138 lines
5.3 KiB
Python
138 lines
5.3 KiB
Python
#!/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 = 8 # 8 批（2026-08-13 由4改8：4272只/4批=1068只×0.9s=961s超600s超时，改8批每批534只×0.9s=481s安全）
|
||
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
|
||
|
||
# 2026-08-14 并发采集：ThreadPoolExecutor 3 并发 + Semaphore 限速（防东财封）
|
||
# 原因：实际 2-3s/只（非估算0.9s），8批502只×2.5s=1255s>600s超时
|
||
# 3并发：502/3≈167串行×2.5s=418s安全；3并发+SLEEP=每秒2-3请求（频率可控）
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
import threading
|
||
_rate_lock = threading.Lock()
|
||
_req_times = []
|
||
|
||
def _fetch_one(code):
|
||
"""单只采集（并发内）"""
|
||
nonlocal ok, fail, empty, new
|
||
try:
|
||
arts = fetch_stock_news(code, max_pages=1)
|
||
if not arts:
|
||
with _rate_lock:
|
||
empty += 1
|
||
return None
|
||
n = 0
|
||
for a in arts:
|
||
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
|
||
with _rate_lock:
|
||
new += n
|
||
ok += 1
|
||
return code
|
||
except Exception as e:
|
||
with _rate_lock:
|
||
fail += 1
|
||
if fail <= 5:
|
||
print(f" FAIL {code}: {str(e)[:60]}", flush=True)
|
||
return None
|
||
|
||
_done = 0
|
||
with ThreadPoolExecutor(max_workers=3) as pool:
|
||
futures = {pool.submit(_fetch_one, c): c for c in codes}
|
||
for f in as_completed(futures):
|
||
_done += 1
|
||
if _done % 50 == 0:
|
||
conn.commit()
|
||
print(f" [{_done}/{len(codes)}] ok={ok} empty={empty} fail={fail} 新增={new} | {time.time()-t0:.0f}s", flush=True)
|
||
|
||
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()
|