sanguo/sanguo_vnpy_v2
3
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(data): T5 backfill ③全文 ④PDF+容量闸门 [nas]
CI/CD / test (push) Successful in 28s
Details
CI/CD / nas-deploy (push) Successful in 2s
Details
CI/CD / nas-verify (push) Successful in 6s
Details

Browse Source 
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 绿
This commit is contained in:
Claude Dev
2026-09-06 23:24:29 +08:00
parent 10c486c64d
commit 1c3a61eb12
2 changed files with 375 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/data_platform/corpus_download.py
+196 -18
View File
@@ -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()