llm-speed-tester/app.py

# -*- coding: utf-8 -*-
"""LLM 速度测试台 - Flask 主应用"""
import io
import json

import requests
from flask import Flask, jsonify, request, send_file, send_from_directory

import config
import database as db
from llm_providers import DEFAULT_URLS, ProviderError, call_stream
from tester import TestRunner

app = Flask(__name__, static_folder="static", static_url_path="")
app.json.ensure_ascii = False

db.init_db()

RUNNERS = {}  # test_id -> TestRunner


@app.route("/")
def index():
    return send_from_directory(app.static_folder, "index.html")


@app.route("/api/health")
def health():
    running = [tid for tid, r in RUNNERS.items() if r.is_alive()]
    return jsonify({"ok": True, "port": config.PORT, "running_tests": running})


def _fill_defaults(cfg):
    p = cfg.get("provider", "openai")
    if not cfg.get("base_url"):
        cfg["base_url"] = DEFAULT_URLS.get(p, "")
    if cfg.get("temperature") is None:
        cfg["temperature"] = 0.7
    return cfg


# ───────────────────────── 提供商配置 ─────────────────────────

@app.route("/api/configs", methods=["GET"])
def list_configs():
    return jsonify(db.list_configs())


@app.route("/api/configs", methods=["POST"])
def add_config():
    cfg = request.get_json(force=True) or {}
    if not cfg.get("name"):
        return jsonify({"ok": False, "error": "请填写配置名称"}), 400
    cid = db.add_config(cfg)
    return jsonify({"ok": True, "id": cid})


@app.route("/api/configs/<int:cid>", methods=["GET"])
def get_one_config(cid):
    c = db.get_config(cid)
    if not c:
        return jsonify({"ok": False, "error": "配置不存在"}), 404
    return jsonify(c)


@app.route("/api/configs/<int:cid>", methods=["PUT"])
def update_config(cid):
    cfg = request.get_json(force=True) or {}
    old = db.get_config(cid)
    if not old:
        return jsonify({"ok": False, "error": "配置不存在"}), 404
    db.update_config(cid, cfg)
    return jsonify({"ok": True, "id": cid})


@app.route("/api/configs/<int:cid>", methods=["DELETE"])
def del_config(cid):
    db.delete_config(cid)
    return jsonify({"ok": True})


@app.route("/api/configs/test", methods=["POST"])
def test_config():
    cfg = _fill_defaults(request.get_json(force=True) or {})
    if not cfg.get("api_key"):
        return jsonify({"ok": False, "error": "请填写 API Key"}), 400
    try:
        # 连接测试：只要流式请求成功返回（哪怕正文为空/只有思维链）都算连通
        m = call_stream(cfg, "你好，请简要回答：1+1=？",
                        {"max_tokens": 32, "avoid_cache": False})
        note = ""
        if not (m.get("output_tokens") or m.get("output_chars")):
            note = "（连接正常，但本次未返回正文内容，可能为推理型模型）"
        return jsonify({"ok": True, "total_ms": m["total_ms"], "metrics": m, "note": note})
    except ProviderError as e:
        return jsonify({"ok": False, "error": str(e)})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)})


# ───────────────────────── 测试 ─────────────────────────

@app.route("/api/tests", methods=["POST"])
def start_test():
    body = request.get_json(force=True) or {}
    cfg = _fill_defaults(body.get("config") or {})
    gen = body.get("gen") or {}
    if not cfg.get("api_key"):
        return jsonify({"ok": False, "error": "请填写 API Key"}), 400
    if not cfg.get("model"):
        return jsonify({"ok": False, "error": "请填写模型名称"}), 400
    tid = db.create_test(cfg, gen)
    runner = TestRunner(tid, cfg, gen)
    RUNNERS[tid] = runner
    runner.start()
    return jsonify({"ok": True, "id": tid})


@app.route("/api/tests", methods=["GET"])
def list_tests():
    try:
        limit = int(request.args.get("limit", 100))
    except ValueError:
        limit = 100
    return jsonify(db.list_tests(max(1, min(limit, 1000))))


@app.route("/api/tests/<int:tid>", methods=["GET"])
def get_test(tid):
    t = db.get_test(tid)
    if not t:
        return jsonify({"ok": False, "error": "测试不存在"}), 404
    t["runs"] = db.get_runs(tid)
    t["logs"] = db.get_logs(tid)
    _mask_cfg(t.get("config"))
    return jsonify(t)


