21 changed files with 157 additions and 1781 deletions
--- a/.gitignore
+++ b/.gitignore
@ -25,5 +25,4 @@ config/conf*.json
 sh/*
 temp/*
 nohup.out
-scripts/*
+scripts/*
 !scripts/sd_download.py
--- a/CLAUDE.md
+++ b/CLAUDE.md
@ -163,8 +163,6 @@ Key model:
 - `Paper`: stores DOI/OpenAlex metadata, OA flags, abstract/fulltext state, fetch status, failure reason, and local file save helpers
 - `PaperAbstract`: separate abstract storage
 Read access to paper APIs (`GET api/resm/paper/` and the PDF view) requires either a valid API Key or a logged-in JWT user. API Keys live in `apps.system.models.ApiKey` (plaintext `key` auto-generated on save, managed via Django admin) and are validated by `apps.utils.apikey_auth.ApiKeyAuthentication` from the `X-API-Key` header or `?api_key=` query param.
 The paper fetch pipeline in `apps/resm/tasks.py` currently includes:
 - metadata ingestion from OpenAlex
@ -174,21 +172,10 @@ 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. The audit checks physical XML/PDF files even when
 legacy `has_fulltext_*` flags are stale. PDF first-page layout guesses always remain low
 confidence and are never bulk-applied; PDF metadata also rejects filenames and generic titles.
 `rollback_pdf_title_corrections` restores PDF-derived corrections from `title_raw` and is a
 dry-run unless `--apply` is passed.
 Download behavior is stateful:
- `fetch_status` is a coarse lock: `"downloading"` for the elsevier/openalex keep-alive chains, `"downloading_pdf"` for the `download_pdf` chain; the `send_download_fulltext_task` concurrency gate counts only `downloading_pdf` so the other chains' locks cannot starve it (queries exclude via `fetch_status__startswith="downloading"`)
+- `fetch_status="downloading"` is used as a coarse lock
 - `fail_reason` accumulates fetch failures
 - files are stored under `media/papers/<year>/<month>/<day>/`
--- a/apps/ops/views.py
+++ b/apps/ops/views.py
@ -250,11 +250,6 @@ class CacheView(APIView):
        cache.set(key, vdata['value'], timeout=vdata["timeout"])
        return Response()
    @swagger_auto_schema(operation_summary="删除key", responses={200: None})
    def delete(self, request, key):
        cache.delete(key)
        return Response()
 class DrfRequestLogViewSet(ListModelMixin, CustomGenericViewSet):
    """list:请求日志
--- a/apps/resm/d_scihub.py
+++ b/apps/resm/d_scihub.py
@ -1,11 +1,8 @@
 import argparse
 import asyncio
 import logging
 import os
 import re
 from pathlib import Path
 from typing import Optional
 from urllib.parse import urljoin
 from playwright.async_api import async_playwright, Page, Browser
 # 尝试导入 playwright-stealth
@ -292,146 +289,6 @@ _SCIHUB_DOMAINS = [
    "sci-hub.se",
 ]
 # bot 挑战页标题/正文关键词: 命中则需浏览器兜底
 _CHALLENGE_MARKERS = ("just a moment", "cloudflare", "checking your browser",
                      "ddos-guard", "ddos guard", "attention required")
 # Sci-Hub 缓存当前存活域名, 避免每篇都从头探测已挂的域名
 _ALIVE_DOMAIN_KEY = "scihub_alive_domain"
 _ALIVE_DOMAIN_TTL = 600  # 秒
 def _normalize_pdf_url(raw: str, base_url: str) -> str:
    """把 Sci-Hub 页面里抓到的 PDF 地址归一成可直接请求的绝对 URL。
    处理: JSON 转义 \\/ 、#fragment、协议相对 // 、站内相对路径。"""
    raw = raw.strip().replace("\\/", "/").split("#")[0]
    if raw.startswith("//"):
        return "https:" + raw
    if raw.startswith("http"):
        return raw
    return urljoin(base_url, raw)
 def _extract_scihub_pdf_url(html: str, base_url: str) -> Optional[str]:
    """从 Sci-Hub 页面 HTML 提取内嵌 PDF 地址, 覆盖各域名/版本的常见结构;
    找不到返回 None。lxml 优先(结构化), 正则兜底(应对畸形 HTML/JS 赋值)。"""
    try:
        from lxml import etree
        root = etree.HTML(html)
        if root is not None:
            for xp in ('//embed[@id="pdf"]/@src',
                       '//iframe[@id="pdf"]/@src',
                       '//div[@id="article"]//embed/@src',
                       '//div[@id="article"]//iframe/@src',
                       '//embed[@type="application/pdf"]/@src'):
                hit = root.xpath(xp)
                if hit and hit[0].strip():
                    return _normalize_pdf_url(hit[0], base_url)
    except Exception:
        pass
    # 正则兜底: 按钮 location.href='...pdf...' / 任意 src|href 指向 .pdf
    for pat in (r"location\.href\s*=\s*['\"]([^'\"]+?\.pdf[^'\"]*)['\"]",
                r"(?:src|href)\s*=\s*['\"]([^'\"]*?/downloads/[^'\"]+?\.pdf[^'\"]*)['\"]",
                r"(?:src|href)\s*=\s*['\"]([^'\"]*?\.pdf(?:\?[^'\"]*)?)['\"]"):
        m = re.search(pat, html, re.I)
        if m and m.group(1).strip():
            return _normalize_pdf_url(m.group(1), base_url)
    return None
 def _save_pdf_bytes(content: bytes, output: str) -> bool:
    """校验 PDF 魔数 + 最小体积后落盘; 不合格返回 False。"""
    if not content or content[:4] != b"%PDF" or len(content) < 10240:
        return False
    os.makedirs(os.path.dirname(os.path.abspath(output)), exist_ok=True)
    with open(output, "wb") as f:
        f.write(content)
    return True
 def download_by_doi_http(doi: str, output: str, timeout: int = 30) -> tuple[bool, str]:
    """curl-cffi 轻量下载(worker 内首选, 不起浏览器): 逐域名取 Sci-Hub 页面,
    解析内嵌 PDF 地址后直接下载并校验魔数。绝大多数无挑战的情况几秒内完成。
    返回 (ok, err)。err 前缀:
      - scihub_error_empty_doi:      DOI 为空
      - scihub_curl_cffi_not_installed
      - scihub_need_play:            命中 bot 挑战, 需浏览器兜底(交独立脚本)
      - scihub_error_pdf_not_found:  页面正常但无 PDF / 所有域名 HTTP 均失败
    """
    doi = (doi or "").strip()
    if not doi:
        return False, "scihub_error_empty_doi"
    try:
        import curl_cffi.requests as cf
    except ImportError:
        return False, "scihub_curl_cffi_not_installed"
    try:
        from django.core.cache import cache
    except Exception:
        cache = None
    # 存活域名优先, 其余按默认顺序补上(去重)
    domains = list(_SCIHUB_DOMAINS)
    if cache is not None:
        alive = cache.get(_ALIVE_DOMAIN_KEY)
        if alive in domains:
            domains = [alive] + [d for d in domains if d != alive]
    challenged = False
    for domain in domains:
        page_url = f"https://{domain}/{doi}"
        try:
            resp = cf.get(page_url, impersonate="chrome131",
                          headers={"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8"},
                          timeout=timeout, allow_redirects=True)
        except Exception as e:
            logger.info(f"{domain} 请求失败: {e}")
            continue
        ct = resp.headers.get("content-type", "")
        # 有些域名直接把 PDF 返回在 DOI 页
        if resp.content[:4] == b"%PDF" or "application/pdf" in ct:
            if _save_pdf_bytes(resp.content, output):
                if cache is not None:
                    cache.set(_ALIVE_DOMAIN_KEY, domain, timeout=_ALIVE_DOMAIN_TTL)
                logger.info(f"✓ HTTP 直取 PDF 成功({domain}): {output}")
                return True, ""
        if resp.status_code != 200:
            if resp.status_code in (403, 503):  # 多为挑战/封禁
                challenged = True
            continue
        try:
            text = resp.text
        except Exception:
            continue
        if any(k in text[:3000].lower() for k in _CHALLENGE_MARKERS):
            challenged = True
            continue
        pdf_url = _extract_scihub_pdf_url(text, page_url)
        if not pdf_url:
            continue
        try:
            pr = cf.get(pdf_url, impersonate="chrome131",
                        headers={"Referer": page_url, "Accept": "application/pdf,*/*"},
                        timeout=timeout, allow_redirects=True)
        except Exception as e:
            logger.info(f"{domain} PDF 下载失败: {e}")
            continue
        if pr.status_code == 200 and _save_pdf_bytes(pr.content, output):
            if cache is not None:
                cache.set(_ALIVE_DOMAIN_KEY, domain, timeout=_ALIVE_DOMAIN_TTL)
            logger.info(f"✓ HTTP 解析下载成功({domain}): {output}")
            return True, ""
    if challenged:
        return False, "scihub_need_play: bot 挑战, 需浏览器兜底"
    return False, "scihub_error_pdf_not_found: 所有域名 HTTP 均未获取 PDF"
 def download_paper_by_doi(doi: str, output: Optional[str] = None, headless: bool = True) -> tuple[bool, str]:
    """
--- a/apps/resm/management/commands/audit_paper_titles.py
+++ b/apps/resm/management/commands/audit_paper_titles.py
@ -1,179 +0,0 @@
 """审计异常标题，并可用全文中的标题安全矫正。"""
 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}"
        ))
--- a/apps/resm/management/commands/fix_preview_pdf.py
+++ b/apps/resm/management/commands/fix_preview_pdf.py
@ -1,67 +1,34 @@
-"""一次性修复: 纠正被误标为全文 PDF 的历史记录(Elsevier 摘要预览页 / 损坏文件)。
+"""一次性修复: 把误标为全文 PDF 的 Elsevier "摘要预览页"(1 页)纠正回未下载状态。
 背景:
  Elsevier Article API 对未授权 / in-press 文章, application/pdf 端点会返回仅含
-  摘要的 1 页预览 PDF(魔数仍是 %PDF、体积也不小); 另有部分历史记录把 HTML 错误页 /
+  摘要的 1 页预览 PDF(魔数仍是 %PDF、体积也不小), 而全文 XML 却能正常拿到。旧抓取
-  被截断的垃圾当 PDF 存了。旧抓取逻辑只校验魔数 + 体积, 都会误标 has_fulltext_pdf=True。
+  逻辑只校验魔数 + 体积, 误将预览页落库并置 has_fulltext_pdf=True。
-本命令核对本地 PDF, 分两类处理:
+本命令重新核对本地 PDF 的页数, 对 <= 1 页者:
-  - 预览页(1 页):   has_fulltext_pdf 置回 False; 文件**仅在 --delete-file 时**删除。
+  - has_fulltext_pdf 置回 False
-  - 损坏文件(非 PDF / pypdf 解析失败): has_fulltext_pdf 置回 False; 文件**总是删除**
+  - 若该论文有 XML 全文(has_fulltext_xml=True), 保留 has_fulltext=True;
-    (dry-run 除外), 因为它根本不是有效全文, 留着无用且会污染下游解析。
+    否则(此前只有这张假预览页冒充全文)一并把 has_fulltext 回退为 False,
-  两类在缺少 XML 全文(has_fulltext_xml=False)时, 一并把 has_fulltext 回退 False,
+    让它能重新进入下载链路去找真正的全文。
-  让其重新进入下载链路去找真正的全文; 并追加 fail_reason 标记供抓取任务排除。
+  - 追加 fail_reason 'elsevier_pdf_preview_only' (供 Elsevier 补抓队列排除, 避免无限重试)
  - 可选: 删除本地预览 PDF 文件 (--delete-file)
-性能:
+文件读取依赖本地存在 PDF (在跑抓取的服务器上执行)。建议先 --dry-run 看统计。
  读文件 + pypdf 解析是 CPU/IO 密集, 用 ProcessPoolExecutor 并行(--workers, 默认 CPU 核数);
  数据库写入留在主进程串行(坏文件仅占少数, 非瓶颈, 也避免子进程共享 DB 连接)。
 安全前提:
  "损坏"只在铁证下判定 —— 文件不以 %PDF 开头, 或已装 pypdf 且解析直接失败。
  若未装 pypdf 且魔数正常但页数判不出, 归为 unknown, **不处理、绝不删除**。
  强烈建议先 `pip install pypdf` 再跑, 否则只能处理魔数明显不符的坏文件。
 用法:
  python manage.py fix_preview_pdf --dry-run
-  python manage.py fix_preview_pdf                      # 纠正标记 + 删坏文件, 保留预览页文件
+  python manage.py fix_preview_pdf --delete-file
  python manage.py fix_preview_pdf --delete-file        # 并删除预览页文件
  python manage.py fix_preview_pdf --workers 16         # 指定并发进程数
 """
 import os
 from concurrent.futures import ProcessPoolExecutor
 from django.conf import settings
 from django.core.management.base import BaseCommand
 from django.utils import timezone
 from apps.resm.models import Paper
