Add unknown_analysis/src/unknown_analysis/cli.py

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()