cluster-knaster/core/src/knaster/cli.py

"""Kommandozeile: serve, validate, inventory, placement, registry export, ontology."""

from __future__ import annotations

import argparse
import sys
from pathlib import Path

import yaml


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(prog="knaster")
    sub = parser.add_subparsers(dest="cmd", required=True)

    p_serve = sub.add_parser("serve", help="Router-Service starten")
    p_serve.add_argument("--host", default="0.0.0.0")
    p_serve.add_argument("--port", type=int, default=8080)

    p_val = sub.add_parser("validate", help="Konfiguration prüfen")
    p_val.add_argument("--root", default=".")
    p_val.add_argument("--site", default="knaster")

    p_inv = sub.add_parser("inventory", help="Inventar")
    inv_sub = p_inv.add_subparsers(dest="inv_cmd", required=True)
    p_col = inv_sub.add_parser("collect", help="lokales System auslesen")
    p_col.add_argument("--name")

    p_plan = sub.add_parser("placement", help="Placement-Plan")
    p_plan.add_argument("--inventory", required=True)
    p_plan.add_argument("--catalog", default="catalog")

    p_tts = sub.add_parser("tts-serve", help="OpenAI-kompatiblen TTS-Dienst (Piper o. ä.) starten")
    p_tts.add_argument("--host", default="0.0.0.0")
    p_tts.add_argument("--port", type=int, default=8090)

    p_bench = sub.add_parser("bench", help="Benchmarks für den Deployment-Tag")
    bench_sub = p_bench.add_subparsers(dest="bench_cmd", required=True)
    p_bt = bench_sub.add_parser("tts", help="Real-Time-Factor eines TTS-Endpunkts messen")
    p_bt.add_argument("--url", required=True, help="Basis-URL, z. B. http://litellm:4000 oder http://tts:8090")
    p_bt.add_argument("--model", default="tts-de")
    p_bt.add_argument("--api-key", default="")
    p_bt.add_argument("--runs", type=int, default=3)
    p_bt.add_argument("--text", default=None)

    p_reg = sub.add_parser("registry", help="Registry")
    reg_sub = p_reg.add_subparsers(dest="reg_cmd", required=True)
    reg_sub.add_parser("export", help="effektiven Stand (inkl. Laufzeitzustand) als YAML ausgeben")

    p_onto = sub.add_parser("ontology", help="Ontologie der Datenhaltung (SPEC-ontology)")
    p_onto.add_argument("--path", default=None, help="Ontologie-Datei (Standard: ontology/know.yaml)")
    onto_sub = p_onto.add_subparsers(dest="onto_cmd", required=True)
    onto_sub.add_parser("validate", help="Modell prüfen")
    p_exp = onto_sub.add_parser("export", help="Turtle, Graph-JSON und/oder HTML-Ansicht erzeugen")
    p_exp.add_argument("--ttl")
    p_exp.add_argument("--graph")
    p_exp.add_argument("--html")
    p_exp.add_argument("--template", default=None, help="Standard: viewer.html.tmpl neben der Ontologie")
    p_snap = onto_sub.add_parser("snapshot", help="n8n-Workflows lesen und als JSON speichern (nur GET)")
    p_snap.add_argument("--out", required=True)
    p_chk = onto_sub.add_parser("check", help="Ontologie mit der Datenhaltung in n8n abgleichen")
    p_chk.add_argument("--snapshot", help="Workflows aus Datei statt aus der n8n-API")
    p_chk.add_argument("--json", action="store_true")
    p_app = onto_sub.add_parser("apply", help="freigegebene Entscheidungen in die Ontologie übernehmen")
    p_app.add_argument("--dry-run", action="store_true")

    args = parser.parse_args(argv)

    if args.cmd == "ontology":
        return _ontology(args)

    if args.cmd == "serve":
        import uvicorn

        uvicorn.run("knaster.api:create_app", factory=True, host=args.host, port=args.port)
        return 0

    if args.cmd == "tts-serve":
        import uvicorn

        uvicorn.run("knaster.tts_server:create_tts_app", factory=True, host=args.host, port=args.port)
        return 0

    if args.cmd == "bench":
        from .bench import bench_tts

        result = bench_tts(args.url, args.model, api_key=args.api_key, runs=args.runs, text=args.text)
        print(yaml.safe_dump(result, sort_keys=False, allow_unicode=True))
        return 0 if result["runs"] else 1

    if args.cmd == "validate":
        from .validate import validate_repo

        issues = validate_repo(Path(args.root), args.site)
        for level, msg in issues:
            print(f"{level.upper():8} {msg}")
        errors = [i for i in issues if i[0] == "error"]
        print(f"{len(errors)} Fehler, {len(issues) - len(errors)} Warnungen")
        return 1 if errors else 0

    if args.cmd == "inventory":
        from .inventory.profile import collect

        print(yaml.safe_dump({"nodes": [collect(args.name).model_dump()]}, sort_keys=False, allow_unicode=True))
        return 0

    if args.cmd == "placement":
        from .inventory.placement import load_catalog, load_nodes, plan

        roles, models = load_catalog(args.catalog)
        result = plan(load_nodes(args.inventory), roles, models)
        print(yaml.safe_dump(result.model_dump(), sort_keys=False, allow_unicode=True))
        return 1 if any(not u.optional for u in result.unplaced) else 0

    if args.cmd == "registry":
        from .registry import Registry
        from .settings import Settings
        from .store import Store

        settings = Settings.from_env()
        registry = Registry.from_file(settings.site_dir / "registry.yaml", Store(settings.db_url))
        print(yaml.safe_dump(registry.export(), sort_keys=False, allow_unicode=True))
        return 0
    return 2