-from apps.resm.pdf_utils import classify_pdf_file
+from apps.resm.tasks import _pdf_page_count
 def _pdf_path(doi, pub_date):
    """按 doi + publication_date 推算 PDF 落盘路径(不创建目录, 只读用)。"""
    safe = doi.replace("/", "_")
    if pub_date is None:
        d = os.path.join(settings.BASE_DIR, "media/papers", "unknown")
    else:
        d = os.path.join(settings.BASE_DIR, "media/papers",
                         str(pub_date.year), str(pub_date.month), str(pub_date.day))
    return os.path.join(d, f"{safe}.pdf")
 def _batched(iterable, size):
    batch = []
    for item in iterable:
        batch.append(item)
        if len(batch) >= size:
            yield batch
            batch = []
    if batch:
        yield batch
 class Command(BaseCommand):
-    help = "纠正被误标为全文的 Elsevier 预览页 / 损坏 PDF(多进程并发)"
+    help = "纠正被误标为全文的 Elsevier 摘要预览 PDF(1 页)"
    def add_arguments(self, parser):
        parser.add_argument("--dry-run", action="store_true",
@ -69,98 +36,71 @@ class Command(BaseCommand):
        parser.add_argument("--limit", type=int, default=0,
                            help="最多处理多少条 (0=不限)")
        parser.add_argument("--delete-file", action="store_true",
-                            help="同时删除预览页文件(坏文件无论该开关都会删)")
+                            help="同时删除本地预览 PDF 文件")
        parser.add_argument("--workers", type=int, default=0,
                            help="并发进程数 (0=CPU 核数)")
        parser.add_argument("--batch", type=int, default=2000,
                            help="每批处理多少条(控制内存与进度粒度)")
    def handle(self, *args, **opts):
        dry = opts["dry_run"]
        limit = opts["limit"]
-        del_preview = opts["delete_file"]
+        del_file = opts["delete_file"]
        batch = max(1, opts["batch"])
        workers = opts["workers"] or (os.cpu_count() or 4)
        qs = Paper.objects.filter(
            has_fulltext_pdf=True, doi__startswith="10.1016"
        ).order_by("id")
        total = qs.count()
        self.stdout.write(
-            f"候选(has_fulltext_pdf=True 且 DOI 以 10.1016 开头): {total}; "
+            f"候选(has_fulltext_pdf=True 且 DOI 以 10.1016 开头): {total}")
            f"workers={workers} batch={batch}"
            + ("  (dry-run)" if dry else ""))
-        rows_iter = qs.values(
+        checked = fixed = only_pdf = missing = unreadable = 0
-            "id", "doi", "publication_date", "has_fulltext_xml", "fail_reason"
+        for paper in qs.iterator():
-        ).iterator(chunk_size=batch)
+            if limit and checked >= limit:
                break
            checked += 1
-        checked = preview = broken = only_pdf = deleted = 0
+            path = paper.init_paper_path("pdf")
-        missing = unknown = 0
+            if not os.path.exists(path):
                missing += 1
                continue
            try:
                with open(path, "rb") as f:
                    content = f.read()
            except OSError:
                unreadable += 1
                continue
-        with ProcessPoolExecutor(max_workers=workers) as ex:
+            pages = _pdf_page_count(content)
-            stop = False
+            if pages is None:
-            for chunk in _batched(rows_iter, batch):
+                unreadable += 1
-                if stop:
+                continue
-                    break
+            if pages > 1:
-                paths = [_pdf_path(r["doi"], r["publication_date"]) for r in chunk]
+                continue  # 真全文, 跳过
                results = ex.map(classify_pdf_file, paths, chunksize=32)
                for r, (_path, kind, pages) in zip(chunk, results):
                    if limit and checked >= limit:
                        stop = True
                        break
                    checked += 1
-                    if kind == "missing":
+            fixed += 1
-                        missing += 1
+            only_pdf_case = not paper.has_fulltext_xml
-                        continue
+            if only_pdf_case:
-                    if kind in ("ok", "unknown", "unreadable"):
+                only_pdf += 1
-                        if kind != "ok":
+            self.stdout.write(
-                            unknown += 1
+                f"[preview {pages}p]{' (only-pdf)' if only_pdf_case else ''} "
-                        continue
+                f"{paper.doi}  {path}")
            if dry:
                continue
