#!/usr/bin/env python3
"""
Voice-Sidecar für Mission Control 2.0 — lokales STT + gestuftes TTS für „Mit Hermes reden".

WARUM ein eigener Dienst? Die ML-Stacks (faster-whisper, piper, chatterbox + torch) brauchen
ihr eigenes Python-3.12-venv — das MC2-Backend läuft auf Python 3.14 und kann sie nicht
importieren. Genau wie der Mem0-Sidecar (mem0_service/) kapselt dieser schlanke FastAPI-Dienst
die schwere Voice-Logik und exponiert sie auf localhost. MC2 (backend/routers/voice.py) proxyt
ihn nach außen; der Browser-Voice-Client (Frontend „Sprechen"-Tab) redet nie direkt mit ihm.

Pipeline-Rolle:
- STT : faster-whisper (Default `medium`, int8, CPU, Sprache=de) — Mikro-Audio → Text.
- TTS : GESTUFT, Engine im Request wählbar (kein Lock-in):
        * `piper`      — schneller Standard, CPU, quasi-sofort, robustes Deutsch (thorsten).
        * `chatterbox` — Premium/Wunschstimme (MIT), Voice-Cloning, dt. über Multilingual.
                         Lazy-Load (Modell erst beim ersten Aufruf) → Dienststart bleibt schnell.
                         Device via VOICE_CHATTERBOX_DEVICE (cpu | cuda); ROCm/iGPU (Strix Halo)
                         per HSA_OVERRIDE_GFX_VERSION=11.0.0 als späterer Umschalter.

Läuft als systemd-User-Dienst (deploy/voice-service.service) im ~/.voice/venv (Python 3.12).
Bind: 127.0.0.1 (nur lokal; MC2 proxyt nach außen).
"""

import io
import json
import logging
import os
import subprocess
import tempfile
from contextlib import asynccontextmanager
from pathlib import Path

from fastapi import FastAPI, File, Form, HTTPException, UploadFile
from fastapi.responses import Response
from pydantic import BaseModel

log = logging.getLogger("voice_service")

# --- Konfiguration (alles über Env überschreibbar; Defaults = Box-Stand) ----------
PORT = int(os.environ.get("VOICE_PORT", "8650"))
# STT
STT_MODEL = os.environ.get("VOICE_STT_MODEL", "medium")       # base|small|medium|large-v3
STT_DEVICE = os.environ.get("VOICE_STT_DEVICE", "cpu")
STT_COMPUTE = os.environ.get("VOICE_STT_COMPUTE", "int8")     # int8=CPU-schonend
STT_LANG = os.environ.get("VOICE_STT_LANG", "de")
# Piper (Default-TTS): Verzeichnis mit *.onnx (+ *.onnx.json) Stimmen + das Piper-Binary.
# Wir nutzen das offizielle Binary (statt des piper-tts-PyPI-Pakets), weil dessen Abhängigkeit
# piper-phonemize auf neueren Linux-Systemen keine Wheels hat. Das Binary bringt espeak-ng mit.
VOICES_DIR = Path(os.environ.get("VOICE_PIPER_DIR", str(Path.home() / ".voice" / "voices")))
PIPER_DEFAULT = os.environ.get("VOICE_PIPER_DEFAULT", "de_DE-thorsten-medium")
PIPER_BIN = os.environ.get("VOICE_PIPER_BIN", str(Path.home() / ".voice" / "piper" / "piper"))
# Chatterbox (Premium-TTS)
CHATTERBOX_DEVICE = os.environ.get("VOICE_CHATTERBOX_DEVICE", "cpu")
CHATTERBOX_LANG = os.environ.get("VOICE_CHATTERBOX_LANG", "de")
# Optionaler Referenz-WAV für Voice-Cloning (10 s Sprachprobe). Leer = Chatterbox-Standardstimme.
CHATTERBOX_REF = os.environ.get("VOICE_CHATTERBOX_REF", "")
# Vom Nutzer hochgeladene Klon-Referenz (persistiert über Restarts). active.txt zeigt auf die Datei.
REFS_DIR = Path(os.environ.get("VOICE_REFS_DIR", str(Path.home() / ".voice" / "refs")))
ACTIVE_REF = REFS_DIR / "active.txt"


