#!/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 # Opt-in: also publish the zaaktype so OpenZaak's Zaken API accepts a zaak against # it (a concept zaaktype is rejected with `not-published`). Off by default — the # S-01 compose seed keeps it a concept (ADR-0002). The ACL integration test # (S-04a, #46) sets OZ_PUBLISH=1. Publishing requires ≥2 statustypen, ≥1 roltype # and ≥1 resultaattype; the resultaattype is validated against the external # Selectielijst reference API, so this path needs outbound access to it. See ADR-0006. PUBLISH = os.environ.get("OZ_PUBLISH", "").lower() in ("1", "true", "yes") SELECTIELIJST = os.environ.get( "OZ_SELECTIELIJST", "https://selectielijst.openzaak.nl/api/v1").rstrip("/") 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 selectielijst(path): """GET the external Selectielijst reference API (no auth). Used only when publishing.""" req = urllib.request.Request(f"{SELECTIELIJST}{path}", headers={"Accept": "application/json"}) with urllib.request.urlopen(req, timeout=30) as r: return json.loads(r.read()) def publish_zaaktype(zt): """Add the relations OpenZaak requires to publish, then publish (idempotent). Publish validation (verified against OpenZaak 1.28.2) demands: ≥2 statustypen (begin + eind), ≥1 roltype, ≥1 resultaattype. A resultaattype needs a Selectielijst `selectielijstklasse` whose procestype matches the zaaktype's `selectielijstProcestype`, plus a `resultaattypeomschrijving`. """ have_st = {s.get("volgnummer") for s in find(f"/statustypen?zaaktype={zt['url']}&status=alles")} for volgnummer, omschrijving in [(1, "Ontvangen"), (2, "Afgehandeld")]: if volgnummer not in have_st: st, body = api("POST", "/statustypen", { "omschrijving": omschrijving, "zaaktype": zt["url"], "volgnummer": volgnummer}) if st != 201: sys.exit(f"create statustype {volgnummer} -> {st}: {json.dumps(body, indent=2)}") print(f"create statustype {volgnummer} ({omschrijving})") if find(f"/roltypen?zaaktype={zt['url']}&status=alles"): print("skip roltype Aanvrager") else: st, body = api("POST", "/roltypen", { "zaaktype": zt["url"], "omschrijving": "Aanvrager", "omschrijvingGeneriek": "initiator"}) if st != 201: sys.exit(f"create roltype -> {st}: {json.dumps(body, indent=2)}") print("create roltype Aanvrager") if find(f"/resultaattypen?zaaktype={zt['url']}&status=alles"): print("skip resultaattype Geregistreerd") else: resultaat = selectielijst("/resultaten?pageSize=1")["results"][0] omschrijvingen = selectielijst("/resultaattypeomschrijvingen") oms = (omschrijvingen if isinstance(omschrijvingen, list) else omschrijvingen["results"])[0]["url"] # The selectielijstklasse and the zaaktype must share a procestype. st, body = api("PATCH", zt["url"], {"selectielijstProcestype": resultaat["procesType"]}) if st != 200: sys.exit(f"set procestype -> {st}: {json.dumps(body, indent=2)}") st, body = api("POST", "/resultaattypen", { "zaaktype": zt["url"], "omschrijving": "Geregistreerd", "resultaattypeomschrijving": oms, "selectielijstklasse": resultaat["url"], "archiefnominatie": "blijvend_bewaren", "brondatumArchiefprocedure": {"afleidingswijze": "afgehandeld"}, }) if st != 201: sys.exit(f"create resultaattype -> {st}: {json.dumps(body, indent=2)}") print("create resultaattype Geregistreerd") if zt.get("concept", True): st, body = api("POST", f"{zt['url']}/publish") if st != 200: sys.exit(f"publish zaaktype -> {st}: {json.dumps(body, indent=2)}") print(f"publish zaaktype BIG-REGISTRATIE ({zt['url']})") else: print("skip publish (already published)") 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") # 4. Optionally publish. By default the zaaktype stays a concept: publishing # requires roltypen, resultaattypen and statustypen, beyond the "lean / # schema-mandatory" zaaktype S-01 asks for (ADR-0002). Set OZ_PUBLISH=1 to add # those relations and publish — needed so a real zaak POST is accepted, which # the ACL integration test (S-04a, #46) exercises. See ADR-0006. if PUBLISH: # Re-fetch: the bsn-eigenschap branch above may hold a stale concept flag. zt = next(z for z in find(f"/zaaktypen?catalogus={cat['url']}&status=alles") if z.get("identificatie") == "BIG-REGISTRATIE") publish_zaaktype(zt) # 5. 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" state = "published" if PUBLISH else "concept" # Machine-readable line so callers (e.g. infra/run-domain-check.sh) can capture the # zaaktype URL to configure the ACL's default-fill (ADR-0003/0009). zt_url = next(z["url"] for z in zaaktypen if z.get("identificatie") == "BIG-REGISTRATIE") print(f"ZAAKTYPE_URL {zt_url}") print(f"OK — BIG catalogus seeded (BIG-REGISTRATIE {state} + bsn eigenschap)") if __name__ == "__main__": main()