diff --git a/README.md b/README.md index 3aca103..e48a1cd 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ 一款面向 AI 领域的**自动采集 → 智能分析 → 邮件推送**的新闻跟踪系统。后台定时从数据源采集资讯，先规则打分、再 LLM 深度分析，把与你兴趣画像相关且重要的资讯**实时发邮件提醒**，并每天 10:00 发一份**AI 资讯日报**。带完整网页管理台，兴趣画像/数据源/通知参数全部可配。 -> 当前版本 **v1.2.0**：数据源**完整编辑**（所有配置可改，模态框）+ **定制监控类型**（无权重，按「推送标准」由大模型判断是否推送）。 +> 当前版本 **v1.3.0**：数据源**独立采集周期**（每源可覆盖全局）+ **定制监控与新闻监控完全分离**（独立间隔/独立汇总）+ **历史采样留档与提取**。 --- @@ -29,7 +29,9 @@ 2. **规则打分**（立即）：`analysis.analyze_article()` 对每条新资讯算**兴趣相关度**（关键词命中权重 + 领域匹配 + 关注公司命中）和**重要度启发式**（发布/融资/禁令等强信号词、金额量级、时效性、数据源权重、公司影响），得到**综合分 total_score (0-100)**。 3. **LLM 深度分析**（后台异步）：规则分 ≥ `llm_threshold`（默认 60）的资讯进入 `llm_analyze()`，**使用网页激活的大模型接口**（默认 SiliconFlow，可一键切换；调用失败自动切换下一个可用接口），输出重要度 1-10、相关度、分类、一句话结论，并与规则分融合更新。 4. **实时通知**：扫描「重要 + 已分析完成 + 未通知」的资讯，综合分 ≥ `realtime_threshold`（默认 80）→ 发 **🔥 重要AI资讯实时提醒** 邮件，标记 `notified=1`。 -5. **每日日报**（默认每天 10:00）：汇总往前 `summary_window_hours`（默认 24h）内综合分 ≥50 的资讯，按分排序取前 `max_summary_items`（默认 15）条发 **📰 AI资讯日报**，含领域分布统计。 +5. **每日日报**（新闻机制，默认每天 10:00）：汇总往前 `summary_window_hours`（默认 24h）内普通源综合分 ≥50 的资讯，按分排序取前 `max_summary_items`（默认 15）条发 **📰 AI资讯日报**，含领域分布统计。 +6. **定制监控汇总**（定制机制，默认每天 18:00，独立时间单独配置）：汇总窗口内定制源**命中推送标准**(is_important=1)的资讯发 **🎯 定制监控汇总** 邮件。 +7. **历史采样**：每次采集（无论成功/失败）写入 `source_snapshots`，数据源页可查看，也可用 `/api/sources/history`、`/api/sources/articles` 提取给自动流程。 --- @@ -40,6 +42,9 @@ - 🤖 **大模型接口可配置**：设置页可增删改/测试大模型接口，预置 SiliconFlow（默认）/ DeepSeek 官方 / Autodl / Local Qwen 四个，**一键切换即时生效**，激活接口失败自动切换下一个可用接口 - 🔥 **实时重要资讯邮件**：综合分达到阈值自动推送，单批最多 10 条 - 🎯 **定制监控数据源**：数据源可选「定制监控」方式——**不设权重**，填写「推送标准」后由大模型逐条判断是否达到标准，达到即实时邮件推送（页面可直接「🧪 测试推送标准」）；命中推送的资讯在邮件/列表中带「🎯 定制监控命中」标识 +- 🔀 **两套独立机制**：新闻监控与定制监控**完全分开**——各自的采集间隔、汇总时间/窗口/条数单独配置（设置页「定制监控机制」区块）；定制源命中会单独发「🎯 定制监控汇总」邮件，不混入新闻日报 +- ⏱ **每源独立采集周期**：全局有统一采集间隔（新闻30分/定制15分），每个数据源可用本源「采集周期」字段自定义覆盖（0=跟随全局） +- 📜 **历史采样留档**：每次采集记录时间/条数/成功失败到 `source_snapshots`，数据源页「📜 历史采样」可查看每次快照及本源采集到的资讯；`/api/sources/history`、`/api/sources/articles` 供自动流程提取历史数据 - 📰 **每日 AI 资讯日报**：默认每天 10:00，按领域统计 + 重点资讯卡片 - 🎯 **兴趣画像管理**：关键词（带权重）、领域、关注公司，全部可在网页维护，改完一键「重新打分」 - 🔗 **数据源管理**：7 个内置数据源模板，**完整编辑**（模态框一次性改名称/类型/URL/权重/描述/监控方式/推送标准）、启停、删除；普通源带权重，定制监控源填「推送标准」 @@ -145,10 +150,10 @@ cd works/news-tracker | `/dashboard` | 仪表盘：总资讯/重要资讯/待分析/已通知 + 近7天趋势 + 领域分布 | | `/news` | 资讯列表：按领域/重要/关键词搜索，分页 | | `/news/` | 资讯详情：规则分/LLM分/实体/分析结论 + **页面可读全文** | -| `/sources` | 数据源管理：**完整编辑**（模态框）/启停/删除，支持普通(权重)与定制监控(推送标准)，真实源状态可见 | +| `/sources` | 数据源管理：**完整编辑**（模态框）/启停/删除，支持普通(权重)与定制监控(推送标准)，每源独立采集周期，「📜 历史采样」可查看每次采样快照与本源资讯 | | `/profile` | 兴趣画像：关键词/领域/公司维护 | | `/logs` | 通知日志（实时/日报发送记录） | -| `/settings` | 设置：自动化参数 + 邮件配置 + **大模型接口管理（一键切换/测试）** | +| `/settings` | 设置：自动化参数（新闻机制）+ **定制监控机制（独立配置）** + 邮件配置 + **大模型接口管理（一键切换/测试）** | --- @@ -157,7 +162,9 @@ cd works/news-tracker | 接口 | 方法 | 说明 | |------|------|------| | `/api/stats` | GET | 统计数据（总数/重要/趋势/领域分布） | -| `/api/sources` | POST | `action=add\|update\|delete\|toggle\|test_standard` 数据源管理（update 可改全部配置；test_standard=用大模型试测推送标准） | +| `/api/sources` | POST | `action=add\|update\|delete\|toggle\|test_standard` 数据源管理（update 可改全部配置含采集周期；test_standard=用大模型试测推送标准） | +| `/api/sources/history` | GET | `?source_id=&limit=` 数据源历史采样记录（每次采集时间/条数/状态/备注，自动流程提取用） | +| `/api/sources/articles` | GET | `?source_id=&page=&page_size=&q=` 数据源历史采集到的资讯（分页/搜索，自动流程提取用） | | `/api/profile` | POST | `action=add\|delete`，`kind=keyword\|domain\|company` 画像维护 | | `/api/settings` | POST | 更新 `auto` / `mail` 设置 | | `/api/llm` | POST | 大模型接口：`add\|update\|delete\|switch\|toggle\|test`（switch=一键切换） | @@ -169,7 +176,8 @@ cd works/news-tracker ## 数据模型 -- **sources**：数据源（name/type/url/weight/**kind**/**monitor_standard**/enabled/status/last_fetch/last_count；`kind`=normal 普通按权重打分 / custom 定制监控按推送标准 LLM 判断；`monitor_standard`=定制监控的推送标准说明；status 标记真实源采集 ok/error） +- **sources**：数据源（name/type/url/weight/**kind**/**monitor_standard**/**scan_interval_min**/enabled/status/last_fetch/last_count；`kind`=normal 普通按权重打分 / custom 定制监控按推送标准 LLM 判断；`monitor_standard`=定制监控的推送标准说明；`scan_interval_min`=本源采集间隔分钟数，0=跟随所属机制全局值；status 标记真实源采集 ok/error） +- **source_snapshots**：历史采样记录（source_id/fetched_at/count/status/detail，每次采集留档） - **articles**：资讯（title/url/content/summary/**full_text(页面可读全文)**/domain/entities/importance/relevance/total_score/is_important/analysis/llm_status/notified/status/published_at/collected_at） - **keywords** / **domains** / **companies**：兴趣画像 - **llm_providers**：大模型接口（name/base_url/api_key/model/active/enabled；active=当前激活） @@ -178,6 +186,8 @@ cd works/news-tracker `llm_status`：`pending`(待分析) → `done`(完成) / `skipped`(未达标跳过) / `error`(失败) +**采集调度**：`scheduler.collect_once()` 按**每源独立周期**采集（本源 `scan_interval_min` > 0 用本源值，否则普通源用 `scan_interval_min`、定制源用 `custom_scan_interval_min`）；调度循环 30s 轮询到期源。新闻源受 `auto_collect` 控制、定制源受 `custom_enabled` 控制，互不影响。手动「立即采集」忽略周期全量采集。 + **定制监控（kind=custom）**：无权重（规则打分时权重项按 0 处理，仅作展示）；采集后**无条件进入 LLM 分析**（不受 `llm_threshold` 限制），LLM 按该源的 `monitor_standard`（推送标准）判断 `meets_standard`，达到 → `is_important=1` → 实时邮件推送；未达到 → 不推送。命中推送的资讯也会纳入每日日报。 **综合分公式**：`total_score = 0.4 × 相关度 + 0.6 × 重要度`（LLM 分析后按 `0.6×规则分 + 0.4×LLM分量` 融合） @@ -211,4 +221,4 @@ cd works/news-tracker ## Git - 仓库：`hz4th_coder/news-tracker` -- 版本：`v1.0.0`（初版）→ `v1.0.1`（补 README）→ `v1.1.0`（真实网页采集+全文入库 / 大模型接口多预置一键切换 / 数据源可编辑）→ **`v1.2.0`**（数据源完整编辑模态框 / 定制监控类型：无权重+推送标准+LLM判断推送+测试按钮） +- 版本：`v1.0.0`（初版）→ `v1.0.1`（补 README）→ `v1.1.0`（真实网页采集+全文入库 / 大模型接口多预置一键切换 / 数据源可编辑）→ `v1.2.0`（数据源完整编辑模态框 / 定制监控类型：无权重+推送标准+LLM判断推送+测试按钮）→ **`v1.3.0`**（每源独立采集周期 / 定制监控与新闻监控分离独立配置+独立汇总 / 历史采样留档+查看+提取API） diff --git a/app.py b/app.py index 584726f..8cf0ca7 100644 --- a/app.py +++ b/app.py @@ -117,7 +117,8 @@ def api_sources(): sid = db.add_source(data.get("name", ""), data.get("type", ""), data.get("url", ""), data.get("description", ""), float(data.get("weight", 1.0)), kind=data.get("kind", "normal"), - monitor_standard=data.get("monitor_standard", "")) + monitor_standard=data.get("monitor_standard", ""), + scan_interval_min=int(data.get("scan_interval_min", 0) or 0)) return jsonify({"ok": True, "id": sid}) if action == "update": db.update_source(data["id"], name=data.get("name"), type=data.get("type"), @@ -125,6 +126,7 @@ def api_sources(): weight=float(data.get("weight", 1.0)), kind=data.get("kind", "normal"), monitor_standard=data.get("monitor_standard", ""), + scan_interval_min=int(data.get("scan_interval_min", 0) or 0), enabled=1 if data.get("enabled") else 0) return jsonify({"ok": True}) if action == "delete": @@ -158,6 +160,60 @@ def api_sources(): return jsonify({"ok": False, "error": "unknown action"}) +@app.route("/api/sources/history") +def api_source_history(): + """数据源历史采样记录（供页面查看 / 自动流程提取） + GET /api/sources/history?source_id=1&limit=50 + 返回: {source: {...}, snapshots: [{id,fetched_at,count,status,detail}], total, limit} + """ + source_id = request.args.get("source_id", type=int) + if not source_id: + return jsonify({"ok": False, "error": "source_id 必填"}) + s = db.get_source(source_id) + if not s: + return jsonify({"ok": False, "error": "数据源不存在"}) + limit = min(500, request.args.get("limit", 50, type=int)) + snaps = db.list_source_snapshots(source_id, limit=limit) + return jsonify({"ok": True, "source": s, "snapshots": snaps, "total": len(snaps), "limit": limit}) + + +@app.route("/api/sources/articles") +def api_source_articles(): + """数据源历史采集到的资讯（自动流程提取用） + GET /api/sources/articles?source_id=1&page=1&page_size=20&q=关键词 + 返回: {ok, source, articles: [...], page, page_size, total} + """ + source_id = request.args.get("source_id", type=int) + if not source_id: + return jsonify({"ok": False, "error": "source_id 必填"}) + s = db.get_source(source_id) + if not s: + return jsonify({"ok": False, "error": "数据源不存在"}) + page = max(1, request.args.get("page", 1, type=int)) + page_size = min(100, max(1, request.args.get("page_size", 20, type=int))) + q = request.args.get("q", "") + conn = db.get_conn() + where, args = "a.source_id=?", [source_id] + if q: + where += " AND (a.title LIKE ? OR a.content LIKE ? OR a.summary LIKE ?)" + args += [f"%{q}%", f"%{q}%", f"%{q}%"] + total = conn.execute(f"SELECT COUNT(*) c FROM articles a WHERE {where}", args).fetchone()["c"] + rows = conn.execute( + f"SELECT a.* FROM articles a WHERE {where} ORDER BY a.collected_at DESC, a.id DESC " + f"LIMIT ? OFFSET ?", args + [page_size, (page - 1) * page_size]).fetchall() + conn.close() + out = [] + for r in rows: + d = dict(r) + try: + d["entities"] = json.loads(d["entities"] or "[]") + except Exception: + d["entities"] = [] + out.append(d) + return jsonify({"ok": True, "source": s, "articles": out, "page": page, + "page_size": page_size, "total": total}) + + @app.route("/api/profile", methods=["POST"]) def api_profile(): data = request.get_json(force=True) or {} @@ -197,6 +253,10 @@ def api_settings(): cur = db.get_all_settings().get("mail", {}) cur.update(data["mail"]) db.set_setting("mail", cur) + if "custom" in data and isinstance(data["custom"], dict): + cur = db.get_all_settings().get("custom", {}) + cur.update(data["custom"]) + db.set_setting("custom", cur) return jsonify({"ok": True}) @@ -268,7 +328,7 @@ def api_actions(): data = request.get_json(force=True) or {} action = data.get("action") if action == "collect": - n = scheduler.collect_once() + n = scheduler.collect_once(force=True) return jsonify({"ok": True, "added": n}) if action == "llm": r = analysis.batch_llm_analyze(limit=int(data.get("limit", 10))) @@ -276,6 +336,9 @@ def api_actions(): if action == "summary": n = scheduler.send_daily_summary() return jsonify({"ok": True, "sent": n}) + if action == "custom_summary": + n = scheduler.send_custom_summary() + return jsonify({"ok": True, "sent": n}) if action == "seed": n = simulate.seed_all() return jsonify({"ok": True, "added": n}) @@ -306,12 +369,16 @@ def not_found(e): # ---------------- 启动 ---------------- def main(): db.init_db() + # 确保定制监控机制配置存在（老库升级） + if db.get_setting("custom") is None: + db.set_setting("custom", dict(config.CUSTOM_DEFAULTS)) # 首次初始化：写入默认数据源 / 兴趣画像 / 默认设置 / 模拟数据 if db.get_setting("initialized") != 1: for s in config.DEFAULT_SOURCES: db.add_source(s["name"], s["type"], s["url"], s["description"], s["weight"], kind=s.get("kind", "normal"), - monitor_standard=s.get("monitor_standard", "")) + monitor_standard=s.get("monitor_standard", ""), + scan_interval_min=s.get("scan_interval_min", 0)) for kw, w in config.DEFAULT_KEYWORDS: db.add_keyword(kw, w) for d, w in config.DEFAULT_DOMAINS: @@ -327,6 +394,7 @@ def main(): db.set_active_provider(pid) db.set_setting("initialized", 1) db.set_setting("auto", dict(config.AUTO_DEFAULTS)) + db.set_setting("custom", dict(config.CUSTOM_DEFAULTS)) db.set_setting("mail", dict(config.MAIL_DEFAULTS)) simulate.seed_all() analysis.run_llm_background() diff --git a/config.py b/config.py index c28857e..8b73bc3 100644 --- a/config.py +++ b/config.py @@ -67,8 +67,8 @@ MAIL_DEFAULTS = { # ---------------- 自动化（默认值，可在设置区修改） ---------------- AUTO_DEFAULTS = { - "auto_collect": 1, # 是否自动定时采集 - "scan_interval_min": 30, # 采集扫描间隔（分钟） + "auto_collect": 1, # 是否自动定时采集（新闻监控机制） + "scan_interval_min": 30, # 采集扫描间隔（分钟，新闻源统一值，每源可覆盖） "realtime_threshold": 80, # 总分 >= 该值 → 实时邮件通知 "llm_threshold": 60, # 规则分 >= 该值 → 进入 LLM 深度分析 "realtime_enabled": 1, # 是否启用实时重要资讯邮件 @@ -78,6 +78,16 @@ AUTO_DEFAULTS = { "max_summary_items": 15, # 汇总邮件最多条目数 } +# ---------------- 定制监控机制（与新闻监控分开单独配置） ---------------- +CUSTOM_DEFAULTS = { + "custom_enabled": 1, # 是否启用定制监控机制（采集/判断/推送/汇总） + "custom_scan_interval_min": 15, # 定制源统一采集间隔（分钟，每源可覆盖；比新闻更频繁） + "custom_summary_enabled": 1, # 是否单独发「定制监控汇总」邮件 + "custom_summary_time": "18:00", # 定制监控汇总时间（独立于新闻日报） + "custom_summary_window_hours": 24, + "custom_max_summary_items": 20, # 定制汇总邮件最多条目数 +} + # ---------------- 默认兴趣画像（可在网页修改） ---------------- DEFAULT_KEYWORDS = [ ("大模型", 8), ("人工智能", 6), ("AI", 6), ("芯片", 7), ("GPU", 6), diff --git a/crawler.py b/crawler.py index c696eb9..328c31a 100644 --- a/crawler.py +++ b/crawler.py @@ -232,9 +232,12 @@ def fetch_source(source): example.com 占位源返回模拟数据；真实源抓取失败抛异常（由 fetch_all 捕获并标记 error，不塞模拟数据）。 """ url = (source.get("url") or "").strip() - # 模拟源（example.com 占位）→ 用仿真数据填充 + # 模拟源（example.com 占位）→ 用仿真数据填充（补齐 source_id，保证定制监控识别正确） if not url or "example.com" in url: - return simulate.items_for_source(source["id"], source.get("type", "")) + items = simulate.items_for_source(source["id"], source.get("type", "")) + for it in items: + it["source_id"] = source["id"] + return items per = config.CRAWL_DEFAULTS["per_source_links"] full = config.CRAWL_DEFAULTS["full_fetch_links"] html = fetch_page(url) diff --git a/db.py b/db.py index 3221644..62abc98 100644 --- a/db.py +++ b/db.py @@ -20,6 +20,7 @@ CREATE TABLE IF NOT EXISTS sources ( weight REAL DEFAULT 1.0, kind TEXT DEFAULT 'normal', -- normal 普通(按权重打分) / custom 定制监控(按推送标准LLM判断) monitor_standard TEXT DEFAULT '', -- 定制监控：推送标准说明 + scan_interval_min INTEGER DEFAULT 0, -- 本源采集间隔(分钟)，0=跟随所属机制的全局值 enabled INTEGER DEFAULT 1, status TEXT DEFAULT 'ok', -- ok / error last_fetch TEXT DEFAULT '', @@ -27,6 +28,17 @@ CREATE TABLE IF NOT EXISTS sources ( created_at TEXT DEFAULT '' ); +CREATE TABLE IF NOT EXISTS source_snapshots ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source_id INTEGER NOT NULL, + fetched_at TEXT DEFAULT '', -- 本次采样时间 + count INTEGER DEFAULT 0, -- 本次采到条数 + status TEXT DEFAULT 'ok', -- ok / error + detail TEXT DEFAULT '', -- 错误信息 / 备注 + created_at TEXT DEFAULT '' +); +CREATE INDEX IF NOT EXISTS idx_snapshots_source ON source_snapshots(source_id, id); + CREATE TABLE IF NOT EXISTS articles ( id INTEGER PRIMARY KEY AUTOINCREMENT, source_id INTEGER DEFAULT 0, @@ -113,12 +125,25 @@ def init_db(): os.makedirs(config.DATA_DIR, exist_ok=True) conn = get_conn() conn.executescript(_SCHEMA) - # 迁移：旧库补充 sources.kind / monitor_standard（定制监控类型） + # 迁移：旧库补充 sources.kind / monitor_standard / scan_interval_min cols = [r["name"] for r in conn.execute("PRAGMA table_info(sources)").fetchall()] if "kind" not in cols: conn.execute("ALTER TABLE sources ADD COLUMN kind TEXT DEFAULT 'normal'") if "monitor_standard" not in cols: conn.execute("ALTER TABLE sources ADD COLUMN monitor_standard TEXT DEFAULT ''") + if "scan_interval_min" not in cols: + conn.execute("ALTER TABLE sources ADD COLUMN scan_interval_min INTEGER DEFAULT 0") + # 迁移：历史采样表（每次采集留档） + conn.execute("""CREATE TABLE IF NOT EXISTS source_snapshots ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source_id INTEGER NOT NULL, + fetched_at TEXT DEFAULT '', + count INTEGER DEFAULT 0, + status TEXT DEFAULT 'ok', + detail TEXT DEFAULT '', + created_at TEXT DEFAULT '' + )""") + conn.execute("CREATE INDEX IF NOT EXISTS idx_snapshots_source ON source_snapshots(source_id, id)") # 迁移：旧库补充 full_text 列（存页面可读正文） cols = [r["name"] for r in conn.execute("PRAGMA table_info(articles)").fetchall()] if "full_text" not in cols: @@ -187,12 +212,12 @@ def get_source(sid): return dict(row) if row else None -def add_source(name, type_, url, desc, weight=1.0, kind="normal", monitor_standard=""): +def add_source(name, type_, url, desc, weight=1.0, kind="normal", monitor_standard="", scan_interval_min=0): conn = get_conn() cur = conn.execute( - "INSERT INTO sources(name,type,url,description,weight,kind,monitor_standard,created_at) " - "VALUES(?,?,?,?,?,?,?,?)", - (name, type_, url, desc, weight, kind, monitor_standard, now_str()), + "INSERT INTO sources(name,type,url,description,weight,kind,monitor_standard,scan_interval_min,created_at) " + "VALUES(?,?,?,?,?,?,?,?,?)", + (name, type_, url, desc, weight, kind, monitor_standard, scan_interval_min, now_str()), ) conn.commit() conn.close() @@ -200,7 +225,8 @@ def add_source(name, type_, url, desc, weight=1.0, kind="normal", monitor_standa def update_source(sid, **fields): - allowed = {"name", "type", "url", "description", "weight", "kind", "monitor_standard", "enabled"} + allowed = {"name", "type", "url", "description", "weight", "kind", "monitor_standard", + "scan_interval_min", "enabled"} sets, vals = [], [] for k, v in fields.items(): if k in allowed: @@ -232,6 +258,28 @@ def update_source_fetch(sid, status="ok", count=0): conn.close() +# ---------------- 历史采样（每次采集留档，可查看/自动流程提取） ---------------- +def add_source_snapshot(source_id, count=0, status="ok", detail=""): + conn = get_conn() + conn.execute( + "INSERT INTO source_snapshots(source_id,fetched_at,count,status,detail,created_at) " + "VALUES(?,?,?,?,?,?)", + (source_id, now_str(), count, status, detail, now_str()), + ) + conn.commit() + conn.close() + + +def list_source_snapshots(source_id, limit=50): + conn = get_conn() + rows = conn.execute( + "SELECT * FROM source_snapshots WHERE source_id=? ORDER BY id DESC LIMIT ?", + (source_id, limit), + ).fetchall() + conn.close() + return [dict(r) for r in rows] + + # ---------------- articles ---------------- def article_exists(url): conn = get_conn() @@ -370,15 +418,36 @@ def article_stats(): def latest_articles_for_summary(window_hours=24): - """汇总窗口内（含未通知、未汇总的）资讯 - 普通源按 total_score>=50；定制监控命中推送标准(is_important=1)的也纳入，不看分数。""" + """新闻机制汇总：普通源(normal)，窗口内 total_score>=50""" conn = get_conn() rows = conn.execute( - "SELECT a.*, s.name AS source_name FROM articles a " + "SELECT a.*, s.name AS source_name, s.kind AS source_kind FROM articles a " "LEFT JOIN sources s ON a.source_id=s.id " "WHERE a.published_at >= datetime('now','localtime','-%d hours') " - "AND (a.total_score>=50 OR (s.kind='custom' AND a.is_important=1)) " - "ORDER BY a.total_score DESC LIMIT 200" % int(window_hours) + "AND (s.kind IS NULL OR s.kind != 'custom') " + "AND a.total_score>=50 ORDER BY a.total_score DESC LIMIT 200" % int(window_hours) + ).fetchall() + conn.close() + out = [] + for r in rows: + d = dict(r) + try: + d["entities"] = json.loads(d["entities"] or "[]") + except Exception: + d["entities"] = [] + out.append(d) + return out + + +def custom_articles_for_summary(window_hours=24): + """定制监控汇总：定制源(custom)，窗口内命中推送标准(is_important=1)的资讯""" + conn = get_conn() + rows = conn.execute( + "SELECT a.*, s.name AS source_name, s.kind AS source_kind FROM articles a " + "LEFT JOIN sources s ON a.source_id=s.id " + "WHERE a.published_at >= datetime('now','localtime','-%d hours') " + "AND s.kind='custom' AND a.is_important=1 " + "ORDER BY a.published_at DESC, a.id DESC LIMIT 200" % int(window_hours) ).fetchall() conn.close() out = [] diff --git a/notifier.py b/notifier.py index 61dca32..baa1460 100644 --- a/notifier.py +++ b/notifier.py @@ -106,7 +106,7 @@ def send_realtime(articles): def send_daily_summary(articles, window_label): - """每日汇总（默认每天10点）：昨天至今的重要/相关资讯""" + """新闻机制：每日汇总（默认每天10点）——普通源的重要/相关资讯""" if not articles: return 0 top = articles[: int(db.get_setting("max_summary_items", config.AUTO_DEFAULTS["max_summary_items"]))] @@ -133,3 +133,25 @@ def send_daily_summary(articles, window_label): if art and art.get("status") != "summarized": db.update_article(aid, status="summarized") return len(top) + + +def send_custom_summary(articles, window_label): + """定制监控机制：独立汇总邮件——定制源在窗口内命中推送标准的资讯（单独配置/单独时间）""" + if not articles: + return 0 + top = articles[: int(db.get_setting("custom_max_summary_items", + config.CUSTOM_DEFAULTS["custom_max_summary_items"]))] + cards = "".join(_card(a) for a in top) + subject = f"🎯 定制监控汇总 {window_label} · 命中{len(articles)}条" + html = _html_wrap( + "定制监控命中汇总", + f""" +