def active_ref() -> str:
    """Pfad der aktuell aktiven Klon-Referenz (Upload bevorzugt, sonst Env-Default)."""
    try:
        p = ACTIVE_REF.read_text(encoding="utf-8").strip()
        if p and os.path.exists(p):
            return p
    except OSError:
        pass
    return CHATTERBOX_REF
# ElevenLabs (Premium-Cloud-TTS): saubere, native deutsche Stimmen (Voice Library). Key wird zur
# Laufzeit gelesen (env ODER ~/.hermes/.env), damit Nachtragen ohne Code-Deploy reicht. Default-
# Modell = Flash v2.5 (multilingual, 0,5 Credit/Zeichen → schont das Free-Tier-Kontingent).
# multilingual_v2 = natürlichstes Modell (statt Flash, das auf Tempo optimiert/flacher ist).
ELEVEN_MODEL = os.environ.get("VOICE_ELEVENLABS_MODEL", "eleven_multilingual_v2")
ELEVEN_DEFAULT_VOICE = os.environ.get("VOICE_ELEVENLABS_VOICE", "21m00Tcm4TlvDq8ikWAM")
# Stimm-Settings: niedrige Stability = ausdrucksstärker/lebendiger (hohe = monoton/roboterhaft).
ELEVEN_STABILITY = float(os.environ.get("VOICE_ELEVENLABS_STABILITY", "0.4"))
ELEVEN_SIMILARITY = float(os.environ.get("VOICE_ELEVENLABS_SIMILARITY", "0.8"))
ELEVEN_STYLE = float(os.environ.get("VOICE_ELEVENLABS_STYLE", "0.35"))
HERMES_ENV = Path.home() / ".hermes" / ".env"
# Edge-TTS: native deutsche Azure-Neural-Stimmen, gratis & ohne Key (kein Cloning → kein Akzent).
EDGE_DEFAULT = os.environ.get("VOICE_EDGE_VOICE", "de-DE-KatjaNeural")


def _eleven_key() -> str:
    """ElevenLabs-Key aus Prozess-Env oder (un-auskommentiert) aus ~/.hermes/.env."""
    k = os.environ.get("ELEVENLABS_API_KEY")
    if k:
        return k.strip()
    try:
        for line in HERMES_ENV.read_text(encoding="utf-8").splitlines():
            s = line.strip()
            if s.startswith("ELEVENLABS_API_KEY=") and not s.startswith("#"):
                return s.split("=", 1)[1].strip().strip('"').strip("'")
    except OSError:
        pass
    return ""


# =================================================================================
# STT — faster-whisper (lazy Singleton)
# =================================================================================
_stt = None


def stt_model():
    global _stt
    if _stt is None:
        from faster_whisper import WhisperModel
        log.info("Lade faster-whisper '%s' (%s/%s) …", STT_MODEL, STT_DEVICE, STT_COMPUTE)
        _stt = WhisperModel(STT_MODEL, device=STT_DEVICE, compute_type=STT_COMPUTE)
    return _stt


def transcribe(audio_bytes: bytes, suffix: str, language: str) -> str:
    # PyAV (in faster-whisper) dekodiert webm/opus/ogg/wav am robustesten aus einer Datei.
    with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tf:
        tf.write(audio_bytes)
        tmp = tf.name
    try:
        segments, _info = stt_model().transcribe(
            tmp, language=language or None, vad_filter=True, beam_size=5,
        )
        return "".join(s.text for s in segments).strip()
    finally:
        try:
            os.unlink(tmp)
        except OSError:
            pass


# =================================================================================
# TTS — Piper (offizielles Binary; Stimme = ONNX-Datei + .json daneben)
# =================================================================================
def piper_tts(text: str, voice: str) -> bytes:
    name = voice or PIPER_DEFAULT
    onnx = VOICES_DIR / f"{name}.onnx"
    if not onnx.exists():
        raise HTTPException(404, f"Piper-Stimme '{name}' nicht gefunden ({onnx}).")
    if not os.path.exists(PIPER_BIN):
        raise HTTPException(500, f"Piper-Binary fehlt ({PIPER_BIN}).")
    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tf:
        out = tf.name
    try:
        proc = subprocess.run(
            [PIPER_BIN, "-m", str(onnx), "-f", out],
            input=text.encode("utf-8"), capture_output=True, timeout=60,
        )
        if proc.returncode != 0:
            raise HTTPException(500, f"Piper-Fehler: {proc.stderr.decode('utf-8', 'replace')[:300]}")
        return Path(out).read_bytes()
    finally:
        try:
            os.unlink(out)
        except OSError:
            pass


