"""Cartesia TTS provider — HTTP bytes endpoint, pure aiohttp.

Targets the chunked-bytes HTTP API which maps cleanly to Patter's
``TTSProvider.synthesize(text) -> AsyncIterator[bytes]`` contract and
requires no vendor SDK. Cartesia also exposes a richer WebSocket streaming
mode with word timestamps and sentence tokenization; that mode is not used
here because the HTTP API already meets Patter's TTFB target while keeping
the dependency surface minimal.
"""

from __future__ import annotations

import logging
import os
from enum import IntEnum, StrEnum
from typing import ClassVar, Any, AsyncIterator, Literal, Optional

logger = logging.getLogger("getpatter.providers.cartesia_tts")

from getpatter.providers.base import TTSProvider

# Lazy import: aiohttp is declared as an optional dep for this provider.
try:  # pragma: no cover - trivial import guard
    import aiohttp
except ImportError:  # pragma: no cover
    aiohttp = None  # type: ignore

CARTESIA_BASE_URL = "https://api.cartesia.ai"
# Cartesia API version pin — kept in sync with our STT integration and the
# Cartesia Line skill. ``2025-04-16`` is the current GA snapshot.
CARTESIA_API_VERSION = "2025-04-16"

# Cartesia's "Katie — Friendly Fixer" is a sensible default. The voice ID
# is stable across the sonic-2 / sonic-3 model bump.
CARTESIA_DEFAULT_VOICE_ID = "f786b574-daa5-4673-aa0c-cbe3e8534c02"


class CartesiaTTSModel(StrEnum):
    """Cartesia TTS model identifiers."""

    SONIC_3 = "sonic-3"
    SONIC_2 = "sonic-2"
    SONIC = "sonic"
    SONIC_PREVIEW = "sonic-preview"


class CartesiaTTSEncoding(StrEnum):
    """Audio encodings accepted by Cartesia TTS output_format."""

    PCM_S16LE = "pcm_s16le"
    PCM_F32LE = "pcm_f32le"
    PCM_MULAW = "pcm_mulaw"
    PCM_ALAW = "pcm_alaw"


class CartesiaTTSContainer(StrEnum):
    """Cartesia TTS output container formats."""

    RAW = "raw"
    WAV = "wav"
    MP3 = "mp3"


class CartesiaSampleRate(IntEnum):
    """Common PCM sample rates for Cartesia TTS output."""

    HZ_8000 = 8000
    HZ_16000 = 16000
    HZ_22050 = 22050
    HZ_24000 = 24000
    HZ_44100 = 44100


class CartesiaVoiceSpeed(StrEnum):
    """Cartesia speed presets accepted via ``generation_config.speed``."""

    FASTEST = "fastest"
    FAST = "fast"
    NORMAL = "normal"
    SLOW = "slow"
    SLOWEST = "slowest"


# Backwards-compat aliases for existing imports.
TTSEncoding = Literal["pcm_s16le"]
TTSVoiceSpeed = Literal["fastest", "fast", "normal", "slow", "slowest"]


