from __future__ import annotations

import asyncio
import hashlib
import os
import tempfile
import time
import wave
from dataclasses import dataclass
from pathlib import Path
from typing import AsyncIterator


def _env_bool(name: str, default: bool) -> bool:
    raw = (os.getenv(name, "") or "").strip().lower()
    if not raw:
        return default
    return raw in {"1", "true", "yes", "on"}


def _env_int(name: str, default: int) -> int:
    try:
        return int(os.getenv(name, str(default)))
    except ValueError:
        return default


@dataclass(frozen=True, slots=True)
class InitialGreetingCacheKey:
    agent: str
    text: str
    provider: str
    voice: str
    language: str
    sample_rate: int

    @property
    def digest(self) -> str:
        raw = "\x1f".join((self.agent, self.text, self.provider, self.voice, self.language, str(self.sample_rate)))
        return hashlib.sha256(raw.encode("utf-8")).hexdigest()


class InitialGreetingAudioCache:
    """Read-through cache for an exact, first-turn TTS response.

    The caller is responsible for only offering first-turn text. Cache entries are
    keyed by the rendered text and TTS settings, so a changed greeting never
    reuses stale audio.
    """

    def __init__(self) -> None:
        self.enabled = _env_bool("INITIAL_GREETING_AUDIO_CACHE_ENABLED", True)
        self._max_chars = max(1, _env_int("INITIAL_GREETING_AUDIO_CACHE_MAX_CHARS", 1_000))
        self._ttl_s = max(1, _env_int("INITIAL_GREETING_AUDIO_CACHE_TTL_S", 3_600))
        self._max_entries = max(1, _env_int("INITIAL_GREETING_AUDIO_CACHE_MAX_ENTRIES", 32))
        self._max_bytes = max(44, _env_int("INITIAL_GREETING_AUDIO_CACHE_MAX_BYTES", 16 * 1024 * 1024))
        self._max_agent_variants = max(1, _env_int("INITIAL_GREETING_AUDIO_CACHE_MAX_VARIANTS_PER_AGENT", 1))
        self._disable_ttl_s = max(1, _env_int("INITIAL_GREETING_AUDIO_CACHE_DISABLE_TTL_S", 300))
        self._agent_variants: dict[str, set[str]] = {}
        self._agent_disabled_until: dict[str, float] = {}
        self._dir = Path(os.getenv("INITIAL_GREETING_AUDIO_CACHE_DIR", "./cache/initial-greetings"))

    def key_for(
        self,
        *,
        agent: str,
        text: str,
        provider: str,
        voice: str,
        language: str,
        sample_rate: int,
    ) -> InitialGreetingCacheKey | None:
        normalized = " ".join(str(text or "").split())
        agent_name = (agent or "unknown").strip().lower()
        now = time.monotonic()
        if self._agent_disabled_until.get(agent_name, 0.0) > now:
            return None
        variants = self._agent_variants.setdefault(agent_name, set())
        if normalized not in variants and len(variants) >= self._max_agent_variants:
            self._agent_disabled_until[agent_name] = now + self._disable_ttl_s
            return None
        variants.add(normalized)
        if not self.enabled or not normalized or len(normalized) > self._max_chars:
            return None
        return InitialGreetingCacheKey(
            agent=(agent or "unknown").strip().lower(),
            text=normalized,
            provider=(provider or "unknown").strip().lower(),
            voice=(voice or "default").strip(),
            language=(language or "pt-BR").strip().lower(),
            sample_rate=max(1, int(sample_rate)),
        )

    def path_for(self, key: InitialGreetingCacheKey) -> Path:
        return self._dir / f"{key.digest}.wav"

    def has(self, key: InitialGreetingCacheKey) -> bool:
        path = self.path_for(key)
        try:
            stat = path.stat()
        except FileNotFoundError:
            return False
        if stat.st_size <= 44 or time.time() - stat.st_mtime >= self._ttl_s:
            self.discard(key)
            return False
        os.utime(path, None)
        return True

    def frames(self, key: InitialGreetingCacheKey) -> AsyncIterator[object]:
        return self._materialize_frames(key)

    def discard(self, key: InitialGreetingCacheKey) -> None:
        self.path_for(key).unlink(missing_ok=True)

    def _materialize_frames(self, key: InitialGreetingCacheKey) -> AsyncIterator[object]:
        path = self.path_for(key)
        try:
            with wave.open(str(path), "rb") as wav:
                channels = wav.getnchannels()
                sample_width = wav.getsampwidth()
                sample_rate = wav.getframerate()
                if channels != 1 or sample_width != 2:
                    raise ValueError(f"Unsupported cached greeting WAV format: {path}")
                pcm = wav.readframes(10**9)
            if not pcm:
                raise ValueError(f"Cached greeting WAV is empty: {path}")
        except (OSError, EOFError, ValueError, wave.Error):
            self.discard(key)
            raise

        async def _frames() -> AsyncIterator[object]:
            from livekit import rtc

            samples_per_frame = max(1, round(sample_rate * 20 / 1000))
            frame_bytes = samples_per_frame * 2 * channels
            for offset in range(0, len(pcm), frame_bytes):
                chunk = pcm[offset : offset + frame_bytes]
                if not chunk:
                    continue
                if len(chunk) < frame_bytes:
                    chunk = chunk + (b"\x00" * (frame_bytes - len(chunk)))
                yield rtc.AudioFrame(
                    data=chunk,
                    sample_rate=sample_rate,
                    num_channels=channels,
                    samples_per_channel=len(chunk) // (2 * channels),
                )
                await asyncio.sleep(0)

        return _frames()

    async def store_pcm(self, key: InitialGreetingCacheKey, pcm: bytes) -> None:
        if not pcm or self.has(key):
            return
        await asyncio.to_thread(self._store_pcm_sync, key, pcm)

    def _store_pcm_sync(self, key: InitialGreetingCacheKey, pcm: bytes) -> None:
        self._dir.mkdir(parents=True, exist_ok=True)
        target = self.path_for(key)
        if self.has(key):
            return
        with tempfile.NamedTemporaryFile(dir=self._dir, suffix=".wav", delete=False) as handle:
            tmp_path = Path(handle.name)
        try:
            with wave.open(str(tmp_path), "wb") as wav:
                wav.setnchannels(1)
                wav.setsampwidth(2)
                wav.setframerate(key.sample_rate)
                wav.writeframes(pcm)
            os.replace(tmp_path, target)
            self._enforce_limits_sync()
        finally:
            tmp_path.unlink(missing_ok=True)

    def _enforce_limits_sync(self) -> None:
        entries = [path for path in self._dir.glob("*.wav") if path.is_file()]
        for path in entries[:]:
            if time.time() - path.stat().st_mtime >= self._ttl_s:
                path.unlink(missing_ok=True)
                entries.remove(path)
        entries.sort(key=lambda path: path.stat().st_mtime)
        total_bytes = sum(path.stat().st_size for path in entries)
        while entries and (len(entries) > self._max_entries or total_bytes > self._max_bytes):
            evicted = entries.pop(0)
            total_bytes -= evicted.stat().st_size
            evicted.unlink(missing_ok=True)
