# mypy: disable-error-code="arg-type,assignment,union-attr" """ Recommendation Service — Auswertung der Empfehlungs-Regeln gegen ein Profil. - Lädt alle live (is_live=1, status='published') Regeln - Lädt Tenant-Overrides - Iteriert Regeln, evaluiert ``conditions`` JSONB gegen Profil-Dict - Wendet Overrides an (Klassifikation überschreiben oder Regel deaktivieren) - Gruppiert in required/recommended/optional mit Begründung """ from typing import Any, Optional from sqlalchemy.orm import Session from compliance.db.template_rule_models import ( TemplateRuleDB, TemplateRuleVersionDB, TenantRuleOverrideDB, ) from compliance.schemas.template_rule import ( RecommendationRequest, RecommendationResult, RecommendedItem, ) # ---------------------------------------------------------------- condition eval def _eval_clause(clause: dict[str, Any], profile: dict[str, Any]) -> bool: """Evaluate eine einzelne Bedingungsklausel gegen das Profil-Dict.""" field = clause.get("field") op = clause.get("op") expected = clause.get("value") if not field or not op: return False actual = profile.get(field) if op == "exists": return actual is not None and actual != "" and actual is not False if op == "eq": return actual == expected if op == "neq": return actual != expected if op == "in": if not isinstance(expected, list): return False return actual in expected if op == "not_in": if not isinstance(expected, list): return True return actual not in expected if op in ("gt", "gte", "lt", "lte"): try: a = _to_number(actual) e = _to_number(expected) except (TypeError, ValueError): return False if op == "gt": return a > e if op == "gte": return a >= e if op == "lt": return a < e if op == "lte": return a <= e if op == "truthy": return bool(actual) if op == "falsy": return not bool(actual) return False def _to_number(v: Any) -> float: if isinstance(v, (int, float)): return float(v) if isinstance(v, str): # Buckets wie "50_249" → nimm untere Schwelle if "_" in v: head = v.split("_")[0] return float(head) return float(v) raise ValueError(f"Cannot coerce to number: {v}") def _eval_condition(condition: dict[str, Any], profile: dict[str, Any]) -> bool: """Evaluate die gesamte Bedingung (all/any über Klauseln).""" if not condition: return True # leere Bedingung → immer wahr kind = condition.get("kind", "all") clauses = condition.get("clauses") or [] if not clauses: return True if kind == "all": return all(_eval_clause(c, profile) for c in clauses) if kind == "any": return any(_eval_clause(c, profile) for c in clauses) return False def _build_reason( rule: TemplateRuleDB, version: TemplateRuleVersionDB, profile: dict[str, Any], override: Optional[TenantRuleOverrideDB], ) -> str: """Menschenlesbare Begründung für den Reviewer im Workspace.""" parts: list[str] = [] conds = version.conditions or {} clauses = conds.get("clauses") or [] matched = [ c for c in clauses if _eval_clause(c, profile) ] if matched: bits = [] for c in matched[:3]: field = c.get("field", "?") op = c.get("op", "?") val = c.get("value", "?") actual = profile.get(field) bits.append(f"{field} ({actual}) {op} {val}") parts.append("Trifft zu: " + "; ".join(bits)) else: parts.append(f"Always-rule für '{rule.document_type}'") if version.source_citation: parts.append(f"Quelle: {version.source_citation}") if override: if override.override_classification: parts.append( f"Tenant-Override: {version.classification} → " f"{override.override_classification} ({override.reason})" ) else: parts.append(f"Tenant-Override: deaktiviert ({override.reason})") return " · ".join(parts) # ---------------------------------------------------------------- service class RecommendationService: def __init__(self, db: Session) -> None: self.db = db def recommend( self, request: RecommendationRequest, tenant_id: Optional[str] = None, ) -> RecommendationResult: profile = dict(request.profile or {}) if request.compliance_depth_level: profile.setdefault("compliance_depth_level", request.compliance_depth_level) # Live-Versionen aller Regeln laden live_versions = ( self.db.query(TemplateRuleVersionDB) .filter(TemplateRuleVersionDB.is_live == 1) .all() ) if not live_versions: return RecommendationResult( required=[], recommended=[], optional=[], profile_used=profile, ) rule_ids = [v.rule_id for v in live_versions] rules = { r.id: r for r in self.db.query(TemplateRuleDB) .filter(TemplateRuleDB.id.in_(rule_ids)) .all() } # Tenant-Overrides overrides: dict[Any, TenantRuleOverrideDB] = {} if tenant_id: for o in ( self.db.query(TenantRuleOverrideDB) .filter(TenantRuleOverrideDB.tenant_id == tenant_id) .filter(TenantRuleOverrideDB.rule_id.in_(rule_ids)) .all() ): overrides[o.rule_id] = o buckets: dict[str, list[RecommendedItem]] = { "required": [], "recommended": [], "optional": [], } for v in live_versions: rule = rules.get(v.rule_id) if not rule: continue if not _eval_condition(v.conditions or {}, profile): continue override = overrides.get(v.rule_id) base_class = v.classification effective_class = base_class if override: if override.override_classification is None: # Override = deaktiviert continue effective_class = override.override_classification reason = _build_reason(rule, v, profile, override) item = RecommendedItem( document_type=rule.document_type, title=rule.title, rule_id=str(rule.id), rule_key=rule.rule_key, classification=effective_class, base_classification=base_class, source_citation=v.source_citation or "", reason=reason, override_applied=override is not None, ) buckets[effective_class].append(item) # Sortiert nach document_type stabil for k in buckets: buckets[k].sort(key=lambda i: i.document_type) return RecommendationResult( required=buckets["required"], recommended=buckets["recommended"], optional=buckets["optional"], profile_used=profile, ) __all__ = ["RecommendationService"]