workbuddy-portal/workbuddy_portal/web/views.py

# -*- coding: utf-8 -*-
"""页面路由（Jinja 模板）。

分工：
  /           概览（KPI + 入口）
  /dashboard  ECharts 交互大屏（独立静态页，登录后可达，数据走 /api/bundle）
  /tasks      任务管理：调度开关/时刻、手动触发、运行历史
  /config     配置管理：Cookie / UA / 采集参数 / 改密码
  /logs       日志管理：采集逐次明细 + 应用日志尾部
  /records    数据明细：分页、筛选、搜索、导出
"""
import csv
import io
import os
import sqlite3

from flask import (Blueprint, current_app, flash, jsonify, redirect, render_template,
                   request, send_from_directory, url_for)

from .. import collect, config, db, query, scheduler
from ..security import (admin_required, clear_fail, current_user, is_locked, lock_left,
                        login_ok, login_required, login_session, logout_session,
                        note_fail, safe_next)

bp = Blueprint("views", __name__)


def _ip():
    return request.headers.get("X-Forwarded-For", request.remote_addr or "").split(",")[0].strip()


# ---------------- 登录 ----------------
def _login_ctx(**kw):
    """登录页共用的上下文：锁定阈值/会话时长都从配置读，避免模板里写死数字。"""
    kw.setdefault("max_fails", config.MAX_LOGIN_FAILS)
    kw.setdefault("lock_minutes", config.LOGIN_LOCK_MINUTES)
    kw.setdefault("session_hours", config.SESSION_HOURS)
    return kw


@bp.route("/login", methods=["GET", "POST"])
def login():
    nxt = request.values.get("next") or ""
    if request.method == "POST":
        ip = _ip()
        if is_locked(ip):
            n = lock_left(ip)
            flash("登录失败次数过多，请 %d 秒后再试" % n, "error")
            return render_template("login.html", **_login_ctx(next_url=nxt)), 429
        username = (request.form.get("username") or "").strip()
        pwd = request.form.get("password") or ""
        conn = db.get_db()
        user = login_ok(conn, username, pwd)
        if user is None:
            n = note_fail(ip)
            db.audit(conn, "login_failed", username, "第 %d 次失败" % n, ip)
            flash("用户名或密码不正确（剩余尝试 %d 次）" % max(0, config.MAX_LOGIN_FAILS - n), "error")
            # 必须把 next 显式回填：失败后 request.args 为空，
            # 若模板从 request.args 取值会导致跳转目标丢失（历史 bug）。
            return render_template("login.html", **_login_ctx(username=username, next_url=nxt)), 401
        clear_fail(ip)
        login_session(user)
        db.audit(conn, "login", username, "登录成功", ip)
        return redirect(safe_next(nxt, url_for("views.overview")))
    if current_user():
        return redirect(url_for("views.overview"))
    return render_template("login.html", **_login_ctx(next_url=nxt))


@bp.post("/logout")
@login_required
def logout_post():
    """退出登录改为 POST + CSRF：GET 型退出会被 <img src> 这类请求静默触发。"""
    u = current_user()
    if u:
        db.audit(db.get_db(), "logout", u["username"], "", _ip())
    logout_session()
    flash("已退出登录", "ok")
    return redirect(url_for("views.login"))


@bp.get("/logout")
def logout():
    """保留 GET 入口：老书签/旧页面不会 405，但只做跳转不执行退出。"""
    if current_user():
        flash("为安全起见，退出登录请点页面右上角的「退出」按钮", "warn")
        return redirect(url_for("views.overview"))
    return redirect(url_for("views.login"))


