zcbot/evaluation/__main__.py

"""命令行入口：`.venv/Scripts/python.exe -m evaluation ...`。"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

from .audit import audit_repository, audit_security_repository
from .client import EvalClientError, ZcbotClient
from .combine import combine_reports
from .inspect_bridge import export_inspect_jsonl
from .models import EvalConfigError
from .report import summarize, write_reports
from .runner import RunSettings, run_suite
from .suite import load_suite


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_SUITE = ROOT / "evaluation" / "datasets" / "smoke.json"
DEFAULT_CONFIG = ROOT / "evaluation" / "config.example.json"


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="zcbot 黑盒技术评测")
    sub = parser.add_subparsers(dest="command", required=True)
    validate = sub.add_parser("validate", help="只校验任务集，不调用 zcbot")
    validate.add_argument("--suite", type=Path, default=DEFAULT_SUITE)

    audit = sub.add_parser("audit", help="运行本地工程质量审计并生成报告")
    audit.add_argument("--suite", type=Path, default=DEFAULT_SUITE)
    audit.add_argument("--output", type=Path, default=ROOT / "evaluation" / "reports")
    audit.add_argument("--timeout-s", type=float, default=900)

    inspect_export = sub.add_parser(
        "export-inspect", help="把 suite 导出为 Inspect AI JSONL"
    )
    inspect_export.add_argument("--suite", type=Path, default=DEFAULT_SUITE)
    inspect_export.add_argument("--output", type=Path, required=True)

    combine = sub.add_parser("combine", help="合并已有 JSON 报告")
    combine.add_argument(
        "--report",
        type=Path,
        action="append",
        required=True,
        help="输入报告 JSON；可重复传入",
    )
    combine.add_argument(
        "--output",
        type=Path,
        default=ROOT / "evaluation" / "reports" / "latest",
    )

    run = sub.add_parser("run", help="执行任务集并生成 JSON/Markdown 报告")
    run.add_argument("--suite", type=Path, default=DEFAULT_SUITE)
    run.add_argument("--config", type=Path, default=DEFAULT_CONFIG)
    run.add_argument("--output", type=Path, default=ROOT / "evaluation" / "reports")
    run.add_argument("--repetitions", type=int)
    run.add_argument("--timeout-s", type=float, default=900)
    run.add_argument("--allow-remote", action="store_true")
    run.add_argument(
        "--with-audit",
        action="store_true",
        help="在线任务结束后同时执行本地工程审计，补齐 engineering 维度",
    )
    run.add_argument(
        "--execute",
        action="store_true",
        help="确认实际调用模型并产生费用；缺少此参数时拒绝执行",
    )
    return parser


def main(argv: list[str] | None = None) -> int:
    args = _parser().parse_args(argv)
    try:
        if args.command == "combine":
            summary = combine_reports(args.report)
            json_path, md_path = write_reports(summary, args.output)
            print(f"[OK] provisional_score={summary['provisional_score']}")
            print(f"[OK] total_score={summary['total_score']}")
            print(f"[OK] json={json_path}")
            print(f"[OK] markdown={md_path}")
            return 0
        suite = load_suite(args.suite)
        if args.command == "validate":
            print(
                f"[OK] suite={suite.name} version={suite.version} "
                f"cases={len(suite.cases)}"
            )
            return 0
        if args.command == "audit":
            engineering = audit_repository(
                ROOT, timeout_s=args.timeout_s, progress=print
            )
            security = audit_security_repository(
                ROOT, timeout_s=args.timeout_s, progress=print
            )
            summary = summarize(suite, [engineering, security], repo_root=ROOT)
            json_path, md_path = write_reports(
                summary, args.output, stem="engineering-audit"
            )
            print(f"[OK] engineering_score={engineering.score * 100:.2f}")
            print(f"[OK] security_score={security.score * 100:.2f}")
            print(f"[OK] json={json_path}")
            print(f"[OK] markdown={md_path}")
            return 0
        if args.command == "export-inspect":
            count = export_inspect_jsonl(suite, args.output)
            print(f"[OK] inspect_jsonl={args.output} samples={count}")
            return 0
        if not args.execute:
            print("[ERR] run 需要显式传 --execute（会调用模型并产生费用）")
            return 2
        config = json.loads(args.config.read_text(encoding="utf-8"))
        if not isinstance(config, dict):
            raise EvalConfigError("config 顶层必须是 JSON object")
        with ZcbotClient.from_config(
            config, allow_remote=args.allow_remote
        ) as client:
            results = run_suite(
                suite,
                client,
                settings=RunSettings(
                    repetitions=args.repetitions,
                    timeout_s=args.timeout_s,
                ),
                progress=print,
            )
        if args.with_audit:
            results.extend(
                [
                    audit_repository(
                        ROOT, timeout_s=args.timeout_s, progress=print
                    ),
                    audit_security_repository(
                        ROOT, timeout_s=args.timeout_s, progress=print
                    ),
                ]
            )
        summary = summarize(suite, results, repo_root=ROOT)
        json_path, md_path = write_reports(summary, args.output)
        print(f"[OK] json={json_path}")
        print(f"[OK] markdown={md_path}")
        return 0
    except (
        EvalConfigError,
        EvalClientError,
        OSError,
        ValueError,
        json.JSONDecodeError,
    ) as exc:
        print(f"[ERR] {exc}")
        return 2


if __name__ == "__main__":
    sys.exit(main())