news-tracker/app.py

# -*- 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/<int:aid>")
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/logs/delete", methods=["POST"])
def api_logs_delete():
    """批量删除通知日志。POST {ids: [1,2,3]} 或 {all: true}（删除筛选条件下的全部）"""
    data = request.get_json(force=True) or {}
    ids = [int(x) for x in (data.get("ids") or []) if str(x).isdigit()]
    if data.get("all"):
        # 删除当前筛选条件下的全部
        f_type = data.get("type", "") or None
        f_status = data.get("status", "") or None
        q = data.get("q", "") or None
        ids = db.list_log_ids(type_=f_type, status=f_status, q=q)
    n = db.delete_logs(ids)
    return jsonify({"ok": True, "deleted": n})


@app.route("/api/logs/<int:lid>")
def api_log_one(lid):
    """取单条通知日志（含 body 正文）"""
    row = db.get_log(lid)
    if not row:
        return jsonify({"ok": False, "error": "not found"})
    d = dict(row)
    try:
        d["article_ids"] = json.loads(d["article_ids"] or "[]")
    except Exception:
        d["article_ids"] = []
    return jsonify({"ok": True, "log": d})


@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("/api/articles/batch")
def api_articles_batch():
    """按 ID 批量取资讯（id+title+url），用于旧日志无 body 时展示链接。"""
    ids = [int(x) for x in request.args.get("ids", "").split(",") if x.strip().isdigit()]
    if not ids:
        return jsonify({"ok": True, "articles": []})
    conn = db.get_conn()
    marks = ",".join("?" * len(ids))
    rows = conn.execute(f"SELECT id, title, url FROM articles WHERE id IN ({marks})", ids).fetchall()
    conn.close()
    return jsonify({"ok": True, "articles": [dict(r) for r in rows]})


@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(),
                           llm_stats=db.llm_stats_summary(), llm_daily=db.llm_stats_daily(14))


# ---------------- 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 == "move":
        # 优先级上移/下移：direction=-1 上移（优先），+1 下移
        ok = db.reorder_provider(data["id"], int(data.get("direction", 0)))
        return jsonify({"ok": ok, "error": "无法移动（已在边界或不存在）"} if not ok else {"ok": True})
    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("📮 新闻智能跟踪系统测试", "<h3>测试成功</h3><p>邮件通知链路正常。</p>")
            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()