276 行
8.9 KiB
Python
可执行文件
276 行
8.9 KiB
Python
可执行文件
#!/usr/bin/env python3
|
|
"""Generate Change Failure Rate reports from regression JSON artifacts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import re
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from typing import Iterable
|
|
|
|
WINDOW_PATTERN = re.compile(r"^(\d+)d$")
|
|
|
|
|
|
@dataclass
|
|
class RegressionRun:
|
|
timestamp: datetime
|
|
commit_sha: str
|
|
failed: bool
|
|
report_file: Path
|
|
|
|
|
|
@dataclass
|
|
class HotfixEvent:
|
|
commit_sha: str
|
|
occurred_at: datetime
|
|
|
|
|
|
def parse_iso_datetime(value: str) -> datetime:
|
|
normalized = value.strip()
|
|
if normalized.endswith("Z"):
|
|
normalized = normalized[:-1] + "+00:00"
|
|
parsed = datetime.fromisoformat(normalized)
|
|
if parsed.tzinfo is None:
|
|
parsed = parsed.replace(tzinfo=timezone.utc)
|
|
return parsed.astimezone(timezone.utc)
|
|
|
|
|
|
def parse_windows(value: str) -> list[int]:
|
|
windows: list[int] = []
|
|
for chunk in value.split(","):
|
|
item = chunk.strip()
|
|
if not item:
|
|
continue
|
|
matched = WINDOW_PATTERN.match(item)
|
|
if not matched:
|
|
raise ValueError(f"invalid window '{item}', expected like 7d/30d/90d")
|
|
days = int(matched.group(1))
|
|
if days <= 0:
|
|
raise ValueError(f"window days must be >0: {item}")
|
|
windows.append(days)
|
|
if not windows:
|
|
windows.append(30)
|
|
return sorted(set(windows))
|
|
|
|
|
|
def load_regression_runs(input_dir: Path) -> list[RegressionRun]:
|
|
runs: list[RegressionRun] = []
|
|
|
|
for path in sorted(input_dir.rglob("*.json")):
|
|
try:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
continue
|
|
|
|
schema_version = str(payload.get("schema_version", "")).strip()
|
|
if schema_version != "1.0":
|
|
continue
|
|
|
|
generated_at_raw = payload.get("generated_at")
|
|
overall = payload.get("overall") if isinstance(payload.get("overall"), dict) else {}
|
|
git = payload.get("git") if isinstance(payload.get("git"), dict) else {}
|
|
|
|
if not generated_at_raw or not isinstance(generated_at_raw, str):
|
|
continue
|
|
|
|
commit_sha = str(git.get("sha", "unknown")).strip() or "unknown"
|
|
status = str(overall.get("status", "")).strip().lower()
|
|
|
|
try:
|
|
generated_at = parse_iso_datetime(generated_at_raw)
|
|
except Exception:
|
|
continue
|
|
|
|
runs.append(
|
|
RegressionRun(
|
|
timestamp=generated_at,
|
|
commit_sha=commit_sha,
|
|
failed=status != "passed",
|
|
report_file=path,
|
|
)
|
|
)
|
|
|
|
return sorted(runs, key=lambda x: x.timestamp)
|
|
|
|
|
|
def load_hotfix_events(path: Path | None) -> list[HotfixEvent]:
|
|
if path is None:
|
|
return []
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"hotfix csv not found: {path}")
|
|
|
|
events: list[HotfixEvent] = []
|
|
with path.open(encoding="utf-8", newline="") as fp:
|
|
reader = csv.DictReader(fp)
|
|
for row in reader:
|
|
commit_sha = (row.get("commit_sha") or "").strip()
|
|
occurred_at_raw = (row.get("occurred_at") or "").strip()
|
|
if not commit_sha or not occurred_at_raw:
|
|
continue
|
|
try:
|
|
occurred_at = parse_iso_datetime(occurred_at_raw)
|
|
except Exception:
|
|
continue
|
|
events.append(HotfixEvent(commit_sha=commit_sha, occurred_at=occurred_at))
|
|
|
|
return sorted(events, key=lambda x: x.occurred_at)
|
|
|
|
|
|
def ratio(numerator: int, denominator: int) -> float:
|
|
if denominator == 0:
|
|
return 0.0
|
|
return numerator / denominator
|
|
|
|
|
|
def has_hotfix_hit_within_24h(run_times: Iterable[datetime], events: list[HotfixEvent]) -> bool:
|
|
event_times = [event.occurred_at for event in events]
|
|
for run_time in run_times:
|
|
for event_time in event_times:
|
|
seconds = (event_time - run_time).total_seconds()
|
|
if 0 <= seconds <= 24 * 3600:
|
|
return True
|
|
return False
|
|
|
|
|
|
def build_window_metrics(
|
|
runs: list[RegressionRun],
|
|
hotfix_events: list[HotfixEvent],
|
|
now: datetime,
|
|
window_days: int,
|
|
) -> dict[str, object]:
|
|
cutoff = now - timedelta(days=window_days)
|
|
window_runs = [run for run in runs if run.timestamp >= cutoff]
|
|
failed_runs = [run for run in window_runs if run.failed]
|
|
|
|
commits: dict[str, list[RegressionRun]] = {}
|
|
for run in window_runs:
|
|
commits.setdefault(run.commit_sha, []).append(run)
|
|
|
|
hotfix_by_commit: dict[str, list[HotfixEvent]] = {}
|
|
for event in hotfix_events:
|
|
if event.occurred_at >= cutoff:
|
|
hotfix_by_commit.setdefault(event.commit_sha, []).append(event)
|
|
|
|
failed_commit_count = 0
|
|
for commit_sha, commit_runs in commits.items():
|
|
run_failed = any(run.failed for run in commit_runs)
|
|
hotfix_hit = has_hotfix_hit_within_24h(
|
|
(run.timestamp for run in commit_runs),
|
|
hotfix_by_commit.get(commit_sha, []),
|
|
)
|
|
if run_failed or hotfix_hit:
|
|
failed_commit_count += 1
|
|
|
|
total_runs = len(window_runs)
|
|
total_commits = len(commits)
|
|
failed_runs_count = len(failed_runs)
|
|
|
|
run_cfr = ratio(failed_runs_count, total_runs)
|
|
commit_cfr = ratio(failed_commit_count, total_commits)
|
|
|
|
return {
|
|
"window": f"{window_days}d",
|
|
"cutoff": cutoff.isoformat().replace("+00:00", "Z"),
|
|
"total_runs": total_runs,
|
|
"failed_runs": failed_runs_count,
|
|
"run_cfr": round(run_cfr, 6),
|
|
"run_cfr_pct": round(run_cfr * 100, 2),
|
|
"total_commits": total_commits,
|
|
"failed_commits": failed_commit_count,
|
|
"commit_cfr": round(commit_cfr, 6),
|
|
"commit_cfr_pct": round(commit_cfr * 100, 2),
|
|
}
|
|
|
|
|
|
def render_markdown(result: dict[str, object]) -> str:
|
|
lines = [
|
|
"# Change Failure Rate 报告",
|
|
"",
|
|
f"- 生成时间: {result['generated_at']}",
|
|
f"- 输入目录: `{result['input_dir']}`",
|
|
f"- 报告总数: {result['reports_scanned']}",
|
|
"",
|
|
"| Window | Total Runs | Failed Runs | Run CFR | Total Commits | Failed Commits | Commit CFR |",
|
|
"| --- | ---: | ---: | ---: | ---: | ---: | ---: |",
|
|
]
|
|
|
|
windows: dict[str, dict[str, object]] = result["windows"] # type: ignore[assignment]
|
|
for key in sorted(windows.keys(), key=lambda x: int(x[:-1])):
|
|
item = windows[key]
|
|
lines.append(
|
|
f"| {key} | {item['total_runs']} | {item['failed_runs']} | {item['run_cfr_pct']}% | "
|
|
f"{item['total_commits']} | {item['failed_commits']} | {item['commit_cfr_pct']}% |"
|
|
)
|
|
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Generate Change Failure Rate report")
|
|
parser.add_argument("--input-dir", required=True, help="Directory that stores regression JSON artifacts")
|
|
parser.add_argument("--window", default="30d", help="Comma separated windows, e.g. 7d,30d,90d")
|
|
parser.add_argument("--output-json", help="Optional output JSON file path")
|
|
parser.add_argument("--output-md", help="Optional output markdown file path")
|
|
parser.add_argument("--hotfix-csv", help="Optional hotfix CSV path (columns: commit_sha,occurred_at)")
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
input_dir = Path(args.input_dir).resolve()
|
|
if not input_dir.exists():
|
|
print(f"[ERR] input dir not found: {input_dir}", file=sys.stderr)
|
|
return 1
|
|
|
|
try:
|
|
windows = parse_windows(args.window)
|
|
except ValueError as exc:
|
|
print(f"[ERR] {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
try:
|
|
hotfix_events = load_hotfix_events(Path(args.hotfix_csv).resolve() if args.hotfix_csv else None)
|
|
except Exception as exc:
|
|
print(f"[ERR] {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
runs = load_regression_runs(input_dir)
|
|
now = datetime.now(timezone.utc)
|
|
|
|
result: dict[str, object] = {
|
|
"generated_at": now.isoformat().replace("+00:00", "Z"),
|
|
"input_dir": str(input_dir),
|
|
"reports_scanned": len(runs),
|
|
"hotfix_events": len(hotfix_events),
|
|
"windows": {},
|
|
}
|
|
|
|
for days in windows:
|
|
result["windows"][f"{days}d"] = build_window_metrics(runs, hotfix_events, now, days) # type: ignore[index]
|
|
|
|
rendered_json = json.dumps(result, ensure_ascii=False, indent=2) + "\n"
|
|
|
|
if args.output_json:
|
|
output_json = Path(args.output_json)
|
|
output_json.parent.mkdir(parents=True, exist_ok=True)
|
|
output_json.write_text(rendered_json, encoding="utf-8")
|
|
|
|
if args.output_md:
|
|
output_md = Path(args.output_md)
|
|
output_md.parent.mkdir(parents=True, exist_ok=True)
|
|
output_md.write_text(render_markdown(result), encoding="utf-8")
|
|
|
|
if not args.output_json and not args.output_md:
|
|
print(rendered_json, end="")
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|