@app.route("/api/tests/<int:tid>/logs", methods=["GET"])
def get_logs(tid):
    after = int(request.args.get("after", 0))
    data = db.get_logs_after(tid, after)
    if data is None:
        return jsonify({"ok": False, "error": "测试不存在"}), 404
    return jsonify(data)


@app.route("/api/tests/<int:tid>/cancel", methods=["POST"])
def cancel_test(tid):
    r = RUNNERS.get(tid)
    if r and r.is_alive():
        r.request_cancel()
        return jsonify({"ok": True, "msg": "正在停止..."})
    return jsonify({"ok": False, "msg": "测试未在运行"})


@app.route("/api/tests/<int:tid>", methods=["DELETE"])
def del_test(tid):
    db.delete_test(tid)
    RUNNERS.pop(tid, None)
    return jsonify({"ok": True})


# ───────────────────────── 图表（data-chart-tool 折线图：预填充左轴虚线 / 解码右轴实线） ─────────────────────────

def _build_chart_csv(t):
    """由测试汇总 by_length 构建画图 CSV：上下文长度, 预填充速度(tok/s), 解码速度(tok/s)"""
    s = t.get("summary") or {}
    by = s.get("by_length") or {}
    lens = sorted(int(k) for k in by)
    rows = []
    for L in lens:
        bl = by.get(str(L)) if str(L) in by else by.get(L) or {}
        pre = bl.get("avg_prefill_speed")
        dec = bl.get("avg_decode_speed")
        if pre is None or dec is None:
            continue
        rows.append([L, round(pre, 2), round(dec, 2)])
    if not rows:
        return None
    csv_lines = ["上下文长度, 预填充速度(tok/s), 解码速度(tok/s)"]
    for L, pre, dec in rows:
        csv_lines.append("%d, %.2f, %.2f" % (L, pre, dec))
    return {"csv": "\n".join(csv_lines), "rows": rows}


def _build_concurrency_chart_csv(t):
    """由测试汇总 by_concurrency 构建画图 CSV：并发数, 预填充速度(tok/s), 解码速度(tok/s)"""
    s = t.get("summary") or {}
    by = s.get("by_concurrency") or {}
    rows = []
    for C in sorted(int(k) for k in by):
        bl = by.get(str(C)) if str(C) in by else by.get(C) or {}
        pre = bl.get("avg_prefill_speed")
        dec = bl.get("avg_decode_speed")
        if pre is None or dec is None:
            continue
        rows.append([C, round(pre, 2), round(dec, 2)])
    if not rows:
        return None
    csv_lines = ["并发数, 预填充速度(tok/s), 解码速度(tok/s)"]
    for C, pre, dec in rows:
        csv_lines.append("%d, %.2f, %.2f" % (C, pre, dec))
    return {"csv": "\n".join(csv_lines), "rows": rows}


def _chart_payload(t, csv_text):
    """组装 data-chart-tool /api/chart 请求体（双Y轴折线图：长度对比）"""
    title = ("%s %s" % (t.get("model") or "", t.get("name") or "速度对比")).strip()
    return {
        "data": csv_text,
        "chartType": "line",
        "title": title,
        "theme": "default",
        "showLegend": True,
        "showGrid": True,
        "showLabel": False,
        "smoothLine": True,
        "dualYAxis": True,
        "leftAxisName": "预填充速度(tok/s)",
        "rightAxisName": "解码速度(tok/s)",
        "seriesTypes": ["line", "line"],
        "seriesAxis": [0, 1],
        "seriesStyles": ["dashed", "solid"],  # 预填充=左轴虚线，解码=右轴实线
        "width": 1000,
        "height": 560,
        "pixelRatio": 2,
    }


def _concurrency_chart_payload(t, csv_text):
    """并发对比折线图请求体（X=并发数，双Y轴：预填充左虚线 / 解码右实线）"""
    title = ("%s %s · 并发对比" % (t.get("model") or "", t.get("name") or "速度对比")).strip()
    p = _chart_payload(t, csv_text)
    p["title"] = title
    return p


def _fmt_num(v):
    if v is None or v == "":
        return ""
    try:
        return "%.2f" % float(v)
    except Exception:
        return ""


