"""TTS voice enumeration for all supported providers.

Provides list_voices(provider) -> list[dict] so the GUI can populate a
voice selector dropdown for whichever TTS backend is active.
"""

from __future__ import annotations

import logging
from importlib.util import find_spec
from pathlib import Path

logger = logging.getLogger(__name__)

# Default directory for XTTS speaker WAV files (relative to project root)
_DEFAULT_SPEAKERS_DIR = Path(__file__).resolve().parent.parent / "voices"

# Cache for edge-tts voice list (avoids repeated async network calls)
_edge_tts_cache: list[dict] | None = None


def list_voices(provider: str, *, speakers_dir: Path | None = None) -> list[dict]:
    """List available voices for the given TTS provider.

    Args:
        provider: One of 'xtts', 'edge-tts', 'pyttsx3'.
        speakers_dir: Override directory for XTTS speaker WAV files.

    Returns:
        List of dicts with keys: ``id``, ``name``, ``language``, ``gender``
        (``gender`` may be ``None``).  Returns ``[]`` if the provider
        dependencies are missing or the directory is empty.
    """
    p = provider.lower().strip()
    if p == "xtts":
        return _list_xtts_voices(speakers_dir=speakers_dir)
    if p in ("edge-tts", "edge_tts", "edgetts"):
        return _list_edge_tts_voices()
    if p == "pyttsx3":
        return _list_pyttsx3_voices()
    logger.warning("Unknown TTS provider for voice listing: %s", provider)
    return []


def _list_xtts_voices(*, speakers_dir: Path | None = None) -> list[dict]:
    """List WAV speaker files in the XTTS speakers directory."""
    directory = speakers_dir or _DEFAULT_SPEAKERS_DIR
    if not directory.is_dir():
        logger.debug("XTTS speakers directory not found: %s", directory)
        return []
    voices: list[dict] = []
    for wav_file in sorted(directory.glob("*.wav")):
        stem = wav_file.stem
        voices.append(
            {
                "id": str(wav_file),
                "name": stem.replace("_", " ").replace("-", " ").title(),
                "language": "en",
                "gender": None,
                "engine": "xtts",
            }
        )
    return voices


def _list_edge_tts_voices() -> list[dict]:
    """List voices available from edge-tts (result cached for the session)."""
    global _edge_tts_cache
    if _edge_tts_cache is not None:
        return _edge_tts_cache

    try:
        import asyncio

        import edge_tts

        async def _fetch() -> list[dict]:
            raw = await edge_tts.list_voices()
            return [
                {
                    "id": v.get("ShortName", ""),
                    "name": v.get("FriendlyName", v.get("ShortName", "")),
                    "language": v.get("Locale", ""),
                    "gender": v.get("Gender"),
                }
                for v in raw
            ]

        _edge_tts_cache = [{**v, "engine": "edge-tts"} for v in asyncio.run(_fetch())]
        return _edge_tts_cache
    except ImportError:
        logger.debug("edge-tts not installed; no voices available")
        return []
    except Exception as exc:
        logger.warning("Failed to list edge-tts voices: %s", exc)
        return []


def _list_pyttsx3_voices() -> list[dict]:
    """List voices from the pyttsx3 engine."""
    try:
        import pyttsx3

        engine = pyttsx3.init()
        raw_voices = engine.getProperty("voices") or []
        voices: list[dict] = []
        for v in raw_voices:
            voice_id = getattr(v, "id", "") or ""
            name = getattr(v, "name", "") or voice_id
            lang = ""
            languages = getattr(v, "languages", None)
            if isinstance(languages, (list, tuple)) and languages:
                lang = str(languages[0])
            voices.append(
                {
                    "id": voice_id,
                    "name": name,
                    "language": lang,
                    "gender": None,
                    "engine": "pyttsx3",
                }
            )
        engine.stop()
        return voices
    except ImportError:
        logger.debug("pyttsx3 not installed; no voices available")
        return []
    except Exception as exc:
        logger.warning("Failed to list pyttsx3 voices: %s", exc)
        return []


