Add unknowns_analysis/src/unknowns_analysis/cli.py

unknowns_analysis/src/unknowns_analysis/cli.py

@@ -0,0 +1,73 @@
+import argparse
+import json
+import os
+from pathlib import Path
+import pandas as pd
+from typing import Any
+
+from unknowns_analysis import core
+
+
+def main() -> None:
+    """Command-line entrypoint for Unknowns Analysis script."""
+    parser = argparse.ArgumentParser(
+        description="Analysiert Unknowns und deren Einfluss auf die WARN-Rate."
+    )
+    parser.add_argument(
+        "--unknowns",
+        required=True,
+        help="Pfad zur CSV-Datei mit Unknown-Metriken."
+    )
+    parser.add_argument(
+        "--warns",
+        required=True,
+        help="Pfad zur CSV-Datei mit WARN-Daten."
+    )
+    parser.add_argument(
+        "--output",
+        required=False,
+        default="output/unknowns_summary.json",
+        help="Zielpfad für die JSON-Ergebnisdatei."
+    )
+    args = parser.parse_args()
+
+    unknowns_path = Path(args.unknowns)
+    warns_path = Path(args.warns)
+    output_path = Path(args.output)
+
+    if not unknowns_path.exists():
+        raise FileNotFoundError(f"Unknowns-Datei nicht gefunden: {unknowns_path}")
+    if not warns_path.exists():
+        raise FileNotFoundError(f"WARN-Datei nicht gefunden: {warns_path}")
+
+    # CSV laden und in Listen von Dicts umwandeln
+    try:
+        unknowns_df = pd.read_csv(unknowns_path)
+        warns_df = pd.read_csv(warns_path)
+    except Exception as e:
+        raise ValueError(f"Fehler beim Lesen der CSV-Dateien: {e}") from e
+
+    unknowns_data = unknowns_df.to_dict(orient="records")
+    warn_data = warns_df.to_dict(orient="records")
+
+    # Analyse durchführen
+    results: dict[str, Any] = core.analyze_unknowns(unknowns_data, warn_data)
+
+    # Validierung der Felder (input_validation_required)
+    required_fields = {"total_unknowns", "warn_increases", "warn_stable"}
+    missing = required_fields - results.keys()
+    if missing:
+        raise ValueError(f"Fehlende Felder im Analyseergebnis: {missing}")
+
+    # Output-Verzeichnis sicherstellen
+    os.makedirs(output_path.parent, exist_ok=True)
+
+    # Ergebnis speichern
+    with open(output_path, "w", encoding="utf-8") as f:
+        json.dump(results, f, indent=2, ensure_ascii=False)
+
+    print(f"Analyse abgeschlossen. Ergebnis gespeichert in: {output_path}")
+
+
+if __name__ == "__main__":
+    main()