"""Lokales TTS über piper (CPU-freundlich, offline).

Bevorzugt die **in-process** piper-Python-API: Das Stimmmodell wird EINMAL geladen
und prozessweit zwischengespeichert (lru_cache) - so entfaellt der teure Modell-Start
pro Satz (~2 s), der bei einem Subprozess-pro-Aufruf anfiel. Fehlt das Paket
(`pip install -e .[local]`), wird automatisch auf das piper-Binary zurueckgefallen.

piper liefert s16le-Mono-PCM in der Sample-Rate des Modells; das Gateway erwartet
24000 Hz -> bei Abweichung wird resampelt (in-process via audioop, sonst ffmpeg).
"""

from __future__ import annotations

import asyncio
import io
import json
import shutil
import wave
from functools import lru_cache
from pathlib import Path

from app.providers.tts.base import TTSProvider

try:  # bevorzugter Pfad: in-process, Modell bleibt geladen
    from piper import PiperVoice  # type: ignore

    _PIPER_LIB = True
except Exception:  # pragma: no cover - Paket optional
    PiperVoice = None  # type: ignore
    _PIPER_LIB = False


def _wrap_wav(pcm: bytes, sample_rate: int) -> bytes:
    buf = io.BytesIO()
    with wave.open(buf, "wb") as w:
        w.setnchannels(1)
        w.setsampwidth(2)
        w.setframerate(sample_rate)
        w.writeframes(pcm)
    return buf.getvalue()


@lru_cache(maxsize=4)
def _load_voice(model_path: str):
    """Laedt ein piper-Stimmmodell einmalig (prozessweit gecacht)."""
    return PiperVoice.load(model_path)


def _resample(pcm: bytes, src_rate: int, dst_rate: int) -> bytes:
    """Resampelt s16le-Mono in-process. audioop (stdlib) bevorzugt, sonst numpy."""
    try:
        import audioop  # in Python 3.13 entfernt -> Fallback unten

        converted, _ = audioop.ratecv(pcm, 2, 1, src_rate, dst_rate, None)
        return converted
    except Exception:
        import numpy as np

        src = np.frombuffer(pcm, dtype="<i2").astype(np.float32)
        n_out = max(1, round(len(src) * dst_rate / src_rate))
        x_old = np.arange(len(src))
        x_new = np.linspace(0, len(src) - 1, n_out)
        out = np.interp(x_new, x_old, src)
        return np.clip(out, -32768, 32767).astype("<i2").tobytes()


class PiperTTSProvider(TTSProvider):
    def __init__(
        self,
        bin_path: str = "piper",
        voices_dir: str = "",
        voice: str = "",
        target_rate: int = 24000,
    ):
        self.bin_path = (bin_path or "piper").strip()
        self.voices_dir = Path(voices_dir).expanduser() if voices_dir else Path.cwd()
        self.voice = (voice or "").strip()
        self.target_rate = int(target_rate)

    def _model_paths(self, voice: str | None) -> tuple[Path, Path]:
        # Angeforderte Stimme zuerst; passt sie nicht (z. B. eine Cloud-Stimme wie
        # "Zephyr"/"alloy" aus der Route), auf die konfigurierte Default-Stimme zurueckfallen.
        if not (voice or self.voice or "").strip():
            raise ValueError("Piper-Stimme ist nicht gesetzt (PIPER_VOICE)")
        tried: list[str] = []
        for candidate in (voice, self.voice):
            name = (candidate or "").strip()
            if not name or name in tried:
                continue
            tried.append(name)
            model = Path(name).expanduser()
            if not model.suffix:  # nur ein Name -> im Voices-Verzeichnis suchen
                model = self.voices_dir / f"{name}.onnx"
            if model.exists():
                return model, Path(f"{model}.json")
        raise RuntimeError(
            f"Piper-Stimmmodell nicht gefunden (versucht: {', '.join(tried)}) in {self.voices_dir}"
        )

    def _native_rate(self, config: Path) -> int:
        try:
            with open(config) as fh:
                return int(json.load(fh)["audio"]["sample_rate"])
        except (OSError, KeyError, ValueError, TypeError):
            return self.target_rate  # Konfig unlesbar -> Resampling überspringen

    async def synthesize(
        self,
        text: str,
        voice: str | None = None,
        audio_format: str = "pcm",
        language: str | None = None,  # ignoriert: die Piper-Stimme kodiert die Sprache
    ) -> bytes:
        if not text or not text.strip():
            raise ValueError("TTS input text is empty")

        model, config = self._model_paths(voice)
        pcm, native_rate = await asyncio.to_thread(
            self._synthesize_sync, str(model), str(config), text.strip()
        )
        if native_rate != self.target_rate:
            pcm = await asyncio.to_thread(_resample, pcm, native_rate, self.target_rate)

        if audio_format == "wav":
            return _wrap_wav(pcm, self.target_rate)
        return pcm

    def _synthesize_sync(self, model: str, config: str, text: str) -> tuple[bytes, int]:
        if _PIPER_LIB:
            voice = _load_voice(model)
            pcm = bytearray()
            rate = self.target_rate
            for chunk in voice.synthesize(text):
                pcm += chunk.audio_int16_bytes
                rate = chunk.sample_rate
            if not pcm:
                raise RuntimeError("Piper lieferte kein Audio")
            return bytes(pcm), rate
        # Fallback: piper-Binary (Subprozess pro Aufruf, langsamer).
        return self._run_piper_binary(model, config, text)

    def _run_piper_binary(self, model: str, config: str, text: str) -> tuple[bytes, int]:
        import subprocess

        if not (shutil.which(self.bin_path) or Path(self.bin_path).exists()):
            raise RuntimeError(f"Piper-Binary nicht gefunden: {self.bin_path}")
        cmd = [self.bin_path, "-m", model, "-c", config, "--output-raw"]
        proc = subprocess.run(cmd, input=text.encode("utf-8"), capture_output=True)
        if proc.returncode != 0:
            detail = proc.stderr.decode("utf-8", "replace").strip()[:300]
            raise RuntimeError(f"Piper-Fehler ({proc.returncode}): {detail}")
        if not proc.stdout:
            raise RuntimeError("Piper lieferte kein Audio")
        return proc.stdout, self._native_rate(Path(config))