Files
breakpilot-compliance/scripts/obligation_discovery/export_join_keys.py
T
Benjamin Admin 6523286af6 feat(registry-quality): scope-Achse — 2 out_of_scope + derived_obligation (User Option 2)
User-Entscheidung 2026-07-01 zum Scope-Audit: Adressat der Norm != Handlungspflicht des
Herstellers. Neue `scope`-Attribut-Achse (Enum, KEINE neue Objektklasse -> Freeze v1.0
unberuehrt): in_scope (default) / out_of_scope / derived_obligation.

- sanctions + market_surveillance_safeguard -> out_of_scope (reine Staats-/Durchsetzungs-
  bestimmungen; Praezedenz CSIRT/ENISA im CRA-Vuln-Cut). Aus join_keys gefiltert.
- notified_body_requirements -> derived_obligation (Norm adressiert primaer die notifizierte
  Stelle, erzeugt aber mittelbare Herstellerpflichten: NB einbeziehen + Unterlagen +
  Konformitaetsbewertung) + scope_split_candidate (spaetere Aufspaltung Normadressat <->
  abgeleitete Herstellerpflicht). BLEIBT im Set (Prinzip: Wissen nicht zu frueh verwerfen).
- export_join_keys.py filtert scope==out_of_scope + fuehrt scope je Eintrag -> join_keys
  126->124 (MaschVO 31->29; 123 in_scope + 1 derived_obligation).
- scope_audit.py jetzt 3-Wege-klassifikations-bewusst (0 unklassifizierte Reste) +
  apply_scope_classification.py (deterministisch). Fuer jeden kuenftigen Cut mitlaufen.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-07-01 13:22:38 +02:00

56 lines
2.5 KiB
Python

"""Exportiert den OBLIGATION_ID-Join-Key-Vertrag aus den Registry-Artefakten.
Die obligation_id ist der stabile Brueckenschluessel zwischen Legal Knowledge Graph
(citation_spans haengen an obligation_id) und Compliance Execution Graph
(control_mapping.source_norm -> obligation_id). citation_units = die legal_basis-Anker,
ueber die beide Seiten heute (vor obligation_id-Adoption) bruecken koennen.
DISZIPLIN: obligation_id wird RE-GELINKT, NIE neu vergeben (Pendant zu span_id/control_uuid).
python3 scripts/obligation_discovery/export_join_keys.py obligations/cra.json obligations/cra_authentication.json
"""
from __future__ import annotations
import argparse
import json
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("registries", nargs="+")
ap.add_argument("--out", default="obligations/obligation_join_keys.json")
a = ap.parse_args()
keys = []
for path in a.registries:
reg = json.load(open(path, encoding="utf-8"))
for o in reg.get("obligations", []):
if o.get("scope") == "out_of_scope":
continue # Institutions-/Enforcement-Recht (Adressat != Hersteller) -> nicht im Join-Vertrag
citation_units = [b.get("anchor", "") for b in o.get("legal_basis", []) if b.get("anchor")]
keys.append({
"obligation_id": o["id"],
"regulation": reg.get("regulation", ""),
"family": o.get("family", ""),
"tier": o.get("tier", ""),
"scope": o.get("scope", "in_scope"),
"citation_units": citation_units,
"source_role": o.get("source_role", ""),
})
out = {
"schema_version": "obligation_join_keys_v1",
"contract": "obligation_id ist der stabile Join-Key. Legal Knowledge Graph haengt "
"citation_spans an obligation_id; Compliance Execution Graph mappt "
"control_mapping.source_norm -> obligation_id. Interim-Bruecke = citation_units. "
"obligation_id NIE neu vergeben (re-link).",
"count": len(keys),
"obligation_ids": keys,
}
json.dump(out, open(a.out, "w", encoding="utf-8"), ensure_ascii=False, indent=1)
from collections import Counter
print(f"exportiert: {a.out} ({len(keys)} obligation_ids)")
print("Regulierungen:", dict(Counter(k["regulation"] for k in keys)))
print("Familien:", dict(Counter(k["family"] for k in keys)))
if __name__ == "__main__":
main()