-                    # kind in ('preview', 'broken'): 纠正标记
+            paper.has_fulltext_pdf = False
-                    do_delete = (kind == "broken") or del_preview
+            update_fields = ["has_fulltext_pdf", "update_time"]
-                    only_pdf_case = not r["has_fulltext_xml"]
+            # 没有 XML 全文时, 之前的 has_fulltext 只是被这张假预览页置上的, 一并回退
-                    if kind == "preview":
+            if not paper.has_fulltext_xml:
-                        preview += 1
+                paper.has_fulltext = False
-                        tag = f"preview {pages}p"
+                update_fields.insert(0, "has_fulltext")
-                        reason = "elsevier_pdf_preview_only"
+            paper.save(update_fields=update_fields)
-                    else:
+            if "elsevier_pdf_preview_only" not in (paper.fail_reason or ""):
-                        broken += 1
+                paper.save_fail_reason("elsevier_pdf_preview_only")
-                        tag = "broken"
+            if del_file:
-                        reason = "pdf_broken"
+                try:
-                    if only_pdf_case:
+                    os.remove(path)
-                        only_pdf += 1
+                except OSError:
-                    self.stdout.write(
+                    pass
                        f"[{tag}]{' (only-pdf)' if only_pdf_case else ''}"
                        f"{' +rm' if do_delete else ''} {r['doi']}  {_path}")
                    if dry:
                        continue
                    fr = r["fail_reason"]
                    if reason not in (fr or ""):
                        fr = f"{fr};{reason}" if fr else f";{reason}"
                    upd = {"has_fulltext_pdf": False, "fail_reason": fr,
                           "update_time": timezone.now()}
                    if only_pdf_case:
                        upd["has_fulltext"] = False
                    Paper.objects.filter(id=r["id"]).update(**upd)
                    if do_delete:
                        try:
                            os.remove(_path)
                            deleted += 1
                        except OSError:
                            pass
                self.stdout.write(
                    f"  进度 checked={checked}/{total} preview={preview} "
                    f"broken={broken} deleted={deleted} missing={missing} "
                    f"unknown={unknown}")
        self.stdout.write(self.style.SUCCESS(
-            f"完成 检查={checked} 预览页={preview} 坏文件={broken} "
+            f"检查={checked} 预览页修复={fixed} (其中无XML全文/一并回退has_fulltext={only_pdf}) "
-            f"(无XML全文一并回退has_fulltext={only_pdf}) 删除文件={deleted} "
+            f"文件缺失={missing} 无法解析={unreadable}"
            f"文件缺失={missing} 未知/跳过={unknown}"
            + ("  (dry-run, 未写库)" if dry else "")
        ))
--- a/apps/resm/management/commands/rollback_pdf_title_corrections.py
+++ b/apps/resm/management/commands/rollback_pdf_title_corrections.py
@ -1,54 +0,0 @@
 """回滚误用 PDF 候选矫正的标题。"""
 from django.core.management.base import BaseCommand
 from django.utils import timezone
 from apps.resm.models import Paper
 from apps.resm.title_utils import assess_title
 class Command(BaseCommand):
    help = "将 PDF 自动矫正标题恢复为 title_raw；默认 dry-run"
    def add_arguments(self, parser):
        parser.add_argument("--doi", action="append", default=[], help="只回滚指定 DOI，可重复")
        parser.add_argument("--keep-doi", action="append", default=[], help="保留的 DOI，可重复")
        parser.add_argument("--apply", action="store_true", help="实际回滚；默认只预览")
    def handle(self, *args, **opts):
        keep = [doi.strip().lower() for doi in opts["keep_doi"] if doi.strip()]
        selected = [doi.strip().lower() for doi in opts["doi"] if doi.strip()]
        qs = Paper.objects.filter(
            title_quality_status=Paper.TITLE_CORRECTED,
            title_source__in=("pdf_metadata", "pdf_first_page"),
            title_raw__isnull=False,
        ).exclude(title_raw="").order_by("id")
        if selected:
            qs = qs.filter(doi__in=selected)
        if keep:
            qs = qs.exclude(doi__in=keep)
        count = 0
        for paper in qs.iterator(chunk_size=500):
            count += 1
            self.stdout.write(
                f"[rollback] {paper.doi} {paper.title!r} -> {paper.title_raw!r}"
            )
            if not opts["apply"]:
                continue
            restored = paper.title_raw
            paper.title = restored
            paper.title_raw = None
            paper.title_source = None
            paper.title_quality_status = (
                Paper.TITLE_SUSPECT if assess_title(restored).suspect
                else Paper.TITLE_UNCHECKED
            )
            paper.title_verified_at = None
            paper.update_time = timezone.now()
            paper.save(update_fields=[
                "title", "title_raw", "title_source", "title_quality_status",
                "title_verified_at", "update_time",
            ])
        mode = "apply" if opts["apply"] else "dry-run"
        self.stdout.write(self.style.SUCCESS(f"完成({mode}) rollback={count}"))
--- a/apps/resm/migrations/0011_seed_briefing_monitors.py
+++ b/apps/resm/migrations/0011_seed_briefing_monitors.py
@ -1,87 +0,0 @@
 """种子数据:对接《全球材料前沿动态简报》三、前沿科技 检索清单,补充期刊 / 关键词监控。
 简报已列但 0009 已收录的期刊(Ceramics International / CCR / CCC / Construction and
 Building Materials)不重复添加;此处只补简报新增项:
 - 一级检索源(Nature/Science 系顶刊):Nature Materials、Nature Communications、
  Communications Materials、Science Advances、Nature Reviews Materials、Scientific Reports
 - 二级检索源补充(建材 TOP):Engineering Structures、Materials Today
 - 统一检索关键词(简报第三节):低碳水泥 / 储能建材 / 碳化机理 / 固废基胶凝 / 建材碳捕集 等
  (OpenAlex 语料为英文,故 value 用英文搜索词,name 标中文)
 期刊监控只按 ISSN 过滤、不带主题词,Nature Communications / Scientific Reports 等综合性
 大刊会拉入非建材论文;简报要求的"建材主题 + TOP5 筛选"需在下游按关键词二次筛选,本表不承担。
 全部复用每天 05:00 的 monitor_papers 周期任务(0009 已注册),无需新增调度。
 get_or_create 保证迁移可安全重跑。
 """
 from django.db import migrations
 from apps.utils.snowflake import idWorker
 # 一级检索源:Nature/Science 系材料类顶刊(简报「前沿科技」一级)
 JOURNALS_TIER1 = [
    ("1476-1122", "Nature Materials"),
    ("2041-1723", "Nature Communications"),
    ("2662-4443", "Communications Materials"),
    ("2375-2548", "Science Advances"),
    ("2058-8437", "Nature Reviews Materials"),
    ("2045-2322", "Scientific Reports"),
 ]
 NOTE_TIER1 = "前沿顶刊"
 # 二级检索源补充:建材 / 无机非金属国际 TOP(简报已列、0009 未收录的)
 JOURNALS_TIER2 = [
    ("0141-0296", "Engineering Structures"),
    ("1369-7021", "Materials Today"),
 ]
 NOTE_TIER2 = "建材TOP顶刊"
 # 统一检索关键词(简报第三节,英文搜索词 + 中文名)
 SEARCHES = [
    ("low carbon cement", "低碳水泥"),
    ("energy storage building material", "储能建筑材料"),
    ("concrete carbonation", "混凝土碳化机理"),
    ("geopolymer", "工业固废基地聚物"),
    ("supplementary cementitious material", "固废基胶凝材料"),
    ("carbon capture cement", "建材碳捕集"),
 ]
 NOTE_SEARCH = "低碳建材前沿"
 def seed(apps, schema_editor):
    PaperMonitor = apps.get_model("resm", "PaperMonitor")
    for issn, name in JOURNALS_TIER1:
        PaperMonitor.objects.get_or_create(
            type="journal", value=issn,
            defaults={"id": idWorker.get_id(), "name": name, "note": NOTE_TIER1,
                      "is_active": True, "days": 7},
        )
    for issn, name in JOURNALS_TIER2:
        PaperMonitor.objects.get_or_create(
            type="journal", value=issn,
            defaults={"id": idWorker.get_id(), "name": name, "note": NOTE_TIER2,
                      "is_active": True, "days": 7},
        )
    for term, name in SEARCHES:
        PaperMonitor.objects.get_or_create(
            type="search", value=term,
            defaults={"id": idWorker.get_id(), "name": name, "note": NOTE_SEARCH,
                      "is_active": True, "days": 7},
        )
 def unseed(apps, schema_editor):
    PaperMonitor = apps.get_model("resm", "PaperMonitor")
    journals = [i for i, _ in JOURNALS_TIER1] + [i for i, _ in JOURNALS_TIER2]
    PaperMonitor.objects.filter(type="journal", value__in=journals).delete()
    PaperMonitor.objects.filter(type="search", value__in=[t for t, _ in SEARCHES]).delete()
 class Migration(migrations.Migration):
    dependencies = [
        ("resm", "0010_seed_ensure_fetch_running"),
    ]
    operations = [
        migrations.RunPython(seed, unseed),
    ]
--- a/apps/resm/migrations/0012_paper_title_quality.py
+++ b/apps/resm/migrations/0012_paper_title_quality.py
@ -1,75 +0,0 @@
 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),
        ),
    ]
