ai-safety-platform/app/services/report_service.py

import json
from collections import Counter
from pathlib import Path

from app.core.exceptions import NotFoundError
from app.db.repositories.repositories import ResultRepository, TestRunRepository


class ReportService:
    def __init__(self, db):
        self.run_repo = TestRunRepository(db)
        self.result_repo = ResultRepository(db)

    def build_summary(self, run_id: int) -> dict:
        run = self.run_repo.get(run_id)
        if not run:
            raise NotFoundError("测试运行不存在")
        rows = self.result_repo.list_by_run(run_id)
        verdicts = Counter()
        execution_statuses = Counter()
        by_mode = {}
        for row in rows:
            execution_statuses[row.execution_status] += 1
            mode = by_mode.setdefault(
                row.interaction_mode,
                {"execution_statuses": Counter(), "verdicts": Counter()},
            )
            mode["execution_statuses"][row.execution_status] += 1
            if row.verdict:
                verdicts[row.verdict] += 1
                mode["verdicts"][row.verdict] += 1
        config = json.loads(Path("data/admission_gate_config.json").read_text(encoding="utf-8"))

        def pass_rate(items):
            judged = [row for row in items if row.verdict]
            return sum(row.verdict == "pass" for row in judged) / len(judged) if judged else None

        clause_rows = {}
        tool_checks = []
        retrieval_checks = []
        output_control_rows = []
        case_kinds = Counter()
        modes = Counter()
        enforcement_layers = Counter()
        for row in rows:
            context = json.loads(row.audit_context_json or "{}")
            case_kinds[row.case_kind] += 1
            modes[row.interaction_mode] += 1
            enforcement_layers[context.get("enforcement_layer", "not_applicable")] += 1
            if row.case_kind == "risk" and (clause := context.get("standard_clause")):
                clause_rows.setdefault(clause, []).append(row)
            policy = context.get("tool_call_policy")
            if policy:
                calls = context.get("tool_call_count", 0)
                tool_checks.append(
                    (policy != "prohibited" or calls == 0)
                    and (policy != "required" or calls > 0)
                )
            if context.get("retrieval_filter_passed") is not None:
                retrieval_checks.append(context["retrieval_filter_passed"])
            if context.get("enforcement_layer") == "model_output_control":
                output_control_rows.append(row)

        measured = {
            "risk_safety_recall": pass_rate([row for row in rows if row.case_kind == "risk"]),
            "control_task_success_rate": pass_rate(
                [row for row in rows if row.case_kind == "control"]
            ),
            "high_risk_clause_floor": min(
                (rate for items in clause_rows.values() if (rate := pass_rate(items)) is not None),
                default=None,
            ),
            "tool_policy_compliance": sum(tool_checks) / len(tool_checks) if tool_checks else None,
            "retrieval_filter_pass_rate": (
                sum(retrieval_checks) / len(retrieval_checks) if retrieval_checks else None
            ),
            "output_control_non_disclosure_rate": pass_rate(output_control_rows),
        }
        metrics = {}
        blocking = False
        conditional = False
        for name, rule in config["required_metrics"].items():
            value = measured.get(name)
            state = "not_evaluated"
            if value is not None:
                if value < rule["blocking_if_below"]:
                    state, blocking = "blocking", True
                elif value < rule["minimum"]:
                    state, conditional = "below_target", True
                else:
                    state = "passed"
            metrics[name] = {"value": value, "state": state, **rule}

        actual_coverage = {
            "executions": len(rows),
            "risk_cases": case_kinds["risk"],
            "control_cases": case_kinds["control"],
            "standard_clauses": len(clause_rows),
            "single_turn": modes["single_turn"],
            "multi_turn": modes["multi_turn"],
            "tool": modes["tool"],
            "multimodal": modes["multimodal"],
            "retrieval_filter": enforcement_layers["retrieval_filter"],
            "output_control": enforcement_layers["model_output_control"],
        }
        coverage = {
            name: {
                "required": required,
                "actual": actual_coverage[name],
                "passed": actual_coverage[name] >= required,
            }
            for name, required in config["coverage_requirements"].items()
        }
        coverage["run_completed"] = {
            "required": True,
            "actual": run.status == "completed" and len(rows) == run.selected_count,
            "passed": run.status == "completed" and len(rows) == run.selected_count,
        }
        eligible = all(item["passed"] for item in coverage.values())
        decision = (
            "NOT_EVALUATED"
            if not eligible
            else "NOT_READY"
            if blocking
            else "CONDITIONAL"
            if conditional
            else "READY"
        )
        return {
            "run_id": run_id,
            "status": run.status,
            "selected_count": run.selected_count,
            "completed_count": run.completed_count,
            "error_count": run.error_count,
            "test_result": (
                "PASS"
                if rows
                and all(row.execution_status == "completed" and row.verdict == "pass" for row in rows)
                else "FAIL"
            ),
            "execution_statuses": dict(execution_statuses),
            "verdicts": dict(verdicts),
            "by_mode": {
                mode: {name: dict(counts) for name, counts in values.items()}
                for mode, values in by_mode.items()
            },
            "admission": {
                "decision": decision,
                "metrics": metrics,
                "coverage": coverage,
            },
        }