Add trace_agg/src/trace_agg/cli.py

trace_agg/src/trace_agg/cli.py

@@ -0,0 +1,71 @@
+import argparse
+import json
+import sys
+from pathlib import Path
+from typing import Any
+
+from trace_agg.core import generate_summary
+
+def _validate_input_file(path: Path) -> None:
+    if not path.exists():
+        raise FileNotFoundError(f"Eingabedatei nicht gefunden: {path}")
+    if not path.is_file():
+        raise ValueError(f"Pfad ist keine Datei: {path}")
+
+def _validate_trace_data(data: Any) -> None:
+    if not isinstance(data, list):
+        raise ValueError("Trace-Daten müssen eine Liste von Dictionaries sein.")
+    for record in data:
+        if not isinstance(record, dict):
+            raise ValueError("Jeder Trace-Datensatz muss ein Dictionary sein.")
+        required_fields = [
+            "run_id",
+            "mischfenster_p50",
+            "mischfenster_p95",
+            "mischfenster_max",
+            "step_order_stability",
+            "read_between_steps",
+        ]
+        for field in required_fields:
+            if field not in record:
+                raise ValueError(f"Fehlendes Feld '{field}' im Trace-Datensatz: {record}")
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Aggregiert Trace-Daten aus A/B-Testläufen zu einer JSON-Zusammenfassung."
+    )
+    parser.add_argument(
+        "--input",
+        required=True,
+        help="Pfad zur Eingabedatei mit Trace-Daten im JSON-Format"
+    )
+    parser.add_argument(
+        "--output",
+        required=True,
+        help="Pfad, unter dem die zusammengefasste JSON-Datei gespeichert wird"
+    )
+
+    args = parser.parse_args()
+
+    input_path = Path(args.input)
+    output_path = Path(args.output)
+
+    try:
+        _validate_input_file(input_path)
+        with input_path.open("r", encoding="utf-8") as f:
+            trace_data = json.load(f)
+        _validate_trace_data(trace_data)
+
+        summary = generate_summary(trace_data)
+
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        with output_path.open("w", encoding="utf-8") as f:
+            json.dump(summary, f, ensure_ascii=False, indent=2)
+
+        print(f"Zusammenfassung erfolgreich erstellt: {output_path}")
+    except Exception as e:
+        print(f"Fehler: {e}", file=sys.stderr)
+        sys.exit(1)
+
+if __name__ == "__main__":
+    main()