782916a283
Give EvalRun.summary a typed RunSummary value (unified RunError, lenient legacy parsing) so readers stop reaching into a schemaless dict, and route every cross-run rollup — dashboard, scenario ranking, trend, campaign report — through one aggregate_runs seam. Fixes the divergence where stats averaged pass_rate over completed-only runs while the campaign report counted faults as 0.0. Cross-run rule (ADR-0004): genuine faults count 0.0, user-cancelled runs are excluded from both denominators.
557 lines
22 KiB
Python
557 lines
22 KiB
Python
"""Report generation for evaluation runs."""
|
||
|
||
import json
|
||
from collections import defaultdict
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any, Optional
|
||
|
||
from jinja2 import Template
|
||
|
||
from agenteval.evaluation.metrics import aggregate_runs
|
||
from agenteval.models import Campaign, EvalRun, RunStatus, RunSummary
|
||
from agenteval.storage.db import DATA_DIR, iso_utc
|
||
from agenteval.storage.repository import RunRepository, ScenarioRepository, TargetRepository
|
||
from agenteval.utils.llm import extract_reply_text
|
||
|
||
HTML_TEMPLATE = """<!DOCTYPE html>
|
||
<html lang="zh-CN">
|
||
<head>
|
||
<meta charset="UTF-8">
|
||
<title>评测报告 - {{ report.run_id }}</title>
|
||
<style>
|
||
body { font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif; margin: 40px; background: #f5f5f5; }
|
||
.container { max-width: 960px; margin: 0 auto; background: #fff; padding: 32px; border-radius: 8px; box-shadow: 0 2px 8px rgba(0,0,0,0.05); }
|
||
h1 { margin-top: 0; }
|
||
.summary { display: grid; grid-template-columns: repeat(auto-fit, minmax(140px, 1fr)); gap: 16px; margin: 24px 0; }
|
||
.card { background: #fafafa; border-radius: 6px; padding: 16px; text-align: center; }
|
||
.card .value { font-size: 24px; font-weight: 700; }
|
||
.card .label { color: #666; font-size: 14px; margin-top: 4px; }
|
||
.case { border: 1px solid #e0e0e0; border-radius: 6px; margin: 16px 0; padding: 16px; }
|
||
.case-title { font-weight: 600; margin-bottom: 8px; }
|
||
.turn { background: #f9f9f9; border-radius: 4px; padding: 12px; margin: 8px 0; }
|
||
.message-label { color: #666; font-size: 12px; }
|
||
.rule { display: flex; align-items: center; gap: 8px; margin: 6px 0; }
|
||
.badge { padding: 2px 8px; border-radius: 4px; font-size: 12px; }
|
||
.pass { background: #e6f7e6; color: #2e7d32; }
|
||
.fail { background: #ffebee; color: #c62828; }
|
||
pre { white-space: pre-wrap; word-break: break-word; background: #f5f5f5; padding: 8px; border-radius: 4px; }
|
||
</style>
|
||
</head>
|
||
<body>
|
||
<div class="container">
|
||
<h1>评测报告</h1>
|
||
<p>评测对象：{{ report.target_name }}（{{ report.target_id }}）</p>
|
||
<p>评测场景：{{ report.scenario_name }}（{{ report.scenario_id }}）</p>
|
||
<p>执行时间：{{ report.started_at }} 至 {{ report.completed_at or '进行中' }}</p>
|
||
|
||
<div class="summary">
|
||
<div class="card">
|
||
<div class="value">{{ report.summary.total_cases }}</div>
|
||
<div class="label">用例总数</div>
|
||
</div>
|
||
<div class="card">
|
||
<div class="value">{{ report.summary.passed_cases }}</div>
|
||
<div class="label">通过用例</div>
|
||
</div>
|
||
<div class="card">
|
||
<div class="value">{{ report.summary.total_rules }}</div>
|
||
<div class="label">规则总数</div>
|
||
</div>
|
||
<div class="card">
|
||
<div class="value">{{ "%.2f"|format(report.summary.pass_rate * 100) }}%</div>
|
||
<div class="label">规则通过率</div>
|
||
</div>
|
||
</div>
|
||
|
||
{% for case in report.cases %}
|
||
<div class="case">
|
||
<div class="case-title">用例 {{ case.case_id }}</div>
|
||
{% for turn in case.turns %}
|
||
<div class="turn">
|
||
<div class="message-label">用户消息</div>
|
||
<pre>{{ turn.sent_text }}</pre>
|
||
<div class="message-label">智能体回复（{{ turn.latency_ms }}ms）</div>
|
||
<pre>{{ turn.reply_text or '（无回复）' }}</pre>
|
||
</div>
|
||
{% endfor %}
|
||
<div>
|
||
{% for result in case.results %}
|
||
<div class="rule">
|
||
<span class="badge {{ 'pass' if result.passed else 'fail' }}">{{ '通过' if result.passed else '失败' }}</span>
|
||
<span>{{ result.rule_type }}: {{ result.reason }}</span>
|
||
</div>
|
||
{% endfor %}
|
||
</div>
|
||
</div>
|
||
{% endfor %}
|
||
</div>
|
||
</body>
|
||
</html>
|
||
"""
|
||
|
||
|
||
def _extract_text(data: Any) -> str:
|
||
return extract_reply_text(data)
|
||
|
||
|
||
def generate_report(run_id: str, session=None) -> dict[str, Any]:
|
||
"""Build a structured report dict for a run."""
|
||
run_repo = RunRepository(session)
|
||
target_repo = TargetRepository(session)
|
||
scenario_repo = ScenarioRepository(session)
|
||
|
||
run = run_repo.get(run_id)
|
||
if not run:
|
||
raise ValueError(f"run not found: {run_id}")
|
||
|
||
target = target_repo.get(run.target_id)
|
||
scenario = scenario_repo.get(run.scenario_id)
|
||
|
||
turns = run_repo.get_turns(run_id)
|
||
results = run_repo.get_results(run_id)
|
||
|
||
# Group by case
|
||
case_map: dict[str, dict[str, Any]] = {}
|
||
for turn in turns:
|
||
case_map.setdefault(turn.case_id, {"turns": [], "results": [], "all_replied": True})
|
||
sent = turn.get_sent_message()
|
||
reply = turn.get_reply()
|
||
if reply is None:
|
||
case_map[turn.case_id]["all_replied"] = False
|
||
case_map[turn.case_id]["turns"].append(
|
||
{
|
||
"round": turn.round_index,
|
||
"sent_text": _extract_text(sent.get("msgBody")),
|
||
"reply_text": _extract_text(reply.get("msgBody") if reply else None),
|
||
"latency_ms": turn.latency_ms,
|
||
"question_msg_id": turn.question_msg_id,
|
||
}
|
||
)
|
||
|
||
for result in results:
|
||
case_map.setdefault(result.case_id, {"turns": [], "results": [], "all_replied": True})
|
||
case_map[result.case_id]["results"].append(
|
||
{
|
||
"rule_type": result.rule_type,
|
||
"passed": result.passed,
|
||
"score": result.score,
|
||
"reason": result.reason,
|
||
}
|
||
)
|
||
|
||
summary = run.summary or RunSummary()
|
||
errored_case_ids = {e.get("case_id") for e in summary.case_errors}
|
||
# 权威判定：引擎经 judgement.combine_case_outcome 算一次写入 summary；
|
||
# 老 run 没有该字段时退回从持久化结果反推（WEIGHTED/ANY 只能近似）。
|
||
authoritative = summary.case_outcomes
|
||
|
||
cases = []
|
||
for case_id in sorted(case_map.keys()):
|
||
item = case_map[case_id]
|
||
if case_id in authoritative:
|
||
outcome = authoritative[case_id]
|
||
connectivity = outcome.connectivity
|
||
passed = outcome.passed
|
||
else:
|
||
# 连通用例：无任何判定结果，且每轮都收到回复、无用例级错误（CONTEXT.md）
|
||
connectivity = (
|
||
not item["results"]
|
||
and bool(item["turns"])
|
||
and item["all_replied"]
|
||
and case_id not in errored_case_ids
|
||
)
|
||
if connectivity:
|
||
passed = True
|
||
elif not item["results"]:
|
||
# 故障用例（无结果且非连通）＝不通过（ADR-0002）
|
||
passed = False
|
||
else:
|
||
passed = all(r["passed"] for r in item["results"])
|
||
cases.append(
|
||
{
|
||
"case_id": case_id,
|
||
"passed": passed,
|
||
"connectivity": connectivity,
|
||
"turns": sorted(item["turns"], key=lambda x: x["round"]),
|
||
"results": item["results"],
|
||
}
|
||
)
|
||
|
||
total_cases = summary.total_cases
|
||
passed_cases = summary.passed_cases
|
||
connectivity_count = sum(1 for c in cases if c["connectivity"])
|
||
judged_total = total_cases - connectivity_count
|
||
# 连通用例按引擎口径计通过，判定型通过数 = 总通过数 - 连通用例数
|
||
judged_pass_rate = round((passed_cases - connectivity_count) / judged_total, 4) if judged_total > 0 else None
|
||
|
||
return {
|
||
"run_id": run.id,
|
||
"target_id": run.target_id,
|
||
"target_name": target.name if target else "未知",
|
||
"scenario_id": run.scenario_id,
|
||
"scenario_name": scenario.name if scenario else "未知",
|
||
"scenario_version": run.scenario_version,
|
||
"triggered_by": run.triggered_by.value,
|
||
"status": run.status.value,
|
||
"started_at": iso_utc(run.started_at),
|
||
"completed_at": iso_utc(run.completed_at),
|
||
"summary": {
|
||
"total_cases": total_cases,
|
||
"passed_cases": passed_cases,
|
||
"failed_cases": summary.failed_cases,
|
||
"total_rules": summary.total_rules,
|
||
"passed_rules": summary.passed_rules,
|
||
"pass_rate": summary.pass_rate if summary.pass_rate is not None else 0.0,
|
||
"connectivity_cases": connectivity_count,
|
||
"judged_pass_rate": judged_pass_rate,
|
||
},
|
||
"cases": cases,
|
||
}
|
||
|
||
|
||
def generate_compare_report(run_id_1: str, run_id_2: str, session=None) -> dict[str, Any]:
|
||
"""Build a side-by-side comparison dict for two runs of the same scenario."""
|
||
report_a = generate_report(run_id_1, session)
|
||
report_b = generate_report(run_id_2, session)
|
||
|
||
# Cross-scenario case_ids never overlap, so every case would be flagged
|
||
# "changed" and the diff would be meaningless — reject early.
|
||
if report_a.get("scenario_id") != report_b.get("scenario_id"):
|
||
raise ValueError("compare report requires both runs to use the same scenario")
|
||
# 同场景还须同考纲版本才可比（ADR-0001）
|
||
if report_a.get("scenario_version") != report_b.get("scenario_version"):
|
||
raise ValueError(
|
||
"compare report requires the same scenario version "
|
||
f"(A: v{report_a.get('scenario_version')}, B: v{report_b.get('scenario_version')})"
|
||
)
|
||
|
||
def _summary_delta(key: str) -> float:
|
||
return report_b["summary"][key] - report_a["summary"][key]
|
||
|
||
# Case-level diff: match by case_id
|
||
cases_a = {c["case_id"]: c for c in report_a.get("cases", [])}
|
||
cases_b = {c["case_id"]: c for c in report_b.get("cases", [])}
|
||
all_case_ids = sorted(set(cases_a) | set(cases_b))
|
||
|
||
case_diffs = []
|
||
for cid in all_case_ids:
|
||
ca = cases_a.get(cid)
|
||
cb = cases_b.get(cid)
|
||
|
||
def _case_passed(c):
|
||
# None 仅表示该 run 没有这个用例；判定本身读权威 passed 字段
|
||
return c["passed"] if c else None
|
||
|
||
case_diffs.append(
|
||
{
|
||
"case_id": cid,
|
||
"connectivity": bool((ca and ca.get("connectivity")) or (cb and cb.get("connectivity"))),
|
||
"run_a_passed": _case_passed(ca),
|
||
"run_b_passed": _case_passed(cb),
|
||
"changed": _case_passed(ca) != _case_passed(cb),
|
||
"run_a_results": ca["results"] if ca else [],
|
||
"run_b_results": cb["results"] if cb else [],
|
||
}
|
||
)
|
||
|
||
return {
|
||
"run_a": {
|
||
"run_id": run_id_1,
|
||
"target_name": report_a.get("target_name"),
|
||
"scenario_name": report_a.get("scenario_name"),
|
||
"scenario_version": report_a.get("scenario_version"),
|
||
"triggered_by": report_a.get("triggered_by"),
|
||
"status": report_a.get("status"),
|
||
"started_at": report_a.get("started_at"),
|
||
"summary": report_a["summary"],
|
||
},
|
||
"run_b": {
|
||
"run_id": run_id_2,
|
||
"target_name": report_b.get("target_name"),
|
||
"scenario_name": report_b.get("scenario_name"),
|
||
"scenario_version": report_b.get("scenario_version"),
|
||
"triggered_by": report_b.get("triggered_by"),
|
||
"status": report_b.get("status"),
|
||
"started_at": report_b.get("started_at"),
|
||
"summary": report_b["summary"],
|
||
},
|
||
"delta": {
|
||
"pass_rate": round(_summary_delta("pass_rate"), 4),
|
||
"passed_cases": int(_summary_delta("passed_cases")),
|
||
"passed_rules": int(_summary_delta("passed_rules")),
|
||
},
|
||
"cases": case_diffs,
|
||
"changed_cases": sum(1 for c in case_diffs if c["changed"]),
|
||
}
|
||
|
||
|
||
def _to_utc(dt: Optional[datetime]) -> Optional[datetime]:
|
||
if dt is None:
|
||
return None
|
||
return dt.replace(tzinfo=timezone.utc) if dt.tzinfo is None else dt
|
||
|
||
|
||
def _aggregate_runs(runs: list[EvalRun]) -> dict[str, Any]:
|
||
"""Delegates to the single cross-run aggregation seam (ADR-0004)."""
|
||
return aggregate_runs(runs)
|
||
|
||
|
||
def summarize_campaign_progress(campaign: Campaign, runs: list[EvalRun]) -> dict[str, Any]:
|
||
"""Compact list-row progress: completed vs *planned* total, plus pass_rate.
|
||
|
||
Unlike ``campaign_progress`` (live window position), this powers the list
|
||
view. ``planned_total`` is the sum of plan-entry counts — a fixed target the
|
||
campaign works toward, so the progress bar fills from 0 rather than tracking
|
||
a growing spawned count. ``overall_pass_rate`` reuses ``_aggregate_runs`` so
|
||
the list figure matches the report exactly (ADR-0002: failures count as 0.0).
|
||
"""
|
||
return {
|
||
"completed_runs": sum(1 for r in runs if r.status == RunStatus.COMPLETED),
|
||
"planned_total": sum(entry.count for entry in campaign.plan),
|
||
"overall_pass_rate": _aggregate_runs(runs)["pass_rate"],
|
||
}
|
||
|
||
|
||
def generate_campaign_report(
|
||
campaign: Campaign,
|
||
runs: list[EvalRun],
|
||
*,
|
||
scenario_names: Optional[dict[str, str]] = None,
|
||
bucket_count: int = 12,
|
||
) -> dict[str, Any]:
|
||
"""Build a dual-axis periodic report for a campaign from its child Runs.
|
||
|
||
Axis 1 (time trend): child Runs bucketed by their position in the service
|
||
window, each bucket carrying pass_rate / availability / latency. Axis 2
|
||
(capability summary): the same measures grouped by scenario across the whole
|
||
window. Pure function — no I/O; ``scenario_names`` maps ids to display names.
|
||
|
||
``time_scale`` is used *only* to place each Run into the right window-time
|
||
bucket (so a compressed dev run still reports "hour 0-2, 2-4, ..."); it never
|
||
changes any aggregated number, keeping figures comparable across lines.
|
||
"""
|
||
scenario_names = scenario_names or {}
|
||
window = float(campaign.window_seconds)
|
||
bucket_seconds = window / bucket_count if bucket_count else window
|
||
campaign_start = _to_utc(campaign.started_at)
|
||
|
||
# ── Axis 1: time trend ────────────────────────────────────────────────
|
||
buckets: dict[int, list[EvalRun]] = defaultdict(list)
|
||
for run in runs:
|
||
run_start = _to_utc(run.started_at)
|
||
if campaign_start is None or run_start is None:
|
||
offset = 0.0
|
||
else:
|
||
offset = (run_start - campaign_start).total_seconds() * campaign.time_scale
|
||
offset = max(0.0, min(offset, window))
|
||
idx = min(int(offset / bucket_seconds), bucket_count - 1) if bucket_seconds else 0
|
||
buckets[idx].append(run)
|
||
|
||
time_trend = []
|
||
for idx in range(bucket_count):
|
||
agg = _aggregate_runs(buckets.get(idx, []))
|
||
time_trend.append({
|
||
"bucket_index": idx,
|
||
"start_seconds": round(idx * bucket_seconds, 3),
|
||
"end_seconds": round((idx + 1) * bucket_seconds, 3),
|
||
**agg,
|
||
})
|
||
|
||
# ── Axis 2: capability summary (by scenario) ──────────────────────────
|
||
by_scenario: dict[str, list[EvalRun]] = defaultdict(list)
|
||
for run in runs:
|
||
by_scenario[run.scenario_id].append(run)
|
||
capability_summary = []
|
||
for sid, sruns in by_scenario.items():
|
||
agg = _aggregate_runs(sruns)
|
||
capability_summary.append({
|
||
"scenario_id": sid,
|
||
"scenario_name": scenario_names.get(sid, (sid or "")[:8]),
|
||
**agg,
|
||
})
|
||
capability_summary.sort(key=lambda s: s["run_count"], reverse=True)
|
||
|
||
overall = _aggregate_runs(runs)
|
||
return {
|
||
"campaign_id": campaign.id,
|
||
"name": campaign.name,
|
||
"target_id": campaign.target_id,
|
||
"status": campaign.status.value,
|
||
"window_seconds": campaign.window_seconds,
|
||
"time_scale": campaign.time_scale,
|
||
"started_at": iso_utc(campaign.started_at),
|
||
"completed_at": iso_utc(campaign.completed_at),
|
||
"summary": {
|
||
"total_runs": len(runs),
|
||
"completed_runs": sum(1 for r in runs if r.status == RunStatus.COMPLETED),
|
||
"overall_pass_rate": overall["pass_rate"],
|
||
"overall_availability": overall["availability"],
|
||
"avg_latency_ms": overall["avg_latency_ms"],
|
||
},
|
||
"time_trend": time_trend,
|
||
"capability_summary": capability_summary,
|
||
}
|
||
|
||
|
||
def render_campaign_markdown_report(
|
||
campaign: Campaign,
|
||
runs: list[EvalRun],
|
||
*,
|
||
scenario_names: Optional[dict[str, str]] = None,
|
||
) -> str:
|
||
"""Render the dual-axis campaign report as Markdown (reuses the export path)."""
|
||
report = generate_campaign_report(campaign, runs, scenario_names=scenario_names)
|
||
s = report["summary"]
|
||
|
||
def _pct(v: Optional[float]) -> str:
|
||
return "—" if v is None else f"{v * 100:.1f}%"
|
||
|
||
def _ms(v: Optional[float]) -> str:
|
||
return "—" if v is None else f"{v:.0f}ms"
|
||
|
||
lines: list[str] = [
|
||
f"# 活动周期报告 — {report['name']}",
|
||
"",
|
||
f"**状态**: {report['status']} ",
|
||
f"**窗口**: {report['window_seconds']}s（倍速 {report['time_scale']}） ",
|
||
f"**开始时间**: {report['started_at'] or '-'} ",
|
||
f"**完成时间**: {report['completed_at'] or '-'} ",
|
||
"",
|
||
"## 汇总",
|
||
"",
|
||
"| 指标 | 数值 |",
|
||
"|------|------|",
|
||
f"| 子运行总数 | {s['total_runs']} |",
|
||
f"| 已完成 | {s['completed_runs']} |",
|
||
f"| 整窗通过率 | {_pct(s['overall_pass_rate'])} |",
|
||
f"| 整窗可用性 | {_pct(s['overall_availability'])} |",
|
||
f"| 平均时延 | {_ms(s['avg_latency_ms'])} |",
|
||
"",
|
||
"## 时间趋势",
|
||
"",
|
||
"| 时段(秒) | 运行数 | 通过率 | 可用性 | 时延 |",
|
||
"|------|------|------|------|------|",
|
||
]
|
||
for b in report["time_trend"]:
|
||
lines.append(
|
||
f"| {b['start_seconds']:.0f}–{b['end_seconds']:.0f} | {b['run_count']} | "
|
||
f"{_pct(b['pass_rate'])} | {_pct(b['availability'])} | {_ms(b['avg_latency_ms'])} |"
|
||
)
|
||
lines += [
|
||
"",
|
||
"## 能力汇总",
|
||
"",
|
||
"| 场景 | 运行数 | 通过率 | 可用性 | 时延 |",
|
||
"|------|------|------|------|------|",
|
||
]
|
||
for c in report["capability_summary"]:
|
||
lines.append(
|
||
f"| {c['scenario_name']} | {c['run_count']} | {_pct(c['pass_rate'])} | "
|
||
f"{_pct(c['availability'])} | {_ms(c['avg_latency_ms'])} |"
|
||
)
|
||
return "\n".join(lines)
|
||
|
||
|
||
def render_markdown_report(run_id: str, session=None) -> str:
|
||
"""Render a report as Markdown string."""
|
||
report = generate_report(run_id, session)
|
||
s = report["summary"]
|
||
judged_rate = s.get("judged_pass_rate")
|
||
judged_rate_text = "—" if judged_rate is None else f"{judged_rate * 100:.1f}%"
|
||
lines: list[str] = [
|
||
f"# 评测报告 — {report.get('scenario_name', run_id)}",
|
||
"",
|
||
f"**评测对象**: {report.get('target_name', '-')} ",
|
||
f"**评测场景**: {report.get('scenario_name', '-')} ",
|
||
f"**状态**: {report.get('status', '-')} ",
|
||
f"**开始时间**: {report.get('started_at', '-')} ",
|
||
f"**完成时间**: {report.get('completed_at', '-')} ",
|
||
"",
|
||
"## 汇总",
|
||
"",
|
||
"| 指标 | 数值 |",
|
||
"|------|------|",
|
||
f"| 总用例数 | {s['total_cases']} |",
|
||
f"| 通过用例 | {s['passed_cases']} |",
|
||
f"| 失败用例 | {s['failed_cases']} |",
|
||
f"| 总规则数 | {s['total_rules']} |",
|
||
f"| 通过规则 | {s['passed_rules']} |",
|
||
f"| 通过率 | {s['pass_rate'] * 100:.1f}% |",
|
||
f"| 连通用例 | {s.get('connectivity_cases', 0)} |",
|
||
f"| 判定型通过率 | {judged_rate_text} |",
|
||
"",
|
||
"## 用例明细",
|
||
"",
|
||
]
|
||
|
||
for case in report.get("cases", []):
|
||
if case.get("connectivity"):
|
||
badge = "🔗"
|
||
else:
|
||
badge = "✅" if case.get("passed") else "❌"
|
||
title = f"### {badge} 用例 `{case['case_id']}`"
|
||
if case.get("connectivity"):
|
||
title += "（连通用例，未配置判定标准）"
|
||
lines.append(title)
|
||
lines.append("")
|
||
|
||
for turn in case.get("turns", []):
|
||
lines.append(f"**第 {turn['round']} 轮**")
|
||
lines.append("")
|
||
lines.append(f"> **用户**: {turn.get('sent_text', '—')}")
|
||
lines.append("")
|
||
reply = turn.get("reply_text") or "（无回复）"
|
||
lines.append(f"> **智能体**: {reply}")
|
||
if turn.get("latency_ms") is not None:
|
||
lines.append(f"> *延迟: {turn['latency_ms']}ms*")
|
||
lines.append("")
|
||
|
||
if case.get("results"):
|
||
lines.append("**规则评估结果**")
|
||
lines.append("")
|
||
lines.append("| 规则 | 结果 | 评分 | 说明 |")
|
||
lines.append("|------|------|------|------|")
|
||
for r in case["results"]:
|
||
badge = "✅" if r["passed"] else "❌"
|
||
score = f"{r['score']:.2f}" if r.get("score") is not None else "-"
|
||
lines.append(f"| {r['rule_type']} | {badge} | {score} | {r.get('reason', '')} |")
|
||
lines.append("")
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
def render_json_report(run_id: str, session=None) -> str:
|
||
"""Render a report as JSON string."""
|
||
report = generate_report(run_id, session)
|
||
return json.dumps(report, ensure_ascii=False, indent=2)
|
||
|
||
|
||
def render_html_report(run_id: str, session=None) -> str:
|
||
"""Render a report as HTML string."""
|
||
report = generate_report(run_id, session)
|
||
template = Template(HTML_TEMPLATE)
|
||
return template.render(report=report)
|
||
|
||
|
||
def save_report(run_id: str, fmt: str = "html", output_dir: Optional[Path] = None) -> Path:
|
||
"""Generate and save a report to disk."""
|
||
output_dir = output_dir or DATA_DIR / "reports"
|
||
output_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
if fmt == "html":
|
||
content = render_html_report(run_id)
|
||
path = output_dir / f"report_{run_id}_{timestamp}.html"
|
||
elif fmt == "json":
|
||
content = render_json_report(run_id)
|
||
path = output_dir / f"report_{run_id}_{timestamp}.json"
|
||
elif fmt == "markdown":
|
||
content = render_markdown_report(run_id)
|
||
path = output_dir / f"report_{run_id}_{timestamp}.md"
|
||
else:
|
||
raise ValueError(f"unsupported report format: {fmt}")
|
||
|
||
path.write_text(content, encoding="utf-8")
|
||
return path
|