Add run_summary_export/src/run_summary_export/core.py

run_summary_export/src/run_summary_export/core.py

@@ -0,0 +1,84 @@
+import csv
+import json
+import os
+from pathlib import Path
+from typing import List, Union, Optional
+
+
+class RunSummary:
+    """Repräsentiert die zentralen Kennwerte eines Runs."""
+
+    def __init__(self, run_id: str, p95: float, p99: float, retry_free_count: int, publish_reorder_count: int) -> None:
+        self.run_id = run_id
+        self.p95 = p95
+        self.p99 = p99
+        self.retry_free_count = retry_free_count
+        self.publish_reorder_count = publish_reorder_count
+        self._validate()
+
+    def _validate(self) -> None:
+        assert isinstance(self.run_id, str) and self.run_id, "run_id muss ein nicht-leerer String sein."
+        assert isinstance(self.p95, (float, int)), "p95 muss eine Zahl sein."
+        assert isinstance(self.p99, (float, int)), "p99 muss eine Zahl sein."
+        assert isinstance(self.retry_free_count, int) and self.retry_free_count >= 0, (
+            "retry_free_count muss eine nicht-negative ganze Zahl sein."
+        )
+        assert isinstance(self.publish_reorder_count, int) and self.publish_reorder_count >= 0, (
+            "publish_reorder_count muss eine nicht-negative ganze Zahl sein."
+        )
+
+    def to_dict(self) -> dict:
+        return {
+            "run_id": self.run_id,
+            "p95": float(self.p95),
+            "p99": float(self.p99),
+            "retry_free_count": int(self.retry_free_count),
+            "publish_reorder_count": int(self.publish_reorder_count),
+        }
+
+
+def export_summary(
+    summary_data: Union[RunSummary, List[RunSummary]],
+    format: str,
+    output_path: Optional[str] = None,
+) -> Optional[str]:
+    """Exportiert eine RunSummary- oder RunSummary-Liste in CSV oder JSON.
+
+    Args:
+        summary_data: Eine oder mehrere RunSummary-Instanzen.
+        format: 'csv' oder 'json'.
+        output_path: Zielpfad, optional. Wird standardmäßig aus 'output/run_summary.<format>' abgeleitet.
+
+    Returns:
+        Pfad zur erstellten Datei oder None bei fehlerhaftem Format.
+    """
+    if isinstance(summary_data, RunSummary):
+        data_list = [summary_data]
+    elif isinstance(summary_data, list) and all(isinstance(s, RunSummary) for s in summary_data):
+        data_list = summary_data
+    else:
+        raise TypeError("summary_data muss eine RunSummary-Instanz oder eine Liste davon sein.")
+
+    fmt = format.lower().strip()
+    if fmt not in {"csv", "json"}:
+        return None
+
+    if not output_path:
+        output_dir = Path("output")
+        output_dir.mkdir(parents=True, exist_ok=True)
+        output_path = str(output_dir / f"run_summary.{fmt}")
+
+    # Serialisierung basierend auf Format
+    if fmt == "json":
+        with open(output_path, "w", encoding="utf-8") as f:
+            json.dump([item.to_dict() for item in data_list], f, ensure_ascii=False, indent=2)
+
+    elif fmt == "csv":
+        fieldnames = list(data_list[0].to_dict().keys())
+        with open(output_path, "w", newline="", encoding="utf-8") as f:
+            writer = csv.DictWriter(f, fieldnames=fieldnames)
+            writer.writeheader()
+            for item in data_list:
+                writer.writerow(item.to_dict())
+
+    return output_path