Add rerun_analysis/src/rerun_analysis/cli.py

rerun_analysis/src/rerun_analysis/cli.py

@@ -0,0 +1,68 @@
+import argparse
+import json
+import logging
+from pathlib import Path
+from typing import List
+
+import pandas as pd
+
+from rerun_analysis.core import AuditRecord, calculate_rerun_effects
+
+
+logging.basicConfig(
+    level=logging.INFO,
+    format='[%(asctime)s] %(levelname)s %(message)s',
+)
+logger = logging.getLogger(__name__)
+
+
+def _load_audit_records(csv_path: Path) -> List[AuditRecord]:
+    if not csv_path.exists():
+        raise FileNotFoundError(f"Audit CSV-Datei nicht gefunden: {csv_path}")
+
+    df = pd.read_csv(csv_path)
+    required_columns = {"stratum", "decision_before", "decision_after"}
+    missing = required_columns - set(df.columns)
+    if missing:
+        raise ValueError(f"Fehlende Spalten in Auditdatei: {missing}")
+
+    records: List[AuditRecord] = []
+    for _, row in df.iterrows():
+        record = AuditRecord(
+            stratum=str(row["stratum"]),
+            pinned=False,  # Pinned-State nicht in CSV, Standardwert False
+            decision_before=str(row["decision_before"]),
+            decision_after=str(row["decision_after"]),
+        )
+        records.append(record)
+    return records
+
+
+def _save_results_json(results_obj, output_path: Path) -> None:
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    with output_path.open('w', encoding='utf-8') as f:
+        json.dump(results_obj.to_json(), f, indent=2, ensure_ascii=False)
+    logger.info(f"Ergebnisse erfolgreich gespeichert: {output_path}")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Analyse von CI-Rerun-Effekten." )
+    parser.add_argument('--input', required=True, help='Pfad zur Audit-CSV-Datei mit CI-Läufen.')
+    parser.add_argument('--output', required=True, help='Pfad, unter dem die Zusammenfassung im JSON-Format gespeichert wird.')
+    args = parser.parse_args()
+
+    input_path = Path(args.input)
+    output_path = Path(args.output)
+
+    logger.info(f"Lade Auditdaten aus {input_path}...")
+    audit_records = _load_audit_records(input_path)
+
+    logger.info("Berechne Rerun-Kennzahlen...")
+    results = calculate_rerun_effects(audit_records)
+
+    logger.info("Speichere Ergebnisse...")
+    _save_results_json(results, output_path)
+
+
+if __name__ == '__main__':
+    main()