"""审计异常标题，并可用全文中的标题安全矫正。""" import csv import os from django.conf import settings from django.core.management.base import BaseCommand, CommandError from django.db.models.functions import Length from django.utils import timezone from apps.resm.models import Paper from apps.resm.title_utils import ( FulltextTitleCandidate, assess_title, clean_title, extract_title_from_pdf, extract_title_from_xml, is_better_title, is_usable_title, ) def _paper_path(paper, ext): safe_doi = paper.doi.replace("/", "_") if paper.publication_date is None: directory = os.path.join(settings.BASE_DIR, "media", "papers", "unknown") else: d = paper.publication_date directory = os.path.join( settings.BASE_DIR, "media", "papers", str(d.year), str(d.month), str(d.day) ) return os.path.join(directory, f"{safe_doi}.{ext}") def _fulltext_candidate(paper): candidates = [] xml_path = _paper_path(paper, "xml") pdf_path = _paper_path(paper, "pdf") # 历史状态位可能与磁盘不一致；审计以实际文件为准，避免“文件存在但 flag=false”漏检。 if paper.has_fulltext_xml or os.path.isfile(xml_path): candidate = extract_title_from_xml(xml_path, paper.doi) if candidate: candidates.append(candidate) if paper.has_fulltext_pdf or os.path.isfile(pdf_path): candidate = extract_title_from_pdf( pdf_path, paper.doi, paper.first_author or "" ) if candidate: candidates.append(candidate) rank = {"high": 3, "medium": 2, "low": 1} return max(candidates, key=lambda item: rank[item.confidence], default=None) class Command(BaseCommand): help = "审计摘要化/网页化标题，并可用 fulltext XML/PDF 标题矫正" def add_arguments(self, parser): parser.add_argument("--doi", action="append", default=[], help="只处理指定 DOI，可重复") parser.add_argument("--min-length", type=int, default=300, help="初筛标题最小长度") parser.add_argument("--limit", type=int, default=0, help="最多检查多少条，0=不限") parser.add_argument("--apply", action="store_true", help="实际写库；默认仅审计") parser.add_argument( "--allow-medium", action="store_true", help="允许采用 medium 置信候选；默认仅 XML+DOI 的 high 候选自动矫正", ) parser.add_argument("--csv", default="", help="将审计结果写入 CSV") parser.add_argument("--title", default="", help="单个 --doi 的人工确认标题") parser.add_argument("--source", default="manual", help="人工标题来源标记") def handle(self, *args, **opts): dois = [doi.strip().lower() for doi in opts["doi"] if doi.strip()] manual_title = clean_title(opts["title"]) if manual_title and len(dois) != 1: raise CommandError("--title 必须与且只能与一个 --doi 一起使用") if manual_title and not is_usable_title(manual_title): raise CommandError("--title 不是可接受的标题") qs = Paper.objects.all().order_by("id") if dois: qs = qs.filter(doi__in=dois) else: qs = qs.annotate(title_len=Length("title")).filter( title_len__gte=max(1, opts["min_length"]) ) if opts["limit"]: qs = qs[:opts["limit"]] csv_file = None writer = None if opts["csv"]: csv_file = open(opts["csv"], "w", newline="", encoding="utf-8-sig") writer = csv.DictWriter(csv_file, fieldnames=( "doi", "openalex_id", "old_title", "reasons", "candidate_title", "candidate_source", "confidence", "evidence", "action", )) writer.writeheader() checked = suspect_count = corrected = candidates = 0 try: for paper in qs.iterator(chunk_size=500): checked += 1 assessment = assess_title(paper.title) if not assessment.suspect and not manual_title: continue suspect_count += int(assessment.suspect) if manual_title: candidate = FulltextTitleCandidate( title=manual_title, source=opts["source"][:30], confidence="high", evidence="manual_confirmed", ) else: candidate = _fulltext_candidate(paper) if candidate: candidates += 1 allowed = candidate and ( candidate.confidence == "high" or (opts["allow_medium"] and candidate.confidence == "medium") ) should_correct = bool( candidate and allowed and (manual_title or is_better_title(paper.title, candidate.title)) ) action = "candidate" if not candidate: action = "no_candidate" elif not allowed: action = "needs_review" elif not should_correct: action = "rejected" if opts["apply"]: if should_correct: if not paper.title_raw: paper.title_raw = paper.title paper.title = candidate.title paper.title_source = candidate.source paper.title_quality_status = Paper.TITLE_CORRECTED paper.title_verified_at = timezone.now() paper.save(update_fields=[ "title", "title_raw", "title_source", "title_quality_status", "title_verified_at", "update_time", ]) corrected += 1 action = "corrected" elif assessment.suspect and paper.title_quality_status == Paper.TITLE_UNCHECKED: paper.title_quality_status = Paper.TITLE_SUSPECT paper.save(update_fields=["title_quality_status", "update_time"]) action = "marked_suspect" row = { "doi": paper.doi, "openalex_id": paper.openalex_id or "", "old_title": paper.title_raw if action == "corrected" else paper.title, "reasons": ",".join(assessment.reasons), "candidate_title": candidate.title if candidate else "", "candidate_source": candidate.source if candidate else "", "confidence": candidate.confidence if candidate else "", "evidence": candidate.evidence if candidate else "", "action": action, } if writer: writer.writerow(row) self.stdout.write( f"[{action}] {paper.doi} len={assessment.plain_length} " f"reasons={row['reasons']} candidate={row['candidate_source'] or '-'} " f"confidence={row['confidence'] or '-'}" ) finally: if csv_file: csv_file.close() mode = "apply" if opts["apply"] else "dry-run" self.stdout.write(self.style.SUCCESS( f"完成({mode}) checked={checked} suspect={suspect_count} " f"fulltext_candidates={candidates} corrected={corrected}" ))