"""erp-card — a build card: the part of a module spec a builder needs to pass its stories, nothing else.
erp-card <module.yaml> [--shared shared.yaml] [--stories FA-16..FA-26,FA-30] [-o card.md]
Keeps (verbatim where it is a contract): every story as one line (id, actor, given/when/then), the workflow transitions
(key, from>to, by, gate, effect, approval, requires), the routes the module owns and the seams it calls (route, request and
response lines verbatim, permission, rule), entities with natural keys and reference tables, master data, jobs,
permissions, config_rows, in-code rules, tests (access, invariants, scenarios), platform constraints, scale, migration, AI
steps, deferred items, and every sap_parity row whose `ours` is not `covered`.
Drops the prose a builder does not code against: summary, facets, actor descriptions, the "As a ..." story sentence,
screens, customer_varies, variation_points, reference, deviation_support, and `covered` parity rows. When --shared is
given, the VARCHAR keys it declares are listed (the L4 rule's inputs).
--stories ID[,ID|ID..ID] (0.9.4) cuts the card to a slice: only those stories, the jobs and seams (owned or called) that name
them, the entities (named by those stories, plus one hop of relationships), permissions and sap_parity rows
those stories or kept seams mention, and scenarios touching them. The shared checklist, cross-cutting sections (workflows, config_rows, platform constraints, scale, AI
steps, deferred, migration, row_visibility, access/invariant tests) and the footer stay. An unknown id is an error.
"""
from __future__ import annotations
import json
import re
import sys
from pathlib import Path
from typing import Optional
import yaml
CHECKLIST = """\
- Self-approval: store the actor's email AND sub; refuse when ANY identity of the approver matches.
- Row visibility: no email must not mean "see everything" - scope by sub; "all" only with read_all.
- Business "today": from the shared business_time (never date.today()/utcnow()); it never goes backwards.
- Concurrency: anything that must not exceed a total takes plain-INSERT claim rows in the same transaction.
- Bounds: every amount/rate/quantity validated (finite, decimals, digits) -> 422 before NUMERIC overflow -> 500.
- Number series: never reuse or fill a gap below the high-water mark; a claim table keyed on the rendered number.
- Migration: a second run writes nothing; a SIGNED run is frozen.
- Issued/posted documents store their computed amounts; reads never recompute.
- Fakes answer exactly what the callee's SPEC route answers (shape and status codes).
- Tool/use-case names exactly as in the spec.
- No SERIAL/sequences; shared VARCHAR keys are never int()'d; DO UPDATE SET only takes EXCLUDED.<col>."""
def _j(v) -> str:
return json.dumps(v, ensure_ascii=False, default=str, separators=(", ", ": "))
def _one(s) -> str:
return " ".join(str(s).split())
def parse_stories(arg: str, known: list) -> list:
"""`FA-16..FA-26,FA-30` -> ids in spec order. A range is by the id's NUMBER (same prefix, a <= n <= b), never by where the
stories sit in the file: specs grow out of order (SO-25 after SO-32), and a positional range silently dropped a slice."""
want, ids = set(), list(known)
def num(i):
m = re.fullmatch(r"(.*?)(\d+)", i)
return (m.group(1), int(m.group(2))) if m else (i, None)
for part in (p.strip() for p in arg.split(",") if p.strip()):
if ".." in part:
a, b = (x.strip() for x in part.split("..", 1))
if a not in ids or b not in ids:
raise SystemExit(f"erp-card: range {part}: unknown story id")
(pa, na), (pb, nb) = num(a), num(b)
if na is None or nb is None or pa != pb or na > nb:
raise SystemExit(f"erp-card: range {part}: give two ids of one prefix, low..high")
want.update(i for i in ids if num(i)[0] == pa and num(i)[1] is not None and na <= num(i)[1] <= nb)
elif part in ids:
want.add(part)
else:
raise SystemExit(f"erp-card: unknown story id {part}")
return [i for i in ids if i in want]
def _mentions(blob: str, word: str) -> bool:
return bool(word) and re.search(r"(?<![A-Za-z0-9_])" + re.escape(word) + r"(?![A-Za-z0-9_])", blob) is not None
def slice_spec(spec: dict, ids: list) -> dict:
"""The part of `spec` the stories `ids` touch (see the module docstring)."""
keep = set(ids)
out = dict(spec)
out["stories"] = [s for s in spec.get("stories", []) if s["id"] in keep]
out["jobs"] = [j for j in spec.get("jobs", []) if j.get("story") in keep]
out["seams"] = [x for x in spec.get("seams", []) if x.get("story") in keep]
text = _j([out["stories"], out["jobs"], out["seams"]])
low = text.lower()
ents = [e for e in spec.get("entities", []) if e["id"] in low or e["id"].replace("_", " ") in low or _mentions(text, e.get("reference_table") or "")]
rel = _j([e.get("relationships") for e in ents]) # one hop: what a kept entity relates to
out["entities"] = [e for e in spec.get("entities", []) if e in ents or _mentions(rel, e["id"]) or _mentions(rel, e["id"].replace("_", " "))]
# config_rows are small and cross-cutting (settings, rates): they stay whole
# transitions are the state machine and stories name them in prose ("proposes a retirement"), so workflows stay whole
held = {p for a in spec.get("actors", []) if any(s["actor"] == a["id"] for s in out["stories"]) for p in a.get("permissions", []) or []}
perms = dict(spec.get("permissions", {}))
perms["catalog"] = [c for c in perms.get("catalog", []) if c in held or _mentions(text, c)]
out["permissions"] = perms
out["sap_parity"] = [r for r in spec.get("sap_parity", []) if any(_mentions(_j(r), i) for i in keep)]
t = dict(spec.get("tests", {}))
t["scenarios"] = [sc for sc in t.get("scenarios", []) or [] if keep & set(sc.get("stories", []))]
out["tests"] = t
return out
def _strs(items) -> list:
"""List items as text: older specs mix plain strings with small mappings ({name: note})."""
return [i if isinstance(i, str) else _j(i) for i in (items or [])]
def build(spec: dict, shared: Optional[dict] = None, stories: Optional[str] = None) -> str:
sliced = None
if stories:
sliced = parse_stories(stories, [s["id"] for s in spec.get("stories", [])])
spec = slice_spec(spec, sliced)
me = spec["id"]
o = [f"# Build card: {me} (v{spec.get('version')}, {spec.get('maturity')})", ""]
if sliced:
o += [f"SLICE: stories {', '.join(sliced)} only; seams, jobs, entities, permissions, parity rows and scenarios are cut to what they "
"reference. Anything a story needs that is not here is in the full module.yaml.", ""]
if spec.get("permissions", {}).get("namespace"):
o.append(f"namespace `{spec['permissions']['namespace']}`; actors: " + ", ".join(a["id"] for a in spec.get("actors", [])))
o += ["", "## Stories"]
for s in spec.get("stories", []):
extra = "".join(f" [{k}: {_one(s[k])}]" for k in ("trigger", "owner", "note") if s.get(k))
o.append(f"- {s.get('id', '-')} ({s.get('actor', '-')}): GIVEN {_one(s.get('given', '-'))} WHEN {_one(s.get('when', '-'))} THEN {_one(s.get('then', '-'))}{extra}")
o += ["", "## Workflows"]
for w in spec.get("workflows", []):
o.append(f"### {w.get('entity', '-')} ({w.get('process', '')}): {', '.join(w.get('states', '-'))}; terminal: {', '.join(_strs(w.get('terminal', [])))}")
for t in w["transitions"]:
bits = [f"{t.get('key') or t.get('name') or t.get('id', '-')}: {t.get('from', '-')} > {t.get('to', '-')} by {t.get('by', '-')}"]
for k in ("gate", "effect", "requires"):
if t.get(k):
bits.append(f"{k}: {_one(t[k]) if not isinstance(t[k], (list, dict)) else _j(t[k])}")
if t.get("approval"):
bits.append("APPROVAL")
o.append("- " + "; ".join(bits))
if w.get("note"):
o.append(f" note: {_one(w.get('note', '-'))}")
seams = spec.get("seams", [])
me = spec.get("id")
def inbound(s) -> bool:
# a spec that predates `direction:` says who owns the route: owned here = inbound
return s["direction"] == "inbound" if s.get("direction") else s.get("owner") == me
for title, pick in (("Routes this module owns (inbound seams)", inbound),
("Seams this module calls (outbound; request lines are the contract)", lambda s: not inbound(s))):
o += ["", f"## {title}"]
for s in seams:
if not pick(s):
continue
o.append(f"- {s.get('story', '-')} {s.get('route', '-')} (with {s.get('with', '-')}, owner {s.get('owner', '-')}, perm {s.get('permission', '-')})")
o.append(f" request: {_one(s.get('request', '-'))}")
o.append(f" response: {_one(s.get('response', '-'))}")
o.append(f" rule: {_one(s.get('rule', '-'))}")
for k in ("status", "mismatch"):
if s.get(k):
o.append(f" {k}: {_one(s[k])}")
o += ["", "## Entities (natural key; table)"]
for e in spec.get("entities", []):
o.append(f"- {e.get('id', '-')}: key {_one(e.get('natural_key', '-'))}; table {e.get('reference_table')}; master {e.get('master_data')}; "
f"attrs {', '.join(_strs(e.get('key_attributes', [])))}; rel {'; '.join(_strs(e.get('relationships', [])))}")
if spec.get("master_data"):
o += ["", "## Master data", _j(spec["master_data"])]
o += ["", "## Jobs"]
for j in spec.get("jobs", []):
o.append(f"- {j.get('story', '-')}: {_one(j.get('trigger', '-'))} -> {_one(j.get('runs', '-'))}; idempotent: {_one(j.get('idempotent', '-'))}; where: {_j(j.get('where_options'))}")
p = spec.get("permissions", {})
o += ["", "## Permissions", "catalog: " + ", ".join(p.get("catalog", [])), "machine: " + _j(p.get("machine", [])),
"row_visibility: " + _j(p.get("row_visibility", {}))]
o += ["", "## config_rows"]
for c in spec.get("config_rows", []):
o.append(f"- table {c.get('table', '-')}: {_one(c.get('changes', '-'))}")
for s in c.get("settings", []):
o.append(f" - {s.get('key', '-')} ({s.get('type', '-')}, default {s.get('default')!r}): {_one(s.get('changes', ''))}")
if spec.get("in_code_not_rows"):
o += ["", "## In code, not rows"] + [f"- {_one(x)}" for x in spec["in_code_not_rows"]]
t = spec.get("tests", {})
o += ["", "## Tests"]
for k in ("acceptance",):
if t.get(k):
o.append(f"- {k}: {_one(t[k])}")
for k in ("access", "invariants"):
for x in t.get(k, []) or []:
o.append(f"- {k}: {_one(x)}")
for sc in t.get("scenarios", []) or []:
o.append(f"- scenario {sc.get('id')} {sc.get('name')}: stories {','.join(sc.get('stories', []))}; "
+ "; ".join(f"{k} {_one(v)}" for k, v in sc.items() if k not in ("id", "name", "stories")))
for key, title in (("platform_constraints", "Platform constraints"), ("scale", "Scale"), ("ai_steps", "AI steps"),
("deferred", "Deferred (not in this build)"), ("migration", "Migration")):
if spec.get(key):
o += ["", f"## {title}", _j(spec[key])]
gaps = [r for r in spec.get("sap_parity", []) if r.get("ours") != "covered"]
o += ["", "## SAP parity: ours != covered"] + [f"- {r.get('sap', '-')}: {r.get('ours', '-')}" + "".join(f"; {k}: {_one(v)}" for k, v in r.items() if k not in ("sap", "ours"))
for r in gaps] if gaps else ["", "## SAP parity: ours != covered", "- (none)"]
if shared:
vc = sorted(k for k, v in shared.get("keys", {}).items() if str(v.get("type", "")).upper().startswith("VARCHAR"))
bg = sorted(k for k, v in shared.get("keys", {}).items() if str(v.get("type", "")).upper().startswith("BIGINT"))
o += ["", "## Shared keys", "VARCHAR (never int()): " + ", ".join(vc), "BIGINT: " + ", ".join(bg)]
o += ["", "## Bug-class checklist", CHECKLIST, "",
"Before you report done: run `erp-lint <brick-dir> --method-dir <erp-method>` and enable `erp_platform.contract.checked_seams` "
"in tests/conftest.py (every Fake answers its callee's spec). Cards drop prose, not contract: when in doubt the full "
"module.yaml wins.", ""]
return "\n".join(o)
def main(argv: Optional[list] = None) -> int:
a = list(sys.argv[1:] if argv is None else argv)
shared = out = stories = None
for flag in ("--shared", "-o", "--stories"):
if flag in a:
i = a.index(flag)
val = a[i + 1]
del a[i:i + 2]
if flag == "-o":
out = val
elif flag == "--stories":
stories = val
else:
shared = yaml.safe_load(Path(val).read_text())
if len(a) != 1:
print("usage: erp-card <module.yaml> [--shared shared.yaml] [--stories ID,ID..ID] [-o card.md]", file=sys.stderr)
return 2
text = build(yaml.safe_load(Path(a[0]).read_text()), shared, stories)
if out:
Path(out).write_text(text)
else:
sys.stdout.write(text)
return 0
if __name__ == "__main__":
sys.exit(main())