ss-tools/merge_spec.py

# #region Tooling.MergeSpecModule [C:3] [TYPE Module]
# @LAYER Infra
# @BRIEF Merge one or many feature spec packages into a single review file.
# @RELATION DEPENDS_ON -> [Doc.Specify.Templates]
# @RATIONALE A reviewer LLM consumes the whole package as one artifact; batch mode
#   (ranges, multiple numbers, or "all") produces a single merged file per feature group.
# @REJECTED Merging only *.md — rejected because new packages carry contracts/openapi.yaml,
#   contracts/ux/*, and prototype/index.html that must be reviewed; a reviewer that never
#   sees the OpenAPI or prototype cannot judge contract closure.

# #region Tooling.Merge.Spec [C:3] [TYPE Module]
# @LAYER Infra

from __future__ import annotations

import re
import sys
from datetime import datetime
from pathlib import Path

REVIEW_PROMPT = (
    "Another LLM created this feature package. Your task: conduct an independent "
    "orthogonal spec review. Evaluate readiness, find contradictions, gaps, "
    "implementation risks, and prepare a structured report with corrections. "
    "Focus on spec review, not rewriting the implementation."
)

# Extensions that are safe to inline into a merged markdown review file.
MERGEABLE_EXTENSIONS = {".md", ".yaml", ".yml", ".html"}
# Extensions that are skipped (binary or non-review data).
SKIP_SUFFIXES = (
    ".json", ".py", ".pyc", ".zip", ".png", ".jpg", ".jpeg", ".webp", ".pdf",
)

# Canonical artifact order for each feature's merged output.
# Each stage: (match_type, path_value, section_label)
#   exact  = exact relative-path match (single file)
#   prefix = all files under that directory prefix
#   glob   = fnmatch-style pattern against relative path
CANONICAL_STAGES = (
    # Layer 1: Requirements & UX narrative
    ("exact", "spec.md",                 "SPEC — Feature Specification"),
    ("exact", "ux_reference.md",         "UX REFERENCE — Interaction Narrative"),
    ("prefix", "checklists/",            "CHECKLISTS — Requirements Quality"),

    # Layer 2: UX Design contracts
    ("exact", "contracts/ux/alternatives.md",   "UX ALTERNATIVES — Design Space Explored"),
    ("exact", "contracts/ux/decisions.md",      "UX DECISIONS — Final Choices"),
    ("exact", "contracts/ux/screen-models.md",  "UX SCREEN MODELS — Model Inventory"),
    ("exact", "contracts/ux/api-ux.md",         "UX API CONTRACT — Endpoints & Shapes"),
    ("exact", "contracts/ux/design-tokens.md",  None),  # force-first within prefix below
    ("prefix", "contracts/ux/",         "UX DESIGN — Per-Screen Contracts"),

    # Layer 3: Implementation plan + API
    ("exact", "plan.md",                "PLAN — Implementation Plan"),
    ("exact", "research.md",            "RESEARCH — Technical Decisions"),
    ("exact", "data-model.md",          "DATA MODEL — Entities & Relations"),
    ("exact", "contracts/modules.md",   "CONTRACTS — Module & Function Contracts"),
    ("exact", "contracts/openapi.yaml", "OPENAPI — REST/Event API Contract"),
    ("prefix", "contracts/",            "CONTRACTS — Remaining"),
    ("exact", "quickstart.md",          "QUICKSTART — Dev Onboarding"),

    # Layer 4: Traceability
    ("exact", "traceability.md",        "TRACEABILITY — Requirements Matrix"),

    # Layer 5: Execution
    ("exact", "tasks.md",               "TASKS — Implementation Tasks"),

    # Layer 6: Prototype
    ("exact", "prototype/manifest.md",  "PROTOTYPE — State/Manifest"),
    ("exact", "prototype/index.html",   "PROTOTYPE — Interactive HTML"),
    ("glob", "prototype/*",             "PROTOTYPE — Other"),
)