async def synthesize_sample(
    provider: str,
    voice_id: str,
    text: str = "Hello, I'm Rex.",
) -> bytes:
    """Synthesize a short audio sample for the given voice and provider.

    Args:
        provider: One of 'xtts', 'edge-tts', 'pyttsx3'.
        voice_id: Provider-specific voice identifier returned by ``list_voices``.
        text: Phrase to synthesize (max 50 characters; longer text is truncated).

    Returns:
        Raw audio bytes (WAV for xtts/pyttsx3, MP3 for edge-tts).
        Raises ``RuntimeError`` if synthesis fails or the provider is unsupported.
    """
    return await synthesize_speech(provider, voice_id, text[:50])


async def synthesize_speech(provider: str, voice_id: str, text: str) -> bytes:
    """Synthesize one bounded assistant response for renderer playback."""
    text = text.strip()[:4000]
    if not text:
        raise RuntimeError("Speech text is required")

    p = provider.lower().strip()
    if p == "xtts":
        return await _synthesize_xtts(voice_id, text)
    if p in ("edge-tts", "edge_tts", "edgetts"):
        return await _synthesize_edge_tts(voice_id, text)
    if p == "pyttsx3":
        return await _synthesize_pyttsx3(voice_id, text)
    raise RuntimeError(f"Unsupported TTS provider for sample synthesis: {provider}")


async def _synthesize_xtts(voice_id: str, text: str) -> bytes:
    """Synthesize audio using Coqui XTTS with the given speaker WAV file."""
    import asyncio
    import os
    import tempfile

    def _run() -> bytes:
        if find_spec("TTS") is None:
            raise RuntimeError("Coqui TTS is not installed")
        from rex.compat import ensure_transformers_compatibility

        ensure_transformers_compatibility()
        from TTS.api import TTS

        tts = TTS("tts_models/multilingual/multi-dataset/xtts_v2", gpu=False)
        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
            tmpfile = f.name
        try:
            tts.tts_to_file(
                text,
                speaker_wav=voice_id,
                language="en",
                file_path=tmpfile,
            )
            with open(tmpfile, "rb") as fh:
                return fh.read()
        finally:
            if os.path.exists(tmpfile):
                os.unlink(tmpfile)

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


async def _synthesize_edge_tts(voice_id: str, text: str) -> bytes:
    """Synthesize audio using edge-tts (returns MP3 bytes)."""
    import os
    import tempfile

    try:
        import edge_tts
    except ImportError as exc:
        raise RuntimeError("edge-tts is not installed") from exc

    with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f:
        tmpfile = f.name
    try:
        communicate = edge_tts.Communicate(text, voice_id)
        await communicate.save(tmpfile)
        with open(tmpfile, "rb") as fh:
            return fh.read()
    finally:
        if os.path.exists(tmpfile):
            os.unlink(tmpfile)


async def _synthesize_pyttsx3(voice_id: str, text: str) -> bytes:
    """Synthesize audio using pyttsx3 (returns WAV bytes)."""
    import asyncio
    import os
    import tempfile

    def _run() -> bytes:
        try:
            import pyttsx3
        except ImportError as exc:
            raise RuntimeError("pyttsx3 is not installed") from exc

        engine = pyttsx3.init()
        if voice_id:
            engine.setProperty("voice", voice_id)
        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
            tmpfile = f.name
        try:
            engine.save_to_file(text, tmpfile)
            engine.runAndWait()
            engine.stop()
            with open(tmpfile, "rb") as fh:
                return fh.read()
        finally:
            if os.path.exists(tmpfile):
                os.unlink(tmpfile)

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


def clear_edge_tts_cache() -> None:
    """Clear the cached edge-tts voice list (useful for testing)."""
    global _edge_tts_cache
    _edge_tts_cache = None


__all__ = ["list_voices", "synthesize_sample", "synthesize_speech", "clear_edge_tts_cache"]