From 2fe9c4cd1fe763328200c635b401f48ac54c9d7e Mon Sep 17 00:00:00 2001 From: caoqianming Date: Tue, 21 Jul 2026 14:16:34 +0800 Subject: [PATCH] =?UTF-8?q?fix(resm):=20download=5Fpdf=E9=93=BE=E6=94=B9?= =?UTF-8?q?=E7=94=A8=E7=8B=AC=E7=AB=8B=E9=94=81downloading=5Fpdf,=20?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=B9=B6=E5=8F=91=E9=97=B8=E9=97=A8=E8=A2=AB?= =?UTF-8?q?=E5=85=B6=E4=BB=96=E9=93=BE=E5=8D=A1=E6=AD=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit send_download_fulltext_task 的 can_send_more 原来统计全库 fetch_status=downloading, 而 elsevier/openalex 保活链常驻十几个粗锁, 导致 download_pdf 永远发不出任务 (oa_url/Sci-Hub/Elsevier 主下载路径完全停摆)。现 download_pdf 用 downloading_pdf 独立锁, 闸门只数自己的在途任务; 各链查询改为 startswith 排除两种锁, release_working_paper 同步释放两种锁。 另修复: download_pdf 遇已被其他链锁定的论文时, finally 中 fetch_end 会误清 别的链持有的锁; Paper.DoesNotExist 时 finally 引用未定义的 paper。 Co-Authored-By: Claude Fable 5 --- CLAUDE.md | 2 +- apps/resm/models.py | 2 +- apps/resm/tasks.py | 30 +++++++++++++++++------------- 3 files changed, 19 insertions(+), 15 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 086a2dd..646aab2 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -177,7 +177,7 @@ The paper fetch pipeline in `apps/resm/tasks.py` currently includes: Download behavior is stateful: -- `fetch_status="downloading"` is used as a coarse lock +- `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"`) - `fail_reason` accumulates fetch failures - files are stored under `media/papers////` diff --git a/apps/resm/models.py b/apps/resm/models.py index e294066..1853ad2 100644 --- a/apps/resm/models.py +++ b/apps/resm/models.py @@ -27,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 + fetch_status = models.CharField(max_length=20, null=True, blank=True) # downloading(elsevier/openalex链) / downloading_pdf(download_pdf链) fail_reason = models.TextField(null=True, blank=True) source = models.CharField( diff --git a/apps/resm/tasks.py b/apps/resm/tasks.py index 2d4c9eb..a3ec563 100644 --- a/apps/resm/tasks.py +++ b/apps/resm/tasks.py @@ -524,7 +524,7 @@ def get_pdf_from_openalex(number_of_task: int =10): count = 0 qs = Paper.objects.filter(is_oa=True, has_fulltext=False).exclude( - fetch_status="downloading").exclude(fail_reason__contains="openalex_pdf_not_found")[:number_of_task] + fetch_status__startswith="downloading").exclude(fail_reason__contains="openalex_pdf_not_found")[:number_of_task] if not qs.exists(): return "done" # 不自重发, 交给 beat 轮询拉起 msg = "" @@ -652,13 +652,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="downloading" + qs = qs.exclude(fetch_status__startswith="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="downloading" + ).exclude(fetch_status__startswith="downloading" ).exclude(fail_reason__contains="elsevier_pdf_preview_only" ).filter(doi__startswith="10.1016") @@ -674,7 +674,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 == "downloading": + if paper.fetch_status and paper.fetch_status.startswith("downloading"): continue paper.fetch(status="downloading") try: @@ -696,7 +696,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 == "downloading": + if paper.fetch_status and paper.fetch_status.startswith("downloading"): continue paper.fetch(status="downloading") try: @@ -719,8 +719,10 @@ def get_abstract_from_elsevier(number_of_task:int = 20, exclude_failed:bool=True return f'{err_msg}, abs {count_abs}, fulltext {count_fulltext}, pdf {count_pdf}' def get_actual_running_count(): - """获取实际在下载的任务数""" - return Paper.objects.filter(fetch_status='downloading').count() + """获取本下载链路(download_pdf)实际在下载的任务数。 + 只数 downloading_pdf, 不含 elsevier/openalex 抓取链的 downloading 粗锁, + 否则那两条链常驻的十几个锁会把本链路的并发闸门永久卡死。""" + return Paper.objects.filter(fetch_status='downloading_pdf').count() def can_send_more(max_running): return get_actual_running_count() < max_running @@ -731,7 +733,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='downloading' + fetch_status__startswith='downloading' ).exclude(fail_reason__contains="download_pdf_tried") if not qs.exists(): return "done" @@ -757,7 +759,8 @@ def send_download_fulltext_task(number_of_task=100): @shared_task(base=CustomTask) def release_working_paper(minutes=10): - qs = Paper.objects.filter(fetch_status="downloading") + # startswith: 同时释放 downloading(elsevier/openalex 链) 和 downloading_pdf(download_pdf 链) + qs = Paper.objects.filter(fetch_status__startswith="downloading") count = 0 for paper in qs: if paper.update_time < timezone.now() - timedelta(minutes=minutes): @@ -770,11 +773,12 @@ 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 = Paper.objects.get(id=paper_id) - if paper.fetch_status == "downloading": - return - paper.fetch("downloading") + paper.fetch("downloading_pdf") msg = "no_method_to_get_pdf" current_from = "" if paper.oa_url: