# -*- coding: utf-8 -*- """ 新闻智能跟踪系统 - Flask 主应用 网页:仪表盘 / 资讯列表 / 资讯详情 / 数据源 / 兴趣画像 / 通知日志 / 设置 API:采集 / LLM分析 / 汇总 / 画像维护 / 设置维护 """ from datetime import datetime, timedelta import json from flask import Flask, render_template, request, jsonify, redirect, url_for import requests import config import db import simulate import analysis import notifier import scheduler app = Flask(__name__, static_folder=config.STATIC_DIR, template_folder=config.TEMPLATE_DIR) # ---------------- 页面 ---------------- @app.route("/") def index(): return redirect(url_for("dashboard")) @app.route("/dashboard") def dashboard(): stats = db.article_stats() latest = db.list_articles(limit=10, order="a.collected_at DESC") important = db.list_articles(is_important=1, limit=10, order="a.total_score DESC") dom_rows = db.get_conn().execute( "SELECT domain, COUNT(*) c FROM articles GROUP BY domain ORDER BY c DESC").fetchall() domain_stats = [{"name": r["domain"] or "未分类", "count": r["c"]} for r in dom_rows] return render_template("dashboard.html", stats=stats, latest=latest, important=important, domain_stats=domain_stats) @app.route("/news") def news(): domain = request.args.get("domain", "") important = request.args.get("important", "") q = request.args.get("q", "") page = max(1, int(request.args.get("page", 1))) per = 20 filters = {} if domain: filters["domain"] = domain if important == "1": filters["is_important"] = 1 if q: filters["q"] = q total = db.count_articles(domain=filters.get("domain"), is_important=filters.get("is_important")) if q: total = len(db.list_articles(q=q)) pages = max(1, (total + per - 1) // per) articles = db.list_articles(offset=(page - 1) * per, limit=per, **filters) domains = db.get_conn().execute( "SELECT DISTINCT domain FROM articles WHERE domain<>'' ORDER BY domain").fetchall() return render_template("news.html", articles=articles, domains=[d["domain"] for d in domains], domain=domain, important=important, q=q, page=page, pages=pages, total=total) @app.route("/news/") def news_detail(aid): art = db.get_article(aid) if not art: return "not found", 404 return render_template("detail.html", art=art) @app.route("/sources") def sources(): return render_template("sources.html", sources=db.list_sources()) @app.route("/profile") def profile(): return render_template("profile.html", keywords=db.list_keywords(), domains=db.list_domains(), companies=db.list_companies()) @app.route("/logs") def logs(): f_type = request.args.get("type", "") f_status = request.args.get("status", "") q = request.args.get("q", "") page = max(1, int(request.args.get("page", 1))) per = 20 total, logs = db.list_logs(page=page, page_size=per, type_=f_type or None, status=f_status or None, q=q or None) pages = max(1, (total + per - 1) // per) return render_template("logs.html", logs=logs, f_type=f_type, f_status=f_status, q=q, page=page, pages=pages, total=total) @app.route("/api/logs") def api_logs(): """通知日志分页查询(JSON) GET /api/logs?page=&page_size=&type=&status=&q= """ f_type = request.args.get("type", "") f_status = request.args.get("status", "") q = request.args.get("q", "") page = max(1, request.args.get("page", 1, type=int)) page_size = min(100, max(1, request.args.get("page_size", 20, type=int))) total, rows = db.list_logs(page=page, page_size=page_size, type_=f_type or None, status=f_status or None, q=q or None) return jsonify({"ok": True, "logs": rows, "page": page, "page_size": page_size, "total": total, "pages": max(1, (total + page_size - 1) // page_size)}) @app.route("/api/errors") def api_errors(): """系统错误日志(最近 N 条)""" limit = min(200, request.args.get("limit", 100, type=int)) return jsonify({"ok": True, "errors": db.list_system_errors(limit=limit)}) @app.route("/settings") def settings_page(): auto = db.get_all_settings() # 归一化(兼容老库嵌套数据),并补齐默认值,保证模板字段齐全 a0 = auto.get("auto") if not isinstance(a0, dict) or "auto_collect" not in a0: a0 = {} for k in config.AUTO_DEFAULTS: a0[k] = (auto.get("auto") or {}).get(k, config.AUTO_DEFAULTS[k]) auto["auto"] = a0 for k, dft in (("mail", config.MAIL_DEFAULTS), ("custom", config.CUSTOM_DEFAULTS), ("webcapture", config.WEBCAPTURE_DEFAULTS), ("errnotify", config.ERRNOTIFY_DEFAULTS)): v = auto.get(k) if not isinstance(v, dict): auto[k] = dict(dft) else: merged = dict(dft) merged.update(v) auto[k] = merged return render_template("settings.html", auto=auto, providers=db.list_providers()) # ---------------- API ---------------- @app.route("/api/stats") def api_stats(): stats = db.article_stats() dom_rows = db.get_conn().execute( "SELECT domain, COUNT(*) c FROM articles GROUP BY domain ORDER BY c DESC").fetchall() stats["domains"] = [{"name": r["domain"] or "未分类", "count": r["c"]} for r in dom_rows] trend_rows = db.get_conn().execute( "SELECT substr(collected_at,1,10) d, COUNT(*) c FROM articles " "GROUP BY d ORDER BY d DESC LIMIT 7").fetchall() stats["trend"] = [{"date": r["d"], "count": r["c"]} for r in reversed(trend_rows)] return jsonify(stats) @app.route("/api/sources", methods=["POST"]) def api_sources(): data = request.get_json(force=True) or {} action = data.get("action") if action == "add": 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", ""), scan_interval_min=int(data.get("scan_interval_min", 0) or 0), fetch_method=data.get("fetch_method", "auto") or "auto", capture_params=data.get("capture_params", "{}") or "{}") return jsonify({"ok": True, "id": sid}) if action == "update": db.update_source(data["id"], name=data.get("name"), type=data.get("type"), url=data.get("url"), description=data.get("description"), 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), fetch_method=data.get("fetch_method", "auto") or "auto", capture_params=data.get("capture_params", "{}") or "{}", enabled=1 if data.get("enabled") else 0) return jsonify({"ok": True}) if action == "delete": db.delete_source(data["id"]) return jsonify({"ok": True}) if action == "toggle": s = db.get_source(data["id"]) db.update_source(data["id"], enabled=0 if s["enabled"] else 1) return jsonify({"ok": True}) if action == "enable_all": n = db.set_all_sources_enabled(1) return jsonify({"ok": True, "count": n}) if action == "disable_all": n = db.set_all_sources_enabled(0) return jsonify({"ok": True, "count": n}) if action == "test_standard": # 定制监控:用大模型测试「推送标准」是否好使(给一段示例内容看判不判得出) standard = (data.get("monitor_standard") or "").strip() sample = (data.get("sample") or "").strip() if not standard: return jsonify({"ok": False, "error": "请先填写推送标准"}) prompt = ( "你是一位资讯监控专员。用户配置了一个定制监控数据源,并设定了「推送标准」。\n" f"【推送标准】\n{standard}\n\n" f"【待判断内容】\n{sample or '(未提供示例内容,请自行用一句典型的需推送场景作答,说明是否达到标准)'}\n\n" "请判断该内容是否达到推送标准。只输出一个 JSON 对象(不要任何其他文字):\n" '{"meets_standard": true或false, "reason": "判断理由(40字内中文)"}' ) try: content, _name = analysis._llm_chat(prompt) parsed = json.loads(content) return jsonify({"ok": True, "meets": bool(parsed.get("meets_standard")), "reason": parsed.get("reason", "")}) except Exception as e: return jsonify({"ok": False, "error": str(e)}) 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 {} action = data.get("action") kind = data.get("kind") if action == "add": if kind == "keyword": db.add_keyword(data.get("name", ""), int(data.get("weight", 5))) elif kind == "domain": db.add_domain(data.get("name", ""), int(data.get("weight", 5))) elif kind == "company": db.add_company(data.get("name", "")) return jsonify({"ok": True}) if action == "delete": if kind == "keyword": db.delete_keyword(data["id"]) elif kind == "domain": db.delete_domain(data["id"]) elif kind == "company": db.delete_company(data["id"]) return jsonify({"ok": True}) return jsonify({"ok": False, "error": "unknown action"}) @app.route("/api/settings", methods=["POST"]) def api_settings(): data = request.get_json(force=True) or {} auto = {} for k in config.AUTO_DEFAULTS: if k in data: auto[k] = data[k] if auto: cur = db.get_all_settings() cur.update(auto) db.set_setting("auto", auto) if "mail" in data and isinstance(data["mail"], dict): 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) if "webcapture" in data and isinstance(data["webcapture"], dict): cur = db.get_all_settings().get("webcapture", {}) cur.update(data["webcapture"]) db.set_setting("webcapture", cur) if "errnotify" in data and isinstance(data["errnotify"], dict): cur = db.get_all_settings().get("errnotify", {}) cur.update(data["errnotify"]) db.set_setting("errnotify", cur) return jsonify({"ok": True}) @app.route("/api/llm", methods=["POST"]) def api_llm(): """大模型接口管理:增删改 / 一键切换 / 测试""" data = request.get_json(force=True) or {} action = data.get("action") if action == "add": pid = db.add_provider( data.get("name", ""), data.get("base_url", ""), data.get("api_key", ""), data.get("model", ""), active=1 if data.get("active") else 0, enabled=1 if data.get("enabled", 1) else 0, ) if data.get("active"): db.set_active_provider(pid) return jsonify({"ok": True, "id": pid}) if action == "update": db.update_provider(data["id"], name=data.get("name"), base_url=data.get("base_url"), api_key=data.get("api_key"), model=data.get("model"), enabled=1 if data.get("enabled", 1) else 0) return jsonify({"ok": True}) if action == "delete": db.delete_provider(data["id"]) return jsonify({"ok": True}) if action == "switch": db.set_active_provider(data["id"]) p = db.get_provider(data["id"]) return jsonify({"ok": True, "name": p["name"] if p else ""}) if action == "toggle": p = db.get_provider(data["id"]) if not p: return jsonify({"ok": False, "error": "not found"}) db.update_provider(data["id"], enabled=0 if p["enabled"] else 1) return jsonify({"ok": True}) if action == "test": # 用指定接口(或当前激活接口)发一条测试消息 cfg = None if data.get("id"): p = db.get_provider(data["id"]) if p and p.get("base_url"): cfg = {"name": p["name"], "base_url": p["base_url"].rstrip("/"), "api_key": p.get("api_key", ""), "model": p.get("model", "")} try: if cfg is None: import analysis cfg = analysis.get_llm_cfg() r = requests.post( f"{cfg['base_url']}/chat/completions", headers={"Authorization": f"Bearer {cfg['api_key']}", "Content-Type": "application/json"}, json={"model": cfg["model"], "messages": [{"role": "user", "content": "请回复:连接正常"}], "max_tokens": 60, "temperature": 0.3}, timeout=60, ) r.raise_for_status() content = r.json()["choices"][0]["message"]["content"] return jsonify({"ok": True, "name": cfg["name"], "model": cfg["model"], "reply": content}) except Exception as e: return jsonify({"ok": False, "error": str(e)}) return jsonify({"ok": False, "error": "unknown action"}) @app.route("/api/actions", methods=["POST"]) def api_actions(): data = request.get_json(force=True) or {} action = data.get("action") if action == "collect": 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))) return jsonify({"ok": True, **r}) 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}) if action == "reanalyze": # 重新跑规则打分(如改了兴趣画像后) conn = db.get_conn() ids = [r["id"] for r in conn.execute("SELECT id FROM articles").fetchall()] conn.close() cnt = 0 for aid in ids: analysis.analyze_article(aid) cnt += 1 return jsonify({"ok": True, "count": cnt}) if action == "test_mail": try: notifier.send_email("📮 新闻智能跟踪系统测试", "

