#!/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 from journey import suite as case_suite, test_type 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", "test_type", "suite", "client_suite", "platform", "result", "executed", "evidence", "notes", "category", "title", "tag", "risk", "driver", "maps_to", "blocked_by", "issue_id", "issue_type", "expected", "observed", "artifact_hash", "environment", "evidence_path", "reproduction", "next_step", ) 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 issue_index(run: dict[str, Any]) -> dict[str, dict[str, Any]]: """Index canonical defects and compatibility test_issues by testcase.""" indexed: dict[str, dict[str, Any]] = {} raw = run.get("defects") if not isinstance(raw, list): raw = run.get("test_issues") if not isinstance(raw, list): return indexed for issue in raw: if not isinstance(issue, dict): continue testcase = str(issue.get("testcase") or issue.get("id") or "") if testcase: indexed.setdefault(testcase, issue) return indexed 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) issues = issue_index(run) 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 {} issue = dict(issues.get(cid, {})) if isinstance(item.get("issue"), dict): issue.update(item["issue"]) client_suite_value = case.get("client_suite") or ( case.get("suite") if test_type(case) == "journey" else "" ) writer.writerow( { "id": cid, "test_type": test_type(case), "suite": case_suite(case), "client_suite": client_suite_value 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 "", "blocked_by": one_line(item.get("blocked_by")), "issue_id": issue.get("id") or item.get("issue_id") or "", "issue_type": issue.get("issue_type") or item.get("issue_type") or "", "expected": one_line(issue.get("expected") or item.get("expected")), "observed": one_line(issue.get("observed") or item.get("observed")), "artifact_hash": issue.get("artifact_hash") or item.get("artifact_hash") or "", "environment": one_line(issue.get("environment") or item.get("environment")), "evidence_path": one_line(issue.get("evidence_path") or item.get("evidence_path")), "reproduction": one_line(issue.get("reproduction") or item.get("reproduction")), "next_step": one_line(issue.get("next_step") or item.get("next_step")), } ) 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())