news-tracker/scheduler.py

# -*- 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, persist_key):
    """到点触发的通用逻辑。time_key 存 'HH:MM'，send_fn 返回发送条数。
    last_day 存上次发送的日期字符串 'YYYY-MM-DD'（None=尚未发送）。
    修复：每天只发一次——last_day 必须是纯字符串，不能嵌套元组；
    并把已发送日期持久化到 settings，服务重启后也不会当天重复发。"""
    day_key = now.strftime("%Y-%m-%d")
    if last_day is None:
        last_day = db.get_setting(persist_key, "")
    if last_day == 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()
            db.set_setting(persist_key, day_key)
            return 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", "summary_last_day")
            last_custom_summary_day = _maybe_daily(now, last_custom_summary_day, "custom_summary_time",
                                                   send_custom_summary, "custom_summary", "custom_summary_last_day")
        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