diff --git a/run_summary_export/src/run_summary_export/cli.py b/run_summary_export/src/run_summary_export/cli.py new file mode 100644 index 0000000..b9a996c --- /dev/null +++ b/run_summary_export/src/run_summary_export/cli.py @@ -0,0 +1,71 @@ +import argparse +import json +import os +from pathlib import Path +from typing import Any, List + +from run_summary_export.core import export_summary, RunSummary + + +# CLI entrypoint for Run Summary Export Tool +def _load_input(path: str) -> List[RunSummary]: + with open(path, 'r', encoding='utf-8') as f: + data = json.load(f) + + # validate structure + if isinstance(data, dict): + data = [data] + if not isinstance(data, list): + raise ValueError('Input JSON must be an object or list of objects.') + + summaries: List[RunSummary] = [] + for item in data: + if not isinstance(item, dict): + raise ValueError('Each entry must be a dict.') + missing_fields = [fld for fld in ('run_id', 'p95', 'p99', 'retry_free_count', 'publish_reorder_count') if fld not in item] + if missing_fields: + raise ValueError(f'Missing fields in input: {missing_fields}') + summaries.append( + RunSummary( + run_id=str(item['run_id']), + p95=float(item['p95']), + p99=float(item['p99']), + retry_free_count=int(item['retry_free_count']), + publish_reorder_count=int(item['publish_reorder_count']) + ) + ) + return summaries + + +def main() -> None: + parser = argparse.ArgumentParser(description='Run Summary Export Tool') + parser.add_argument('--input', required=True, help='Pfad zur JSON-Datei mit Run-Summary-Daten.') + parser.add_argument('--format', required=True, choices=['csv', 'json'], help="Ausgabeformat ('csv' oder 'json').") + parser.add_argument('--output', required=False, help='Zielpfad der Exportdatei.') + + args = parser.parse_args() + + input_path = args.input + output_format = args.format.lower() + output_path = args.output + + if not os.path.exists(input_path): + raise FileNotFoundError(f'Input file not found: {input_path}') + + summaries = _load_input(input_path) + + export_path = export_summary(summaries, output_format) + + if output_path and export_path: + # rename/move the generated file to the user-provided output path + os.replace(export_path, output_path) + export_path = output_path + + if export_path is None: + raise SystemExit(f'Export failed: unsupported format {output_format!r}') + + print(f'Export completed successfully -> {export_path}') + + +if __name__ == '__main__': + main()