测试成功

邮件通知链路正常。

") return jsonify({"ok": True, "msg": "测试邮件已发送"}) except Exception as e: return jsonify({"ok": False, "error": str(e)}) if action == "test_error_mail": try: notifier.test_error_notify() return jsonify({"ok": True, "msg": "测试错误通知邮件已发送"}) except Exception as e: return jsonify({"ok": False, "error": str(e)}) if action == "test_webcapture": try: cfg = dict(config.WEBCAPTURE_DEFAULTS) cfg.update(db.get_all_settings().get("webcapture", {})) base = (cfg.get("api_url") or "").rstrip("/") r = requests.post(f"{base}/api/capture", json={"url": "https://example.com", "action": "text", "wait_time": 800}, timeout=30) r.raise_for_status() d = r.json() if d.get("success"): return jsonify({"ok": True, "title": d.get("title", ""), "text": (d.get("text") or "")[:80], "api_url": base}) return jsonify({"ok": False, "error": d.get("error", "接口返回失败")}) except Exception as e: return jsonify({"ok": False, "error": str(e)}) return jsonify({"ok": False, "error": "unknown action"}) @app.errorhandler(404) def not_found(e): return render_template("404.html"), 404 @app.errorhandler(Exception) def handle_exception(e): """未捕获异常 → 记录系统错误并邮件通知(频率/静默由设置控制),返回 500 页""" from werkzeug.exceptions import HTTPException if isinstance(e, HTTPException): return e try: notifier.report_error("系统", "未捕获异常", f"{type(e).__name__}: {e}") except Exception: pass return render_template("500.html"), 500 # ---------------- 启动 ---------------- def main(): db.init_db() # 确保定制监控机制配置存在(老库升级) if db.get_setting("custom") is None: db.set_setting("custom", dict(config.CUSTOM_DEFAULTS)) # 确保网页提取服务 / 错误通知配置存在(老库升级) if db.get_setting("webcapture") is None: db.set_setting("webcapture", dict(config.WEBCAPTURE_DEFAULTS)) if db.get_setting("errnotify") is None: db.set_setting("errnotify", dict(config.ERRNOTIFY_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", ""), 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: db.add_domain(d, w) for c in config.DEFAULT_COMPANIES: db.add_company(c) # 预置大模型接口(可一键切换) if not db.list_providers(): for p in config.LLM_PROVIDERS_DEFAULT: pid = db.add_provider(p["name"], p["base_url"], p["api_key"], p["model"], active=1 if p.get("active") else 0) if p.get("active"): 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() stop_event, t = scheduler.start_scheduler() print(f"✅ {config.SERVICE_NAME} 启动完成") print(f" Web: http://0.0.0.0:{config.SERVICE_PORT}/") print(f" 每日汇总: {db.get_setting('summary_time', config.AUTO_DEFAULTS['summary_time'])}" f" | 采集间隔: {db.get_setting('scan_interval_min', 30)}分钟" f" | 实时阈值: {db.get_setting('realtime_threshold', 80)}") app.run(host=config.SERVICE_HOST, port=config.SERVICE_PORT, threaded=True) if __name__ == "__main__": main()