diff --git a/run_summary_export/src/run_summary_export/core.py b/run_summary_export/src/run_summary_export/core.py new file mode 100644 index 0000000..b5eb0b3 --- /dev/null +++ b/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.' 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