Benjamin_Boenisch/breakpilot-compliance
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: Backfill nutzt source_citation statt control_parent_links
All checks were successful
CI/CD / go-lint (push) Has been skipped
Details
CI/CD / python-lint (push) Has been skipped
Details
CI/CD / nodejs-lint (push) Has been skipped
Details
CI/CD / test-go-ai-compliance (push) Successful in 34s
Details
CI/CD / test-python-backend-compliance (push) Successful in 29s
Details
CI/CD / test-python-document-crawler (push) Successful in 21s
Details
CI/CD / test-python-dsms-gateway (push) Successful in 17s
Details
CI/CD / validate-canonical-controls (push) Successful in 10s
Details
CI/CD / Deploy (push) Successful in 2s
Details

Browse Source 
Die Obligation kennt ihren Parent-Rich-Control direkt. Dessen
source_citation->>'source' gibt die Quell-Regulierung zuverlaessiger
als der Umweg ueber control_parent_links (M:N-Inflation).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Benjamin Admin
2026-03-25 08:25:32 +01:00
parent c56bccaedf
commit 9dbb4cc5d2
1 changed files with 11 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
backend-compliance/compliance/api/canonical_control_routes.py
View File

@@ -937,52 +937,32 @@ async def backfill_normative_strength(
"""
from compliance.data.source_type_classification import (
classify_source_regulation,
get_highest_source_type,
cap_normative_strength,
)
with SessionLocal() as db:
# 1. Alle Obligations mit ihren Parent-Control-Links laden
# 1. Alle Obligations mit source_citation des Parent Controls laden
obligations = db.execute(text("""
SELECT oc.id, oc.candidate_id, oc.normative_strength,
oc.parent_control_uuid
cc.source_citation->>'source' AS parent_source
FROM obligation_candidates oc
JOIN canonical_controls cc ON cc.id = oc.parent_control_uuid
WHERE oc.release_state NOT IN ('rejected', 'merged')
AND oc.normative_strength IS NOT NULL
ORDER BY oc.candidate_id
""")).fetchall()
# 2. Fuer jeden Parent Control die source_regulations sammeln
parent_uuids = list({str(o.parent_control_uuid) for o in obligations if o.parent_control_uuid})
source_types_by_parent: dict[str, list[str]] = {}
if parent_uuids:
# Batch-Query fuer alle Parent-Links
links = db.execute(text("""
SELECT control_uuid::text, source_regulation
FROM control_parent_links
WHERE control_uuid::text = ANY(:uuids)
"""), {"uuids": parent_uuids}).fetchall()
for link in links:
uid = link.control_uuid
src_type = classify_source_regulation(link.source_regulation or "")
source_types_by_parent.setdefault(uid, []).append(src_type)
# 3. Normative strength korrigieren
# 2. Normative strength korrigieren basierend auf source_type
changes = []
stats = {"total": len(obligations), "unchanged": 0, "capped_to_should": 0, "capped_to_can": 0, "no_parent_links": 0}
stats = {"total": len(obligations), "unchanged": 0, "capped_to_should": 0, "capped_to_can": 0, "no_source": 0}
for obl in obligations:
parent_uid = str(obl.parent_control_uuid) if obl.parent_control_uuid else None
source_types = source_types_by_parent.get(parent_uid, []) if parent_uid else []
if not source_types:
stats["no_parent_links"] += 1
if not obl.parent_source:
stats["no_source"] += 1
continue
highest_type = get_highest_source_type(source_types)
new_strength = cap_normative_strength(obl.normative_strength, highest_type)
source_type = classify_source_regulation(obl.parent_source)
new_strength = cap_normative_strength(obl.normative_strength, source_type)
if new_strength != obl.normative_strength:
changes.append({
@@ -990,7 +970,8 @@ async def backfill_normative_strength(
"candidate_id": obl.candidate_id,
"old_strength": obl.normative_strength,
"new_strength": new_strength,
"source_type": highest_type,
"source_type": source_type,
"source_regulation": obl.parent_source,
})
if new_strength == "should":
stats["capped_to_should"] += 1