fix: news_collector并发采集——ThreadPoolExecutor 3并发+Semaphore限速(实际2-3s/只非0.9s估算,502只×2.5s=1255s>600s,3并发→418s安全)

deploy/profile-scripts/news_collector_full.py

@@ -75,15 +75,26 @@ def main():
     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):
+
+    # 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)  # 第一页 50 条
+            arts = fetch_stock_news(code, max_pages=1)
             if not arts:
-                empty += 1
-                continue
+                with _rate_lock:
+                    empty += 1
+                return None
             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:
@@ -96,16 +107,25 @@ def main():
                             n += 1
                     except Exception:
                         pass
-            new += n
-            ok += 1
+            with _rate_lock:
+                new += n
+                ok += 1
+            return code
         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)
+            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()