feat(resm): correct suspect titles from fulltext

CLAUDE.md

@@ -174,6 +174,13 @@ The paper fetch pipeline in `apps/resm/tasks.py` currently includes:
 - PDF fetch from Elsevier
 - Sci-Hub fallback
 - task fan-out and stuck-download release
+- suspicious-title detection, targeted OpenAlex title refresh, and high-confidence
+  correction from fulltext XML when the XML DOI matches the paper
+
+Title quality is tracked on `Paper` with the original value retained only when a correction
+is applied. `python manage.py audit_paper_titles` performs a dry-run audit by default;
+`--apply` writes only high-confidence XML corrections unless `--allow-medium` is explicitly
+provided for reviewed PDF candidates.
 
 Download behavior is stateful:
 

apps/resm/management/commands/audit_paper_titles.py

@@ -0,0 +1,176 @@
+"""审计异常标题，并可用全文中的标题安全矫正。"""
+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 = []
+    if paper.has_fulltext_xml:
+        candidate = extract_title_from_xml(_paper_path(paper, "xml"), paper.doi)
+        if candidate:
+            candidates.append(candidate)
+    if paper.has_fulltext_pdf:
+        candidate = extract_title_from_pdf(
+            _paper_path(paper, "pdf"), 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}"
+        ))

apps/resm/migrations/0012_paper_title_quality.py

@@ -0,0 +1,75 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    # resm_paper 现有数百万行；质量状态索引用 CONCURRENTLY 创建，避免长时间阻塞写入。
+    atomic = False
+
+    dependencies = [
+        ("resm", "0011_seed_briefing_monitors"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="paper",
+            name="title_raw",
+            field=models.TextField(blank=True, null=True),
+        ),
+        migrations.AddField(
+            model_name="paper",
+            name="title_source",
+            field=models.CharField(blank=True, max_length=30, null=True),
+        ),
+        migrations.SeparateDatabaseAndState(
+            database_operations=[
+                migrations.AddField(
+                    model_name="paper",
+                    name="title_quality_status",
+                    field=models.CharField(
+                        choices=[
+                            ("unchecked", "未检查"),
+                            ("suspect", "疑似异常"),
+                            ("verified", "已验证"),
+                            ("corrected", "已矫正"),
+                        ],
+                        default="unchecked",
+                        max_length=20,
+                    ),
+                ),
+            ],
+            state_operations=[
+                migrations.AddField(
+                    model_name="paper",
+                    name="title_quality_status",
+                    field=models.CharField(
+                        choices=[
+                            ("unchecked", "未检查"),
+                            ("suspect", "疑似异常"),
+                            ("verified", "已验证"),
+                            ("corrected", "已矫正"),
+                        ],
+                        db_index=True,
+                        default="unchecked",
+                        max_length=20,
+                    ),
+                ),
+            ],
+        ),
+        migrations.RunSQL(
+            sql=(
+                "CREATE INDEX CONCURRENTLY IF NOT EXISTS "
+                "resm_paper_title_quality_status_idx "
+                "ON resm_paper (title_quality_status);"
+            ),
+            reverse_sql=(
+                "DROP INDEX CONCURRENTLY IF EXISTS "
+                "resm_paper_title_quality_status_idx;"
+            ),
+        ),
+        migrations.AddField(
+            model_name="paper",
+            name="title_verified_at",
+            field=models.DateTimeField(blank=True, null=True),
+        ),
+    ]

apps/resm/models.py

@@ -5,12 +5,30 @@ import os
 # Create your models here.
 
 class Paper(BaseModel):
