market/scripts/gen-collection-children.py

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.10"
# dependencies = ["pyyaml>=6.0"]
# ///
"""Generate `children` for collection entries in skills/<id>/entry.json.

A collection entry points at an upstream repo that holds several sibling skills
(e.g. larksuite/cli has skills/lark-im, skills/lark-doc, ...). One market card
covers them all; the client lets the user tick which ones to install. That needs
a declared child list so the detail page renders without touching the network.

For each requested entry this script:
  1. clones the pinned ref (shallow, falling back to fetch-by-SHA);
  2. walks the tree for SKILL.md files, skipping test fixtures and vendored dirs;
  3. reads frontmatter name/description/version;
  4. writes `children` back into entry.json, leaving every other field untouched.

Usage:
  python3 scripts/gen-collection-children.py                 # all entries that already declare children
  python3 scripts/gen-collection-children.py larksuite-cli   # specific ids
  python3 scripts/gen-collection-children.py --check          # verify without writing
"""

from __future__ import annotations

import argparse
import json
import os
import re
import subprocess
import sys
import tempfile
from pathlib import Path

import yaml

REPO_ROOT = Path(__file__).resolve().parent.parent
SKILLS_DIR = REPO_ROOT / "skills"

# Path segments that never hold a publishable skill: test fixtures for skill
# linters look exactly like real skills (larksuite/cli ships six of them under
# scripts/skill-format-check/tests/), so excluding them is not optional.
EXCLUDED_SEGMENTS = {
    ".git", "node_modules", "upstream", ".cache", "vendor", "internal",
    "test", "tests", "__tests__", "testdata", "test-data",
    "fixture", "fixtures", "__fixtures__",
    "example", "examples", "template", "templates",
}

# Descriptions run to several hundred characters upstream; the picker only has
# room for one line, so keep the first sentence.
MAX_SHORT_DESC = 160


def run(cmd: list[str], cwd: Path | None = None) -> subprocess.CompletedProcess:
    try:
        return subprocess.run(cmd, cwd=cwd, capture_output=True, text=True, timeout=300,
                              env={**os.environ, "GIT_TERMINAL_PROMPT": "0", "GCM_INTERACTIVE": "never"})
    except subprocess.TimeoutExpired:
        return subprocess.CompletedProcess(cmd, 124, "", "timed out")


def clone_pinned(repo_url: str, branch: str, ref: str | None, dest: Path) -> None:
    """Clone at the pinned ref. Mirrors the client's cloneGitToTemp behaviour."""
    result = run(["git", "clone", "--quiet", "--single-branch", "-b", branch,
                  "--depth", "100", repo_url, str(dest)])
    if result.returncode != 0:
        raise RuntimeError(f"clone failed: {result.stderr.strip()[:300]}")

    if not ref:
        return

    if run(["git", "checkout", "--quiet", ref], cwd=dest).returncode == 0:
        return

    # Pinned commit fell outside the shallow window — fetch that object alone.
    # Both failures carry stderr: "cannot fetch pinned ref <sha>" alone cannot
    # tell a missing commit object from a permissions or network problem.
    fetched = run(["git", "fetch", "--quiet", "origin", ref, "--depth", "1"], cwd=dest)
    if fetched.returncode != 0:
        raise RuntimeError(f"cannot fetch pinned ref {ref}: {fetched.stderr.strip()[:300]}")

    checked_out = run(["git", "checkout", "--quiet", ref], cwd=dest)
    if checked_out.returncode != 0:
        raise RuntimeError(f"cannot check out pinned ref {ref}: {checked_out.stderr.strip()[:300]}")


def parse_frontmatter(path: Path) -> dict:
    try:
        text = path.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return {}
    return parse_frontmatter_text(text)


def parse_frontmatter_text(text: str) -> dict:
    match = re.match(r"^---\r?\n(.*?)\r?\n---", text, re.S)
    if not match:
        return {}
    try:
        data = yaml.safe_load(match.group(1))
    except yaml.YAMLError:
        return {}
    return data if isinstance(data, dict) else {}