--- a/apps/resm/models.py
+++ b/apps/resm/models.py
@ -5,30 +5,12 @@ 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)
    # ===== 作者（最小可用集）=====
@ -45,7 +27,7 @@ class Paper(BaseModel):
    has_fulltext = models.BooleanField(default=False, db_index=True)
    has_fulltext_xml = models.BooleanField(default=False, db_index=True)
    has_fulltext_pdf = models.BooleanField(default=False, db_index=True)
-    fetch_status = models.CharField(max_length=20, null=True, blank=True) # downloading(elsevier/openalex链) / downloading_pdf(download_pdf链)
+    fetch_status = models.CharField(max_length=20, null=True, blank=True) # downloading
    fail_reason = models.TextField(null=True, blank=True)
    source = models.CharField(
--- a/apps/resm/pdf_utils.py
+++ b/apps/resm/pdf_utils.py
@ -1,94 +0,0 @@
 """PDF 解析/分类工具(纯 stdlib + pypdf, 不依赖 Django)。
 独立成模块, 以便 ProcessPoolExecutor 的子进程能安全导入(fork/spawn 均可),
 不会牵连 Django 模型与配置。tasks.py 从这里复用这些函数。
 """
 import os
 import re
 def _pdf_page_count(content: bytes):
    """返回 PDF 页数; 无法确定时返回 None。
    优先用 pypdf 精确解析; 未安装或解析异常时退化为字节扫描
    (对未压缩对象树有效, Elsevier 的摘要预览页正属此类)。"""
    try:
        from io import BytesIO
        import logging
        # 坏 PDF 会让 pypdf 刷大量恢复日志, 这里只关心页数, 静音其 logger
        logging.getLogger("pypdf").setLevel(logging.CRITICAL)
        from pypdf import PdfReader
        return len(PdfReader(BytesIO(content), strict=False).pages)
    except ImportError:
        pass
    except Exception:
        return None
    try:
        counts = [int(m) for m in re.findall(rb"/Count\s+(\d+)", content)]
        if counts:
            return max(counts)
        n = len(re.findall(rb"/Type\s*/Page(?![sR])", content))
        if n:
            return n
    except Exception:
        pass
    return None
 def _is_elsevier_preview_pdf(content: bytes) -> bool:
    """判断 Elsevier 返回的 PDF 是否为"摘要预览页"。
    Elsevier Article API 对未授权 / in-press 文章, application/pdf 端点会返回
    仅含摘要的 1 页预览 PDF(魔数仍是 %PDF、体积也不小), 全文 XML 却可能正常。
    判据: 能确定页数且 <= 1 页。无法确定页数时返回 False(从宽, 不误杀真全文)。"""
    pages = _pdf_page_count(content)
    return pages is not None and pages <= 1
 def _inspect_pdf(content: bytes):
    """对历史落库的 PDF 文件分类, 返回 (kind, pages)。
    kind:
      'broken'  - 非 PDF(魔数不符)或 pypdf 解析直接失败 -> 可安全删除重抓
      'preview' - 1 页摘要预览页
      'ok'      - 多页, 视为真全文, 不处理
      'unknown' - 魔数正常但页数判不出(通常因未装 pypdf) -> 不处理, 绝不当坏文件
    pages: 页数; 无法确定为 None。"""
    if not content or b"%PDF" not in content[:1024]:
        return "broken", 0
    try:
        from io import BytesIO
        import logging
        logging.getLogger("pypdf").setLevel(logging.CRITICAL)
        from pypdf import PdfReader
    except ImportError:
        # 没装 pypdf: 只能靠字节扫描, 判不出就 unknown(从宽, 不误判为坏)
        pages = _pdf_page_count(content)
        if pages is None:
            return "unknown", None
        return ("preview" if pages <= 1 else "ok"), pages
    try:
        pages = len(PdfReader(BytesIO(content), strict=False).pages)
    except Exception:
        return "broken", None
    if pages <= 0:
        return "broken", pages
    return ("preview" if pages == 1 else "ok"), pages
 def classify_pdf_file(path: str):
    """并发 worker 入口: 读取并分类单个 PDF 文件路径。
    返回 (path, kind, pages)。除 _inspect_pdf 的四种 kind 外, 另有 IO 结果:
      'missing'    - 文件不存在
      'unreadable' - 打开失败(权限等)
    设计为纯函数(仅 stdlib + pypdf), 可被进程池安全 pickle / 导入。"""
    try:
        if not os.path.exists(path):
            return path, "missing", None
        with open(path, "rb") as f:
            content = f.read()
    except OSError:
        return path, "unreadable", None
    kind, pages = _inspect_pdf(content)
    return path, kind, pages
--- a/apps/resm/tasks.py
+++ b/apps/resm/tasks.py
@ -11,9 +11,8 @@ from lxml import etree
 from celery import current_app
 from datetime import datetime, timedelta
 import random
-from .pdf_utils import _is_elsevier_preview_pdf
+import re
-from .title_utils import assess_title, extract_title_from_xml, is_better_title
+from .d_oaurl import download_from_url_playwright
 from uuid import uuid4
 import asyncio
 import sys
 import os
@ -59,12 +58,9 @@ def run_async(coro):
 OPENALEX_SELECT_FIELDS = [
    "id", "doi", "title", "publication_date",
    "open_access", "authorships", "primary_location", "publication_year",
-    "display_name", "content_urls", "type"
+    "display_name", "content_urls"
 ]
 # OpenAlex 抓取的 work 类型,"|" 表示 OR;记录的真实 type 会写回 Paper.type
 OPENALEX_WORK_TYPES = "article|book|book-chapter"
 def _apply_keyword_search(pager, keywords: str, search: str):
    """把 keywords / search 过滤条件套到 pyalex 的 pager 上。
@ -91,13 +87,10 @@ def _build_paper_from_record(record, keywords: str, search: str) -> Paper:
    paper.o_keywords = keywords
    paper.o_search = search
    paper.source = "openalex"
-    paper.type = (record.get("type") or "article")[:20]
+    paper.type = "article"
    paper.openalex_id = record["id"].split("/")[-1]
    paper.doi = record["doi"].replace("https://doi.org/", "")
-    paper.title = record.get("title") or record["display_name"]
+    paper.title = 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"]:
@ -153,16 +146,14 @@ def _crawl_openalex_query(base_pager, keywords: str, search: str, cache_key: str
@shared_task(base=CustomTask)
-def get_paper_meta_from_openalex(publication_year:int, keywords:str="", search:str="", end_year:int=None,
+def get_paper_meta_from_openalex(publication_year:int, keywords:str="", search:str="", end_year:int=None):
                                 types:str=OPENALEX_WORK_TYPES):
    if not (keywords or search):
        raise Exception("keywords or search must be provided")
-    # types 进 key:OpenAlex 游标只对完全相同的查询有效,换类型过滤必须换 checkpoint
+    cache_key = f"openalex_cursor_{publication_year}_{keywords}{search}"
    cache_key = f"openalex_cursor_{publication_year}_{keywords}{search}_{types}"
    base = Works().filter(
        publication_year=publication_year,
        has_doi=True,
-        type=types
+        type="article"
    )
    base = _apply_keyword_search(base, keywords, search)
    _crawl_openalex_query(base, keywords, search, cache_key=cache_key,
@ -177,8 +168,7 @@ def get_paper_meta_from_openalex(publication_year:int, keywords:str="", search:s
                    "publication_year": publication_year + 1,
                    "keywords": keywords,
                    "search": search,
-                    "end_year": end_year,
+                    "end_year": end_year
                    "types": types
                },
                countdown=5
            )
@ -196,7 +186,7 @@ def _fetch_openalex_published_since(keywords: str, search: str, from_publication
    """
    base = Works().filter(
        has_doi=True,
-        type=OPENALEX_WORK_TYPES,
+        type="article",
        from_publication_date=from_publication_date,
    )
    base = _apply_keyword_search(base, keywords, search)
@ -230,49 +220,12 @@ 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"
@shared_task(base=CustomTask)
 def backfill_paper_meta_from_openalex(from_publication_date: str, to_publication_date: str = None,
-                                      combo_index: int = 0, combos=None,
+                                      combo_index: int = 0, combos=None):
                                      types: str = OPENALEX_WORK_TYPES):
    """按发表日期一次性回补论文索引,支持断点续传(应对 OpenAlex 配额限制)。
    本任务只负责策略:抓哪些查询组合、按什么发表日期区间、怎么 chain。
@ -299,7 +252,7 @@ def backfill_paper_meta_from_openalex(from_publication_date: str, to_publication
        })]
    def ckey(kw, search):
-        return f"backfill_cursor_{from_publication_date}|{to_publication_date}|{kw}|{search}|{types}"
+        return f"backfill_cursor_{from_publication_date}|{to_publication_date}|{kw}|{search}"
    # 跳过已完成的组合
    while combo_index < len(combos):
@ -314,7 +267,7 @@ def backfill_paper_meta_from_openalex(from_publication_date: str, to_publication
    kw, search = combos[combo_index]
    cursor_key = ckey(kw, search)
    base = Works().filter(
-        has_doi=True, type=types, from_publication_date=from_publication_date,
+        has_doi=True, type="article", from_publication_date=from_publication_date,
    )
    if to_publication_date:
        base = base.filter(to_publication_date=to_publication_date)
@ -341,7 +294,6 @@ def backfill_paper_meta_from_openalex(from_publication_date: str, to_publication
            "to_publication_date": to_publication_date,
            "combo_index": combo_index + 1,
            "combos": combos,
            "types": types,
        },
        countdown=3,
    )
@ -365,7 +317,7 @@ def monitor_papers(monitor_id: str = None):
    results = []
    for m in qs:
        from_pub = (timezone.now() - timedelta(days=m.days or 30)).date().isoformat()
-        base = Works().filter(has_doi=True, type=OPENALEX_WORK_TYPES, from_publication_date=from_pub)
+        base = Works().filter(has_doi=True, type="article", from_publication_date=from_pub)
        kw, search = "", ""
        if m.type == PaperMonitor.TYPE_JOURNAL:
            base = base.filter(primary_location={"source": {"issn": m.value}})
@ -414,9 +366,6 @@ 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")
@ -535,24 +484,6 @@ def touch_alive(def_name: str):
    """标记该自触发链仍在运行。"""
    cache.set(def_name + ":alive", 1, timeout=ALIVE_TTL)
 def claim_chain(def_name: str, chain_id: str):
    """自触发链去重: 同名任务同时只允许一条链存活。
    cache 里记录当前唯一合法链的 chain_id。无 chain_id 的调用(beat 点火/手动 .delay/
    历史遗留任务)视为新链接管, 旧链下一轮发现 chain_id 不匹配即自杀, 收敛到单链。
    返回本任务应携带的 chain_id; 返回 None 表示本任务是重复链, 应立即退出。
    """
    key = def_name + ":chain"
    current = cache.get(key)
    if chain_id and current and chain_id != current:
        return None  # 已有更新的链在跑, 本链退出
    if not chain_id:
        chain_id = uuid4().hex  # 新点火: 接管成为唯一链
        cache.set(key, chain_id, timeout=None)
    elif not current:
        cache.set(key, chain_id, timeout=None)  # cache 丢失(如 redis 重启)后自愈
    return chain_id
 def is_alive(def_name: str):
    return cache.get(def_name + ":alive") is not None
@ -567,13 +498,10 @@ def ensure_fetch_running():
    return f"ensure_fetch_running started: {started}"
@shared_task(base=CustomTask)
-def get_pdf_from_openalex(number_of_task: int =10, chain_id: str = None):
+def get_pdf_from_openalex(number_of_task: int =10):
    def_name = get_pdf_from_openalex.name
    if not show_task_run(def_name):
        return "stoped"
    chain_id = claim_chain(def_name, chain_id)
    if chain_id is None:
        return "duplicate chain, exit"
    touch_alive(def_name)
    # 限流退避中: 不打 API, 慢节奏自重发只为维持 alive, 等 exceed 标记自然过期。
@ -581,14 +509,14 @@ def get_pdf_from_openalex(number_of_task: int =10, chain_id: str = None):
    if cache.get("openalex_api_exceed"):
        current_app.send_task(
            "apps.resm.tasks.get_pdf_from_openalex",
-            kwargs={"number_of_task": number_of_task, "chain_id": chain_id},
+            kwargs={"number_of_task": number_of_task},
            countdown=60,
        )
        return "openalex_api_exceed, backing off"
    count = 0
    qs = Paper.objects.filter(is_oa=True, has_fulltext=False).exclude(
-        fetch_status__startswith="downloading").exclude(fail_reason__contains="openalex_pdf_not_found")[:number_of_task]
+        fetch_status="downloading").exclude(fail_reason__contains="openalex_pdf_not_found")[:number_of_task]
    if not qs.exists():
        return "done"  # 不自重发, 交给 beat 轮询拉起
    msg = ""
@ -609,7 +537,6 @@ def get_pdf_from_openalex(number_of_task: int =10, chain_id: str = None):
            "apps.resm.tasks.get_pdf_from_openalex",
            kwargs={
                "number_of_task": number_of_task,
                "chain_id": chain_id,
            },
            countdown=countdown,
        )
@ -668,27 +595,46 @@ def _elsevier_fetch_xml(req, paper):
        paper.has_fulltext_xml = True
    paper.save_file_xml(xml_str)
-    update_fields = ["has_abstract", "has_abstract_xml",
+    paper.save(update_fields=["has_abstract", "has_abstract_xml",
-                     "has_fulltext", "has_fulltext_xml", "update_time"]
+                              "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
 def _pdf_page_count(content: bytes):
    """返回 PDF 页数; 无法确定时返回 None。
    优先用 pypdf 精确解析; 未安装或解析异常时退化为字节扫描
    (对未压缩对象树有效, Elsevier 的摘要预览页正属此类)。"""
    try:
        from io import BytesIO
        from pypdf import PdfReader
        return len(PdfReader(BytesIO(content), strict=False).pages)
    except ImportError:
        pass
    except Exception:
        return None
    try:
        counts = [int(m) for m in re.findall(rb"/Count\s+(\d+)", content)]
        if counts:
            return max(counts)
        n = len(re.findall(rb"/Type\s*/Page(?![sR])", content))
        if n:
            return n
    except Exception:
        pass
    return None
 def _is_elsevier_preview_pdf(content: bytes) -> bool:
    """判断 Elsevier 返回的 PDF 是否为"摘要预览页"。
    Elsevier Article API 对未授权 / in-press 文章, application/pdf 端点会返回
    仅含摘要的 1 页预览 PDF(魔数仍是 %PDF、体积也不小), 全文 XML 却可能正常。
    判据: 能确定页数且 <= 1 页。无法确定页数时返回 False(从宽, 不误杀真全文)。"""
    pages = _pdf_page_count(content)
    return pages is not None and pages <= 1
 def _elsevier_fetch_pdf(req, paper):
    """同一 DOI 取 PDF, 成功落库返回 True。"""
    try:
@ -718,16 +664,13 @@ def _elsevier_fetch_pdf(req, paper):
@shared_task(base=CustomTask)
 def get_abstract_from_elsevier(number_of_task:int = 20, exclude_failed:bool=True,
-                               pdf_number_of_task:int = 20, chain_id: str = None):
+                               pdf_number_of_task:int = 20):
    """Elsevier 单端点合并任务: 同一 DOI 先取 XML(摘要/全文标记), 若有全文则内联
    再取一次 PDF; 并补抓历史上已有全文标记但缺 PDF 的论文。原 get_pdf_from_elsevier 已并入。
    number_of_task: 阶段1(摘要+内联 PDF)每轮上限; pdf_number_of_task: 阶段2(存量补 PDF)每轮上限。"""
    def_name = get_abstract_from_elsevier.name
    if not show_task_run(def_name):
        return "stoped"
    chain_id = claim_chain(def_name, chain_id)
    if chain_id is None:
        return "duplicate chain, exit"
    touch_alive(def_name)
    # 待抓摘要(并顺带取 PDF)
@ -736,13 +679,13 @@ def get_abstract_from_elsevier(number_of_task:int = 20, exclude_failed:bool=True
        qs = qs.filter(fail_reason=None)
    else:
        qs = qs.exclude(fail_reason__contains="elsevier_")
-    qs = qs.exclude(fetch_status__startswith="downloading"
+    qs = qs.exclude(fetch_status="downloading"
    ).filter(doi__startswith="10.1016").order_by("?")
    # 存量补 PDF: 已有全文标记但还没下到 PDF
    qs_pdf = Paper.objects.filter(
        has_fulltext=True, has_fulltext_pdf=False, has_abstract=True
-    ).exclude(fetch_status__startswith="downloading"
+    ).exclude(fetch_status="downloading"
    ).exclude(fail_reason__contains="elsevier_pdf_preview_only"
    ).filter(doi__startswith="10.1016")
@ -758,7 +701,7 @@ def get_abstract_from_elsevier(number_of_task:int = 20, exclude_failed:bool=True
        for paper in qs[:number_of_task]:
            if not show_task_run(def_name):
                break
-            if paper.fetch_status and paper.fetch_status.startswith("downloading"):
+            if paper.fetch_status == "downloading":
                continue
            paper.fetch(status="downloading")
            try:
@ -780,7 +723,7 @@ def get_abstract_from_elsevier(number_of_task:int = 20, exclude_failed:bool=True
            for paper in qs_pdf[:pdf_number_of_task]:
                if not show_task_run(def_name):
                    break
-                if paper.fetch_status and paper.fetch_status.startswith("downloading"):
+                if paper.fetch_status == "downloading":
                    continue
                paper.fetch(status="downloading")
                try:
@ -797,17 +740,14 @@ def get_abstract_from_elsevier(number_of_task:int = 20, exclude_failed:bool=True
                "number_of_task": number_of_task,
                "exclude_failed": exclude_failed,
                "pdf_number_of_task": pdf_number_of_task,
                "chain_id": chain_id,
            },
            countdown=countdown,
        )
    return f'{err_msg}, abs {count_abs}, fulltext {count_fulltext}, pdf {count_pdf}'
 def get_actual_running_count():
-    """获取本下载链路(download_pdf)实际在下载的任务数。
+    """获取实际在下载的任务数"""
-    只数 downloading_pdf, 不含 elsevier/openalex 抓取链的 downloading 粗锁,
+    return Paper.objects.filter(fetch_status='downloading').count()
    否则那两条链常驻的十几个锁会把本链路的并发闸门永久卡死。"""
    return Paper.objects.filter(fetch_status='downloading_pdf').count()
 def can_send_more(max_running):
    return get_actual_running_count() < max_running
@ -818,7 +758,7 @@ def send_download_fulltext_task(number_of_task=100):
    # 不再用 fail_reason=None —— 否则被 openalex 保活链失败标记蹭上 fail_reason 的论文会被
    # 永久遮蔽, 其 oa_url/elsevier/scihub 兜底路径永远不会被尝试。
    qs = Paper.objects.filter(has_fulltext=False, is_oa=True).exclude(
-        fetch_status__startswith='downloading'
+        fetch_status='downloading'
    ).exclude(fail_reason__contains="download_pdf_tried")
    if not qs.exists():
        return "done"
@ -844,8 +784,7 @@ def send_download_fulltext_task(number_of_task=100):
@shared_task(base=CustomTask)
 def release_working_paper(minutes=10):
-    # startswith: 同时释放 downloading(elsevier/openalex 链) 和 downloading_pdf(download_pdf 链)
+    qs = Paper.objects.filter(fetch_status="downloading")
    qs = Paper.objects.filter(fetch_status__startswith="downloading")
    count = 0
    for paper in qs:
        if paper.update_time < timezone.now() - timedelta(minutes=minutes):
@ -858,12 +797,11 @@ def download_pdf(paper_id):
    """
    下载单个论文的PDF
    """
    paper = Paper.objects.get(id=paper_id)
    if paper.fetch_status and paper.fetch_status.startswith("downloading"):
        # 已被任一链锁定: 直接返回, 不能走 finally 的 fetch_end, 否则会误清别的链持有的锁
        return
    try:
-        paper.fetch("downloading_pdf")
+        paper = Paper.objects.get(id=paper_id)
        if paper.fetch_status == "downloading":
            return
        paper.fetch("downloading")
        msg = "no_method_to_get_pdf"
        current_from = ""
        if paper.oa_url:
@ -891,7 +829,7 @@ def download_pdf(paper_id):
 def save_pdf_from_oa_url(paper: Paper):
-    from .d_oaurl import download_pdf_with_curl_cffi
+    from .d_oaurl import download_pdf_with_curl_cffi, download_from_url_playwright
    # 策略1: 直接请求
    try:
@ -924,10 +862,16 @@ def save_pdf_from_oa_url(paper: Paper):
        paper.save(update_fields=["has_fulltext", "has_fulltext_pdf", "update_time"])
        return "success"
-    # 内联 Playwright 回退已移除(在 worker 里起浏览器太重, 且会开有头窗口)。
+    # 策略3: Playwright（最终回退）
-    # 打 oa_url_need_play 标记, 交由独立脚本 scripts/get_pdf_by_playwright.py 兜底。
+    is_ok, err_msg = run_async(download_from_url_playwright(paper.oa_url, paper_path))
-    paper.save_fail_reason(f"oa_url_need_play: {err_msg}")
+    if is_ok:
-    return f"oa_url_need_play: {err_msg}"
+        paper.has_fulltext = True
        paper.has_fulltext_pdf = True
        paper.save(update_fields=["has_fulltext", "has_fulltext_pdf", "update_time"])
        return "success"
    paper.save_fail_reason(f"oa_url_all_methods_failed: {err_msg}")
    return f"oa_url_all_methods_failed: {err_msg}"
 def save_pdf_from_openalex(paper:Paper):
    if cache.get("openalex_api_exceed"):
@ -949,12 +893,10 @@ def save_pdf_from_openalex(paper:Paper):
            message = res.json().get("message", "")
        except ValueError:
            message = res.text
-        # 文案历史上出现过 "Insufficient credits" 和 "Insufficient budget"(每日额度,
+        if "Insufficient credits" in message:
        # UTC 午夜重置), 放宽到 "Insufficient" 统一匹配, 避免落进 2 分钟短退避分支空转
        if "Insufficient" in message:
            # 额度耗尽: 退避 1 小时
            cache.set("openalex_api_exceed", True, timeout=3600)
-            return f"openalex_pdf_error: {message[:100]}"
+            return "openalex_pdf_error: Insufficient credits"
        # 普通限流(请求过频): 短退避 2 分钟, 避免立刻重试再撞 429
        cache.set("openalex_api_exceed", True, timeout=120)
        return f"openalex_pdf_error: 429 {message[:100]}"
@ -989,10 +931,8 @@ def save_pdf_from_elsevier(paper:Paper):
        return f"elsevier_status_error: {res.status_code} {res.text}"
 def save_pdf_from_scihub(paper:Paper):
-    # worker 内只走 curl-cffi 轻量解析, 不起浏览器; 撞挑战页返回 scihub_need_play,
+    from .d_scihub import download_paper_by_doi
-    # 交由独立脚本用浏览器兜底(同 oa_url need_play 的分工)。
+    is_ok, err_msg = download_paper_by_doi(paper.doi, paper.init_paper_path("pdf"))
    from .d_scihub import download_by_doi_http
    is_ok, err_msg = download_by_doi_http(paper.doi, paper.init_paper_path("pdf"))
    if is_ok:
        paper.has_fulltext = True
        paper.has_fulltext_pdf = True
@ -1001,4 +941,4 @@ def save_pdf_from_scihub(paper:Paper):
    else:
        paper.save_fail_reason(err_msg)
        return err_msg
-# https://sci.bban.top/pdf/10.1016/j.conbuildmat.2020.121016.pdf?download=true
+# https://sci.bban.top/pdf/10.1016/j.conbuildmat.2020.121016.pdf?download=true
--- a/apps/resm/test_title_utils.py
+++ b/apps/resm/test_title_utils.py
@ -1,126 +0,0 @@
 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")
    def test_rejects_layout_filename_and_keeps_first_page_low_confidence(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="article-layout.indd")
        )
        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.source, "pdf_first_page")
        self.assertEqual(candidate.confidence, "low")
 if __name__ == "__main__":
    unittest.main()
--- a/apps/resm/title_utils.py
+++ b/apps/resm/title_utils.py
@ -1,274 +0,0 @@
 """论文标题质量检测与全文标题提取工具。
 本模块不依赖 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",
    "no job name",
 }
 _FILE_TITLE_RE = re.compile(r"(?:\.eps|\.indd|\.docx?|\.pdf|\.tex)$", re.I)
