新闻智能跟踪系统 v1.0.0: AI资讯自动采集+智能分析+邮件通知+网页管理台

scheduler.py

@@ -0,0 +1,132 @@
+# -*- coding: utf-8 -*-
+"""
+新闻智能跟踪系统 - 后台调度器
+1. 定时采集（scan_interval_min 分钟一次）：拉取 → 入库 → 规则分析 → 后台LLM深度分析 → 实时通知
+2. 每日汇总（默认 10:00）：汇总昨天至今重要资讯邮件
+"""
+import threading
+import time
+from datetime import datetime, timedelta
+
+import config
+import db
+import simulate
+import analysis
+import notifier
+
+
+def collect_once():
+    """执行一次采集全流程，返回新增数"""
+    if not int(db.get_setting("auto_collect", config.AUTO_DEFAULTS["auto_collect"])):
+        return 0
+    sources = db.list_sources(only_enabled=True)
+    if not sources:
+        return 0
+    items = simulate.fetch_simulated()
+    added = 0
+    for it in items:
+        if db.article_exists(it["url"]):
+            continue
+        it["source_id"] = simulate._source_id_for_domain(it["domain"])
+        aid = db.add_article(it)
+        analysis.analyze_article(aid)
+        added += 1
+    # 后台 LLM 深度分析
+    analysis.run_llm_background()
+    if added:
+        for s in sources:
+            db.update_source_fetch(s["id"], status="ok", count=added)
+    # 实时通知
+    try:
+        send_realtime_if_needed()
+    except Exception:
+        pass
+    return added
+
+
+def send_realtime_if_needed():
+    """扫描已分析完成、重要、未通知的资讯，发实时邮件"""
+    if not int(db.get_setting("realtime_enabled", config.AUTO_DEFAULTS["realtime_enabled"])):
+        return 0
+    # LLM 深度分析完成后，重新判定重要度并通知
+    arts = db.list_articles(is_important=1, order="a.total_score DESC", limit=20)
+    pending = [a for a in arts if not a["notified"] and a["llm_status"] != "pending"]
+    if not pending:
+        return 0
+    # 批量发（控制每封数量）
+    batch = pending[:10]
+    return notifier.send_realtime(batch)
+
+
+def send_daily_summary():
+    """每日汇总：昨天至今的重要/相关资讯"""
+    if not int(db.get_setting("summary_enabled", config.AUTO_DEFAULTS["summary_enabled"])):
+        return 0
+    window = int(db.get_setting("summary_window_hours", config.AUTO_DEFAULTS["summary_window_hours"]))
+    articles = db.latest_articles_for_summary(window)
+    if not articles:
+        return 0
+    start = (datetime.now() - timedelta(hours=window)).strftime("%m-%d %H:%M")
+    end = datetime.now().strftime("%m-%d %H:%M")
+    label = f"{start} ~ {end}"
+    return notifier.send_daily_summary(articles, label)
+
+
+def _next_summary_run():
+    """计算下一次汇总时间点（默认每天 10:00，可配置）"""
+    hm = str(db.get_setting("summary_time", config.AUTO_DEFAULTS["summary_time"]))
+    try:
+        hh, mm = hm.split(":")
+        hh, mm = int(hh), int(mm)
+    except Exception:
+        hh, mm = 10, 0
+    now = datetime.now()
+    nxt = now.replace(hour=hh, minute=mm, second=0, microsecond=0)
+    if nxt <= now:
+        nxt = nxt + timedelta(days=1)
+    return nxt
+
+
+def scheduler_loop(stop_event):
+    last_summary_day = None
+    while not stop_event.is_set():
+        try:
+            # 每日汇总
+            now = datetime.now()
+            day_key = now.strftime("%Y-%m-%d")
+            if last_summary_day != day_key:
+                hm = str(db.get_setting("summary_time", config.AUTO_DEFAULTS["summary_time"]))[:5]
+                if now.strftime("%H:%M") >= hm and now.hour >= int(hm.split(":")[0]):
+                    try:
+                        send_daily_summary()
+                        last_summary_day = day_key
+                    except Exception as e:
+                        db.add_log("summary", "每日汇总异常", 0, [], status="error", detail=str(e))
+        except Exception:
+            pass
+
+        # 定时采集（以分钟为单位）
+        interval = int(db.get_setting("scan_interval_min", config.AUTO_DEFAULTS["scan_interval_min"]))
+        next_scan = time.time() + interval * 60
+        # 在等待期间兼顾实时通知（LLM 分析完成后推送）
+        while time.time() < next_scan and not stop_event.is_set():
+            try:
+                send_realtime_if_needed()
+            except Exception:
+                pass
+            stop_event.wait(min(30, max(5, interval * 60)))
+
+        if stop_event.is_set():
+            break
+        try:
+            collect_once()
+        except Exception as e:
+            db.add_log("realtime", "采集异常", 0, [], status="error", detail=str(e))
+    return
+
+
+def start_scheduler():
+    stop_event = threading.Event()
+    t = threading.Thread(target=scheduler_loop, args=(stop_event,), daemon=True)
+    t.start()
+    return stop_event, t