From 4881d100714bfff1324919532482cba09d90cbcc Mon Sep 17 00:00:00 2001 From: claude_dev Date: Tue, 22 Sep 2026 08:17:14 +0800 Subject: [PATCH] =?UTF-8?q?feat(data=5Fplatform):=20corpus=20flash=5Fmeta?= =?UTF-8?q?=20=E5=85=A8=E5=B8=82=E5=9C=BA=E5=BF=AB=E8=AE=AF=E5=9F=9F?= =?UTF-8?q?=E4=B8=8A=E7=BA=BF(spec=20=C2=A720.12.3=20=E8=A1=A5=E5=85=85?= =?UTF-8?q?=E7=B1=BB=E9=87=87=E7=BA=B3)=E2=80=94=E2=80=94=E9=87=91?= =?UTF-8?q?=E5=8D=81=20flash=5Fnewest.js=20=E5=BF=AB=E7=85=A7+=E4=B8=9C?= =?UTF-8?q?=E8=B4=A2=20kuaixun=20=E7=BF=BB=E9=A1=B5(T-1=20=E8=BE=B9?= =?UTF-8?q?=E7=95=8C=E5=81=9C=E7=95=8C,20=E9=A1=B5=E7=A1=AC=E9=A1=B6)?= =?UTF-8?q?=E4=B8=A4=E6=BA=90,=E4=B8=8E=20news=5Fmeta=20=E4=B8=AA=E8=82=A1?= =?UTF-8?q?=E6=96=B0=E9=97=BB=E8=AF=AD=E4=B9=89=E5=B9=B6=E5=88=97=E4=B8=8D?= =?UTF-8?q?=20join;flash=5Fid=3D=E6=BA=90=E5=89=8D=E7=BC=80+=E6=BA=90?= =?UTF-8?q?=E7=AB=99=E7=A8=B3=E5=AE=9Aid=E8=B7=A8=E6=BA=90=E5=91=BD?= =?UTF-8?q?=E5=90=8D=E7=A9=BA=E9=97=B4=E9=9A=94=E7=A6=BB,=E5=BD=93?= =?UTF-8?q?=E6=97=A5marker=E5=B9=82=E7=AD=89,=E6=BA=90=E7=BA=A7=E9=9A=94?= =?UTF-8?q?=E7=A6=BB=E5=8D=95=E6=BA=90=E5=A4=B1=E8=B4=A5=E4=B8=8D=E7=82=B8?= =?UTF-8?q?=E5=8F=A6=E4=B8=80=E6=BA=90;run=5Fcorpus.sh=20=E7=AC=AC?= =?UTF-8?q?=E4=B8=89lane=20rc=E9=9A=94=E7=A6=BB=E4=B8=8D=E8=BF=9B=E8=81=9A?= =?UTF-8?q?=E5=90=88(=E4=B8=A4=E4=B8=BBlane=E5=85=A8=E8=AE=A9=E8=B7=AF?= =?UTF-8?q?=E4=B9=9F=E4=B8=8D=E6=BC=8F=E5=BD=93=E5=A4=A9=E5=BF=AB=E8=AE=AF?= =?UTF-8?q?);=E9=95=9C=E5=83=8F=E8=BD=A6=20corpus=20=E5=9B=9B=E5=AD=90?= =?UTF-8?q?=E5=9F=9F=E6=89=A9=E4=BA=94(=E6=96=B0=E5=9F=9F=E6=8C=87?= =?UTF-8?q?=E7=BA=B9=E5=85=A8=E6=96=B0=E9=A6=96=E8=B7=91=E8=87=AA=E5=8A=A8?= =?UTF-8?q?=E5=85=A8=E9=87=8F=E6=8E=A8);corpus=5Freader.get=5Fflash=20?= =?UTF-8?q?=E7=A9=BAdf=E5=B8=A6schema;TDD=20+13=E4=BE=8B=E5=85=A8=E5=A5=97?= =?UTF-8?q?465=E7=BB=BF;Mac=E7=9C=9F=E7=BD=91=E7=BB=9C=E5=86=92=E7=83=9F?= =?UTF-8?q?=3D=E9=87=91=E5=8D=8150+=E4=B8=9C=E8=B4=A2884=E6=96=B0,21s?= =?UTF-8?q?=E5=AE=8C=E8=B5=9B,dt=E8=B7=A8=E5=BA=A6T-1=E8=BE=B9=E7=95=8C?= =?UTF-8?q?=E5=AE=9E=E8=AF=81,=E5=B9=82=E7=AD=89=E5=A4=8D=E8=B7=91?= =?UTF-8?q?=E9=9B=B6=E8=AF=B7=E6=B1=82;=E9=A6=96=E7=8F=AD=E9=94=9A?= =?UTF-8?q?=E7=82=B9=3D=E6=AC=A1=E6=97=A506:0x=20flash=20lane=20exit=3D0?= =?UTF-8?q?=E8=A1=8C=20[nas]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../2026-07-21-data-source-fusion-design.md | 3 +- sanguo_data/corpus_reader.py | 18 ++ scripts/data_platform/corpus_download.py | 131 +++++++++++++- scripts/nas_sync/run_corpus_standalone.sh | 11 ++ scripts/nas_sync/vps_mirror_lib.py | 3 +- tests/data_platform/test_corpus_download.py | 171 ++++++++++++++++++ tests/data_platform/test_corpus_reader.py | 41 +++++ tests/data_platform/test_vps_mirror_lib.py | 17 ++ 8 files changed, 389 insertions(+), 6 deletions(-) diff --git a/docs/superpowers/specs/2026-07-21-data-source-fusion-design.md b/docs/superpowers/specs/2026-07-21-data-source-fusion-design.md index fdf8ea94..1078877b 100644 --- a/docs/superpowers/specs/2026-07-21-data-source-fusion-design.md +++ b/docs/superpowers/specs/2026-07-21-data-source-fusion-design.md @@ -1639,7 +1639,8 @@ schema 特殊低估交集教训）；限速纪律沿 §19.9 采集纪律表；VP - **语义定位**：现 corpus news=news_meta/news_fulltext 为**个股新闻**（东财个股面）；金十+东财快讯=**全市场宏观/政策/突发流**（财联社类）——语义互补非重叠。**东财公告 np-anotice 不入此类**（ann_meta 已采公告元数据，它是对照源，见 §20.12.4）。 - **容量评估（2026-09-22 实测底数）**：VPS C 盘剩 68.0G/总 193.2G；VPS corpus+dmsk 镜像现 264M；NAS corpus 权威 415M（news_meta 121M/ann_meta 119M/news_fulltext 20M/ann_pdf 1.2M/state 173M）。快讯增量估 <1M/日（金十 ~70 条快照+东财 50×N 翻页，meta 级）→ **年 <0.4G/机，双机 <1G vs 剩余 68G=无压力**。 -- **设计要点（实施时 TDD 细化）**：新子域 `corpus/flash_meta/dt=YYYY-MM-DD/`（title/content/publish_time/source∈{jin10,em_kuaixun}/dedup 键=id/newsid）；采集挂 run_corpus.sh 班内（06:00 顺带，两源独立 try/except 互不影响主 rc）；镜像车 corpus 四子域扩为五子域（journal 指纹 diff 同机制）；消费接口 corpus_reader 加 flash 域入口（空 df 带 schema）。**实施待用户点开工，本节先定设计。** +- **设计要点（实施时 TDD 细化）**：新子域 `corpus/flash_meta/dt=YYYY-MM-DD/`（title/content/publish_time/source∈{jin10,em_kuaixun}/dedup 键=id/newsid）；采集挂 run_corpus.sh 班内（06:00 顺带，两源独立 try/except 互不影响主 rc）；镜像车 corpus 四子域扩为五子域（journal 指纹 diff 同机制）；消费接口 corpus_reader 加 flash 域入口（空 df 带 schema）。 +- **实施记录（09-22 当日 TDD 落地）**：`corpus_download.py --lane flash`（金十 `_parse_jin10_payload` JS 变量抽取+东财 `_parse_em_kuaixun_payload` JSONP；`norm_flash_row` 归一 flash_id=源前缀+源站 id；`fetch_em_kuaixun` 翻页 20 页硬顶+页内最老条目<昨日 00:00 停界；当日 marker 幂等；源级隔离单源失败不炸另一源；rc 语义照旧 failed>0=1）；run_corpus_standalone.sh 第三 lane（**rc 隔离不进聚合**，独立跑不 gate 于 daily/backfill——两主 lane 全让路也不漏当天快讯）；vps_mirror_lib FAMILIES corpus 五子域（新域=指纹全新首跑自动全量推一次）；corpus_reader.get_flash(start,end,sources)+FLASH_COLUMNS 空 schema 兜底。TDD +13 例（parse 垃圾兜底/norm title-null 截 80/lane 落盘/账本去重复跑/marker 当日跳过/源隔离/翻页停界×2/reader 读+空 schema+日期窗/mirror 子域+指纹），全套 465 绿。**Mac 真网络冒烟**：金十 50 条+东财 884 新（跨页 dup 16 批内去重拦截），21s 完赛 1.2s 域节奏，dt 跨度=昨日 20:24→今晨（T-1 边界实证）；幂等复跑 units=0 零请求。**首班锚点=次日 06:0x cron.log「flash lane exit=0(隔离)」行+当晚 push.log corpus 段含 flash_meta 分区**。 #### 20.12.4 对照类裁决：跨 vendor 仲裁探针工具箱（不入采集管线、不落库、按需现拉） diff --git a/sanguo_data/corpus_reader.py b/sanguo_data/corpus_reader.py index 49813769..b521cfd6 100644 --- a/sanguo_data/corpus_reader.py +++ b/sanguo_data/corpus_reader.py @@ -9,6 +9,9 @@ schema 契约（2026-09-17 NAS 实测落档，§20.11.3 数据契约③）： sec_code(裸 6 位)/sec_name/title/short_title/ann_time/ann_type(|| 分隔多码)/ ann_type_name/column_id/adjunct_url/adjunct_size/first_seen_date - ann_pdf: /ann_pdf//.pdf +- flash_meta(§20.12.3, 2026-09-22): flash_id(jin10:/em_kuaixun: 前缀+源站稳定 id)/ + source/publish_time/title/content/important/channel/first_seen_date—— + 全市场宏观快讯流(金十+东财kuaixun), 与 news_meta 个股新闻语义并列不 join 稳定 id（数据契约②配方）：art_code/announcement_id 为源站稳定 id；爬虫层 跨天重复行会在数据里原样出现——核内零去重，消费方按 id 自行幂等。 @@ -48,6 +51,21 @@ def ann_pdf_dir(year): return os.path.join(pr.corpus_root(), "ann_pdf", str(year)) +FLASH_COLUMNS = ["flash_id", "source", "publish_time", "title", "content", + "important", "channel", "first_seen_date"] + + +def get_flash(start=None, end=None, sources=None): + """全市场快讯流（§20.12.3，dt 分区直读）。sources={"jin10","em_kuaixun"} + 过滤；空树返带 schema 的空 df（容错分级契约）。""" + df = pr.read_partitions(pr.corpus_root(), "flash_meta", start, end) + if df.empty: + return df.reindex(columns=FLASH_COLUMNS) + if sources: + df = df[df["source"].isin(set(sources))] + return df + + def get_watermark(): """镜像水位线（None=尚无成功推送记录）。""" return pr.read_watermark(pr.corpus_root()) diff --git a/scripts/data_platform/corpus_download.py b/scripts/data_platform/corpus_download.py index c8d47c19..f2a3227b 100644 --- a/scripts/data_platform/corpus_download.py +++ b/scripts/data_platform/corpus_download.py @@ -7,6 +7,8 @@ 新闻全文(昨日新 art_code)/当日 PDF(五类服务端码+业绩快报标题谓词) --lane backfill ①ann_meta 2000起 ②news_meta np-listapi ③新闻全文(账本即进度, 404墓碑) ④PDF六类新→旧(容量闸门<200G暂停段) + --lane flash 全市场快讯域 flash_meta(§20.12.3): 金十最新快照~70条 + 东财 + kuaixun 翻页到昨日 00:00 边界; 快照型不深回补, 当日 marker 幂等 --until HH:MM 墙钟自停: 完成当前 unit 后 checkpoint 退出(rc=3) --limit N 冒烟: 每段最多 N unit, 绝不落 marker(5m 09-17 教训) @@ -58,6 +60,9 @@ PDF_BASE = "http://static.cninfo.com.cn/" SEARCH_API_URL = "https://search-api-web.eastmoney.com/search/jsonp" NP_LIST_URL = ("https://np-listapi.eastmoney.com/comm/web/getListInfo" "?client=web&mTypeAndCode={mkt}.{code}&type=1&pageSize=100&pageIndex={page}") +JIN10_URL = "https://www.jin10.com/flash_newest.js" +EM_KUAIXUN_URL = ("https://newsapi.eastmoney.com/kuaixun/v1/getlist_102_" + "ajaxResult_50_{page}_.html") PAGE_SIZE = 30 # 巨潮服务端封顶 FIVE_CAT = ";".join([ @@ -76,14 +81,17 @@ POOL_A_PREFIXES = ("00", "30", "60", "68", "43", "83", "87", "88", "92") # A股 PDF_PAUSE_FREE_GB = 200 # 容量闸门: 卷空闲低于此 → 暂停 PDF 段(spec §18.3) PDF_CAPACITY_CHECK_GB = 50 # 每落 50GB 自查一次 df -RATE = {"cninfo": 1.0, "eastmoney": 1.0, "article": 2.0} # 每域最小间隔(秒), 慢爬纪律 +RATE = {"cninfo": 1.0, "eastmoney": 1.0, "article": 2.0, "jin10": 1.0} # 每域最小间隔(秒), 慢爬纪律 ANN_RETRY_WAIT_S = 600.0 # 巨潮 504 尾重试前等待(网关抖动恢复窗; 测试归零) JITTER = (0.0, 0.3) UA = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"} ID_KEYS = {"ann_meta": ["announcement_id"], "news_meta": ["art_code", "stock_code"], - "news_fulltext": ["art_code"]} + "news_fulltext": ["art_code"], + "flash_meta": ["flash_id"]} +FLASH_EM_MAX_PAGES = 20 # 快照型时效数据: 50条/页×20≈2-3天窗, 不深回补 +FLASH_LOOKBACK_DAYS = 1 # 东财快讯翻页拉到昨日 00:00 边界即停 log = logging.getLogger("corpus") @@ -519,6 +527,119 @@ def download_pdf(client, url, dest): return len(r.content) +# ---------- flash lane(spec §20.12.3: 全市场快讯域, 金十+东财快讯) ---------- + +def _parse_jin10_payload(text): + """金十 flash_newest.js: `var newest = [...]` JS 变量 → raw item 列表。""" + m = re.search(r"=\s*(\[.*\])\s*;?\s*$", (text or "").strip(), re.S) + if not m: + return [] + try: + items = json.loads(m.group(1)) + except ValueError: + return [] + return items if isinstance(items, list) else [] + + +def _parse_em_kuaixun_payload(text): + """东财 kuaixun: `var ajaxResult={"LivesList":[...]}` JSONP → raw item 列表。""" + m = re.search(r"=\s*(\{.*\})\s*;?\s*$", (text or "").strip(), re.S) + if not m: + return [] + try: + j = json.loads(m.group(1)) + except ValueError: + return [] + return j.get("LivesList") or [] + + +def norm_flash_row(source, raw): + """快讯归一(§20.12.3 schema): flash_id=源前缀+源站稳定 id, 跨源命名空间隔离。""" + if source == "jin10": + data = raw.get("data") or {} + content = _strip_em(data.get("content") or "") or None + title = _strip_em(data.get("title") or "") or None + if title is None and content: + title = content[:80] # 金十 title 常为 null, content 截 80 兜底 + channel = "|".join(str(c) for c in (raw.get("channel") or [])) or None + return { + "flash_id": "jin10:" + str(raw.get("id") or ""), + "source": source, + "publish_time": raw.get("time"), + "title": title, + "content": content, + "important": raw.get("important"), + "channel": channel, + "first_seen_date": dt.date.today().isoformat(), + } + digest = _strip_em(raw.get("digest") or "") or None + return { + "flash_id": "em_kuaixun:" + str(raw.get("newsid") or raw.get("id") or ""), + "source": source, + "publish_time": raw.get("showtime"), + "title": _strip_em(raw.get("title") or "") or None, + "content": digest, + "important": None, + "channel": None, + "first_seen_date": dt.date.today().isoformat(), + } + + +def fetch_jin10_flash(client): + r = client.request("GET", JIN10_URL, + headers={"User-Agent": UA["User-Agent"], + "Referer": "https://www.jin10.com/"}) + return _parse_jin10_payload(r.text) + + +def fetch_em_kuaixun(client, until_str): + """翻页拉东财快讯(新→旧), 页内最老条目 < until_str 即停(时效数据不深回补)。""" + out = [] + for page in range(1, FLASH_EM_MAX_PAGES + 1): + r = client.request("GET", EM_KUAIXUN_URL.format(page=page), + headers={"User-Agent": UA["User-Agent"], + "Referer": "https://kuaixun.eastmoney.com/"}) + rows = _parse_em_kuaixun_payload(r.text) + if not rows: + break + out.extend(rows) + oldest = min((it.get("showtime") or "9999") for it in rows) + if oldest < until_str: + break + return out + + +def run_flash(ctx): + """快照型两源(§20.12.3): 当日 marker 幂等; 源级隔离(单源失败不炸另一源)。""" + ledgers = {d: IdLedger(d) for d in ("flash_meta",)} + recents = {d: RecentIndex(d) for d in ("flash_meta",)} + cl_j = DomainClient("jin10") + cl_e = DomainClient("eastmoney") + ctx.clients = {"jin10": cl_j, "eastmoney": cl_e} + ctx.set_stores(ledgers) + _reconcile_ledgers(ledgers, recents) + today = dt.date.today().isoformat() + until = dt.datetime.combine( + dt.date.today() - dt.timedelta(days=FLASH_LOOKBACK_DAYS), + dt.time.min).strftime("%Y-%m-%d %H:%M:%S") + for stage, fetch in ( + ("jin10", lambda: fetch_jin10_flash(cl_j)), + ("em_kuaixun", lambda: fetch_em_kuaixun(cl_e, until)), + ): + if is_done(ctx.lane, stage, today): + continue + try: + raw = fetch() + except Exception as e: # TransportError/DomainCooldown/解析错 + log.warning("flash %s fail: %s", stage, e) + ctx.failed += 1 + continue + rows = [norm_flash_row(stage, it) for it in raw if it.get("id") or it.get("newsid")] + new = _absorb_new("flash_meta", rows, ledgers, recents) + log.info("flash %s: fetched=%d new=%d", stage, len(rows), len(new)) + ctx.unit_done(stage, today) + + # ---------- PDF 谓词(六类=五服务端码+业绩快报标题) ---------- _PDF_LABELS = [(EXPRESS_KEYWORD, "业绩快报"), ("业绩预告", "业绩预告"), @@ -1257,7 +1378,7 @@ def run_lane(lane, until=None, limit=None): _DAY_BUFFERS.clear() _ensure_log() ctx = Ctx(lane=lane, until=until, limit=limit) - pool = load_stock_pool() + pool = [] if lane == "flash" else load_stock_pool() log.info("lane=%s start: %d stocks, until=%s, limit=%s", lane, len(pool), until, limit) t0 = time.monotonic() @@ -1266,6 +1387,8 @@ def run_lane(lane, until=None, limit=None): run_daily(ctx, pool) elif lane == "backfill": run_backfill(ctx, pool) + elif lane == "flash": + run_flash(ctx) else: log.error("unknown lane: %s", lane) ctx.failed += 1 @@ -1289,7 +1412,7 @@ def run_lane(lane, until=None, limit=None): def main(): ap = argparse.ArgumentParser(description="NAS 基本面语料库采集(spec §18)") - ap.add_argument("--lane", required=True, choices=["daily", "backfill"]) + ap.add_argument("--lane", required=True, choices=["daily", "backfill", "flash"]) ap.add_argument("--until", default=None, help="HH:MM 墙钟自停") ap.add_argument("--limit", type=int, default=None, help="冒烟: 每 stage 最多 N unit") ap.add_argument("--refresh-pool", action="store_true") diff --git a/scripts/nas_sync/run_corpus_standalone.sh b/scripts/nas_sync/run_corpus_standalone.sh index ca04cca6..e1818bd4 100644 --- a/scripts/nas_sync/run_corpus_standalone.sh +++ b/scripts/nas_sync/run_corpus_standalone.sh @@ -70,6 +70,17 @@ GID_ADMINS=101 # administrators(Synology ACL 授权组,金丝雀实测必带) else echo "=== daily 非零($rc_daily), 跳过 backfill lane(次日自愈) ===" fi + # ===== flash lane(spec §20.12.3, 2026-09-22: 全市场快讯域, 金十+东财kuaixun) ===== + # 秒级完赛; rc 隔离不进聚合(快讯丢一天无实害, 次日快照自然覆盖近期); + # 独立跑不 gate 于 daily/backfill rc——两主 lane 全让路也不该漏当天快讯。 + "$DOCKER" run --rm --name sanguo-corpus --user "${UID_ADMIN}:${GID_ADMIN}" \ + --group-add "${GID_ADMINS}" --no-healthcheck --entrypoint python \ + -v /volume1/stock:/volume1/stock \ + -v "$APP":/app:ro \ + sanguo_vnpy_v2:lock-aligned \ + /app/scripts/data_platform/corpus_download.py --lane flash + rc_flash=$? + echo "=== $(date '+%F %T') corpus flash lane exit=$rc_flash(隔离,不进聚合) ===" } >> "$LOG" 2>&1 # ===== VPS 镜像推送(2026-09-18 用户确认 DSM 复用;spec §20.11.2,§20.11.6 遗留①闭环) ===== # 挂 corpus 完赛尾部=完赛即推(~10:1x,同日新鲜度);dmsk 五域 07:15 已完,指纹 diff diff --git a/scripts/nas_sync/vps_mirror_lib.py b/scripts/nas_sync/vps_mirror_lib.py index e56239be..38c13bea 100644 --- a/scripts/nas_sync/vps_mirror_lib.py +++ b/scripts/nas_sync/vps_mirror_lib.py @@ -28,7 +28,8 @@ import sys FAMILIES = { "corpus": { - "domains": ["news_meta", "news_fulltext", "ann_meta", "ann_pdf"], + "domains": ["news_meta", "news_fulltext", "ann_meta", "ann_pdf", + "flash_meta"], # §20.12.3 快讯域(09-22): 新域=指纹全新, 首跑自动全量推一次 "local_root": "/volume1/stock/corpus", "remote_root": "C:/sanguo_vnpy_v2/data/corpus", }, diff --git a/tests/data_platform/test_corpus_download.py b/tests/data_platform/test_corpus_download.py index 91a156d9..d62f29bf 100644 --- a/tests/data_platform/test_corpus_download.py +++ b/tests/data_platform/test_corpus_download.py @@ -1074,3 +1074,174 @@ def test_capacity_gate_pauses_pdf_only(root, monkeypatch): monkeypatch.setattr(cd, "_disk_free_gb", lambda path: 645.0) cd.run_lane("backfill") assert len(downloads2) == 1 + + +# ---------- flash lane(spec §20.12.3, 2026-09-22: 全市场快讯域, 金十+东财快讯) ---------- + +def _jin10_item(fid, when_str, content="央行开展逆回购操作...", important=0, title=None): + return {"id": fid, "time": when_str, "type": 0, + "data": {"title": title, "content": content}, + "important": important, "tags": [], "channel": [5, 7], "remark": []} + + +def _em_item(nid, when_str, title="东财快讯标题", digest="快讯摘要内容"): + return {"sort": "1", "id": nid, "newsid": nid, "title": title, + "digest": digest, "showtime": when_str, + "url_w": "http://finance.eastmoney.com/a/x.html"} + + +JIN10_JS_TMPL = "var newest = %s;" +EM_JSONP_TMPL = 'var ajaxResult={"rc":1,"me":"","LivesList":%s};' + + +def _wire_flash(monkeypatch, jin10_items=None, em_items=None, jin10_exc=None): + monkeypatch.setattr(cd, "load_stock_pool", lambda **kw: POOL) + if jin10_exc is not None: + jin10_mock = MagicMock(side_effect=jin10_exc) + else: + jin10_mock = MagicMock(return_value=jin10_items or []) + monkeypatch.setattr(cd, "fetch_jin10_flash", jin10_mock) + monkeypatch.setattr(cd, "fetch_em_kuaixun", MagicMock(return_value=em_items or [])) + return jin10_mock + + +def _now_str(minus_min=10): + return (dt.datetime.now() - dt.timedelta(minutes=minus_min)).strftime( + "%Y-%m-%d %H:%M:%S") + + +def test_parse_jin10_payload_extracts_items(): + items = [_jin10_item("J1", _now_str(), content="内容A"), + _jin10_item("J2", _now_str(), title="标题B", content="内容B")] + raw = cd._parse_jin10_payload( + JIN10_JS_TMPL % json.dumps(items, ensure_ascii=False)) + assert len(raw) == 2 + assert raw[0]["id"] == "J1" + assert raw[1]["data"]["title"] == "标题B" + + +def test_parse_jin10_payload_garbage_returns_empty(): + assert cd._parse_jin10_payload("") == [] + assert cd._parse_jin10_payload("var newest = broken;") == [] + assert cd._parse_jin10_payload('var newest = ["截断') == [] + + +def test_parse_em_kuaixun_payload_extracts_items(): + items = [_em_item("E1", _now_str()), _em_item("E2", _now_str(60))] + raw = cd._parse_em_kuaixun_payload( + EM_JSONP_TMPL % json.dumps(items, ensure_ascii=False)) + assert len(raw) == 2 + assert raw[0]["newsid"] == "E1" + + +def test_norm_flash_row_jin10_title_fallback(): + row = cd.norm_flash_row("jin10", _jin10_item("J1", _now_str(), title=None, + content="c" * 200)) + assert row["flash_id"] == "jin10:J1" + assert row["publish_time"] == _now_str() or row["publish_time"] + assert row["title"] == "c" * 80 # title 空时 content 截 80 兜底 + assert row["content"] == "c" * 200 + assert row["channel"] == "5|7" + assert row["important"] == 0 + + +def test_norm_flash_row_em(): + row = cd.norm_flash_row("em_kuaixun", _em_item("E1", "2026-09-21 18:30:00")) + assert row["flash_id"] == "em_kuaixun:E1" + assert row["publish_time"] == "2026-09-21 18:30:00" + assert row["title"] == "东财快讯标题" + assert row["content"] == "快讯摘要内容" + assert row["important"] is None + + +def test_flash_lane_writes_both_sources(root, monkeypatch): + import pandas as pd + _wire_flash(monkeypatch, + jin10_items=[_jin10_item("J1", _now_str()), + _jin10_item("J2", _now_str(30))], + em_items=[_em_item("E1", _now_str()), + _em_item("E2", _now_str(45))]) + rc = cd.run_lane("flash") + assert rc == 0 + part = root / "flash_meta" / f"dt={dt.date.today().isoformat()}" + df = pd.read_parquet(part) + assert len(df) == 4 + assert set(df["source"]) == {"jin10", "em_kuaixun"} + today = dt.date.today().isoformat() + assert cd.is_done("flash", "jin10", today) + assert cd.is_done("flash", "em_kuaixun", today) + + +def test_flash_lane_dedup_second_run(root, monkeypatch): + import pandas as pd + _wire_flash(monkeypatch, + jin10_items=[_jin10_item("J1", _now_str())], + em_items=[_em_item("E1", _now_str())]) + cd.run_lane("flash") + # 次日重拉同款(账本拦截): 手动清 marker 模拟, id 不变 + cd.run_lane("flash") + part = root / "flash_meta" / f"dt={dt.date.today().isoformat()}" + df = pd.read_parquet(part) + assert len(df) == 2 # 零新增 + + +def test_flash_lane_marker_skips_same_day(root, monkeypatch): + today = dt.date.today().isoformat() + cd.mark_done("flash", "jin10", today) + cd.mark_done("flash", "em_kuaixun", today) + jmock = _wire_flash(monkeypatch, jin10_items=[_jin10_item("J9", _now_str())], + em_items=[_em_item("E9", _now_str())]) + rc = cd.run_lane("flash") + assert rc == 0 + jmock.assert_not_called() # 当日已拉, 零请求 + assert not (root / "flash_meta").exists() or \ + not list((root / "flash_meta").rglob("part-*.parquet")) + + +def test_flash_lane_source_isolation(root, monkeypatch): + import pandas as pd + _wire_flash(monkeypatch, + jin10_exc=cd.TransportError("jin10 down"), + em_items=[_em_item("E1", _now_str())]) + rc = cd.run_lane("flash") + assert rc == 1 # 单源失败抬 rc=1 + df = pd.read_parquet(root / "flash_meta" / f"dt={dt.date.today().isoformat()}") + assert len(df) == 1 # 另一源照常落盘 + assert df.iloc[0]["source"] == "em_kuaixun" + + +class _FakeResp: + def __init__(self, text): + self.text = text + + +class _FakeCl: + def __init__(self, pages): + self.pages = pages + self.calls = 0 + + def request(self, method, url, **kw): + r = self.pages[self.calls] + self.calls += 1 + return _FakeResp(r) + + +def test_fetch_em_kuaixun_stops_at_boundary(): + today = dt.date.today() + yest = (today - dt.timedelta(days=1)).strftime("%Y-%m-%d 07:00:00") + older = (today - dt.timedelta(days=2)).strftime("%Y-%m-%d 09:00:00") + now = _now_str() + page1 = EM_JSONP_TMPL % json.dumps( + [_em_item("E1", now), _em_item("E2", now), _em_item("E3", yest)], + ensure_ascii=False) + page2 = EM_JSONP_TMPL % json.dumps([_em_item("E4", older)], ensure_ascii=False) + cl = _FakeCl([page1, page2]) + until = (today - dt.timedelta(days=1)).strftime("%Y-%m-%d 00:00:00") + rows = cd.fetch_em_kuaixun(cl, until) + assert len(rows) == 4 # 页1 全收, 页2 收到即停(最老<边界) + assert cl.calls == 2 + # 首页最老即越界 → 单页即停(page2 全为前天条目, 边界=昨天 00:00) + cl2 = _FakeCl([page2, page1]) + rows2 = cd.fetch_em_kuaixun(cl2, until) + assert len(rows2) == 1 + assert cl2.calls == 1 diff --git a/tests/data_platform/test_corpus_reader.py b/tests/data_platform/test_corpus_reader.py index 632d0559..9478a0de 100644 --- a/tests/data_platform/test_corpus_reader.py +++ b/tests/data_platform/test_corpus_reader.py @@ -58,3 +58,44 @@ class TestAnn: def test_ann_pdf_dir(self, tree): assert cr.ann_pdf_dir(2026) == str(tree / "ann_pdf" / "2026") + + +class TestFlash: + """§20.12.3 快讯域: dt 直读/sources 过滤/空树带 schema。""" + + def test_get_flash_reads_partitions(self, tree): + (tree / "flash_meta" / "dt=2026-09-17").mkdir(parents=True) + pd.DataFrame({ + "flash_id": ["jin10:J1", "em_kuaixun:E1"], + "source": ["jin10", "em_kuaixun"], + "publish_time": ["2026-09-17 07:12:16", "2026-09-17 08:00:00"], + "title": ["t1", "t2"], "content": ["c1", "c2"], + "important": [0, None], "channel": ["5|7", None], + "first_seen_date": ["2026-09-17", "2026-09-17"], + }).to_parquet(tree / "flash_meta" / "dt=2026-09-17" / "part-0.parquet", + index=False) + df = cr.get_flash() + assert len(df) == 2 + assert set(df["source"]) == {"jin10", "em_kuaixun"} + only = cr.get_flash(sources=["jin10"]) + assert list(only["flash_id"]) == ["jin10:J1"] + + def test_get_flash_empty_tree_schema(self, tree): + df = cr.get_flash() + assert df.empty + assert list(df.columns) == cr.FLASH_COLUMNS + + def test_get_flash_date_range(self, tree): + (tree / "flash_meta" / "dt=2026-09-16").mkdir(parents=True) + (tree / "flash_meta" / "dt=2026-09-17").mkdir(parents=True) + for d in ("2026-09-16", "2026-09-17"): + pd.DataFrame({ + "flash_id": [f"jin10:{d}"], "source": ["jin10"], + "publish_time": [f"{d} 07:00:00"], "title": ["t"], + "content": ["c"], "important": [0], "channel": [None], + "first_seen_date": [d], + }).to_parquet(tree / "flash_meta" / f"dt={d}" / "part-0.parquet", + index=False) + assert len(cr.get_flash(start="2026-09-17")) == 1 + assert len(cr.get_flash(end="2026-09-16")) == 1 + assert len(cr.get_flash(start="2026-09-16", end="2026-09-17")) == 2 diff --git a/tests/data_platform/test_vps_mirror_lib.py b/tests/data_platform/test_vps_mirror_lib.py index 31a252a6..1068726a 100644 --- a/tests/data_platform/test_vps_mirror_lib.py +++ b/tests/data_platform/test_vps_mirror_lib.py @@ -111,3 +111,20 @@ class TestWatermark: payload = lib.watermark_payload(lib.scan_family(tree, ["news_meta", "ann_pdf"])) assert payload["domains"] == {"news_meta": "2026-09-17", "ann_pdf": "2026"} assert "generated_at" in payload + + +class TestFlashDomain: + """§20.12.3: flash_meta 入 corpus 族五子域, journal 语义自动吸收。""" + + def test_flash_domain_in_families(self): + assert "flash_meta" in lib.FAMILIES["corpus"]["domains"] + assert len(lib.FAMILIES["corpus"]["domains"]) == 5 + + def test_flash_partition_scanned_and_pushed(self, tree, tmp_path): + _mk_partition(tree, "flash_meta", "dt=2026-09-22", [("part-0.parquet", 8)]) + inv = lib.scan_family(tree, lib.FAMILIES["corpus"]["domains"]) + assert "flash_meta/dt=2026-09-22" in inv + # 空 journal 下全量 planned(新域=指纹全新) + to_push, _ = lib.diff({}, inv) + keys = [k for k in to_push if k.startswith("flash_meta/")] + assert keys == ["flash_meta/dt=2026-09-22"]