def piper_list() -> list[dict]:
    if not VOICES_DIR.exists():
        return []
    return [
        {"engine": "piper", "id": p.stem, "label": p.stem, "clonable": False}
        for p in sorted(VOICES_DIR.glob("*.onnx"))
    ]


# =================================================================================
# TTS — Chatterbox (lazy; Multilingual für Deutsch; optional Voice-Cloning)
# =================================================================================
_chatterbox = None


def chatterbox_model():
    global _chatterbox
    if _chatterbox is None:
        log.info("Lade Chatterbox (Multilingual, device=%s) — einmalig, dauert kurz …", CHATTERBOX_DEVICE)
        from chatterbox.mtl_tts import ChatterboxMultilingualTTS
        _chatterbox = ChatterboxMultilingualTTS.from_pretrained(device=CHATTERBOX_DEVICE)
    return _chatterbox


def chatterbox_tts(text: str, language: str, ref_path: str) -> bytes:
    import soundfile as sf
    model = chatterbox_model()
    kwargs = {"language_id": language or CHATTERBOX_LANG}
    ref = ref_path or active_ref()
    if ref and os.path.exists(ref):
        kwargs["audio_prompt_path"] = ref          # Zero-Shot Voice-Cloning aus Referenz
    # Chatterbox hat einen sporadischen Bug (alignment_stream_analyzer → NoneType), der einzelne
    # Generierungen fehlschlagen lässt → im Voice-Loop „Stimme kam nicht". Ein Retry klappt meist.
    wav = None
    last_err: Exception | None = None
    for attempt in range(3):
        try:
            wav = model.generate(text, **kwargs)
            break
        except Exception as exc:  # noqa: BLE001
            last_err = exc
            log.warning("Chatterbox-Generation fehlgeschlagen (Versuch %d/3): %s", attempt + 1, exc)
    if wav is None:
        raise HTTPException(502, f"Chatterbox-Generation fehlgeschlagen: {last_err}")
    # wav = torch.Tensor [1, N] @ model.sr → in WAV-Bytes serialisieren.
    import numpy as np
    arr = wav.squeeze(0).detach().cpu().numpy().astype(np.float32)
    # Peak-Normalisierung gegen Übersteuern/Clipping (Klon-Output ist oft zu laut → kratzig).
    peak = float(np.max(np.abs(arr))) if arr.size else 0.0
    if peak > 0:
        arr = arr / peak * float(os.environ.get("VOICE_CHATTERBOX_PEAK", "0.9"))
    buf = io.BytesIO()
    sf.write(buf, arr, int(model.sr), format="WAV", subtype="PCM_16")
    return buf.getvalue()


def chatterbox_list() -> list[dict]:
    # Chatterbox hat keine festen „Stimm-Dateien": Standardstimme + optionale Klon-Referenz.
    items = [{"engine": "chatterbox", "id": "default", "label": "Chatterbox (Standard, dt.)", "clonable": True}]
    if active_ref():
        items.append({"engine": "chatterbox", "id": "clone", "label": "Chatterbox (deine Referenz)", "clonable": True})
    return items


