"""Web UI for HiveMind — Chat, Admin, Training, Sessions."""
from __future__ import annotations

import asyncio
import json
import logging
import base64
from pathlib import Path
from typing import Any

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, UploadFile, File, Form
from fastapi.responses import HTMLResponse, JSONResponse, StreamingResponse

log = logging.getLogger(__name__)

# ── Community: anonymisiertes Query-Tracking (fire-and-forget) ──────────────
def _community_track_query_bg(node: Any, meta: dict) -> None:
    """Sendet das erste Thema aus `meta` asynchron ans Community-API."""
    topics = meta.get("topics") if isinstance(meta, dict) else None
    topic  = (topics[0] if topics else None) or meta.get("intent", "") if meta else ""
    topic  = str(topic).strip().lower()[:100]
    if not topic or topic in {"", "none", "unknown"}:
        return

    async def _post():
        try:
            import httpx
            relay = (getattr(node.config.network, 'relay_servers', None) or ['https://hive.1seele.de'])[0]
            url   = relay.rstrip('/') + '/community.php?action=track-query'
            async with httpx.AsyncClient(timeout=5.0) as client:
                await client.post(url, json={"topic": topic})
        except Exception:
            pass

    try:
        loop = asyncio.get_event_loop()
        if loop.is_running():
            asyncio.create_task(_post())
    except Exception:
        pass

# Load HTML from file or fallback
_HTML_PATH = Path(__file__).parent / "dashboard.html"

# ── Global SSE event bus ─────────────────────────────────────────────
# Jeder /api/events-Client hält eine asyncio.Queue. broadcast_event() legt
# Ereignisse in alle Queues — wird von telegram.py aufgerufen.
_sse_clients: set[asyncio.Queue] = set()


async def broadcast_event(event: dict) -> None:
    """Sende ein Ereignis an alle verbundenen Dashboard-Clients."""
    payload = "data: " + json.dumps(event, ensure_ascii=False) + "\n\n"
    dead: list[asyncio.Queue] = []
    for q in list(_sse_clients):
        try:
            q.put_nowait(payload)
        except asyncio.QueueFull:
            dead.append(q)
    for q in dead:
        _sse_clients.discard(q)


def _marketplace_url(node) -> str:
    """Marketplace-Basis-URL: zuerst config.network.marketplace_url, dann relay_servers[0]."""
    cfg_url = getattr(node.config.network, 'marketplace_url', '') or ''
    if cfg_url:
        return cfg_url.rstrip('/')
    relay = (getattr(node.config.network, 'relay_servers', None) or ['https://hive.1seele.de'])[0]
    return relay.rstrip('/')


