220 lines
8.8 KiB
Python
220 lines
8.8 KiB
Python
# -*- 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
|