@app.route("/api/tests/<int:tid>/concurrency-chart-data")
def test_concurrency_chart_data(tid):
    t = db.get_test(tid)
    if not t:
        return jsonify({"ok": False, "error": "测试不存在"}), 404
    built = _build_concurrency_chart_csv(t)
    if not built:
        return jsonify({"ok": False, "error": "无并发分组采样数据（本次测试可能只测了单流），无法画图"}), 400
    return jsonify({
        "ok": True,
        "csv": built["csv"],
        "rows": built["rows"],
        "payload": _concurrency_chart_payload(t, built["csv"]),
    })


@app.route("/api/tests/<int:tid>/concurrency-chart")
def test_concurrency_chart(tid):
    """并发对比折线图 PNG（X=并发数，预填充左轴虚线 / 解码右轴实线）"""
    t = db.get_test(tid)
    if not t:
        return jsonify({"ok": False, "error": "测试不存在"}), 404
    built = _build_concurrency_chart_csv(t)
    if not built:
        return jsonify({"ok": False, "error": "无并发分组采样数据，无法画图"}), 400
    try:
        resp = requests.post(config.CHART_API_BASE + "/api/chart",
                             json=_concurrency_chart_payload(t, built["csv"]), timeout=60)
    except requests.RequestException as e:
        return jsonify({"ok": False, "error": "图表服务不可用: %s" % e}), 502
    if resp.status_code != 200:
        return jsonify({"ok": False, "error": "图表生成失败(%d): %s" % (resp.status_code, resp.text[:300])}), 502
    return send_file(io.BytesIO(resp.content), mimetype="image/png")


@app.route("/api/chart", methods=["POST"])
def chart_proxy():
    """通用图表代理：把任意 data-chart-tool /api/chart 请求体转发，返回 PNG（多测试对比用）"""
    payload = request.get_json(force=True) or {}
    try:
        resp = requests.post(config.CHART_API_BASE + "/api/chart", json=payload, timeout=60)
    except requests.RequestException as e:
        return jsonify({"ok": False, "error": "图表服务不可用: %s" % e}), 502
    if resp.status_code != 200:
        return jsonify({"ok": False, "error": "图表生成失败(%d): %s" % (resp.status_code, resp.text[:300])}), 502
    return send_file(io.BytesIO(resp.content), mimetype="image/png")


# ───────────────────────── 多测试对比 ─────────────────────────

@app.route("/api/compare", methods=["POST"])
def compare_tests():
    """把多个测试结果放在一起对比：返回对比表 + 柱状图CSV + 并发折线图CSV"""
    body = request.get_json(force=True) or {}
    ids = [int(x) for x in (body.get("ids") or []) if str(x).isdigit()][:20]
    rows = []
    for tid in ids:
        t = db.get_test(tid)
        if not t:
            continue
        s = t.get("summary") or {}
        g = t.get("gen") or {}
        cls = sorted(set(int(x) for x in (s.get("concurrency_levels") or g.get("concurrency_levels") or [1])))
        label = "#%d %s" % (t["id"], (t.get("name") or t.get("model") or "未命名"))
        rows.append({
            "id": t["id"],
            "label": label,
            "created_at": t.get("created_at", ""),
            "name": t.get("name", ""),
            "provider": t.get("provider", ""),
            "model": t.get("model", ""),
            "status": t.get("status", ""),
            "concurrency_levels": cls,
            "by_concurrency": s.get("by_concurrency") or {},
            "samples_ok": s.get("samples_ok"),
            "samples_total": s.get("samples_total"),
            "avg_ttft_ms": s.get("avg_ttft_ms"),
            "avg_prefill_speed": s.get("avg_prefill_speed"),
            "avg_decode_speed": s.get("avg_decode_speed"),
            "avg_stream_decode": s.get("avg_stream_decode"),
            "avg_output_tokens": s.get("avg_output_tokens"),
            "avg_total_ms": s.get("avg_total_ms"),
        })
    if not rows:
        return jsonify({"ok": False, "error": "未找到可对比的测试"}), 400

    # 柱状图：预填充 vs 解码（X=测试）
    bar_csv_lines = ["测试, 预填充速度(tok/s), 解码速度(tok/s)"]
    for r in rows:
        bar_csv_lines.append("%s, %s, %s" % (r["label"], _fmt_num(r["avg_prefill_speed"]), _fmt_num(r["avg_decode_speed"])))
    bar_csv = "\n".join(bar_csv_lines)
    bar_payload = {
        "data": bar_csv, "chartType": "bar",
        "title": "多测试速度对比（预填充空心 / 解码实心）",
        "theme": "default", "showLegend": True, "showGrid": True, "showLabel": True,
        "smoothLine": True,
        "seriesTypes": ["bar", "bar"],
        "seriesStyles": ["hollow", "solid"],
        "width": 1000, "height": 520, "pixelRatio": 2,
    }

    # 折线图：解码速度随并发变化（取各测试共同并发档，>=2 档才有意义）
    line_part = None
    if rows:
        common = sorted(set.intersection(*[set(r["concurrency_levels"]) for r in rows]))
        if len(common) >= 2:
            headers = ["并发数"] + [r["label"] for r in rows]
            lines = []
            for C in common:
                cells = [str(C)]
                for r in rows:
                    bc = r.get("by_concurrency") or {}
                    bl = bc.get(str(C)) if str(C) in bc else bc.get(C) or {}
                    cells.append(_fmt_num(bl.get("avg_decode_speed")))
                lines.append(", ".join(cells))
            line_csv = "\n".join([", ".join(headers)] + lines)
            nser = len(rows)
            line_payload = {
                "data": line_csv, "chartType": "line",
                "title": "解码速度随并发变化（各测试对比）",
                "theme": "default", "showLegend": True, "showGrid": True, "showLabel": False,
                "smoothLine": True,
                "seriesTypes": ["line"] * nser,
                "seriesStyles": (["solid", "dashed", "dotted"] * nser)[:nser],
                "width": 1000, "height": 520, "pixelRatio": 2,
            }
            line_part = {"csv": line_csv, "payload": line_payload, "levels": common}

    return jsonify({"ok": True, "rows": rows, "bar_csv": bar_csv,
                    "bar_payload": bar_payload, "line": line_part})