def _ontology(args: argparse.Namespace) -> int:
    import json

    from .ontology import (
        Ontology,
        apply_decisions,
        build_snapshot,
        check,
        fetch_workflows,
        to_graph,
        to_html,
        to_turtle,
    )
    from .settings import Settings

    settings = Settings.from_env()
    path = Path(args.path) if args.path else settings.ontology_path
    onto = Ontology.from_file(path)

    if args.onto_cmd == "validate":
        errors = onto.validate_model()
        for e in errors:
            print(f"ERROR    {e}")
        print(f"{len(onto.classes)} Klassen, {len(onto.relations)} Beziehungen, {len(onto.vocabularies)} Vokabulare, "
              f"{len(errors)} Fehler")
        return 1 if errors else 0

    if args.onto_cmd == "export":
        if args.ttl:
            Path(args.ttl).write_text(to_turtle(onto), encoding="utf-8")
        if args.graph:
            Path(args.graph).write_text(json.dumps(to_graph(onto), ensure_ascii=False, indent=1), encoding="utf-8")
        if args.html:
            template = Path(args.template) if args.template else path.parent / "viewer.html.tmpl"
            Path(args.html).write_text(to_html(onto, template), encoding="utf-8")
        return 0

    if args.onto_cmd in {"snapshot", "check"}:
        if getattr(args, "snapshot", None):
            data = json.loads(Path(args.snapshot).read_text(encoding="utf-8"))
            workflows = data.get("data", data) if isinstance(data, dict) else data
        else:
            if not (settings.n8n_host and settings.n8n_api_key):
                print("N8N_HOST und N8N_API_KEY fehlen (oder --snapshot angeben)", file=sys.stderr)
                return 2
            workflows = fetch_workflows(settings.n8n_host, settings.n8n_api_key)
        if args.onto_cmd == "snapshot":
            Path(args.out).write_text(json.dumps({"data": workflows}, ensure_ascii=False), encoding="utf-8")
            print(f"{len(workflows)} Workflows gespeichert in {args.out} (enthält Workflow-Code, nicht committen)")
            return 0
        from .store import Store

        decided = {d["key"] for d in Store(settings.db_url).list_ontology_decisions()}
        findings = check(onto, build_snapshot(workflows), decided=decided)
        if args.json:
            print(json.dumps([f.model_dump() for f in findings], ensure_ascii=False, indent=1))
        else:
            for f in findings:
                mark = "→ Vorschlag" if f.change else ""
                print(f"{f.severity.upper():7} {f.summary} {mark}".rstrip())
            print(f"{len(findings)} Befunde, davon {sum(1 for f in findings if f.change)} Änderungsvorschläge")
        return 1 if any(f.severity == "high" for f in findings) else 0

    if args.onto_cmd == "apply":
        from .store import Store

        store = Store(settings.db_url)
        pending = store.list_ontology_decisions(applied=False)
        applied, messages = apply_decisions(onto, pending)
        for m in messages:
            print(m)
        errors = onto.validate_model()
        if errors:
            print("Ontologie wäre ungültig, nichts gespeichert:\n" + "\n".join(errors), file=sys.stderr)
            return 1
        if applied and not args.dry_run:
            header = "".join(line + "\n" for line in path.read_text(encoding="utf-8").splitlines()
                             if line.startswith("#"))
            onto.save(path, header)
            store.mark_ontology_applied(applied)
            print(f"{len(applied)} Entscheidungen übernommen, Version {onto.version}. "
                  "Danach `knaster ontology export` und committen.")
        return 0
    return 2


if __name__ == "__main__":
    sys.exit(main())