@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_usable_pdf_metadata_title(value: Optional[str]) -> bool:
    """PDF Title 属性经常是排版文件名，使用比普通标题更严格的门槛。"""
    title = clean_title(value)
    assessment = assess_title(title)
    if not is_usable_title(title):
        return False
    if assessment.plain_length < 15 or assessment.word_count < 3:
        return False
    if title.casefold() in _GENERIC_PDF_TITLES or _FILE_TITLE_RE.search(title):
        return False
    if re.fullmatch(r"(?:issn\s*)?[\dXx-]{7,}", title):
        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 _is_usable_pdf_metadata_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="low",
            evidence=f"pdf_first_page+{identity}" if identity else "pdf_first_page_only",
        )
    return None
--- a/apps/resm/views.py
+++ b/apps/resm/views.py
@ -1,21 +1,18 @@
 from django.shortcuts import get_object_or_404
 from django.http import FileResponse, Http404
 from rest_framework.response import Response
-from rest_framework.decorators import api_view, permission_classes, authentication_classes
+from rest_framework.decorators import api_view, permission_classes
-from rest_framework.permissions import IsAuthenticated
+from rest_framework.permissions import AllowAny
 from rest_framework.settings import api_settings
 from .models import Paper, PaperAbstract
 from .serializers import PaperListSerializer
 from .filters import PaperFilterSet
 from apps.utils.viewsets import CustomGenericViewSet, CustomListModelMixin
 from apps.utils.mixins import CustomRetrieveModelMixin
 from apps.utils.apikey_auth import ApiKeyAuthentication, HasApiKey
 import os
