fix(data): corpus --limit 改每段独立预算+pdf五类循环补门控 [nas]

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

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