hz4th_coder/news-tracker
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v1.3.0: 每源独立采集周期 + 定制监控与新闻监控分离(独立配置/独立汇总) + 历史采样留档与提取API

Browse Source
This commit is contained in:
hz4th_coder
2026-08-29 19:29:28 +08:00
parent 2cad7a2976
commit de08530958
12 changed files with 489 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scheduler.py
+123 -54
View File
@@ -1,8 +1,13 @@
# -*- coding: utf-8 -*-
"""
新闻智能跟踪系统 - 后台调度器
1. 定时采集（scan_interval_min 分钟一次）：拉取 → 入库 → 规则分析 → 后台LLM深度分析 → 实时通知
2. 每日汇总（默认 10:00）：汇总昨天至今重要资讯邮件
两套独立机制（互不混用，各自单独配置）：
1. 新闻监控机制（普通源 normal）：统一采集间隔 scan_interval_min + 每日新闻日报 + 实时重要资讯
2. 定制监控机制（定制源 custom）：统一采集间隔 custom_scan_interval_min + 定制监控汇总 + 命中实时推送
每个数据源都可用 scan_interval_min 覆盖所属机制的全局采集间隔（0=跟随全局）。
每次采集都会写入 source_snapshots 历史采样表，供查看与自动流程提取。
"""
import threading
import time
@@ -10,23 +15,66 @@ from datetime import datetime, timedelta
import config
import db
import simulate
import simulate # noqa: F401 (保留引用，crawler 内部使用)
import crawler
import analysis
import notifier
def collect_once():
"""执行一次采集全流程，返回新增数
真实 URL 源走网页抓取+正文清洗+全文入库；模拟源(example.com)回退仿真数据。
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 控制，两套独立。
"""
if not int(db.get_setting("auto_collect", config.AUTO_DEFAULTS["auto_collect"])):
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, per_source = crawler.fetch_all()
added = 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])
for it in items:
if db.article_exists(it["url"]):
continue
@@ -35,7 +83,7 @@ def collect_once():
added += 1
# 后台 LLM 深度分析
analysis.run_llm_background()
# 实时通知
# 实时通知（新闻重要资讯 + 定制监控命中）
try:
send_realtime_if_needed()
except Exception:
@@ -44,21 +92,33 @@ def collect_once():
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"]
"""扫描已分析完成、重要、未通知的资讯发实时邮件。
普通源受 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"]))
@@ -71,56 +131,65 @@ def send_daily_summary():
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"]))
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, mm = hm.split(":")
hh, mm = int(hh), int(mm)
hh = int(hm.split(":")[0])
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
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))
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:
# 每日汇总
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))
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:
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))
# 实时通知（LLM 分析完成后推送）
try:
send_realtime_if_needed()
except Exception:
pass
# 30s 轮询粒度，兼顾每源自定义的短周期（如 5 分钟）
stop_event.wait(30)
return