@api_view(['GET'])
-@authentication_classes([ApiKeyAuthentication] + api_settings.DEFAULT_AUTHENTICATION_CLASSES)
+@permission_classes([AllowAny])
@permission_classes([HasApiKey | IsAuthenticated])
 def paper_pdf_view(request, pk):
    paper = get_object_or_404(Paper, pk=pk)
    if not paper.has_fulltext_pdf:
@ -44,12 +41,11 @@ class PaperViewSet(CustomGenericViewSet, CustomListModelMixin, CustomRetrieveMod
    ordering = ["-publication_date", "-create_time"]
    def get_authenticators(self):
        authenticators = super().get_authenticators()
        if self.request.method == 'GET':
-            return [ApiKeyAuthentication()] + authenticators
+            return []
-        return authenticators
+        return super().get_authenticators()
    def get_permissions(self):
        if self.request.method == 'GET':
-            return [(HasApiKey | IsAuthenticated)()]
+            return [AllowAny()]
-        return super().get_permissions()
+        return super().get_permissions()
--- a/apps/system/admin.py
+++ b/apps/system/admin.py
@ -1,5 +1,5 @@
 from django.contrib import admin
-from .models import User, Dept, Role, Permission, DictType, Dictionary, File, ApiKey
+from .models import User, Dept, Role, Permission, DictType, Dictionary, File
 # Register your models here.
 admin.site.register(User)
 admin.site.register(Dept)
@ -8,12 +8,3 @@ admin.site.register(Permission)
 admin.site.register(DictType)
 admin.site.register(Dictionary)
 admin.site.register(File)
@admin.register(ApiKey)
 class ApiKeyAdmin(admin.ModelAdmin):
    list_display = ('name', 'key', 'is_active', 'expires_at', 'last_used_at', 'create_time')
    list_filter = ('is_active',)
    search_fields = ('name', 'key')
    readonly_fields = ('key', 'last_used_at')
    fields = ('name', 'key', 'is_active', 'expires_at', 'scopes', 'last_used_at')
--- a/apps/system/migrations/0009_apikey.py
+++ b/apps/system/migrations/0009_apikey.py
@ -1,37 +0,0 @@
 # Generated by Django 4.2.27 on 2026-07-06 02:30
 from django.conf import settings
 from django.db import migrations, models
 import django.db.models.deletion
 import django.utils.timezone
 class Migration(migrations.Migration):
    dependencies = [
        ('system', '0008_merge_20260116_1509'),
    ]
    operations = [
        migrations.CreateModel(
            name='ApiKey',
            fields=[
                ('id', models.CharField(editable=False, help_text='主键ID', max_length=20, primary_key=True, serialize=False, verbose_name='主键ID')),
                ('create_time', models.DateTimeField(default=django.utils.timezone.now, help_text='创建时间', verbose_name='创建时间')),
                ('update_time', models.DateTimeField(auto_now=True, help_text='修改时间', verbose_name='修改时间')),
                ('is_deleted', models.BooleanField(default=False, help_text='删除标记', verbose_name='删除标记')),
                ('name', models.CharField(help_text='调用方标识', max_length=100, verbose_name='名称')),
                ('key', models.CharField(db_index=True, editable=False, max_length=64, unique=True, verbose_name='密钥')),
                ('is_active', models.BooleanField(default=True, verbose_name='启用')),
                ('expires_at', models.DateTimeField(blank=True, help_text='留空则永不过期', null=True, verbose_name='过期时间')),
                ('scopes', models.JSONField(blank=True, default=list, help_text='预留, 如 ["resm.paper"]', verbose_name='权限范围')),
                ('last_used_at', models.DateTimeField(blank=True, editable=False, null=True, verbose_name='最后使用')),
                ('create_by', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='%(class)s_create_by', to=settings.AUTH_USER_MODEL, verbose_name='创建人')),
                ('update_by', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='%(class)s_update_by', to=settings.AUTH_USER_MODEL, verbose_name='最后编辑人')),
            ],
            options={
                'verbose_name': 'API密钥',
                'verbose_name_plural': 'API密钥',
            },
        ),
    ]
