.pouch/skills/tester/scripts/generate_run_csv.py

#!/usr/bin/env python3
"""从 runs/*.yaml 生成给人看的成绩单 CSV。不要手改 CSV。"""

from __future__ import annotations

import argparse
import csv
import re
import sys
from collections import Counter
from pathlib import Path
from typing import Any

try:
    import yaml
except ImportError:  # pragma: no cover
    print("需要 PyYAML: python3 -m pip install pyyaml", file=sys.stderr)
    raise SystemExit(2)

from generate_cases_csv import load_cases, one_line, resolve_root, tester_dir

RESULTS = ("pass", "fail", "environment-fail", "skipped", "not-run")
RUN_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
EXECUTED_YES = frozenset({"pass", "fail", "environment-fail"})
EXECUTED_NO = frozenset({"skipped", "not-run"})
FIELDS = (
    "id",
    "client_suite",
    "platform",
    "result",
    "executed",
    "evidence",
    "notes",
    "category",
    "title",
    "tag",
    "risk",
    "driver",
    "maps_to",
)


def executed_cell(result: str) -> str:
    if result in EXECUTED_YES:
        return "yes"
    if result in EXECUTED_NO:
        return "no"
    return ""


def run_yaml_files(root: Path) -> list[Path]:
    runs = tester_dir(root) / "runs"
    if not runs.is_dir():
        return []
    return sorted(path for path in runs.glob("*.yaml") if path.is_file())


def latest_run_yaml(root: Path) -> Path | None:
    files = run_yaml_files(root)
    return files[-1] if files else None


def new_run_id(scope: str, suffix: str | None = None) -> str:
    """Return a collision-resistant filename/id for a new run snapshot."""
    from datetime import datetime, timezone
    from secrets import token_hex

    clean_scope = re.sub(r"[^A-Za-z0-9._-]+", "-", scope).strip("-") or "run"
    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    clean_suffix = re.sub(r"[^A-Za-z0-9._-]+", "-", suffix or "").strip("-")
    random_part = token_hex(3)
    parts = [stamp, clean_scope]
    if clean_suffix:
        parts.append(clean_suffix)
    parts.append(random_part)
    return "-".join(parts)


def load_run(path: Path) -> dict[str, Any]:
    data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
    if not isinstance(data, dict):
        raise SystemExit(f"{path} 必须是 mapping")
    results = data.get("results")
    if not isinstance(results, list):
        raise SystemExit(f"{path} 缺少 results 列表")
    run_id = data.get("runId")
    if run_id is not None and (
        not isinstance(run_id, str) or RUN_ID_RE.fullmatch(run_id) is None
    ):
        raise SystemExit(f"{path} 的 runId 格式无效")
    return data


def results_by_id(run: dict[str, Any]) -> dict[str, dict[str, Any]]:
    out: dict[str, dict[str, Any]] = {}
    for item in run.get("results") or []:
        if not isinstance(item, dict) or not item.get("id"):
            continue
        out[str(item["id"])] = item
    return out


def case_index(root: Path) -> dict[str, dict[str, Any]]:
    out: dict[str, dict[str, Any]] = {}
    for case in load_cases(root):
        cid = str(case.get("id") or "")
        if cid:
            out[cid] = case
    return out


def export_run_csv(
    root: Path,
    run_path: Path | None = None,
    out: Path | None = None,
) -> Path:
    path = run_path or latest_run_yaml(root)
    if path is None:
        raise SystemExit(f"找不到运行快照: {tester_dir(root) / 'runs'}/*.yaml")
    path = path.resolve()
    run = load_run(path)
    cases = case_index(root)
    dest = out or path.with_suffix(".csv")
    dest.parent.mkdir(parents=True, exist_ok=True)
    with dest.open("w", encoding="utf-8-sig", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=list(FIELDS), extrasaction="ignore")
        writer.writeheader()
        for item in run.get("results") or []:
            if not isinstance(item, dict) or not item.get("id"):
                continue
            cid = str(item["id"])
            result = str(item.get("result") or "")
            case = cases.get(cid) or {}
            writer.writerow(
                {
                    "id": cid,
                    "client_suite": case.get("client_suite") or case.get("suite") or "",
                    "platform": one_line(case.get("platform")),
                    "result": result,
                    "executed": executed_cell(result),
                    "evidence": one_line(item.get("evidence")),
                    "notes": one_line(item.get("notes")),
                    "category": case.get("category") or "",
                    "title": case.get("title") or "",
                    "tag": case.get("tag") or "",
                    "risk": case.get("risk") or "",
                    "driver": case.get("driver") or "",
                    "maps_to": case.get("maps_to") or "",
                }
            )
    return dest


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="从 runs/*.yaml 生成成绩单 CSV")
    parser.add_argument("--root", default=".", help="项目根")
    parser.add_argument("--run", help="运行快照 yaml，默认 runs/ 里文件名最新的一份")
    parser.add_argument("--out", help="输出路径，默认与 yaml 同名 .csv")
    parser.add_argument(
        "--no-refresh-cases",
        action="store_true",
        help="兼容旧命令；cases.csv 已不会被运行结果刷新",
    )
    args = parser.parse_args(argv)
    root = resolve_root(args.root)
    run_path = Path(args.run).resolve() if args.run else None
    out = Path(args.out).resolve() if args.out else None
    path = export_run_csv(
        root,
        run_path=run_path,
        out=out,
    )
    counts: Counter[str] = Counter()
    with path.open(encoding="utf-8-sig", newline="") as handle:
        rows = list(csv.DictReader(handle))
    for row in rows:
        counts[row.get("result") or ""] += 1
    bits = " ".join(f"{name}={counts.get(name, 0)}" for name in RESULTS)
    print(f"wrote {path}  rows={len(rows)}  {bits}")
    return 0


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