Add metrics_reporting/src/metrics_reporting/cli.py

metrics_reporting/src/metrics_reporting/cli.py

@@ -0,0 +1,45 @@
+import argparse
+import json
+from pathlib import Path
+from typing import Any, Dict
+from metrics_reporting import core
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Analyse und Reporting von Log-basierten Metriken zur Driftbewertung.")
+    parser.add_argument("--input", required=True, help="Pfad zur Eingabe-Log-Datei im JSON-Format.")
+    parser.add_argument("--threshold", required=True, type=float, help="Drift-Schwellenwert zur Klassifizierung auffälliger Metriken.")
+    parser.add_argument("--output", required=False, help="Pfad zur Ausgabedatei mit der Metrik-Zusammenfassung.")
+
+    args = parser.parse_args()
+
+    input_path = Path(args.input)
+    if not input_path.exists() or not input_path.is_file():
+        raise FileNotFoundError(f"Eingabe-Datei nicht gefunden: {input_path}")
+
+    try:
+        with input_path.open("r", encoding="utf-8") as f:
+            logs = json.load(f)
+    except json.JSONDecodeError as e:
+        raise ValueError(f"Ungültiges JSON in Eingabedatei: {e}")
+
+    if not isinstance(logs, list) or not all(isinstance(item, dict) for item in logs):
+        raise TypeError("Erwartet wird eine Liste von Log-Dictionaries im JSON-Format.")
+
+    threshold = args.threshold
+    if not isinstance(threshold, (float, int)):
+        raise TypeError("Threshold muss vom Typ float sein.")
+
+    metric_summary: Dict[str, Any] = core.report_metrics(logs, float(threshold))
+
+    if args.output:
+        output_path = Path(args.output)
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        with output_path.open("w", encoding="utf-8") as f:
+            json.dump(metric_summary, f, indent=2, ensure_ascii=False)
+    else:
+        print(json.dumps(metric_summary, indent=2, ensure_ascii=False))
+
+
+if __name__ == "__main__":
+    main()