"""生成 HTML 可视化报告""" import json import logging import os from . import config logger = logging.getLogger(__name__) def generate_report(report: dict) -> str: """从报告数据生成 HTML 报告文件，返回输出路径""" html = build_html(report) os.makedirs(config.REPORT_DIR, exist_ok=True) output_path = os.path.join(config.REPORT_DIR, "report.html") with open(output_path, "w", encoding="utf-8") as f: f.write(html) logger.info("报告已生成: %s", output_path) return output_path def _diff_color(difficulty: str) -> str: """难度对应的颜色""" return { "简单": "#22c55e", "中等": "#f59e0b", "耗时": "#ef4444", "压力": "#8b5cf6", }.get(difficulty, "#6b7280") def _stats_rows(results: list) -> str: """构建详细数据表格行 HTML""" rows = [] for r in results: stats = r.get("statistics", {}) if not stats: continue rows.append( f'' f'{r["case_name"]}' f'{r["difficulty"]}' f'{r["status"]}' f'{stats.get("mean_ms", "N/A")}' f'{stats.get("std_ms", "N/A")}' f'{stats.get("min_ms", "N/A")}' f'{stats.get("max_ms", "N/A")}' f'{stats.get("p95_ms", "N/A")}' f'' ) return "\n".join(rows) def _chart_data(results: list) -> tuple: """提取图表数据：(labels, means, p95s)""" labels, means, p95s = [], [], [] for r in results: stats = r.get("statistics", {}) if stats: labels.append(r["case_name"]) means.append(stats.get("mean_ms", 0)) p95s.append(stats.get("p95_ms", 0)) return labels, means, p95s def _performance_analysis(results: list) -> str: """性能分析摘要""" _, means, _ = _chart_data(results) if len(means) < 2: return "" best_idx = means.index(min(means)) worst_idx = means.index(max(means)) ratio = (means[worst_idx] / means[best_idx]) if means[best_idx] > 0 else "N/A" avg = sum(means) / len(means) return ( f'
' f'

性能分析

' f'
' ) def _concurrency_summary(results: list) -> str: """提取并发压力测试摘要（case_08），无则返回空串""" for r in results: raw = r.get("raw_data", {}) or {} if r.get("case_id") == "case_08_concurrency" and raw: def g(key, suffix=""): val = raw.get(key) return "N/A" if val is None else f"{val}{suffix}" keep = raw.get("tps_keep_rate_pct", 0) keep_color = "#22c55e" if keep >= 60 else ("#f59e0b" if keep > 0 else "#ef4444") err = raw.get("error_rate_pct", 0) err_color = "#22c55e" if err == 0 else "#ef4444" return ( f'
' f'

⚡ 并发压力测试摘要

' f'
' f'
并发数
{g("concurrency")}
' f'
基准 TPS
{g("baseline_tps")}
' f'
并发 TPS
{g("concurrent_tps")}
' f'
TPS 保持率
{g("tps_keep_rate_pct", "%")}
' f'
P95 延迟
{g("e2e_p95_ms", "ms")}
' f'
P99 延迟
{g("e2e_p99_ms", "ms")}
' f'
错误率
{g("error_rate_pct", "%")}
' f'
' ) return "" def build_html(report: dict) -> str: """构建 HTML 报告""" summary = report.get("summary", {}) results = report.get("results", []) env = report.get("environment", {}) config_ = report.get("config", {}) timestamp = report.get("timestamp", "") run_id = report.get("run_id", "") case_names, case_means, case_p95s = _chart_data(results) return f""" 大模型速度测试报告 - {run_id}

大模型速度测试报告

Run ID: {run_id} | 时间: {timestamp} | 用例: {config_.get('total_cases', 0)} | 重复: {config_.get('repeats', 0)}次
模型: {env.get('model', 'N/A')} | 平台: {env.get('platform', 'N/A')}
总用例数
{summary.get('total_cases', 0)}
通过
{summary.get('passed', 0)}
失败
{summary.get('failed', 0)}
总耗时
{summary.get('total_elapsed_seconds', 0):.2f}s
{_concurrency_summary(results)} {_performance_analysis(results)}

性能对比柱状图

P95延迟对比

详细数据

{_stats_rows(results)}
用例名称难度状态平均延迟 (ms) 标准差最小最大P95

环境信息

模型{env.get('model', 'N/A')}
运行平台{env.get('platform', 'N/A')}
运行ID{run_id}
测试时间{timestamp}
"""