#!/usr/bin/env python3
"""Generate the public shortcut catalog from real test results."""
from __future__ import annotations
import datetime as dt
import json
import re
import subprocess
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
READ_PATH = ROOT / "docs" / "shortcut-real-read-results.json"
WRITE_PATH = ROOT / "docs" / "shortcut-real-write-results.json"
GO_PATH = ROOT / "internal" / "shortcut" / "public_catalog_generated.go"
CATALOG_PATH = ROOT / "docs" / "shortcut-public-catalog.json"
FOLLOWUP_MD_PATH = ROOT / "docs" / "shortcut-real-test-followups.md"
FOLLOWUP_JSON_PATH = ROOT / "docs" / "shortcut-real-test-followups.json"
def load(path: Path) -> dict[str, Any]:
return json.loads(path.read_text(encoding="utf-8"))
def squish(text: str, limit: int = 180) -> str:
text = re.sub(r"\s+", " ", text or "").strip()
if len(text) <= limit:
return text
return text[: limit - 1] + "…"
def error_text(row: dict[str, Any]) -> str:
return f"{row.get('stderr') or ''}\n{row.get('stdout') or ''}"
def parse_error_payload(row: dict[str, Any]) -> dict[str, Any]:
for key in ("stderr", "stdout"):
raw = (row.get(key) or "").strip()
if not raw:
continue
try:
value = json.loads(raw)
except json.JSONDecodeError:
continue
if isinstance(value, dict):
err = value.get("error")
if isinstance(err, dict):
return err
return value
return {}
def evidence(row: dict[str, Any]) -> str:
payload = parse_error_payload(row)
msg = payload.get("message")
if isinstance(msg, str) and msg.strip():
return squish(msg, 220)
return squish(error_text(row), 220)
def collect() -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
public: list[dict[str, Any]] = []
followups: list[dict[str, Any]] = []
for suite, path in (("read", READ_PATH), ("write", WRITE_PATH)):
data = load(path)
for row in data.get("results", []):
item = {
"suite": suite,
"service": row.get("service") or "",
"command": row.get("command") or "",
"risk": row.get("risk") or "",
"status": row.get("status") or "",
}
if row.get("status") == "real-ok":
public.append(item)
continue
followups.append({
**item,
"failure_category": row.get("failure_category") or "unclassified",
"fixability": row.get("fixability") or "",
"diagnosis": row.get("diagnosis") or "",
"evidence": evidence(row),
})
public.sort(key=lambda r: (r["service"], r["command"]))
followups.sort(key=lambda r: (r["suite"], r["service"], r["command"]))
return public, followups
def go_quote(s: str) -> str:
return json.dumps(s, ensure_ascii=False)
def write_go(rows: list[dict[str, Any]]) -> None:
lines = [
"// Code generated by scripts/gen_shortcut_public_catalog.py; DO NOT EDIT.",
"",
"package shortcut",
"",
"// publicShortcutCatalog contains the generated shortcut surface used by",
"// command discovery and skill generation.",
"func generatedPublicShortcutCatalog() map[string]struct{} {",
"\treturn map[string]struct{}{",
]
for row in rows:
key = f"{row['service']}\x00{row['command']}"
lines.append(f"\t\t{go_quote(key)}: {{}},")
lines.extend(["\t}", "}", ""])
GO_PATH.write_text("\n".join(lines), encoding="utf-8")
subprocess.run(["gofmt", "-w", str(GO_PATH)], check=True)
def md_escape(s: str) -> str:
return str(s).replace("|", "\\|").replace("\n", "
")
def write_followup_md(rows: list[dict[str, Any]]) -> None:
generated = dt.datetime.now().isoformat(timespec="seconds")
lines = [
"# Shortcut 真实测试跟进清单",
"",
f"生成时间：`{generated}`",
"",
"来源：`docs/shortcut-real-read-results.json` 与 `docs/shortcut-real-write-results.json`。",
"",
"口径：记录真实后端测试中需要继续定位的 case，用于 CR 和问题分派；Agent 使用入口以公开 shortcut catalog 和产品 skill 为准。",
"",
f"总计：{len(rows)} 条。",
"",
"| # | suite | shortcut | risk | status | category | fixability | 处理依据 |",
"|---:|---|---|---|---|---|---|---|",
]
for idx, row in enumerate(rows, 1):
shortcut = f"{row['service']} {row['command']}"
basis = row["diagnosis"] or row["evidence"]
lines.append(
f"| {idx} | {md_escape(row['suite'])} | `{md_escape(shortcut)}` | "
f"{md_escape(row['risk'])} | {md_escape(row['status'])} | "
f"{md_escape(row['failure_category'])} | {md_escape(row['fixability'])} | "
f"{md_escape(squish(basis, 160))} |"
)
lines.append("")
rendered = "\n".join(lines)
if FOLLOWUP_MD_PATH.exists():
existing = FOLLOWUP_MD_PATH.read_text(encoding="utf-8")
without_time = re.compile(r"生成时间：`[^`]+`")
if without_time.sub("生成时间：``", existing) == without_time.sub(
"生成时间：``", rendered
):
return
FOLLOWUP_MD_PATH.write_text(rendered, encoding="utf-8")
def write_json(public: list[dict[str, Any]], followups: list[dict[str, Any]]) -> None:
generated = dt.datetime.now().isoformat()
write_json_if_changed(CATALOG_PATH, {
"count": len(public),
"results": public,
}, generated)
write_json_if_changed(FOLLOWUP_JSON_PATH, {
"count": len(followups),
"results": followups,
}, generated)
def write_json_if_changed(path: Path, payload: dict[str, Any], generated: str) -> None:
if path.exists():
existing = load(path)
existing.pop("generated_at", None)
if existing == payload:
return
path.write_text(json.dumps({
"generated_at": generated,
**payload,
}, ensure_ascii=False, indent=2), encoding="utf-8")
def main() -> int:
public, followups = collect()
write_go(public)
write_followup_md(followups)
write_json(public, followups)
print(f"public={len(public)} followups={len(followups)}")
print(f"written: {GO_PATH}")
print(f"written: {CATALOG_PATH}")
print(f"written: {FOLLOWUP_MD_PATH}")
print(f"written: {FOLLOWUP_JSON_PATH}")
return 0
if __name__ == "__main__":
raise SystemExit(main())