Niek Otten
f2e575b427
OpenZaak rejects a zaak against a concept zaaktype (not-published). Add an opt-in OZ_PUBLISH path that creates the relations publish requires — two statustypen, a roltype, and a resultaattype whose Selectielijst procestype is matched onto the zaaktype — then publishes. Default stays concept (ADR-0002); only the ACL integration test flips it on. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
218 lines
9.8 KiB
Python
218 lines
9.8 KiB
Python
#!/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"
|
|
print(f"OK — BIG catalogus seeded (BIG-REGISTRATIE {state} + bsn eigenschap)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|