erp/.claude/skills/dolibarr-sandbox-write/scripts/promote-apply.sh

#!/usr/bin/env bash
# Replay a promote manifest against a target — sandbox (rehearsal) or prod (real).
#
# Resolves the manifest's symbolic @refs to the ids actually created during this
# run, so dependent ops (an invoice -> its just-created thirdparty) wire up on the
# target. Sandbox uses dol-write.sh; prod uses the gated dol-prod-write.sh.
#
#   promote-apply.sh <manifest.json> [--target sandbox|prod]
#
# --target prod requires, in the environment (never stored):
#   DOLIBARR_PROD_WRITE_KEY=<prod write key>
#   ARCO_PROMOTE_CONFIRM=I-UNDERSTAND-THIS-WRITES-PROD
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
MANIFEST="${1:?usage: promote-apply.sh <manifest.json> [--target sandbox|prod]}"; shift || true
TARGET="sandbox"
while [[ $# -gt 0 ]]; do
  case "$1" in
    --target) TARGET="${2:?}"; shift 2 ;;
    *) echo "promote-apply.sh: unknown arg '$1'" >&2; exit 2 ;;
  esac
done
case "${TARGET}" in
  sandbox) export DOL_WRITE="${SCRIPT_DIR}/dol-write.sh" ;;
  prod)    export DOL_WRITE="${SCRIPT_DIR}/dol-prod-write.sh" ;;
  *) echo "promote-apply.sh: --target must be sandbox|prod" >&2; exit 2 ;;
esac
echo ">>> promote-apply target=${TARGET} (writes via $(basename "${DOL_WRITE}"))" >&2

python3 - "$MANIFEST" "$SCRIPT_DIR" <<'PY'
import json, sys, subprocess, os
manifest_path, script_dir = sys.argv[1], sys.argv[2]
ops = json.load(open(manifest_path))
OP_SCRIPT = {"thirdparty": "thirdparty-create.sh", "invoice": "invoice-create.sh",
             "creditnote": "creditnote-create.sh", "payment": "payment-record.sh"}
refmap = {}

import urllib.parse
DOL_WRITE = os.environ.get("DOL_WRITE")  # GET wrapper for the chosen target

# Business-key lookup of a PRE-EXISTING entity on the target, so a manifest can
# reference records it does not create (e.g. an invoice for an existing client).
# Resolves against the *target*, so #thirdparty:name=X becomes the sandbox id on
# --target sandbox and the prod id on --target prod (portability for real changes).
ENTITY_LOOKUP = {
    "thirdparty": ("/thirdparties", {"name": "t.nom", "code": "t.code_client",
                                     "supplier_code": "t.code_fournisseur"}),
    "invoice": ("/invoices", {"ref": "t.ref"}),
    "supplierinvoice": ("/supplierinvoices", {"ref": "t.ref", "ref_supplier": "t.ref_supplier"}),
}
def lookup(spec):
    try:
        ent, rest = spec.split(":", 1); field, value = rest.split("=", 1)
    except ValueError:
        sys.exit("promote-apply: bad lookup '#%s' (use #entity:field=value)" % spec)
    if ent not in ENTITY_LOOKUP:
        sys.exit("promote-apply: unknown lookup entity '%s'" % ent)
    endpoint, cols = ENTITY_LOOKUP[ent]
    if field not in cols:
        sys.exit("promote-apply: cannot look %s up by '%s' (try %s)" % (ent, field, "/".join(cols)))
    flt = "(%s:=:'%s')" % (cols[field], value.replace("'", "''"))
    path = "%s?limit=2&sqlfilters=%s" % (endpoint, urllib.parse.quote(flt))
    r = subprocess.run([DOL_WRITE, "GET", path], capture_output=True, text=True, env=os.environ)
    if r.returncode != 0:
        sys.stderr.write(r.stdout + r.stderr + "\n")
        sys.exit("promote-apply: lookup '#%s' query failed" % spec)
    try:
        rows = json.loads(r.stdout)
    except Exception:
        rows = []
    rows = rows if isinstance(rows, list) else []
    if len(rows) == 0:
        sys.exit("promote-apply: lookup '#%s' matched nothing on the target" % spec)
    if len(rows) > 1:
        sys.exit("promote-apply: lookup '#%s' is ambiguous (%d matches) — use a unique key" % (spec, len(rows)))
    return int(rows[0]["id"])

def resolve(v):
    if isinstance(v, str) and v.startswith("@"):
        k = v[1:]
        if k not in refmap:
            sys.exit("promote-apply: unresolved ref @%s (is it created earlier in the manifest?)" % k)
        return refmap[k]
    if isinstance(v, str) and v.startswith("#"):
        return lookup(v[1:])
    if isinstance(v, dict):
        return {kk: resolve(vv) for kk, vv in v.items()}
    if isinstance(v, list):
        return [resolve(x) for x in v]
    return v
for i, op in enumerate(ops, 1):
    t = op["op"]; script = OP_SCRIPT.get(t)
    if not script:
        sys.exit("promote-apply: unknown op '%s'" % t)
    inp = resolve(op.get("input", {}))
    r = subprocess.run([os.path.join(script_dir, script)], input=json.dumps(inp),
                       capture_output=True, text=True, env=os.environ)
    if r.returncode != 0:
        sys.stderr.write(r.stdout + r.stderr + "\n")
        sys.exit("promote-apply: op %d (%s) FAILED" % (i, t))
    out = r.stdout.strip()
    try:
        rid = json.loads(out).get("id")
    except Exception:
        rid = out if out.isdigit() else None
    ref = op.get("ref")
    if ref and rid is not None:
        refmap[ref] = int(rid) if str(rid).isdigit() else rid
    print("  [%d/%d] %-11s %-8s -> id=%s" % (i, len(ops), t, ("@" + ref) if ref else "", rid))
print("OK — promote complete. ref -> id: %s" % json.dumps(refmap))
PY