fix(data): daily增量marker按日折叠——首日全量done后次日整段跳过增量的day-2致命坑 [nas]

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

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

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