@app.route("/api/tests/<int:tid>/chart-data")
def test_chart_data(tid):
    t = db.get_test(tid)
    if not t:
        return jsonify({"ok": False, "error": "测试不存在"}), 404
    built = _build_chart_csv(t)
    if not built:
        return jsonify({"ok": False, "error": "无成功采样数据，无法画图"}), 400
    return jsonify({
        "ok": True,
        "csv": built["csv"],
        "rows": built["rows"],
        "payload": _chart_payload(t, built["csv"]),
    })


@app.route("/api/tests/<int:tid>/chart")
def test_chart(tid):
    """用 data-chart-tool 生成折线图 PNG（预填充左轴虚线 / 解码右轴实线）"""
    t = db.get_test(tid)
    if not t:
        return jsonify({"ok": False, "error": "测试不存在"}), 404
    built = _build_chart_csv(t)
    if not built:
        return jsonify({"ok": False, "error": "无成功采样数据，无法画图"}), 400
    try:
        resp = requests.post(config.CHART_API_BASE + "/api/chart",
                             json=_chart_payload(t, built["csv"]), timeout=60)
    except requests.RequestException as e:
        return jsonify({"ok": False, "error": "图表服务不可用: %s" % e}), 502
    if resp.status_code != 200:
        return jsonify({"ok": False, "error": "图表生成失败(%d): %s" % (resp.status_code, resp.text[:300])}), 502
    return send_file(io.BytesIO(resp.content), mimetype="image/png")


@app.route("/api/tests/<int:tid>/export.json")
def export_json(tid):
    t = db.get_test(tid)
    if not t:
        return jsonify({"ok": False, "error": "测试不存在"}), 404
    t["runs"] = db.get_runs(tid)
    t["logs"] = db.get_logs(tid)
    _mask_cfg(t.get("config"))
    return jsonify(t)


def _mask_cfg(cfg):
    """对外隐藏 API Key，仅保留前几位便于识别"""
    if cfg and cfg.get("api_key"):
        k = cfg["api_key"]
        cfg["api_key"] = k[:4] + "****" if len(k) > 6 else "****"
    return cfg


# ───────────────────────── Excel 导出 ─────────────────────────

