fix(resm): download_pdf链改用独立锁downloading_pdf, 修复并发闸门被其他链卡死
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 <noreply@anthropic.com>
This commit is contained in:
parent
0705b72859
commit
2fe9c4cd1f
|
|
@ -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/<year>/<month>/<day>/`
|
||||
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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
|
||||
"""
|
||||
try:
|
||||
paper = Paper.objects.get(id=paper_id)
|
||||
if paper.fetch_status == "downloading":
|
||||
if paper.fetch_status and paper.fetch_status.startswith("downloading"):
|
||||
# 已被任一链锁定: 直接返回, 不能走 finally 的 fetch_end, 否则会误清别的链持有的锁
|
||||
return
|
||||
paper.fetch("downloading")
|
||||
try:
|
||||
paper.fetch("downloading_pdf")
|
||||
msg = "no_method_to_get_pdf"
|
||||
current_from = ""
|
||||
if paper.oa_url:
|
||||
|
|
|
|||
Loading…
Reference in New Issue