dingtalk-workspace-cli/scripts/gen_shortcut_public_catalog.py

#!/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", "<br>")


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("生成时间：`<generated>`", existing) == without_time.sub(
            "生成时间：`<generated>`", 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())