register-referentie/infra/openzaak/seed_catalogus.py

#!/usr/bin/env python3
"""Idempotent seed of the BIG catalogus into OpenZaak via the ZTC API.

Creates (if absent):
  - catalogus "BIG"
  - a lean "BIG-registratie" zaaktype (only schema-mandatory fields)
  - a "bsn" eigenschap on that zaaktype
  - then publishes the zaaktype.

Auth uses the JWT client provisioned by setup_configuration (see ADR-0002).
Stdlib only — no pip deps. Re-running is safe (matches existing by identifier).
"""
import base64, hashlib, hmac, json, os, sys, time, urllib.error, urllib.request

BASE = os.environ.get("OZ_BASE", "http://localhost:8000")
CLIENT_ID = os.environ.get("OZ_CLIENT_ID", "big-reference-seed")
SECRET = os.environ.get("OZ_SECRET", "insecure-dev-secret-change-me")
ZTC = f"{BASE}/catalogi/api/v1"
RSIN = "517439943"  # elfproef-valid test RSIN


def token():
    b64 = lambda b: base64.urlsafe_b64encode(b).rstrip(b"=")
    hdr = {"alg": "HS256", "typ": "JWT"}
    pl = {"iss": CLIENT_ID, "iat": int(time.time()), "client_id": CLIENT_ID,
          "user_id": "seed", "user_representation": "seed"}
    seg = b64(json.dumps(hdr, separators=(",", ":")).encode()) + b"." + \
          b64(json.dumps(pl, separators=(",", ":")).encode())
    sig = b64(hmac.new(SECRET.encode(), seg, hashlib.sha256).digest())
    return (seg + b"." + sig).decode()


def api(method, path, body=None):
    url = path if path.startswith("http") else f"{ZTC}{path}"
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(url, data=data, method=method, headers={
        "Authorization": "Bearer " + token(),
        "Content-Type": "application/json",
        "Accept": "application/json",
    })
    try:
        with urllib.request.urlopen(req, timeout=30) as r:
            return r.status, json.loads(r.read() or "null")
    except urllib.error.HTTPError as e:
        return e.code, json.loads(e.read() or "null")


def find(path):
    status, body = api("GET", path)
    if status != 200:
        sys.exit(f"GET {path} -> {status}: {body}")
    return body.get("results", [])


def main():
    # 1. Catalogus
    existing = [c for c in find(f"/catalogussen?domein=BIG") if c.get("domein") == "BIG"]
    if existing:
        cat = existing[0]
        print(f"skip   catalogus BIG ({cat['url']})")
    else:
        st, cat = api("POST", "/catalogussen", {
            "domein": "BIG", "rsin": RSIN,
            "contactpersoonBeheerNaam": "BIG Beheer",
        })
        if st != 201:
            sys.exit(f"create catalogus -> {st}: {cat}")
        print(f"create catalogus BIG ({cat['url']})")

    # 2. Zaaktype (concept)
    # status=alles so concept zaaktypen are matched too (else we'd duplicate).
    zts = [z for z in find(f"/zaaktypen?catalogus={cat['url']}&status=alles")
           if z.get("identificatie") == "BIG-REGISTRATIE"]
    if zts:
        zt = zts[0]
        print(f"skip   zaaktype BIG-REGISTRATIE ({zt['url']}) concept={zt.get('concept')}")
    else:
        st, zt = api("POST", "/zaaktypen", {
            "identificatie": "BIG-REGISTRATIE",
            "omschrijving": "BIG-registratie",
            "vertrouwelijkheidaanduiding": "openbaar",
            "doel": "Registratie van een zorgprofessional in het BIG-register",
            "aanleiding": "Aanvraag tot registratie",
            "indicatieInternOfExtern": "extern",
            "handelingInitiator": "indienen",
            "onderwerp": "BIG-registratie",
            "handelingBehandelaar": "behandelen",
            "doorlooptijd": "P30D",
            "opschortingEnAanhoudingMogelijk": False,
            "verlengingMogelijk": False,
            "publicatieIndicatie": False,
            "productenOfDiensten": [],
            "referentieproces": {"naam": "BIG-registratie"},
            "catalogus": cat["url"],
            "besluittypen": [],
            "gerelateerdeZaaktypen": [],
            "beginGeldigheid": "2026-01-01",
            "versiedatum": "2026-01-01",
            "verantwoordelijke": RSIN,
        })
        if st != 201:
            sys.exit(f"create zaaktype -> {st}: {json.dumps(zt, indent=2)}")
        print(f"create zaaktype BIG-REGISTRATIE ({zt['url']})")

    # 3. bsn eigenschap (only addable while concept)
    eigs = [e for e in find(f"/eigenschappen?zaaktype={zt['url']}&status=alles")
            if e.get("naam") == "bsn"]
    if eigs:
        print("skip   eigenschap bsn")
    elif zt.get("concept", True):
        st, eig = api("POST", "/eigenschappen", {
            "naam": "bsn",
            "definitie": "Burgerservicenummer van de zorgprofessional",
            "zaaktype": zt["url"],
            "specificatie": {"groep": "aanvrager", "formaat": "tekst",
                             "lengte": "9", "kardinaliteit": "1", "waardenverzameling": []},
        })
        if st != 201:
            sys.exit(f"create eigenschap -> {st}: {json.dumps(eig, indent=2)}")
        print("create eigenschap bsn")
    else:
        print("warn   zaaktype already published; cannot add bsn eigenschap")

    # Intentionally NOT published. Publishing requires roltypen, resultaattypen
    # and statustypen, which go beyond the "lean / schema-mandatory" zaaktype this
    # slice asks for; they arrive with the workflow/zaak slices. See ADR-0002.

    # 4. Verify the JWT client can list the zaaktype (concepts included).
    zaaktypen = find(f"/zaaktypen?catalogus={cat['url']}&status=alles")
    names = [z.get("identificatie") for z in zaaktypen]
    print(f"zaaktypen in BIG: {names}")
    assert "BIG-REGISTRATIE" in names, "BIG-REGISTRATIE not listed"
    print("OK — BIG catalogus seeded (BIG-REGISTRATIE concept + bsn eigenschap)")


if __name__ == "__main__":
    main()