--- a/apps/system/models.py
+++ b/apps/system/models.py
@ -263,33 +263,3 @@ class MySchedule(CommonAModel):
        IntervalSchedule, on_delete=models.PROTECT, null=True, blank=True)
    crontab = models.ForeignKey(
        CrontabSchedule, on_delete=models.PROTECT, null=True, blank=True)
 class ApiKey(CommonAModel):
    """
    外部程序调用的API Key
    key保存时自动生成, 明文存库, admin中可直接查看
    """
    name = models.CharField('名称', max_length=100, help_text='调用方标识')
    key = models.CharField('密钥', max_length=64, unique=True, db_index=True, editable=False)
    is_active = models.BooleanField('启用', default=True)
    expires_at = models.DateTimeField('过期时间', null=True, blank=True, help_text='留空则永不过期')
    scopes = models.JSONField('权限范围', default=list, blank=True, help_text='预留, 如 ["resm.paper"]')
    last_used_at = models.DateTimeField('最后使用', null=True, blank=True, editable=False)
    class Meta:
        verbose_name = 'API密钥'
        verbose_name_plural = verbose_name
    def __str__(self):
        return self.name
    @staticmethod
    def generate_key() -> str:
        import secrets
        return 'pk_' + secrets.token_urlsafe(36)
    def save(self, *args, **kwargs):
        if not self.key:
            self.key = self.generate_key()
        return super().save(*args, **kwargs)
--- a/apps/utils/apikey_auth.py
+++ b/apps/utils/apikey_auth.py
@ -1,44 +0,0 @@
 from datetime import timedelta
 from django.contrib.auth.models import AnonymousUser
 from django.utils import timezone
 from rest_framework import exceptions
 from rest_framework.authentication import BaseAuthentication
 from rest_framework.permissions import BasePermission
 from apps.system.models import ApiKey
 class ApiKeyAuthentication(BaseAuthentication):
    """
    API Key认证: 取请求头 X-API-Key 或查询参数 api_key
    通过后 request.auth 为 ApiKey 实例, request.user 为匿名用户
    未携带Key时返回None, 交给后续认证类处理
    """
    def authenticate(self, request):
        raw_key = request.META.get('HTTP_X_API_KEY') or request.GET.get('api_key')
        if not raw_key:
            return None
        try:
            apikey = ApiKey.objects.get(key=raw_key)
        except ApiKey.DoesNotExist:
            raise exceptions.AuthenticationFailed('无效的API Key')
        if not apikey.is_active:
            raise exceptions.AuthenticationFailed('API Key已禁用')
        if apikey.expires_at and apikey.expires_at <= timezone.now():
            raise exceptions.AuthenticationFailed('API Key已过期')
        now = timezone.now()
        # last_used_at节流更新, 避免每次请求都写库
        if apikey.last_used_at is None or now - apikey.last_used_at > timedelta(minutes=5):
            ApiKey.objects.filter(pk=apikey.pk).update(last_used_at=now)
        return (AnonymousUser(), apikey)
 class HasApiKey(BasePermission):
    """
    持有效API Key即放行
    """
    def has_permission(self, request, view):
        return isinstance(request.auth, ApiKey)