def skill_paths(repo_dir: Path, revision: str = "HEAD") -> list[str]:
    """Use Git's exact filenames, independent of filesystem case folding."""
    if (repo_dir / ".git").exists():
        result = run(["git", "ls-tree", "-rz", "--full-tree", revision], cwd=repo_dir)
        if result.returncode:
            raise RuntimeError(f"cannot list skill files: {result.stderr.strip()[:300]}")
        paths = []
        for record in result.stdout.split("\0"):
            if not record:
                continue
            info, path = record.split("\t", 1)
            mode, kind, _ = info.split()
            if kind == "blob" and mode != "120000" and path.split("/")[-1] == "SKILL.md":
                paths.append(path)
        return sorted(paths, key=lambda path: tuple(path.split("/")))

    # Support local authoring fixtures without a Git repository. os.walk yields
    # actual directory entries; glob("SKILL.md") can match lowercase skill.md.
    return sorted(
        ((Path(directory) / name).relative_to(repo_dir).as_posix()
         for directory, _, names in os.walk(repo_dir)
         for name in names
         if name == "SKILL.md" and not (Path(directory) / name).is_symlink()),
        key=lambda path: tuple(path.split("/")),
    )


def skill_frontmatter(repo_dir: Path, path: str, revision: str = "HEAD") -> dict:
    if (repo_dir / ".git").exists():
        result = run(["git", "show", f"{revision}:{path}"], cwd=repo_dir)
        if result.returncode:
            raise RuntimeError(f"cannot read {path}: {result.stderr.strip()[:300]}")
        return parse_frontmatter_text(result.stdout)
    return parse_frontmatter(repo_dir / path)


def first_sentence(text: str) -> str:
    flat = " ".join(str(text).split())
    match = re.search(r"^(.{1,%d}?[。.!?！？])\s" % MAX_SHORT_DESC, flat + " ")
    candidate = match.group(1) if match else flat
    if len(candidate) > MAX_SHORT_DESC:
        candidate = candidate[:MAX_SHORT_DESC].rstrip() + "…"
    return candidate


def normalize_version(raw) -> str | None:
    """Mirror the client's normalizeSemver: upstream writes `1.0` and `v1.2.3`."""
    if raw is None:
        return None
    core = str(raw).strip().lstrip("vV").split("-")[0].split("+")[0]
    match = re.match(r"^(\d+)(?:\.(\d+))?(?:\.(\d+))?$", core)
    if not match:
        return None
    major, minor, patch = match.groups()
    return f"{int(major)}.{int(minor or 0)}.{int(patch or 0)}"


def discover_children(repo_dir: Path, revision: str = "HEAD") -> list[dict]:
    children: list[dict] = []
    seen: dict[str, str] = {}

    # Convention first: when the repo has a top-level skills/ directory, that is
    # the published set. Scanning the whole tree instead picks up decoys such as
    # larksuite/cli's internal/qualitygate/skillscan/testdata/skills/lark-demo,
    # which is a linter fixture, not a shippable skill.
    paths = skill_paths(repo_dir, revision)
    has_skills = any(path.startswith("skills/") for path in paths)

    for path in paths:
        if has_skills and not path.startswith("skills/"):
            continue
        rel = Path(path)
        segments = [s.lower() for s in rel.parts[:-1]]
        if any(seg in EXCLUDED_SEGMENTS for seg in segments):
            continue
        if not segments:
            continue  # repo root itself is a single skill, not a collection

        fm = skill_frontmatter(repo_dir, path, revision)
        metadata = fm.get("metadata")
        if not fm.get("name") or (isinstance(metadata, dict) and metadata.get("internal") is True):
            continue

        child_id = rel.parts[-2]
        if not re.match(r"^[a-z0-9-]+$", child_id):
            print(f"  ! skipped {rel.parent}: directory name is not a valid skill id", file=sys.stderr)
            continue
        if child_id in seen:
            print(f"  ! skipped {rel.parent}: id '{child_id}' already taken by {seen[child_id]}", file=sys.stderr)
            continue
        seen[child_id] = str(rel.parent)

        # Git catalog paths use POSIX separators on every contributor platform.
        child: dict = {"id": child_id, "path": rel.parent.as_posix()}
        name = fm.get("display_name") or fm.get("name")
        if name and str(name) != child_id:
            child["name"] = str(name)
        version = normalize_version(fm.get("version"))
        if version:
            child["version"] = version
        description = fm.get("description")
        if description:
            child["i18n"] = {
                "zh-CN": {"shortDesc": first_sentence(description)},
                "en-US": {"shortDesc": first_sentence(description)},
            }
        children.append(child)

    return children


