feat: 缺口2修复——news_collector_full全市场新闻采集(4013只东财search-api按code搜,只存近3天,4批cron盘前7:30-8:00)+4个包装脚本

2026-08-12 09:09:53 +08:00
parent 8879815b0f
commit e876e85a85
1 changed files with 117 additions and 0 deletions
@@ -0,0 +1,117 @@
+#!/usr/bin/env python3
+"""news_collector_full.py — 全市场新闻采集（stock_news 日常刷新）
+
+背景（2026-08-12 架构补缺）：
+  stock_news 403万行 8/4 停更。现有 news_collector/ths_news 只覆盖 v8.1 的 58 只，
+  未注册 cron。p_oversold 的 news3 因子需要全市场个股近3日新闻数。
+
+数据源：东财 search-api（news_collector.fetch_stock_news，keyword=code 翻页）
+  老莫改直连后实测可用（600519 返回 50 条含当日最新）。
+策略：全市场 4013 只按 code 逐个搜，每只取第一页(50条)只存近3天。
+  4 个 cron 分批(--batch 0/1/2/3 各~1000只)，0.4s 间隔防封，INSERT OR IGNORE 幂等。
+
+调度：盘前 4 批（8:20/8:30/8:40/8:50）+ 盘后 4 批（16:40/16:45/16:50/16:55）
+规范：单例守卫(5.3) + 限速防封 + 分批 commit + 近3天过滤
+"""
+import sys, os, json, time, sqlite3, fcntl
+from pathlib import Path
+from datetime import datetime, timedelta
+
+sys.path.insert(0, str(Path(__file__).parent))
+from news_collector import fetch_stock_news, init_table
+
+DB_PATH = Path("/home/hmo/MoFin/data/mofin.db")
+SLEEP = 0.4          # 请求间隔（防东财封）
+BATCHES = 4          # 4 批
+RECENT_DAYS = 3      # 只存近 3 天新闻（news3 因子口径）
+
+
+def _singleton_guard(tag):
+    lock_dir = Path("/tmp/mofin_locks")
+    lock_dir.mkdir(exist_ok=True)
+    try:
+        fd = os.open(str(lock_dir / f"{tag}.lock"), os.O_CREAT | os.O_RDWR)
+        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        return fd
+    except OSError:
+        print(f"[{tag}] 已有实例在运行，退出", flush=True)
+        sys.exit(0)
+
+
+def get_codes(conn, batch):
+    """全市场 A 股（stock_daily distinct），按 batch 分批"""
+    rows = conn.execute(
+        "SELECT DISTINCT code FROM stock_daily WHERE length(code)=6 ORDER BY code"
+    ).fetchall()
+    codes = [str(r[0]) for r in rows]
+    return [c for i, c in enumerate(codes) if i % BATCHES == batch]
+
+
+def main():
+    batch = 0
+    limit = 0
+    i = 1
+    while i < len(sys.argv):
+        a = sys.argv[i]
+        if a.startswith("--batch"):
+            batch = int(a.split("=")[-1] if "=" in a else sys.argv[i + 1])
+        if a.startswith("--limit"):
+            limit = int(a.split("=")[-1] if "=" in a else sys.argv[i + 1])
+        i += 1
+    tag = f"news_collector_full_b{batch}"
+    _fd = _singleton_guard(tag)
+
+    t0 = time.time()
+    print(f"[{tag}] {datetime.now().strftime('%H:%M:%S')} 全市场新闻采集(batch={batch}) 开始", flush=True)
+
+    conn = sqlite3.connect(str(DB_PATH), timeout=30)
+    init_table(conn)
+    codes = get_codes(conn, batch)
+    if limit > 0:
+        codes = codes[:limit]
+        print(f"  [测试模式] 只跑前 {limit} 只", flush=True)
+    print(f"  股票池: {len(codes)} 只 (batch {batch}/{BATCHES})", flush=True)
+
+    since = (datetime.now() - timedelta(days=RECENT_DAYS)).strftime("%Y-%m-%d")
+    cur = conn.cursor()
+    ok = fail = empty = new = 0
+    for idx, code in enumerate(codes, 1):
+        try:
+            arts = fetch_stock_news(code, max_pages=1)  # 第一页 50 条
+            if not arts:
+                empty += 1
+                continue
+            n = 0
+            for a in arts:
+                # 只存近 RECENT_DAYS 天（news3 口径），date 格式 'YYYY-MM-DD HH:MM' 或 'YYYY-MM-DD HH:MM:SS'
+                dstr = (a.get("date") or "")[:10]
+                if dstr and dstr >= since:
+                    try:
+                        cur.execute(
+                            "INSERT OR IGNORE INTO stock_news (code, title, content, source, date, url) "
+                            "VALUES (?,?,?,?,?,?)",
+                            (code, a.get("title", ""), a.get("content", ""),
+                             a.get("source", "eastmoney"), a.get("date", ""), a.get("url", "")))
+                        if cur.rowcount:
+                            n += 1
+                    except Exception:
+                        pass
+            new += n
+            ok += 1
+        except Exception as e:
+            fail += 1
+            if fail <= 5:
+                print(f"  FAIL {code}: {str(e)[:60]}", flush=True)
+        if idx % 50 == 0:
+            conn.commit()
+            print(f"  [{idx}/{len(codes)}] ok={ok} empty={empty} fail={fail} 新增={new} | {time.time()-t0:.0f}s", flush=True)
+        time.sleep(SLEEP)
+
+    conn.commit()
+    conn.close()
+    dt = time.time() - t0
+    print(f"[{tag}] 完成: {ok}/{len(codes)} 成功, {empty} 无新闻, {fail} 失败, 新增 {new} 条, 耗时 {dt:.0f}s", flush=True)
+
+
+if __name__ == "__main__":
+    main()