diff --git a/scripts/data_platform/corpus_download.py b/scripts/data_platform/corpus_download.py
new file mode 100644
index 0000000..1c80488
--- /dev/null
+++ b/scripts/data_platform/corpus_download.py
@@ -0,0 +1,846 @@
+# -*- coding: utf-8 -*-
+"""corpus_download.py — NAS 基本面语料库(新闻+公告)采集唯一入口(spec §18, 2026-09-06)。
+
+运行形态: 一次性容器 docker run --rm(wrapper=/volume1/stock/corpus/run_corpus.sh),
+与 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)
+ --until HH:MM 墙钟自停: 完成当前 unit 后 checkpoint 退出(rc=3)
+ --limit N 冒烟: 每段最多 N unit, 绝不落 marker(5m 09-17 教训)
+
+断点续传: unit marker(state/markers/, tmp+rename 原子);重启=从最老未 done unit 续跑。
+退出码: 0=完成 1=致命(unit 失败/域名冷却) 2=限流让路(429) 3=墙钟/锁让路。
+
+单实例锁: fcntl.flock(state/corpus.lock) — 跨容器语义(内核级, bind mount 同一
+inode;进程死自动释放,无陈旧锁)。5m 的 pid 活性检测在 docker run --rm 各自独立
+PID namespace 下失效,故 corpus 用 flock(契约同款: 第二实例让路退出)。
+
+id 去重: IdLedger(全局 int64 hash 账本, state/ids_*.parquet) + RecentIndex(近
+DEDUP_WINDOW_DAYS 日分区键索引, 兜住账本 flush 周期内的崩溃窗口)双层;append-only
+不重写历史分区。news_meta 键=(art_code,stock_code) 复合(同一稿提及多股=meta 多行,
+fulltext 层 art_code 全局一条, spec §18.2)。
+
+接口契约(docs/fundamental_corpus_probes_20260906.md 实测):
+- 巨潮 pool: GET .../new/data/szse_stock.json → stockList[{code,orgId}](含退市+北交)
+- 巨潮查询: POST .../new/hisAnnouncement/query, stock="code,orgId", pageSize≤30,
+ 翻页按 totalAnnouncement;业绩快报无服务端类别码(无效码被静默忽略=全量!),
+ 六类=五码 category_ndbg/bndbg/yjdbg/sjdbg/yjygjxz_szsh + 客户端标题谓词「业绩快报」
+- 东财日增: search-api-web jsonp(akshare stock_news_em 同款, 带 content 摘要,最近100)
+- 东财回补: np-listapi(无摘要,历史~3年,翻到空页或 page≥200 止)
+"""
+import argparse
+import datetime as dt
+import fcntl
+import hashlib
+import json
+import logging
+import os
+import random
+import re
+import sys
+import time
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+import requests
+
+# ---------- 常量(测试可 monkeypatch) ----------
+
+CORPUS_ROOT = Path(os.environ.get("CORPUS_ROOT", "/volume1/stock/corpus"))
+
+POOL_URL = "http://www.cninfo.com.cn/new/data/szse_stock.json"
+QUERY_URL = "http://www.cninfo.com.cn/new/hisAnnouncement/query"
+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}")
+
+PAGE_SIZE = 30 # 巨潮服务端封顶
+FIVE_CAT = ";".join([
+ "category_ndbg_szsh", "category_bndbg_szsh", "category_yjdbg_szsh",
+ "category_sjdbg_szsh", "category_yjygjxz_szsh",
+]) # 业绩快报无码, 标题谓词补充
+EXPRESS_KEYWORD = "业绩快报"
+BACKFILL_FROM_YEAR = 2000
+NEWS_BACKFILL_MAX_PAGE = 200 # np-listapi 硬顶(实测 page200 空)
+
+COOLDOWN_ERRORS = 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股/基金/转债
+
+RATE = {"cninfo": 1.0, "eastmoney": 1.0} # 每域最小间隔(秒), 慢爬纪律
+JITTER = (0.0, 0.3)
+
+ID_KEYS = {"ann_meta": ["announcement_id"],
+ "news_meta": ["art_code", "stock_code"],
+ "news_fulltext": ["art_code"]}
+
+log = logging.getLogger("corpus")
+
+
+# ---------- 异常 ----------
+
+class TransportError(Exception):
+ """传输类失败(超时/5xx/连接): unit 不标 done, 计入域名连续错。"""
+
+
+class DomainCooldown(Exception):
+ """域名冷却: 连续错×10 或 429。rate_limited=True → 退出码走 2(限流让路)。"""
+
+ def __init__(self, domain, rate_limited=False):
+ super().__init__(f"domain {domain} cooled (rate_limited={rate_limited})")
+ self.rate_limited = rate_limited
+
+
+class HttpDeterministicError(Exception):
+ """确定性 4xx(非 429): 如 PDF 404。记日志跳过, 绝不烧重试。"""
+
+ def __init__(self, status, url=""):
+ super().__init__(f"HTTP {status} {url}")
+ self.status = status
+
+
+class WallClockStop(Exception):
+ """--until 墙钟到: 完成当前 unit 后抛出, 顶层收拾落盘并 rc=3。"""
+
+
+# ---------- 单实例锁(flock, 跨容器) ----------
+
+def acquire_lock():
+ """取锁返回 fd(进程存活期间持有); 已被持有 → None(让路)。
+
+ flock 由内核在文件描述符上持有 — bind mount 两侧(两个容器/容器与宿主)
+ 看到同一 inode, 互斥跨容器成立;进程退出内核自动释放,无陈旧锁文件问题。
+ """
+ lock = CORPUS_ROOT / "state" / "corpus.lock"
+ lock.parent.mkdir(parents=True, exist_ok=True)
+ fd = os.open(str(lock), os.O_CREAT | os.O_RDWR, 0o666)
+ try:
+ fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+ except BlockingIOError:
+ os.close(fd)
+ return None
+ return fd
+
+
+# ---------- 域名客户端(限速/连续错/429) ----------
+
+class DomainClient:
+ def __init__(self, name):
+ self.name = name
+ self.session = requests.Session()
+ self.session.trust_env = False # 直连不走代理(慢爬纪律)
+ self.consecutive_errors = 0
+ self.rate_limited = False
+ self.cooled = False
+ self.n_requests = 0
+ self._last = 0.0
+
+ def _pace(self):
+ interval = RATE.get(self.name, 1.0) + random.uniform(*JITTER)
+ wait = self._last + interval - time.monotonic()
+ if wait > 0:
+ time.sleep(wait)
+ self._last = time.monotonic()
+ self.n_requests += 1
+
+ def _transport_fail(self, msg):
+ self.consecutive_errors += 1
+ if self.consecutive_errors >= COOLDOWN_ERRORS:
+ self.cooled = True
+ raise DomainCooldown(self.name)
+ raise TransportError(msg)
+
+ def request(self, method, url, **kw):
+ if self.cooled:
+ raise DomainCooldown(self.name)
+ self._pace()
+ try:
+ r = self.session.request(method, url, timeout=25, **kw)
+ except requests.RequestException as e:
+ self._transport_fail(f"{type(e).__name__}: {e}")
+ if r.status_code == 429:
+ self.rate_limited = True
+ self.cooled = True
+ raise DomainCooldown(self.name, rate_limited=True)
+ if r.status_code >= 500:
+ self._transport_fail(f"HTTP {r.status_code}")
+ if r.status_code >= 400:
+ raise HttpDeterministicError(r.status_code, url)
+ self.consecutive_errors = 0
+ return r
+
+
+# ---------- 股票池 ----------
+
+def filter_stock_pool(pairs):
+ """A 股前缀白名单(含北交), 排 B股(200/900)/基金(15/16/51/56/58)/转债等。"""
+ return [(c, o) for c, o in pairs if c[:2] in POOL_A_PREFIXES]
+
+
+def load_stock_pool(refresh=False):
+ """(code, orgId) 全池: 巨潮 szse_stock.json(实测含退市 000003/北交 920001),
+ 本地缓存 POOL_TTL_DAYS 天。orgId=按股查询的硬前提。"""
+ cache = CORPUS_ROOT / "state" / "cninfo_pool.json"
+ if cache.exists() and not refresh:
+ age = time.time() - cache.stat().st_mtime
+ if age < POOL_TTL_DAYS * 86400:
+ j = json.loads(cache.read_text(encoding="utf-8"))
+ return sorted(j.items())
+ client = DomainClient("cninfo")
+ r = client.request("GET", POOL_URL)
+ pairs = [(it["code"], it["orgId"])
+ for it in r.json().get("stockList") or []]
+ pairs = filter_stock_pool(pairs)
+ tmp = cache.with_suffix(".tmp")
+ tmp.write_text(json.dumps(dict(pairs), ensure_ascii=False), encoding="utf-8")
+ os.replace(tmp, cache)
+ log.info("pool refreshed: %d stocks", len(pairs))
+ return pairs
+
+
+# ---------- 归一化(纯函数) ----------
+
+def _strip_em(s):
+ if not s:
+ return s
+ return (s.replace("", "").replace("", "")
+ .replace("(", "").replace(")", ""))
+
+
+def _epoch_ms_to_naive(ms):
+ if not ms:
+ return None
+ tz8 = dt.timezone(dt.timedelta(hours=8))
+ return dt.datetime.fromtimestamp(ms / 1000, tz=tz8).replace(tzinfo=None)
+
+
+def norm_ann_row(raw, first_seen):
+ t = raw.get("announcementTime")
+ ts = _epoch_ms_to_naive(t)
+ return {
+ "announcement_id": str(raw.get("announcementId") or ""),
+ "sec_code": raw.get("secCode"),
+ "sec_name": raw.get("secName"),
+ "org_id": raw.get("orgId"),
+ "title": _strip_em(raw.get("announcementTitle")),
+ "short_title": raw.get("shortTitle"),
+ "content": raw.get("announcementContent") or None,
+ "ann_time": ts.strftime("%Y-%m-%d %H:%M:%S") if ts else None,
+ "ann_type": raw.get("announcementType"),
+ "ann_type_name": raw.get("announcementTypeName"),
+ "column_id": raw.get("columnId"),
+ "important": raw.get("important"),
+ "adjunct_url": raw.get("adjunctUrl"),
+ "adjunct_size": raw.get("adjunctSize"),
+ "batch_num": raw.get("batchNum"),
+ "first_seen_date": first_seen,
+ }
+
+
+def norm_news_search_row(raw, stock_code):
+ summary = _strip_em(raw.get("content") or "")
+ summary = summary.replace(" ", "").replace("\r\n", " ") or None
+ return {
+ "art_code": str(raw.get("code") or ""),
+ "stock_code": stock_code,
+ "show_time": raw.get("date"),
+ "title": _strip_em(raw.get("title")),
+ "summary": summary,
+ "media_name": raw.get("mediaName") or None,
+ "url": "http://finance.eastmoney.com/a/{}.html".format(raw.get("code")),
+ "first_seen_date": dt.date.today().isoformat(),
+ }
+
+
+def norm_news_np_row(raw, stock_code):
+ """np-listapi 回补段: summary/media_name 物理不存在 → None 如实(spec §18.2)。"""
+ return {
+ "art_code": str(raw.get("Art_Code") or ""),
+ "stock_code": stock_code,
+ "show_time": raw.get("Art_ShowTime"),
+ "title": _strip_em(raw.get("Art_Title")),
+ "summary": None,
+ "media_name": None,
+ "url": raw.get("Art_Url"),
+ "first_seen_date": dt.date.today().isoformat(),
+ }
+
+
+_TAG_RE = re.compile(r"<[^>]+>")
+_WS_RE = re.compile(r"[ \t\f\v ]+")
+_MULTI_NL = re.compile(r"\n\s*\n+")
+
+
+def html_to_text(html):
+ html = re.sub(r"(?is)<(script|style)[^>]*>.*?", " ", html or "")
+ text = _TAG_RE.sub("\n", html)
+ text = (text.replace(" ", " ").replace("&", "&")
+ .replace("<", "<").replace(">", ">").replace(""", '"'))
+ lines = [ln.strip() for ln in text.splitlines()]
+ return _MULTI_NL.sub("\n", _WS_RE.sub(" ", "\n".join(lines))).strip()
+
+
+# ---------- id 去重: 全局账本 + 近窗索引 ----------
+
+def _hash_id(s):
+ return int.from_bytes(hashlib.blake2b(s.encode("utf-8"),
+ digest_size=8).digest(), "big") & (2**63 - 1)
+
+
+def _row_key(domain, row):
+ return "|".join(str(row.get(c) or "") for c in ID_KEYS[domain])
+
+
+class IdLedger:
+ """全局 id 账本(state/ids_.parquet, int64 hash 有序数组)。
+
+ 覆盖 spec「短窗去重」: 全局精确去重(§18.5 要求 announcement_id 全局无重)。
+ FLUSH_EVERY 周期落盘; 崩溃窗口由 RecentIndex(读分区)兜底。
+ """
+
+ def __init__(self, domain):
+ self.path = CORPUS_ROOT / "state" / f"ids_{domain}.parquet"
+ if self.path.exists():
+ self._arr = pd.read_parquet(self.path)["h"].to_numpy()
+ else:
+ self._arr = np.empty(0, dtype=np.int64)
+ self._pending = []
+
+ def add(self, ids):
+ self._pending.extend(_hash_id(i) for i in ids)
+
+ def has_any(self, ids):
+ hs = np.array([_hash_id(i) for i in ids], dtype=np.int64)
+ out = []
+ for h in hs:
+ i = np.searchsorted(self._arr, h)
+ out.append(bool(i < len(self._arr) and self._arr[i] == h)
+ or h in self._pending)
+ return out
+
+ def flush(self):
+ if not self._pending:
+ return len(self._arr)
+ merged = np.unique(np.concatenate(
+ [self._arr, np.array(self._pending, dtype=np.int64)]))
+ tmp = self.path.with_suffix(".tmp")
+ pd.DataFrame({"h": merged}).to_parquet(tmp, index=False)
+ os.replace(tmp, self.path)
+ self._arr = merged
+ self._pending = []
+ return len(self._arr)
+
+
+class RecentIndex:
+ """近 N 日分区键索引: 账本 flush 周期内的崩溃恢复兜底(分区即真相)。"""
+
+ def __init__(self, domain, days=DEDUP_WINDOW_DAYS):
+ self.domain = domain
+ self.keys = set()
+ today = dt.date.today()
+ for k in range(days + 1):
+ part = (CORPUS_ROOT / domain / f"dt={(today - dt.timedelta(days=k)).isoformat()}"
+ / "part-0.parquet")
+ if part.exists():
+ df = pd.read_parquet(part, columns=ID_KEYS[domain])
+ for row in df.to_dict("records"):
+ self.keys.add("|".join(str(row[c]) for c in ID_KEYS[domain]))
+
+ def has_any(self, ids):
+ return [i in self.keys for i in ids]
+
+ def add(self, ids):
+ self.keys.update(ids)
+
+
+# ---------- 分区追加(append-only + 原子) ----------
+
+def append_parquet(domain, rows, id_col):
+ """追进今天分区 part-0.parquet; 同日同 id 只进一次; tmp+rename 原子。"""
+ if not rows:
+ return 0
+ cols = ID_KEYS[domain] if isinstance(id_col, list) else [id_col]
+ day = dt.date.today().isoformat()
+ d = CORPUS_ROOT / domain / f"dt={day}"
+ d.mkdir(parents=True, exist_ok=True)
+ part = d / "part-0.parquet"
+ df_new = pd.DataFrame(rows)
+ if part.exists():
+ existing = pd.read_parquet(part)
+ keys = {"|".join(str(r[c]) for c in cols)
+ for r in existing[cols].to_dict("records")}
+ keep = ["|".join(str(r[c]) for c in cols) not in keys
+ for r in df_new[cols].to_dict("records")]
+ df_new = df_new[keep]
+ if df_new.empty:
+ return 0
+ df_new.index = range(len(existing), len(existing) + len(df_new))
+ df_new = pd.concat([existing, df_new])
+ tmp = d / f".part-0.{os.getpid()}.tmp"
+ try:
+ df_new.to_parquet(tmp, index=False)
+ os.replace(tmp, part)
+ except Exception:
+ tmp.unlink(missing_ok=True)
+ raise
+ return len(df_new)
+
+
+# ---------- unit marker ----------
+
+def _marker_path(lane, stage, unit):
+ return CORPUS_ROOT / "state" / "markers" / lane / stage / f"{unit}.done"
+
+
+def is_done(lane, stage, unit):
+ return _marker_path(lane, stage, unit).exists()
+
+
+def mark_done(lane, stage, unit):
+ p = _marker_path(lane, stage, unit)
+ p.parent.mkdir(parents=True, exist_ok=True)
+ tmp = p.with_suffix(".tmp")
+ tmp.write_text(dt.datetime.now().isoformat(), encoding="utf-8")
+ os.replace(tmp, p)
+
+
+# ---------- fetch 层 ----------
+
+def fetch_cninfo_announcements(client, code, org_id, start, end, category=""):
+ data = {"pageNum": "1", "pageSize": str(PAGE_SIZE), "column": "szse",
+ "tabName": "fulltext", "plate": "", "stock": f"{code},{org_id}",
+ "searchkey": "", "secid": "", "category": category, "trade": "",
+ "seDate": f"{start}~{end}", "sortName": "", "sortType": "",
+ "isHLtitle": "true"}
+ out, page = [], 1
+ while True:
+ data["pageNum"] = str(page)
+ r = client.request("POST", QUERY_URL, data=data)
+ try:
+ j = r.json()
+ except ValueError as e:
+ raise TransportError(f"json decode: {e}") from e
+ rows = j.get("announcements") or []
+ out.extend(rows)
+ total = j.get("totalAnnouncement") or 0
+ if len(rows) < PAGE_SIZE or page * PAGE_SIZE >= total or page >= 500:
+ return out
+ page += 1
+
+
+def fetch_news_recent(client, code):
+ inner = {"uid": "", "keyword": code, "type": ["cmsArticleWebOld"],
+ "client": "web", "clientType": "web", "clientVersion": "curr",
+ "param": {"cmsArticleWebOld": {"searchScope": "default",
+ "sort": "default", "pageIndex": 1,
+ "pageSize": 100, "preTag": "",
+ "postTag": ""}}}
+ params = {"cb": "cb", "param": json.dumps(inner, ensure_ascii=False),
+ "_": str(int(time.time() * 1000))}
+ r = client.request("GET", SEARCH_API_URL, params=params)
+ s = r.text.strip()
+ j = json.loads(s[s.index("(") + 1:s.rindex(")")])
+ return (j.get("result") or {}).get("cmsArticleWebOld") or []
+
+
+def fetch_news_backfill_page(client, code, mkt_prefix, page):
+ url = NP_LIST_URL.format(mkt=mkt_prefix, code=code, page=page)
+ r = client.request("GET", url)
+ return (r.json().get("data") or {}).get("list") or []
+
+
+def fetch_article_html(client, url):
+ return client.request("GET", url).text
+
+
+def download_pdf(client, url, dest):
+ r = client.request("GET", url)
+ dest.parent.mkdir(parents=True, exist_ok=True)
+ tmp = dest.with_suffix(".tmp")
+ tmp.write_bytes(r.content)
+ os.replace(tmp, dest)
+ return len(r.content)
+
+
+# ---------- PDF 谓词(六类=五服务端码+业绩快报标题) ----------
+
+_PDF_LABELS = [(EXPRESS_KEYWORD, "业绩快报"), ("业绩预告", "业绩预告"),
+ ("半年", "半年报"), ("三季度", "三季报"), ("一季度", "一季报"),
+ ("年度报告", "年报"), ("年报", "年报")]
+
+
+def collect_pdf_candidates(rows):
+ """无服务端码的「业绩快报」靠标题谓词(自描述文档类, 误报可忽略)。"""
+ return [r for r in rows
+ if EXPRESS_KEYWORD in (r.get("announcementTitle") or "")]
+
+
+def _pdf_label(title):
+ for kw, label in _PDF_LABELS:
+ if kw in (title or ""):
+ return label
+ return None
+
+
+def _report_year(title, ann_time):
+ m = re.search(r"(19|20)\d{2}", title or "")
+ if m:
+ return m.group(0)
+ return ann_time[:4] if ann_time else None
+
+
+# ---------- 运行上下文 ----------
+
+class Ctx:
+ def __init__(self, lane, until=None, limit=None):
+ self.lane = lane
+ self.limit = limit
+ self.deadline = None
+ if until:
+ hh, mm = until.split(":")
+ self.deadline = dt.datetime.combine(
+ dt.date.today(), dt.time(int(hh), int(mm)))
+ self.failed = 0
+ self.rate_limited = False
+ self.hard_cool = False
+ self.wallclock = False
+ self.units = 0
+ self.depth = {}
+
+ def stop_now(self):
+ """完成当前 unit 后调用: 过墙钟 → 抛 WallClockStop(checkpoint 语义)。"""
+ if self.deadline and dt.datetime.now() >= self.deadline:
+ self.wallclock = True
+ raise WallClockStop()
+
+ def rc(self):
+ if self.failed or self.hard_cool:
+ return 1
+ if self.rate_limited:
+ return 2
+ if self.wallclock:
+ return 3
+ return 0
+
+
+def _filter_new(domain, norm_rows, ledger, recent):
+ keys = [_row_key(domain, r) for r in norm_rows]
+ known_l = ledger.has_any(keys)
+ known_r = recent.has_any(keys)
+ return [r for r, kl, kr in zip(norm_rows, known_l, known_r)
+ if not kl and not kr]
+
+
+def _record_depth(ctx, source, code, date_str):
+ if not date_str:
+ return
+ cur = ctx.depth.setdefault(source, {}).get(code)
+ if cur is None or date_str < cur:
+ ctx.depth[source][code] = date_str
+
+
+def _save_depth(ctx):
+ p = CORPUS_ROOT / "state" / "depth_summary.json"
+ merged = {}
+ if p.exists():
+ try:
+ merged = json.loads(p.read_text(encoding="utf-8"))
+ except ValueError:
+ log.warning("depth_summary.json 损坏, 重写(幂等无损)")
+ for src, stocks in ctx.depth.items():
+ for code, d in stocks.items():
+ cur = merged.get(src, {}).get(code)
+ if cur is None or d < cur:
+ merged.setdefault(src, {})[code] = d
+ tmp = p.with_suffix(".tmp")
+ tmp.write_text(json.dumps(merged, ensure_ascii=False, indent=1), encoding="utf-8")
+ os.replace(tmp, p)
+
+
+def _maybe_flush(ledgers, since_flush):
+ if since_flush >= FLUSH_EVERY:
+ for led in ledgers.values():
+ led.flush()
+ return 0
+ return since_flush
+
+
+# ---------- daily lane ----------
+
+def _run_ann_increment(ctx, pool, cl_cn, ledgers, recents):
+ today = dt.date.today()
+ yday = today - dt.timedelta(days=1)
+ first_seen = today.isoformat()
+ candidates = []
+ since_flush = 0
+ for code, org in pool:
+ if ctx.limit is None:
+ if is_done(ctx.lane, "ann", code):
+ continue
+ elif ctx.units >= ctx.limit:
+ break
+ try:
+ rows = fetch_cninfo_announcements(cl_cn, code, org,
+ yday.isoformat(), today.isoformat())
+ except DomainCooldown as e:
+ log.warning("ann 域冷却 @%s: %s", code, e)
+ ctx.rate_limited |= e.rate_limited
+ ctx.hard_cool |= not e.rate_limited
+ return candidates
+ except (TransportError, HttpDeterministicError) as e:
+ ctx.failed += 1
+ log.warning("ann unit %s 失败(不标done): %s", code, e)
+ 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"])
+ 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])
+ log.info("ann %s: +%d/%d", code, len(new), len(rows))
+ if ctx.limit is None:
+ mark_done(ctx.lane, "ann", code)
+ ctx.units += 1
+ since_flush = _maybe_flush(ledgers, since_flush + 1)
+ ctx.stop_now()
+ return candidates
+
+
+def _run_news_increment(ctx, pool, cl_em, ledgers, recents, new_articles):
+ for code, _org in pool:
+ if ctx.limit is None:
+ if is_done(ctx.lane, "news", code):
+ continue
+ elif ctx.units >= ctx.limit:
+ break
+ try:
+ rows = fetch_news_recent(cl_em, code)
+ except DomainCooldown as e:
+ log.warning("news 域冷却 @%s: %s", code, 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("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"])
+ 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"]))
+ log.info("news %s: +%d/%d", code, len(new), len(rows))
+ if ctx.limit is None:
+ mark_done(ctx.lane, "news", code)
+ ctx.units += 1
+ ctx.stop_now()
+
+
+def _run_fulltext(ctx, cl_em, ledgers, recents, new_articles):
+ seen = set()
+ for art_code, url, show_time in new_articles:
+ if art_code in seen or is_done(ctx.lane, "fulltext", art_code):
+ continue
+ try:
+ html = fetch_article_html(cl_em, url)
+ except DomainCooldown as e:
+ log.warning("fulltext 域冷却 @%s: %s", art_code, e)
+ ctx.rate_limited |= e.rate_limited
+ ctx.hard_cool |= not e.rate_limited
+ return
+ except HttpDeterministicError as e:
+ log.warning("fulltext %s 确定性失败(标done跳过): %s", art_code, e)
+ if ctx.limit is None:
+ mark_done(ctx.lane, "fulltext", art_code)
+ seen.add(art_code)
+ continue
+ except TransportError as e:
+ ctx.failed += 1
+ log.warning("fulltext %s 失败(不标done): %s", art_code, e)
+ continue
+ row = {"art_code": art_code, "url": url, "show_time": show_time,
+ "content_text": html_to_text(html),
+ "fetch_date": dt.date.today().isoformat()}
+ if not any(ledgers["news_fulltext"].has_any([art_code])
+ + recents["news_fulltext"].has_any([art_code])):
+ append_parquet("news_fulltext", [row], "art_code")
+ ledgers["news_fulltext"].add([art_code])
+ recents["news_fulltext"].add([art_code])
+ if ctx.limit is None:
+ mark_done(ctx.lane, "fulltext", art_code)
+ seen.add(art_code)
+ ctx.units += 1
+ ctx.stop_now()
+
+
+def _run_pdf_daily(ctx, pool, cl_cn, ledgers, candidates):
+ today = dt.date.today()
+ yday = today - dt.timedelta(days=1)
+ for code, org in pool:
+ try:
+ rows = fetch_cninfo_announcements(cl_cn, code, org,
+ yday.isoformat(), today.isoformat(),
+ category=FIVE_CAT)
+ except DomainCooldown as e:
+ log.warning("pdf 域冷却 @%s: %s", code, e)
+ ctx.rate_limited |= e.rate_limited
+ ctx.hard_cool |= not e.rate_limited
+ break
+ except (TransportError, HttpDeterministicError) as e:
+ ctx.failed += 1
+ log.warning("pdf fivecat %s 失败: %s", code, e)
+ continue
+ candidates.extend(rows)
+ ctx.stop_now()
+ seen = set()
+ index_rows = []
+ for raw in candidates:
+ aid = str(raw.get("announcementId") or "")
+ if not aid or aid in seen:
+ continue
+ seen.add(aid)
+ if is_done(ctx.lane, "pdf", aid):
+ continue
+ ann_time = _epoch_ms_to_naive(raw.get("announcementTime"))
+ year = (ann_time or dt.datetime.now()).strftime("%Y")
+ dest = CORPUS_ROOT / "ann_pdf" / year / f"{aid}.pdf"
+ if dest.exists() and dest.stat().st_size > 0:
+ continue
+ url = PDF_BASE + (raw.get("adjunctUrl") or "")
+ try:
+ size = download_pdf(cl_cn, url, dest)
+ except HttpDeterministicError as e:
+ if e.status == 404:
+ log.warning("pdf %s 404 missing(标done不重试): %s", aid, url)
+ if ctx.limit is None:
+ mark_done(ctx.lane, "pdf", aid)
+ continue
+ ctx.failed += 1
+ log.warning("pdf %s 失败: %s", aid, e)
+ continue
+ except (TransportError, DomainCooldown) as e:
+ ctx.failed += 1
+ log.warning("pdf %s 失败(不标done): %s", aid, e)
+ continue
+ title = raw.get("announcementTitle") or ""
+ index_rows.append({"announcement_id": aid,
+ "sec_code": raw.get("secCode"),
+ "ann_type_name": _pdf_label(title),
+ "report_year": _report_year(title,
+ ann_time.strftime("%Y-%m-%d") if ann_time else None),
+ "local_path": str(dest), "bytes": size,
+ "fetch_date": today.isoformat()})
+ if ctx.limit is None:
+ 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))
+
+
+def run_daily(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")
+ 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_pdf_daily(ctx, pool, cl_cn, ledgers, candidates)
+ for led in ledgers.values():
+ led.flush()
+
+
+# ---------- backfill lane(T2 占位, ①②=T4 ③④=T5) ----------
+
+def run_backfill(ctx, pool):
+ log.warning("backfill lane 未实施(T4/T5): 让路退出, daily 数据不受影响")
+ return
+
+
+# ---------- lane 入口 ----------
+
+def _ensure_log():
+ if log.handlers:
+ return
+ (CORPUS_ROOT / "logs").mkdir(parents=True, exist_ok=True)
+ ts = dt.datetime.now().strftime("%Y%m%d_%H%M%S")
+ fh = logging.FileHandler(CORPUS_ROOT / "logs" / f"corpus_{ts}.log",
+ encoding="utf-8")
+ fh.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
+ log.addHandler(fh)
+ log.addHandler(logging.StreamHandler())
+ log.setLevel(logging.INFO)
+
+
+def run_lane(lane, until=None, limit=None):
+ _ensure_log()
+ ctx = Ctx(lane=lane, until=until, limit=limit)
+ pool = load_stock_pool()
+ log.info("lane=%s start: %d stocks, until=%s, limit=%s",
+ lane, len(pool), until, limit)
+ t0 = time.monotonic()
+ try:
+ if lane == "daily":
+ run_daily(ctx, pool)
+ elif lane == "backfill":
+ run_backfill(ctx, pool)
+ else:
+ log.error("unknown lane: %s", lane)
+ ctx.failed += 1
+ except WallClockStop:
+ log.info("墙钟到(--until %s): 完成当前 unit 后 checkpoint 退出", until)
+ finally:
+ _save_depth(ctx)
+ 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()
+
+
+def main():
+ ap = argparse.ArgumentParser(description="NAS 基本面语料库采集(spec §18)")
+ ap.add_argument("--lane", required=True, choices=["daily", "backfill"])
+ 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")
+ args = ap.parse_args()
+ fd = acquire_lock()
+ if fd is None:
+ print("另一实例持有锁, 让路退出", file=sys.stderr)
+ sys.exit(3)
+ try:
+ rc = run_lane(args.lane, until=args.until, limit=args.limit)
+ sys.exit(rc)
+ finally:
+ os.close(fd)
+
+
+if __name__ == "__main__":
+ main()
diff --git a/tests/data_platform/test_corpus_download.py b/tests/data_platform/test_corpus_download.py
new file mode 100644
index 0000000..2896654
--- /dev/null
+++ b/tests/data_platform/test_corpus_download.py
@@ -0,0 +1,534 @@
+# -*- coding: utf-8 -*-
+"""TDD for corpus_download.py — NAS 基本面语料库(spec §18.7 T2)。
+
+全 mock 零网络;日期全部相对 today(禁写死);CORPUS_ROOT 重定向 tmp_path。
+契约来源: docs/fundamental_corpus_probes_20260906.md 实测 + spec §18。
+"""
+import datetime as dt
+import fcntl
+import json
+import sys
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from scripts.data_platform import corpus_download as cd
+
+
+# ---------- Fixtures ----------
+
+@pytest.fixture(autouse=True)
+def _fast(monkeypatch):
+ """限速归零 + 抖动归零: 测试不 sleep。"""
+ monkeypatch.setattr(cd, "RATE", {k: 0.0 for k in cd.RATE})
+ monkeypatch.setattr(cd, "JITTER", (0.0, 0.0))
+
+
+@pytest.fixture
+def root(tmp_path, monkeypatch):
+ monkeypatch.setattr(cd, "CORPUS_ROOT", tmp_path)
+ 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
+
+
+POOL = [("000001", "gssz0000001"), ("600519", "gssh600519")]
+
+
+def _ann_raw(aid="1225275126", code="000001", title="关于xxx的公告",
+ ts_ms=1777563098000, **over):
+ row = {
+ "id": None, "secCode": code, "secName": "平安银行", "orgId": "gssz0000001",
+ "announcementId": aid, "announcementTitle": title,
+ "announcementTime": ts_ms, "adjunctUrl": f"finalpage/2026-04-30/{aid}.PDF",
+ "adjunctSize": 244, "adjunctType": "PDF", "storageTime": None,
+ "columnId": "09020202||250101||251302", "pageColumn": "SZZB",
+ "announcementType": "01010503||010112||010301", "associateAnnouncement": None,
+ "important": None, "batchNum": None, "announcementContent": "",
+ "orgName": None, "tileSecName": "平安银行", "shortTitle": title,
+ "announcementTypeName": None, "secNameList": None,
+ }
+ row.update(over)
+ return row
+
+
+def _resp(json_data=None, status=200, text=""):
+ r = MagicMock()
+ r.status_code = status
+ r.json.return_value = json_data if json_data is not None else {}
+ r.text = text
+ r.content = b""
+ return r
+
+
+def _cninfo_page(rows, total=None):
+ total = len(rows) if total is None else total
+ return {"announcements": rows, "totalAnnouncement": total, "hasMore": False}
+
+
+# ---------- 股票池过滤 ----------
+
+def test_pool_prefix_filter():
+ """A 股前缀白名单: 00/30/60/68/43/83/87/92 过,B股(200/900)/基金(15/51/56)拒。"""
+ raw = [("000001", "o1"), ("300750", "o2"), ("600519", "o3"), ("688981", "o4"),
+ ("920001", "o5"), ("430047", "o6"), ("830799", "o7"), ("871981", "o8"),
+ ("200002", "x1"), ("900901", "x2"), ("159915", "x3"), ("510300", "x4"),
+ ("561880", "x5"), ("110038", "x6"), ("113050", "x7"), ("161725", "x8")]
+ got = cd.filter_stock_pool(raw)
+ codes = [c for c, _ in got]
+ assert codes == ["000001", "300750", "600519", "688981", "920001",
+ "430047", "830799", "871981"]
+
+
+# ---------- 归一化 ----------
+
+def test_norm_ann_row_epoch_ms_to_naive_ts():
+ row = cd.norm_ann_row(_ann_raw(ts_ms=1777563098000), first_seen="2026-09-07")
+ # 1777563098s UTC+8 = 2026-04-30 (Asia/Shanghai, 无时区后缀)
+ assert row["ann_time"].startswith("2026-04-30")
+ assert row["announcement_id"] == "1225275126"
+ assert row["sec_code"] == "000001"
+ assert row["first_seen_date"] == "2026-09-07"
+ assert row["ann_type"] == "01010503||010112||010301"
+ assert row["ann_type_name"] is None
+
+
+def test_norm_ann_row_strips_em_and_keeps_nulls():
+ row = cd.norm_ann_row(_ann_raw(title="2022年度业绩快报"),
+ first_seen="2026-09-07")
+ assert row["title"] == "2022年度业绩快报"
+ assert row["important"] is None
+ assert row["batch_num"] is None
+
+
+def test_norm_news_search_row():
+ raw = {"code": "art123", "date": "2026-09-06 10:11:51",
+ "title": "贵州茅台(600519.SH)发布", "content": "正文摘要 有 全角",
+ "mediaName": "界面新闻", "image": ""}
+ row = cd.norm_news_search_row(raw, stock_code="600519")
+ assert row["art_code"] == "art123"
+ assert row["stock_code"] == "600519"
+ assert row["title"] == "贵州茅台(600519.SH)发布"
+ assert "" not in row["summary"] and " " not in row["summary"]
+ assert row["media_name"] == "界面新闻"
+ assert row["url"] == "http://finance.eastmoney.com/a/art123.html"
+ assert row["show_time"] == "2026-09-06 10:11:51"
+
+
+def test_norm_news_np_row_summary_null():
+ """np-listapi 回补段: summary/media_name 物理不存在 → None 如实(spec §18.2)。"""
+ raw = {"Art_Code": "art9", "Art_ShowTime": "2023-09-01 08:00:00",
+ "Art_Title": "旧新闻", "Art_Url": "http://finance.eastmoney.com/a/art9.html",
+ "Np_dst": "CMS", "Art_SortStart": 1}
+ row = cd.norm_news_np_row(raw, stock_code="600519")
+ assert row["summary"] is None
+ assert row["media_name"] is None
+ assert row["art_code"] == "art9"
+
+
+def test_html_to_text():
+ html = (""
+ " 第一段
第二段
")
+ text = cd.html_to_text(html)
+ assert "var x" not in text and ".a{}" not in text
+ assert "第一段" in text and "第二段" in text
+
+
+# ---------- IdLedger(全局 id 去重账本, int64 hash) ----------
+
+def test_ledger_roundtrip_and_dup_detect(root):
+ led = cd.IdLedger("ann_meta")
+ led.add(["a1", "a2", "a3"])
+ assert led.flush() == 3
+ led2 = cd.IdLedger("ann_meta")
+ assert led2.has_any(["a2", "zzz"]) == [True, False]
+
+
+def test_ledger_empty_init(root):
+ led = cd.IdLedger("ann_meta")
+ assert led.has_any(["anything"]) == [False]
+
+
+# ---------- RecentIndex(近 N 日分区 id 索引) ----------
+
+def test_recent_index_reads_partitions(root):
+ today = dt.date.today()
+ d = root / "ann_meta" / f"dt={today.isoformat()}"
+ d.mkdir(parents=True)
+ import pandas as pd
+ pd.DataFrame({"announcement_id": ["old1", "old2"]}).to_parquet(d / "part-0.parquet")
+ idx = cd.RecentIndex("ann_meta", days=7)
+ assert idx.has_any(["old1", "new1"]) == [True, False]
+
+
+# ---------- parquet 追加(append-only + 原子) ----------
+
+def test_append_parquet_dedup_and_atomic(root):
+ import pandas as pd
+ today = dt.date.today().isoformat()
+ rows1 = [{"announcement_id": "a1", "title": "x"}, {"announcement_id": "a2", "title": "y"}]
+ cd.append_parquet("ann_meta", rows1, id_col="announcement_id")
+ part = root / "ann_meta" / f"dt={today}" / "part-0.parquet"
+ df = pd.read_parquet(part)
+ assert len(df) == 2
+ # 同日追加含重复 id → 只进新 id
+ rows2 = [{"announcement_id": "a2", "title": "y2"}, {"announcement_id": "a3", "title": "z"}]
+ cd.append_parquet("ann_meta", rows2, id_col="announcement_id")
+ df = pd.read_parquet(part)
+ assert sorted(df["announcement_id"]) == ["a1", "a2", "a3"]
+ assert not list((root / "ann_meta" / f"dt={today}").glob("*.tmp")) # 无残留 tmp
+
+
+def test_append_parquet_crash_leaves_no_partial(root):
+ """写中途抛错 → 不留 tmp/半截文件(下次重拉幂等前提)。"""
+ import pandas as pd
+ with patch.object(pd.DataFrame, "to_parquet", side_effect=RuntimeError("disk")):
+ with pytest.raises(RuntimeError):
+ cd.append_parquet("ann_meta", [{"announcement_id": "a1"}],
+ id_col="announcement_id")
+ today = dt.date.today().isoformat()
+ assert not list((root / "ann_meta" / f"dt={today}").glob("*"))
+
+
+# ---------- unit marker ----------
+
+def test_marker_done_and_skip(root):
+ assert not cd.is_done("daily", "ann", "000001")
+ cd.mark_done("daily", "ann", "000001")
+ assert cd.is_done("daily", "ann", "000001")
+ assert not cd.is_done("daily", "ann", "600519")
+ assert not cd.is_done("backfill", "ann", "000001") # lane 隔离
+
+
+# ---------- 单实例锁(flock, 跨容器语义) ----------
+
+def test_lock_acquire_second_refused_release_ok(root):
+ fd1 = cd.acquire_lock()
+ assert fd1 is not None
+ fd2 = cd.acquire_lock()
+ assert fd2 is None # 让路
+ import os
+ os.close(fd1) # 模拟进程退出(内核释放 flock)
+ assert cd.acquire_lock() is not None
+
+
+def test_lock_stale_never(tmp_path):
+ """flock 由内核持有,进程死即释放 — 无陈旧锁文件问题(优于 pid 检测)。"""
+ lock = tmp_path / "corpus.lock"
+ lock.write_text("999999") # 内容无所谓
+ import os
+ f = open(lock, "r+")
+ fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
+ f.close() # 关闭即释放
+ f2 = open(lock, "r+")
+ fcntl.flock(f2, fcntl.LOCK_EX | fcntl.LOCK_NB) # 不抛 = 无陈旧
+ f2.close()
+
+
+# ---------- DomainClient(限速/冷却/429) ----------
+
+def test_client_cooldown_after_10_consecutive_errors(root):
+ c = cd.DomainClient("cninfo")
+ sess = MagicMock()
+ sess.request = MagicMock(side_effect=cd.requests.ConnectionError("timeout"))
+ c.session = sess
+ with pytest.raises(cd.DomainCooldown):
+ for _ in range(11):
+ try:
+ c.request("POST", "http://x")
+ except cd.TransportError:
+ pass # 未达阈值的传输错让单元层处理, 这里只数到冷却
+ assert c.consecutive_errors == 10
+
+
+def test_client_429_marks_rate_limited(root):
+ c = cd.DomainClient("cninfo")
+ sess = MagicMock()
+ sess.request = MagicMock(return_value=_resp(status=429))
+ c.session = sess
+ with pytest.raises(cd.DomainCooldown):
+ c.request("POST", "http://x")
+ assert c.rate_limited is True
+
+
+def test_client_success_resets_error_counter(root):
+ c = cd.DomainClient("cninfo")
+ seq = [_resp(status=500)] * 9 + [_resp(json_data={"ok": 1})] * 3
+ sess = MagicMock()
+ sess.request = MagicMock(side_effect=seq)
+ c.session = sess
+ for r in seq:
+ try:
+ c.request("POST", "http://x")
+ except cd.TransportError:
+ pass
+ assert c.consecutive_errors == 0
+
+
+# ---------- fetch 层(分页/两天窗) ----------
+
+def test_fetch_cninfo_paginates_till_short_page(root):
+ c = cd.DomainClient("cninfo")
+ pages = [_resp(json_data=_cninfo_page([_ann_raw(aid=f"a{i}") for i in range(30)],
+ total=45)),
+ _resp(json_data=_cninfo_page([_ann_raw(aid="a45")], total=45))]
+ sess = MagicMock()
+ sess.request = MagicMock(side_effect=pages)
+ c.session = sess
+ rows = cd.fetch_cninfo_announcements(c, "000001", "gssz0000001",
+ "2026-09-05", "2026-09-06")
+ assert len(rows) == 31
+ sent = sess.request.call_args_list[0][1]["data"]
+ assert sent["seDate"] == "2026-09-05~2026-09-06" # 两天窗原样
+ assert sent["stock"] == "000001,gssz0000001" # orgId 必带
+
+
+def test_fetch_cninfo_empty_is_legal(root):
+ c = cd.DomainClient("cninfo")
+ sess = MagicMock()
+ sess.request = MagicMock(return_value=_resp(json_data=_cninfo_page([])))
+ c.session = sess
+ rows = cd.fetch_cninfo_announcements(c, "000001", "o", "2026-09-05", "2026-09-06")
+ assert rows == []
+
+
+def test_fetch_news_recent_parses_jsonp(root):
+ c = cd.DomainClient("eastmoney")
+ inner = [{"code": "art1", "date": "2026-09-06 01:02:03", "title": "t",
+ "content": "c", "mediaName": "m", "image": ""}]
+ text = "cb(" + json.dumps({"result": {"cmsArticleWebOld": inner}}) + ")"
+ sess = MagicMock()
+ sess.request = MagicMock(return_value=_resp(text=text))
+ c.session = sess
+ rows = cd.fetch_news_recent(c, "600519")
+ assert len(rows) == 1 and rows[0]["code"] == "art1"
+ p = sess.request.call_args_list[0][1]["params"]
+ assert json.loads(p["param"])["keyword"] == "600519"
+
+
+def test_fetch_news_backfill_page(root):
+ c = cd.DomainClient("eastmoney")
+ lst = [{"Art_Code": "a1", "Art_ShowTime": "2024-01-01 00:00:00",
+ "Art_Title": "t", "Art_Url": "http://x/a1.html", "Np_dst": "CMS"}]
+ sess = MagicMock()
+ sess.request = MagicMock(return_value=_resp(
+ json_data={"data": {"list": lst}}))
+ c.session = sess
+ rows = cd.fetch_news_backfill_page(c, "600519", mkt_prefix="1", page=3)
+ assert rows[0]["Art_Code"] == "a1"
+ url = sess.request.call_args_list[0][0][1]
+ assert "mTypeAndCode=1.600519" in url and "pageIndex=3" in url
+
+
+# ---------- daily lane 集成(全 mock) ----------
+
+def _wire_daily(monkeypatch, root, ann_pages=None, news_rows=None,
+ article_html="正文内容"):
+ ann_pages = ann_pages or [_cninfo_page([_ann_raw(aid="a1", code="000001")])]
+ news_rows = news_rows if news_rows is not None else [
+ {"code": "art1", "date": "2026-09-06 10:00:00", "title": "新闻一",
+ "content": "摘要", "mediaName": "源", "image": ""}]
+ monkeypatch.setattr(cd, "load_stock_pool", lambda **kw: POOL)
+ monkeypatch.setattr(cd, "fetch_cninfo_announcements",
+ MagicMock(return_value=list(ann_pages[0]["announcements"])))
+ monkeypatch.setattr(cd, "fetch_news_recent", MagicMock(return_value=news_rows))
+ monkeypatch.setattr(cd, "fetch_article_html", MagicMock(return_value=article_html))
+ monkeypatch.setattr(cd, "download_pdf", MagicMock(return_value=1024))
+
+
+def test_daily_happy_path_writes_all_domains(root, monkeypatch):
+ import pandas as pd
+ _wire_daily(monkeypatch, root)
+ rc = cd.run_lane("daily")
+ assert rc == 0
+ today = dt.date.today().isoformat()
+ ann = pd.read_parquet(root / "ann_meta" / f"dt={today}" / "part-0.parquet")
+ assert len(ann) == 1 and ann.iloc[0]["announcement_id"] == "a1"
+ news = pd.read_parquet(root / "news_meta" / f"dt={today}" / "part-0.parquet")
+ assert len(news) == 2 # 每股 1 条 × 2 股
+ ft = pd.read_parquet(root / "news_fulltext" / f"dt={today}" / "part-0.parquet")
+ assert len(ft) == 1 and "正文内容" in ft.iloc[0]["content_text"]
+ # markers: ann/news 每股 done;fulltext 每 art done
+ assert cd.is_done("daily", "ann", "000001") and cd.is_done("daily", "ann", "600519")
+ assert cd.is_done("daily", "news", "600519")
+ assert cd.is_done("daily", "fulltext", "art1")
+
+
+def test_daily_rerun_idempotent_no_dup(root, monkeypatch):
+ import pandas as pd
+ _wire_daily(monkeypatch, root)
+ assert cd.run_lane("daily") == 0
+ # markers 全 done → 第二趟零 fetch;强制清 marker 重跑同数据 → 零重复行
+ for code, _ in POOL:
+ (root / "state" / "markers" / "daily" / "ann" / f"{code}.done").unlink()
+ (root / "state" / "markers" / "daily" / "news" / f"{code}.done").unlink()
+ assert cd.run_lane("daily") == 0
+ today = dt.date.today().isoformat()
+ ann = pd.read_parquet(root / "ann_meta" / f"dt={today}" / "part-0.parquet")
+ news = pd.read_parquet(root / "news_meta" / f"dt={today}" / "part-0.parquet")
+ assert len(ann) == 1 and len(news) == 2 # id 去重,零重写
+
+
+def test_daily_two_day_window_relative_dates(root, monkeypatch):
+ """seDate = (today-1)~today 相对构造,漏日自愈契约。"""
+ _wire_daily(monkeypatch, root)
+ cd.run_lane("daily")
+ sedate = cd.fetch_cninfo_announcements.call_args[0][3:5]
+ t = dt.date.today()
+ assert sedate == ((t - dt.timedelta(days=1)).isoformat(), t.isoformat())
+
+
+def test_daily_two_day_window_after_missed_day(root, monkeypatch):
+ """停跑一天 → 次日单次执行补齐: fetch 窗含昨天,昨日公告照入。"""
+ import pandas as pd
+ _wire_daily(monkeypatch, root)
+ cd.run_lane("daily")
+ # 清 marker 模拟「另一天再来」;同 id 仍是昨天发的 → 已在账,不重
+ for code, _ in POOL:
+ (root / "state" / "markers" / "daily" / "ann" / f"{code}.done").unlink()
+ cd.run_lane("daily")
+ today = dt.date.today().isoformat()
+ ann = pd.read_parquet(root / "ann_meta" / f"dt={today}" / "part-0.parquet")
+ assert len(ann) == 1
+
+
+def test_daily_unit_fail_not_marked_and_rc1(root, monkeypatch):
+ """fetch 单股失败 → 该股不标 done,它股继续,整体 rc=1 留痕。"""
+ import pandas as pd
+ _wire_daily(monkeypatch, root)
+ calls = {"n": 0}
+
+ def flaky(client, code, org, start, end, category=""):
+ calls["n"] += 1
+ if code == "000001":
+ raise cd.TransportError("boom")
+ return [_ann_raw(aid="a2", code=code)]
+
+ monkeypatch.setattr(cd, "fetch_cninfo_announcements", flaky)
+ rc = cd.run_lane("daily")
+ assert rc == 1
+ assert not cd.is_done("daily", "ann", "000001")
+ assert cd.is_done("daily", "ann", "600519")
+ today = dt.date.today().isoformat()
+ ann = pd.read_parquet(root / "ann_meta" / f"dt={today}" / "part-0.parquet")
+ assert len(ann) == 1 # 只有好股的
+
+
+def test_daily_domain_cooldown_skips_domain_but_others_run(root, monkeypatch):
+ """巨潮连续错×10 → 该域冷却,东财域照跑;rc=1。"""
+ import pandas as pd
+ monkeypatch.setattr(cd, "load_stock_pool", lambda **kw: POOL)
+ monkeypatch.setattr(cd, "fetch_cninfo_announcements",
+ MagicMock(side_effect=cd.TransportError("down")))
+ monkeypatch.setattr(cd, "fetch_news_recent", MagicMock(return_value=[
+ {"code": "art1", "date": "2026-09-06 10:00:00", "title": "t",
+ "content": "c", "mediaName": "m", "image": ""}]))
+ monkeypatch.setattr(cd, "fetch_article_html",
+ MagicMock(return_value="x"))
+ monkeypatch.setattr(cd, "download_pdf", MagicMock(return_value=1))
+ rc = cd.run_lane("daily")
+ assert rc == 1
+ today = dt.date.today().isoformat()
+ news = pd.read_parquet(root / "news_meta" / f"dt={today}" / "part-0.parquet")
+ assert len(news) == 2 # 东财域全量完成
+ assert not cd.is_done("daily", "ann", "000001") # 巨潮域整体未标
+
+
+def test_daily_429_gives_rc2(root, monkeypatch):
+ monkeypatch.setattr(cd, "load_stock_pool", lambda **kw: POOL)
+ monkeypatch.setattr(cd, "fetch_cninfo_announcements",
+ MagicMock(return_value=[]))
+ monkeypatch.setattr(cd, "fetch_news_recent",
+ MagicMock(side_effect=cd.DomainCooldown("eastmoney",
+ rate_limited=True)))
+ monkeypatch.setattr(cd, "fetch_article_html", MagicMock(return_value="x"))
+ monkeypatch.setattr(cd, "download_pdf", MagicMock(return_value=1))
+ # 东财域 429 冷却 → rc=2 限流让路
+ rc = cd.run_lane("daily")
+ assert rc == 2
+
+
+def test_daily_limit_smoke_writes_but_no_markers(root, monkeypatch):
+ """--limit 冒烟: 真数据可写,绝不落 marker(5m 09-17 教训同款)。"""
+ _wire_daily(monkeypatch, root)
+ rc = cd.run_lane("daily", limit=1)
+ assert rc == 0
+ assert not any((root / "state" / "markers").rglob("*.done"))
+
+
+def test_daily_express_title_routes_to_pdf(root, monkeypatch):
+ """标题含「业绩快报」→ 进当日 PDF 待下清单(无服务端码,客户端谓词)。"""
+ todo = cd.collect_pdf_candidates(
+ [_ann_raw(aid="e1", title="2025年度业绩快报"),
+ _ann_raw(aid="n1", title="2025年年度报告")])
+ assert [t["announcementId"] for t in todo] == ["e1"]
+
+
+def test_daily_pdf_downloads_fivecat_and_express(root, monkeypatch):
+ import pandas as pd
+ _wire_daily(monkeypatch, root)
+
+ def fake_download(client, url, dest):
+ dest.parent.mkdir(parents=True, exist_ok=True)
+ dest.write_bytes(b"%PDF-1.4 fake")
+ return len(b"%PDF-1.4 fake")
+
+ fivecat_rows = [_ann_raw(aid="r1", title="2025年年度报告")]
+ monkeypatch.setattr(cd, "fetch_cninfo_announcements",
+ MagicMock(return_value=fivecat_rows))
+ monkeypatch.setattr(cd, "download_pdf", fake_download)
+ cd.run_lane("daily")
+ idx = pd.read_parquet(root / "state" / "ann_pdf_index.parquet")
+ assert len(idx) == 1
+ assert idx.iloc[0]["announcement_id"] == "r1"
+ assert idx.iloc[0]["report_year"] == "2025"
+ assert (root / "ann_pdf" / "2026" / "r1.pdf").read_bytes().startswith(b"%PDF")
+
+
+# ---------- 墙钟自停 ----------
+
+def test_wallclock_stops_after_current_unit(root, monkeypatch):
+ """--until 已过 → 完成当前 unit 后退出,rc=3,已完 unit 有 marker。"""
+ _wire_daily(monkeypatch, root)
+ past = (dt.datetime.now() - dt.timedelta(minutes=1)).strftime("%H:%M")
+ rc = cd.run_lane("daily", until=past)
+ assert rc == 3
+ assert cd.is_done("daily", "ann", "000001") # 首 unit 完成后停
+
+
+def test_wallclock_future_runs_to_completion(root, monkeypatch):
+ _wire_daily(monkeypatch, root)
+ future = dt.datetime.now() + dt.timedelta(minutes=30)
+ if future.date() != dt.date.today(): # 跨午夜 → 取当天最晚时刻
+ future = dt.datetime.combine(dt.date.today(), dt.time(23, 59))
+ assert cd.run_lane("daily", until=future.strftime("%H:%M")) == 0
+
+
+# ---------- depth_summary(探测优先) ----------
+
+def test_depth_summary_records_earliest(root, monkeypatch):
+ _wire_daily(monkeypatch, root)
+ cd.run_lane("daily")
+ j = json.loads((root / "state" / "depth_summary.json").read_text())
+ assert "ann" in j or "news" in j
+
+
+# ---------- main/退出码 ----------
+
+def test_main_lock_busy_yields_exit3(root, monkeypatch):
+ import os
+ fd = cd.acquire_lock()
+ assert fd is not None
+ monkeypatch.setattr(sys, "argv", ["corpus_download.py", "--lane", "daily"])
+ with pytest.raises(SystemExit) as e:
+ cd.main()
+ os.close(fd)
+ assert e.value.code == 3 # 让路(2/3 类)
+
+
+def test_main_daily_exit0(root, monkeypatch):
+ _wire_daily(monkeypatch, root)
+ monkeypatch.setattr(sys, "argv", ["corpus_download.py", "--lane", "daily"])
+ with pytest.raises(SystemExit) as e:
+ cd.main()
+ assert e.value.code == 0