ORDER_HINT = (
    "Artifact order: spec → ux_reference → checklists → "
    "UX (alternatives → decisions → screen-models → api-ux → per-screen → design-tokens) → "
    "plan → research → data-model → modules → openapi → contracts → quickstart → "
    "traceability → tasks → prototype. Non-mergeable files (.json/.py/binary) are skipped."
)


def relative_key(path: Path, root: Path) -> str:
    return path.relative_to(root).as_posix()


def is_mergeable(path: Path) -> bool:
    if path.suffix in SKIP_SUFFIXES:
        return False
    if any(part in path.parts for part in ("__pycache__", ".git")):
        return False
    return path.suffix in MERGEABLE_EXTENSIONS


def ordered_artifacts(target_dir: Path) -> list[Path]:
    """Order reviewable files in a feature dir per CANONICAL_STAGES."""
    files = [p for p in target_dir.rglob("*") if p.is_file() and is_mergeable(p)]
    remaining = {relative_key(p, target_dir): p for p in files}

    ordered: list[Path] = []
    seen: set[str] = set()

    def take(path: Path) -> None:
        rel = relative_key(path, target_dir)
        if rel in seen:
            return
        ordered.append(path)
        seen.add(rel)
        remaining.pop(rel, None)

    for stage_type, stage_value, _label in CANONICAL_STAGES:
        if stage_type == "exact":
            path = remaining.pop(stage_value, None)
            if path is not None:
                take(path)
        elif stage_type == "prefix":
            matches = sorted(
                (p for rel, p in remaining.items() if rel.startswith(stage_value)),
                key=lambda p: relative_key(p, target_dir),
            )
            for p in matches:
                take(p)
        elif stage_type == "glob":
            import fnmatch
            matches = sorted(
                (p for rel, p in remaining.items() if fnmatch.fnmatch(rel, stage_value)),
                key=lambda p: relative_key(p, target_dir),
            )
            for p in matches:
                take(p)

    # Any remaining reviewable files — deterministic order at the end.
    for rel in sorted(remaining):
        take(remaining[rel])

    return ordered


def resolve_targets(specs_dir: Path, tokens: list[str]) -> list[Path]:
    """Resolve CLI tokens into concrete feature directories.

    Supports:
      - single number:  "038"
      - number range:   "036-041"  (inclusive)
      - explicit list:  "036 038 044"
      - dir name:       "042-dashboard-scenario-registry"
      - all:            "all"
    """
    targets: list[Path] = []

    def find_by_number(num: str) -> Path | None:
        for item in specs_dir.iterdir():
            if item.is_dir() and item.name.startswith(f"{num}-"):
                return item
        return None

    for token in tokens:
        token = token.strip()
        if not token:
            continue
        if token.lower() == "all":
            for item in sorted(specs_dir.iterdir(), key=lambda p: p.name):
                if item.is_dir() and re.match(r"^\d{3}-", item.name):
                    targets.append(item)
            continue
        range_match = re.fullmatch(r"(\d{3})-(\d{3})", token)
        if range_match:
            lo, hi = int(range_match.group(1)), int(range_match.group(2))
            if lo > hi:
                lo, hi = hi, lo
            for num in range(lo, hi + 1):
                d = find_by_number(f"{num:03d}")
                if d is not None:
                    targets.append(d)
                else:
                    print(f"  [warn] no spec for number {num:03d} — skipped")
            continue
        if re.fullmatch(r"\d{3}", token):
            d = find_by_number(token)
            if d is not None:
                targets.append(d)
            else:
                print(f"  [warn] no spec for number {token} — skipped")
            continue
        # Treat as an explicit relative path / dir name.
        cand = specs_dir / token
        if cand.is_dir():
            targets.append(cand)
        else:
            print(f"  [warn] unrecognized target '{token}' — skipped")

    # De-duplicate while preserving order.
    seen: set[Path] = set()
    unique: list[Path] = []
    for t in targets:
        if t.resolve() not in seen:
            seen.add(t.resolve())
            unique.append(t)
    return unique


