feat(use-case-controls): Adressat-Achse — out-of-scope advisory + additiver GOV-Tag
2-Pass-Haiku-Klassifikation (konservativ + Re-Confirm jeder Nicht-unternehmen- Einstufung) der Review-Tier-Atome: wer muss die Pflicht erfuellen? - Migration 155: atom_classification.addressee (unternehmen/oeffentliche_stelle/ aufsichtsbefugnis/staat_eu/dritter/meta), additiv, kein CHECK. [migration-approved] - Service: addressee + applicable + is_gov pro Control; include_out_of_scope-Param (Default false -> out-of-scope advisory ausgeblendet, NIE geloescht); out_of_scope_count. Pure Helper addressee_applicable/addressee_is_gov (+ Tests). - Route: optionaler include_out_of_scope-Query (contract-safe, additiv). - Frontend: GOV-Chip (additiv) + "kein Kunden-Pruefaspekt"-Chip + 1-Klick-Toggle zum Einblenden der out-of-scope-Atome. Daten: 40.859 Adressat-Tags auf macmini geladen (81% applicable, 19% advisory, 3.146 GOV). Konservativ: NULL/Unklar = applicable. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -57,6 +57,11 @@ async def controls_for_use_case(
|
||||
description="atom-grain: 'core'=nur validierte Kern-Pflichten (Default), "
|
||||
"'all'=alle inkl. 'zur Prüfung'-Stufe",
|
||||
),
|
||||
include_out_of_scope: bool = Query(
|
||||
False,
|
||||
description="atom-grain: out-of-scope-Adressaten (Aufsichtsbefugnis/"
|
||||
"Mitgliedstaat/Dritter/meta) einblenden (Default: advisory ausgeblendet)",
|
||||
),
|
||||
limit: int = Query(50, ge=1, le=200),
|
||||
offset: int = Query(0, ge=0),
|
||||
svc: UseCaseControlsService = Depends(get_use_case_controls_service),
|
||||
@@ -64,4 +69,7 @@ async def controls_for_use_case(
|
||||
"""Controls for a topic. Atom-grain (Haiku: relevant + sub_topic) wenn vorhanden,
|
||||
sonst master-grain Seed."""
|
||||
with translate_domain_errors():
|
||||
return svc.controls_for_use_case(use_case, primary_only, limit, offset, sub_topic, tier)
|
||||
return svc.controls_for_use_case(
|
||||
use_case, primary_only, limit, offset, sub_topic, tier,
|
||||
include_out_of_scope,
|
||||
)
|
||||
|
||||
@@ -58,6 +58,23 @@ def source_type(license_rule: Optional[int]) -> str:
|
||||
return "own_library" if license_rule == 3 else "derived"
|
||||
|
||||
|
||||
_OUT_OF_SCOPE_ADDRESSEES = ("aufsichtsbefugnis", "staat_eu", "dritter", "meta")
|
||||
|
||||
|
||||
def addressee_applicable(addressee: Optional[str]) -> bool:
|
||||
"""An obligation is applicable to a (potential) customer unless its addressee
|
||||
is clearly someone else: a supervisory authority's power, a member state / EU
|
||||
institution, a foreign third party, or pure meta. NULL = not yet classified =
|
||||
treated as applicable (conservative — nothing hidden by default)."""
|
||||
return addressee not in _OUT_OF_SCOPE_ADDRESSEES
|
||||
|
||||
|
||||
def addressee_is_gov(addressee: Optional[str]) -> bool:
|
||||
"""Public-body-as-obligor → an additive GOV hint (Kommune/Stadt = potential
|
||||
public-sector customer). The atom keeps its use-case; this is only a tag."""
|
||||
return addressee == "oeffentliche_stelle"
|
||||
|
||||
|
||||
# Representative member (most severe, then lowest control_id) carries the
|
||||
# human-readable title/objective — master_controls.canonical_name is only the
|
||||
# merge token, so we surface a real member control per master.
|
||||
@@ -92,6 +109,7 @@ _LIST_SQL = text("""
|
||||
# seed. Preferred whenever the use-case has been processed.
|
||||
_ATOM_LIST_SQL = text("""
|
||||
SELECT ac.control_uuid, ac.sub_topic, ac.canonical_obligation, ac.relevant,
|
||||
ac.addressee,
|
||||
cc.control_id, cc.title, cc.objective, cc.severity, cc.license_rule,
|
||||
cpl.source_regulation, cpl.source_article
|
||||
FROM atom_classification ac
|
||||
@@ -102,6 +120,8 @@ _ATOM_LIST_SQL = text("""
|
||||
WHERE cpl.control_uuid = ac.control_uuid LIMIT 1
|
||||
) cpl ON true
|
||||
WHERE ac.use_case = :uc AND (:all = true OR ac.relevant = true)
|
||||
AND (:incl_oos = true OR ac.addressee IS NULL
|
||||
OR ac.addressee NOT IN ('aufsichtsbefugnis','staat_eu','dritter','meta'))
|
||||
AND (:sub IS NULL OR ac.sub_topic = :sub)
|
||||
ORDER BY ac.relevant DESC, ac.sub_topic NULLS LAST,
|
||||
CASE cc.severity WHEN 'critical' THEN 0 WHEN 'high' THEN 1
|
||||
@@ -163,6 +183,7 @@ class UseCaseControlsService:
|
||||
offset: int = 0,
|
||||
sub_topic: Optional[str] = None,
|
||||
tier: str = "core",
|
||||
include_out_of_scope: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
"""Controls for ``use_case``. Prefers the atom-grain Haiku classification
|
||||
(precise + sub-topic-organized) when present; falls back to the
|
||||
@@ -170,7 +191,10 @@ class UseCaseControlsService:
|
||||
|
||||
``tier`` (atom-grain only): 'core' = validated obligations only (default,
|
||||
keeps the agent/CRA callers precise); 'all' = everything incl. the
|
||||
'review' tier (shown, flagged) so the human browse view loses nothing."""
|
||||
'review' tier (shown, flagged) so the human browse view loses nothing.
|
||||
``include_out_of_scope``: by default out-of-scope addressees (authority
|
||||
power / member-state / foreign / meta) are hidden (advisory, never
|
||||
deleted); set true to surface them."""
|
||||
if not is_valid_use_case(use_case):
|
||||
raise NotFoundError(f"Unknown use_case '{use_case}'")
|
||||
uc = REGISTRY[use_case]
|
||||
@@ -179,7 +203,8 @@ class UseCaseControlsService:
|
||||
tier = tier if tier in ("core", "all") else "core"
|
||||
|
||||
if self._has_atom_grain(use_case):
|
||||
return self._atom_grain(uc, lim, off, sub_topic, tier)
|
||||
return self._atom_grain(uc, lim, off, sub_topic, tier,
|
||||
bool(include_out_of_scope))
|
||||
|
||||
# --- master-grain fallback (recall seed) ---
|
||||
count_sql = (
|
||||
@@ -226,28 +251,38 @@ class UseCaseControlsService:
|
||||
|
||||
def _atom_grain(
|
||||
self, uc, lim: int, off: int, sub_topic: Optional[str], tier: str = "core",
|
||||
include_out_of_scope: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
all_flag = tier == "all"
|
||||
counts = self.db.execute(text(
|
||||
"SELECT count(*) FILTER (WHERE relevant), "
|
||||
"count(*) FILTER (WHERE NOT relevant) "
|
||||
"count(*) FILTER (WHERE NOT relevant), "
|
||||
"count(*) FILTER (WHERE addressee IN "
|
||||
" ('aufsichtsbefugnis','staat_eu','dritter','meta') "
|
||||
" AND (:all = true OR relevant = true)) "
|
||||
"FROM atom_classification "
|
||||
"WHERE use_case = :uc AND (:sub IS NULL OR sub_topic = :sub)"
|
||||
), {"uc": uc.key, "sub": sub_topic}).first()
|
||||
), {"uc": uc.key, "all": all_flag, "sub": sub_topic}).first()
|
||||
core_count = int((counts[0] if counts else 0) or 0)
|
||||
review_count = int((counts[1] if counts else 0) or 0)
|
||||
total = core_count + review_count if all_flag else core_count
|
||||
oos_count = int((counts[2] if counts else 0) or 0)
|
||||
tier_total = core_count + review_count if all_flag else core_count
|
||||
total = tier_total if include_out_of_scope else tier_total - oos_count
|
||||
facet = {
|
||||
row[0]: int(row[1])
|
||||
for row in self.db.execute(text(
|
||||
"SELECT COALESCE(sub_topic, '(none)'), count(*) "
|
||||
"FROM atom_classification WHERE use_case = :uc "
|
||||
"AND (:all = true OR relevant = true) "
|
||||
"AND (:incl_oos = true OR addressee IS NULL OR addressee NOT IN "
|
||||
" ('aufsichtsbefugnis','staat_eu','dritter','meta')) "
|
||||
"GROUP BY 1 ORDER BY 2 DESC"
|
||||
), {"uc": uc.key, "all": all_flag}).fetchall()
|
||||
), {"uc": uc.key, "all": all_flag,
|
||||
"incl_oos": include_out_of_scope}).fetchall()
|
||||
}
|
||||
rows = self.db.execute(_ATOM_LIST_SQL, {
|
||||
"uc": uc.key, "all": all_flag, "sub": sub_topic, "lim": lim, "off": off,
|
||||
"uc": uc.key, "all": all_flag, "incl_oos": include_out_of_scope,
|
||||
"sub": sub_topic, "lim": lim, "off": off,
|
||||
}).fetchall()
|
||||
controls = [
|
||||
{
|
||||
@@ -263,6 +298,9 @@ class UseCaseControlsService:
|
||||
"relevant": bool(r.relevant),
|
||||
"tier": tier_label(r.relevant),
|
||||
"source_type": source_type(r.license_rule),
|
||||
"addressee": r.addressee,
|
||||
"applicable": addressee_applicable(r.addressee),
|
||||
"is_gov": addressee_is_gov(r.addressee),
|
||||
}
|
||||
for r in rows
|
||||
]
|
||||
@@ -270,6 +308,8 @@ class UseCaseControlsService:
|
||||
"use_case": uc.key, "label": uc.label, "group": uc.group,
|
||||
"granularity": "atom", "tier": tier, "total": int(total),
|
||||
"core_count": core_count, "review_count": review_count,
|
||||
"out_of_scope_count": oos_count,
|
||||
"include_out_of_scope": bool(include_out_of_scope),
|
||||
"limit": lim, "offset": off,
|
||||
"sub_topic": sub_topic, "subtopic_counts": facet, "controls": controls,
|
||||
}
|
||||
|
||||
@@ -9,6 +9,8 @@ import pytest
|
||||
from compliance.domain import NotFoundError
|
||||
from compliance.services.use_case_controls import (
|
||||
UseCaseControlsService,
|
||||
addressee_applicable,
|
||||
addressee_is_gov,
|
||||
relevance_score,
|
||||
source_type,
|
||||
tier_label,
|
||||
@@ -71,3 +73,21 @@ def test_source_type_own_library_vs_derived():
|
||||
assert source_type(1) == "derived"
|
||||
assert source_type(2) == "derived"
|
||||
assert source_type(None) == "derived"
|
||||
|
||||
|
||||
def test_addressee_applicable_defaults_to_true_when_unknown():
|
||||
# NULL / company / public body = applicable (nothing hidden by default)
|
||||
assert addressee_applicable(None) is True
|
||||
assert addressee_applicable("unternehmen") is True
|
||||
assert addressee_applicable("oeffentliche_stelle") is True
|
||||
|
||||
|
||||
def test_addressee_applicable_false_for_out_of_scope():
|
||||
for ad in ("aufsichtsbefugnis", "staat_eu", "dritter", "meta"):
|
||||
assert addressee_applicable(ad) is False
|
||||
|
||||
|
||||
def test_addressee_is_gov_only_for_public_body():
|
||||
assert addressee_is_gov("oeffentliche_stelle") is True
|
||||
assert addressee_is_gov("unternehmen") is False
|
||||
assert addressee_is_gov(None) is False
|
||||
|
||||
Reference in New Issue
Block a user