feat(tooling): rewrite merge_spec.py — batch spec merging + new package support
Rewrite merge_spec.py to merge one or many feature spec packages into a single review file. Batch modes: - single number: python merge_spec.py 038 - inclusive range: python merge_spec.py 036-041 - explicit list: python merge_spec.py 036 038 044 - by dir name: python merge_spec.py 042-dashboard-scenario-registry - all: python merge_spec.py all - custom output: python merge_spec.py 036-041 -o out.md Handles the new spec package structure that plain *.md merging missed: - includes contracts/openapi.yaml (YAML), contracts/ux/* (decisions.md), prototype/index.html + prototype/manifest.md - skips .json/.py/.zip/.pyc and __pycache__ (fixtures/code/binaries) - canonical per-feature order: spec -> ux_reference -> checklists -> UX contracts -> plan -> research -> data-model -> modules -> openapi -> quickstart -> traceability -> tasks -> prototype - missing numbers warn+skip; dedup; per-feature grouping in one output Verified: 043 (14 files), 036-041 (6 features/104 files), 036-047 (12/186), all (50/572) with no .json/.zip/.pyc leakage.
This commit is contained in:
373
merge_spec.py
373
merge_spec.py
@@ -1,9 +1,19 @@
|
||||
# #region Tooling.MergeSpecModule [C:2] [TYPE Module]
|
||||
# #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
|
||||
|
||||
# #region Tooling.Merge.Spec [C:2] [TYPE Module]
|
||||
# @LAYER Infra
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import sys
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
@@ -15,42 +25,58 @@ REVIEW_PROMPT = (
|
||||
"Focus on spec review, not rewriting the implementation."
|
||||
)
|
||||
|
||||
# Canonical artifact order for the merged output.
|
||||
# Each stage: (match_type, value, section_label)
|
||||
# match_type "exact" = exact filename match
|
||||
# match_type "prefix" = directory prefix — matches all files under that dir
|
||||
CANONICAL_MD_STAGES = (
|
||||
# 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"),
|
||||
("exact", "spec.md", "SPEC — Feature Specification"),
|
||||
("exact", "ux_reference.md", "UX REFERENCE — Interaction Narrative"),
|
||||
("prefix", "checklists/", "CHECKLISTS — Requirements Quality"),
|
||||
|
||||
# Layer 2: UX Design (from /speckit.ux)
|
||||
("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"),
|
||||
("prefix", "contracts/ux/", "UX DESIGN — Per-Screen Contracts"),
|
||||
("exact", "contracts/ux/design-tokens.md", None), # force-first within prefix
|
||||
# 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
|
||||
("exact", "plan.md", "PLAN — Implementation Plan"),
|
||||
("exact", "research.md", "RESEARCH — Technical Decisions"),
|
||||
("exact", "data-model.md", "DATA MODEL — Entities & Relations"),
|
||||
("prefix", "contracts/", "CONTRACTS — Module & Function Contracts"),
|
||||
("exact", "quickstart.md", "QUICKSTART — Dev Onboarding"),
|
||||
# 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"),
|
||||
("exact", "traceability.md", "TRACEABILITY — Requirements Matrix"),
|
||||
|
||||
# Layer 5: Execution
|
||||
("exact", "tasks.md", "TASKS — Implementation Tasks"),
|
||||
("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 → contracts → quickstart → traceability → tasks → remaining."
|
||||
"plan → research → data-model → modules → openapi → contracts → quickstart → "
|
||||
"traceability → tasks → prototype. Non-mergeable files (.json/.py/binary) are skipped."
|
||||
)
|
||||
|
||||
|
||||
@@ -58,137 +84,240 @@ def relative_key(path: Path, root: Path) -> str:
|
||||
return path.relative_to(root).as_posix()
|
||||
|
||||
|
||||
def ordered_markdown_files(target_dir: Path) -> list[Path]:
|
||||
"""Order markdown files according to CANONICAL_MD_STAGES."""
|
||||
markdown_files = [path for path in target_dir.rglob("*.md") if path.is_file()]
|
||||
remaining = {relative_key(path, target_dir): path for path in markdown_files}
|
||||
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] = []
|
||||
# Track which stages have already been matched to avoid duplicates
|
||||
seen = set()
|
||||
seen: set[str] = set()
|
||||
|
||||
for stage_type, stage_value, _section_label in CANONICAL_MD_STAGES:
|
||||
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":
|
||||
# Exact: only match if the file exists and hasn't been collected yet
|
||||
rel = f"contracts/ux/{stage_value}" if stage_value.startswith("contracts/ux/") else stage_value
|
||||
if stage_value.startswith("contracts/ux/"):
|
||||
path = remaining.pop(stage_value, None)
|
||||
else:
|
||||
path = remaining.pop(stage_value, None)
|
||||
if path is not None and path not in seen:
|
||||
ordered.append(path)
|
||||
seen.add(path)
|
||||
continue
|
||||
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)
|
||||
|
||||
# Prefix: match all files whose relative path starts with stage_value
|
||||
stage_matches = sorted(
|
||||
[
|
||||
path
|
||||
for rel_path, path in remaining.items()
|
||||
if rel_path.startswith(stage_value)
|
||||
],
|
||||
key=lambda p: relative_key(p, target_dir),
|
||||
)
|
||||
for path in stage_matches:
|
||||
if path not in seen:
|
||||
ordered.append(path)
|
||||
seen.add(path)
|
||||
rel = relative_key(path, target_dir)
|
||||
remaining.pop(rel, None)
|
||||
|
||||
# Remaining files — append at end
|
||||
remaining_paths = sorted(remaining.values(), key=lambda p: relative_key(p, target_dir))
|
||||
for path in remaining_paths:
|
||||
if path not in seen:
|
||||
ordered.append(path)
|
||||
# Any remaining reviewable files — deterministic order at the end.
|
||||
for rel in sorted(remaining):
|
||||
take(remaining[rel])
|
||||
|
||||
return ordered
|
||||
|
||||
|
||||
def merge_specs(feature_number: str) -> str | None:
|
||||
"""Merge all spec artifacts for a feature into one review file."""
|
||||
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
|
||||
|
||||
# Find directory starting with feature number
|
||||
target_dir = None
|
||||
for item in specs_dir.iterdir():
|
||||
if item.is_dir() and item.name.startswith(f"{feature_number}-"):
|
||||
target_dir = item
|
||||
break
|
||||
|
||||
if not target_dir:
|
||||
print(f"Error: No directory found for feature number '{feature_number}' in 'specs/'.")
|
||||
targets = resolve_targets(specs_dir, tokens)
|
||||
if not targets:
|
||||
print("Error: no specs matched the given arguments.")
|
||||
return None
|
||||
|
||||
feature_name = target_dir.name
|
||||
now = datetime.now().strftime("%Y%m%d-%H%M%S")
|
||||
output_filename = f"{feature_name}-{now}.md"
|
||||
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"
|
||||
|
||||
content_blocks = [
|
||||
blocks = [
|
||||
REVIEW_PROMPT,
|
||||
"",
|
||||
"=" * 80,
|
||||
ORDER_HINT,
|
||||
f"Feature: {feature_name}",
|
||||
f"Features: {', '.join(t.name for t in targets)}",
|
||||
f"Generated: {datetime.now().isoformat()}",
|
||||
"=" * 80,
|
||||
"",
|
||||
]
|
||||
|
||||
files_to_merge = ordered_markdown_files(target_dir)
|
||||
total_files = 0
|
||||
for target in targets:
|
||||
total_files += merge_one_feature(target, blocks)
|
||||
|
||||
for file_path in files_to_merge:
|
||||
relative_path = file_path.relative_to(target_dir)
|
||||
try:
|
||||
content = file_path.read_text(encoding="utf-8")
|
||||
except Exception as e:
|
||||
print(f"Skipping {file_path}: {e}")
|
||||
continue
|
||||
blocks.append("")
|
||||
blocks.append("=" * 80)
|
||||
blocks.append(f"Features merged: {len(targets)} | Files merged: {total_files}")
|
||||
blocks.append("=" * 80)
|
||||
blocks.append("")
|
||||
|
||||
# Determine section label
|
||||
section_label = str(relative_path)
|
||||
for stage_type, stage_value, label in CANONICAL_MD_STAGES:
|
||||
if label is None:
|
||||
continue
|
||||
if stage_type == "exact" and str(relative_path) == stage_value:
|
||||
section_label = label
|
||||
break
|
||||
if stage_type == "prefix" and str(relative_path).startswith(stage_value):
|
||||
section_label = f"{label} — {relative_path.name}"
|
||||
break
|
||||
merged = "\n".join(blocks)
|
||||
Path(output).write_text(merged, encoding="utf-8")
|
||||
print(f"Created: {output} ({len(targets)} features, {total_files} files)")
|
||||
return output
|
||||
|
||||
content_blocks.append("")
|
||||
content_blocks.append("-" * 60)
|
||||
content_blocks.append(f"## {section_label}")
|
||||
content_blocks.append(f"Source: {relative_path}")
|
||||
content_blocks.append("-" * 60)
|
||||
content_blocks.append("")
|
||||
content_blocks.append(content)
|
||||
content_blocks.append("")
|
||||
|
||||
# Summary
|
||||
content_blocks.append("")
|
||||
content_blocks.append("=" * 80)
|
||||
content_blocks.append(f"Files merged: {len(files_to_merge)}")
|
||||
content_blocks.append("=" * 80)
|
||||
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
|
||||
|
||||
merged = "\n".join(content_blocks)
|
||||
output_filename = output_filename
|
||||
Path(output_filename).write_text(merged, encoding="utf-8")
|
||||
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:]
|
||||
|
||||
print(f"Successfully created: {output_filename} ({len(files_to_merge)} files)")
|
||||
return output_filename
|
||||
result = merge_specs(tokens, output=out_path)
|
||||
return 0 if result else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) < 2:
|
||||
print("Usage: python merge_spec.py <feature_number>")
|
||||
sys.exit(1)
|
||||
|
||||
merge_specs(sys.argv[1])
|
||||
sys.exit(main(sys.argv))
|
||||
|
||||
# #endregion Tooling.Merge.Spec
|
||||
|
||||
|
||||
Reference in New Issue
Block a user