# ---------------- 概览 ----------------
@bp.get("/")
@login_required
def overview():
    conn = db.get_db()
    mf = query.manifest(conn)
    t = query.totals(conn)
    today = db.now_str()[:10]
    st = query.summary(conn, today, today)
    d30 = query.summary(conn, _shift(-29), today)
    # 昨日对比：昨日整日 vs 今日（残日），让「今天偏少」有参照
    y = _shift(-1)
    yest = query.summary(conn, y, y)
    dims = query.dims(conn)
    # 注意：这里的 SQL 必须把模板用到的列都选出来（模板渲染 r.fetched，
    # 少选一列并不会报错，只会静默渲染成空白 —— 历史 bug）。
    runs = conn.execute(
        "SELECT id,trigger,status,started_at,duration_ms,fetched,added,dup,total,conflicts,message"
        " FROM collect_runs ORDER BY id DESC LIMIT 8").fetchall()
    return render_template("overview.html", mf=mf, totals=t, today_stat=st, stat30=d30,
                           yesterday=yest, yday=y,
                           models=dims["model"][:8], clients=dims["client"],
                           runs=runs, sch=_sch_info(conn), active="overview")


def _shift(days):
    from datetime import datetime, timedelta
    return (datetime.now() + timedelta(days=days)).strftime("%Y-%m-%d")


# ---------------- 大屏（独立 ECharts 页）----------------
@bp.get("/dashboard")
@bp.get("/dashboard/")
@login_required
def dashboard():
    # 大屏是独立静态页，但它也是导航的一项 —— 用 redirect 到一个带 active 的
    # 路由做不到（静态页由 send_from_directory 直接吐文件），所以这里只负责发文件；
    # 导航高亮改为让大屏页自己提供「返回后台」入口（见 index.html 的 topbar）。
    return send_from_directory(os.path.join(current_app.static_folder, "dashboard"), "index.html")


# ---------------- 任务管理 ----------------
def _sch_info(conn):
    sch = scheduler.get_scheduler()
    nxt = scheduler.next_run_at(conn)
    last = conn.execute("SELECT * FROM collect_runs ORDER BY id DESC LIMIT 1").fetchone()
    return {
        "running": sch.running,
        "enabled": db.get_bool(conn, "schedule_enabled", True),
        "times": scheduler.slots(conn),
        "next_run": nxt.strftime("%Y-%m-%d %H:%M:%S") if nxt else None,
        "catch_up": db.get_bool(conn, "catch_up", True),
        "interval": sch.interval,
        "last": dict(last) if last else None,
        "lock": os.path.exists(collect.LOCK_PATH),
        "now": db.now_str(),
    }


