# -*- 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 crawler import analysis import notifier def collect_once(): """执行一次采集全流程,返回新增数 真实 URL 源走网页抓取+正文清洗+全文入库;模拟源(example.com)回退仿真数据。 """ 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, per_source = crawler.fetch_all() added = 0 for it in items: if db.article_exists(it["url"]): continue aid = db.add_article(it) analysis.analyze_article(aid) added += 1 # 后台 LLM 深度分析 analysis.run_llm_background() # 实时通知 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