+    TITLE_UNCHECKED = "unchecked"
+    TITLE_SUSPECT = "suspect"
+    TITLE_VERIFIED = "verified"
+    TITLE_CORRECTED = "corrected"
+    TITLE_QUALITY_CHOICES = (
+        (TITLE_UNCHECKED, "未检查"),
+        (TITLE_SUSPECT, "疑似异常"),
+        (TITLE_VERIFIED, "已验证"),
+        (TITLE_CORRECTED, "已矫正"),
+    )
     # ===== 全局唯一标识 =====
     openalex_id = models.TextField(unique=True, verbose_name="OpenAlex ID", null=True, blank=True)
     doi = models.TextField(unique=True, verbose_name="DOI")
     # ===== 基本信息 =====
     type = models.CharField(max_length=20, db_index=True)
     title = models.TextField()
+    # title_raw 仅在矫正时保存旧值，避免为全部历史数据复制一份大字段。
+    title_raw = models.TextField(null=True, blank=True)
+    title_source = models.CharField(max_length=30, null=True, blank=True)
+    title_quality_status = models.CharField(
+        max_length=20, choices=TITLE_QUALITY_CHOICES,
+        default=TITLE_UNCHECKED, db_index=True,
+    )
+    title_verified_at = models.DateTimeField(null=True, blank=True)
     publication_date = models.DateField(null=True, blank=True)
     publication_year = models.IntegerField(db_index=True)
     # ===== 作者（最小可用集）=====

apps/resm/tasks.py

@@ -12,6 +12,7 @@ from celery import current_app
 from datetime import datetime, timedelta
 import random
 from .pdf_utils import _is_elsevier_preview_pdf
+from .title_utils import assess_title, extract_title_from_xml, is_better_title
 from uuid import uuid4
 import asyncio
 import sys
@@ -93,7 +94,10 @@ def _build_paper_from_record(record, keywords: str, search: str) -> Paper:
     paper.type = (record.get("type") or "article")[:20]
     paper.openalex_id = record["id"].split("/")[-1]
     paper.doi = record["doi"].replace("https://doi.org/", "")
-    paper.title = record["display_name"]
+    paper.title = record.get("title") or record["display_name"]
+    paper.title_source = "openalex"
+    if assess_title(paper.title).suspect:
+        paper.title_quality_status = Paper.TITLE_SUSPECT
     paper.publication_date = record["publication_date"]
     paper.publication_year = record["publication_year"]
     if record["open_access"]:
@@ -226,6 +230,42 @@ def update_paper_meta_from_openalex(days: int = 30, per_combo_max: int = None):
     return f"openalex update: combos={n_combos}, new_papers={new_papers}, since={from_publication_date}"
 
 
+@shared_task(base=CustomTask)
+def refresh_suspect_titles_from_openalex(limit: int = 100):
+    """定向刷新疑似异常标题，吸收 OpenAlex/Crossref 后续元数据修正。
+
+    常规抓取使用 bulk_create(ignore_conflicts=True)，不会更新已有 DOI；本任务只检查
+    已标记 suspect 的记录，并且仅在新标题通过质量检查且明显优于旧值时覆盖。
+    """
+    qs = Paper.objects.filter(
+        title_quality_status=Paper.TITLE_SUSPECT,
+        openalex_id__isnull=False,
+    ).exclude(openalex_id="").order_by("id")[:max(1, limit)]
+    checked = corrected = failed = 0
+    for paper in qs:
+        checked += 1
+        try:
+            record = Works()[paper.openalex_id]
+        except Exception:
+            failed += 1
+            continue
+        candidate = (record or {}).get("title") or (record or {}).get("display_name")
+        if not is_better_title(paper.title, candidate):
+            continue
+        if not paper.title_raw:
+            paper.title_raw = paper.title
+        paper.title = assess_title(candidate).title
+        paper.title_source = "openalex_refresh"
+        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
+    return f"title refresh: checked={checked}, corrected={corrected}, failed={failed}"
+
+
 BACKFILL_STOP_KEY = "backfill_paper_meta_stop"
 
 
@@ -374,6 +414,9 @@ def _build_paper_from_sd_result(r, qs_text: str):
     paper.o_search = qs_text
     paper.doi = str(doi).replace("https://doi.org/", "")
     paper.title = title
