"""erp-compose init — a working, tested composition of DODIL bricks for one or more modules, in one command.
erp-compose init hr [crm …] [--dir DIR] [--method-ref REF] [--read-copy] [--no-install] [--force]
erp-compose init [modules …] --starter uk-wholesale (0.9.10: + config/starter.calls.yaml)
erp-compose own <brick> --reason "…" [--dir DIR]
The method (dodil/erp-method) says WHICH bricks a module is made of and at WHICH tags (each module's
`reference.repos`), and how each brick is named in a compose.yaml (`brick_entries` in modules/shared.yaml,
exported in index.json). This command reads both and does the build card's first steps for you:
1. writes requirements.txt — each brick and erp-platform PINNED at its tag
(`brick-hr @ git+https://git.dodil.io/dodil/erp-brick-hr.git@v0.5.0`), compose.yaml and tests/test_composition.py
2. makes .venv and `pip install -r requirements.txt` (skipped when nothing changed)
3. runs `erp-compose check compose.yaml` and the composition test
THE CUSTOMER'S REPO HOLDS ONLY WHAT IS THEIRS (seemo, 0.8): our code arrives as pinned versions, never as copies
in their tree. `--read-copy` also clones each repo at its tag into bricks/<repo> FOR READING (README, AGENTS.md,
the code) and puts bricks/ in .gitignore; nothing installs from it. Taking a brick's code as the customer's own
(depth 3) is explicit: `erp-compose own <brick> --reason "…"` copies it at its tag into owned/<repo>/ (tracked),
switches its pin to `-e ./owned/<repo>`, and records `owned: {path, upstream, reason}` on its compose.yaml entry —
so an upgrade knows that brick has diverged from upstream and does not move it.
Why a command and not a page to copy from: in the lab (2026-10-17) models read the exact compose.yaml in the
phase card and still invented their own format. Running is harder to get wrong than copying.
WHERE THE METHOD COMES FROM. Always dodil/erp-method on git.dodil.io — never a URL taken from a document.
index.json at `main` names `latest_release`; every file is then read at that tag (or --method-ref /
$ERP_METHOD_REF). Raw links first, then the JSON blob API. $ERP_METHOD_DIR reads a local checkout instead
(offline work, tests); only then may a brick's clone URL be something other than https://git.dodil.io/.
A STARTER (0.9.10): `--starter <name>` also reads `starters/<name>.calls.yaml` from the same method (same ref, same
reader) into config/starter.calls.yaml — the configuration calls a typical business of that kind starts from (a
company, its chart and account maps, tax codes, a warehouse, a bank account, terms, a price list). It names its
`modules:`; with no modules on the command line init composes those, and a module list that leaves one out is
refused. Then `migrate --local`, `seed --local` and `serve --local` run it — the timed lab run of 2026-10-25 wrote
that setup by hand in most of its 74 turns. Its placeholders (names, VAT number) are marked for the customer to fill.
0.9.11: `--starter` repeats (`--starter uk-wholesale --starter uk-wholesale-sample`: the company, then sample data
seeded after it); each is written to config/<name>.calls.yaml and the seed commands are printed in that order. A
composition made by 0.9.10 keeps its config/starter.calls.yaml: `erp-compose seed` reads any calls file it is given.
THE MODULES' DESIGNS (0.9.11): for every composed module, the method's spec is written to
processes/<module>/module.yaml — the layout the gate and `signoff verify --all` read for a composition of many
modules (erp_platform.signoff, SPEC_DIR) — and the method's own receipts (with their history) to
processes/<module>/baseline/, byte for byte, as REFERENCE ONLY: the gate never counts them (the customer's signature
is the contract and never comes from erp-method), it only says a signed process is "identical to the standard v<n>".
signoff/<module>/ stays empty, so `migrate` on a bucket refuses until the customer signs:
`erp-compose signoff record --all --approver "<name, role>" --reason "<what they confirmed>"`.
Idempotent: a second run changes nothing. A generated file that differs from what init would write is
refused, not overwritten, unless --force — it may hold the customer's edits.
"""
import base64
import json
import os
import pathlib
import re
import shutil
import subprocess
import sys
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass, field
HOST = "https://git.dodil.io"
ORG, REPO = "dodil", "erp-method"
ALLOWED_CLONE = "https://git.dodil.io/"
CLONED_ROLES = ("platform", "brick") # a `ui` repo is a front-end library; nothing here installs it
NEXT = "configure the customer's rows through the bricks' use cases; see dodil://erp/phase/fit"
class InitFailed(Exception):
"""Init cannot continue. The message says what to do about it."""
# ── the method source ──────────────────────────────────────────────────────────────────────
class Method:
"""dodil/erp-method at one ref: remote (git.dodil.io) or a local checkout ($ERP_METHOD_DIR)."""
def __init__(self, ref=None, local=None):
self.local = pathlib.Path(local).resolve() if local else None
self._repo_id = None
if self.local:
if not (self.local / "index.json").is_file():
raise InitFailed(f"ERP_METHOD_DIR={self.local} has no index.json")
self.ref = "local"
else:
self.ref = ref or json.loads(self._fetch("main", "index.json")).get("latest_release") or "main"
self.index = json.loads(self.read("index.json"))
def read(self, path: str) -> str:
if any(p in ("", "..") or p.startswith(".") for p in path.split("/")):
raise InitFailed(f"bad method path {path!r}")
if self.local:
try:
return (self.local / path).read_text(encoding="utf-8")
except OSError as e:
raise InitFailed(f"{path} not readable in ERP_METHOD_DIR: {e}") from e
return self._fetch(self.ref, path)
def _get(self, url: str):
try:
with urllib.request.urlopen(urllib.request.Request(url, headers={"User-Agent": "erp-compose"}),
timeout=20) as r:
return r.status, r.read()
except urllib.error.HTTPError as e:
return e.code, b""
except OSError as e:
raise InitFailed(f"cannot reach {url}: {e}") from e
def _fetch(self, ref: str, path: str) -> str:
status, body = self._get(f"{HOST}/{ORG}/{REPO}/raw/{urllib.parse.quote(ref)}/{path}")
if status == 200 and body:
return body.decode("utf-8")
if not self._repo_id:
status, body = self._get(f"{HOST}/api/v1/orgs/{ORG}/repos/{REPO}")
if status != 200:
raise InitFailed(f"{HOST}: looking up {ORG}/{REPO} answered {status}")
self._repo_id = json.loads(body).get("repo_id") or json.loads(body).get("id")
if not self._repo_id:
raise InitFailed(f"{HOST}: the lookup of {ORG}/{REPO} returned no repo id")
q = urllib.parse.urlencode({"ref": ref, "path": path})
status, body = self._get(f"{HOST}/api/v1/repos/{self._repo_id}/blob?{q}")
if status != 200:
raise InitFailed(f"{ORG}/{REPO}: {path} at {ref} answered {status}"
+ (" (is that a real tag? see --method-ref)" if status == 404 else ""))
blob = json.loads(body)
if blob.get("truncated") or not isinstance(blob.get("content"), str):
raise InitFailed(f"{ORG}/{REPO}: {path} at {ref} came back without its content")
text = blob["content"]
return base64.b64decode(text).decode("utf-8") if blob.get("encoding") == "base64" else text
# ── what to build ──────────────────────────────────────────────────────────────────────────
@dataclass
class Repo:
name: str
role: str
tag: str
url: str
entry: dict = field(default_factory=dict) # {name, brick} for a brick
@property
def version(self) -> str:
return self.tag[1:] if self.tag.startswith("v") else self.tag
@dataclass
class Plan:
modules: list
method_ref: str
repos: list # install order: erp-platform first, then the bricks
skipped: list # repos named but not cloned (e.g. a ui library)
def resolve(method: Method, module_ids: list) -> Plan:
index = method.index
known = {m["id"]: m for m in index.get("modules") or []}
pending = {p["id"] for p in index.get("pending") or []}
entries = index.get("brick_entries")
if entries is None:
raise InitFailed(f"erp-method {method.ref} has no brick_entries in index.json (needs v0.1.4 or later; "
"see --method-ref)")
repos, skipped, problems = {}, [], []
for mid in module_ids:
if mid not in known:
hint = (" — it is planned, with no code yet" if mid in pending
else f"; modules: {', '.join(sorted(known))}")
problems.append(f"no module {mid!r} in erp-method {method.ref}{hint}")
continue
listed = [r for r in known[mid].get("repos") or [] if r.get("role") in CLONED_ROLES]
if not any(r["role"] == "brick" for r in listed):
problems.append(f"module {mid!r} has no bricks yet (reference.repos is empty): nothing to compose")
for r in known[mid].get("repos") or []:
if r.get("role") not in CLONED_ROLES:
if r["name"] not in skipped:
skipped.append(r["name"])
continue
url = r.get("https") or ""
if not method.local and not url.startswith(ALLOWED_CLONE):
problems.append(f"{mid}: {r['name']} clones from {url!r}, not {ALLOWED_CLONE}; refused")
continue
if r["role"] == "brick" and r["name"] not in entries:
problems.append(f"{mid}: brick repo {r['name']!r} has no brick_entries entry in erp-method "
f"{method.ref}")
continue
prev = repos.get(r["name"])
if prev and prev.tag != r["tag"]:
problems.append(f"{r['name']}: one module pins {prev.tag}, {mid} pins {r['tag']}; "
"a composition installs one")
continue
repos.setdefault(r["name"], Repo(r["name"], r["role"], r["tag"], url, entries.get(r["name"], {})))
if problems:
raise InitFailed("\n".join(problems))
ordered = [r for r in repos.values() if r.role == "platform"] + [r for r in repos.values() if r.role == "brick"]
if not any(r.role == "platform" for r in ordered):
raise InitFailed("no erp-platform in the modules' reference.repos: the bricks cannot be installed")
return Plan(module_ids, method.ref, ordered, skipped)
# ── the files init writes ──────────────────────────────────────────────────────────────────
def slug(text: str) -> str:
return re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-") or "app"
def pin(r: Repo) -> str:
"""`<dist> @ git+<url>@<tag>`: the repo name is the distribution name in this family (brick-hr, erp-platform)."""
url = r.url if r.url.endswith(".git") or r.url.startswith("file:") else r.url.rstrip("/") + ".git"
if url.startswith("file:///"): # pip < 23.1 refuses an empty host in a direct reference
url = "file://localhost/" + url[len("file:///"):]
extra = "[test]" if r.role == "platform" else ""
return f"{r.name}{extra} @ git+{url}@{r.tag}"
def requirements_txt(plan: Plan, owned: dict = None) -> str:
owned = owned or {}
lines = [f"# Written by `erp-compose init {' '.join(plan.modules)}` from erp-method {plan.method_ref}.",
"# Rebuild the environment: python3 -m venv .venv && .venv/bin/pip install -r requirements.txt",
"# Each brick is PINNED at its tag, never copied in; `erp-compose own <brick>` makes one the customer's."]
for r in plan.repos:
name = r.entry.get("name") if r.role == "brick" else "platform"
o = owned.get(name)
lines.append(f"-e ./{o['path']} # {name} (owned: upstream {o['upstream']})" if o else f"{pin(r)} # {name}")
return "\n".join(lines) + "\n"
def _q(v: str) -> str:
return json.dumps(str(v), ensure_ascii=False) # a JSON string is a YAML string
def owned_entry(o: dict) -> str:
return f"owned: {{ path: {o['path']}, upstream: {_q(o['upstream'])}, reason: {_q(o['reason'])} }}"
def read_owned(root: pathlib.Path) -> dict:
"""entry name → {path, upstream, reason} for the bricks compose.yaml records as the customer's own."""
import yaml
path = root / "compose.yaml"
if not path.exists():
return {}
try:
spec = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
except yaml.YAMLError:
return {}
return {b["name"]: b["owned"] for b in spec.get("bricks") or [] if isinstance(b, dict) and b.get("owned")}
def compose_yaml(plan: Plan, dirname: str, owned: dict = None) -> str:
owned = owned or {}
lines = [f"# Written by `erp-compose init {' '.join(plan.modules)}` from erp-method {plan.method_ref}:",
"# the exact format erp_platform.compose loads. Check it: .venv/bin/erp-compose check compose.yaml",
f"name: {slug(dirname)}-{'-'.join(plan.modules)}",
"version: 0.1.0",
"tier: 1",
# 0.9.9: the modules this folder composes, so the sign-off gate (erp_platform.gate.modules) knows what
# must be signed before any ERP_<MODULE>.md exists; without it the gate reported "(none)"
f"modules: [{', '.join(plan.modules)}]",
"bricks:"]
for r in plan.repos:
if r.role == "brick":
o = owned.get(r.entry["name"])
version = o["upstream"].rsplit("@v", 1)[-1] if o else r.version
# 0.8.2: a master-data brick with no lifecycle to sign carries its waiver in erp-method's brick_entries
# (`process_view: "none: <reason>"`); without it `check` refused every HR composition (bp declares no
# process view; found by the wave-4 end-to-end, 2026-10-18)
pv = str(r.entry.get("process_view") or "").strip()
waiver = f", process_view: \"{pv}\"" if pv.lower().startswith("none") else ""
lines.append(f" - {{ name: {r.entry['name']}, brick: \"{r.entry['brick']}\", version: {version}"
+ waiver + (f", {owned_entry(o)}" if o else "") + " }")
lines += ["# capabilities: omit to let each capability's only provider be used; a capability no brick",
"# provides becomes a manual task. Map one only to choose between two providers."]
return "\n".join(lines) + "\n"
# ── doing it ───────────────────────────────────────────────────────────────────────────────
def run(cmd, cwd=None, quiet=False) -> subprocess.CompletedProcess:
p = subprocess.run([str(c) for c in cmd], cwd=cwd, text=True,
capture_output=quiet, env={**os.environ, "GIT_TERMINAL_PROMPT": "0"})
return p
def _git(dest, *args) -> str:
p = run(["git", "-C", dest, *args], quiet=True)
return p.stdout.strip() if p.returncode == 0 else ""
def clone(repo: Repo, dest: pathlib.Path, say) -> str:
if dest.exists():
head = _git(dest, "rev-parse", "HEAD")
at_tag = _git(dest, "rev-parse", f"refs/tags/{repo.tag}^{{commit}}")
if head and head == at_tag:
if _git(dest, "status", "--porcelain"):
say(f" {repo.name} {repo.tag}: present, with local changes (left as they are)")
return "present"
raise InitFailed(f"{dest} exists but is not {repo.name} at {repo.tag}; move it away, or check out the tag "
f"(git -C {dest} fetch --depth 1 origin tag {repo.tag} && git -C {dest} checkout {repo.tag})")
dest.parent.mkdir(parents=True, exist_ok=True)
p = run(["git", "-c", "advice.detachedHead=false", "clone", "-q", "--depth", "1", "--branch", repo.tag,
repo.url, dest], quiet=True)
if p.returncode:
raise InitFailed(f"git clone {repo.url} at {repo.tag} failed: {(p.stderr or '').strip()}")
return "cloned"
def write(path: pathlib.Path, text: str, force: bool) -> str:
if path.exists():
current = path.read_text(encoding="utf-8")
if current == text:
return "unchanged"
if not force:
raise InitFailed(f"{path} exists and differs from what init writes (it may hold your edits). "
"Keep it, or re-run with --force to replace it.")
path.write_text(text, encoding="utf-8")
return "replaced"
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(text, encoding="utf-8")
return "written"
def venv_python() -> str:
return os.environ.get("ERP_COMPOSE_PYTHON") or shutil.which("python3") or sys.executable
def install(root: pathlib.Path, plan: Plan, reqs: str, say) -> str:
py = root / ".venv" / "bin" / "python"
if not py.exists():
p = run([venv_python(), "-m", "venv", root / ".venv"], quiet=True)
if p.returncode:
raise InitFailed(f"python3 -m venv .venv failed: {(p.stderr or '').strip()}")
stamp = root / ".venv" / ".erp-compose-installed"
if stamp.exists() and stamp.read_text(encoding="utf-8") == reqs:
return "unchanged"
say(" pip install -r requirements.txt")
p = run([py, "-m", "pip", "install", "-q", "--disable-pip-version-check", "-r", "requirements.txt"], cwd=root,
quiet=True)
if p.returncode:
tail = "\n".join((p.stderr or p.stdout or "").strip().splitlines()[-15:])
raise InitFailed(f"pip install failed:\n{tail}")
stamp.write_text(reqs, encoding="utf-8")
return "installed"
def _say(line: str) -> None:
print(line, flush=True) # before the subprocesses write to the same terminal
def gitignore(root: pathlib.Path, line: str) -> str:
path = root / ".gitignore"
text = path.read_text(encoding="utf-8") if path.exists() else ""
if line in text.splitlines():
return "unchanged"
path.write_text(text + ("" if not text or text.endswith("\n") else "\n") + line + "\n", encoding="utf-8")
return "written"
STARTER_FILE = "config/starter.calls.yaml"
_STARTER_NAME = re.compile(r"^[a-z0-9][a-z0-9-]{0,63}$")
def read_starter(method: "Method", name: str) -> tuple:
"""(text, modules) of starters/<name>.calls.yaml at the method's ref."""
import yaml
if not _STARTER_NAME.match(name or ""):
raise InitFailed(f"--starter {name!r}: a starter is named in lower-case letters, digits and dashes")
try:
text = method.read(f"starters/{name}.calls.yaml")
except InitFailed as e:
raise InitFailed(f"--starter {name}: erp-method {method.ref} has no starters/{name}.calls.yaml ({e})") from None
doc = yaml.safe_load(text)
if not isinstance(doc, dict) or not isinstance(doc.get("calls"), list) or not doc.get("modules"):
raise InitFailed(f"starters/{name}.calls.yaml: expected a mapping with modules: [...] and calls: [...]")
return text, [str(m) for m in doc["modules"]]
def starter_file(name: str) -> str:
"""0.9.11: each starter in its own file (a second starter used to overwrite the first in STARTER_FILE)."""
return f"config/{name}.calls.yaml"
def starter_next(rels=(STARTER_FILE,)) -> str:
rels = [rels] if isinstance(rels, str) else list(rels)
return (" the starter" + ("s, in this order" if len(rels) > 1 else "")
+ " (placeholders marked PLACEHOLDER: the customer's own name, address, VAT number):\n"
" .venv/bin/erp-compose migrate --local compose.yaml\n"
+ "".join(f" .venv/bin/erp-compose seed --local compose.yaml {r}\n" for r in rels) +
" .venv/bin/erp-compose serve --local compose.yaml "
"(seeded today: the business clock starts tomorrow, when rows a brick dates from tomorrow are in force)")
def init(module_ids, root=".", method_ref=None, no_install=False, force=False, read_copy=False, unsigned=None,
say=_say, starter=None) -> int:
root = pathlib.Path(root).resolve()
root.mkdir(parents=True, exist_ok=True)
method = None
starters = [starter] if isinstance(starter, str) else list(dict.fromkeys(starter or ()))
starter_texts = [] # [(name, text)] in the order given: the order they are seeded
if starters:
method = Method(ref=method_ref or os.environ.get("ERP_METHOD_REF"), local=os.environ.get("ERP_METHOD_DIR"))
named, wanted_all = list(module_ids), []
for name in starters:
text, wanted = read_starter(method, name)
starter_texts.append((name, text))
wanted_all += [m for m in wanted if m not in wanted_all]
missing = [m for m in wanted if m not in (named or wanted_all)]
if missing:
raise InitFailed(f"--starter {name} configures {', '.join(missing)} too: add "
f"{'it' if len(missing) == 1 else 'them'} to the modules, or name none to take the "
"starter's")
if not module_ids:
module_ids = wanted_all
from erp_platform import gate
designed = [m for m in dict.fromkeys(module_ids) if (root / f"ERP_{m.upper()}.md").exists()]
if designed:
# 0.8: a module that is designed here is built only from its signed BPMN (a playbook's is verified at plan)
if not gate.enforce(f"erp-compose init {' '.join(module_ids)}", root, mods=designed, unsigned=unsigned,
can_boot=False, say=say):
raise InitFailed("refused: the design is not signed (above)")
else:
say(f" no ERP_<MODULE>.md here for {', '.join(module_ids)}: before the build, show the customer the process "
"(`erp-compose bpmn --from ERP_<MODULE>.md`) and record their signature (`erp-compose signoff record`); "
"`plan` and `migrate` refuse a module without it")
method = method or Method(ref=method_ref or os.environ.get("ERP_METHOD_REF"), local=os.environ.get("ERP_METHOD_DIR"))
plan = resolve(method, list(dict.fromkeys(module_ids)))
say(f"erp-compose init {' '.join(plan.modules)} — erp-method {plan.method_ref}, into {root}")
owned = read_owned(root)
if read_copy:
for r in plan.repos:
state = clone(r, root / "bricks" / r.name, say)
say(f" {state:<9} bricks/{r.name} {r.tag} (a copy to READ; nothing installs from it)")
say(f" {gitignore(root, 'bricks/'):<9} .gitignore (bricks/ is never the customer's code)")
for name in plan.skipped:
say(f" not cloned {name} (not a brick: nothing in the composition imports it)")
for r in plan.repos:
o = owned.get(r.entry.get("name")) if r.role == "brick" else None
if o and o["upstream"] != f"{r.name}@{r.tag}":
say(f" OWNED {r.entry['name']}: ./{o['path']} diverged from {o['upstream']}; the method now pins "
f"{r.tag} — not moved (merge upstream by hand if the customer wants it)")
reqs, test_src = requirements_txt(plan, owned), method.read(method.index["method"]["test_composition"])
files = [("requirements.txt", reqs), ("compose.yaml", compose_yaml(plan, root.name, owned)),
("tests/test_composition.py", test_src)]
starter_of = {starter_file(n): n for n, _t in starter_texts}
files += [(starter_file(n), t) for n, t in starter_texts]
for rel, text in files:
say(f" {write(root / rel, text, force):<9} {rel}" + (f" (starter {starter_of[rel]}, erp-method {method.ref})"
if rel in starter_of else ""))
designs(method, plan.modules, root, force, say)
starter_text = starter_texts or None
seed_files = [starter_file(n) for n, _t in starter_texts]
if no_install:
say("\n--no-install: no .venv, no install, no checks. Next:\n"
" python3 -m venv .venv && .venv/bin/pip install -r requirements.txt\n"
" .venv/bin/erp-compose check compose.yaml && .venv/bin/python -m pytest -q tests/"
+ ("\n" + starter_next(seed_files) if starter_text is not None else ""))
return 0
say(f" {install(root, plan, reqs, say):<9} .venv")
bin_ = root / ".venv" / "bin"
say("\n$ erp-compose check compose.yaml")
if run([bin_ / "erp-compose", "check", "compose.yaml"], cwd=root).returncode:
raise InitFailed("erp-compose check found problems (above); the bricks at these tags do not compose")
say("\n$ python -m pytest -q tests/")
if run([bin_ / "python", "-m", "pytest", "-q", "-p", "no:cacheprovider", "tests/"], cwd=root).returncode:
raise InitFailed("tests/test_composition.py failed (above)")
bricks = ", ".join(f"{r.entry['name']} {r.version}" for r in plan.repos if r.role == "brick")
say(f"\nReady: {bricks} composed in compose.yaml; check clean, composition test passing.\n"
"Next:\n"
+ (f" 1. {NEXT}\n (edit {', '.join(seed_files)} and replay it with `erp-compose seed`; never an INSERT into a brick's table)\n"
if starter_text is not None else
f" 1. {NEXT}\n (a calls file replayed with `erp-compose seed`; never an INSERT into a brick's table)\n") +
" 2. extend: a need no brick covers is a new brick (erp-platform templates/brick-example/, BRICKS.md) added to compose.yaml\n"
" 3. own: a signed change to a brick itself: erp-compose own <brick> --reason \"<CR / deviation>\" (dodil://erp/phase/build)\n"
" read a brick's README / AGENTS.md / code: re-run with --read-copy (bricks/, git-ignored)\n"
" run it on this machine (dev tier, SQLite): .venv/bin/erp-compose migrate --local compose.yaml && "
".venv/bin/erp-compose serve --local compose.yaml\n"
" run it on DataK3: BUCKET=<bucket> .venv/bin/erp-compose migrate compose.yaml, then serve"
+ ("\n" + starter_next(seed_files) if starter_text is not None else ""))
return 0
def designs(method: "Method", module_ids: list, root: pathlib.Path, force: bool, say) -> None:
"""0.9.11: processes/<module>/module.yaml and processes/<module>/baseline/ (the method's receipts + history, for
reference — never signoff/, which only the customer's signature fills), from the method, unedited."""
from erp_platform import signoff
known = {m["id"]: m for m in method.index.get("modules") or []}
listed = method.index.get("files") or []
for mid in module_ids:
spec = (known.get(mid) or {}).get("spec")
if not spec:
continue
rel = f"{signoff.SPEC_DIR}/{mid}/module.yaml"
say(f" {write(root / rel, method.read(spec), force):<9} {rel} (the module's spec, erp-method {method.ref})")
prefix = f"{pathlib.PurePosixPath(spec).parent}/signoff/{mid}/"
receipts = [f for f in listed if f.startswith(prefix)]
states = {}
base = f"{signoff.SPEC_DIR}/{mid}/{signoff.BASELINE}"
for f in receipts:
state = write(root / base / f[len(prefix):], method.read(f), force)
states[state] = states.get(state, 0) + 1
if receipts:
say(f" {'+'.join(states):<9} {base}/ ({len(receipts)} files: the method's standard process, for "
f"reference — NOT a signature; signoff/{mid}/ is filled only by the customer: "
"`erp-compose signoff record --all`)")
USAGE = ("usage: erp-compose init <module> [<module> …] [--starter NAME …] [--dir DIR] [--method-ref REF] [--read-copy] "
"[--no-install] [--force] [--unsigned \"<reason>\"]\n"
" erp-compose init --starter NAME [--starter NAME …] (no modules: the starters'; seeded in order)")
def main(argv) -> int:
args, mods, opts = list(argv), [], {"dir": ".", "method_ref": None, "no_install": False, "force": False,
"read_copy": False, "unsigned": None, "starter": []}
while args:
a = args.pop(0)
if a in ("--dir", "--method-ref", "--unsigned", "--starter"):
if not args:
print(f"{a} needs a value\n{USAGE}", file=sys.stderr)
return 2
if a == "--starter": # 0.9.11: repeatable, seeded in the order given
opts["starter"].append(args.pop(0))
else:
opts[a[2:].replace("-", "_")] = args.pop(0)
elif a in ("--no-install", "--force", "--read-copy"):
opts[a[2:].replace("-", "_")] = True
elif a in ("-h", "--help"):
print(USAGE + "\n\n" + __doc__.split("\n\n", 1)[1])
return 0
elif a.startswith("-"):
print(f"unknown option {a}\n{USAGE}", file=sys.stderr)
return 2
else:
mods.append(a)
if not mods and not opts["starter"]:
print(USAGE, file=sys.stderr)
return 2
try:
return init(mods, root=opts["dir"], method_ref=opts["method_ref"], no_install=opts["no_install"],
force=opts["force"], read_copy=opts["read_copy"], unsigned=opts["unsigned"],
starter=opts["starter"])
except InitFailed as e:
print(f"erp-compose init: {e}", file=sys.stderr)
return 1
# ── own: a brick's code becomes the customer's (depth 3) ───────────────────────────────────
_PIN = re.compile(r"^(?P<dist>[A-Za-z0-9_.-]+)(?:\[[^\]]*\])? @ git\+(?P<url>\S+)@(?P<tag>[^\s@]+)\s*#\s*(?P<name>\S+)\s*$")
def own(brick: str, reason: str, root=".", say=_say) -> int:
"""Copy the brick at its pinned tag into owned/<repo>/ (tracked: it is the customer's code now), pin it to that
path, and record why and from what upstream on its compose.yaml entry. Refuses without a reason."""
root = pathlib.Path(root).resolve()
if not reason or not reason.strip():
raise InitFailed("--reason is required: the signed change (CR / deviation) that needs the brick's own code")
reqs_path, comp_path = root / "requirements.txt", root / "compose.yaml"
if not reqs_path.exists() or not comp_path.exists():
raise InitFailed(f"{root} has no requirements.txt / compose.yaml: run `erp-compose init <module>` first")
mine = {k: o for k, o in read_owned(root).items() if brick in (k, str(o.get("upstream", "")).split("@")[0])}
if mine:
raise InitFailed(f"{brick} is already the customer's own ({next(iter(mine.values()))['path']})")
lines = reqs_path.read_text(encoding="utf-8").splitlines()
hit = next(((i, m) for i, m in ((i, _PIN.match(l)) for i, l in enumerate(lines))
if m and brick in (m["name"], m["dist"]) and m["name"] != "platform"), None)
if hit is None:
raise InitFailed(f"no pinned brick {brick!r} in requirements.txt (a brick's entry name or repo, e.g. hr or brick-hr)")
i, m = hit
name, dist, tag = m["name"], m["dist"], m["tag"]
dest = root / "owned" / dist
if dest.exists():
raise InitFailed(f"{dest} exists; move it away first")
dest.parent.mkdir(parents=True, exist_ok=True)
url = m["url"].replace("file://localhost/", "file:///", 1)
p = run(["git", "-c", "advice.detachedHead=false", "clone", "-q", "--depth", "1", "--branch", tag, url,
dest], quiet=True)
if p.returncode:
raise InitFailed(f"git clone {m['url']} at {tag} failed: {(p.stderr or '').strip()}")
shutil.rmtree(dest / ".git") # the customer's code now, tracked in THEIR repo
o = {"path": f"owned/{dist}", "upstream": f"{dist}@{tag}", "reason": reason.strip()}
lines[i] = f"-e ./{o['path']} # {name} (owned: upstream {o['upstream']})"
reqs_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
text = comp_path.read_text(encoding="utf-8")
pat = re.compile(rf"^(\s*-\s*\{{\s*name:\s*{re.escape(name)}\s*,.*?)\s*\}}\s*$", re.M)
if not pat.search(text):
raise InitFailed(f"compose.yaml has no one-line entry `- {{ name: {name}, … }}` to record the ownership on")
comp_path.write_text(pat.sub(lambda x: f"{x.group(1)}, {owned_entry(o)} }}", text, count=1), encoding="utf-8")
say(f"owned {name}: {dist} {tag} → ./{o['path']} (tracked), pinned -e ./{o['path']}; compose.yaml records "
f"upstream {o['upstream']} and the reason")
py = root / ".venv" / "bin" / "python"
if py.exists():
p = run([py, "-m", "pip", "install", "-q", "--disable-pip-version-check", "-e", f"./{o['path']}"], cwd=root,
quiet=True)
if p.returncode:
raise InitFailed(f"pip install -e ./{o['path']} failed: {(p.stderr or p.stdout or '').strip()[-800:]}")
say(f" .venv now runs ./{o['path']}")
return 0
def own_main(argv) -> int:
import argparse
ap = argparse.ArgumentParser(prog="erp-compose own", description="Make a brick's code the customer's own (depth 3).")
ap.add_argument("brick", help="the brick's entry name in compose.yaml (hr) or its repo (brick-hr)")
ap.add_argument("--reason", required=True, help="the signed change that needs it (CR-n / D-n, in words)")
ap.add_argument("--dir", default=".")
a = ap.parse_args(argv)
try:
return own(a.brick, a.reason, a.dir)
except InitFailed as e:
print(f"erp-compose own: {e}", file=sys.stderr)
return 1