diff --git a/unknown_analysis/src/unknown_analysis/cli.py b/unknown_analysis/src/unknown_analysis/cli.py new file mode 100644 index 0000000..c4c82e0 --- /dev/null +++ b/unknown_analysis/src/unknown_analysis/cli.py @@ -0,0 +1,65 @@ +import argparse +import json +from pathlib import Path +from typing import Any +import sys + +from unknown_analysis import core + + +def _load_json(path: Path) -> list[dict[str, Any]]: + """Loads log data from a JSON file and validates structure.""" + if not path.exists() or not path.is_file(): + raise FileNotFoundError(f"Input file not found: {path}") + with path.open('r', encoding='utf-8') as f: + data = json.load(f) + if not isinstance(data, list): + raise ValueError("Expected log data to be a list of dictionaries.") + for entry in data: + if not isinstance(entry, dict): + raise ValueError("Each log entry must be a dictionary.") + required_fields = {"artifact_key", "status", "cause", "path", "error"} + if not required_fields.issubset(entry.keys()): + raise ValueError(f"Missing required fields in log entry: {entry}") + return data + + +def _write_json(data: Any, path: Path) -> None: + """Writes data as JSON to the specified path.""" + path.parent.mkdir(parents=True, exist_ok=True) + with path.open('w', encoding='utf-8') as f: + json.dump(data, f, indent=2, ensure_ascii=False) + + +def main(argv: list[str] | None = None) -> None: + """Entry point for the Unknown Analysis CLI.""" + parser = argparse.ArgumentParser(description="Unknown Artifact Analysis CLI") + parser.add_argument("--input", required=True, help="Pfad zur JSON-Logdatei") + parser.add_argument("--outdir", required=True, help="Zielverzeichnis für Output-Dateien") + + args = parser.parse_args(argv) + + input_path = Path(args.input) + outdir = Path(args.outdir) + + try: + log_data = _load_json(input_path) + except (FileNotFoundError, ValueError) as e: + print(f"Fehler beim Laden der Logdaten: {e}", file=sys.stderr) + sys.exit(1) + + unknown_rates = core.calculate_unknown_rates(log_data) + top_switches = core.get_top_pass_unknown_switches(log_data) + + rates_path = outdir / "unknown_rates.json" + switches_path = outdir / "top_switches.json" + + _write_json(unknown_rates, rates_path) + _write_json(top_switches, switches_path) + + print(f"Unknown rates saved to: {rates_path}") + print(f"Top switches saved to: {switches_path}") + + +if __name__ == "__main__": + main()