133 lines
4.5 KiB
Python
133 lines
4.5 KiB
Python
# -*- 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
|