@app.route("/api/tests/<int:tid>/export.xlsx")
def export_xlsx(tid):
    t = db.get_test(tid)
    if not t:
        return jsonify({"ok": False, "error": "测试不存在"}), 404
    t["runs"] = db.get_runs(tid)
    t["logs"] = db.get_logs(tid)
    try:
        data = _build_xlsx(t)
    except Exception as e:
        return jsonify({"ok": False, "error": "导出失败: %s" % e}), 500
    return send_file(data, as_attachment=True,
                     download_name="llm_speed_test_%d.xlsx" % tid,
                     mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")


def _build_xlsx(t):
    from openpyxl import Workbook
    from openpyxl.styles import Alignment, Font, PatternFill

    s = t.get("summary") or {}
    g = t.get("gen") or {}
    cfg = t.get("config") or {}
    by_length = s.get("by_length") or {}
    runs = t.get("runs") or []
    logs = t.get("logs") or []

    wb = Workbook()
    head_fill = PatternFill("solid", fgColor="2A3550")
    head_font = Font(color="FFFFFF", bold=True)
    title_font = Font(bold=True, size=12)

    def style_header(ws, row, ncol):
        for c in range(1, ncol + 1):
            cell = ws.cell(row=row, column=c)
            cell.fill = head_fill
            cell.font = head_font
            cell.alignment = Alignment(horizontal="center", vertical="center")

    # ── Sheet1 汇总 ──
    ws = wb.active
    ws.title = "汇总"
    ws.append(["LLM 速度测试报告"])
    ws.cell(1, 1).font = Font(bold=True, size=14)
    info = [
        ["测试编号", "#%d" % t["id"]],
        ["测试名称", t.get("name") or "（未命名）"],
        ["创建时间", t.get("created_at", "")],
        ["状态", t.get("status", "")],
        ["提供商", t.get("provider", "")],
        ["模型", t.get("model", "")],
        ["Base URL", cfg.get("base_url") or "（默认）"],
        ["上下文长度列表", " / ".join(str(x) for x in (g.get("context_lengths") or []))],
        ["并发数列表", " / ".join(str(x) for x in (s.get("concurrency_levels") or g.get("concurrency_levels") or [1]))],
        ["生成长度(max tokens)", g.get("max_tokens", 128)],
        ["每个长度采样次数", g.get("samples", 2)],
        ["预热(空转)", "开" if g.get("warmup", True) else "关"],
        ["避免缓存", "开" if g.get("avoid_cache") else "关"],
        ["采样(成功/总数)", "%s / %s" % (s.get("samples_ok"), s.get("samples_total"))],
        ["校准 字符/token", s.get("calibration_chars_per_token") or "—"],
        ["错误信息", t.get("error") or ""],
    ]
    for row in info:
        ws.append(row)
    ws.cell(15, 1).font = title_font
    r0 = len(info) + 2
    overall = [
        ["首字延迟(ms)", s.get("avg_ttft_ms"), s.get("max_ttft_ms"), s.get("min_ttft_ms")],
        ["预填充速度(tok/s)", s.get("avg_prefill_speed"), s.get("max_prefill_speed"), s.get("min_prefill_speed")],
        ["解码速度(tok/s)", s.get("avg_decode_speed"), s.get("max_decode_speed"), s.get("min_decode_speed")],
        ["提示词(tok)", s.get("avg_prompt_tokens"), None, None],
        ["输出(tok)", s.get("avg_output_tokens"), None, None],
        ["总耗时(ms)", s.get("avg_total_ms"), s.get("max_total_ms"), s.get("min_total_ms")],
    ]
    ws.cell(r0, 1, "整体统计指标（平均 / 最大 / 最小）").font = title_font
    for j, c in enumerate(["指标", "平均", "最大", "最小"], start=1):
        ws.cell(row=r0 + 1, column=j, value=c)
    style_header(ws, r0 + 1, 4)
    for i, row in enumerate(overall, start=r0 + 2):
        for j, v in enumerate(row, start=1):
            ws.cell(row=i, column=j, value=v)

    # 按上下文长度分组
    r1 = r0 + len(overall) + 3
    ws.cell(r1, 1, "按上下文长度分组").font = title_font
    cols = ["上下文长度(tok)", "采样(成功/总数)", "首字ms", "预填充tok/s", "解码tok/s", "提示词tok", "输出tok", "总耗时ms"]
    ws.append([])
    for j, c in enumerate(cols, start=1):
        ws.cell(row=r1 + 1, column=j, value=c)
    style_header(ws, r1 + 1, len(cols))
    if by_length:
        rr = r1 + 2
        for L in sorted(int(k) for k in by_length):
            bl = by_length[str(L)] if str(L) in by_length else by_length[L]
            ws.cell(row=rr, column=1, value=L)
            ws.cell(row=rr, column=2, value="%s / %s" % (bl.get("samples_ok"), bl.get("samples_total")))
            ws.cell(row=rr, column=3, value=bl.get("avg_ttft_ms"))
            ws.cell(row=rr, column=4, value=bl.get("avg_prefill_speed"))
            ws.cell(row=rr, column=5, value=bl.get("avg_decode_speed"))
            ws.cell(row=rr, column=6, value=bl.get("avg_prompt_tokens"))
            ws.cell(row=rr, column=7, value=bl.get("avg_output_tokens"))
            ws.cell(row=rr, column=8, value=bl.get("avg_total_ms"))
            rr += 1
    else:
        ws.cell(row=r1 + 2, column=1, value="（无成功采样数据）")

    # 按并发数分组（多测试/多并发对比核心数据）
    by_conc = s.get("by_concurrency") or {}
    r2 = r1 + (len(by_length) if by_length else 1) + 3
    ws.cell(r2, 1, "按并发数分组（整批吞吐，tok/s）").font = title_font
    ccols = ["并发数", "采样(成功/总数)", "首字ms", "预填充tok/s", "解码tok/s", "单流均解码tok/s", "输出tok", "总耗时ms"]
    ws.append([])
    for j, c in enumerate(ccols, start=1):
        ws.cell(row=r2 + 1, column=j, value=c)
    style_header(ws, r2 + 1, len(ccols))
    if by_conc:
        rr = r2 + 2
        for C in sorted(int(k) for k in by_conc):
            bl = by_conc[str(C)] if str(C) in by_conc else by_conc[C]
            ws.cell(row=rr, column=1, value=C)
            ws.cell(row=rr, column=2, value="%s / %s" % (bl.get("samples_ok"), bl.get("samples_total")))
            ws.cell(row=rr, column=3, value=bl.get("avg_ttft_ms"))
            ws.cell(row=rr, column=4, value=bl.get("avg_prefill_speed"))
            ws.cell(row=rr, column=5, value=bl.get("avg_decode_speed"))
            ws.cell(row=rr, column=6, value=bl.get("avg_stream_decode"))
            ws.cell(row=rr, column=7, value=bl.get("avg_output_tokens"))
            ws.cell(row=rr, column=8, value=bl.get("avg_total_ms"))
            rr += 1
    else:
        ws.cell(row=r2 + 2, column=1, value="（仅单流，无并发分组）")
    for col, w in zip("ABCDEFGH", [12, 20, 12, 14, 14, 16, 12, 14]):
        ws.column_dimensions[col].width = max(w, ws.column_dimensions[col].width or 0)

    # ── Sheet2 采样明细 ──
    ws2 = wb.create_sheet("采样明细")
    h2 = ["序号", "上下文长度tok", "并发", "流(成功/总数)", "提示词tok", "缓存tok", "首字ms", "预填充tok/s",
          "输出tok", "解码tok/s", "单流均解码tok/s", "总耗时ms", "备注"]
    ws2.append(h2)
    style_header(ws2, 1, len(h2))
    for i, r in enumerate(runs, start=1):
        m = r.get("metrics") or {}
        ws2.append([
            i,
            r.get("context_length") or m.get("context_length") or "",
            m.get("concurrency") or 1,
            "%s/%s" % (m.get("streams_ok"), m.get("streams_total")) if m.get("streams_total") else 1,
            m.get("prompt_tokens") or "",
            m.get("cached_tokens") if m.get("cached_tokens") else "",
            m.get("ttft_ms"),
            m.get("prefill_speed"),
            m.get("output_tokens"),
            m.get("decode_speed"),
            m.get("avg_stream_decode"),
            m.get("total_ms"),
            r.get("error") or "OK",
        ])
    for col, w in zip("ABCDEFGHIJKLM", [8, 14, 8, 14, 12, 10, 12, 14, 12, 14, 16, 12, 30]):
        ws2.column_dimensions[col].width = w

    # ── Sheet3 日志 ──
    ws3 = wb.create_sheet("日志")
    ws3.append(["相对时间(s)", "级别", "内容"])
    style_header(ws3, 1, 3)
    for l in logs:
        ws3.append([l.get("rel", 0), l.get("level", ""), l.get("msg", "")])
    for col, w in zip("ABC", [14, 10, 90]):
        ws3.column_dimensions[col].width = w

    bio = io.BytesIO()
    wb.save(bio)
    bio.seek(0)
    return bio


if __name__ == "__main__":
    app.run(host=config.HOST, port=config.PORT, threaded=True, debug=False)