Add rerun_analysis/src/rerun_analysis/core.py

rerun_analysis/src/rerun_analysis/core.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+import logging
+from dataclasses import dataclass
+from typing import List, Dict
+import pandas as pd
+
+logger = logging.getLogger(__name__)
+
+
+class InvalidAuditDataError(ValueError):
+    """Wird ausgelöst, wenn Eingabedaten ungültig sind."""
+    pass
+
+
+@dataclass
+class AuditRecord:
+    """Repräsentiert einen einzelnen Audit-Datensatz für einen CI-Lauf."""
+    stratum: str
+    pinned: bool
+    decision_before: str
+    decision_after: str
+
+    def __post_init__(self) -> None:
+        valid_decisions = {"PASS", "WARN", "FAIL"}
+        if self.decision_before not in valid_decisions or self.decision_after not in valid_decisions:
+            raise InvalidAuditDataError(
+                f"Ungültige Entscheidung(en) gefunden: before={self.decision_before}, after={self.decision_after}"
+            )
+
+
+@dataclass
+class RerunAnalysisResults:
+    """Aggregierte Ergebnisse der Rerun-Analyse."""
+    helps: int
+    shifts: int
+    hurts: int
+
+    def to_json(self) -> Dict[str, int]:
+        return {"helps": self.helps, "shifts": self.shifts, "hurts": self.hurts}
+
+
+def calculate_rerun_effects(audit_data: List[AuditRecord]) -> RerunAnalysisResults:
+    """Berechnet Kennzahlen zu den Auswirkungen von CI-Reruns.
+
+    Args:
+        audit_data: Liste von AuditRecord-Objekten.
+
+    Returns:
+        RerunAnalysisResults: Aggregiertes Ergebnis.
+    """
+    if not audit_data:
+        logger.warning("Leere Auditdaten übergeben.")
+        return RerunAnalysisResults(helps=0, shifts=0, hurts=0)
+
+    df = pd.DataFrame([record.__dict__ for record in audit_data])
+
+    # Validierungs-Checks
+    required_cols = {"decision_before", "decision_after"}
+    if not required_cols.issubset(df.columns):
+        raise InvalidAuditDataError(f"Fehlende Spalten in Auditdaten: {required_cols - set(df.columns)}")
+
+    # Hilfsfunktion: Klassifizierung der Auswirkungen
+    def classify(before: str, after: str) -> str:
+        mapping = {"PASS": 3, "WARN": 2, "FAIL": 1}
+        before_val, after_val = mapping.get(before, 0), mapping.get(after, 0)
+        if after_val > before_val:
+            return "help"
+        elif after_val < before_val:
+            return "hurt"
+        elif before_val == after_val and before != after:
+            return "shift"
+        else:
+            return "neutral"
+
+    df["effect"] = df.apply(lambda r: classify(r["decision_before"], r["decision_after"]), axis=1)
+
+    helps = int((df["effect"] == "help").sum())
+    shifts = int((df["effect"] == "shift").sum())
+    hurts = int((df["effect"] == "hurt").sum())
+
+    logger.info("Analyse abgeschlossen: helps=%d, shifts=%d, hurts=%d", helps, shifts, hurts)
+
+    result = RerunAnalysisResults(helps=helps, shifts=shifts, hurts=hurts)
+
+    # CI-Ready: Assertions
+    assert all(isinstance(v, int) for v in (result.helps, result.shifts, result.hurts)), \
+        "Ergebniswerte müssen vom Typ int sein"
+
+    return result