baodan/scripts/tools/evaluate_plan_goldens.py

#!/usr/bin/env python3
"""批量评估计划书解析金标准，输出 JSON 与 Markdown 报告。"""
from __future__ import annotations

import argparse
import json
from dataclasses import asdict, dataclass
from decimal import Decimal, InvalidOperation
from pathlib import Path


@dataclass
class EvaluationReport:
    status: str
    sample_count: int
    minimum_samples: int
    scalar_accuracy: float
    cell_accuracy: float
    target_row_recall: float
    evidence_coverage: float
    differences: list[dict]


def evaluate(manifest_path: Path, predictions_dir: Path) -> EvaluationReport:
    manifest = _read_json(manifest_path)
    samples = manifest.get("samples") or []
    minimum = int(manifest.get("minimumDevelopmentSamples") or 10)
    scalar_total = scalar_correct = 0
    cell_total = cell_correct = 0
    row_total = row_recalled = 0
    evidence_total = evidence_correct = 0
    differences: list[dict] = []

    for sample in samples:
        sample_id = str(sample.get("id") or "")
        label_path = manifest_path.parent / str(sample.get("label") or "")
        prediction_path = predictions_dir / f"{sample_id}.json"
        if not sample_id or not label_path.is_file() or not prediction_path.is_file():
            differences.append({"sampleId": sample_id, "type": "missing_file"})
            continue
        expected = _read_json(label_path)
        actual = _read_json(prediction_path)

        expected_scalars = expected.get("scalars") or {}
        actual_scalars = actual.get("scalars") or {}
        for field_path, expected_field in expected_scalars.items():
            scalar_total += 1
            actual_field = actual_scalars.get(field_path)
            if _field_equal(expected_field, actual_field):
                scalar_correct += 1
            else:
                differences.append({"sampleId": sample_id, "type": "scalar", "path": field_path})
            if not bool((expected_field or {}).get("allowMissing")):
                evidence_total += 1
                if _has_evidence(actual_field):
                    evidence_correct += 1

        expected_rows = _rows_by_key(expected.get("benefitRows") or [])
        actual_rows = _rows_by_key(actual.get("benefitRows") or [])
        row_total += len(expected_rows)
        row_recalled += len(set(expected_rows).intersection(actual_rows))
        for row_key, expected_row in expected_rows.items():
            actual_row = actual_rows.get(row_key) or {}
            for field_name, expected_field in (expected_row.get("values") or {}).items():
                cell_total += 1
                actual_field = (actual_row.get("values") or {}).get(field_name)
                if _field_equal(expected_field, actual_field):
                    cell_correct += 1
                else:
                    differences.append({
                        "sampleId": sample_id,
                        "type": "cell",
                        "path": f"{row_key}.{field_name}",
                    })

    status = "pass" if len(samples) >= minimum and not differences else "blocked"
    return EvaluationReport(
        status=status,
        sample_count=len(samples),
        minimum_samples=minimum,
        scalar_accuracy=_ratio(scalar_correct, scalar_total),
        cell_accuracy=_ratio(cell_correct, cell_total),
        target_row_recall=_ratio(row_recalled, row_total),
        evidence_coverage=_ratio(evidence_correct, evidence_total),
        differences=differences,
    )


def write_report(report: EvaluationReport, output_dir: Path) -> None:
    output_dir.mkdir(parents=True, exist_ok=True)
    payload = asdict(report)
    (output_dir / "plan-golden-report.json").write_text(
        json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    lines = [
        "# 计划书金标准评测报告",
        "",
        f"- 状态：`{report.status}`",
        f"- 样本：{report.sample_count}/{report.minimum_samples}",
        f"- 标量精确率：{report.scalar_accuracy:.2%}",
        f"- 金额单元格精确率：{report.cell_accuracy:.2%}",
        f"- 目标行召回率：{report.target_row_recall:.2%}",
        f"- 证据覆盖率：{report.evidence_coverage:.2%}",
        f"- 差异数：{len(report.differences)}",
    ]
    (output_dir / "plan-golden-report.md").write_text("\n".join(lines) + "\n", encoding="utf-8")


def _rows_by_key(rows: list[dict]) -> dict[str, dict]:
    result = {}
    for row in rows:
        key = "|".join(str(row.get(name) or "") for name in (
            "tableId", "scenarioType", "policyYear", "rowVariant"
        ))
        if key.strip("|"):
            result[key] = row
    return result


def _field_equal(expected, actual) -> bool:
    if not isinstance(expected, dict) or not isinstance(actual, dict):
        return False
    return (
        _normalized_value(expected.get("value")) == _normalized_value(actual.get("value"))
        and expected.get("currency") == actual.get("currency")
        and expected.get("unit") == actual.get("unit")
    )


def _normalized_value(value):
    if isinstance(value, bool) or value is None:
        return value
    try:
        return Decimal(str(value))
    except (InvalidOperation, ValueError):
        return str(value).strip()


def _has_evidence(field) -> bool:
    if not isinstance(field, dict):
        return False
    return any(
        isinstance(item, dict) and item.get("pageNumber") and item.get("bbox")
        for item in field.get("evidence") or []
    )


def _ratio(numerator: int, denominator: int) -> float:
    return numerator / denominator if denominator else 0.0


def _read_json(path: Path) -> dict:
    return json.loads(path.read_text(encoding="utf-8"))


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--manifest", type=Path, default=Path("tests/fixtures/plan_goldens/manifest.json"))
    parser.add_argument("--predictions", type=Path, required=True)
    parser.add_argument("--output", type=Path, default=Path("tmp/plan-golden-report"))
    args = parser.parse_args()
    report = evaluate(args.manifest, args.predictions)
    write_report(report, args.output)
    return 0 if report.status == "pass" else 2


if __name__ == "__main__":
    raise SystemExit(main())