# =================================================================================
# TTS — ElevenLabs (Premium-Cloud; native dt. Stimmen, Voice Library). Nur HTTP (stdlib).
# =================================================================================
def elevenlabs_tts(text: str, voice: str) -> bytes:
    import urllib.error
    import urllib.request

    key = _eleven_key()
    if not key:
        raise HTTPException(503, "ELEVENLABS_API_KEY nicht gesetzt (in ~/.hermes/.env eintragen).")
    vid = voice or ELEVEN_DEFAULT_VOICE
    body = json.dumps({
        "text": text,
        "model_id": ELEVEN_MODEL,
        "voice_settings": {
            "stability": ELEVEN_STABILITY,
            "similarity_boost": ELEVEN_SIMILARITY,
            "style": ELEVEN_STYLE,
            "use_speaker_boost": True,
        },
    }).encode("utf-8")
    req = urllib.request.Request(
        f"https://api.elevenlabs.io/v1/text-to-speech/{vid}",
        data=body,
        headers={"xi-api-key": key, "Content-Type": "application/json", "Accept": "audio/mpeg"},
    )
    import time
    # Free-Tier hat ein Concurrency-Limit → parallele Sätze geben 429. Retry mit Backoff fängt das ab.
    last = ""
    for attempt in range(4):
        try:
            with urllib.request.urlopen(req, timeout=60) as r:
                return r.read()
        except urllib.error.HTTPError as e:
            last = f"{e.code}: {e.read()[:200].decode('utf-8', 'replace')}"
            if e.code in (429, 500, 502, 503, 529) and attempt < 3:
                time.sleep(0.5 * (attempt + 1))
                continue
            raise HTTPException(502, f"ElevenLabs-Fehler {last}")
        except urllib.error.URLError as e:
            last = str(e)
            if attempt < 3:
                time.sleep(0.5 * (attempt + 1))
                continue
            raise HTTPException(502, f"ElevenLabs nicht erreichbar: {last}")
    raise HTTPException(502, f"ElevenLabs-Fehler (nach Retries): {last}")


def elevenlabs_list() -> list[dict]:
    import urllib.request

    key = _eleven_key()
    if not key:
        return []
    try:
        req = urllib.request.Request("https://api.elevenlabs.io/v1/voices", headers={"xi-api-key": key})
        with urllib.request.urlopen(req, timeout=10) as r:
            data = json.loads(r.read())
        # NUR eigene Stimmen (generated/cloned/professional) — Library-/premade-Stimmen sperrt das
        # Free-Tier per API (402). So landet im Picker nie eine unbrauchbare Stimme.
        return [
            {"engine": "elevenlabs", "id": v["voice_id"], "label": v.get("name", v["voice_id"])}
            for v in data.get("voices", [])
            if v.get("category") != "premade"
        ]
    except Exception:  # noqa: BLE001
        log.warning("ElevenLabs-Stimmenliste fehlgeschlagen", exc_info=True)
        return []


# =================================================================================
# TTS — Edge-TTS (native dt. Azure-Neural-Stimmen, gratis, kein Key, KEIN Cloning → kein Akzent)
# =================================================================================
def edge_tts_synth(text: str, voice: str) -> bytes:
    import asyncio
    import edge_tts

    v = voice or EDGE_DEFAULT

    async def _run() -> bytes:
        buf = bytearray()
        communicate = edge_tts.Communicate(text, v)
        async for chunk in communicate.stream():
            if chunk["type"] == "audio":
                buf.extend(chunk["data"])
        return bytes(buf)

    try:
        return asyncio.run(_run())
    except Exception as exc:  # noqa: BLE001
        raise HTTPException(502, f"Edge-TTS-Fehler: {exc}")


_edge_cache: list | None = None


def edge_list() -> list[dict]:
    """Deutsche Edge/Azure-Stimmen (einmal vom Dienst geholt + gecacht). Weibliche zuerst."""
    global _edge_cache
    if _edge_cache is not None:
        return _edge_cache
    try:
        import asyncio
        import edge_tts

        voices = asyncio.run(edge_tts.list_voices())
        out = []
        for v in voices:
            if not v.get("Locale", "").startswith("de-"):
                continue
            short = v["ShortName"]                       # z.B. de-DE-KatjaNeural
            gender = "weiblich" if v.get("Gender") == "Female" else "männlich"
            name = short.split("-")[-1].replace("Neural", "").replace("Multilingual", " (multiling.)")
            out.append({"engine": "edge", "id": short,
                        "label": f"{name} · {v['Locale']} · {gender}",
                        "clonable": False, "_female": v.get("Gender") == "Female"})
        out.sort(key=lambda i: (not i.pop("_female"), i["id"]))
        _edge_cache = out
    except Exception:  # noqa: BLE001
        log.warning("Edge-Stimmenliste fehlgeschlagen", exc_info=True)
        _edge_cache = []
    return _edge_cache


# =================================================================================
# FastAPI
# =================================================================================
@asynccontextmanager
async def lifespan(_app: FastAPI):
    logging.basicConfig(level=logging.INFO)
    try:
        stt_model()  # STT beim Start vorwärmen (Modell aus HF-Cache laden)
        log.info("Voice-Sidecar bereit (STT '%s', Piper-Dir %s).", STT_MODEL, VOICES_DIR)
    except Exception:
        log.exception("STT-Vorwärmen fehlgeschlagen (Dienst läuft, /health meldet Detail).")
    yield


