diff --git a/scripts/data_platform/corpus_download.py b/scripts/data_platform/corpus_download.py index 1c80488..6c3375b 100644 --- a/scripts/data_platform/corpus_download.py +++ b/scripts/data_platform/corpus_download.py @@ -536,6 +536,17 @@ def _filter_new(domain, norm_rows, ledger, recent): if not kl and not kr] +def _absorb_new(domain, norm_rows, ledgers, recents): + """过滤已知 id → 追加分区+账本+近窗(daily/backfill 共用)。返回新行。""" + new = _filter_new(domain, norm_rows, ledgers[domain], recents[domain]) + if new: + append_parquet(domain, new, ID_KEYS[domain]) + keys = [_row_key(domain, r) for r in new] + ledgers[domain].add(keys) + recents[domain].add(keys) + return new + + def _record_depth(ctx, source, code, date_str): if not date_str: return @@ -598,12 +609,8 @@ def _run_ann_increment(ctx, pool, cl_cn, ledgers, recents): continue candidates.extend(collect_pdf_candidates(rows)) norm = [norm_ann_row(r, first_seen) for r in rows] - new = _filter_new("ann_meta", norm, ledgers["ann_meta"], recents["ann_meta"]) + new = _absorb_new("ann_meta", norm, ledgers, recents) if new: - append_parquet("ann_meta", new, "announcement_id") - keys = [_row_key("ann_meta", r) for r in new] - ledgers["ann_meta"].add(keys) - recents["ann_meta"].add(keys) for r in new: _record_depth(ctx, "ann", r["sec_code"] or code, (r["ann_time"] or "")[:10]) @@ -635,12 +642,8 @@ def _run_news_increment(ctx, pool, cl_em, ledgers, recents, new_articles): log.warning("news unit %s 失败(不标done): %s", code, e) continue norm = [norm_news_search_row(r, code) for r in rows] - new = _filter_new("news_meta", norm, ledgers["news_meta"], recents["news_meta"]) + new = _absorb_new("news_meta", norm, ledgers, recents) if new: - append_parquet("news_meta", new, ["art_code", "stock_code"]) - keys = [_row_key("news_meta", r) for r in new] - ledgers["news_meta"].add(keys) - recents["news_meta"].add(keys) for r in new: _record_depth(ctx, "news", code, (r["show_time"] or "")[:10]) new_articles.append((r["art_code"], r["url"], r["show_time"])) @@ -778,11 +781,101 @@ def run_daily(ctx, pool): led.flush() -# ---------- backfill lane(T2 占位, ①②=T4 ③④=T5) ---------- +# ---------- backfill lane(①ann_meta ②news_meta;③④=T5) ---------- + +def _run_ann_backfill(ctx, pool, cl_cn, ledgers, recents): + """①公告元数据回补 2000→今, unit=(stock,year), 逐年按池序(探测优先:空年=合法)。""" + today = dt.date.today() + first_seen = today.isoformat() + since_flush = 0 + for year in range(BACKFILL_FROM_YEAR, today.year + 1): + 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: + return + if is_done(ctx.lane, "ann", unit): + continue + try: + rows = fetch_cninfo_announcements(cl_cn, code, org, + f"{year}-01-01", year_end) + except DomainCooldown as e: + log.warning("backfill ann 域冷却 @%s: %s", unit, e) + ctx.rate_limited |= e.rate_limited + ctx.hard_cool |= not e.rate_limited + return + except (TransportError, HttpDeterministicError) as e: + ctx.failed += 1 + log.warning("backfill ann %s 失败(不标done): %s", unit, e) + continue + norm = [norm_ann_row(r, first_seen) for r in rows] + new = _absorb_new("ann_meta", norm, ledgers, recents) + for r in new: + _record_depth(ctx, "ann", r["sec_code"] or code, + (r["ann_time"] or "")[:10]) + if new: + log.info("backfill ann %s: +%d/%d", unit, len(new), len(rows)) + if ctx.limit is None: + mark_done(ctx.lane, "ann", unit) + ctx.units += 1 + since_flush = _maybe_flush(ledgers, since_flush + 1) + ctx.stop_now() + + +def _run_news_backfill(ctx, pool, cl_em, ledgers, recents): + """②新闻标题级回补(np-listapi), unit=(stock,page), 翻到空页或 page≥200 止。""" + for code, _org in pool: + if is_done(ctx.lane, "news_end", code): + continue + mkt = "1" if code.startswith("6") else "0" + 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: + return + if is_done(ctx.lane, "news", unit): + continue + try: + rows = fetch_news_backfill_page(cl_em, code, mkt, page) + except DomainCooldown as e: + log.warning("backfill news 域冷却 @%s: %s", unit, e) + ctx.rate_limited |= e.rate_limited + ctx.hard_cool |= not e.rate_limited + return + except (TransportError, HttpDeterministicError) as e: + ctx.failed += 1 + log.warning("backfill news %s 失败(断点留此页): %s", unit, e) + break + if not rows: + if ctx.limit is None: + mark_done(ctx.lane, "news", unit) + mark_done(ctx.lane, "news_end", code) + log.info("backfill news %s: 到头(空页@%d)", code, page) + break + norm = [norm_news_np_row(r, code) for r in rows] + new = _absorb_new("news_meta", norm, ledgers, recents) + for r in new: + _record_depth(ctx, "news", code, (r["show_time"] or "")[:10]) + if ctx.limit is None: + mark_done(ctx.lane, "news", unit) + ctx.units += 1 + if page == NEWS_BACKFILL_MAX_PAGE: + capped = True + ctx.stop_now() + if capped and ctx.limit is None: + mark_done(ctx.lane, "news_end", code) + log.info("backfill news %s: page=%d 硬顶止", code, NEWS_BACKFILL_MAX_PAGE) + def run_backfill(ctx, pool): - log.warning("backfill lane 未实施(T4/T5): 让路退出, daily 数据不受影响") - return + 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") + _run_ann_backfill(ctx, pool, cl_cn, ledgers, recents) + _run_news_backfill(ctx, pool, cl_em, ledgers, recents) + for led in ledgers.values(): + led.flush() # ---------- lane 入口 ---------- diff --git a/tests/data_platform/test_corpus_download.py b/tests/data_platform/test_corpus_download.py index 2896654..4861d56 100644 --- a/tests/data_platform/test_corpus_download.py +++ b/tests/data_platform/test_corpus_download.py @@ -532,3 +532,154 @@ def test_main_daily_exit0(root, monkeypatch): with pytest.raises(SystemExit) as e: cd.main() assert e.value.code == 0 + + +# ---------- backfill lane ①ann_meta ②news_meta(T4) ---------- + +def _wire_backfill_ann(monkeypatch, rows_by_unit): + """rows_by_unit: {(code, start, end): rows} — 按 unit 窗口回放。 + 必须同时锁死 news 段 fetch(零网络铁律——曾漏锁致真打 np-listapi 40s)。""" + monkeypatch.setattr(cd, "load_stock_pool", lambda **kw: POOL) + monkeypatch.setattr(cd, "fetch_news_backfill_page", MagicMock(return_value=[])) + calls = [] + + def fake(client, code, org, start, end, category=""): + calls.append((code, start, end)) + return rows_by_unit.get((code, start, end), []) + + monkeypatch.setattr(cd, "fetch_cninfo_announcements", fake) + return calls + + +def test_backfill_ann_units_walk_years_asc_from_2000(root, monkeypatch): + """unit=(stock,year), 2000→今年逐年, 每年内按池序; 空年=合法 done。""" + import pandas as pd + calls = _wire_backfill_ann(monkeypatch, {("000001", "2001-01-01", "2001-12-31"): + [_ann_raw(aid="b1")]}) + rc = cd.run_lane("backfill") + assert rc == 0 + t = dt.date.today() + n_years = t.year - 2000 + 1 + assert len(calls) == n_years * 2 # 每股每年 1 req(空窗 1 页即返) + assert calls[0] == ("000001", "2000-01-01", "2000-12-31") + assert calls[-1][1].startswith(t.isoformat()[:4]) # 今年 + # 今年窗口右端=今天(相对), 历年=12-31 + assert calls[-1][2] == t.isoformat() + assert cd.is_done("backfill", "ann", "000001_2001") + assert cd.is_done("backfill", "ann", "000001_2000") # 空年也 done + today = t.isoformat() + ann = pd.read_parquet(root / "ann_meta" / f"dt={today}" / "part-0.parquet") + assert len(ann) == 1 and ann.iloc[0]["announcement_id"] == "b1" + + +def test_backfill_ann_marker_skip_on_rerun(root, monkeypatch): + """done unit 零重拉(断点续传实证之 mock 版);数据幂等零重复。""" + import pandas as pd + _wire_backfill_ann(monkeypatch, {("000001", "2001-01-01", "2001-12-31"): + [_ann_raw(aid="b1")]}) + cd.run_lane("backfill") + calls2 = _wire_backfill_ann(monkeypatch, {("000001", "2001-01-01", "2001-12-31"): + [_ann_raw(aid="b1")]}) + cd.run_lane("backfill") + assert calls2 == [] # 全部 done → 零请求 + today = dt.date.today().isoformat() + ann = pd.read_parquet(root / "ann_meta" / f"dt={today}" / "part-0.parquet") + assert len(ann) == 1 + + +def test_backfill_ann_resume_from_oldest_undone(root, monkeypatch): + """手动标 2000/2001 done → 重跑只碰 2002 起。""" + cd.mark_done("backfill", "ann", "000001_2000") + cd.mark_done("backfill", "ann", "000001_2001") + cd.mark_done("backfill", "ann", "600519_2000") + cd.mark_done("backfill", "ann", "600519_2001") + calls = _wire_backfill_ann(monkeypatch, {}) + cd.run_lane("backfill") + starts = {c[1] for c in calls} + assert "2000-01-01" not in starts and "2001-01-01" not in starts + + +def test_backfill_ann_unit_fail_not_marked(root, monkeypatch): + calls = _wire_backfill_ann(monkeypatch, {}) + monkeypatch.setattr(cd, "fetch_cninfo_announcements", + MagicMock(side_effect=cd.TransportError("boom"))) + rc = cd.run_lane("backfill") + assert rc == 1 + assert not cd.is_done("backfill", "ann", "000001_2000") + + +def _wire_news_backfill(monkeypatch, pages): + """pages: {page_idx: rows} — 翻页回放。""" + monkeypatch.setattr(cd, "load_stock_pool", lambda **kw: [("600519", "o1")]) + monkeypatch.setattr(cd, "fetch_cninfo_announcements", + MagicMock(return_value=[])) + urls = [] + + def fake(client, code, mkt, page): + urls.append((code, mkt, page)) + return pages.get(page, []) + + monkeypatch.setattr(cd, "fetch_news_backfill_page", fake) + return urls + + +def test_backfill_news_walks_pages_until_empty(root, monkeypatch): + import pandas as pd + np_rows = [{"Art_Code": f"a{i}", "Art_ShowTime": f"2024-01-0{i} 00:00:00", + "Art_Title": f"t{i}", "Art_Url": f"http://x/a{i}.html", + "Np_dst": "CMS"} for i in (1, 2, 3)] + urls = _wire_news_backfill(monkeypatch, {1: np_rows[:2], 2: np_rows[2:]}) + rc = cd.run_lane("backfill") + assert rc == 0 + assert urls[:2] == [("600519", "1", 1), ("600519", "1", 2)] + assert urls[2] == ("600519", "1", 3) # 第3页空 → 止 + assert len(urls) == 3 + assert cd.is_done("backfill", "news", "600519_p1") + assert cd.is_done("backfill", "news_end", "600519") # 空页=股票到头 + today = dt.date.today().isoformat() + news = pd.read_parquet(root / "news_meta" / f"dt={today}" / "part-0.parquet") + assert len(news) == 3 + assert news.iloc[0]["summary"] is None # np 段无摘要如实 + + +def test_backfill_news_end_sentinel_skips_stock(root, monkeypatch): + cd.mark_done("backfill", "news_end", "600519") + urls = _wire_news_backfill(monkeypatch, {1: ["x"]}) + cd.run_lane("backfill") + assert urls == [] # 到头股零请求 + + +def test_backfill_news_page_cap_200(root, monkeypatch): + """page≥200 止(实测硬顶), 到 200 即算该股完结。""" + row = {"Art_Code": "a", "Art_ShowTime": "2020-01-01 00:00:00", + "Art_Title": "t", "Art_Url": "http://x/a.html", "Np_dst": "CMS"} + urls = _wire_news_backfill(monkeypatch, + {p: [dict(row)] for p in range(1, 201)}) + cd.run_lane("backfill") + assert len(urls) == 200 + assert cd.is_done("backfill", "news_end", "600519") + + +def test_backfill_news_mkt_prefix_mapping(root, monkeypatch): + urls = _wire_news_backfill(monkeypatch, {}) + cd.run_lane("backfill") + assert urls[0][1] == "1" # 600519 → SH=1 + urls2 = _wire_news_backfill(monkeypatch, {}) # 只换 fetch 窗口 + monkeypatch.setattr(cd, "load_stock_pool", lambda **kw: [("000001", "o")]) + cd.run_lane("backfill") + assert urls2[0][1] == "0" # 000001 → SZ=0 + + +def test_backfill_limit_smoke_no_markers(root, monkeypatch): + _wire_backfill_ann(monkeypatch, {}) + rc = cd.run_lane("backfill", limit=3) + assert rc == 0 + assert not any((root / "state" / "markers" / "backfill").rglob("*.done")) + + +def test_backfill_wallclock_rc3(root, monkeypatch): + _wire_backfill_ann(monkeypatch, {}) + past = (dt.datetime.now() - dt.timedelta(minutes=1)).strftime("%H:%M") + rc = cd.run_lane("backfill", until=past) + assert rc == 3 + assert cd.is_done("backfill", "ann", "000001_2000") # 首 unit 完成后停