feat(data): T2 corpus_download.py daily lane+断点+锁+36测全绿 [nas]
spec §18.7 T2(corpus 栈唯一新脚本,单脚本红线): - daily 四段: 公告增量(巨潮 T-1~T 两天窗自愈)/新闻增量(search-api 带摘要, (art_code,stock_code) 复合键=同稿多股 meta 多行)/新闻全文(昨日新 art_code, art_code 全局一条)/当日 PDF(五类服务端码 category 码+业绩快报标题谓词—— 实测 yjkb 码不存在,无效码被服务端静默忽略返回全量,故走客户端标题过滤) - 断点: unit marker tmp+rename;--until 墙钟=完成当前 unit 后 rc=3;--limit 冒烟 绝不落 marker(5m 09-17 教训) - 单实例锁: fcntl.flock(state/corpus.lock)——跨容器语义(bind mount 同 inode, 内核持有/进程死自动释放);5m 的 pid 活性检测在 docker run --rm 各自独立 PID namespace 下失效,corpus 换 flock(让路契约同款,锁忙 rc=3) - id 去重双层: IdLedger 全局账本(state/ids_*.parquet int64 hash,满足 §18.5 announcement_id 全局无重)+RecentIndex 近30日分区索引(兜账本flush周期崩溃窗) - 失败语义: 传输错 unit 不标done;真空=合法done;4xx确定性跳过不烧重试; 域名连续错×10或429冷却它域继续;rc 0完成/1致命/2限流/3墙钟或锁 - 股票池: 巨潮 szse_stock.json(实测含退市000003+北交920001,6247条)A前缀过滤, 非NAS副本库——orgId 为按股查询硬前提且退市覆盖更全(spec 偏差T6回写) - 测试 36 个全 mock 零网络零写死日期;全套 253 绿
This commit is contained in:
@@ -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("<em>", "").replace("</em>", "")
|
||||
.replace("(<em>", "").replace("</em>)", ""))
|
||||
|
||||
|
||||
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)[^>]*>.*?</\1>", " ", 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_<domain>.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": "<em>",
|
||||
"postTag": "</em>"}}}
|
||||
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()
|
||||
Reference in New Issue
Block a user