From 52756e35abf91368eeca37641883b9fe011762a6 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Mon, 7 Sep 2026 18:53:41 +0800 Subject: [PATCH] =?UTF-8?q?fix(data):=20daily=E5=A2=9E=E9=87=8Fmarker?= =?UTF-8?q?=E6=8C=89=E6=97=A5=E6=8A=98=E5=8F=A0=E2=80=94=E2=80=94=E9=A6=96?= =?UTF-8?q?=E6=97=A5=E5=85=A8=E9=87=8Fdone=E5=90=8E=E6=AC=A1=E6=97=A5?= =?UTF-8?q?=E6=95=B4=E6=AE=B5=E8=B7=B3=E8=BF=87=E5=A2=9E=E9=87=8F=E7=9A=84?= =?UTF-8?q?day-2=E8=87=B4=E5=91=BD=E5=9D=91=20[nas]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 09-07 首跑夜复查发现: daily ann/news marker 无日期维度,首跑日已写满 6168+6168 → 次日 06:00 首班会整段跳过增量采集,且 backfill 仍在跑会掩盖 症状。修=unit 名前挂 dt=YYYY-MM-DD/ 子目录(_dunit)+日初清旧日目录与 历史平铺 marker(_prune_daily_increment_markers);fulltext/pdf 段键=内容 id(不可变一次永逸)保持平铺不走折叠。契约测试: 次日重新拉窗承接新公告/ 当日断点续传不变/旧日目录日初清。 --- scripts/data_platform/corpus_download.py | 34 ++++++++++-- tests/data_platform/test_corpus_download.py | 58 ++++++++++++++++++--- 2 files changed, 80 insertions(+), 12 deletions(-) diff --git a/scripts/data_platform/corpus_download.py b/scripts/data_platform/corpus_download.py index 42a4eb6..56c91aa 100644 --- a/scripts/data_platform/corpus_download.py +++ b/scripts/data_platform/corpus_download.py @@ -724,6 +724,30 @@ def _save_depth(ctx): # ---------- daily lane ---------- +def _dunit(code, day): + """daily 增量段(ann/news) marker 单元名: 按日折叠(dt=YYYY-MM-DD/ 子目录)。 + 增量窗每日一新,断点续传只保护当日——无日期维度会让首日全量 done 后次日 + 整段跳过增量采集(09-07 首跑夜抓到的 day-2 坑)。fulltext/pdf 段键=内容 id + (公告/文章不可变),保持平铺一次永逸,不走本折叠。""" + return f"dt={day.isoformat()}/{code}" + + +def _prune_daily_increment_markers(): + """daily 日初清理: ann/news 增量 marker 只留今日 dt= 目录,并清历史遗留的 + 平铺 *.done(首跑日产物,无日期维度)——两者都是次日跳过坑与无限增长防线。""" + today = dt.date.today().isoformat() + for stage in ("ann", "news"): + d = CORPUS_ROOT / "state" / "markers" / "daily" / stage + if not d.exists(): + continue + for child in d.iterdir(): + if child.is_dir(): + if child.name != f"dt={today}": + shutil.rmtree(child, ignore_errors=True) + elif child.suffix == ".done": + child.unlink(missing_ok=True) + + def _run_ann_increment(ctx, pool, cl_cn, ledgers, recents): ctx.reset_stage() today = dt.date.today() @@ -732,7 +756,7 @@ def _run_ann_increment(ctx, pool, cl_cn, ledgers, recents): candidates = [] for code, org in pool: if ctx.limit is None: - if is_done(ctx.lane, "ann", code): + if is_done(ctx.lane, "ann", _dunit(code, today)): continue elif ctx.budget_exhausted(): break @@ -756,16 +780,17 @@ def _run_ann_increment(ctx, pool, cl_cn, ledgers, recents): _record_depth(ctx, "ann", r["sec_code"] or code, (r["ann_time"] or "")[:10]) log.info("ann %s: +%d/%d", code, len(new), len(rows)) - ctx.unit_done("ann", code, mark=ctx.limit is None) + ctx.unit_done("ann", _dunit(code, today), mark=ctx.limit is None) ctx.stop_now() return candidates def _run_news_increment(ctx, pool, cl_em, ledgers, recents, new_articles): ctx.reset_stage() + today = dt.date.today() for code, _org in pool: if ctx.limit is None: - if is_done(ctx.lane, "news", code): + if is_done(ctx.lane, "news", _dunit(code, today)): continue elif ctx.budget_exhausted(): break @@ -787,7 +812,7 @@ def _run_news_increment(ctx, pool, cl_em, ledgers, recents, new_articles): _record_depth(ctx, "news", code, (r["show_time"] or "")[:10]) new_articles.append((r["art_code"], r["url"], r["show_time"])) log.info("news %s: +%d/%d", code, len(new), len(rows)) - ctx.unit_done("news", code, mark=ctx.limit is None) + ctx.unit_done("news", _dunit(code, today), mark=ctx.limit is None) ctx.stop_now() @@ -921,6 +946,7 @@ def run_daily(ctx, pool): ctx.clients = {"cninfo": cl_cn, "eastmoney": cl_em, "article": cl_art} ctx.set_stores(ledgers) _reconcile_ledgers(ledgers, recents) + _prune_daily_increment_markers() candidates = _run_ann_increment(ctx, pool, cl_cn, ledgers, recents) _run_news_increment(ctx, pool, cl_em, ledgers, recents, new_articles := []) _run_fulltext(ctx, cl_art, ledgers, recents, new_articles) diff --git a/tests/data_platform/test_corpus_download.py b/tests/data_platform/test_corpus_download.py index b3b1386..10e1441 100644 --- a/tests/data_platform/test_corpus_download.py +++ b/tests/data_platform/test_corpus_download.py @@ -418,19 +418,58 @@ def test_daily_happy_path_writes_all_domains(root, monkeypatch): ft = pd.read_parquet(root / "news_fulltext" / f"dt={today}" / "part-0.parquet") assert len(ft) == 1 and "正文内容" in ft.iloc[0]["content_text"] # markers: ann/news 每股 done;fulltext 每 art done - assert cd.is_done("daily", "ann", "000001") and cd.is_done("daily", "ann", "600519") - assert cd.is_done("daily", "news", "600519") + assert cd.is_done("daily", "ann", f"dt={today}/000001") \ + and cd.is_done("daily", "ann", f"dt={today}/600519") + assert cd.is_done("daily", "news", f"dt={today}/600519") assert cd.is_done("daily", "fulltext", "art1") +def test_daily_increment_next_day_refetches(root, monkeypatch): + """daily ann/news 增量 marker 按日折叠(09-07 首跑夜抓到的 day-2 致命坑): + marker 无日期维度→首日全量 done 后次日整段跳过增量采集(backfill 仍在跑会掩盖)。 + 契约: 次日必须重新拉窗承接新公告/新闻;当日断点续传语义不变;旧日 dt= 目录日初清。""" + import pandas as pd + _wire_daily(monkeypatch, root) + day1 = dt.date.today() + assert cd.run_lane("daily") == 0 + assert cd.is_done("daily", "ann", f"dt={day1.isoformat()}/000001") + + class _NextDay(dt.date): + @classmethod + def today(cls): + return day1 + dt.timedelta(days=1) + monkeypatch.setattr(cd.dt, "date", _NextDay) + _wire_daily(monkeypatch, root, + ann_pages=[_cninfo_page([_ann_raw(aid="a2", code="000001")])], + news_rows=[{"code": "art2", "date": "2026-09-07 10:00:00", + "title": "次日新闻", "content": "摘要", + "mediaName": "源", "image": ""}]) + assert cd.run_lane("daily") == 0 + day2 = (day1 + dt.timedelta(days=1)).isoformat() + ann = pd.read_parquet(root / "ann_meta" / f"dt={day2}" / "part-0.parquet") + assert list(ann["announcement_id"]) == ["a2"] # 次日新公告承接 + news = pd.read_parquet(root / "news_meta" / f"dt={day2}" / "part-0.parquet") + assert "art2" in set(news["art_code"]) + ft = pd.read_parquet(root / "news_fulltext" / f"dt={day2}" / "part-0.parquet") + assert list(ft["art_code"]) == ["art2"] + # 旧日 dt= 目录日初被清(无限增长防线);当日再跑零 fetch(同日断点续传不变) + assert not (root / "state" / "markers" / "daily" / "ann" / f"dt={day1.isoformat()}").exists() + m = cd.fetch_cninfo_announcements + inc = lambda: sum(1 for c in m.call_args_list if not c.kwargs.get("category")) + n_before = inc() # 增量段调用(无 category);fivecat 全池扫描(mark=False)不计 + assert cd.run_lane("daily") == 0 + assert inc() == n_before + + def test_daily_rerun_idempotent_no_dup(root, monkeypatch): import pandas as pd _wire_daily(monkeypatch, root) assert cd.run_lane("daily") == 0 # markers 全 done → 第二趟零 fetch;强制清 marker 重跑同数据 → 零重复行 + today = dt.date.today().isoformat() for code, _ in POOL: - (root / "state" / "markers" / "daily" / "ann" / f"{code}.done").unlink() - (root / "state" / "markers" / "daily" / "news" / f"{code}.done").unlink() + (root / "state" / "markers" / "daily" / "ann" / f"dt={today}" / f"{code}.done").unlink() + (root / "state" / "markers" / "daily" / "news" / f"dt={today}" / f"{code}.done").unlink() assert cd.run_lane("daily") == 0 today = dt.date.today().isoformat() ann = pd.read_parquet(root / "ann_meta" / f"dt={today}" / "part-0.parquet") @@ -453,8 +492,9 @@ def test_daily_two_day_window_after_missed_day(root, monkeypatch): _wire_daily(monkeypatch, root) cd.run_lane("daily") # 清 marker 模拟「另一天再来」;同 id 仍是昨天发的 → 已在账,不重 + today = dt.date.today().isoformat() for code, _ in POOL: - (root / "state" / "markers" / "daily" / "ann" / f"{code}.done").unlink() + (root / "state" / "markers" / "daily" / "ann" / f"dt={today}" / f"{code}.done").unlink() cd.run_lane("daily") today = dt.date.today().isoformat() ann = pd.read_parquet(root / "ann_meta" / f"dt={today}" / "part-0.parquet") @@ -476,8 +516,9 @@ def test_daily_unit_fail_not_marked_and_rc1(root, monkeypatch): monkeypatch.setattr(cd, "fetch_cninfo_announcements", flaky) rc = cd.run_lane("daily") assert rc == 1 - assert not cd.is_done("daily", "ann", "000001") - assert cd.is_done("daily", "ann", "600519") + today = dt.date.today().isoformat() + assert not cd.is_done("daily", "ann", f"dt={today}/000001") + assert cd.is_done("daily", "ann", f"dt={today}/600519") today = dt.date.today().isoformat() ann = pd.read_parquet(root / "ann_meta" / f"dt={today}" / "part-0.parquet") assert len(ann) == 1 # 只有好股的 @@ -592,7 +633,8 @@ def test_wallclock_stops_after_current_unit(root, monkeypatch): past = (dt.datetime.now() - dt.timedelta(minutes=1)).strftime("%H:%M") rc = cd.run_lane("daily", until=past) assert rc == 3 - assert cd.is_done("daily", "ann", "000001") # 首 unit 完成后停 + today = dt.date.today().isoformat() + assert cd.is_done("daily", "ann", f"dt={today}/000001") # 首 unit 完成后停 def test_wallclock_future_runs_to_completion(root, monkeypatch):