class CartesiaTTS(TTSProvider):
    """Cartesia TTS over the HTTP ``/tts/bytes`` endpoint.

    Output is PCM_S16LE at the configured sample rate (default 16000 Hz so it
    lines up with Patter's telephony pipeline without a resample step).

    Default model is ``sonic-3`` (GA snapshot ``sonic-3-2026-01-12``) — the
    current Cartesia GA model, with a documented ~90 ms TTFB target. The
    sonic-2 voice IDs (e.g. the default Katie voice) remain compatible on
    sonic-3 so the upgrade is drop-in.

    Telephony optimization
    ----------------------
    The constructor default ``sample_rate=16000`` is correct for web
    playback, dashboard previews, and 16 kHz pipelines. For real phone
    calls use the carrier-specific factories instead:

    * :meth:`for_twilio` — emits ``pcm_mulaw`` @ 8 kHz, Twilio's exact wire
      codec, so the pipeline skips resampling AND PCM -> mu-law encoding
      (bit-clean passthrough). The sender reads the declared output format
      via :meth:`source_audio_format` rather than assuming a 16 kHz source,
      so this no longer chipmunks.
    * :meth:`for_telnyx` — emits ``pcm_mulaw`` @ 8 kHz. The SDK pins the
      Telnyx wire to PCMU/mu-law @ 8 kHz, so this flows end-to-end with
      zero resampling or transcoding — same passthrough win as the Twilio
      factory.
    """

    #: Stable pricing/dashboard key — read by stream-handler/metrics.
    provider_key: ClassVar[str] = "cartesia_tts"

    def __init__(
        self,
        api_key: Optional[str] = None,
        *,
        model: str = "sonic-3",
        voice: str = CARTESIA_DEFAULT_VOICE_ID,
        language: str = "en",
        sample_rate: int = 16000,
        encoding: str = "pcm_s16le",
        speed: Optional[str | float] = None,
        emotion: Optional[str | list[str]] = None,
        volume: Optional[float] = None,
        base_url: str = CARTESIA_BASE_URL,
        api_version: str = CARTESIA_API_VERSION,
        session: Optional["aiohttp.ClientSession"] = None,
    ) -> None:
        if aiohttp is None:
            raise ImportError(
                "aiohttp is required for CartesiaTTS. "
                "Install with: pip install getpatter[cartesia]"
            )

        resolved_key = api_key or os.environ.get("CARTESIA_API_KEY")
        if not resolved_key:
            raise ValueError(
                "Cartesia API key is required, either as argument or set "
                "CARTESIA_API_KEY environment variable"
            )

        self.api_key = resolved_key
        self.model = model
        self.voice = voice
        self.language = language
        self.sample_rate = sample_rate
        self.encoding = encoding
        self.speed = speed
        self.emotion = [emotion] if isinstance(emotion, str) else emotion
        self.volume = volume
        self.base_url = base_url
        self.api_version = api_version
        self._owns_session = session is None
        self._session = session

    def __repr__(self) -> str:
        # Never leak the API key in repr / logs.
        return (
            f"CartesiaTTS(model={self.model!r}, voice={self.voice!r}, "
            f"language={self.language!r}, sample_rate={self.sample_rate})"
        )

    # ------------------------------------------------------------------
    # Telephony factories
    # ------------------------------------------------------------------

    @classmethod
    def for_twilio(
        cls,
        api_key: Optional[str] = None,
        **kwargs: Any,
    ) -> "CartesiaTTS":
        """Build an instance pre-configured for Twilio Media Streams.

        Emits ``pcm_mulaw`` @ 8 kHz — exactly Twilio's wire codec — so the
        pipeline takes the passthrough path: zero resampling, zero PCM ->
        mu-law encoding, bit-clean audio. This generalises the ElevenLabs
        ``set_telephony_carrier`` mu-law-native trick to Cartesia.

        Previously this requested ``pcm_s16le`` @ 16 kHz because the sender
        hardcoded a 16 kHz -> 8 kHz decimator; now the sender reads the
        declared format (:meth:`source_audio_format`) so an 8 kHz mu-law
        source flows straight to the wire with correct pitch.
        """
        kwargs.pop("sample_rate", None)
        kwargs.pop("encoding", None)
        return cls(api_key=api_key, encoding="pcm_mulaw", sample_rate=8000, **kwargs)

    @classmethod
    def for_telnyx(
        cls,
        api_key: Optional[str] = None,
        **kwargs: Any,
    ) -> "CartesiaTTS":
        """Build an instance pre-configured for Telnyx bidirectional media.

        The SDK's ``streaming_start`` pins the Telnyx wire to PCMU/mu-law @
        8 kHz, so emitting ``pcm_mulaw`` @ 8 kHz flows end-to-end with zero
        resampling or transcoding — the same passthrough win as
        :meth:`for_twilio`.
        """
        kwargs.pop("sample_rate", None)
        kwargs.pop("encoding", None)
        return cls(api_key=api_key, encoding="pcm_mulaw", sample_rate=8000, **kwargs)

    def source_audio_format(self) -> "AudioFormat":
        """Declare the audio format this adapter emits, so the pipeline sender
        derives the correct resample ratio (or skips it for mu-law passthrough)
        instead of assuming a fixed 16 kHz source. See ``getpatter.audio.format``.
        """
        from getpatter.audio.format import AudioFormat

        if self.encoding == "pcm_mulaw":
            return AudioFormat(encoding="mulaw", sample_rate=int(self.sample_rate))
        if self.encoding == "pcm_alaw":
            return AudioFormat(encoding="alaw", sample_rate=int(self.sample_rate))
        return AudioFormat(encoding="pcm_s16le", sample_rate=int(self.sample_rate))

    def _ensure_session(self) -> "aiohttp.ClientSession":
        if self._session is None:
            self._session = aiohttp.ClientSession()
            self._owns_session = True
        return self._session

    def _build_payload(self, text: str) -> dict[str, Any]:
        voice: dict[str, Any] = {"mode": "id", "id": self.voice}

        payload: dict[str, Any] = {
            "model_id": self.model,
            "voice": voice,
            "transcript": text,
            "output_format": {
                "container": "raw",
                "encoding": self.encoding,
                "sample_rate": self.sample_rate,
            },
            "language": self.language,
        }

        generation_config: dict[str, Any] = {}
        if self.speed is not None:
            generation_config["speed"] = self.speed
        if self.emotion:
            generation_config["emotion"] = self.emotion[0]
        if self.volume is not None:
            generation_config["volume"] = self.volume
        if generation_config:
            payload["generation_config"] = generation_config

        return payload

    def _record_synthesis_cost(self, text: str) -> None:
        """Emit ``patter.cost.tts_chars`` for the synthesised text."""
        try:
            from getpatter.observability.attributes import record_patter_attrs

            record_patter_attrs(
                {
                    "patter.cost.tts_chars": len(text),
                    "patter.tts.provider": "cartesia_tts",
                }
            )
        except Exception:  # pragma: no cover — defense in depth
            logger.debug("_record_synthesis_cost failed", exc_info=True)

    async def synthesize(self, text: str) -> AsyncIterator[bytes]:
        """Stream raw PCM_S16LE bytes for ``text`` over HTTP."""
        self._record_synthesis_cost(text)
        session = self._ensure_session()

        headers = {
            "X-API-Key": self.api_key,
            "Cartesia-Version": self.api_version,
            "Content-Type": "application/json",
        }

        async with session.post(
            f"{self.base_url}/tts/bytes",
            headers=headers,
            json=self._build_payload(text),
            timeout=aiohttp.ClientTimeout(total=30),
        ) as resp:
            resp.raise_for_status()
            async for chunk in resp.content.iter_chunked(4096):
                if chunk:
                    yield chunk

    async def warmup(self) -> None:
        """Pre-call HTTP warmup for the Cartesia ``/tts/bytes`` endpoint.

        Issues a lightweight ``GET <base_url>/voices`` so DNS, TLS, and
        HTTP/2 are already up by the time the first :meth:`synthesize`
        POST lands. Best-effort: 5 s timeout, all exceptions swallowed
        at DEBUG.

        Billing safety: ``GET /voices`` is a free metadata read on
        Cartesia's REST surface (per https://docs.cartesia.ai). It does
        not consume any synthesis credits. The actual synthesis is billed
        only when ``POST /tts/bytes`` runs with a non-empty ``transcript``.

        Note: Cartesia TTS uses the HTTP path (vs the WebSocket variant
        Cartesia also exposes) — connection warmup is therefore HTTP-GET
        based, not WebSocket pre-handshake. The latency win is smaller
        (~50-150 ms vs the ~200-500 ms of a WS prewarm) but still real.
        """
        try:
            session = self._ensure_session()
            headers = {
                "X-API-Key": self.api_key,
                "Cartesia-Version": self.api_version,
            }
            async with session.get(
                f"{self.base_url}/voices",
                headers=headers,
                timeout=aiohttp.ClientTimeout(total=5),
            ) as resp:
                # Drain the body so the underlying connection returns to
                # the pool ready for the next request.
                await resp.read()
        except Exception as exc:  # noqa: BLE001 - best-effort
            logger.debug("Cartesia TTS warmup failed (best-effort): %s", exc)

    async def close(self) -> None:
        """Close the underlying session (idempotent)."""
        if self._session is not None and self._owns_session:
            await self._session.close()
            self._session = None
