build_pitfall_skill.py

#!/usr/bin/env python3
"""Build Cursor skill from 踩坑大全 markdown. See repo README."""
from __future__ import annotations

import argparse
import json
import re
from collections import defaultdict
from pathlib import Path
from typing import Dict, List, Optional, Set, Tuple

REPO_ROOT = Path(__file__).resolve().parent.parent
DEFAULT_ALIASES_PATH = REPO_ROOT / "config" / "tool_aliases.yaml"

from _domains import (  # noqa: E402
    domain_build_meta,
    domain_sources_dir,
    list_domain_slugs,
    load_domains_config,
    update_domain_tool_registry,
)


def _strip_category_heading(title: str) -> str:
    title = title.strip()
    if "(" in title and title.endswith(")"):
        return title.rsplit("(", 1)[0].strip()
    return title


def parse_pitfall_markdown(text: str, source_name: str) -> List[dict]:
    lines = text.splitlines()
    entries: List[dict] = []
    category = "uncategorized"
    i = 0
    n = len(lines)
    skip_headers = ("table of contents", "summary table")

    while i < n:
        line = lines[i]
        if line.startswith("## ") and not line.startswith("###"):
            heading = line[3:].strip()
            low = heading.lower()
            if any(low.startswith(s) for s in skip_headers) or heading.startswith("|-"):
                i += 1
                continue
            category = _strip_category_heading(heading)
            i += 1
            continue

        if line.startswith("### "):
            block = [line]
            i += 1
            while i < n:
                if lines[i].strip() == "---":
                    i += 1
                    break
                if lines[i].startswith("### "):
                    break
                if lines[i].startswith("## ") and not lines[i].startswith("###"):
                    break
                block.append(lines[i])
                i += 1
            body = "\n".join(block)
            entries.append({"category": category, "body": body, "source_file": source_name})
            continue
        i += 1
    return entries


def extract_tools(body: str) -> Optional[List[str]]:
    m = re.search(r"^\*\*Tools involved:\*\*\s*(.+)$", body, re.MULTILINE)
    if not m:
        return None
    parts = [p.strip() for p in m.group(1).split(",") if p.strip()]
    return parts if parts else None


def extract_issue_link(body: str) -> Optional[str]:
    m = re.search(r"https://github\.com/[^]\s)]+/issues/\d+", body)
    return m.group(0) if m else None


def extract_confidence(body: str) -> Optional[float]:
    m = re.search(r"^\*\*Confidence:\*\*\s*([\d.]+)", body, re.MULTILINE)
    if not m:
        return None
    try:
        return float(m.group(1))
    except ValueError:
        return None


def default_canonical(raw: str) -> str:
    """Fallback when no alias: trim and replace spaces with underscores."""
    return raw.strip().replace(" ", "_")


def load_tool_aliases(path: Path) -> Dict[str, str]:
    """Lowercase key -> canonical tool name. Minimal YAML subset (no PyYAML dep)."""
    if not path.exists():
        return {}
    out: Dict[str, str] = {}
    in_aliases = False
    for line in path.read_text(encoding="utf-8").splitlines():
        stripped = line.strip()
        if not stripped or stripped.startswith("#"):
            continue
        if re.match(r"^aliases:\s*$", stripped):
            in_aliases = True
            continue
        if in_aliases:
            if not (line.startswith(" ") or line.startswith("\t")):
                in_aliases = False
                continue
            content = line.strip()
            if ":" not in content:
                continue
            idx = content.index(":")
            k = content[:idx].strip().strip("'\"")
            v = content[idx + 1 :].strip().strip("'\"")
            if k and v:
                out[k.lower()] = v
    return out


