Add data_analysis/src/data_analysis/cli.py

data_analysis/src/data_analysis/cli.py

@@ -0,0 +1,46 @@
+import argparse
+import json
+from pathlib import Path
+import pandas as pd
+from data_analysis.core import analyze_data
+
+
+def main() -> None:
+    """CLI-Einstiegspunkt zum Analysieren von Near-Expiry-Daten."""
+    parser = argparse.ArgumentParser(description="Analysiert Near-Expiry-Unpinned Datensätze.")
+    parser.add_argument("--input", required=True, help="Pfad zur Eingabedatei (CSV).")
+    parser.add_argument("--output", required=True, help="Pfad zur Ausgabedatei (JSON).")
+
+    args = parser.parse_args()
+
+    input_path = Path(args.input)
+    output_path = Path(args.output)
+
+    if not input_path.exists():
+        raise FileNotFoundError(f"Eingabedatei nicht gefunden: {input_path}")
+
+    # CSV-Einlesung
+    df = pd.read_csv(input_path)
+
+    # Input-Validierung
+    expected_columns = {"context", "pinned_status", "delta_t", "warn"}
+    if not expected_columns.issubset(df.columns):
+        missing = expected_columns - set(df.columns)
+        raise ValueError(f"Fehlende Spalten im Input: {missing}")
+
+    data = df.to_dict(orient="records")
+
+    # Analyse durchführen
+    result = analyze_data(data)
+
+    # Ergebnis prüfen und schreiben
+    if not isinstance(result, dict):
+        raise TypeError("Analyseergebnis ist kein Dictionary.")
+
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    with open(output_path, "w", encoding="utf-8") as f:
+        json.dump(result, f, indent=2, ensure_ascii=False)
+
+
+if __name__ == "__main__":
+    main()