"""Text-to-speech service using piper-tts (in-process, CPU).

Voice model files (.onnx + .onnx.json sidecar) live in two directories:

- /opt/piper-voices/   — bundled at image build time, read-only.
- /data/voices/        — admin-downloaded at runtime, persisted across
                          restarts via the same volume as other /data/*.

A single PiperVoice is kept loaded at a time (one voice per request would
be too slow; keeping all warm would use a lot of RAM for a feature most
users won't enable). The active voice is whichever one the user picked
via Settings → Voice — selection changes trigger reload_tts_model().
"""
from __future__ import annotations

import asyncio
import io
import json
import logging
import time
import wave
from pathlib import Path
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from piper import PiperVoice

logger = logging.getLogger(__name__)

# Two directories scanned for voices. Order matters for tie-breaking when
# both contain the same voice id — /data wins (admin-downloaded override).
_BUNDLED_VOICES_DIR = Path("/opt/piper-voices")
_USER_VOICES_DIR = Path("/data/voices")

_DEFAULT_VOICE_ID = "en_US-amy-medium"

_voice: "PiperVoice | None" = None
_voice_id: str | None = None  # id of the currently-loaded voice
_voice_lock = asyncio.Lock()
_load_error: str | None = None


def _voice_dirs() -> list[Path]:
    """Directories scanned for voice files, in precedence order (later wins)."""
    return [_BUNDLED_VOICES_DIR, _USER_VOICES_DIR]


def _discover_voice(voice_id: str) -> Path | None:
    """Find the .onnx file for `voice_id` across the scanned directories.

    /data wins over /opt if both contain the same id so an admin-downloaded
    voice can override a bundled one if they want.
    """
    found: Path | None = None
    for d in _voice_dirs():
        candidate = d / f"{voice_id}.onnx"
        if candidate.is_file() and (d / f"{voice_id}.onnx.json").is_file():
            found = candidate
    return found


def _scan_voices() -> list[Path]:
    """Return all valid voice files (.onnx + .onnx.json) across all dirs.

    Same-id duplicates are deduplicated; later directories override earlier
    so a /data voice shadows a bundled one with the same name.
    """
    by_id: dict[str, Path] = {}
    for d in _voice_dirs():
        if not d.is_dir():
            continue
        for onnx in sorted(d.glob("*.onnx")):
            sidecar = onnx.with_suffix(".onnx.json")
            if not sidecar.is_file():
                continue
            by_id[onnx.stem] = onnx
    return list(by_id.values())


def _voice_metadata(onnx_path: Path) -> dict:
    """Read the .onnx.json sidecar and return a UI-shaped metadata dict.

    The sidecar's full schema is large; we surface only what the picker
    needs: id (filename stem), label (derived), language, quality, sample_rate.
    """
    sidecar = onnx_path.with_suffix(".onnx.json")
    info: dict = {}
    try:
        info = json.loads(sidecar.read_text())
    except Exception:
        logger.warning("Could not parse piper sidecar %s", sidecar, exc_info=True)

    voice_id = onnx_path.stem
    lang_info = info.get("language") or {}
    audio_info = info.get("audio") or {}
    # Filename convention: <lang>-<dataset>-<quality> e.g. en_US-amy-medium
    parts = voice_id.split("-")
    dataset = parts[1] if len(parts) >= 2 else voice_id
    quality = parts[2] if len(parts) >= 3 else "medium"
    lang_code = lang_info.get("code") or (parts[0] if parts else "")
    region_name = lang_info.get("name_native") or lang_info.get("region") or lang_code

    return {
        "id": voice_id,
        "label": f"{dataset.capitalize()} ({region_name}, {quality})",
        "language": lang_code,
        "quality": quality,
        "sample_rate": audio_info.get("sample_rate", 22050),
    }


async def load_tts_model() -> None:
    """Load the active piper voice. Called once at startup if voice is enabled.

    The active voice is determined by the per-user `voice_tts_voice` setting
    when known; falls back to `_DEFAULT_VOICE_ID` if unset or invalid.
    Because the active voice is per-user but the loaded model is a single
    in-process resource, this loads the default at startup; per-user voice
    selection triggers a reload at request time.
    """
    global _voice, _voice_id, _load_error
    from fabledassistant.services.voice_config import is_voice_enabled

    if not await is_voice_enabled():
        return

    async with _voice_lock:
        if _voice is not None:
            return
        try:
            target_id = _DEFAULT_VOICE_ID
            onnx_path = _discover_voice(target_id)
            if onnx_path is None:
                # No bundled default? Fall back to whatever is on disk.
                discovered = _scan_voices()
                if not discovered:
                    _load_error = (
                        "No piper voice models found in /opt/piper-voices "
                        "or /data/voices"
                    )
                    logger.error(_load_error)
                    return
                onnx_path = discovered[0]
                target_id = onnx_path.stem
                logger.warning(
                    "Default voice %r not found; using %r instead",
                    _DEFAULT_VOICE_ID, target_id,
                )

            logger.info("Loading piper TTS voice %s from %s", target_id, onnx_path)
            loop = asyncio.get_running_loop()
            from piper import PiperVoice

            _voice = await loop.run_in_executor(
                None, lambda: PiperVoice.load(str(onnx_path))
            )
            _voice_id = target_id
            logger.info("Piper TTS voice %s loaded", target_id)
        except Exception:
            _load_error = "Failed to load piper TTS voice"
            logger.exception(_load_error)


