feat(data): T6 验收器 verify_corpus.py+每域请求统计日志 [nas]

- corpus_download 收尾补 stats 行(每域 n_requests/avg_interval/[429]/[cooled])
  =§18.5「日志限速合规」验收依据(平均间隔≥设定值可直读)
- verify_corpus.py 只读验收器(零写入零网络,容器内跑): 三域分区/行数/键重复/
  账本一致性(ledger<uniq=FAIL)/PDF 索引↔文件/marker/depth/容量/最近日志尾部;
  NAS 真数据首验 PASS(三域 dups=0, ledger=uniq, 索引=文件, 644G)
- 晨检一条命令复用(首周人工盯跑工具)

scripts/data_platform/corpus_download.py

@@ -568,6 +568,8 @@ class Ctx:
         self.units = 0
         self.stage_units = 0
         self.depth = {}
+        self.clients = {}
+        self.t0 = time.monotonic()
 
     def reset_stage(self):
         """--limit 预算按段独立(冒烟要覆盖每段;曾因预算共享+pdf 段无门控
@@ -840,6 +842,7 @@ def run_daily(ctx, pool):
     cl_cn = DomainClient("cninfo")
     cl_em = DomainClient("eastmoney")
     cl_art = DomainClient("article")
+    ctx.clients = {"cninfo": cl_cn, "eastmoney": cl_em, "article": cl_art}
     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)
@@ -1083,6 +1086,7 @@ def run_backfill(ctx, pool):
     cl_cn = DomainClient("cninfo")
     cl_em = DomainClient("eastmoney")
     cl_art = DomainClient("article")
+    ctx.clients = {"cninfo": cl_cn, "eastmoney": cl_em, "article": cl_art}
     _run_ann_backfill(ctx, pool, cl_cn, ledgers, recents)
     _run_news_backfill(ctx, pool, cl_em, ledgers, recents)
     _run_fulltext_backfill(ctx, cl_art, ledgers, recents)
@@ -1125,6 +1129,13 @@ def run_lane(lane, until=None, limit=None):
         log.info("墙钟到(--until %s): 完成当前 unit 后 checkpoint 退出", until)
     finally:
         _save_depth(ctx)
+        for name, cl in ctx.clients.items():
+            if cl.n_requests:
+                log.info("stats %s: n_requests=%d avg_interval=%.2fs%s",
+                         name, cl.n_requests,
+                         (time.monotonic() - t0) / cl.n_requests,
+                         " [429]" if cl.rate_limited else
+                         " [cooled]" if cl.cooled else "")
         log.info("lane=%s done in %.0fs: units=%d failed=%d rc=%d",
                  lane, time.monotonic() - t0, ctx.units, ctx.failed, ctx.rc())
     return ctx.rc()

scripts/data_platform/verify_corpus.py

@@ -0,0 +1,119 @@
+# -*- coding: utf-8 -*-
+"""verify_corpus.py — 语料库只读验收器(spec §18.5/T6 晨检, 零写入零网络)。
+
+在一次性容器里跑(同 corpus_download 的挂载):
+  docker run --rm --user 1024:100 --group-add 101 --entrypoint python \
+    -v /volume1/stock:/volume1/stock sanguo_vnpy_v2:lock-aligned \
+    /app/scripts/data_platform/verify_corpus.py [--root /volume1/stock/corpus]
+
+输出: 每域分区/行数/键重复/账本一致性/marker/深度/容量/最近日志尾部 stats 行。
+判定: 任何 DUPS>0 或 LEDGER<ROWS(账本丢键) 记 FAIL;其余为信息项。
+"""
+import argparse
+import glob
+import json
+import sys
+from pathlib import Path
+
+import pandas as pd
+
+ID_KEYS = {"ann_meta": ["announcement_id"],
+           "news_meta": ["art_code", "stock_code"],
+           "news_fulltext": ["art_code"]}
+
+
+def load_domain(root, domain):
+    parts = sorted(glob.glob(str(root / domain / "dt=*" / "part-0.parquet")))
+    if not parts:
+        return None, []
+    return pd.concat([pd.read_parquet(p, columns=ID_KEYS[domain]) for p in parts],
+                     ignore_index=True), parts
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--root", default="/volume1/stock/corpus")
+    args = ap.parse_args()
+    root = Path(args.root)
+    fails = []
+
+    print("=" * 72)
+    print(f"corpus verify @ {root}")
+    for domain in ID_KEYS:
+        df, parts = load_domain(root, domain)
+        if df is None:
+            print(f"[{domain}] 无分区(尚无数据)")
+            continue
+        keys = df.apply(lambda r: "|".join(str(r[c]) for c in ID_KEYS[domain]),
+                        axis=1)
+        dups = int(keys.duplicated().sum())
+        ledger = root / "state" / f"ids_{domain}.parquet"
+        n_led = len(pd.read_parquet(ledger)) if ledger.exists() else 0
+        n_uniq = int(keys.nunique())
+        flag = "FAIL" if (dups or n_led < n_uniq) else "OK"
+        if dups or n_led < n_uniq:
+            fails.append(domain)
+        dt_range = (Path(parts[0]).parent.name, Path(parts[-1]).parent.name)
+        print(f"[{domain}] {flag} partitions={len(parts)} {dt_range[0]}..{dt_range[1]} "
+              f"rows={len(df)} uniq_keys={n_uniq} dups={dups} ledger={n_led}")
+
+    # PDF: 索引行 vs 文件数
+    idx_path = root / "state" / "ann_pdf_index.parquet"
+    pdfs = list((root / "ann_pdf").glob("*/*.pdf"))
+    if idx_path.exists():
+        idx = pd.read_parquet(idx_path)
+        files = {p.stem for p in pdfs}
+        orphan_files = len(files - set(idx["announcement_id"]))
+        missing_files = int((~idx["announcement_id"].isin(files)).sum())
+        bad = idx["bytes"].fillna(0).le(0).sum() if "bytes" in idx else 0
+        flag = "FAIL" if (missing_files or int(bad)) else "OK"
+        if flag == "FAIL":
+            fails.append("ann_pdf")
+        print(f"[ann_pdf] {flag} index_rows={len(idx)} files={len(pdfs)} "
+              f"missing_files={missing_files} orphan_files={orphan_files} "
+              f"zero_byte_rows={int(bad)}")
+    else:
+        print(f"[ann_pdf] 无索引(files={len(pdfs)})")
+
+    # markers
+    mk = root / "state" / "markers"
+    per = {}
+    if mk.exists():
+        for lane_dir in mk.iterdir():
+            for stage_dir in lane_dir.iterdir():
+                n = len(list(stage_dir.glob("*.done")))
+                per[f"{lane_dir.name}/{stage_dir.name}"] = n
+    print(f"[markers] {per}")
+
+    # depth
+    dpath = root / "state" / "depth_summary.json"
+    if dpath.exists():
+        d = json.loads(dpath.read_text(encoding="utf-8"))
+        print(f"[depth] " + " ".join(f"{k}={len(v)}只" for k, v in d.items()))
+        for src, stocks in d.items():
+            earliest = min(stocks.values()) if stocks else "-"
+            print(f"        {src} earliest={earliest}")
+
+    # 容量
+    try:
+        import shutil
+        free = shutil.disk_usage(str(root)).free / (1 << 30)
+        print(f"[disk] free={free:.0f}G (PDF 闸门阈值 200G: "
+              f"{'暂停中' if free < 200 else '正常'})")
+    except OSError as e:
+        print(f"[disk] 探测失败 {e}")
+
+    # 最近一次运行尾部(stats=限速合规依据)
+    logs = sorted((root / "logs").glob("corpus_*.log"))
+    if logs:
+        tail = logs[-1].read_text(encoding="utf-8").strip().splitlines()[-8:]
+        print(f"[last-log] {logs[-1].name}")
+        for ln in tail:
+            print("        " + ln)
+    print("=" * 72)
+    print("VERIFY " + ("FAIL: " + ",".join(fails) if fails else "PASS"))
+    sys.exit(1 if fails else 0)
+
+
+if __name__ == "__main__":
+    main()