+    paper.title_source = "elsevier"
+    if assess_title(paper.title).suspect:
+        paper.title_quality_status = Paper.TITLE_SUSPECT
     paper.publication_date = pub_date if len(str(pub_date)) == 10 else None
     paper.publication_year = year
     paper.publication_name = r.get("sourceTitle")
@@ -625,8 +668,24 @@ def _elsevier_fetch_xml(req, paper):
         paper.has_fulltext_xml = True
 
     paper.save_file_xml(xml_str)
-    paper.save(update_fields=["has_abstract", "has_abstract_xml",
-                              "has_fulltext", "has_fulltext_xml", "update_time"])
+    update_fields = ["has_abstract", "has_abstract_xml",
+                     "has_fulltext", "has_fulltext_xml", "update_time"]
+    # XML 标题节点且 DOI 匹配才是 high；仅在旧标题已呈摘要化/网页化时自动覆盖。
+    if paper.title_quality_status == Paper.TITLE_SUSPECT or assess_title(paper.title).suspect:
+        candidate = extract_title_from_xml(paper.init_paper_path("xml"), paper.doi)
+        if (candidate and candidate.confidence == "high"
+                and is_better_title(paper.title, candidate.title)):
+            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()
+            update_fields.extend([
+                "title", "title_raw", "title_source", "title_quality_status",
+                "title_verified_at",
+            ])
+    paper.save(update_fields=update_fields)
     return True, has_fulltext, None
 
 

apps/resm/test_title_utils.py

@@ -0,0 +1,102 @@
+import os
+import tempfile
+import types
+import unittest
+from unittest.mock import patch
+
+from apps.resm.title_utils import (
+    assess_title,
+    extract_title_from_pdf,
+    extract_title_from_xml,
+    is_better_title,
+)
+
+
+class TitleAssessmentTests(unittest.TestCase):
+    def test_normal_title_is_not_suspect(self):
+        result = assess_title("Optimal Nonlinear PID Speed Control for an Electric Vehicle")
+        self.assertFalse(result.suspect)
+
+    def test_abstract_stored_as_title_is_suspect(self):
+        text = (
+            "In many situations, a beam opening is necessary near a plastic hinge. "
+            "However, few studies investigated the resulting behavior. "
+            "In this study, nine full-scale joints were tested with several configurations. "
+            "The results show that additional reinforcement improves strength and ductility. "
+        ) * 2
+        result = assess_title(text)
+        self.assertTrue(result.suspect)
+        self.assertIn("abstract_like_prose", result.reasons)
+
+    def test_mathml_markup_does_not_trigger_raw_length_false_positive(self):
+        markup = "<mml:math>" + "<mml:mi>x</mml:mi>" * 100 + "</mml:math>"
+        result = assess_title(f"Behavior of {markup} under pressure")
+        self.assertFalse(result.suspect)
+
+    def test_short_candidate_is_better_than_abstract(self):
+        old = ("This study investigates reinforced concrete joints. " * 20).strip()
+        new = "Cyclic Behavior of Reinforced Concrete Beam-Column Joints"
+        self.assertTrue(is_better_title(old, new))
+
+
+class XmlTitleExtractionTests(unittest.TestCase):
+    def _xml_file(self, body):
+        handle = tempfile.NamedTemporaryFile("w", suffix=".xml", delete=False, encoding="utf-8")
+        self.addCleanup(lambda: os.path.exists(handle.name) and os.unlink(handle.name))
+        handle.write(body)
+        handle.close()
+        return handle.name
+
+    def test_extracts_elsevier_title_and_verifies_doi(self):
+        path = self._xml_file("""<?xml version="1.0"?>
+        <response xmlns:dc="http://purl.org/dc/elements/1.1/"
+                  xmlns:prism="http://prismstandard.org/namespaces/basic/2.0/">
+          <coredata>
+            <prism:doi>10.1234/example.1</prism:doi>
+            <dc:title>Correct &amp; Verified Article Title</dc:title>
+          </coredata>
+        </response>""")
+        candidate = extract_title_from_xml(path, "10.1234/example.1")
+        self.assertIsNotNone(candidate)
+        self.assertEqual(candidate.title, "Correct & Verified Article Title")
+        self.assertEqual(candidate.confidence, "high")
+
+    def test_rejects_xml_for_another_doi(self):
+        path = self._xml_file("""<article>
+          <article-meta><article-id pub-id-type="doi">10.1234/wrong</article-id>
+          <title-group><article-title>A Correct Looking Title</article-title></title-group>
+          </article-meta></article>""")
+        candidate = extract_title_from_xml(path, "10.1234/expected")
+        self.assertIsNone(candidate)
+
+
+class PdfTitleExtractionTests(unittest.TestCase):
+    def test_pdf_metadata_requires_identity_evidence_and_stays_medium(self):
+        handle = tempfile.NamedTemporaryFile("wb", suffix=".pdf", delete=False)
+        self.addCleanup(lambda: os.path.exists(handle.name) and os.unlink(handle.name))
+        handle.write(b"%PDF-placeholder")
+        handle.close()
+
+        page = types.SimpleNamespace(
+            extract_text=lambda: (
+                "Correct Article Title\nMohamed Shamseldin\n"
+                "https://doi.org/10.1234/example.1\nAbstract"
+            )
+        )
+        reader = types.SimpleNamespace(
+            pages=[page],
+            metadata=types.SimpleNamespace(title="Correct Article Title"),
+        )
+        module = types.SimpleNamespace(PdfReader=lambda *args, **kwargs: reader)
+        with patch.dict("sys.modules", {"pypdf": module}):
+            candidate = extract_title_from_pdf(
+                handle.name, "10.1234/example.1", "Mohamed Shamseldin"
+            )
+        self.assertIsNotNone(candidate)
+        self.assertEqual(candidate.title, "Correct Article Title")
+        self.assertEqual(candidate.confidence, "medium")
+        self.assertEqual(candidate.evidence, "pdf_metadata+doi")
+
+
+if __name__ == "__main__":
+    unittest.main()