def apply_presentation_overrides(entry_id: str, children: list[dict]) -> list[dict]:
    """Keep reviewed translations reproducible without overriding source facts."""
    path = SKILLS_DIR / entry_id / "collection-child-overrides.json"
    if not path.exists():
        return children
    overrides = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(overrides, dict):
        raise RuntimeError("child presentation overrides must be an object")
    by_id = {child["id"]: child for child in children}
    for child_id, override in overrides.items():
        if child_id not in by_id:
            raise RuntimeError(f"presentation override names missing upstream child '{child_id}'")
        if not isinstance(override, dict) or set(override) != {"i18n"}:
            raise RuntimeError(f"presentation override for '{child_id}' may only contain i18n")
        locales = override["i18n"]
        if not isinstance(locales, dict) or not locales:
            raise RuntimeError(f"presentation override for '{child_id}' must declare locales")
        for locale, payload in locales.items():
            if locale not in {"en-US", "zh-CN"} or not isinstance(payload, dict) or set(payload) != {"shortDesc"}:
                raise RuntimeError(f"invalid presentation override locale for '{child_id}'")
            if not isinstance(payload["shortDesc"], str) or not payload["shortDesc"].strip():
                raise RuntimeError(f"presentation override for '{child_id}' needs a non-empty summary")
        by_id[child_id]["i18n"] = locales
    return children


def process(entry_id: str, *, check: bool = False) -> bool:
    entry_path = SKILLS_DIR / entry_id / "entry.json"
    if not entry_path.exists():
        print(f"{entry_id}: no entry.json")
        return False

    entry = json.loads(entry_path.read_text(encoding="utf-8"))
    source = entry.get("source", {})
    if source.get("kind") != "git":
        print(f"{entry_id}: only git sources can be collections (kind={source.get('kind')})")
        return False
    if check and source.get("path") is not None:
        print(f"{entry_id}: stale — collection source.path must be omitted")
        return False
    if check and not source.get("ref"):
        print(
            f"{entry_id}: SKIP — source.ref is not pinned; "
            "a mutable source cannot have a deterministic generated-output check"
        )
        return True

    print(f"{entry_id}: cloning {source['repoUrl']} …")
    with tempfile.TemporaryDirectory(prefix=f"collection-{entry_id}-") as tmp:
        repo_dir = Path(tmp) / "repo"
        try:
            clone_pinned(source["repoUrl"], source.get("repoBranch", "main"), source.get("ref"), repo_dir)
        except RuntimeError as err:
            print(f"{entry_id}: {err}")
            return False
        try:
            children = apply_presentation_overrides(entry_id, discover_children(repo_dir))
        except (RuntimeError, ValueError) as err:
            print(f"{entry_id}: {err}")
            return False

    if not children:
        print(f"{entry_id}: no sub-skills found — not a collection?")
        return False

    if check:
        if entry.get("children") != children:
            print(
                f"{entry_id}: stale — declared children differ from the "
                f"{len(children)} children discovered at the pinned source"
            )
            return False
        print(f"{entry_id}: OK ({len(children)} children)")
        return True

    entry.pop("children", None)
    entry["children"] = children
    # source.path is mutually exclusive with children (the client rejects both).
    entry.get("source", {}).pop("path", None)
    entry_path.write_text(json.dumps(entry, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    print(f"{entry_id}: wrote {len(children)} children")
    return True


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--check",
        action="store_true",
        help="compare generated children with entry.json without writing files",
    )
    parser.add_argument("ids", nargs="*", help="collection entry IDs (default: every declared collection)")
    args = parser.parse_args(argv)

    ids = args.ids
    if not ids:
        ids = sorted(
            p.parent.name
            for p in SKILLS_DIR.glob("*/entry.json")
            if "children" in json.loads(p.read_text(encoding="utf-8"))
        )
        if not ids:
            print("no collection entries found; pass ids explicitly")
            return 1

    failed = [entry_id for entry_id in ids if not process(entry_id, check=args.check)]
    if failed:
        print(f"\nfailed: {', '.join(failed)}")
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))