Add gate_v1_function/src/gate_v1_function/core.py

gate_v1_function/src/gate_v1_function/core.py

@@ -0,0 +1,106 @@
+from __future__ import annotations
+import json
+from pathlib import Path
+from typing import List, Optional, Dict, Any
+import pandas as pd
+
+
+class GateDecision:
+    """Datenmodell für GateEntscheidung: decision (str) und reasons (List[str])"""
+
+    def __init__(self, decision: str, reasons: List[str]):
+        assert decision in {"PASS", "REVIEW", "BLOCK"}, f"Ungültige Entscheidung: {decision}"
+        self.decision = decision
+        self.reasons = reasons
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {"decision": self.decision, "reasons": self.reasons}
+
+    def __repr__(self) -> str:
+        return f"GateDecision(decision={self.decision}, reasons={self.reasons})"
+
+
+class DeltaCase:
+    """Repräsentiert eine einzelne Zeile in delta_cases.csv"""
+
+    def __init__(self, stratum: str, from_status: str, to_status: str, unknown_reason: str):
+        self.stratum = stratum
+        self.from_status = from_status
+        self.to_status = to_status
+        self.unknown_reason = unknown_reason
+
+    @classmethod
+    def from_row(cls, row: Dict[str, Any]) -> "DeltaCase":
+        return cls(
+            stratum=str(row.get("stratum", "")),
+            from_status=str(row.get("from_status", "")),
+            to_status=str(row.get("to_status", "")),
+            unknown_reason=str(row.get("unknown_reason", "")),
+        )
+
+
+def _validate_input_files(*paths: Optional[str]) -> None:
+    for p in paths:
+        if p is None:
+            continue
+        path_obj = Path(p)
+        assert path_obj.exists(), f"Eingabedatei nicht gefunden: {p}"
+
+
+def calculate_gate_decision(
+    delta_summary_path: str,
+    delta_cases_path: str,
+    unknown_whitelist_path: Optional[str] = None,
+) -> GateDecision:
+    """Berechnet auf Basis von Delta-Artefakten die Gate-Entscheidung."""
+
+    _validate_input_files(delta_summary_path, delta_cases_path, unknown_whitelist_path)
+
+    # Whitelist laden, falls vorhanden
+    whitelist: List[str] = []
+    if unknown_whitelist_path:
+        with open(unknown_whitelist_path, encoding="utf-8") as f:
+            data = json.load(f)
+            assert isinstance(data, list), "unknown_whitelist.json muss eine Liste von Strings sein"
+            whitelist = [str(item).strip() for item in data]
+
+    # Delta-Summary laden zur eventuellen Kontrolle (aktuell nicht direkt genutzt)
+    with open(delta_summary_path, encoding="utf-8") as f:
+        _ = json.load(f)
+
+    # Delta-Cases als DataFrame laden
+    df = pd.read_csv(delta_cases_path, dtype=str).fillna("")
+    cases: List[DeltaCase] = [DeltaCase.from_row(row) for _, row in df.iterrows()]
+
+    reasons: List[str] = []
+    decision = "PASS"
+
+    # Harte Blocker prüfen
+    for case in cases:
+        if case.from_status == "PASS" and case.to_status in {"FAIL", "Unknown"}:
+            reasons.append(
+                f"Harter Blocker: {case.stratum} wechselt von {case.from_status} zu {case.to_status}."
+            )
+            decision = "BLOCK"
+
+    if decision != "BLOCK":
+        # Soft-Review-Kriterien prüfen
+        for case in cases:
+            if case.from_status == "WARN" and case.to_status == "FAIL":
+                reasons.append(f"Soft Review: {case.stratum} von WARN zu FAIL.")
+            elif (
+                case.to_status == "Unknown"
+                and case.unknown_reason
+                and case.unknown_reason not in whitelist
+            ):
+                reasons.append(
+                    f"Soft Review: {case.stratum} Unknown Grund '{case.unknown_reason}' nicht gelistet."
+                )
+        if reasons:
+            decision = "REVIEW"
+
+    # Keine Gründe → PASS
+    if not reasons:
+        reasons.append("Alle Strata stabil, keine Blocker oder Review-Kriterien gefunden.")
+
+    return GateDecision(decision=decision, reasons=reasons)