apps/resm/title_utils.py

@@ -0,0 +1,256 @@
+"""论文标题质量检测与全文标题提取工具。
+
+本模块不依赖 Django，管理命令、抓取任务和单元测试都可以复用。全文来源的
+可信度按 XML(标题节点且 DOI 匹配) > PDF 元数据(首页身份匹配) > PDF 首页版式
+推断排序；调用方默认只应自动采用 high 置信度候选。
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from html import unescape
+import logging
+import os
+import re
+from typing import Optional
+
+
+_TAG_RE = re.compile(r"<[^>]+>")
+_SPACE_RE = re.compile(r"\s+")
+_WORD_RE = re.compile(r"[\w'-]+", re.UNICODE)
+_DOI_RE = re.compile(r"10\.\d{4,9}/[-._;()/:A-Z0-9]+", re.I)
+_WEB_MARKERS = (
+    "journal help", "user username", "remember me login", "current issue",
+    "paper submission", "guidelines for authors", "view subscribe",
+)
+_DOCUMENT_MARKERS = ("author:", "authors:", "abstract:", "document type:")
+_GENERIC_PDF_TITLES = {
+    "untitled", "microsoft word", "document", "article", "full text", "pdf",
+}
+
+
+@dataclass(frozen=True)
+class TitleAssessment:
+    title: str
+    plain_length: int
+    word_count: int
+    sentence_count: int
+    suspect: bool
+    reasons: tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class FulltextTitleCandidate:
+    title: str
+    source: str
+    confidence: str
+    evidence: str
+
+
+def clean_title(value: Optional[str]) -> str:
+    """去标签、解实体并合并空白，保留公式中的可见文本。"""
+    if not value:
+        return ""
+    text = unescape(_TAG_RE.sub(" ", str(value)))
+    text = _SPACE_RE.sub(" ", text).strip(" \t\r\n\ufeff")
+    if text.lower().startswith("title:"):
+        text = text[6:].strip()
+    return text
+
+
+def normalize_doi(value: Optional[str]) -> str:
+    if not value:
+        return ""
+    value = unescape(str(value)).strip().lower()
+    value = re.sub(r"^(?:doi:\s*|https?://(?:dx\.)?doi\.org/)", "", value)
+    match = _DOI_RE.search(value)
+    return match.group(0).rstrip(".,;)").lower() if match else value.rstrip(".,;)")
+
+
+def assess_title(value: Optional[str]) -> TitleAssessment:
+    plain = clean_title(value)
+    lower = plain.lower()
+    words = len(_WORD_RE.findall(plain))
+    sentences = len(re.findall(r"[.!?](?:\s|$)", plain))
+    reasons = []
+
+    # 去标签后的长度才用于判断，避免 MathML 本身把正常标题撑到几千字符。
+    if len(plain) >= 300 and words >= 50 and sentences >= 3:
+        reasons.append("abstract_like_prose")
+    web_hits = sum(marker in lower for marker in _WEB_MARKERS)
+    if len(plain) >= 200 and web_hits >= 2:
+        reasons.append("webpage_text")
+    doc_hits = sum(marker in lower for marker in _DOCUMENT_MARKERS)
+    if len(plain) >= 300 and doc_hits >= 2:
+        reasons.append("document_metadata")
+    if len(plain) >= 800 and words >= 80:
+        reasons.append("extreme_length")
+
+    return TitleAssessment(
+        title=plain,
+        plain_length=len(plain),
+        word_count=words,
+        sentence_count=sentences,
+        suspect=bool(reasons),
+        reasons=tuple(reasons),
+    )
+
+
+def is_usable_title(value: Optional[str]) -> bool:
+    assessment = assess_title(value)
+    if not assessment.title or assessment.suspect:
+        return False
+    if assessment.plain_length < 5 or assessment.plain_length > 500:
+        return False
+    if assessment.word_count > 70 or assessment.sentence_count > 2:
+        return False
+    return True
+
+
+def is_better_title(current: Optional[str], candidate: Optional[str]) -> bool:
+    old = assess_title(current)
+    new = assess_title(candidate)
+    if not old.suspect or not is_usable_title(new.title):
+        return False
+    normalized_old = re.sub(r"\W+", "", old.title).casefold()
+    normalized_new = re.sub(r"\W+", "", new.title).casefold()
+    if not normalized_new or normalized_new == normalized_old:
+        return False
+    return new.plain_length < old.plain_length * 0.75
+
+
+def _element_text(element) -> str:
+    return clean_title(" ".join(element.itertext()))
+
+
+def extract_title_from_xml(
+    path: str,
+    expected_doi: Optional[str] = None,
+) -> Optional[FulltextTitleCandidate]:
+    """从 Elsevier/JATS/常见全文 XML 中提取文章标题。
+
+    XML 内存在 DOI 时必须与目标论文一致；不一致直接拒绝，防止历史文件错配。
+    """
+    if not os.path.exists(path):
+        return None
+    try:
+        from lxml import etree
+
+        parser = etree.XMLParser(recover=True, huge_tree=True, resolve_entities=False)
+        root = etree.parse(path, parser).getroot()
+    except Exception:
+        return None
+
+    expected = normalize_doi(expected_doi)
+    doi_values = []
+    for node in root.xpath(
+        "//*[local-name()='doi' or local-name()='identifier' "
+        "or (local-name()='article-id' and "
+        "translate(@pub-id-type, 'DOI', 'doi')='doi')]"
+    ):
+        value = normalize_doi(_element_text(node))
+        if value.startswith("10."):
+            doi_values.append(value)
+    if expected and doi_values and expected not in doi_values:
+        return None
+
+    title_xpaths = (
+        "//*[local-name()='coredata']/*[local-name()='title']",
+        "//*[local-name()='article-meta']/*[local-name()='title-group']/*[local-name()='article-title']",
+        "//*[local-name()='article-title']",
+        "/*/*[local-name()='coredata']/*[local-name()='title']",
+    )
+    for xpath in title_xpaths:
+        for node in root.xpath(xpath):
+            title = _element_text(node)
+            if not is_usable_title(title):
+                continue
+            doi_matched = bool(expected and expected in doi_values)
+            return FulltextTitleCandidate(
+                title=title,
+                source="fulltext_xml",
+                confidence="high" if doi_matched else "medium",
+                evidence="xml_title_node+doi" if doi_matched else "xml_title_node",
+            )
+    return None
+
+
+def _identity_evidence(text: str, expected_doi: str, first_author: str) -> str:
+    if expected_doi:
+        found = {normalize_doi(v) for v in _DOI_RE.findall(text)}
+        if normalize_doi(expected_doi) in found:
+            return "doi"
+    if first_author:
+        surname = clean_title(first_author).split()[-1].casefold()
+        if len(surname) >= 3 and surname in text.casefold():
+            return "author"
+    return ""
+
+
+def _title_from_first_page(text: str, first_author: str) -> str:
+    """保守地从首页文本猜标题，仅返回低/中置信候选。"""
+    lines = [clean_title(line) for line in text.splitlines()]
+    lines = [line for line in lines if line]
+    if not lines:
+        return ""
+
+    surname = clean_title(first_author).split()[-1].casefold() if first_author else ""
+    author_index = next(
+        (i for i, line in enumerate(lines[:30]) if len(surname) >= 3 and surname in line.casefold()),
+        None,
+    )
+    if author_index is None or author_index == 0:
+        return ""
+
+    candidates = []
+    for line in lines[max(0, author_index - 4):author_index]:
+        lower = line.casefold()
+        if _DOI_RE.search(line) or "journal" in lower or "issn" in lower:
+            continue
+        if re.fullmatch(r"[\d\s|:/.-]+", line):
+            continue
+        candidates.append(line)
+    title = clean_title(" ".join(candidates))
+    return title if is_usable_title(title) else ""
+
+
+def extract_title_from_pdf(
+    path: str,
+    expected_doi: Optional[str] = None,
+    first_author: Optional[str] = None,
+) -> Optional[FulltextTitleCandidate]:
+    """从 PDF metadata 或首页文本提取标题候选。
+
+    PDF metadata 只有在首页 DOI/作者能确认文件身份时才达到 medium；首页版式推断
+    即使 DOI 匹配也保持 medium，默认不会被管理命令自动写库。
+    """
+    if not os.path.exists(path):
+        return None
+    try:
+        logging.getLogger("pypdf").setLevel(logging.CRITICAL)
+        from pypdf import PdfReader
+
+        reader = PdfReader(path, strict=False)
+        page_text = reader.pages[0].extract_text() or "" if reader.pages else ""
+        metadata = reader.metadata
+    except Exception:
+        return None
+
+    identity = _identity_evidence(page_text, expected_doi or "", first_author or "")
+    meta_title = clean_title(getattr(metadata, "title", "") if metadata else "")
+    if meta_title and meta_title.casefold() not in _GENERIC_PDF_TITLES and is_usable_title(meta_title):
+        return FulltextTitleCandidate(
+            title=meta_title,
+            source="pdf_metadata",
+            confidence="medium" if identity else "low",
+            evidence=f"pdf_metadata+{identity}" if identity else "pdf_metadata_only",
+        )
+
+    page_title = _title_from_first_page(page_text, first_author or "")
+    if page_title:
+        return FulltextTitleCandidate(
+            title=page_title,
+            source="pdf_first_page",
+            confidence="medium" if identity else "low",
+            evidence=f"pdf_first_page+{identity}" if identity else "pdf_first_page_only",
+        )
+    return None