def create_web_app(node: Any, private_key_path: Path | None = None) -> FastAPI:
    """Create the web UI application."""
    app = FastAPI(title=f"HiveMind: {node.name}")

    @app.get("/", response_class=HTMLResponse)
    async def index():
        if _HTML_PATH.exists():
            return _HTML_PATH.read_text(encoding="utf-8")
        return "<h1>dashboard.html not found</h1>"

    @app.get("/assets/{filename}")
    async def serve_asset(filename: str):
        from fastapi.responses import FileResponse
        import mimetypes
        asset_dir = Path(__file__).parent / "assets"
        path = asset_dir / filename
        if not path.exists() or not path.is_file():
            from fastapi import HTTPException
            raise HTTPException(status_code=404, detail="Asset not found")
        mt, _ = mimetypes.guess_type(str(path))
        return FileResponse(str(path), media_type=mt or "application/octet-stream")

    # ─── API: Status ─────────────────────────────────────────────
    @app.get("/api/status")
    async def status():
        return node.status

    @app.get("/api/activity")
    async def activity_status():
        """Live-Aktivitätsstatus für die Dashboard-Statusseite."""
        from pathlib import Path as _Path
        act = dict(getattr(node, "_activity", {}))
        large_path = node.config.model.path or ""
        gw = getattr(node, "_telegram_gateway", None)
        return {
            "activity": act,
            "models": {
                "large": {
                    "name": _Path(large_path).name if large_path else "",
                    "loaded": node.model.loaded,
                },
                "small": {
                    "name": "",
                    "loaded": False,
                    "configured": False,
                },
            },
            "telegram": {
                "configured": bool(node.config.gateway.telegram.token),
                "enabled": node.config.gateway.telegram.enabled,
                "running": bool(gw and gw.running),
            },
        }

    @app.get("/api/events")
    async def events():
        """Server-Sent Events stream — liefert Live-Updates an das Dashboard.

        Ereignis-Typen:
          {"type": "session_update", "session_id": "tg-..."}  ← neue Telegram-Nachricht
          {"type": "ping"}                                    ← Keepalive alle 25 s
        """
        q: asyncio.Queue = asyncio.Queue(maxsize=32)
        _sse_clients.add(q)

        async def _stream():
            try:
                # Keepalive-Task: verhindert Browser-Timeout
                async def _ping():
                    while True:
                        await asyncio.sleep(25)
                        try:
                            q.put_nowait("data: {\"type\":\"ping\"}\n\n")
                        except asyncio.QueueFull:
                            break
                ping_task = asyncio.create_task(_ping())
                try:
                    while True:
                        payload = await q.get()
                        yield payload
                finally:
                    ping_task.cancel()
            finally:
                _sse_clients.discard(q)

        return StreamingResponse(_stream(), media_type="text/event-stream",
                                  headers={"Cache-Control": "no-cache",
                                           "X-Accel-Buffering": "no"})

    @app.post("/api/chat")
    async def chat(data: dict):
        msg = data.get("message", "")
        if not msg:
            return {"response": ""}
        try:
            node.set_activity(
                last="Web-Chat-Anfrage erhalten",
                current="Modell generiert Antwort\u2026",
                next_="Antwort an Browser senden",
                source="web",
            )
        except Exception:
            pass
        response_text, meta = await node.chat(msg)
        try:
            node.set_activity(last="Antwort an Browser gesendet", current=None, next_=None)
        except Exception:
            pass
        # ── Anonymisiertes Query-Tracking ──────────────────────────────────
        try:
            _community_track_query_bg(node, meta)
        except Exception:
            pass
        return {"response": response_text, "meta": meta}

    @app.post("/api/chat/stream")
    async def chat_stream(data: dict):
        """Two-phase streaming chat endpoint (SSE).

        Yields Server-Sent Events:
          data: {"type": "fast", "text": "...", "meta": {...}}   ← immediate
          data: {"type": "full", "text": "...", "meta": {...}}   ← detailed
          data: {"type": "done"}
        """
        msg = data.get("message", "")
        session_id = data.get("session_id", "")
        if not msg:
            async def _empty():
                yield "data: {\"type\": \"done\"}\n\n"
            return StreamingResponse(_empty(), media_type="text/event-stream")

        return StreamingResponse(
            node.chat_streaming(msg, session_id=session_id),
            media_type="text/event-stream",
            headers={
                "Cache-Control": "no-cache",
                "X-Accel-Buffering": "no",   # disable nginx buffering if proxied
            },
        )

    @app.post("/api/command")
    async def command(data: dict):
        cmd = data.get("command", "")
        if cmd == "clear":
            # Clear active session history
            s = node.sessions.active
            s.messages.clear()
            node.sessions.save_active()
            return {"result": "Chat geleert."}
        elif cmd == "status":
            return {"result": json.dumps(node.status, indent=2, ensure_ascii=False)}
        elif cmd == "plugins":
            caps = [{"plugin": c.plugin, "name": c.name, "desc": c.description}
                    for c in node.plugins.all_capabilities()]
            return {"result": json.dumps(caps, indent=2, ensure_ascii=False)}
        elif cmd == "stop":
            import asyncio as _asyncio
            async def _request_stop():
                await _asyncio.sleep(0.3)  # HTTP-Antwort zuerst ausliefern
                log.info("/stop-Befehl empfangen – beende HiveMind sauber.")
                stop_ev = getattr(node, "_stop_event", None)
                if stop_ev is not None:
                    stop_ev.set()
                else:
                    # Fallback falls cli.py kein _stop_event gesetzt hat
                    import os, signal as _signal
                    os.kill(os.getpid(), _signal.SIGINT)
            _asyncio.create_task(_request_stop())
            return {"result": "HiveMind wird sauber beendet..."}
        elif cmd == "help":
            return {"result": (
                "Befehle:\\n"
                "  /help      - Diese Hilfe\\n"
                "  /status    - Node-Status\\n"
                "  /plugins   - Geladene Plugins\\n"
                "  /peers     - Verbundene Peers\\n"
                "  /addpeer host:port - Peer hinzufuegen\\n"
                "  /clear     - Chat leeren\\n"
                "  /memory    - Gespeicherte Fakten\\n"
                "  /stop      - HiveMind sauber beenden"
            )}
        elif cmd == "memory":
            facts = node.memory.all_facts
            custom = node.memory.all_custom
            if not facts and not custom:
                return {"result": "Kein Memory gespeichert."}
            lines = ["Gespeicherte Fakten:"]
            for k, v in facts.items():
                lines.append(f"  {k}: {v}")
            if custom:
                lines.append("Notizen:")
                for n in custom:
                    lines.append(f"  - {n}")
            return {"result": "\\n".join(lines)}
        elif cmd == "peers":
            if not node.network:
                return {"result": "Netzwerk nicht aktiviert."}
            peers = [{"name": p.name or p.node_id, "address": p.address,
                       "online": p.online, "hours_ago": round(p.hours_since_seen, 1)}
                     for p in node.network.peers.all_peers]
            return {"result": json.dumps(peers, indent=2, ensure_ascii=False)}
        elif cmd.startswith("addpeer "):
            addr = cmd.split(" ", 1)[1].strip()
            try:
                from hivemind.network.protocol import PeerInfo as _PI
                host, port = _PI.parse_address(addr)
                if node.network:
                    node.network.peers.add_manual(host, port)
                    import asyncio
                    asyncio.create_task(node.network.connect_to(host, int(port)))
                    return {"result": f"Peer hinzugefuegt: {addr}"}
                return {"result": "Netzwerk nicht aktiviert."}
            except Exception as e:
                return {"result": f"Fehler: {e}"}
        return {"result": f"Unbekannter Befehl: {cmd}"}

    # ─── API: Peers (strukturiert) ─────────────────────────────
    @app.get("/api/peers")
    async def list_peers():
        if not node.network:
            return {"peers": []}
        peers = [
            {
                "id": p.address,           # interner Schlüssel, nicht angezeigt
                "name": p.name or p.node_id or p.address,
                "specialization": getattr(p, "specialization", "") or "",
                "expertise_tags": getattr(p, "expertise_tags", []) or [],
                "online": p.online,
                "hours_ago": round(p.hours_since_seen, 1),
            }
            for p in node.network.peers.all_peers
        ]
        # Relay-Nodes ergänzen (nur anzeigen, nicht doppelt wenn schon per P2P verbunden)
        p2p_names = {p["name"] for p in peers}
        if node.relay and node.relay.connected:
            for rn in node.relay.relay_nodes:
                rname = rn.get("name", rn.get("node_id", ""))
                if rname and rname not in p2p_names:
                    peers.append({
                        "id": None,   # Relay-Node: kein manuelles Entfernen möglich
                        "name": rname,
                        "specialization": rn.get("info", {}).get("specialization", ""),
                        "expertise_tags": rn.get("info", {}).get("expertise_tags", []),
                        "online": True,
                        "hours_ago": 0,
                        "relay_only": True,
                    })
        return {"peers": peers}

    @app.post("/api/peers/remove")
    async def remove_peer(data: dict):
        addr = data.get("id", "")
        if not addr or not node.network:
            return {"success": False, "error": "Ungültige Anfrage"}
        ok = node.network.peers.remove(addr)
        return {"success": ok}

    # ─── API: Plugins ─────────────────────────────────────────────

    def _plugin_dir() -> Path:
        raw = node.config.plugins.get("directory", "./plugins")
        raw = raw.lstrip("./")
        return node.base_dir / (raw or "plugins")

    @app.get("/api/plugins")
    async def list_plugins():
        enabled_names: list = node.config.plugins.get("enabled", [])
        builtin = [node.plugins.get_status(n, enabled_names) for n in node.plugins.BUILTIN]
        community = node.plugins.installed_community(str(_plugin_dir()))
        for p in community:
            p["enabled"] = p["name"] in enabled_names
        return {"builtin": builtin, "community": community}

    @app.post("/api/plugins/toggle")
    async def toggle_plugin(data: dict):
        import yaml as _yaml
        name = data.get("name", "")
        enabled = bool(data.get("enabled", False))
        if not name:
            return JSONResponse({"error": "Name fehlt"}, 400)
        config_path = node.base_dir / "config.yaml"
        cfg: dict = {}
        if config_path.exists():
            cfg = _yaml.safe_load(config_path.read_text()) or {}
        cfg.setdefault("plugins", {})
        enabled_list: list = list(cfg["plugins"].get("enabled",
                                  node.config.plugins.get("enabled", [])))
        pdir = str(_plugin_dir())
        if enabled:
            if name not in enabled_list:
                enabled_list.append(name)
            await node.plugins.enable_live(name, pdir)
        else:
            enabled_list = [n for n in enabled_list if n != name]
            await node.plugins.disable_live(name)
        cfg["plugins"]["enabled"] = enabled_list
        node.config.plugins["enabled"] = enabled_list
        config_path.write_text(_yaml.dump(cfg, allow_unicode=True, default_flow_style=False))
        return {"success": True, "loaded": node.plugins.loaded}

    @app.post("/api/plugins/install")
    async def install_plugin(data: dict):
        import yaml as _yaml
        slug = data.get("slug", "")
        download_url = data.get("download_url", "")
        if not slug or not download_url:
            return JSONResponse({"error": "slug und download_url erforderlich"}, 400)
        pdir = str(_plugin_dir())
        ok, error = await node.plugins.install_from_url(slug, download_url, pdir)
        if not ok:
            return JSONResponse({"error": error}, 400)
        # Persist and enable live
        config_path = node.base_dir / "config.yaml"
        cfg: dict = {}
        if config_path.exists():
            cfg = _yaml.safe_load(config_path.read_text()) or {}
        cfg.setdefault("plugins", {})
        enabled_list: list = list(cfg["plugins"].get("enabled",
                                  node.config.plugins.get("enabled", [])))
        if slug not in enabled_list:
            enabled_list.append(slug)
        cfg["plugins"]["enabled"] = enabled_list
        node.config.plugins["enabled"] = enabled_list
        config_path.write_text(_yaml.dump(cfg, allow_unicode=True, default_flow_style=False))
        await node.plugins.enable_live(slug, pdir)
        # Notify marketplace of download (best-effort)
        try:
            import json as _json
            from urllib.request import urlopen, Request as _Req
            relay = _marketplace_url(node)
            _req = _Req(f"{relay}/marketplace/api.php",
                        data=_json.dumps({"action": "download", "slug": slug}).encode(),
                        headers={"Content-Type": "application/json", "User-Agent": "HiveMind"},
                        method="POST")
            urlopen(_req, timeout=5)
        except Exception:
            pass
        return {"success": True}

    @app.post("/api/plugins/uninstall")
    async def uninstall_plugin(data: dict):
        import yaml as _yaml
        name = data.get("name", "")
        if not name:
            return JSONResponse({"error": "Name fehlt"}, 400)
        if name in node.plugins.BUILTIN:
            return JSONResponse({"error": "Built-in Plugins können nicht deinstalliert werden"}, 400)
        await node.plugins.disable_live(name)
        config_path = node.base_dir / "config.yaml"
        cfg: dict = {}
        if config_path.exists():
            cfg = _yaml.safe_load(config_path.read_text()) or {}
        cfg.setdefault("plugins", {})
        cfg["plugins"]["enabled"] = [n for n in cfg["plugins"].get("enabled", []) if n != name]
        node.config.plugins["enabled"] = cfg["plugins"]["enabled"]
        config_path.write_text(_yaml.dump(cfg, allow_unicode=True, default_flow_style=False))
        node.plugins.uninstall_community(name, str(_plugin_dir()))
        return {"success": True}

    @app.get("/api/marketplace")
    async def marketplace_list(q: str = "", tag: str = "", page: int = 1):
        import json as _json
        from urllib.request import urlopen, Request as _Req
        relay = _marketplace_url(node)
        params = f"?action=list&q={q}&tag={tag}&page={page}&per_page=12"
        try:
            req = _Req(f"{relay}/marketplace/api.php{params}",
                       headers={"User-Agent": "HiveMind"})
            raw = await asyncio.get_event_loop().run_in_executor(
                None, lambda: urlopen(req, timeout=10).read())
            return JSONResponse(_json.loads(raw))
        except Exception as e:
            return JSONResponse({"error": str(e), "plugins": [], "total": 0, "tags": []})

    @app.post("/api/marketplace/submit")
    async def marketplace_submit(data: dict):
        import json as _json
        from urllib.request import urlopen, Request as _Req
        relay = _marketplace_url(node)
        payload = _json.dumps({**data, "action": "submit"}).encode()
        try:
            req = _Req(f"{relay}/marketplace/api.php", data=payload,
                       headers={"Content-Type": "application/json", "User-Agent": "HiveMind"},
                       method="POST")
            raw = await asyncio.get_event_loop().run_in_executor(
                None, lambda: urlopen(req, timeout=10).read())
            return JSONResponse(_json.loads(raw))
        except Exception as e:
            return JSONResponse({"error": str(e)}, 500)

    # ─── API: Sessions ─────────────────────────────────────
    @app.get("/api/sessions")
    async def list_sessions():
        return {"sessions": node.sessions.list_sessions(), "active": node.sessions.active_id}

    @app.post("/api/sessions/create")
    async def create_session(data: dict = None):
        name = (data or {}).get("name", "")
        s = node.sessions.create(name)
        return {"session": s.summary_info}

    @app.post("/api/sessions/switch")
    async def switch_session(data: dict):
        s = node.sessions.switch(data.get("id", ""))
        if s:
            return {"session": s.summary_info, "messages": s.messages}
        return JSONResponse({"error": "Session nicht gefunden"}, 404)

    @app.post("/api/sessions/rename")
    async def rename_session(data: dict):
        ok = node.sessions.rename(data.get("id", ""), data.get("name", ""))
        return {"success": ok}

    @app.delete("/api/sessions/{session_id}")
    async def delete_session(session_id: str):
        ok = node.sessions.delete(session_id)
        return {"success": ok}

    @app.get("/api/sessions/history")
    async def session_history():
        return {"messages": node.sessions.get_history(), "session": node.sessions.active.summary_info}

    # ─── API: Memory ─────────────────────────────────────────────
    @app.get("/api/memory")
    async def get_memory():
        return {"facts": node.memory.all_facts, "custom": node.memory.all_custom, "stats": node.memory.stats}

    @app.post("/api/memory/fact")
    async def add_memory_fact(data: dict):
        node.memory.add_fact(data.get("category", ""), data.get("value", ""))
        return {"success": True}

    @app.delete("/api/memory/fact/{category}")
    async def del_memory_fact(category: str):
        return {"success": node.memory.remove_fact(category)}

    @app.post("/api/memory/custom")
    async def add_memory_custom(data: dict):
        node.memory.add_custom(data.get("note", ""))
        return {"success": True}

    @app.delete("/api/memory/custom/{index}")
    async def del_memory_custom(index: int):
        return {"success": node.memory.remove_custom(index)}

    # ─── Favicon ──────────────────────────────────────────────────
    @app.get("/favicon.ico", include_in_schema=False)
    async def favicon():
        # Simple brain emoji SVG as favicon
        svg = ('<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 100 100">'
               '<text y=".9em" font-size="90">\U0001f9e0</text></svg>')
        return HTMLResponse(content=svg, media_type="image/svg+xml")

    # ─── API: Network Mode (Online/Offline) ────────────────────
    @app.get("/api/network/mode")
    async def network_mode():
        online = bool(node.network or getattr(node, 'relay', None))
        enabled = getattr(getattr(node.config, 'network', None), 'enabled', True)
        return {"online": online, "network_enabled": enabled}

    @app.post("/api/network/mode")
    async def set_network_mode(data: dict):
        import yaml
        go_online = data.get("online", True)
        if go_online and not node.network and not getattr(node, 'relay', None):
            node.config.network.enabled = True
            try:
                from hivemind.network.peerlist import PeerList
                from hivemind.network.peer import P2PNetwork
                from hivemind.network.updater import AutoUpdater
                peers = PeerList(node.base_dir / "data" / "peers.json")
                for addr in (node.config.network.bootstrap_nodes or []):
                    from hivemind.network.protocol import PeerInfo as _PI
                    host, port = _PI.parse_address(addr)
                    peers.add_manual(host, port)
                network = P2PNetwork(
                    node=node, peer_list=peers,
                    listen_port=node.config.network.listen_port,
                )
                updater = AutoUpdater(node, node.base_dir)
                node.updater = updater
                updater.register_handlers(network)
                from hivemind.network.protocol import MsgType
                network.on_message(MsgType.QUERY, node._handle_network_query)
                network.on_message(MsgType.RESPONSE, node._handle_network_response)
                await network.start()
                node.network = network
            except Exception as e:
                log.warning("Failed to start P2P: %s", e)
            relay_urls = getattr(node.config.network, 'relay_servers', []) or []
            if relay_urls and not getattr(node, 'relay', None):
                try:
                    from hivemind.network.relay_client import RelayConnection
                    node.relay = RelayConnection(
                        relay_url=relay_urls[0], node=node,
                        on_message=node._handle_relay_message,
                        on_peers=node._handle_relay_peers,
                    )
                    await node.relay.start()
                except Exception as e:
                    log.warning("Failed to start relay: %s", e)
        elif not go_online:
            node.config.network.enabled = False
            relay = getattr(node, 'relay', None)
            if relay:
                await relay.stop()
                node.relay = None
            if node.network:
                await node.network.stop()
                node.network = None
        # Persist
        config_path = node.base_dir / "config.yaml"
        cfg = yaml.safe_load(config_path.read_text()) if config_path.exists() else {}
        cfg = cfg or {}
        cfg.setdefault("network", {})["enabled"] = go_online
        config_path.write_text(yaml.dump(cfg, allow_unicode=True, default_flow_style=False))
        return {"online": go_online, "success": True}

    # ─── API: Public IP ──────────────────────────────────────────
    @app.get("/api/network/myaddress")
    async def my_address():
        import asyncio
        from urllib.request import urlopen, Request
        port = node.config.network.listen_port if hasattr(node.config, 'network') else 9420

        def _fetch_ip(url, timeout=5):
            try:
                return urlopen(Request(url, headers={"User-Agent": "HiveMind"}), timeout=timeout).read().decode().strip()
            except Exception:
                return None

        loop = asyncio.get_event_loop()
        # Fetch IPv4 and IPv6 in parallel
        ipv4_task = loop.run_in_executor(None, lambda: _fetch_ip("https://api4.ipify.org"))
        ipv6_task = loop.run_in_executor(None, lambda: _fetch_ip("https://api6.ipify.org"))
        ipv4 = await ipv4_task
        ipv6 = await ipv6_task

        # Format addresses
        from hivemind.network.protocol import PeerInfo as _PI
        result = {"port": port, "ipv4": ipv4 or None, "ipv6": ipv6 or None}
        if ipv4:
            result["address_v4"] = f"{ipv4}:{port}"
        if ipv6:
            result["address_v6"] = _PI.format_address(ipv6, port)
        # Primary address: prefer IPv6 if available (DS-Lite compatibility)
        if ipv6:
            result["address"] = result["address_v6"]
            result["ip"] = ipv6
        elif ipv4:
            result["address"] = result["address_v4"]
            result["ip"] = ipv4
        else:
            result["address"] = "unbekannt"
            result["ip"] = "unbekannt"
        # DS-Lite detection: if IPv4 looks like CGN (100.64.x.x-range or shared)
        if ipv4 and ipv6 and ipv4 != ipv6:
            result["ds_lite"] = True
            result["hint"] = "DS-Lite erkannt: IPv4 ist geteilt (CGN). Nutze die IPv6-Adresse!"
        return result

    # ─── API: Profile ────────────────────────────────────────────
    @app.get("/api/profile")
    async def get_profile():
        return {
            "name": node.name,
            "specialization": node.config.node.specialization,
            "expertise_tags": node.config.node.expertise_tags,
            "model": Path(node.config.model.path).stem if node.config.model.path else "",
        }

    @app.post("/api/profile")
    async def save_profile(data: dict):
        import yaml
        node.config.node.specialization = data.get("specialization", "")
        node.config.node.expertise_tags = data.get("expertise_tags", [])
        from hivemind.confidence import ConfidenceScorer
        node.scorer = ConfidenceScorer(
            expertise_tags=node.config.node.expertise_tags,
            specialization=node.config.node.specialization,
        )
        config_path = node.base_dir / "config.yaml"
        cfg = {}
        if config_path.exists():
            cfg = yaml.safe_load(config_path.read_text()) or {}
        if "node" not in cfg:
            cfg["node"] = {}
        cfg["node"]["specialization"] = node.config.node.specialization
        cfg["node"]["expertise_tags"] = node.config.node.expertise_tags
        config_path.write_text(yaml.dump(cfg, allow_unicode=True, default_flow_style=False))
        # Relay sofort mit neuen Profildaten aktualisieren
        if node.relay and node.relay.connected:
            try:
                await node.relay.reregister()
            except Exception as _re:
                log.warning("Relay re-register nach Profiländerung fehlgeschlagen: %s", _re)
        return {"success": True}

    # ─── API: Training ───────────────────────────────────────────
    @app.get("/api/training/stats")
    async def training_stats():
        return node.training.stats

    @app.post("/api/training/export")
    async def training_export(data: dict = None):
        fmt = (data or {}).get("format", "chatml")
        path = node.training.export_jsonl(format_type=fmt)
        return {"path": str(path), "format": fmt}

    @app.get("/api/training/topics")
    async def training_topics():
        return node.training.topic_analysis()

    @app.get("/api/training/lora")
    async def list_lora():
        return {"adapters": node.training.get_lora_adapters()}

    @app.post("/api/training/lora/upload")
    async def upload_lora(file: UploadFile = File(...)):
        data = await file.read()
        path = node.training.save_lora(file.filename or "adapter.gguf", data)
        return {"success": True, "path": str(path), "size_mb": round(len(data) / 1048576, 1)}

    @app.post("/api/training/ingest-url")
    async def training_ingest_url(data: dict):
        """Fetch an article URL, generate Q&A pairs with the local model,
        save them as training data, and optionally add the article to RAG."""
        url = (data.get("url") or "").strip()
        if not url.startswith("http"):
            return JSONResponse(
                {"error": "Ungültige URL — bitte eine http/https-Adresse eingeben."},
                status_code=400,
            )
        n_pairs = max(3, min(int(data.get("n_pairs") or 10), 30))
        also_rag = bool(data.get("also_rag", True))
        try:
            result = await node.training.ingest_url(url, node.model, n_pairs)
            raw_text = result.pop("raw_text", "")
            if also_rag and raw_text:
                title = result.get("title", "Artikel")
                rag_chunks = node.rag.add_document(
                    f"[Artikel] {title}", content=raw_text
                )
                result["rag_chunks"] = rag_chunks
            return result
        except Exception as exc:
            return JSONResponse({"error": str(exc)}, status_code=400)

    # ─── API: RAG ────────────────────────────────────────────────
    @app.get("/api/rag/documents")
    async def rag_documents():
        return {"documents": node.rag.document_list, "stats": node.rag.stats}

    @app.post("/api/rag/upload")
    async def rag_upload(file: UploadFile = File(...)):
        content = (await file.read()).decode("utf-8", errors="replace")
        name = file.filename or "document.txt"
        chunks = node.rag.add_document(name, content=content)
        return {"success": True, "name": name, "chunks": chunks}

    @app.delete("/api/rag/document/{name}")
    async def rag_delete(name: str):
        return {"success": node.rag.remove_document(name)}

    @app.post("/api/rag/import/huggingface")
    async def rag_import_hf(data: dict):
        url = data.get("url", "")
        if not url:
            return JSONResponse({"error": "URL fehlt"}, 400)
        max_rows = data.get("max_rows", 5000)
        text_fields = data.get("text_fields", None)
        split = data.get("split", "train")
        import asyncio
        result = await asyncio.get_event_loop().run_in_executor(
            None, lambda: node.rag.import_huggingface(url, max_rows, text_fields, split))
        if "error" in result:
            return JSONResponse(result, 400)
        return result

    @app.post("/api/rag/search")
    async def rag_search(data: dict):
        query = data.get("query", "")
        return {"results": node.rag.search(query, top_k=data.get("top_k", 5))}

    # ─── API: Keys & Updates ─────────────────────────────────────
    @app.get("/api/keys")
    async def get_keys():
        from hivemind.network.updater import PUBLISHER_PUBLIC_KEY
        priv = ""
        if private_key_path and private_key_path.exists():
            priv = private_key_path.read_text().strip()
        return {"public_key": PUBLISHER_PUBLIC_KEY, "private_key_available": bool(priv),
                "private_key_path": str(private_key_path) if private_key_path else ""}

    @app.post("/api/update/publish")
    async def publish_update(file: UploadFile = File(...), version: str = Form(...), changelog: str = Form("")):
        from hivemind.network.updater import sign_update
        from hivemind.network.protocol import Message, MsgType
        if not private_key_path or not private_key_path.exists():
            return JSONResponse({"error": "Private Key nicht gefunden."}, 400)
        private_key = private_key_path.read_text().strip()
        updates_dir = node.base_dir / "updates"
        updates_dir.mkdir(exist_ok=True)
        zip_path = updates_dir / f"hivemind-{version}.zip"
        content = await file.read()
        zip_path.write_bytes(content)
        try:
            manifest = sign_update(zip_path, private_key, version, changelog)
        except Exception as e:
            return JSONResponse({"error": str(e)}, 500)
        (updates_dir / f"hivemind-{version}.json").write_text(json.dumps(manifest, indent=2))
        sent = 0
        update_payload = {"manifest": manifest, "data_b64": base64.b64encode(content).decode()}
        # Send via direct P2P
        if node.network:
            sent += await node.network.broadcast(Message(type=MsgType.UPDATE_DATA, sender_id=node.id,
                payload=update_payload))
        # Send via Relay
        if node.relay and node.relay.connected:
            relay_ok = await node.relay.broadcast({
                "type": "UPDATE_DATA",
                "manifest": manifest,
                "data_b64": update_payload["data_b64"],
            })
            if relay_ok:
                sent += node.relay.status.get("relay_nodes", 0) or 1
        return {"success": True, "version": version, "sha256": manifest["sha256"],
                "signature": manifest["signature"][:40] + "...", "peers_notified": sent}

    # ─── API: Pending Update ───────────────────────────────────────
    @app.get("/api/update/pending")
    async def update_pending():
        if node.updater and node.updater.pending_update:
            return {"pending": True, **node.updater.pending_update}
        return {"pending": False}

    @app.post("/api/update/apply")
    async def update_apply():
        if not node.updater:
            return JSONResponse({"error": "Updater nicht verfuegbar"}, 400)
        result = await node.updater.apply_pending()
        if "error" in result:
            return JSONResponse(result, 400)
        return result

    @app.post("/api/update/dismiss")
    async def update_dismiss():
        if node.updater:
            node.updater.dismiss_pending()
        return {"success": True}

    # ─── API: PDF Download ─────────────────────────────────────────
    @app.get("/api/pdf/{filename}")
    async def pdf_download(filename: str):
        from fastapi.responses import FileResponse
        pdf_dir = node.base_dir / "data" / "pdf"
        path = pdf_dir / filename
        if not path.exists() or not filename.endswith(".pdf"):
            return JSONResponse({"error": "PDF nicht gefunden"}, 404)
        return FileResponse(path, media_type="application/pdf", filename=filename)

    # ─── API: Relay Status ───────────────────────────────────────
    @app.get("/api/relay/info")
    async def relay_info():
        result = {"relay_client": None}
        if node.relay:
            result["relay_client"] = node.relay.status
        return result

    # ─── API: Telegram Bridge ────────────────────────────────────
    @app.get("/api/telegram/status")
    async def telegram_status():
        import asyncio
        from urllib.request import urlopen, Request
        import json as _json
        token = node.config.gateway.telegram.token
        enabled = node.config.gateway.telegram.enabled
        allowed_users = node.config.gateway.telegram.allowed_users
        gw = getattr(node, "_telegram_gateway", None)
        bot_running = bool(gw and gw.running)
        result: dict = {
            "enabled": enabled,
            "token_set": bool(token),
            "token_masked": ("***" + token[-6:]) if len(token) > 6 else ("*" * len(token)) if token else "",
            "running": bot_running,
            "bot_info": None,
            "bot_error": None,
            "allowed_users": allowed_users,
        }
        if token:
            def _getme():
                try:
                    req = Request(f"https://api.telegram.org/bot{token}/getMe",
                                  headers={"User-Agent": "HiveMind"})
                    return _json.loads(urlopen(req, timeout=5).read())
                except Exception as exc:
                    return {"ok": False, "error": str(exc)}
            data = await asyncio.get_event_loop().run_in_executor(None, _getme)
            if data.get("ok"):
                result["bot_info"] = data["result"]
            else:
                result["bot_error"] = data.get("description") or data.get("error", "Unbekannter Fehler")
        return result

    @app.post("/api/telegram/test")
    async def telegram_test_token(data: dict):
        import asyncio
        from urllib.request import urlopen, Request
        import json as _json
        token = (data.get("token") or "").strip()
        if not token:
            return JSONResponse({"error": "Kein Token angegeben."}, 400)
        def _getme():
            try:
                req = Request(f"https://api.telegram.org/bot{token}/getMe",
                              headers={"User-Agent": "HiveMind"})
                return _json.loads(urlopen(req, timeout=8).read())
            except Exception as exc:
                return {"ok": False, "error": str(exc)}
        resp = await asyncio.get_event_loop().run_in_executor(None, _getme)
        if resp.get("ok"):
            return {"valid": True, "bot": resp["result"]}
        return {"valid": False, "error": resp.get("description") or resp.get("error", "Token ungültig.")}

    @app.post("/api/telegram/save")
    async def telegram_save_config(data: dict):
        import yaml as _yaml
        token = (data.get("token") or "").strip()
        enabled = data.get("enabled")
        allowed_users = data.get("allowed_users", None)
        config_path = node.base_dir / "config.yaml"
        cfg: dict = {}
        if config_path.exists():
            cfg = _yaml.safe_load(config_path.read_text()) or {}
        cfg.setdefault("gateway", {}).setdefault("telegram", {})
        if token:
            cfg["gateway"]["telegram"]["token"] = token
            node.config.gateway.telegram.token = token
        if enabled is not None:
            cfg["gateway"]["telegram"]["enabled"] = bool(enabled)
            node.config.gateway.telegram.enabled = bool(enabled)
        if allowed_users is not None:
            clean = [int(u) for u in allowed_users if str(u).strip().lstrip("-").isdigit()]
            cfg["gateway"]["telegram"]["allowed_users"] = clean
            node.config.gateway.telegram.allowed_users = clean
        config_path.write_text(_yaml.dump(cfg, allow_unicode=True, default_flow_style=False))
        return {"success": True}

    # Lock verhindert gleichzeitige concurrent Aufrufe von /api/telegram/start,
    # die das "two getUpdates"-Konflikt-Problem verursachen würden.
    _tg_start_lock = asyncio.Lock()

    @app.post("/api/telegram/start")
    async def telegram_start():
        token = node.config.gateway.telegram.token
        if not token:
            return JSONResponse({"error": "Kein Token konfiguriert. Bitte zuerst Token speichern."}, 400)

        async with _tg_start_lock:
            gw = getattr(node, "_telegram_gateway", None)
            if gw and gw.running:
                return {"success": True, "already_running": True}

            # Alten Gateway signalisieren zu stoppen (setzt _stop_event).
            # start()'s finally-Block übernimmt den vollständigen PTB-Cleanup
            # inklusive updater.stop() — das wartet auf den laufenden Long-Poll.
            if gw:
                try:
                    await gw.stop()
                except Exception:
                    pass
                node._telegram_gateway = None

            # Auf alten Task warten (max. 15 s): stellt sicher, dass PTB vollständig
            # heruntergefahren ist und keine offene getUpdates-Verbindung bei Telegram
            # mehr existiert, bevor die neue Instanz startet.
            old_task = getattr(node, "_telegram_gateway_task", None)
            if old_task and not old_task.done():
                try:
                    await asyncio.wait_for(asyncio.shield(old_task), timeout=15.0)
                except (asyncio.TimeoutError, asyncio.CancelledError, Exception):
                    old_task.cancel()
                    try:
                        await old_task
                    except (asyncio.CancelledError, Exception):
                        pass
            node._telegram_gateway_task = None

            try:
                from hivemind.gateway.telegram import TelegramGateway

                gw = TelegramGateway(token=token, node=node)
                node._telegram_gateway = gw

                async def _run():
                    try:
                        await gw.start()
                    except Exception as exc:
                        log.error("Telegram gateway task failed: %s", exc, exc_info=True)
                    finally:
                        # Gateway-Referenz immer aufräumen wenn Task endet.
                        if node._telegram_gateway is gw:
                            node._telegram_gateway = None

                node._telegram_gateway_task = asyncio.create_task(_run())
                node.config.gateway.telegram.enabled = True

                # Auf vollständigen Start warten (max. 5 s): polling aktiv = running True.
                # Verhindert, dass status-Polling im Fenster zwischen create_task und
                # _started=True ein "running: false" zurückgibt und einen Neustart triggert.
                for _ in range(50):
                    if gw.running:
                        break
                    await asyncio.sleep(0.1)

                return {"success": True}
            except Exception as exc:
                node._telegram_gateway = None
                return JSONResponse({"error": str(exc)}, 500)

    @app.post("/api/telegram/stop")
    async def telegram_stop():
        gw = getattr(node, "_telegram_gateway", None)
        task = getattr(node, "_telegram_gateway_task", None)
        if gw:
            try:
                await gw.stop()  # setzt _stop_event; start()'s finally räumt PTB auf
            except Exception:
                pass
            node._telegram_gateway = None
        # Auf Task warten statt ihn sofort canceln — sonst wird der PTB-Cleanup
        # (updater.stop / app.shutdown) durch CancelledError unterbrochen.
        if task and not task.done():
            try:
                await asyncio.wait_for(asyncio.shield(task), timeout=15.0)
            except (asyncio.TimeoutError, asyncio.CancelledError, Exception):
                task.cancel()
                try:
                    await task
                except (asyncio.CancelledError, Exception):
                    pass
        node._telegram_gateway_task = None
        node.config.gateway.telegram.enabled = False
        return {"success": True}

    # ─── WebSocket ───────────────────────────────────────────────
    @app.websocket("/ws/chat")
    async def ws_chat(ws: WebSocket):
        await ws.accept()
        try:
            while True:
                msg = await ws.receive_text()
                response_text, meta = await node.chat(msg)
                await ws.send_text(json.dumps({"response": response_text, "meta": meta}, ensure_ascii=False))
        except WebSocketDisconnect:
            pass

    # ─── Crash-Detection ─────────────────────────────────────────
    @app.get("/api/crash-detected")
    async def crash_detected():
        """Gibt zurück, ob HiveMind beim letzten Start gecrasht ist."""
        crashed = getattr(node, "_crashed_on_startup", False)
        log_path = getattr(node, "_last_log_path", "")
        return {"crashed": bool(crashed), "log_path": log_path}

    @app.post("/api/crash-detected/dismiss")
    async def crash_dismiss():
        """Crash-Flag zurücksetzen (nach Modal-Schließen)."""
        node._crashed_on_startup = False
        return {"success": True}

    # ─── Community (Proxy zum Relay-Server) ───────────────────────
    def _community_url(action: str) -> str:
        base = _marketplace_url(node)
        return f"{base}/community.php?action={action}"

    @app.get("/api/community/stats")
    async def community_stats():
        try:
            import httpx
            async with httpx.AsyncClient(timeout=8.0) as client:
                r = await client.get(_community_url("stats"))
                return JSONResponse(r.json(), status_code=r.status_code)
        except Exception as exc:
            return JSONResponse({"error": str(exc)}, 503)

    @app.get("/api/community/top-queries")
    async def community_top_queries(limit: int = 10):
        try:
            import httpx
            async with httpx.AsyncClient(timeout=8.0) as client:
                r = await client.get(_community_url("top-queries"), params={"limit": limit})
                return JSONResponse(r.json(), status_code=r.status_code)
        except Exception as exc:
            return JSONResponse({"error": str(exc)}, 503)

    @app.get("/api/community/leaderboard")
    async def community_leaderboard(limit: int = 10):
        try:
            import httpx
            async with httpx.AsyncClient(timeout=8.0) as client:
                r = await client.get(_community_url("leaderboard"), params={"limit": limit})
                return JSONResponse(r.json(), status_code=r.status_code)
        except Exception as exc:
            return JSONResponse({"error": str(exc)}, 503)

    @app.get("/api/community/plugin-requests")
    async def community_plugin_requests(limit: int = 20, offset: int = 0, status: str = ""):
        try:
            import httpx
            params: dict = {"limit": limit, "offset": offset}
            if status:
                params["status"] = status
            async with httpx.AsyncClient(timeout=8.0) as client:
                r = await client.get(_community_url("plugin-requests"), params=params)
                return JSONResponse(r.json(), status_code=r.status_code)
        except Exception as exc:
            return JSONResponse({"error": str(exc)}, 503)

    @app.post("/api/community/plugin-request")
    async def community_plugin_request(data: dict):
        try:
            import httpx
            data["node_id"]   = node.id
            data["node_name"] = node.name
            async with httpx.AsyncClient(timeout=10.0) as client:
                r = await client.post(_community_url("plugin-request"), json=data)
                return JSONResponse(r.json(), status_code=r.status_code)
        except Exception as exc:
            return JSONResponse({"error": str(exc)}, 503)

    @app.post("/api/community/vote-plugin")
    async def community_vote_plugin(data: dict):
        try:
            import httpx
            async with httpx.AsyncClient(timeout=8.0) as client:
                r = await client.post(_community_url("vote-plugin"), json=data)
                return JSONResponse(r.json(), status_code=r.status_code)
        except Exception as exc:
            return JSONResponse({"error": str(exc)}, 503)

    @app.post("/api/community/report")
    async def community_report(data: dict):
        """Bug/Crash-Report einreichen. Liest Log-Datei automatisch (log_b64)."""
        try:
            import httpx
            data.setdefault("node_id",   node.id)
            data.setdefault("node_name", node.name)
            data.setdefault("hive_ver",  __import__("hivemind").__version__)
            data.setdefault("platform",  __import__("platform").platform())
            # Log-Datei base64-kodiert beifügen wenn nicht schon vorhanden
            if not data.get("log_b64"):
                log_path = getattr(node, "_last_log_path", "")
                if log_path:
                    try:
                        raw = Path(log_path).read_bytes()
                        data["log_b64"] = base64.b64encode(raw).decode()
                    except Exception:
                        pass
            async with httpx.AsyncClient(timeout=30.0) as client:
                r = await client.post(_community_url("report"), json=data)
                return JSONResponse(r.json(), status_code=r.status_code)
        except Exception as exc:
            return JSONResponse({"error": str(exc)}, 503)

    @app.post("/api/community/rate-node")
    async def community_rate_node(data: dict):
        try:
            import httpx
            data.setdefault("node_id",   node.id)
            data.setdefault("node_name", node.name)
            async with httpx.AsyncClient(timeout=8.0) as client:
                r = await client.post(_community_url("rate-node"), json=data)
                return JSONResponse(r.json(), status_code=r.status_code)
        except Exception as exc:
            return JSONResponse({"error": str(exc)}, 503)

    return app