--- a/docs/全球材料前沿动态简报(1).docx
+++ b/docs/全球材料前沿动态简报(1).docx
--- a/scripts/sd_download.py
+++ b/scripts/sd_download.py
@ -1,311 +0,0 @@
 #!/usr/bin/env python
 """独立脚本: 从 ScienceDirect 网页下载排版 PDF。
 与 apps.resm 解耦, 独立运行。核心难点是 Cloudflare 人机校验: Playwright 自建
 浏览器带自动化指纹会被 Turnstile 识破而死循环("are you a robot"), 因此推荐
 连接你手动启动的真实 Chrome(由真人过一次验证), 脚本只负责驱动它下载。
 前提: 运行方 IP 在机构订阅网段(ScienceDirect 按 IP 授权)。
 依赖: playwright, requests, lxml, pypdf(可选, 用于精确判页数)。
 凭证: 默认从项目 config/conf.py 读取 ELSEVIER_API_KEY / ELSEVIER_INST_TOKEN,
      也可用 --pii 直接给 PII 跳过取号。
 用法(推荐 CDP 模式):
  1) 单独起一个带调试端口的 Chrome(独立档案, 不影响日常浏览器):
     Windows:
       & "C:\\Program Files\\Google\\Chrome\\Application\\chrome.exe" \
         --remote-debugging-port=9222 --user-data-dir="D:\\chrome-sd-profile"
     Linux:
       google-chrome --remote-debugging-port=9222 --user-data-dir=/tmp/sd-profile
  2) 在该 Chrome 里手动打开任一 ScienceDirect 文章, 亲手过掉 Cloudflare 验证。
  3) 运行本脚本(可一次多篇):
     python scripts/sd_download.py 10.1016/j.conbuildmat.2026.146897 \
         --cdp http://localhost:9222 --out ./sd_pdfs
  不加 --cdp 时脚本自行启动浏览器(大概率被 Cloudflare 拦, 仅调试用)。
 提示: 批量爬 ScienceDirect 违反 Elsevier 条款且可能导致机构 IP 被封, 仅供少量补抓。
 """
 import argparse
 import asyncio
 import os
 import re
 import sys
 from io import BytesIO
 # 项目根入 sys.path, 以便读取 config/conf.py 的凭证
 ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 if ROOT not in sys.path:
    sys.path.insert(0, ROOT)
 _UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
       "(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36")
 _STEALTH_ARGS = [
    "--disable-blink-features=AutomationControlled",
    "--no-first-run", "--no-default-browser-check",
    "--disable-infobars", "--disable-extensions", "--disable-notifications",
 ]
 _CHALLENGE_KW = ("just a moment", "moment", "checking your browser",
                 "attention required")
 # ------------------------------ PII / PDF 工具 ------------------------------
 def get_creds():
    try:
        from config.conf import ELSEVIER_API_KEY, ELSEVIER_INST_TOKEN
        return ELSEVIER_API_KEY, ELSEVIER_INST_TOKEN
    except Exception as e:
        print(f"[warn] 读取 config/conf.py 凭证失败: {e!r}")
        return None, None
 def fetch_pii(doi):
    """调 Elsevier API(text/xml) 取归一化 PII; 失败返回 None。"""
    import requests
    from lxml import etree
    key, token = get_creds()
    if not key:
        return None
    headers = {"X-ELS-APIKey": key}
    if token:
        headers["X-ELS-Insttoken"] = token
    try:
        r = requests.get(f"https://api.elsevier.com/content/article/doi/{doi}",
                         params={"httpAccept": "text/xml"}, headers=headers,
                         timeout=(3, 30))
    except requests.RequestException as e:
        print(f"[warn] 取 PII 请求失败: {e!r}")
        return None
    if r.status_code != 200:
        print(f"[warn] 取 PII 非 200: {r.status_code}")
        return None
    try:
        root = etree.fromstring(r.content)
    except Exception:
        return None
    nodes = root.xpath("//*[local-name()='pii']/text()")
    if not nodes:
        return None
    return re.sub(r"[^A-Za-z0-9]", "", nodes[0])
 def pdf_page_count(content: bytes):
    """返回页数, 判不出返回 None。优先 pypdf, 退化字节扫描。"""
    try:
        import logging
        logging.getLogger("pypdf").setLevel(logging.CRITICAL)
        from pypdf import PdfReader
        return len(PdfReader(BytesIO(content), strict=False).pages)
    except ImportError:
        pass
    except Exception:
        return None
    try:
        counts = [int(m) for m in re.findall(rb"/Count\s+(\d+)", content)]
        if counts:
            return max(counts)
        n = len(re.findall(rb"/Type\s*/Page(?![sR])", content))
        if n:
            return n
    except Exception:
        pass
    return None
 def classify(content: bytes):
    """(kind, pages): broken / preview(1页) / ok(多页) / unknown。"""
    if not content or b"%PDF" not in content[:1024]:
        return "broken", 0
    pages = pdf_page_count(content)
    if pages is None:
        return "unknown", None
    if pages <= 0:
        return "broken", pages
    return ("preview" if pages == 1 else "ok"), pages
 # ------------------------------ 浏览器下载 ------------------------------
 async def _wait_challenge_cleared(page, rounds=30, interval=2000):
    for _ in range(rounds):
        try:
            await page.wait_for_timeout(interval)
        except Exception:
            pass
        try:
            title = (await page.title()) or ""
        except Exception:
            continue  # 跳转中, 下一轮再看
        if title and not any(k in title.lower() for k in _CHALLENGE_KW):
            return True
    return False
 async def _find_pdfft_href(page):
    try:
        return await page.evaluate(
            "() => { const a=document.querySelector('a[href*=\"pdfft\"]');"
            " return a ? a.href : null; }")
    except Exception:
        return None
 async def _grab_pdf(page, url, timeout):
    # 优先 download 事件
    try:
        async with page.expect_download(timeout=min(timeout, 45000)) as dl:
            try:
                await page.goto(url, timeout=timeout)
            except Exception:
                pass
        download = await dl.value
        path = await download.path()
        if path:
            with open(path, "rb") as f:
                data = f.read()
            if data:
                return data
    except Exception:
        pass
    # 退化: 拦截内联 application/pdf 响应
    captured = {}
    async def on_resp(resp):
        try:
            if "application/pdf" in resp.headers.get("content-type", ""):
                captured["b"] = await resp.body()
        except Exception:
            pass
    page.on("response", on_resp)
    try:
        try:
            await page.goto(url.replace("?download=true", ""), timeout=timeout)
        except Exception:
            pass
        for _ in range(12):
            if captured.get("b"):
                break
            await page.wait_for_timeout(1000)
    finally:
        try:
            page.remove_listener("response", on_resp)
        except Exception:
            pass
    return captured.get("b")
 async def download_one(pii, save_path, cdp_url=None, headless=False,
                       timeout=60000):
    """返回 (ok, msg)。"""
    from playwright.async_api import async_playwright
    article = f"https://www.sciencedirect.com/science/article/pii/{pii}"
    pdf_url = f"{article}/pdfft?download=true"
    async with async_playwright() as p:
        connected = bool(cdp_url)
        if connected:
            browser = await p.chromium.connect_over_cdp(cdp_url)
        else:
            try:
                browser = await p.chromium.launch(headless=headless,
                                                   channel="chrome",
                                                   args=_STEALTH_ARGS)
            except Exception:
                browser = await p.chromium.launch(headless=headless,
                                                   args=_STEALTH_ARGS)
        page = None
        try:
            if connected:
                ctx = browser.contexts[0] if browser.contexts else await browser.new_context()
            else:
                ctx = await browser.new_context(
                    viewport={"width": 1920, "height": 1080},
                    user_agent=_UA, locale="en-US", accept_downloads=True)
            page = await ctx.new_page()
            page.set_default_timeout(timeout)
            if not connected:
                await page.add_init_script(
                    "Object.defineProperty(navigator,'webdriver',{get:()=>false});")
            try:
                await page.goto(article, wait_until="domcontentloaded",
                                timeout=40000)
            except Exception:
                pass
            rounds = 60 if connected else 20
            if not await _wait_challenge_cleared(page, rounds=rounds):
                return False, "cloudflare_not_cleared"
            # reload 拿干净文章页
            try:
                await page.goto(article, wait_until="domcontentloaded",
                                timeout=40000)
                await page.wait_for_timeout(4000)
            except Exception:
                pass
            href = await _find_pdfft_href(page) or pdf_url
            body = await _grab_pdf(page, href, timeout)
            if not body:
                return False, "no_pdf_captured"
            kind, pages = classify(body)
            if kind != "ok":
                head = body[:160].decode("utf-8", "replace").replace("\n", " ")
                return False, f"not_fulltext kind={kind} pages={pages} len={len(body)} head={head!r}"
            os.makedirs(os.path.dirname(os.path.abspath(save_path)), exist_ok=True)
            with open(save_path, "wb") as f:
                f.write(body)
            return True, f"ok pages={pages} len={len(body)}"
        finally:
            try:
                if connected and page:
                    await page.close()
                await browser.close()
            except Exception:
                pass
 # ------------------------------ CLI ------------------------------
 async def _run(args):
    os.makedirs(args.out, exist_ok=True)
    ok_n = 0
    for doi in args.doi:
        doi = doi.strip()
        pii = args.pii or fetch_pii(doi)
        if not pii:
            print(f"[FAIL] {doi}: 取不到 PII")
            continue
        save = os.path.join(args.out, doi.replace("/", "_") + ".pdf")
        print(f"[..] {doi}  PII={pii}  -> {save}")
        ok, msg = await download_one(pii, save, cdp_url=args.cdp or None,
                                     headless=args.headless, timeout=args.timeout * 1000)
        if ok:
            ok_n += 1
            print(f"[OK] {doi}: {msg}")
        else:
            print(f"[FAIL] {doi}: {msg}")
    print(f"完成: {ok_n}/{len(args.doi)} 成功")
 def main():
    ap = argparse.ArgumentParser(description="从 ScienceDirect 网页下载 PDF(独立脚本)")
    ap.add_argument("doi", nargs="+", help="一个或多个 DOI")
    ap.add_argument("--cdp", default="",
                    help="连接手动启动的 Chrome, 如 http://localhost:9222(推荐)")
    ap.add_argument("--out", default="./sd_pdfs", help="PDF 输出目录")
    ap.add_argument("--pii", default="", help="直接指定 PII(仅单篇时用, 跳过 API 取号)")
    ap.add_argument("--headless", action="store_true", help="无头(非 CDP 模式, 调试用)")
    ap.add_argument("--timeout", type=int, default=60, help="单步超时(秒)")
    args = ap.parse_args()
    if sys.platform == "win32":
        asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
    asyncio.run(_run(args))
 if __name__ == "__main__":
    main()