109 lines
5.1 KiB
Python
109 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
PASCAL = re.compile(r"^[A-Z][A-Za-z0-9]*$")
|
|
REQ = re.compile(r"^REQ-[A-Z0-9-]+$")
|
|
ROUTE = re.compile(r"^[a-z0-9][a-z0-9_/{}/.-]*$")
|
|
ALLOWED_METHODS = {"Get", "Post", "Put", "Patch", "Delete"}
|
|
ALLOWED_ROLES = {"Admin", "Advisor", "Approver", "Auditor", "Compliance", "DataOps", "Quant", "Risk", "System"}
|
|
|
|
@dataclass(frozen=True)
|
|
class Spec:
|
|
module: str
|
|
slice_name: str
|
|
requirement_id: str
|
|
http_method: str
|
|
route: str
|
|
role: str
|
|
frontend_feature: str
|
|
|
|
|
|
def validate(spec: Spec) -> None:
|
|
errors: list[str] = []
|
|
if not PASCAL.fullmatch(spec.module): errors.append("module must be PascalCase")
|
|
if not PASCAL.fullmatch(spec.slice_name): errors.append("slice must be PascalCase")
|
|
if not REQ.fullmatch(spec.requirement_id): errors.append("requirement must match REQ-[A-Z0-9-]+")
|
|
if spec.http_method not in ALLOWED_METHODS: errors.append(f"method must be one of {sorted(ALLOWED_METHODS)}")
|
|
if not ROUTE.fullmatch(spec.route) or spec.route.startswith("/"): errors.append("route must be relative lower-case API path")
|
|
if spec.role not in ALLOWED_ROLES: errors.append(f"role must be one of {sorted(ALLOWED_ROLES)}")
|
|
if not re.fullmatch(r"[a-z][a-z0-9-]*", spec.frontend_feature): errors.append("feature must be kebab-case")
|
|
if errors: raise ValueError("; ".join(errors))
|
|
|
|
|
|
def render(template: Path, values: dict[str, str]) -> str:
|
|
text = template.read_text(encoding="utf-8")
|
|
for key, value in values.items(): text = text.replace(f"__{key}__", value)
|
|
unresolved = sorted(set(re.findall(r"__[A-Z0-9_]+__", text)))
|
|
if unresolved: raise ValueError(f"unresolved template tokens in {template.name}: {unresolved}")
|
|
return text
|
|
|
|
|
|
def plan(spec: Spec) -> dict[Path, str]:
|
|
validate(spec)
|
|
values = {
|
|
"MODULE": f"KArtSell.Modules.{spec.module}",
|
|
"SLICE": spec.slice_name,
|
|
"REQUIREMENT_ID": spec.requirement_id,
|
|
"HTTP_METHOD": spec.http_method,
|
|
"ROUTE": spec.route,
|
|
"ROLE": spec.role,
|
|
"FEATURE": spec.frontend_feature,
|
|
}
|
|
backend_dir = ROOT / "src" / f"KArtSell.Modules.{spec.module}" / "Features" / spec.slice_name
|
|
frontend_dir = ROOT / "frontend" / "src" / "features" / spec.frontend_feature
|
|
files: dict[Path, str] = {}
|
|
dotnet = ROOT / "templates" / "dotnet" / "VerticalSlice"
|
|
for name in ["Endpoint.cs", "Request.cs", "Response.cs", "Handler.cs", "Validator.cs", "Sql.cs", "README.md"]:
|
|
files[backend_dir / name] = render(dotnet / f"{name}.template", values)
|
|
vue = ROOT / "templates" / "vue" / "Feature"
|
|
for template_name, output_name in [
|
|
("api.ts.template", "api.ts"), ("queries.ts.template", "queries.ts"),
|
|
("schema.ts.template", "schema.ts"), ("page.vue.template", "pages/FeaturePage.vue"),
|
|
("schema.spec.ts.template", "tests/schema.spec.ts")]:
|
|
files[frontend_dir / output_name] = render(vue / template_name, values)
|
|
return files
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Create a fail-closed Vertical Slice scaffold from reviewed templates.")
|
|
parser.add_argument("--module", required=True)
|
|
parser.add_argument("--slice", dest="slice_name", required=True)
|
|
parser.add_argument("--requirement", required=True)
|
|
parser.add_argument("--method", default="Post")
|
|
parser.add_argument("--route", required=True)
|
|
parser.add_argument("--role", required=True)
|
|
parser.add_argument("--feature", required=True)
|
|
parser.add_argument("--write", action="store_true", help="write files; otherwise print a deterministic plan")
|
|
args = parser.parse_args()
|
|
try:
|
|
spec = Spec(args.module, args.slice_name, args.requirement, args.method, args.route, args.role, args.feature)
|
|
files = plan(spec)
|
|
collisions = [str(path.relative_to(ROOT)) for path in files if path.exists()]
|
|
if collisions: raise FileExistsError("refusing to overwrite existing files: " + ", ".join(collisions))
|
|
manifest = {str(path.relative_to(ROOT)): content for path, content in sorted(files.items())}
|
|
if not args.write:
|
|
print(json.dumps({"status": "DRY_RUN", "files": list(manifest)}, indent=2, ensure_ascii=False))
|
|
return 0
|
|
for path, content in files.items():
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(content, encoding="utf-8")
|
|
marker = ROOT / "docs" / "generated" / f"{spec.requirement_id}_{spec.slice_name}.json"
|
|
marker.parent.mkdir(parents=True, exist_ok=True)
|
|
marker.write_text(json.dumps({"status":"SCAFFOLD_ONLY","spec":spec.__dict__,"files":list(manifest)}, ensure_ascii=False, indent=2)+"\n", encoding="utf-8")
|
|
print(json.dumps({"status": "WRITTEN", "files": list(manifest), "marker": str(marker.relative_to(ROOT))}, indent=2, ensure_ascii=False))
|
|
return 0
|
|
except (ValueError, FileExistsError) as exc:
|
|
print(f"ERROR: {exc}", file=sys.stderr)
|
|
return 2
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|