@bp.get("/tasks")
@login_required
def tasks():
    conn = db.get_db()
    page = _int_arg("page", 1, 1, 10 ** 6)
    size = 20
    total = conn.execute("SELECT COUNT(*) FROM collect_runs").fetchone()[0]
    runs = conn.execute("SELECT * FROM collect_runs ORDER BY id DESC LIMIT ? OFFSET ?",
                        (size, (page - 1) * size)).fetchall()
    s = db.get_settings(conn)
    pages = max(1, (total + size - 1) // size)
    return render_template("tasks.html", runs=runs, sch=_sch_info(conn),
                           s_times=s.get("schedule_times") or "",
                           s_grace=s.get("catch_up_grace_hours") or "12",
                           page=page, pages=pages, total=total,
                           page_window=_page_window(page, pages),
                           active="tasks")


def _int_arg(name, default, lo, hi):
    """健壮地取整数查询参数：非法值回落到 default，不抛异常（历史 500 来源）。"""
    try:
        return max(lo, min(hi, int(request.args.get(name) or default)))
    except (TypeError, ValueError):
        return default


def _page_window(page, pages, span=9):
    """分页器只渲染当前页附近的一段，避免几百个页码链接把页面撑长。"""
    if pages <= span:
        return list(range(1, pages + 1))
    half = span // 2
    lo = max(1, min(page - half, pages - span + 1))
    return list(range(lo, lo + span))


# ---------------- 配置管理 ----------------
@bp.get("/config")
@login_required
def config_page():
    conn = db.get_db()
    s = db.get_settings(conn)
    for k in [k for k in list(s) if config.is_internal_key(k)]:
        s.pop(k, None)
    cookie = (s.pop("cookie", "") or "")
    s["cookie_hint"] = ("%d 字符，结尾 …%s" % (len(cookie), cookie[-16:])) if cookie else ""
    return render_template("config.html", s=s, sch=_sch_info(conn),
                           secret_keys=config.SECRET_KEYS,
                           num_settings=config.NUM_SETTINGS,
                           active="config")


# ---------------- 用户管理 ----------------
@bp.get("/users")
@admin_required
def users_page():
    conn = db.get_db()
    users = conn.execute(
        "SELECT id,username,display_name,is_admin,created_at,last_login_at,login_count"
        " FROM users ORDER BY id").fetchall()
    audits = conn.execute("SELECT * FROM audit_log WHERE action LIKE 'user%'"
                          " ORDER BY id DESC LIMIT 20").fetchall()
    return render_template("users.html", users=users, audits=audits,
                           me=current_user(), active="users")


# ---------------- 日志管理 ----------------
@bp.get("/logs")
@login_required
def logs():
    conn = db.get_db()
    run_id = request.args.get("run")
    detail = None
    if run_id and str(run_id).isdigit():
        detail = conn.execute("SELECT * FROM collect_runs WHERE id=?", (int(run_id),)).fetchone()
    status = request.args.get("status") or ""
    w, p = ("WHERE status = ?", [status]) if status in ("ok", "warn", "error", "running") else ("", [])
    # 操作审计：按动作筛选 + 分页（原来只能看最近 40 条，等于不可查）
    act = request.args.get("act") or ""
    aw, ap = ("WHERE action = ?", [act]) if act else ("", [])
    apage = _int_arg("apage", 1, 1, 10 ** 6)
    asize = 20
    atotal = conn.execute("SELECT COUNT(*) FROM audit_log %s" % aw, ap).fetchone()[0]
    audits = conn.execute("SELECT * FROM audit_log %s ORDER BY id DESC LIMIT ? OFFSET ?" % aw,
                          ap + [asize, (apage - 1) * asize]).fetchall()
    # 注意传的是 sqlite3.Row 列表而不是纯字符串列表：模板要用 a[0]=动作、a[1]=次数，
    # 若在这里就用推导式取 r[0]，模板里的 a[0] 会变成「字符串的第一个字符」。
    actions = conn.execute(
        "SELECT action, COUNT(*) n FROM audit_log GROUP BY action ORDER BY n DESC, action").fetchall()
    page = _int_arg("page", 1, 1, 10 ** 6)
    size = 30
    total = conn.execute("SELECT COUNT(*) FROM collect_runs %s" % w, p).fetchone()[0]
    runs = conn.execute("SELECT id,trigger,status,started_at,duration_ms,fetched,added,dup,total,"
                        "conflicts,exit_code,message FROM collect_runs %s"
                        " ORDER BY id DESC LIMIT ? OFFSET ?" % w, p + [size, (page - 1) * size]).fetchall()
    apages = max(1, (atotal + asize - 1) // asize)
    return render_template("logs.html", runs=runs, detail=detail, audits=audits,
                           actions=actions, act=act, apage=apage, apages=apages, atotal=atotal,
                           apage_window=_page_window(apage, apages, span=7),
                           page=page, pages=max(1, (total + size - 1) // size), total=total,
                           page_window=_page_window(page, max(1, (total + size - 1) // size)),
                           status=status,
                           active="logs")


@bp.get("/logs/tail")
@login_required
def logs_tail():
    n = _int_arg("lines", 200, 10, 2000)
    path = config.APP_LOG
    if not os.path.exists(path):
        return jsonify({"lines": [], "path": path, "size": 0})
    # 只保留尾部 n 行，不把整个日志文件读进内存
    from collections import deque
    with open(path, "r", encoding="utf-8", errors="replace") as f:
        lines = list(deque(f, maxlen=n))
    return jsonify({"lines": [l.rstrip("\n") for l in lines], "path": path,
                    "size": os.path.getsize(path)})


# ---------------- 数据明细 ----------------
def _day_args():
    """取明细页的日期区间。

    规范参数名是 from / to，但模板内部为了避免与 Python 关键字混淆用的是
    frm，历史上也出现过 `?frm=` 的链接，这里一并接受，避免导出/翻页丢筛选条件。
    """
    a = request.args
    frm = query.norm_day(a.get("from") or a.get("frm"))
    to = query.norm_day(a.get("to"))
    if frm and to and frm > to:
        frm, to = to, frm
    return frm, to


@bp.get("/records")
@login_required
def records():
    conn = db.get_db()
    frm, to = _day_args()
    model = request.args.get("model") or None
    client = request.args.get("client") or None
    q = request.args.get("q") or None
    order = request.args.get("order") or "ts_desc"
    page = _int_arg("page", 1, 1, 10 ** 6)
    size = _int_arg("size", 50, 10, query.MAX_PAGE_SIZE)
    data = query.records_page(conn, frm, to, model=model, client=client, q=q,
                              page=page, size=size, order=order)
    # 只算一次 dims：query.dims() 内部有 3 条 GROUP BY，重复调用纯属浪费
    d = query.dims(conn)
    models = [r["name"] for r in d["model"]]
    clients = [r["name"] for r in d["client"]]
    # 注意：不要把含 "items" 键的 dict 直接交给模板——Jinja 的属性查找会先命中
    # dict.items 这个方法而不是数据，所以这里拆出独立变量。
    return render_template("records.html", data=data, rows=data["items"], models=models,
                           clients=clients,
                           page_window=_page_window(page, data["pages"]),
                           f=dict(frm=frm or "", to=to or "", model=model or "",
                                  client=client or "", q=q or "", order=order, size=size),
                           active="records")


@bp.get("/records/export")
@login_required
def records_export():
    """按当前筛选条件导出 CSV（与官网 xlsx 同构的列）。

    用 csv 模块逐行写（原来手工拼字符串，model/client 含逗号或引号时会串列），
    数据用 query.iter_records 流式取，不把整个结果集读进内存。
    """
    from flask import Response
    frm, to = _day_args()
    model = request.args.get("model") or None
    client = request.args.get("client") or None
    q = request.args.get("q") or None
    order = request.args.get("order") or "ts_desc"

    def gen():
        yield "\ufeff"                       # UTF-8 BOM：Excel 直接双击不乱码
        buf = io.StringIO()
        w = csv.writer(buf, lineterminator="\r\n")
        w.writerow(collect.FIELDS)
        yield buf.getvalue()
        # 重要：流式响应的 body 是在视图返回、请求上下文被 pop（teardown 里
        # close_db 已把 g.db 关掉）之后才被 WSGI 服务器逐块拉取的，
        # 所以这里绝不能复用 db.get_db() 的连接——那会直接
        # `sqlite3.ProgrammingError: Cannot operate on a closed database`。
        # 自己开一条连接，并在流结束时关掉。
        own = db.connect()
        try:
            for r in query.iter_records(own, frm, to, model=model, client=client,
                                        q=q, order=order):
                buf.seek(0)
                buf.truncate(0)
                w.writerow([r["request_id"], "%.2f" % r["credits"], r["prompt"] or "",
                            r["model"], r["client"], r["ts"]])
                yield buf.getvalue()
        finally:
            own.close()

    name = "usage_%s_%s.csv" % (frm or "all", to or db.now_str()[:10])
    resp = Response(gen(), mimetype="text/csv; charset=utf-8",
                    headers={"Content-Disposition": 'attachment; filename="%s"' % name})
    # 导出可能很慢，避免 nginx 之类的前置代理先缓冲整个响应体
    resp.headers["X-Accel-Buffering"] = "no"
    return resp


# ---------------- 兼容旧地址 ----------------
@bp.get("/index.html")
def legacy_index():
    return redirect(url_for("views.dashboard"))