From 89da5f71e087da8fef4b1b95fbc68d8ce20d4f7c Mon Sep 17 00:00:00 2001 From: claude_dev Date: Sun, 6 Sep 2026 23:40:53 +0800 Subject: [PATCH] =?UTF-8?q?fix(data):=20corpus=20--limit=20=E6=94=B9?= =?UTF-8?q?=E6=AF=8F=E6=AE=B5=E7=8B=AC=E7=AB=8B=E9=A2=84=E7=AE=97+pdf?= =?UTF-8?q?=E4=BA=94=E7=B1=BB=E5=BE=AA=E7=8E=AF=E8=A1=A5=E9=97=A8=E6=8E=A7?= =?UTF-8?q?=20[nas]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit NAS 真实冒烟(--limit 2)假死 12min 事故根因(x2): ①预算共享: ann 段吃光 limit 后 news/fulltext 即刻 break(假死其一); ②pdf 五类查询循环完全无门控无计数 → 全池 6168 股 @1.3s 扫描 ~2.2h, 周日窗口 0 公告 0 日志=假死其二(docker stop 收尸, log 停在 lane start) 修: Ctx.stage_units+reset_stage()+budget_exhausted() 八段入口重置; 五类循环计数即耗预算;回归测试钉死(fivecat 调用数=预算数,非全池) 连带修正 smoke 语义=文档承诺的「每段最多 N unit」;52+269 全绿 --- scripts/data_platform/corpus_download.py | 41 ++++++++++++++++++--- tests/data_platform/test_corpus_download.py | 19 ++++++++++ 2 files changed, 54 insertions(+), 6 deletions(-) diff --git a/scripts/data_platform/corpus_download.py b/scripts/data_platform/corpus_download.py index e29d412..8c5ccfe 100644 --- a/scripts/data_platform/corpus_download.py +++ b/scripts/data_platform/corpus_download.py @@ -563,8 +563,17 @@ class Ctx: self.hard_cool = False self.wallclock = False self.units = 0 + self.stage_units = 0 self.depth = {} + def reset_stage(self): + """--limit 预算按段独立(冒烟要覆盖每段;曾因预算共享+pdf 段无门控 + 导致全池扫描假死 12min, 2026-09-06 NAS 冒烟实锤)。""" + self.stage_units = 0 + + def budget_exhausted(self): + return self.limit is not None and self.stage_units >= self.limit + def stop_now(self): """完成当前 unit 后调用: 过墙钟 → 抛 WallClockStop(checkpoint 语义)。""" if self.deadline and dt.datetime.now() >= self.deadline: @@ -637,6 +646,7 @@ def _maybe_flush(ledgers, since_flush): # ---------- daily lane ---------- def _run_ann_increment(ctx, pool, cl_cn, ledgers, recents): + ctx.reset_stage() today = dt.date.today() yday = today - dt.timedelta(days=1) first_seen = today.isoformat() @@ -646,7 +656,7 @@ def _run_ann_increment(ctx, pool, cl_cn, ledgers, recents): if ctx.limit is None: if is_done(ctx.lane, "ann", code): continue - elif ctx.units >= ctx.limit: + elif ctx.budget_exhausted(): break try: rows = fetch_cninfo_announcements(cl_cn, code, org, @@ -671,17 +681,19 @@ def _run_ann_increment(ctx, pool, cl_cn, ledgers, recents): if ctx.limit is None: mark_done(ctx.lane, "ann", code) ctx.units += 1 + ctx.stage_units += 1 since_flush = _maybe_flush(ledgers, since_flush + 1) ctx.stop_now() return candidates def _run_news_increment(ctx, pool, cl_em, ledgers, recents, new_articles): + ctx.reset_stage() for code, _org in pool: if ctx.limit is None: if is_done(ctx.lane, "news", code): continue - elif ctx.units >= ctx.limit: + elif ctx.budget_exhausted(): break try: rows = fetch_news_recent(cl_em, code) @@ -704,10 +716,12 @@ def _run_news_increment(ctx, pool, cl_em, ledgers, recents, new_articles): if ctx.limit is None: mark_done(ctx.lane, "news", code) ctx.units += 1 + ctx.stage_units += 1 ctx.stop_now() def _run_fulltext(ctx, cl_em, ledgers, recents, new_articles): + ctx.reset_stage() seen = set() for art_code, url, show_time in new_articles: if art_code in seen or is_done(ctx.lane, "fulltext", art_code): @@ -741,13 +755,17 @@ def _run_fulltext(ctx, cl_em, ledgers, recents, new_articles): mark_done(ctx.lane, "fulltext", art_code) seen.add(art_code) ctx.units += 1 + ctx.stage_units += 1 ctx.stop_now() def _run_pdf_daily(ctx, pool, cl_cn, ledgers, candidates): + ctx.reset_stage() today = dt.date.today() yday = today - dt.timedelta(days=1) for code, org in pool: + if ctx.budget_exhausted(): + break try: rows = fetch_cninfo_announcements(cl_cn, code, org, yday.isoformat(), today.isoformat(), @@ -762,6 +780,8 @@ def _run_pdf_daily(ctx, pool, cl_cn, ledgers, candidates): log.warning("pdf fivecat %s 失败: %s", code, e) continue candidates.extend(rows) + ctx.units += 1 + ctx.stage_units += 1 ctx.stop_now() seen = set() index_rows = [] @@ -804,6 +824,7 @@ def _run_pdf_daily(ctx, pool, cl_cn, ledgers, candidates): if ctx.limit is None: mark_done(ctx.lane, "pdf", aid) ctx.units += 1 + ctx.stage_units += 1 ctx.stop_now() _append_pdf_index(index_rows) @@ -825,6 +846,7 @@ def run_daily(ctx, pool): # ---------- backfill lane(①ann_meta ②news_meta ③fulltext ④PDF;T5) ---------- def _run_ann_backfill(ctx, pool, cl_cn, ledgers, recents): + ctx.reset_stage() """①公告元数据回补 2000→今, unit=(stock,year), 逐年按池序(探测优先:空年=合法)。""" today = dt.date.today() first_seen = today.isoformat() @@ -833,7 +855,7 @@ def _run_ann_backfill(ctx, pool, cl_cn, ledgers, recents): year_end = today.isoformat() if year == today.year else f"{year}-12-31" for code, org in pool: unit = f"{code}_{year}" - if ctx.limit is not None and ctx.units >= ctx.limit: + if ctx.budget_exhausted(): return if is_done(ctx.lane, "ann", unit): continue @@ -859,11 +881,13 @@ def _run_ann_backfill(ctx, pool, cl_cn, ledgers, recents): if ctx.limit is None: mark_done(ctx.lane, "ann", unit) ctx.units += 1 + ctx.stage_units += 1 since_flush = _maybe_flush(ledgers, since_flush + 1) ctx.stop_now() def _run_news_backfill(ctx, pool, cl_em, ledgers, recents): + ctx.reset_stage() """②新闻标题级回补(np-listapi), unit=(stock,page), 翻到空页或 page≥200 止。""" for code, _org in pool: if is_done(ctx.lane, "news_end", code): @@ -872,7 +896,7 @@ def _run_news_backfill(ctx, pool, cl_em, ledgers, recents): capped = False for page in range(1, NEWS_BACKFILL_MAX_PAGE + 1): unit = f"{code}_p{page}" - if ctx.limit is not None and ctx.units >= ctx.limit: + if ctx.budget_exhausted(): return if is_done(ctx.lane, "news", unit): continue @@ -900,6 +924,7 @@ def _run_news_backfill(ctx, pool, cl_em, ledgers, recents): if ctx.limit is None: mark_done(ctx.lane, "news", unit) ctx.units += 1 + ctx.stage_units += 1 if page == NEWS_BACKFILL_MAX_PAGE: capped = True ctx.stop_now() @@ -909,6 +934,7 @@ def _run_news_backfill(ctx, pool, cl_em, ledgers, recents): def _run_fulltext_backfill(ctx, cl_art, ledgers, recents): + ctx.reset_stage() """③新闻全文回补: 扫 news_meta 库存 → 未入账 art 补正文; 账本即进度 (无 marker 文件); 404 墓碑行(content_text=None)入账不再重试。""" rows_all = [] @@ -930,7 +956,7 @@ def _run_fulltext_backfill(ctx, cl_art, ledgers, recents): since_flush = 0 today = dt.date.today().isoformat() for r in todo: - if ctx.limit is not None and ctx.units >= ctx.limit: + if ctx.budget_exhausted(): return try: content = html_to_text(fetch_article_html(cl_art, r["url"])) @@ -954,11 +980,13 @@ def _run_fulltext_backfill(ctx, cl_art, ledgers, recents): ledgers["news_fulltext"].add([str(r["art_code"])]) recents["news_fulltext"].add([str(r["art_code"])]) ctx.units += 1 + ctx.stage_units += 1 since_flush = _maybe_flush(ledgers, since_flush + 1) ctx.stop_now() def _run_pdf_backfill(ctx, cl_cn): + ctx.reset_stage() """④定期+业绩类 PDF 回补(新→旧): ann_meta 库存标题谓词筛六类; 文件在=跳过; 404 墓碑; 容量闸门=段首+每 50GB 自查, <200G 暂停本段并落告警(其余段不受影响)。""" free_gb = _disk_free_gb(CORPUS_ROOT) @@ -995,7 +1023,7 @@ def _run_pdf_backfill(ctx, cl_cn): index_rows = [] today = dt.date.today().isoformat() for r in cand.to_dict("records"): - if ctx.limit is not None and ctx.units >= ctx.limit: + if ctx.budget_exhausted(): break aid = str(r["announcement_id"]) if aid in missing: @@ -1038,6 +1066,7 @@ def _run_pdf_backfill(ctx, cl_cn): "local_path": str(dest), "bytes": size, "fetch_date": today}) ctx.units += 1 + ctx.stage_units += 1 ctx.stop_now() _append_pdf_index(index_rows) diff --git a/tests/data_platform/test_corpus_download.py b/tests/data_platform/test_corpus_download.py index aa15b41..3ff99f3 100644 --- a/tests/data_platform/test_corpus_download.py +++ b/tests/data_platform/test_corpus_download.py @@ -458,6 +458,25 @@ def test_daily_limit_smoke_writes_but_no_markers(root, monkeypatch): assert not any((root / "state" / "markers").rglob("*.done")) +def test_daily_limit_is_per_stage_and_bounds_pdf_fivecat(root, monkeypatch): + """回归(2026-09-06 NAS 冒烟假死 12min): ①--limit 预算按段独立,不被首段 + 吃光;②pdf 五类查询循环必须有门控——曾漏门控致全池 6168 股扫描。""" + _wire_daily(monkeypatch, root) + orig = cd.fetch_cninfo_announcements + calls = {"ann": 0, "fivecat": 0} + + def counting(client, code, org, start, end, category=""): + calls["fivecat" if category else "ann"] += 1 + return orig(client, code, org, start, end, category) + + monkeypatch.setattr(cd, "fetch_cninfo_announcements", counting) + cd.run_lane("daily", limit=1) + assert calls["ann"] == 1 # ann 段 1 发 + assert calls["fivecat"] == 1 # pdf 段独立预算 1 发,而非全池 + news_calls = cd.fetch_news_recent.call_count + assert news_calls == 1 # news 段独立预算未被 ann 挤占 + + def test_daily_express_title_routes_to_pdf(root, monkeypatch): """标题含「业绩快报」→ 进当日 PDF 待下清单(无服务端码,客户端谓词)。""" todo = cd.collect_pdf_candidates(