# -*- coding: utf-8 -*- """ 新闻智能跟踪系统 - 后台调度器 两套独立机制（互不混用，各自单独配置）： 1. 新闻监控机制（普通源 normal）：统一采集间隔 scan_interval_min + 每日新闻日报 + 实时重要资讯 2. 定制监控机制（定制源 custom）：统一采集间隔 custom_scan_interval_min + 定制监控汇总 + 命中实时推送 每个数据源都可用 scan_interval_min 覆盖所属机制的全局采集间隔（0=跟随全局）。 每次采集都会写入 source_snapshots 历史采样表，供查看与自动流程提取。 """ import threading import time from datetime import datetime, timedelta import config import db import simulate # noqa: F401 (保留引用，crawler 内部使用) import crawler import analysis import notifier def source_interval(s): """单个数据源的实际采集间隔(分钟)：优先本源自定义值，否则跟随所属机制全局值""" iv = int(s.get("scan_interval_min") or 0) if iv > 0: return iv if s.get("kind") == "custom": return max(1, int(db.get_setting("custom_scan_interval_min", config.CUSTOM_DEFAULTS["custom_scan_interval_min"]))) return max(1, int(db.get_setting("scan_interval_min", config.AUTO_DEFAULTS["scan_interval_min"]))) def _is_due(s): """判断该源是否到点需要采集""" interval = source_interval(s) last = (s.get("last_fetch") or "").strip() if not last: return True try: lt = datetime.strptime(last, "%Y-%m-%d %H:%M:%S") return (datetime.now() - lt).total_seconds() / 60 >= interval except Exception: return True def collect_once(force=False): """采集到期的数据源，返回新增条数。 force=True：忽略周期，全部采集（用于手动「立即采集」）。 新闻源受 auto_collect 控制，定制源受 custom_enabled 控制，两套独立。 """ auto = int(db.get_setting("auto_collect", config.AUTO_DEFAULTS["auto_collect"])) custom = int(db.get_setting("custom_enabled", config.CUSTOM_DEFAULTS["custom_enabled"])) if not auto and not custom: return 0 sources = db.list_sources(only_enabled=True) if not sources: return 0 items, added = [], 0 for s in sources: if s.get("kind") == "custom": if not custom: continue else: if not auto: continue if not force and not _is_due(s): continue try: got = crawler.fetch_source(s) db.update_source_fetch(s["id"], status="ok", count=len(got)) db.add_source_snapshot(s["id"], len(got), "ok", "") items.extend(got) except Exception as e: db.update_source_fetch(s["id"], status="error", count=0) db.add_source_snapshot(s["id"], 0, "error", str(e)[:300]) notifier.report_error("采集", f"数据源「{s['name']}」采集失败", str(e)[:300]) 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 as e: notifier.report_error("实时通知", "实时重要资讯推送异常", str(e)[:300]) return added def send_realtime_if_needed(): """扫描已分析完成、重要、未通知的资讯发实时邮件。 普通源受 realtime_enabled 控制；定制源受 custom_enabled 控制。""" auto_rt = int(db.get_setting("realtime_enabled", config.AUTO_DEFAULTS["realtime_enabled"])) custom_on = int(db.get_setting("custom_enabled", config.CUSTOM_DEFAULTS["custom_enabled"])) arts = db.list_articles(is_important=1, order="a.total_score DESC", limit=30) pending = [] for a in arts: if a["notified"] or a["llm_status"] == "pending": continue if a.get("source_id"): src = db.get_source(a["source_id"]) if src and src.get("kind") == "custom": if custom_on: pending.append(a) else: if auto_rt: pending.append(a) elif auto_rt: pending.append(a) if not pending: return 0 batch = pending[:10] return notifier.send_realtime(batch) def send_daily_summary(): """新闻机制：每日日报（普通源），默认每天 10:00""" 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 send_custom_summary(): """定制机制：定制监控汇总（定制源命中），独立时间单独配置""" if not int(db.get_setting("custom_summary_enabled", config.CUSTOM_DEFAULTS["custom_summary_enabled"])): return 0 window = int(db.get_setting("custom_summary_window_hours", config.CUSTOM_DEFAULTS["custom_summary_window_hours"])) articles = db.custom_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_custom_summary(articles, label) def _maybe_daily(now, last_day, time_key, send_fn, kind): """到点触发的通用逻辑。time_key 存 'HH:MM'，send_fn 返回发送条数。""" day_key = now.strftime("%Y-%m-%d") if last_day[0] == day_key: return last_day hm = str(db.get_setting(time_key, config.AUTO_DEFAULTS.get(time_key) or config.CUSTOM_DEFAULTS.get(time_key) or "10:00")) try: hh = int(hm.split(":")[0]) except Exception: hh = 0 if now.strftime("%H:%M") >= hm and now.hour >= hh: try: send_fn() last_day = (day_key,) except Exception as e: db.add_log(kind, f"{time_key} 汇总异常", 0, [], status="error", detail=str(e)) notifier.report_error("汇总", f"{time_key} 汇总发送异常", str(e)[:300]) return last_day def scheduler_loop(stop_event): last_summary_day = None last_custom_summary_day = None while not stop_event.is_set(): now = datetime.now() try: last_summary_day = _maybe_daily(now, (last_summary_day,), "summary_time", send_daily_summary, "summary") last_custom_summary_day = _maybe_daily(now, (last_custom_summary_day,), "custom_summary_time", send_custom_summary, "custom_summary") except Exception as e: notifier.report_error("汇总", "定时汇总检查异常", str(e)[:300]) # 采集到期数据源（每源独立周期） try: collect_once() except Exception as e: db.add_log("realtime", "采集异常", 0, [], status="error", detail=str(e)) notifier.report_error("采集", "定时采集异常", str(e)[:300]) # 实时通知（LLM 分析完成后推送） try: send_realtime_if_needed() except Exception as e: notifier.report_error("实时通知", "实时推送检查异常", str(e)[:300]) # 系统错误通知（冷却频率 + 静默时段控制，静默结束后自动补发） try: notifier.flush_pending_errors() except Exception: pass # 定期清理过期错误日志（每天一次） try: db.clear_old_errors(days=30) except Exception: pass # 30s 轮询粒度，兼顾每源自定义的短周期（如 5 分钟） stop_event.wait(30) 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