async def reload_tts_model(voice_id: str | None = None) -> None:
    """Unload the current voice and load `voice_id` (or re-load the default).

    Safe to call at runtime — used when the user changes voice in Settings.
    """
    global _voice, _voice_id, _load_error
    async with _voice_lock:
        _voice = None
        _voice_id = None
        _load_error = None
    if voice_id:
        await _switch_voice(voice_id)
    else:
        await load_tts_model()


async def _switch_voice(voice_id: str) -> None:
    """Internal: load a specific voice by id. No-op if already loaded."""
    global _voice, _voice_id, _load_error
    async with _voice_lock:
        if _voice is not None and _voice_id == voice_id:
            return
        onnx_path = _discover_voice(voice_id)
        if onnx_path is None:
            _load_error = f"Voice {voice_id!r} not found in /opt or /data"
            logger.warning(_load_error)
            return
        try:
            logger.info("Switching piper voice to %s", voice_id)
            loop = asyncio.get_running_loop()
            from piper import PiperVoice

            _voice = await loop.run_in_executor(
                None, lambda: PiperVoice.load(str(onnx_path))
            )
            _voice_id = voice_id
            _load_error = None
        except Exception:
            _load_error = f"Failed to load piper voice {voice_id!r}"
            logger.exception(_load_error)


def tts_available() -> bool:
    return _voice is not None


def list_voices() -> list[dict]:
    """Return metadata for every voice on disk, in stable order."""
    return [_voice_metadata(p) for p in _scan_voices()]


async def synthesise(
    text: str,
    voice: str = _DEFAULT_VOICE_ID,
    speed: float = 1.0,
    voice_blend: list[dict] | None = None,  # accepted for backcompat; ignored
) -> bytes:
    """Synthesise text → WAV bytes (PCM 16-bit mono). Runs in executor.

    `voice_blend` is accepted for backward compatibility with callers that
    still pass it, but piper has no voice-blending equivalent, so it is
    silently ignored. The first voice in the blend (if any) is used; if
    that doesn't exist on disk, the request uses the default voice.

    `speed` is the kokoro-shaped multiplier (1.0=normal, 1.5=faster).
    Piper uses `length_scale` where smaller = faster, so the conversion
    is `length_scale = 1.0 / speed`.
    """
    # Backcompat: if blend was passed, just use the first entry's voice.
    if voice_blend and isinstance(voice_blend, list) and voice_blend:
        first = voice_blend[0]
        if isinstance(first, dict) and first.get("voice"):
            voice = str(first["voice"])

    # Lazy-load / switch if needed.
    if _voice is None or _voice_id != voice:
        await _switch_voice(voice)
    if _voice is None:
        raise RuntimeError(f"TTS voice {voice!r} not loaded: {_load_error}")

    speed = max(0.5, min(2.0, speed))
    length_scale = 1.0 / speed

    def _run() -> bytes:
        # piper.SynthesisConfig is optional; defaults are sensible.
        # We override length_scale only.
        from piper import SynthesisConfig

        config = SynthesisConfig(length_scale=length_scale)

        t0 = time.monotonic()
        buf = io.BytesIO()
        sample_rate = _voice.config.sample_rate  # type: ignore[union-attr]
        total_samples = 0
        with wave.open(buf, "wb") as wav:
            wav.setnchannels(1)
            wav.setsampwidth(2)  # 16-bit PCM
            wav.setframerate(sample_rate)
            try:
                for chunk in _voice.synthesize(text, config):  # type: ignore[union-attr]
                    pcm = chunk.audio_int16_bytes
                    if not pcm:
                        continue
                    wav.writeframes(pcm)
                    total_samples += len(pcm) // 2
            except Exception:
                logger.exception(
                    "Piper synthesis error: %d chars, preview=%r",
                    len(text), text[:80],
                )
                raise

        elapsed = time.monotonic() - t0
        logger.info(
            "Piper synthesis: %d chars → %d samples (%.2fs, %.0f chars/s, voice=%s)",
            len(text), total_samples, elapsed,
            len(text) / elapsed if elapsed > 0 else 0,
            _voice_id,
        )
        if not total_samples:
            return b""
        return buf.getvalue()

    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(None, _run)
