#!/usr/bin/env python3 """Create a fail-closed, idempotent CASAN project shell from versioned templates.""" from __future__ import annotations import argparse import json import os import re import shutil import stat import sys import tempfile from pathlib import Path HERE = Path(__file__).resolve().parent SOURCE_ROOT = HERE.parent.parent SLUG = re.compile(r"^[a-z][a-z0-9-]{1,62}$") FEATURE = re.compile(r"^[0-9]{3}-[a-z0-9-]+$") MODULE = re.compile(r"^MOD-[0-9]{2,}$") TEXT_SUFFIXES = {".json", ".md", ".ts", ".tsx", ".js", ".mjs", ".css", ".html", ".yml", ".yaml", ".conf", ".txt"} class ScaffoldError(RuntimeError): pass def atomic_write(path: Path, content: bytes, mode: int = 0o644) -> str: path.parent.mkdir(parents=True, exist_ok=True) if path.is_symlink(): raise ScaffoldError(f"refusing to overwrite symlink: {path}") if path.exists(): current = path.read_bytes() if current == content: return "unchanged" raise ScaffoldError(f"existing file differs; no files were overwritten: {path}") with tempfile.NamedTemporaryFile(dir=path.parent, delete=False) as handle: handle.write(content) temporary = Path(handle.name) os.chmod(temporary, mode) os.replace(temporary, path) return "created" def rendered(content: bytes, replacements: dict[str, str], suffix: str) -> bytes: if suffix not in TEXT_SUFFIXES and suffix not in {"", ".gitignore"}: return content text = content.decode("utf-8") for token, value in replacements.items(): text = text.replace(token, value) return text.encode("utf-8") def copy_template(template: Path, target: Path, replacements: dict[str, str]) -> tuple[int, int]: created = unchanged = 0 for source in sorted(template.rglob("*")): if not source.is_file(): continue relative = source.relative_to(template) parts = [replacements.get("__PROJECT_SLUG__", "project") if part == "project" else part for part in relative.parts] destination = target.joinpath(*parts) status = atomic_write(destination, rendered(source.read_bytes(), replacements, source.suffix)) created += status == "created" unchanged += status == "unchanged" return created, unchanged def copy_domain_pack(target: Path, slug: str, name: str) -> tuple[int, int]: source = HERE / "templates" / "domain-pack" destination = target / "apps" / slug / "domain" created = unchanged = 0 for path in sorted(source.rglob("*")): if not path.is_file() or path.name == "traceability-map.example.json" or path.name.endswith(".example.jsonl"): continue relative = path.relative_to(source) content = path.read_bytes() if path.suffix in TEXT_SUFFIXES: text = content.decode("utf-8").replace("__PROJECT_SLUG__", slug).replace("__PROJECT_NAME__", name) content = text.encode("utf-8") if path.name == "domain-pack.yaml": text = content.decode("utf-8").replace("id: custom", f"id: {slug}").replace('name: "Custom domain"', f'name: "{name}"') content = text.encode("utf-8") status = atomic_write(destination / relative, content) created += status == "created" unchanged += status == "unchanged" return created, unchanged def install_harness(target: Path) -> tuple[int, int]: created = unchanged = 0 harness_source = SOURCE_ROOT / "packages" / "casan-harness" for source in sorted(harness_source.rglob("*")): if not source.is_file() or "__pycache__" in source.parts or source.suffix == ".pyc": continue relative = source.relative_to(harness_source) destination = target / "packages" / "casan-harness" / relative # The target registry is adoption state, not immutable harness code. Preserve it # after the first install so repeated scaffolds and upgrades remain idempotent. if relative.as_posix() == "level5/project-registry.json" and destination.exists(): unchanged += 1 continue mode = stat.S_IMODE(source.stat().st_mode) status = atomic_write(destination, source.read_bytes(), mode) created += status == "created" unchanged += status == "unchanged" status = atomic_write(target / "bin" / "casan", (SOURCE_ROOT / "bin" / "casan").read_bytes(), 0o755) created += status == "created" unchanged += status == "unchanged" for name in ("casan-project.mjs", "casan-step.mjs", "run-casan-pipeline.mjs", "casan-log.mjs"): status = atomic_write(target / "scripts" / name, (SOURCE_ROOT / "scripts" / name).read_bytes()) created += status == "created" unchanged += status == "unchanged" return created, unchanged def register_project(target: Path, slug: str, name: str) -> None: registry_path = target / "packages" / "casan-harness" / "level5" / "project-registry.json" data = json.loads(registry_path.read_text(encoding="utf-8")) harness_version = next((item.get("harness_version") for item in data.get("projects", []) if item.get("harness_version")), "1.0.0") # A shipped harness may carry source-hub examples. Never register dangling # projects in the consumer repository; preserve only domain packs that exist. data["projects"] = [ item for item in data.get("projects", []) if (target / str(item.get("domain_root", "__missing__"))).is_dir() ] desired = { "project_id": slug, "domain": name, "domain_root": f"apps/{slug}/domain", "manifest": f"apps/{slug}/domain/project.manifest.json", "context_roots": [f"apps/{slug}"], "harness_package": "fpt-casan-sdd-harness", "harness_version": harness_version, "status": "active", } existing = next((item for item in data["projects"] if item.get("project_id") == slug), None) if existing and existing != desired: raise ScaffoldError(f"registry entry already exists with different configuration: {slug}") if not existing: data["projects"].append(desired) registry_path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") def scaffold(args: argparse.Namespace) -> dict: if not SLUG.fullmatch(args.project): raise ScaffoldError("--project must be a lowercase slug (2-63 characters)") if not 2 <= len(args.name) <= 100 or not re.fullmatch(r"[\w][\w .&()'_-]+", args.name, re.UNICODE): raise ScaffoldError("--name must be 2-100 printable letters/numbers with safe punctuation") feature_id = args.feature_id or f"001-{args.project}-app" if not FEATURE.fullmatch(feature_id): raise ScaffoldError("--feature-id must match NNN-slug") if not MODULE.fullmatch(args.module_id): raise ScaffoldError("--module-id must match MOD-NN") target = Path(args.target).expanduser().resolve() if target == Path('/') or target == Path.home(): raise ScaffoldError("refusing broad target path") target.mkdir(parents=True, exist_ok=True) if target.is_symlink(): raise ScaffoldError("target may not be a symlink") replacements = { "__PROJECT_SLUG__": args.project, "__PROJECT_NAME__": args.name, "__FEATURE_ID__": feature_id, "__MODULE_ID__": args.module_id, } template = HERE / "templates" / "project-shell" / args.template if not template.is_dir(): raise ScaffoldError(f"unknown template: {args.template}") created, unchanged = copy_template(template, target, replacements) domain_created, domain_unchanged = copy_domain_pack(target, args.project, args.name) created += domain_created unchanged += domain_unchanged profile = HERE / "quality-profiles" / "enterprise-web-v1.json" schema = HERE / "schemas" / "project-manifest.schema.json" for destination, source in ( (target / "config/casan/quality-profiles/enterprise-web-v1.json", profile), (target / "config/casan/schemas/project-manifest.schema.json", schema), ): status = atomic_write(destination, source.read_bytes()) created += status == "created" unchanged += status == "unchanged" if args.with_harness: harness_created, harness_unchanged = install_harness(target) created += harness_created unchanged += harness_unchanged register_project(target, args.project, args.name) return {"status": "ok", "target": str(target), "project_id": args.project, "template": args.template, "created": created, "unchanged": unchanged} def parser() -> argparse.ArgumentParser: value = argparse.ArgumentParser(description=__doc__) value.add_argument("--target", required=True) value.add_argument("--project", required=True) value.add_argument("--name", required=True) value.add_argument("--feature-id") value.add_argument("--module-id", default="MOD-01") value.add_argument("--template", default="nestjs-react", choices=["nestjs-react"]) value.add_argument("--with-harness", action="store_true") return value def main() -> int: try: result = scaffold(parser().parse_args()) print(json.dumps(result, ensure_ascii=False)) return 0 except (OSError, ValueError, ScaffoldError, json.JSONDecodeError) as error: print(f"CASAN_PROJECT_SCAFFOLD_FAILED: {error}", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())