132 lines
4.8 KiB
Python
132 lines
4.8 KiB
Python
"""合并多份评测 JSON，生成统一的五维评分视图。"""
|
||
from __future__ import annotations
|
||
|
||
import datetime as dt
|
||
import json
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from .models import DEFAULT_DIMENSION_WEIGHTS, EvalConfigError
|
||
from .report import FRAMEWORK_INFO, SECURITY_CAP
|
||
|
||
|
||
def _load_report(path: Path) -> dict[str, Any]:
|
||
try:
|
||
report = json.loads(path.read_text(encoding="utf-8"))
|
||
except json.JSONDecodeError as exc:
|
||
raise EvalConfigError(f"报告不是有效 JSON: {path}: {exc}") from exc
|
||
if not isinstance(report, dict):
|
||
raise EvalConfigError(f"报告顶层必须是 JSON object: {path}")
|
||
if not isinstance(report.get("dimensions"), dict):
|
||
raise EvalConfigError(f"报告缺少 dimensions: {path}")
|
||
if not isinstance(report.get("suite"), dict):
|
||
raise EvalConfigError(f"报告缺少 suite: {path}")
|
||
return report
|
||
|
||
|
||
def combine_reports(paths: list[Path]) -> dict[str, Any]:
|
||
"""按用例数加权合并同维度得分，并保留每份报告的来源信息。"""
|
||
if not paths:
|
||
raise EvalConfigError("至少需要一份 --report")
|
||
|
||
loaded = [(path.resolve(), _load_report(path)) for path in paths]
|
||
dimensions: dict[str, dict[str, Any]] = {}
|
||
for name, weight in DEFAULT_DIMENSION_WEIGHTS.items():
|
||
values: list[tuple[float, int]] = []
|
||
total_cases = 0
|
||
for _, report in loaded:
|
||
item = report["dimensions"].get(name)
|
||
if not isinstance(item, dict) or item.get("score") is None:
|
||
continue
|
||
case_count = int(item.get("case_count", 0))
|
||
values.append((float(item["score"]), max(case_count, 1)))
|
||
total_cases += max(case_count, 0)
|
||
denominator = sum(case_weight for _, case_weight in values)
|
||
score = (
|
||
sum(value * case_weight for value, case_weight in values)
|
||
/ denominator
|
||
if denominator
|
||
else None
|
||
)
|
||
dimensions[name] = {
|
||
"weight": weight,
|
||
"score": round(score, 2) if score is not None else None,
|
||
"case_count": total_cases,
|
||
}
|
||
|
||
present = [item for item in dimensions.values() if item["score"] is not None]
|
||
present_weight = sum(item["weight"] for item in present)
|
||
provisional = (
|
||
sum(item["score"] * item["weight"] for item in present) / present_weight
|
||
if present_weight
|
||
else None
|
||
)
|
||
complete = all(item["score"] is not None for item in dimensions.values())
|
||
total = provisional if complete else None
|
||
|
||
security_failures: list[dict[str, Any]] = []
|
||
cases: list[dict[str, Any]] = []
|
||
sources: list[dict[str, Any]] = []
|
||
for path, report in loaded:
|
||
suite = report["suite"]
|
||
source_label = str(suite.get("name") or path.stem)
|
||
sources.append(
|
||
{
|
||
"path": str(path),
|
||
"suite": source_label,
|
||
"version": str(suite.get("version", "")),
|
||
"generated_at": report.get("generated_at"),
|
||
}
|
||
)
|
||
for case in report.get("cases", []):
|
||
if isinstance(case, dict):
|
||
cases.append({**case, "source_report": source_label})
|
||
cap = report.get("security_cap")
|
||
if isinstance(cap, dict):
|
||
for failure in cap.get("failures", []):
|
||
if isinstance(failure, dict):
|
||
security_failures.append(
|
||
{**failure, "source_report": source_label}
|
||
)
|
||
|
||
cap_applied = bool(
|
||
security_failures and total is not None and total > SECURITY_CAP
|
||
)
|
||
if cap_applied:
|
||
total = SECURITY_CAP
|
||
|
||
first_environment = loaded[0][1].get("environment")
|
||
environment = (
|
||
dict(first_environment) if isinstance(first_environment, dict) else {}
|
||
)
|
||
return {
|
||
"schema_version": 1,
|
||
"suite": {
|
||
"name": "zcbot-unified",
|
||
"version": "+".join(
|
||
str(report["suite"].get("version", ""))
|
||
for _, report in loaded
|
||
),
|
||
},
|
||
"generated_at": dt.datetime.now(dt.timezone.utc).isoformat(),
|
||
"environment": environment,
|
||
"framework": FRAMEWORK_INFO,
|
||
"sources": sources,
|
||
"merge_method": "同维度按报告中的 case_count 加权；五维按固定权重汇总",
|
||
"dimensions": dimensions,
|
||
"pre_cap_score": (
|
||
round(provisional, 2) if provisional is not None else None
|
||
),
|
||
"total_score": round(total, 2) if total is not None else None,
|
||
"provisional_score": (
|
||
round(provisional, 2) if provisional is not None else None
|
||
),
|
||
"is_complete": complete,
|
||
"security_cap": {
|
||
"threshold": SECURITY_CAP,
|
||
"applied": cap_applied,
|
||
"failures": security_failures,
|
||
},
|
||
"cases": cases,
|
||
}
|