From 1c3a61eb1247558e8f95da38ae93f78f4431291a Mon Sep 17 00:00:00 2001 From: claude_dev Date: Sun, 6 Sep 2026 23:24:29 +0800 Subject: [PATCH] =?UTF-8?q?feat(data):=20T5=20backfill=20=E2=91=A2?= =?UTF-8?q?=E5=85=A8=E6=96=87=20=E2=91=A3PDF+=E5=AE=B9=E9=87=8F=E9=97=B8?= =?UTF-8?q?=E9=97=A8=20[nas]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit spec §18.7 T5(backfill lane 补完,四段齐): - ③新闻全文回补: 扫 news_meta 库存→未入账 art 补正文;账本即进度(免 260万 marker 文件);404=墓碑行 content_text=None 入账不重试;文章域独立 限速 2s(RATE['article'],spec @2/s 慢车道) - ④PDF六类回补(新→旧): ann_meta 库存标题谓词筛六类→ann_time desc; 文件在=跳过(文件即进度);404 墓碑 state/pdf_missing.json;索引复用 _append_pdf_index(daily/backfill 同管道) - 容量闸门: 段首+每50GB 自查 _disk_free_gb,<200G 暂停 PDF 段+落 capacity_paused.json 告警(其余段不受影响,空间回升自动续) - 测试51个(root fixture 默认 _disk_free_gb=645:曾吃 Mac 真实磁盘 8G 余量误触发闸门);全套 268 绿 --- scripts/data_platform/corpus_download.py | 214 ++++++++++++++++++-- tests/data_platform/test_corpus_download.py | 180 +++++++++++++++- 2 files changed, 375 insertions(+), 19 deletions(-) diff --git a/scripts/data_platform/corpus_download.py b/scripts/data_platform/corpus_download.py index 6c3375b..e29d412 100644 --- a/scripts/data_platform/corpus_download.py +++ b/scripts/data_platform/corpus_download.py @@ -5,7 +5,8 @@ 与 sanguo-5m 同款;NAS 家宽出口 IP,与 VPS 东财限流零冲突。 --lane daily 四段: 公告增量(巨潮 T-1~T)/新闻增量(东财 search-api 带摘要)/ 新闻全文(昨日新 art_code)/当日 PDF(五类服务端码+业绩快报标题谓词) - --lane backfill ①ann_meta 2000起 ②news_meta np-listapi(未实施=T4, 先让路 rc=0) + --lane backfill ①ann_meta 2000起 ②news_meta np-listapi ③新闻全文(账本即进度, + 404墓碑) ④PDF六类新→旧(容量闸门<200G暂停段) --until HH:MM 墙钟自停: 完成当前 unit 后 checkpoint 退出(rc=3) --limit N 冒烟: 每段最多 N unit, 绝不落 marker(5m 09-17 教训) @@ -38,6 +39,7 @@ import logging import os import random import re +import shutil import sys import time from pathlib import Path @@ -71,8 +73,10 @@ DEDUP_WINDOW_DAYS = 30 FLUSH_EVERY = 50 # 账本/索引落盘周期(unit 计) POOL_TTL_DAYS = 7 POOL_A_PREFIXES = ("00", "30", "60", "68", "43", "83", "87", "88", "92") # A股含BJ, 排B股/基金/转债 +PDF_PAUSE_FREE_GB = 200 # 容量闸门: 卷空闲低于此 → 暂停 PDF 段(spec §18.3) +PDF_CAPACITY_CHECK_GB = 50 # 每落 50GB 自查一次 df -RATE = {"cninfo": 1.0, "eastmoney": 1.0} # 每域最小间隔(秒), 慢爬纪律 +RATE = {"cninfo": 1.0, "eastmoney": 1.0, "article": 2.0} # 每域最小间隔(秒), 慢爬纪律 JITTER = (0.0, 0.3) ID_KEYS = {"ann_meta": ["announcement_id"], @@ -494,6 +498,55 @@ def _report_year(title, ann_time): return ann_time[:4] if ann_time else None +# ---------- 回补扫描/容量工具(T5) ---------- + +def _disk_free_gb(path): + return shutil.disk_usage(str(path)).free / (1 << 30) + + +def _iter_partitions(domain): + base = CORPUS_ROOT / domain + if not base.exists(): + return [] + return sorted(p for p in base.glob("dt=*/part-0.parquet") if p.is_file()) + + +def _append_pdf_index(rows): + if not rows: + return + idx_path = CORPUS_ROOT / "state" / "ann_pdf_index.parquet" + df_new = pd.DataFrame(rows) + if idx_path.exists(): + existing = pd.read_parquet(idx_path) + have = set(existing["announcement_id"]) + df_new = df_new[~df_new["announcement_id"].isin(have)] + if not df_new.empty: + df_new = pd.concat([existing, df_new], ignore_index=True) + if df_new.empty: + return + tmp = idx_path.with_suffix(".tmp") + df_new.to_parquet(tmp, index=False) + os.replace(tmp, idx_path) + log.info("pdf index +%d", len(rows)) + + +def _load_missing(path_key): + p = CORPUS_ROOT / "state" / path_key + if p.exists(): + try: + return set(json.loads(p.read_text(encoding="utf-8"))) + except ValueError: + log.warning("%s 损坏, 重置", p.name) + return set() + + +def _save_missing(path_key, ids): + p = CORPUS_ROOT / "state" / path_key + tmp = p.with_suffix(".tmp") + tmp.write_text(json.dumps(sorted(ids)), encoding="utf-8") + os.replace(tmp, p) + + # ---------- 运行上下文 ---------- class Ctx: @@ -752,20 +805,7 @@ def _run_pdf_daily(ctx, pool, cl_cn, ledgers, candidates): mark_done(ctx.lane, "pdf", aid) ctx.units += 1 ctx.stop_now() - if index_rows: - idx_path = CORPUS_ROOT / "state" / "ann_pdf_index.parquet" - df_new = pd.DataFrame(index_rows) - if idx_path.exists(): - existing = pd.read_parquet(idx_path) - have = set(existing["announcement_id"]) - df_new = df_new[~df_new["announcement_id"].isin(have)] - if not df_new.empty: - df_new = pd.concat([existing, df_new], ignore_index=True) - if not df_new.empty: - tmp = idx_path.with_suffix(".tmp") - df_new.to_parquet(tmp, index=False) - os.replace(tmp, idx_path) - log.info("pdf index +%d", len(index_rows)) + _append_pdf_index(index_rows) def run_daily(ctx, pool): @@ -773,15 +813,16 @@ def run_daily(ctx, pool): recents = {d: RecentIndex(d) for d in ID_KEYS} cl_cn = DomainClient("cninfo") cl_em = DomainClient("eastmoney") + cl_art = DomainClient("article") 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_em, ledgers, recents, new_articles) + _run_fulltext(ctx, cl_art, ledgers, recents, new_articles) _run_pdf_daily(ctx, pool, cl_cn, ledgers, candidates) for led in ledgers.values(): led.flush() -# ---------- backfill lane(①ann_meta ②news_meta;③④=T5) ---------- +# ---------- backfill lane(①ann_meta ②news_meta ③fulltext ④PDF;T5) ---------- def _run_ann_backfill(ctx, pool, cl_cn, ledgers, recents): """①公告元数据回补 2000→今, unit=(stock,year), 逐年按池序(探测优先:空年=合法)。""" @@ -867,13 +908,150 @@ def _run_news_backfill(ctx, pool, cl_em, ledgers, recents): log.info("backfill news %s: page=%d 硬顶止", code, NEWS_BACKFILL_MAX_PAGE) +def _run_fulltext_backfill(ctx, cl_art, ledgers, recents): + """③新闻全文回补: 扫 news_meta 库存 → 未入账 art 补正文; 账本即进度 + (无 marker 文件); 404 墓碑行(content_text=None)入账不再重试。""" + rows_all = [] + for part in _iter_partitions("news_meta"): + df = pd.read_parquet(part, columns=["art_code", "url", "show_time"]) + rows_all.extend(df.to_dict("records")) + if not rows_all: + return + known = ledgers["news_fulltext"].has_any([str(r["art_code"]) + for r in rows_all]) + todo, seen = [], set() + for r, k in zip(rows_all, known): + art = str(r["art_code"]) + if k or art in seen or not r.get("url"): + continue + seen.add(art) + todo.append(r) + log.info("fulltext backfill: %d 待补 / %d 库存行", len(todo), len(rows_all)) + since_flush = 0 + today = dt.date.today().isoformat() + for r in todo: + if ctx.limit is not None and ctx.units >= ctx.limit: + return + try: + content = html_to_text(fetch_article_html(cl_art, r["url"])) + except DomainCooldown as e: + log.warning("fulltext backfill 域冷却: %s", e) + ctx.rate_limited |= e.rate_limited + ctx.hard_cool |= not e.rate_limited + return + except HttpDeterministicError as e: + log.warning("fulltext %s 确定性失败(墓碑): %s", r["art_code"], e) + content = None + except TransportError as e: + ctx.failed += 1 + log.warning("fulltext %s 失败(下次重试): %s", r["art_code"], e) + continue + append_parquet("news_fulltext", + [{"art_code": str(r["art_code"]), "url": r["url"], + "show_time": r.get("show_time"), + "content_text": content, "fetch_date": today}], + "art_code") + ledgers["news_fulltext"].add([str(r["art_code"])]) + recents["news_fulltext"].add([str(r["art_code"])]) + ctx.units += 1 + since_flush = _maybe_flush(ledgers, since_flush + 1) + ctx.stop_now() + + +def _run_pdf_backfill(ctx, cl_cn): + """④定期+业绩类 PDF 回补(新→旧): ann_meta 库存标题谓词筛六类; 文件在=跳过; + 404 墓碑; 容量闸门=段首+每 50GB 自查, <200G 暂停本段并落告警(其余段不受影响)。""" + free_gb = _disk_free_gb(CORPUS_ROOT) + + def _gate(): + nonlocal free_gb + free_gb = _disk_free_gb(CORPUS_ROOT) + if free_gb >= PDF_PAUSE_FREE_GB: + return False + flag = CORPUS_ROOT / "state" / "capacity_paused.json" + tmp = flag.with_suffix(".tmp") + tmp.write_text(json.dumps( + {"ts": dt.datetime.now().isoformat(), "free_gb": round(free_gb, 1), + "threshold_gb": PDF_PAUSE_FREE_GB}), encoding="utf-8") + os.replace(tmp, flag) + log.error("容量闸门: 卷空闲 %.1fG < %dG, 暂停 PDF 回补段(升级用户重拍)", + free_gb, PDF_PAUSE_FREE_GB) + return True + + if _gate(): + return + missing = _load_missing("pdf_missing.json") + frames = [pd.read_parquet( + p, columns=["announcement_id", "sec_code", "title", "ann_time", + "adjunct_url"]) for p in _iter_partitions("ann_meta")] + if not frames: + return + df = pd.concat(frames, ignore_index=True) + cand = df[df["title"].map(lambda t: _pdf_label(t) is not None)] + cand = cand.sort_values("ann_time", ascending=False, kind="stable") + cand = cand.drop_duplicates("announcement_id") + log.info("pdf backfill: %d 候选 / %d 行库存", len(cand), len(df)) + bytes_since_check = 0 + 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: + break + aid = str(r["announcement_id"]) + if aid in missing: + continue + year = (str(r["ann_time"] or ""))[:4] or today[:4] + dest = CORPUS_ROOT / "ann_pdf" / year / f"{aid}.pdf" + if dest.exists() and dest.stat().st_size > 0: + continue + if bytes_since_check >= (PDF_CAPACITY_CHECK_GB << 30): + if _gate(): + break + bytes_since_check = 0 + url = PDF_BASE + (r["adjunct_url"] or "") + try: + size = download_pdf(cl_cn, url, dest) + except HttpDeterministicError as e: + if e.status == 404: + missing.add(aid) + _save_missing("pdf_missing.json", missing) + log.warning("pdf %s 404 墓碑(不重试)", aid) + continue + ctx.failed += 1 + log.warning("pdf %s 失败: %s", aid, e) + continue + except DomainCooldown as e: + log.warning("pdf backfill 域冷却: %s", e) + ctx.rate_limited |= e.rate_limited + ctx.hard_cool |= not e.rate_limited + break + except TransportError as e: + ctx.failed += 1 + log.warning("pdf %s 失败(下次重试): %s", aid, e) + continue + bytes_since_check += size + index_rows.append({"announcement_id": aid, + "sec_code": r.get("sec_code"), + "ann_type_name": _pdf_label(r["title"]), + "report_year": _report_year(r["title"], + str(r["ann_time"] or "")), + "local_path": str(dest), "bytes": size, + "fetch_date": today}) + ctx.units += 1 + ctx.stop_now() + _append_pdf_index(index_rows) + + def run_backfill(ctx, pool): ledgers = {d: IdLedger(d) for d in ID_KEYS} recents = {d: RecentIndex(d) for d in ID_KEYS} cl_cn = DomainClient("cninfo") cl_em = DomainClient("eastmoney") + cl_art = DomainClient("article") _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) + _run_pdf_backfill(ctx, cl_cn) for led in ledgers.values(): led.flush() diff --git a/tests/data_platform/test_corpus_download.py b/tests/data_platform/test_corpus_download.py index 4861d56..aa15b41 100644 --- a/tests/data_platform/test_corpus_download.py +++ b/tests/data_platform/test_corpus_download.py @@ -27,6 +27,8 @@ def _fast(monkeypatch): @pytest.fixture def root(tmp_path, monkeypatch): monkeypatch.setattr(cd, "CORPUS_ROOT", tmp_path) + # 容量闸门默认放行: 本机磁盘余量与测试无关(曾因 Mac 真实 free<200G 误暂停 ④) + monkeypatch.setattr(cd, "_disk_free_gb", lambda path: 645.0) for sub in ("news_meta", "news_fulltext", "ann_meta", "ann_pdf", "state", "logs"): (tmp_path / sub).mkdir(parents=True, exist_ok=True) return tmp_path @@ -609,10 +611,13 @@ def test_backfill_ann_unit_fail_not_marked(root, monkeypatch): def _wire_news_backfill(monkeypatch, pages): - """pages: {page_idx: rows} — 翻页回放。""" + """pages: {page_idx: rows} — 翻页回放。③④ 段 fetch 一并锁死(零网络铁律)。""" monkeypatch.setattr(cd, "load_stock_pool", lambda **kw: [("600519", "o1")]) monkeypatch.setattr(cd, "fetch_cninfo_announcements", MagicMock(return_value=[])) + monkeypatch.setattr(cd, "fetch_article_html", + MagicMock(return_value="x")) + monkeypatch.setattr(cd, "download_pdf", MagicMock(return_value=12)) urls = [] def fake(client, code, mkt, page): @@ -683,3 +688,176 @@ def test_backfill_wallclock_rc3(root, monkeypatch): rc = cd.run_lane("backfill", until=past) assert rc == 3 assert cd.is_done("backfill", "ann", "000001_2000") # 首 unit 完成后停 + + +# ---------- backfill lane ③news_fulltext ④PDF+容量闸门(T5) ---------- + +def _seed_news_meta(root, rows): + """往今天的 news_meta 分区落行(模拟 ①② 已产出的库存)。""" + import pandas as pd + day = dt.date.today().isoformat() + d = root / "news_meta" / f"dt={day}" + d.mkdir(parents=True, exist_ok=True) + pd.DataFrame(rows).to_parquet(d / "part-0.parquet", index=False) + + +_NEWS_SEED = [ + {"art_code": "f1", "stock_code": "600519", "show_time": "2024-01-01 00:00:00", + "title": "t1", "summary": None, "media_name": None, + "url": "http://finance.eastmoney.com/a/f1.html", "first_seen_date": "2026-09-06"}, + {"art_code": "f2", "stock_code": "600519", "show_time": "2024-01-02 00:00:00", + "title": "t2", "summary": None, "media_name": None, + "url": "http://finance.eastmoney.com/a/f2.html", "first_seen_date": "2026-09-06"}, +] + + +def _wire_backfill_fulltext(monkeypatch, html_by_url): + monkeypatch.setattr(cd, "load_stock_pool", lambda **kw: POOL) + monkeypatch.setattr(cd, "fetch_cninfo_announcements", + MagicMock(return_value=[])) + monkeypatch.setattr(cd, "fetch_news_backfill_page", + MagicMock(return_value=[])) + fetched = [] + + def fake(client, url): + fetched.append(url) + v = html_by_url.get(url, "ok") + if isinstance(v, Exception): + raise v + return v + + monkeypatch.setattr(cd, "fetch_article_html", fake) + return fetched + + +def test_fulltext_backfill_fills_from_news_meta(root, monkeypatch): + import pandas as pd + _seed_news_meta(root, _NEWS_SEED) + fetched = _wire_backfill_fulltext(monkeypatch, {}) + rc = cd.run_lane("backfill") + assert rc == 0 + assert sorted(fetched) == ["http://finance.eastmoney.com/a/f1.html", + "http://finance.eastmoney.com/a/f2.html"] + today = dt.date.today().isoformat() + ft = pd.read_parquet(root / "news_fulltext" / f"dt={today}" / "part-0.parquet") + assert len(ft) == 2 and (ft["content_text"] == "ok").all() + # 幂等: 账本已知 → 重扫零 fetch + fetched2 = _wire_backfill_fulltext(monkeypatch, {}) + cd.run_lane("backfill") + assert fetched2 == [] + + +def test_fulltext_backfill_404_tombstone_no_retry(root, monkeypatch): + import pandas as pd + _seed_news_meta(root, _NEWS_SEED) + fetched = _wire_backfill_fulltext( + monkeypatch, {"http://finance.eastmoney.com/a/f1.html": + cd.HttpDeterministicError(404, "x")}) + cd.run_lane("backfill") + today = dt.date.today().isoformat() + ft = pd.read_parquet(root / "news_fulltext" / f"dt={today}" / "part-0.parquet") + by_id = {r["art_code"]: r for r in ft.to_dict("records")} + assert by_id["f1"]["content_text"] is None # 墓碑行(已处理无正文) + assert by_id["f2"]["content_text"] == "ok" + fetched2 = _wire_backfill_fulltext(monkeypatch, {}) + cd.run_lane("backfill") + assert fetched2 == [] # 墓碑也进账本,不再重试 + + +def _seed_ann_meta(root, rows): + import pandas as pd + day = dt.date.today().isoformat() + d = root / "ann_meta" / f"dt={day}" + d.mkdir(parents=True, exist_ok=True) + pd.DataFrame(rows).to_parquet(d / "part-0.parquet", index=False) + + +def _ann_norm_row(aid, title, ann_time): + return {"announcement_id": aid, "sec_code": "600519", "sec_name": None, + "org_id": None, "title": title, "short_title": None, "content": None, + "ann_time": ann_time, "ann_type": None, "ann_type_name": None, + "column_id": None, "important": None, + "adjunct_url": f"finalpage/2026-04-30/{aid}.PDF", + "adjunct_size": 244, "batch_num": None, + "first_seen_date": dt.date.today().isoformat()} + + +def _wire_pdf_backfill(monkeypatch): + monkeypatch.setattr(cd, "load_stock_pool", lambda **kw: POOL) + monkeypatch.setattr(cd, "fetch_cninfo_announcements", + MagicMock(return_value=[])) + monkeypatch.setattr(cd, "fetch_news_backfill_page", + MagicMock(return_value=[])) + monkeypatch.setattr(cd, "fetch_article_html", MagicMock()) + downloads = [] + + def fake_dl(client, url, dest): + dest.parent.mkdir(parents=True, exist_ok=True) + dest.write_bytes(b"%PDF-fake") + downloads.append((url, str(dest))) + return 12 + + monkeypatch.setattr(cd, "download_pdf", fake_dl) + return downloads + + +def test_pdf_backfill_selects_six_types_new_first(root, monkeypatch): + import pandas as pd + _seed_ann_meta(root, [ + _ann_norm_row("p_new", "2025年年度报告", "2025-04-30 10:00:00"), + _ann_norm_row("p_mid", "2023年度业绩快报", "2023-04-20 10:00:00"), + _ann_norm_row("p_old", "2020年半年度报告", "2020-08-28 10:00:00"), + _ann_norm_row("p_skip", "第三届董事会决议公告", "2024-01-02 10:00:00"), + ]) + downloads = _wire_pdf_backfill(monkeypatch) + rc = cd.run_lane("backfill") + assert rc == 0 + aids = [u.split("/")[-1].replace(".PDF", "") for u, _ in downloads] + assert aids == ["p_new", "p_mid", "p_old"] # 新→旧,非六类不选 + assert (root / "ann_pdf" / "2025" / "p_new.pdf").exists() + idx = pd.read_parquet(root / "state" / "ann_pdf_index.parquet") + assert len(idx) == 3 + assert idx.iloc[0]["announcement_id"] == "p_new" + # 幂等: 文件在=跳过,零重下 + downloads2 = _wire_pdf_backfill(monkeypatch) + cd.run_lane("backfill") + assert downloads2 == [] + + +def test_pdf_backfill_404_tombstone(root, monkeypatch): + _seed_ann_meta(root, [_ann_norm_row("p404", "2024年年度报告", + "2024-04-30 10:00:00")]) + + def boom(client, url, dest): + raise cd.HttpDeterministicError(404, url) + + monkeypatch.setattr(cd, "load_stock_pool", lambda **kw: POOL) + monkeypatch.setattr(cd, "fetch_cninfo_announcements", MagicMock(return_value=[])) + monkeypatch.setattr(cd, "fetch_news_backfill_page", MagicMock(return_value=[])) + monkeypatch.setattr(cd, "fetch_article_html", MagicMock()) + monkeypatch.setattr(cd, "download_pdf", boom) + rc = cd.run_lane("backfill") + assert rc == 0 # missing 不算失败 + downloads = _wire_pdf_backfill(monkeypatch) + cd.run_lane("backfill") + assert downloads == [] # 墓碑不重试 + + +def test_capacity_gate_pauses_pdf_only(root, monkeypatch): + """free < 阈值 → PDF 段暂停(其余段照跑), 留 capacity_paused.json 告警。""" + import pandas as pd + _seed_ann_meta(root, [_ann_norm_row("p1", "2025年年度报告", + "2025-04-30 10:00:00")]) + downloads = _wire_pdf_backfill(monkeypatch) + monkeypatch.setattr(cd, "_disk_free_gb", lambda path: 100.0) + rc = cd.run_lane("backfill") + assert rc == 0 # 暂停≠失败,其余段不受影响 + assert downloads == [] # PDF 未下 + flag = root / "state" / "capacity_paused.json" + assert flag.exists() + assert "free_gb" in json.loads(flag.read_text()) + # 恢复(空间回升) → 复跑即补 + downloads2 = _wire_pdf_backfill(monkeypatch) + monkeypatch.setattr(cd, "_disk_free_gb", lambda path: 645.0) + cd.run_lane("backfill") + assert len(downloads2) == 1