172 lines
6.5 KiB
Python
172 lines
6.5 KiB
Python
#!/usr/bin/env python3
|
||
"""le-app-design 验收门：从 calls.jsonl 生成本次运行的集成证据报告。
|
||
|
||
用法:
|
||
python3 evidence_report.py --run-dir <RUN> [--out <RUN>/INTEGRATION-REPORT.md] \
|
||
[--artifact <路径>]... [--title "任务名"] \
|
||
[--require "dep[:note前缀],..."]
|
||
|
||
验收规则（硬门）:
|
||
默认要求三个上游 ui-ux-pro-max / frontend-design / impeccable 各 ≥1 条成功调用
|
||
（exec 且 exit_code==0，或 load 且 exit_code==0）。
|
||
轻量路由用 --require 指定参与该路由的上游子集，如仅审查任务:
|
||
--require "impeccable"
|
||
每项可带 note 前缀限定履职阶段，防止非履职调用（如 S2-research 调研检索）
|
||
混过门，如完整流程的严格口径:
|
||
--require "ui-ux-pro-max:S3-,frontend-design:S2-,impeccable"
|
||
--artifact 指定的每个文件必须存在。
|
||
任一不满足 → 报告标 FAIL，exit 1；全满足 → PASS，exit 0。
|
||
|
||
报告内容: 判定横幅、调用统计表、逐条调用清单（时间/依赖/命令/退出码）、
|
||
产物核验清单、失败与缺口说明。这份报告就是「真实集成」的可检查证据。
|
||
仅 python3 标准库。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import sys
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
GATE_DEPS = ["ui-ux-pro-max", "frontend-design", "impeccable"]
|
||
|
||
|
||
def load_records(ledger: Path) -> list[dict]:
|
||
if not ledger.is_file():
|
||
return []
|
||
records = []
|
||
for line in ledger.read_text(encoding="utf-8", errors="replace").splitlines():
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
rec = json.loads(line)
|
||
except json.JSONDecodeError:
|
||
records.append({"dep": "(unparsable)", "kind": "raw", "exit_code": None, "command": line[:120], "ts": ""})
|
||
continue
|
||
records.append(rec)
|
||
return records
|
||
|
||
|
||
def is_success(rec: dict) -> bool:
|
||
return rec.get("exit_code") == 0 and rec.get("kind") in ("exec", "load")
|
||
|
||
|
||
def one_line(rec: dict) -> str:
|
||
cmd = rec.get("command", "")
|
||
try: # exec 记录的 command 是 JSON 数组字符串
|
||
parts = json.loads(cmd)
|
||
if isinstance(parts, list):
|
||
cmd = " ".join(str(p) for p in parts)
|
||
except (json.JSONDecodeError, TypeError):
|
||
pass
|
||
return cmd if len(cmd) <= 160 else cmd[:157] + "..."
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(description="le-app-design 验收门")
|
||
ap.add_argument("--run-dir", required=True)
|
||
ap.add_argument("--out", default=None, help="报告写入路径，缺省 <run-dir>/INTEGRATION-REPORT.md")
|
||
ap.add_argument("--artifact", action="append", default=[], help="须存在的交付物路径（可多次）")
|
||
ap.add_argument("--title", default="le-app-design 集成证据报告")
|
||
ap.add_argument("--require", default=None,
|
||
help='门规格 "dep[:note前缀],..."；缺省=三个上游全部、不限前缀。轻量路由传子集')
|
||
args = ap.parse_args()
|
||
|
||
# 解析门规格：dep[:note-prefix]
|
||
gate: dict[str, str | None] = {}
|
||
if args.require:
|
||
for item in args.require.split(","):
|
||
item = item.strip()
|
||
if not item:
|
||
continue
|
||
dep, _, prefix = item.partition(":")
|
||
gate[dep] = prefix or None
|
||
else:
|
||
gate = {dep: None for dep in GATE_DEPS}
|
||
|
||
run_dir = Path(args.run_dir).expanduser()
|
||
records = load_records(run_dir / "calls.jsonl")
|
||
out_path = Path(args.out).expanduser() if args.out else run_dir / "INTEGRATION-REPORT.md"
|
||
|
||
stats = {dep: {"ok": 0, "fail": 0} for dep in gate}
|
||
for rec in records:
|
||
dep = rec.get("dep")
|
||
if dep in stats:
|
||
stats[dep]["ok" if is_success(rec) else "fail"] += 1
|
||
|
||
def gate_ok(dep: str) -> bool:
|
||
prefix = gate[dep]
|
||
if prefix is None:
|
||
return stats[dep]["ok"] >= 1
|
||
return any(
|
||
is_success(rec) and str(rec.get("note", "")).startswith(prefix)
|
||
for rec in records if rec.get("dep") == dep
|
||
)
|
||
|
||
gate_pass = all(gate_ok(dep) for dep in gate)
|
||
artifacts_missing = [a for a in args.artifact if not Path(a).expanduser().exists()]
|
||
overall = gate_pass and not artifacts_missing
|
||
|
||
lines: list[str] = []
|
||
lines.append(f"# {args.title}")
|
||
lines.append("")
|
||
lines.append(f"- 运行目录: `{run_dir}`")
|
||
lines.append(f"- 生成时间: {datetime.now(timezone.utc).isoformat(timespec='seconds')}")
|
||
lines.append(f"- 调用记录: {len(records)} 条（`calls.jsonl`）")
|
||
lines.append("")
|
||
lines.append(f"## 判定: {'✅ PASS' if overall else '❌ FAIL'}")
|
||
lines.append("")
|
||
lines.append("| 上游 | 成功调用 | 失败调用 | 门(≥1 成功) |")
|
||
lines.append("|---|---|---|---|")
|
||
for dep in gate:
|
||
s = stats[dep]
|
||
req = f"（限 note={gate[dep]}*）" if gate[dep] else ""
|
||
lines.append(f"| {dep}{req} | {s['ok']} | {s['fail']} | {'✅' if gate_ok(dep) else '❌ 未被真实调用'} |")
|
||
lines.append("")
|
||
|
||
if artifacts_missing:
|
||
lines.append("**缺失产物:**")
|
||
for a in artifacts_missing:
|
||
lines.append(f"- ❌ `{a}`")
|
||
lines.append("")
|
||
elif args.artifact:
|
||
lines.append("**产物核验:**")
|
||
for a in args.artifact:
|
||
lines.append(f"- ✅ `{a}`")
|
||
lines.append("")
|
||
|
||
lines.append("## 调用清单")
|
||
lines.append("")
|
||
lines.append("| # | 时间(UTC) | 依赖 | 类型 | 退出码 | 命令/装载 |")
|
||
lines.append("|---|---|---|---|---|---|")
|
||
for i, rec in enumerate(records, 1):
|
||
lines.append(
|
||
f"| {i} | {rec.get('ts', '')} | {rec.get('dep', '?')} | {rec.get('kind', '?')} "
|
||
f"| {rec.get('exit_code', '?')} | `{one_line(rec)}` |"
|
||
)
|
||
lines.append("")
|
||
|
||
if not overall:
|
||
lines.append("## 缺口说明")
|
||
lines.append("")
|
||
for dep in gate:
|
||
if not gate_ok(dep):
|
||
prefix = f"（履职阶段 note={gate[dep]}*）" if gate[dep] else ""
|
||
lines.append(f"- `{dep}` 无成功调用记录{prefix}——该上游未被真实使用，本次流程**不得标记为成功**。")
|
||
for a in artifacts_missing:
|
||
lines.append(f"- 产物缺失: `{a}`")
|
||
lines.append("")
|
||
|
||
report = "\n".join(lines)
|
||
out_path.parent.mkdir(parents=True, exist_ok=True)
|
||
out_path.write_text(report + "\n", encoding="utf-8")
|
||
print(report)
|
||
print(f"\n[report] {'PASS' if overall else 'FAIL'} → {out_path}", file=sys.stderr)
|
||
return 0 if overall else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|