.gitignore

@@ -26,4 +26,3 @@ sh/*
 temp/*
 nohup.out
 scripts/*
-!scripts/sd_download.py

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>/`
 

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:请求日志
 

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]:
     """

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}"
-        ))

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 错误页 /
-  被截断的垃圾当 PDF 存了。旧抓取逻辑只校验魔数 + 体积, 都会误标 has_fulltext_pdf=True。
+  摘要的 1 页预览 PDF(魔数仍是 %PDF、体积也不小), 而全文 XML 却能正常拿到。旧抓取
+  逻辑只校验魔数 + 体积, 误将预览页落库并置 has_fulltext_pdf=True。
 
-本命令核对本地 PDF, 分两类处理:
-  - 预览页(1 页):   has_fulltext_pdf 置回 False; 文件**仅在 --delete-file 时**删除。
-  - 损坏文件(非 PDF / pypdf 解析失败): has_fulltext_pdf 置回 False; 文件**总是删除**
-    (dry-run 除外), 因为它根本不是有效全文, 留着无用且会污染下游解析。
-  两类在缺少 XML 全文(has_fulltext_xml=False)时, 一并把 has_fulltext 回退 False,
-  让其重新进入下载链路去找真正的全文; 并追加 fail_reason 标记供抓取任务排除。
+本命令重新核对本地 PDF 的页数, 对 <= 1 页者:
+  - has_fulltext_pdf 置回 False
+  - 若该论文有 XML 全文(has_fulltext_xml=True), 保留 has_fulltext=True;
+    否则(此前只有这张假预览页冒充全文)一并把 has_fulltext 回退为 False,
+    让它能重新进入下载链路去找真正的全文。
+  - 追加 fail_reason 'elsevier_pdf_preview_only' (供 Elsevier 补抓队列排除, 避免无限重试)
+  - 可选: 删除本地预览 PDF 文件 (--delete-file)
 
-性能:
-  读文件 + pypdf 解析是 CPU/IO 密集, 用 ProcessPoolExecutor 并行(--workers, 默认 CPU 核数);
-  数据库写入留在主进程串行(坏文件仅占少数, 非瓶颈, 也避免子进程共享 DB 连接)。
-
-安全前提:
-  "损坏"只在铁证下判定 —— 文件不以 %PDF 开头, 或已装 pypdf 且解析直接失败。
-  若未装 pypdf 且魔数正常但页数判不出, 归为 unknown, **不处理、绝不删除**。
-  强烈建议先 `pip install pypdf` 再跑, 否则只能处理魔数明显不符的坏文件。
+文件读取依赖本地存在 PDF (在跑抓取的服务器上执行)。建议先 --dry-run 看统计。
 
 用法:
   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 --workers 16         # 指定并发进程数
+  python manage.py fix_preview_pdf --delete-file
 """
 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
-
-
-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
+from apps.resm.tasks import _pdf_page_count
 
 
 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="同时删除预览页文件(坏文件无论该开关都会删)")
-        parser.add_argument("--workers", type=int, default=0,
-                            help="并发进程数 (0=CPU 核数)")
-        parser.add_argument("--batch", type=int, default=2000,
-                            help="每批处理多少条(控制内存与进度粒度)")
+                            help="同时删除本地预览 PDF 文件")
 
     def handle(self, *args, **opts):
         dry = opts["dry_run"]
         limit = opts["limit"]
-        del_preview = opts["delete_file"]
-        batch = max(1, opts["batch"])
-        workers = opts["workers"] or (os.cpu_count() or 4)
+        del_file = opts["delete_file"]
 
         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"workers={workers} batch={batch}"
-            + ("  (dry-run)" if dry else ""))
+            f"候选(has_fulltext_pdf=True 且 DOI 以 10.1016 开头): {total}")
 
-        rows_iter = qs.values(
-            "id", "doi", "publication_date", "has_fulltext_xml", "fail_reason"
-        ).iterator(chunk_size=batch)
+        checked = fixed = only_pdf = missing = unreadable = 0
+        for paper in qs.iterator():
+            if limit and checked >= limit:
+                break
+            checked += 1
 
-        checked = preview = broken = only_pdf = deleted = 0
-        missing = unknown = 0
+            path = paper.init_paper_path("pdf")
+            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:
-            stop = False
-            for chunk in _batched(rows_iter, batch):
-                if stop:
-                    break
-                paths = [_pdf_path(r["doi"], r["publication_date"]) for r in chunk]
-                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
+            pages = _pdf_page_count(content)
+            if pages is None:
+                unreadable += 1
+                continue
+            if pages > 1:
+                continue  # 真全文, 跳过
 
-                    if kind == "missing":
-                        missing += 1
-                        continue
-                    if kind in ("ok", "unknown", "unreadable"):
-                        if kind != "ok":
-                            unknown += 1
-                        continue
+            fixed += 1
+            only_pdf_case = not paper.has_fulltext_xml
+            if only_pdf_case:
+                only_pdf += 1
+            self.stdout.write(
+                f"[preview {pages}p]{' (only-pdf)' if only_pdf_case else ''} "
+                f"{paper.doi}  {path}")
+            if dry:
+                continue
 
-                    # kind in ('preview', 'broken'): 纠正标记
-                    do_delete = (kind == "broken") or del_preview
-                    only_pdf_case = not r["has_fulltext_xml"]
-                    if kind == "preview":
-                        preview += 1
-                        tag = f"preview {pages}p"
-                        reason = "elsevier_pdf_preview_only"
-                    else:
-                        broken += 1
-                        tag = "broken"
-                        reason = "pdf_broken"
-                    if only_pdf_case:
-                        only_pdf += 1
-                    self.stdout.write(
-                        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}")
+            paper.has_fulltext_pdf = False
+            update_fields = ["has_fulltext_pdf", "update_time"]
+            # 没有 XML 全文时, 之前的 has_fulltext 只是被这张假预览页置上的, 一并回退
+            if not paper.has_fulltext_xml:
+                paper.has_fulltext = False
+                update_fields.insert(0, "has_fulltext")
+            paper.save(update_fields=update_fields)
+            if "elsevier_pdf_preview_only" not in (paper.fail_reason or ""):
+                paper.save_fail_reason("elsevier_pdf_preview_only")
+            if del_file:
+                try:
+                    os.remove(path)
+                except OSError:
+                    pass
 
         self.stdout.write(self.style.SUCCESS(
-            f"完成 检查={checked} 预览页={preview} 坏文件={broken} "
-            f"(无XML全文一并回退has_fulltext={only_pdf}) 删除文件={deleted} "
-            f"文件缺失={missing} 未知/跳过={unknown}"
+            f"检查={checked} 预览页修复={fixed} (其中无XML全文/一并回退has_fulltext={only_pdf}) "
+            f"文件缺失={missing} 无法解析={unreadable}"
             + ("  (dry-run, 未写库)" if dry else "")
         ))

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}"))

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),
-    ]

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),
-        ),
-    ]

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(

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

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
-from .title_utils import assess_title, extract_title_from_xml, is_better_title
-from uuid import uuid4
+import re
+from .d_oaurl import download_from_url_playwright
 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_source = "openalex"
-    if assess_title(paper.title).suspect:
-        paper.title_quality_status = Paper.TITLE_SUSPECT
+    paper.title = record["display_name"]
     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,
-                                 types:str=OPENALEX_WORK_TYPES):
+def get_paper_meta_from_openalex(publication_year:int, keywords:str="", search:str="", end_year:int=None):
     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}_{types}"
+    cache_key = f"openalex_cursor_{publication_year}_{keywords}{search}"
     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,
-                    "types": types
+                    "end_year": end_year
                 },
                 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,
-                                      types: str = OPENALEX_WORK_TYPES):
+                                      combo_index: int = 0, combos=None):
     """按发表日期一次性回补论文索引,支持断点续传(应对 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",
-                     "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)
+    paper.save(update_fields=["has_abstract", "has_abstract_xml",
+                              "has_fulltext", "has_fulltext_xml", "update_time"])
     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_pdf').count()
+    """获取实际在下载的任务数"""
+    return Paper.objects.filter(fetch_status='downloading').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__startswith="downloading")
+    qs = Paper.objects.filter(fetch_status="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 里起浏览器太重, 且会开有头窗口)。
-    # 打 oa_url_need_play 标记, 交由独立脚本 scripts/get_pdf_by_playwright.py 兜底。
-    paper.save_fail_reason(f"oa_url_need_play: {err_msg}")
-    return f"oa_url_need_play: {err_msg}"
+    # 策略3: Playwright（最终回退）
+    is_ok, err_msg = run_async(download_from_url_playwright(paper.oa_url, paper_path))
+    if is_ok:
+        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"(每日额度,
-        # UTC 午夜重置), 放宽到 "Insufficient" 统一匹配, 避免落进 2 分钟短退避分支空转
-        if "Insufficient" in message:
+        if "Insufficient credits" 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,
-    # 交由独立脚本用浏览器兜底(同 oa_url need_play 的分工)。
-    from .d_scihub import download_by_doi_http
-    is_ok, err_msg = download_by_doi_http(paper.doi, paper.init_paper_path("pdf"))
+    from .d_scihub import download_paper_by_doi
+    is_ok, err_msg = download_paper_by_doi(paper.doi, paper.init_paper_path("pdf"))
     if is_ok:
         paper.has_fulltext = True
         paper.has_fulltext_pdf = True

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()

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

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.permissions import IsAuthenticated
-from rest_framework.settings import api_settings
+from rest_framework.decorators import api_view, permission_classes
+from rest_framework.permissions import AllowAny
 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([HasApiKey | IsAuthenticated])
+@permission_classes([AllowAny])
 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 authenticators
+            return []
+        return super().get_authenticators()
 
     def get_permissions(self):
         if self.request.method == 'GET':
-            return [(HasApiKey | IsAuthenticated)()]
+            return [AllowAny()]
         return super().get_permissions()

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')

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密钥',
-            },
-        ),
-    ]

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)

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)

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()