def _section_label(rel: str) -> str:
    for stage_type, stage_value, label in CANONICAL_STAGES:
        if label is None:
            continue
        if stage_type == "exact" and rel == stage_value:
            return label
        if stage_type == "prefix" and rel.startswith(stage_value):
            return f"{label} — {Path(rel).name}"
        if stage_type == "glob":
            import fnmatch
            if fnmatch.fnmatch(rel, stage_value):
                return f"{label} — {Path(rel).name}"
    return rel


def merge_one_feature(target_dir: Path, blocks: list[str]) -> int:
    """Append one feature's merged content to `blocks`. Returns file count."""
    feature_name = target_dir.name
    artifacts = ordered_artifacts(target_dir)
    blocks.append("")
    blocks.append("=" * 80)
    blocks.append(f"FEATURE: {feature_name}")
    blocks.append(f"Files: {len(artifacts)}")
    blocks.append("=" * 80)
    blocks.append("")

    for file_path in artifacts:
        rel = relative_key(file_path, target_dir)
        try:
            content = file_path.read_text(encoding="utf-8")
        except Exception as e:  # noqa: BLE001
            print(f"  [warn] skipping {rel}: {e}")
            continue
        blocks.append("")
        blocks.append("-" * 60)
        blocks.append(f"## {_section_label(rel)}")
        blocks.append(f"Source: {rel}")
        blocks.append("-" * 60)
        blocks.append("")
        blocks.append(content)
        blocks.append("")

    return len(artifacts)


def merge_specs(tokens: list[str], output: str | None = None) -> str | None:
    """Merge all requested spec packages into one review file."""
    specs_dir = Path("specs")
    if not specs_dir.exists():
        print("Error: 'specs' directory not found.")
        return None

    targets = resolve_targets(specs_dir, tokens)
    if not targets:
        print("Error: no specs matched the given arguments.")
        return None

    now = datetime.now().strftime("%Y%m%d-%H%M%S")
    if output is None:
        if len(targets) == 1:
            output = f"{targets[0].name}-{now}.md"
        else:
            first, last = targets[0].name[:3], targets[-1].name[:3]
            output = f"specs-{first}-{last}-{now}.md"

    blocks = [
        REVIEW_PROMPT,
        "",
        "=" * 80,
        ORDER_HINT,
        f"Features: {', '.join(t.name for t in targets)}",
        f"Generated: {datetime.now().isoformat()}",
        "=" * 80,
        "",
    ]

    total_files = 0
    for target in targets:
        total_files += merge_one_feature(target, blocks)

    blocks.append("")
    blocks.append("=" * 80)
    blocks.append(f"Features merged: {len(targets)} | Files merged: {total_files}")
    blocks.append("=" * 80)
    blocks.append("")

    merged = "\n".join(blocks)
    Path(output).write_text(merged, encoding="utf-8")
    print(f"Created: {output} ({len(targets)} features, {total_files} files)")
    return output


def main(argv: list[str]) -> int:
    if len(argv) < 2:
        print("Usage:")
        print("  python merge_spec.py <number>              # single spec, e.g. 038")
        print("  python merge_spec.py <lo>-<hi>             # range, e.g. 036-041")
        print("  python merge_spec.py <n> <n> ...           # explicit list, e.g. 036 040 044")
        print("  python merge_spec.py all                   # every spec package")
        print("  python merge_spec.py <dir-name>            # by directory name")
        print("  python merge_spec.py <n>-<n> -o out.md     # custom output file")
        return 1

    tokens = argv[1:]
    out_path = None
    if "-o" in tokens or "--output" in tokens:
        flag_idx = tokens.index("-o") if "-o" in tokens else tokens.index("--output")
        if flag_idx + 1 >= len(tokens):
            print("Error: -o/--output requires a filename")
            return 1
        out_path = tokens[flag_idx + 1]
        tokens = tokens[:flag_idx] + tokens[flag_idx + 2:]

    result = merge_specs(tokens, output=out_path)
    return 0 if result else 1


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

# #endregion Tooling.Merge.Spec

# #endregion Tooling.MergeSpecModule