def save_tool_aliases(path: Path, aliases: Dict[str, str]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    lines = [
        "# Maps lowercase form of each comma-separated token in **Tools involved:** to the",
        "# canonical tool slug (by-tool filenames, cross_index.json).",
        "# New tokens from domain sources/*.md are added on each build unless --no-update-aliases.",
        "# Edit values here to resolve ambiguities; keys are normalized to lowercase.",
        "",
        "aliases:",
    ]
    for k in sorted(aliases.keys(), key=str.lower):
        v = aliases[k]
        lines.append(f"  {json.dumps(k, ensure_ascii=False)}: {json.dumps(v, ensure_ascii=False)}")
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")


def collect_raw_tool_names(entries: List[dict]) -> Set[str]:
    names: Set[str] = set()
    for entry in entries:
        raw = extract_tools(entry["body"])
        if raw:
            names.update(raw)
    return names


def merge_aliases_for_build(
    alias_path: Path,
    raw_tool_names: Set[str],
    *,
    update_file: bool,
) -> Dict[str, str]:
    """
    Load existing aliases, add any raw tool token (lowercased key) missing with default_canonical(raw).
    In-memory map is always complete for this run. Writes YAML only when update_file and new keys appear
    (or the file did not exist yet).
    """
    aliases = load_tool_aliases(alias_path)
    added = False
    for raw in sorted(raw_tool_names, key=str.lower):
        key = raw.strip().lower()
        if key not in aliases:
            aliases[key] = default_canonical(raw)
            added = True
    if update_file and (added or not alias_path.exists()):
        save_tool_aliases(alias_path, aliases)
    return aliases


def canonicalize_tool(name: str, aliases: Dict[str, str]) -> str:
    key = name.strip().lower()
    if key in aliases:
        return aliases[key]
    return default_canonical(name)


def discover_source_markdowns(sources_dir: Path) -> List[Path]:
    if not sources_dir.is_dir():
        return []
    return sorted(sources_dir.glob("*.md"), key=lambda p: p.name.lower())


def compute_domain_tool_lists(
    sources_dir: Path,
    all_entries: List[dict],
    aliases: Dict[str, str],
) -> Tuple[List[str], List[str]]:
    """
    domain_tools: canonical names derived from each sources/*.md basename.
    cross_tools: canonical tools that appear in at least one multi-tool pitfall
    entry here but are not in domain_tools (interoperability / other-field tools).
    """
    md_paths = discover_source_markdowns(sources_dir)
    domain_tools = sorted(
        {canonicalize_tool(p.stem, aliases) for p in md_paths},
        key=str.lower,
    )
    domain_set = set(domain_tools)
    cross: Set[str] = set()
    for entry in all_entries:
        raw = extract_tools(entry["body"])
        if not raw or len(raw) < 2:
            continue
        tools = list(dict.fromkeys(canonicalize_tool(t, aliases) for t in raw))
        if len(tools) < 2:
            continue
        for t in tools:
            if t not in domain_set:
                cross.add(t)
    cross_tools = sorted(cross, key=str.lower)
    return domain_tools, cross_tools


def compose_auto_skill_description(
    skill_title: str,
    domain_tools: List[str],
    cross_tools: List[str],
) -> str:
    native = ", ".join(domain_tools) if domain_tools else "(none)"
    cross = ", ".join(cross_tools) if cross_tools else "(none)"
    return (
        f"{skill_title}. Domain tools (catalogue homes — use for field membership): {native}. "
        f"Cross partners (co-tools in pitfalls here — not membership): {cross}. "
        f"See skill Domain registry; config/domains.json is authoritative."
    )


def cross_filename_for_tools(tools: List[str]) -> str:
    sorted_tools = sorted(tools, key=lambda t: t.lower())
    return "__".join(sorted_tools) + ".md"


def entry_to_record(entry: dict, tools: List[str]) -> dict:
    body = entry["body"]
    return {
        "tools": tools,
        "category": entry.get("category"),
        "summary": body.splitlines()[0].lstrip("#").strip(),
        "detail": body,
        "fix": "",
        "confidence": extract_confidence(body),
        "issue_url": extract_issue_link(body),
        "source_markdown": body,
    }


def build(
    input_paths: List[Path],
    out_dir: Path,
    aliases: Dict[str, str],
    records_path: Optional[Path] = None,
) -> None:
    all_entries: List[dict] = []
    for p in input_paths:
        all_entries.extend(parse_pitfall_markdown(p.read_text(encoding="utf-8"), p.name))

    by_tool: Dict[str, List[str]] = defaultdict(list)
    cross_groups: Dict[Tuple[str, ...], List[str]] = defaultdict(list)
    seen_single: Set[str] = set()
    seen_cross: Set[tuple] = set()

    records_out: List[dict] = []

    for entry in all_entries:
        raw_tools = extract_tools(entry["body"])
        if not raw_tools:
            continue
        tools = list(dict.fromkeys(canonicalize_tool(t, aliases) for t in raw_tools))
        rec = entry_to_record(entry, tools)
        records_out.append(rec)
        url = extract_issue_link(entry["body"]) or ""

        if len(tools) == 1:
            dedup = f"{tools[0]}|{url}" if url else f"{tools[0]}|{hash(entry['body'])}"
            if dedup in seen_single:
                continue
            seen_single.add(dedup)
            by_tool[tools[0]].append(entry["body"])
            continue

        key_tuple = tuple(sorted(tools, key=lambda t: t.lower()))
        dedup = (url, key_tuple) if url else (str(hash(entry["body"])), key_tuple)
        if dedup in seen_cross:
            continue
        seen_cross.add(dedup)
        cross_groups[key_tuple].append(entry["body"])

    ref = out_dir / "reference"
    bt = ref / "by-tool"
    cr = ref / "cross"
    bt.mkdir(parents=True, exist_ok=True)
    cr.mkdir(parents=True, exist_ok=True)
    # Remove stale files so cross-only tools do not leave old by-tool stubs behind.
    for p in bt.glob("*.md"):
        p.unlink()
    for p in cr.glob("*.md"):
        p.unlink()

    cross_index: List[dict] = []

    for tool, bodies in sorted(by_tool.items(), key=lambda x: x[0].lower()):
        path = bt / f"{tool}.md"
        content = [
            f"# Pitfalls — {tool} (single-tool)",
            "",
            "> Auto-generated. Entries involve only this tool.",
            "",
        ]
        for b in bodies:
            content.append(b)
            content.append("")
            content.append("---")
            content.append("")
        path.write_text("\n".join(content).rstrip() + "\n", encoding="utf-8")

    for tool_tuple, bodies in sorted(cross_groups.items(), key=lambda x: x[0]):
        fname = cross_filename_for_tools(list(tool_tuple))
        rel = f"reference/cross/{fname}"
        path = cr / fname
        content = [
            f"# Cross-tool pitfalls — {' + '.join(tool_tuple)}",
            "",
            "> Auto-generated. Each entry involves multiple tools; not duplicated under by-tool/.",
            "",
        ]
        for b in bodies:
            content.append(b)
            content.append("")
            content.append("---")
            content.append("")
        path.write_text("\n".join(content).rstrip() + "\n", encoding="utf-8")
        cross_index.append({"path": rel, "tools": list(tool_tuple)})

    cross_index.sort(key=lambda x: x["path"])
    (out_dir / "cross_index.json").write_text(
        json.dumps(cross_index, indent=2, ensure_ascii=False) + "\n", encoding="utf-8"
    )

    if records_path:
        records_path.parent.mkdir(parents=True, exist_ok=True)
        with records_path.open("w", encoding="utf-8") as f:
            for rec in records_out:
                f.write(json.dumps(rec, ensure_ascii=False) + "\n")

    # No stub by-tool files for tools that only appear in cross entries — agents use
    # cross_index.json + reference/cross/ for those tools.


def write_skill_md(
    out_dir: Path,
    skill_name: str,
    skill_title: str,
    skill_description: str,
    domain_tools: List[str],
    cross_tools: List[str],
) -> None:
    native = ", ".join(domain_tools) if domain_tools else "(none)"
    cross = ", ".join(cross_tools) if cross_tools else "(none)"
    body = f"""---
name: {skill_name}
description: >-
  {skill_description}
---

# {skill_title}

## Domain registry (mirror of config/domains.json)

- **Domain tools** (`domain_tools`): native catalogue — this field owns `sources/<slug>/{{Tool}}.md` for these names. **Use this list (and those files) to decide whether a tool’s home field is this domain.**
- **Cross-tool partners** (`cross_tools`): names that appear **only** as co-tools in multi-tool pitfalls **recorded in this field’s sources**. They are **not** catalogue homes here.

**Do not** treat `cross_tools` as evidence that a tool **belongs** in this field. A tool listed only under `cross_tools` may belong to another domain or nowhere in this repo yet — classify new contributions using **`domain_tools` / `sources/<field>/{{Tool}}.md`** per slug, plus maintainer intent, not cross-partner lists.

When in doubt, read **`config/domains.json`** for every slug and compare `domain_tools` only for membership.

## When to use

- When working in this domain, before trusting defaults or implicit conventions in the listed tools.
- After drafting code: cross-check assumptions against the reference entries.

## How to look up (required order)

1. Normalize tool names (see `cross_index.json` for exact strings).
2. Read **`cross_index.json`** at the skill root. For each tool in context, collect every entry whose `tools` array contains that tool; read each listed `path` under `reference/cross/` (dedupe). **This is the only place to find pitfalls for tools that have no single-tool entries** (there may be no `reference/by-tool/<Tool>.md` for them).
3. For each relevant tool, **if** `reference/by-tool/<Tool>.md` exists, read it (single-tool pitfalls only). **If it does not exist**, that is normal for cross-only tools — step 2 already covers them.
4. Do **not** list `reference/cross/` by glob; use **only** `cross_index.json`.

## Output expectations

- Call out matching pitfalls with issue links when present.
- If confidence is low or evidence is thin, say so and suggest verifying against upstream docs.

## Maintenance

Regenerate via `python scripts/build_pitfall_skill.py --domain <slug>` (see repository README). Tool name aliases live in `config/tool_aliases.yaml`.
"""
    (out_dir / "SKILL.md").write_text(body, encoding="utf-8")


def _run_one_domain(domain_slug: str, args: argparse.Namespace) -> None:
    meta = domain_build_meta(domain_slug)
    sources_dir = domain_sources_dir(domain_slug)
    out = REPO_ROOT / str(meta["skill_out"])
    records = REPO_ROOT / str(meta["records"])

    if args.inputs is not None:
        input_paths = args.inputs
    else:
        input_paths = discover_source_markdowns(sources_dir)
        if not input_paths:
            raise SystemExit(
                f"No *.md found under {sources_dir}. Add sources or pass --inputs."
            )

    for p in input_paths:
        if not p.exists():
            raise SystemExit(f"Missing input: {p}")

    all_entries: List[dict] = []
    for p in input_paths:
        all_entries.extend(parse_pitfall_markdown(p.read_text(encoding="utf-8"), p.name))
    raw_names = collect_raw_tool_names(all_entries)
    aliases = merge_aliases_for_build(
        args.aliases,
        raw_names,
        update_file=not args.no_update_aliases,
    )

    domain_tools, cross_tools = compute_domain_tool_lists(
        sources_dir, all_entries, aliases
    )
    if not args.no_update_domains:
        update_domain_tool_registry(domain_slug, domain_tools, cross_tools)

    skill_title = str(meta.get("skill_title") or meta["skill_name"])
    skill_description = compose_auto_skill_description(
        skill_title, domain_tools, cross_tools
    )

    out.mkdir(parents=True, exist_ok=True)
    build(input_paths, out, aliases, records_path=records)
    write_skill_md(
        out,
        str(meta["skill_name"]),
        skill_title,
        skill_description,
        domain_tools,
        cross_tools,
    )

    def _rel(p: Path) -> str:
        try:
            return str(p.relative_to(REPO_ROOT))
        except ValueError:
            return str(p)

    print(f"[{domain_slug}] Sources ({len(input_paths)}): {[_rel(p) for p in input_paths]}")
    print(f"[{domain_slug}] Aliases: {_rel(args.aliases)} ({len(aliases)} keys)")
    print(f"[{domain_slug}] Wrote skill to {out.resolve()}")
    print(f"[{domain_slug}] Wrote records to {records.resolve()}")


def main() -> None:
    ap = argparse.ArgumentParser(
        description="Build pitfall skill from sources/<domain>/*.md (see config/domains.json)."
    )
    ap.add_argument(
        "--domain",
        action="append",
        dest="domains",
        metavar="SLUG",
        help="Domain slug(s) from config/domains.json (repeat to build several; default one if omitted)",
    )
    ap.add_argument(
        "--all-domains",
        action="store_true",
        help="Build every domain listed in config/domains.json",
    )
    ap.add_argument(
        "--inputs",
        nargs="+",
        type=Path,
        default=None,
        help="Markdown inputs (default: all *.md under the domain's sources dir)",
    )
    ap.add_argument(
        "--sources-dir",
        type=Path,
        default=None,
        help="Override domain layout: scan this directory for *.md (--domain ignored for inputs)",
    )
    ap.add_argument("--out", type=Path, default=None, help="Override output skill directory")
    ap.add_argument("--records", type=Path, default=None, help="Override JSONL records path")
    ap.add_argument(
        "--aliases",
        type=Path,
        default=DEFAULT_ALIASES_PATH,
        help="YAML file mapping lowercase tool tokens to canonical names",
    )
    ap.add_argument(
        "--no-update-aliases",
        action="store_true",
        help="Do not merge new tool tokens into the aliases file",
    )
    ap.add_argument(
        "--no-update-domains",
        action="store_true",
        help="Do not write domain_tools / cross_tools into config/domains.json",
    )
    args = ap.parse_args()

    if args.all_domains:
        slugs = list_domain_slugs()
        if not slugs:
            raise SystemExit(f"No domains in config/domains.json")
        for slug in slugs:
            _run_one_domain(slug, args)
        return

    if args.sources_dir is not None:
        # Legacy / override path: single build without domain metadata file
        sources_dir = args.sources_dir
        if args.out is None or args.records is None:
            raise SystemExit("With --sources-dir, pass both --out and --records")
        out = args.out
        records = args.records
        if args.inputs is not None:
            input_paths = args.inputs
        else:
            input_paths = discover_source_markdowns(sources_dir)
            if not input_paths:
                raise SystemExit(f"No *.md under {sources_dir}")
        for p in input_paths:
            if not p.exists():
                raise SystemExit(f"Missing input: {p}")
        all_entries: List[dict] = []
        for p in input_paths:
            all_entries.extend(parse_pitfall_markdown(p.read_text(encoding="utf-8"), p.name))
        raw_names = collect_raw_tool_names(all_entries)
        aliases = merge_aliases_for_build(
            args.aliases,
            raw_names,
            update_file=not args.no_update_aliases,
        )
        out.mkdir(parents=True, exist_ok=True)
        build(input_paths, out, aliases, records_path=records)
        dt, ct = compute_domain_tool_lists(sources_dir, all_entries, aliases)
        write_skill_md(
            out,
            "pitfall-skill",
            "Pitfall skill",
            compose_auto_skill_description("Pitfall skill", dt, ct),
            dt,
            ct,
        )
        print(f"Sources ({len(input_paths)}): {input_paths}")
        print(f"Wrote skill to {out.resolve()}")
        print(f"Wrote records to {records.resolve()}")
        return

    if args.out is not None or args.records is not None:
        raise SystemExit("Use --sources-dir with both --out and --records for ad-hoc builds.")

    if args.domains:
        for slug in args.domains:
            _run_one_domain(slug, args)
        return

    domain_slug: str | None = None
    cfg = load_domains_config()
    if "fusion-equilibrium" in cfg:
        domain_slug = "fusion-equilibrium"
    elif cfg:
        domain_slug = sorted(cfg.keys(), key=str.lower)[0]
    else:
        raise SystemExit(
            "Pass --domain <slug> or populate config/domains.json (see README)."
        )
    _run_one_domain(domain_slug, args)


if __name__ == "__main__":
    main()