app = FastAPI(title="MC2 Voice Sidecar", lifespan=lifespan)


class TTSIn(BaseModel):
    text: str
    engine: str = "piper"          # piper | chatterbox
    voice: str = ""                # Piper-Stimmname; bei Chatterbox: "default" | "clone"
    language: str = ""             # überschreibt Default-Sprache
    ref_path: str = ""             # optionaler Klon-Referenz-WAV (Chatterbox)


@app.get("/health")
def health() -> dict:
    return {"ok": True, "stt_model": STT_MODEL,
            "engines": ["elevenlabs", "edge"],
            "elevenlabs_key": bool(_eleven_key())}


@app.get("/voices")
def voices() -> dict:
    # Nur die genutzten Engines: ElevenLabs (eigene Stimme) + Edge (gratis, nativ-deutsch).
    return {"voices": elevenlabs_list() + edge_list(),
            "default": {"engine": "edge", "voice": EDGE_DEFAULT}}


@app.post("/stt")
async def stt(audio: UploadFile = File(...), language: str = Form(default="")) -> dict:
    data = await audio.read()
    if not data:
        raise HTTPException(400, "Leeres Audio.")
    suffix = Path(audio.filename or "rec.webm").suffix or ".webm"
    text = transcribe(data, suffix, language or STT_LANG)
    return {"text": text}


@app.post("/tts")
def tts(body: TTSIn) -> Response:
    text = (body.text or "").strip()
    if not text:
        raise HTTPException(400, "Leerer Text.")
    if body.engine == "edge":
        return Response(content=edge_tts_synth(text, body.voice), media_type="audio/mpeg")
    if body.engine == "elevenlabs":
        return Response(content=elevenlabs_tts(text, body.voice), media_type="audio/mpeg")
    if body.engine == "chatterbox":
        return Response(content=chatterbox_tts(text, body.language, body.ref_path), media_type="audio/wav")
    return Response(content=piper_tts(text, body.voice), media_type="audio/wav")


def _save_reference_as_wav(data: bytes, filename: str) -> str:
    """Upload (mp3/wav/…) → 24 kHz Mono-WAV für Chatterbox. Dekodierung via PyAV (steckt in
    faster-whisper → kein System-ffmpeg nötig); Fallback soundfile, sonst Rohdatei."""
    REFS_DIR.mkdir(parents=True, exist_ok=True)
    raw = REFS_DIR / ("upload" + (Path(filename or "ref").suffix or ".bin"))
    raw.write_bytes(data)
    dest = REFS_DIR / "ref.wav"
    import soundfile as sf
    try:
        from faster_whisper.audio import decode_audio   # PyAV-basiert, kann mp3/m4a/ogg/wav
        arr = decode_audio(str(raw), sampling_rate=24000)
        sf.write(str(dest), arr, 24000, subtype="PCM_16")
        return str(dest)
    except Exception:
        log.warning("PyAV-Dekodierung fehlgeschlagen, versuche soundfile", exc_info=True)
    try:
        arr, srr = sf.read(str(raw))
        sf.write(str(dest), arr, srr, subtype="PCM_16")
        return str(dest)
    except Exception:
        return str(raw)   # Fallback: Chatterbox versucht selbst zu laden


@app.post("/reference")
async def set_reference(audio: UploadFile = File(...)) -> dict:
    """Klon-Referenz hochladen (z.B. ElevenLabs-Erzeugnis) → Chatterbox nutzt sie als Stimme."""
    data = await audio.read()
    if not data:
        raise HTTPException(400, "Leeres Audio.")
    path = _save_reference_as_wav(data, audio.filename or "ref.wav")
    ACTIVE_REF.write_text(path, encoding="utf-8")
    return {"ok": True, "path": path, "bytes": len(data)}


@app.get("/reference")
def get_reference() -> dict:
    p = active_ref()
    return {"active": bool(p), "path": p}


@app.delete("/reference")
def clear_reference() -> dict:
    try:
        ACTIVE_REF.unlink(missing_ok=True)
    except OSError:
        pass
    return {"ok": True}


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=PORT)