汇总时段：{window_label}

+

以下为定制监控源中达到推送标准的资讯（由大模型按各源推送标准判定）：

+ {cards} + """, + ) + send_email(subject, html) + ids = [a["id"] for a in articles if a["id"]] + db.add_log("custom_summary", subject, len(articles), ids, status="ok", detail=f"汇总{len(top)}条") + return len(top) diff --git a/scheduler.py b/scheduler.py index 69b2e5a..d79929b 100644 --- a/scheduler.py +++ b/scheduler.py @@ -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 diff --git a/simulate.py b/simulate.py index a3c47b1..1d3cb1c 100644 --- a/simulate.py +++ b/simulate.py @@ -202,12 +202,14 @@ def _make_item(row): "summary": summary, "domain": domain, "entities": entities, + "source_id": int(src_idx), "published_at": published.strftime("%Y-%m-%d %H:%M:%S"), } def items_for_source(src_id, src_type=""): - """返回指定数据源的模拟条目（真实抓取失败/模拟源时回退用）""" + """返回指定数据源的模拟条目（真实抓取失败/模拟源时回退用） + 无内建匹配条目的源（如新建的定制监控源）→ 合成几条通用资讯，保证链路不空转。""" items = [] for row in _NEWS: if row[3] == int(src_id): @@ -215,6 +217,31 @@ def items_for_source(src_id, src_type=""): if src_type: it["domain"] = src_type items.append(it) + if not items: + s = db.get_source(int(src_id)) if db else None + name = (s.get("name") if s else "") or f"数据源{src_id}" + generic = [ + ("新品发布与里程碑", "该公司/该源今日宣布一项重要产品发布与里程碑进展，引发行业广泛关注，多位分析师上调预期。", + "重大产品发布，行业里程碑。"), + ("重大战略合作", "该公司今日宣布与多家头部企业达成重大战略合作，涉及核心业务扩张与生态布局，市场反应积极。", + "重大战略合作公告。"), + ("关键进展与新动态", "该领域出现关键性进展与新动态，相关技术/产品进入新阶段，预计将显著影响后续走向。", + "关键进展，影响显著。"), + ("股价大幅波动", "受重大消息影响，该相关标的股价今日大幅波动，单日涨跌幅显著，市场情绪剧烈变化。", + "股价大幅波动。"), + ("日常行业动态", "该领域今日整体平稳，行业例行消息与常规动态为主，无实质性重大变化。", "日常动态。"), + ] + import hashlib + from datetime import timedelta + for i, (t, c, sm) in enumerate(generic): + published = datetime.now() - timedelta(hours=2 + i) + url = "https://news.example.com/s/" + hashlib.md5(f"{src_id}-{t}".encode()).hexdigest()[:16] + items.append({ + "title": t, "url": url, "author": name, + "content": c, "summary": sm, "domain": src_type or "综合", + "entities": [], "source_id": int(src_id), + "published_at": published.strftime("%Y-%m-%d %H:%M:%S"), + }) return items diff --git a/static/style.css b/static/style.css index 20bfc40..be1d98b 100644 --- a/static/style.css +++ b/static/style.css @@ -123,6 +123,9 @@ textarea { width:100%; } .modal-body textarea { resize:vertical; } .modal-foot { display:flex; justify-content:flex-end; gap:8px; padding:14px 20px; border-top:1px solid #f3f4f6; } .modal-mask[hidden] { display:none !important; } +.hist-list { max-height:300px; overflow:auto; } +.hist-item { padding:7px 0; border-bottom:1px solid #f9fafb; } +.hist-item:last-child { border-bottom:none; } /* 页面可读全文（干净排版） */ .full-text { margin-top:8px; padding:14px 16px; background:#f9fafb; border:1px solid #e5e7eb; border-radius:8px; font-size:13px; line-height:1.9; color:#374151; white-space:pre-wrap; word-break:break-word; } diff --git a/templates/base.html b/templates/base.html index 1a647b7..bbeb40c 100644 --- a/templates/base.html +++ b/templates/base.html @@ -38,6 +38,15 @@ const API = { body: JSON.stringify(body) }); return r.json(); + }, + get: async (url, params) => { + if (params){ + const qs = Object.entries(params).filter(([k,v])=>v!==''&&v!=null) + .map(([k,v])=>encodeURIComponent(k)+'='+encodeURIComponent(v)).join('&'); + if (qs) url += (url.includes('?')?'&':'?') + qs; + } + const r = await fetch(url); + return r.json(); } }; function toast(msg, ok=true) { diff --git a/templates/settings.html b/templates/settings.html index 21d84a6..148f8c3 100644 --- a/templates/settings.html +++ b/templates/settings.html @@ -67,6 +67,36 @@ +
+

🎯 定制监控机制 + （与新闻监控完全独立：定制源按此间隔采集，命中推送标准单独汇总；每源可用本源「采集周期」覆盖间隔） +

+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+

🤖 大模型接口 （切换后即时生效，智能分析将使用当前接口；失败自动切换下一个可用接口） @@ -132,7 +162,15 @@ async function saveAll(){ smtp_user: G('m_user'), smtp_pass: G('m_pass'), smtp_mode: G('m_mode'), email_to: G('m_to'), sender_name: G('m_name'), }; - const r = await API.json('/api/settings', { ...auto, mail }); + const custom = { + custom_enabled: document.getElementById('c_custom_enabled').checked ? 1 : 0, + custom_scan_interval_min: parseInt(G('c_custom_scan_interval_min'))||15, + custom_summary_enabled: document.getElementById('c_custom_summary_enabled').checked ? 1 : 0, + custom_summary_time: G('c_custom_summary_time')||'18:00', + custom_summary_window_hours: parseInt(G('c_custom_summary_window_hours'))||24, + custom_max_summary_items: parseInt(G('c_custom_max_summary_items'))||20, + }; + const r = await API.json('/api/settings', { ...auto, mail, custom }); toast(r.ok ? '✅ 设置已保存' : '❌ 保存失败', r.ok); } diff --git a/templates/sources.html b/templates/sources.html index 236fb4f..74264b5 100644 --- a/templates/sources.html +++ b/templates/sources.html @@ -9,20 +9,22 @@
- + {% for s in sources %} - + + @@ -32,7 +34,8 @@
ID名称监控方式类型权重状态最近采集新增条数操作
ID名称监控方式类型权重采集周期状态最近采样条数操作
{{ s.id }} {{ s.name }}
{{ s.description }}
{{ s.url }}		 {% if s.kind == 'custom' %}🎯 定制监控{% else %}普通{% endif %} {{ s.type }} {% if s.kind == 'custom' %}—{% else %}{{ s.weight }}{% endif %}{% if s.scan_interval_min %}{{ s.scan_interval_min }}分{% else %}默认{% endif %} {% if s.enabled %}启用{% else %}停用{% endif %} {{ s.last_fetch or '—' }} {{ s.last_count }} +
💡 真实 URL 源自动抓取页面并清洗可读正文入库；example.com 等占位地址走模拟数据。
- 🎯 定制监控：不设权重，填写「推送标准」后由大模型逐条判断是否达到标准，达到即实时邮件推送。 + ⏱ 采集周期：0=跟随所属机制全局值（普通源=设置页采集间隔，定制源=定制监控间隔），填分钟数则本源自定义。
+ 🎯 定制监控：与新闻监控独立，无权重，按「推送标准」由大模型判断是否推送，命中实时推送 + 单独汇总。
@@ -49,10 +52,13 @@ + + +
@@ -82,10 +88,32 @@
+ + + {% endblock %} {% block script %} {% endblock %}