#!/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.parse, 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`. """ # Ontvangen (begin) → Afgehandeld (eind, highest volgnummer). "Geannuleerd" (S-10c) sits between # them: a non-terminal status the document-timeout branch sets, so it never displaces the Afgehandeld # eindstatus the approval path resolves. Keyed by volgnummer on a fresh catalogus (CI reseeds); a # stale local stack must reset its OpenZaak volumes for the renumbering to take effect. have_st = {s.get("volgnummer") for s in find(f"/statustypen?zaaktype={zt['url']}&status=alles")} for volgnummer, omschrijving in [(1, "Ontvangen"), (2, "Geannuleerd"), (3, "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") # Two resultaattypen, keyed by omschrijving so each is created independently (idempotent): # "Geregistreerd" — the approval outcome (S-09b) # "Vervallen" — the document-timeout cancellation outcome (S-10c) # Both selectielijstklassen must share the zaaktype's selectielijstProcestype, so pick two # Selectielijst resultaten from a single procestype and set that procestype on the zaaktype. have_rt = {r.get("omschrijving") for r in find(f"/resultaattypen?zaaktype={zt['url']}&status=alles")} wanted = [("Geregistreerd", "blijvend_bewaren"), ("Vervallen", "vernietigen")] if all(naam in have_rt for naam, _ in wanted): print("skip resultaattypen Geregistreerd + Vervallen") else: # Anchor on the procestype of an arbitrary resultaat, then fetch that procestype's resultaten so # both klassen validate against the zaaktype's selectielijstProcestype. procestype = selectielijst("/resultaten?pageSize=1")["results"][0]["procesType"] resultaten = selectielijst(f"/resultaten?procesType={urllib.parse.quote(procestype, safe='')}")["results"] if len(resultaten) < len(wanted): sys.exit(f"selectielijst procestype has too few resultaten ({len(resultaten)}) for {len(wanted)} resultaattypen") omschrijvingen = selectielijst("/resultaattypeomschrijvingen") oms_list = omschrijvingen if isinstance(omschrijvingen, list) else omschrijvingen["results"] st, body = api("PATCH", zt["url"], {"selectielijstProcestype": procestype}) if st != 200: sys.exit(f"set procestype -> {st}: {json.dumps(body, indent=2)}") for i, (naam, archiefnominatie) in enumerate(wanted): if naam in have_rt: print(f"skip resultaattype {naam}") continue st, body = api("POST", "/resultaattypen", { "zaaktype": zt["url"], "omschrijving": naam, "resultaattypeomschrijving": oms_list[i]["url"], "selectielijstklasse": resultaten[i]["url"], "archiefnominatie": archiefnominatie, "brondatumArchiefprocedure": {"afleidingswijze": "afgehandeld"}, }) if st != 201: sys.exit(f"create resultaattype {naam} -> {st}: {json.dumps(body, indent=2)}") print(f"create resultaattype {naam}") 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 seed_informatieobjecttype(cat, zt): """Create the "Diploma" informatieobjecttype and relate it to the zaaktype (both idempotent). A diploma uploaded in S-10b is filed under this informatieobjecttype; OpenZaak only accepts a document (and its zaak relation) once the informatieobjecttype is published AND allowed for the zaak's zaaktype (a zaaktype-informatieobjecttype relation). Both the relation and this call must run while the zaaktype is still a concept, so seed this *before* publishing the zaaktype. Returns the informatieobjecttype dict. """ iots = [i for i in find(f"/informatieobjecttypen?catalogus={cat['url']}&status=alles") if i.get("omschrijving") == "Diploma"] if iots: iot = iots[0] print(f"skip informatieobjecttype Diploma ({iot['url']}) concept={iot.get('concept')}") else: st, iot = api("POST", "/informatieobjecttypen", { "catalogus": cat["url"], "omschrijving": "Diploma", "vertrouwelijkheidaanduiding": "openbaar", "informatieobjectcategorie": "diploma", "beginGeldigheid": "2026-01-01", }) if st != 201: sys.exit(f"create informatieobjecttype -> {st}: {json.dumps(iot, indent=2)}") print(f"create informatieobjecttype Diploma ({iot['url']})") # Relate it to the zaaktype (must be done while both are concept). relations = find(f"/zaaktype-informatieobjecttypen?zaaktype={zt['url']}&status=alles") if any(r.get("informatieobjecttype") == iot["url"] for r in relations): print("skip zaaktype-informatieobjecttype Diploma") else: st, body = api("POST", "/zaaktype-informatieobjecttypen", { "zaaktype": zt["url"], "informatieobjecttype": iot["url"], "volgnummer": 1, "richting": "inkomend"}) if st != 201: sys.exit(f"relate zaaktype-informatieobjecttype -> {st}: {json.dumps(body, indent=2)}") print("create zaaktype-informatieobjecttype Diploma") return iot def publish_informatieobjecttype(iot): """Publish the informatieobjecttype (idempotent) so documents may reference it.""" if iot.get("concept", True): st, body = api("POST", f"{iot['url']}/publish") if st != 200: sys.exit(f"publish informatieobjecttype -> {st}: {json.dumps(body, indent=2)}") print(f"publish informatieobjecttype Diploma ({iot['url']})") else: print("skip publish informatieobjecttype (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. iot = None 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") # Seed + relate the Diploma informatieobjecttype (S-10b) while the zaaktype is still concept, # then publish both. Publish the informatieobjecttype before the zaaktype so the zaaktype's # relations reference a published type. iot = seed_informatieobjecttype(cat, zt) publish_informatieobjecttype(iot) 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}") # Machine-readable informatieobjecttype URL (S-10b) so callers can configure the ACL's document # default-fill. Only emitted when publishing — a concept informatieobjecttype can't back a document. if iot is not None: print(f"INFORMATIEOBJECTTYPE_URL {iot['url']}") print(f"OK — BIG catalogus seeded (BIG-REGISTRATIE {state} + bsn eigenschap)") if __name__ == "__main__": main()