app/client.py

"""
VoiceCloneClient — a thin router/facade over available TTS providers.

Selects a provider per-request based on:
  1. A forced provider (if the caller supplies one).
  2. The canonical LANGUAGE_CATALOG hint (`chatterbox` / `minimax`).
  3. Availability (is_configured + supports_language).

Handles: retry-once on transient errors, user-readable error messages,
and language discovery merged across all providers.
"""
from __future__ import annotations

import logging
import time
import wave
from pathlib import Path
from typing import Optional

from app.config import LANGUAGE_CATALOG, Settings, get_settings
from app.errors import (
    ConfigError,
    ProviderHTTPError,
    ProviderTimeoutError,
    ProviderUnavailableError,
    ReferenceAudioError,
    UnsupportedLanguageError,
    VoiceCloneError,
)
from app.providers import ChatterboxProvider, MinimaxProvider
from app.providers.base import GenerationResult, HealthStatus, TTSProvider

log = logging.getLogger(__name__)


TRANSIENT_STATUS_CODES = {408, 429, 500, 502, 503, 504}


class VoiceCloneClient:
    """High-level orchestrator used by Gradio + tests."""

    def __init__(
        self,
        settings: Settings | None = None,
        chatterbox: Optional[TTSProvider] = None,
        minimax: Optional[TTSProvider] = None,
    ):
        self.settings = settings or get_settings()
        self.chatterbox: TTSProvider = chatterbox or ChatterboxProvider(settings=self.settings)
        self.minimax: TTSProvider = minimax or MinimaxProvider(settings=self.settings)

    # ------------------------------------------------------------------
    # Language discovery
    # ------------------------------------------------------------------
    def discover_languages(self) -> dict[str, dict[str, str]]:
        """
        Return the languages we will actually expose in the UI.

        Starts from LANGUAGE_CATALOG, then filters to those the configured
        provider(s) can actually serve. Cantonese is enforced as mandatory:
        if no configured provider supports `yue`, we log a loud warning.
        """
        chatterbox_langs = set(self.chatterbox.supported_languages()) \
            if self.chatterbox.is_configured() else set()
        minimax_langs = set(self.minimax.supported_languages()) \
            if self.minimax.is_configured() else set()

        available: dict[str, dict[str, str]] = {}
        for code, meta in LANGUAGE_CATALOG.items():
            if meta["provider_hint"] == "chatterbox" and code in chatterbox_langs:
                available[code] = meta
            elif meta["provider_hint"] == "minimax" and code in minimax_langs:
                available[code] = meta
            elif code in chatterbox_langs or code in minimax_langs:
                # Catalog says X but in practice either provider has it.
                available[code] = meta

        if "yue" not in available:
            log.warning(
                "Cantonese (yue) is not available in the current configuration. "
                "Configure MINIMAX_API_KEY or another yue-capable backend."
            )
        return available

    def cantonese_is_available(self) -> bool:
        return "yue" in self.discover_languages()

    # ------------------------------------------------------------------
    # Health
    # ------------------------------------------------------------------
    def health(self) -> dict[str, HealthStatus]:
        return {
            "chatterbox": self.chatterbox.health(),
            "minimax": self.minimax.health(),
        }

    def any_provider_reachable(self) -> bool:
        return any(h.reachable for h in self.health().values())

    # ------------------------------------------------------------------
    # Reference audio validation
    # ------------------------------------------------------------------
    def validate_reference_audio(self, path: Path) -> tuple[bool, str]:
        """Return (ok, message). Message includes guidance for short clips."""
        if not path.exists():
            return False, "Reference audio file not found."
        size_mb = path.stat().st_size / 1_000_000
        if size_mb > self.settings.max_reference_mb:
            return False, (
                f"Reference audio is {size_mb:.1f} MB which exceeds the "
                f"{self.settings.max_reference_mb} MB limit."
            )
        try:
            with wave.open(str(path), "rb") as w:
                duration = w.getnframes() / float(w.getframerate() or 1)
        except wave.Error as e:
            return False, (
                f"Reference audio is not a valid WAV file ({e}). "
                "Please upload a PCM WAV at 16 kHz or higher."
            )
        if duration < self.settings.min_reference_seconds:
            return False, (
                f"Reference audio is only {duration:.1f}s — minimum is "
                f"{self.settings.min_reference_seconds:.0f}s."
            )
        if duration > self.settings.max_reference_seconds:
            return False, (
                f"Reference audio is {duration:.1f}s — trim to "
                f"≤ {self.settings.max_reference_seconds:.0f}s for best results."
            )
        if duration < self.settings.recommended_reference_seconds_min:
            return True, (
                f"Clip is {duration:.1f}s. Recommended is "
                f"{self.settings.recommended_reference_seconds_min:.0f}–"
                f"{self.settings.recommended_reference_seconds_max:.0f}s "
                f"for a cleaner clone."
            )
        return True, f"Reference OK ({duration:.1f}s)."

    # ------------------------------------------------------------------
    # Provider routing
    # ------------------------------------------------------------------
    def _pick_provider(self, language_code: str,
                       force: Optional[str] = None) -> TTSProvider:
        if force == "chatterbox":
            if not self.chatterbox.is_configured():
                raise ConfigError("Chatterbox backend not configured.")
            return self.chatterbox
        if force == "minimax":
            if not self.minimax.is_configured():
                raise ConfigError("MiniMax backend not configured.")
            return self.minimax

        meta = LANGUAGE_CATALOG.get(language_code)
        hint = meta["provider_hint"] if meta else None

        # First try the catalog hint. If the hinted provider is configured,
        # prefer it even if `/languages` discovery failed transiently — the
        # catalog is our source of truth for what each provider *can* do.
        if hint == "minimax" and self.minimax.is_configured():
            return self.minimax
        if hint == "chatterbox" and self.chatterbox.is_configured():
            return self.chatterbox

        # Fallback: any configured provider that supports the language.
        if self.chatterbox.is_configured() and self.chatterbox.supports_language(language_code):
            return self.chatterbox
        if self.minimax.is_configured() and self.minimax.supports_language(language_code):
            return self.minimax

        raise UnsupportedLanguageError(
            f"No configured provider supports language '{language_code}'. "
            f"Check CHATTERBOX_API_BASE_URL or MINIMAX_API_KEY."
        )

    # ------------------------------------------------------------------
    # Generation
    # ------------------------------------------------------------------
    def clone_voice(
        self,
        *,
        reference_audio_path: Path,
        text: str,
        language_code: str,
        exaggeration: Optional[float] = None,
        cfg_weight: Optional[float] = None,
        temperature: Optional[float] = None,
        force_provider: Optional[str] = None,
    ) -> GenerationResult:
        if not text or not text.strip():
            raise VoiceCloneError("Please enter some text to synthesise.")

        ok, msg = self.validate_reference_audio(reference_audio_path)
        if not ok:
            raise ReferenceAudioError(msg)

        exag = exaggeration if exaggeration is not None else self.settings.default_exaggeration
        cfg = cfg_weight if cfg_weight is not None else self.settings.default_cfg_weight
        tmp = temperature if temperature is not None else self.settings.default_temperature

        # Clamp to documented ranges.
        exag = max(self.settings.exaggeration_range[0],
                   min(self.settings.exaggeration_range[1], exag))
        cfg = max(self.settings.cfg_weight_range[0],
                  min(self.settings.cfg_weight_range[1], cfg))
        tmp = max(self.settings.temperature_range[0],
                  min(self.settings.temperature_range[1], tmp))

        provider = self._pick_provider(language_code, force=force_provider)

        attempt = 0
        last_exc: Exception | None = None
        max_attempts = 2 if self.settings.request_retry_once else 1
        while attempt < max_attempts:
            attempt += 1
            try:
                return provider.generate(
                    text=text,
                    reference_audio_path=reference_audio_path,
                    language_code=language_code,
                    exaggeration=exag,
                    cfg_weight=cfg,
                    temperature=tmp,
                )
            except ProviderTimeoutError as e:
                last_exc = e
                log.warning("provider timeout (attempt %d/%d): %s",
                            attempt, max_attempts, e)
            except ProviderUnavailableError as e:
                last_exc = e
                log.warning("provider unavailable (attempt %d/%d): %s",
                            attempt, max_attempts, e)
            except ProviderHTTPError as e:
                last_exc = e
                if e.status_code in TRANSIENT_STATUS_CODES:
                    log.warning("transient HTTP %d (attempt %d/%d): %s",
                                e.status_code, attempt, max_attempts, e)
                else:
                    # Non-transient: surface immediately.
                    if self.settings.log_provider_errors:
                        log.error("provider error body: %s", e.provider_response)
                    raise
            if attempt < max_attempts:
                time.sleep(1.0)
        assert last_exc is not None
        if self.settings.log_provider_errors and hasattr(last_exc, "provider_response"):
            log.error("final provider error body: %s",
                      getattr(last_exc, "provider_response", None))
        raise last_exc

    def language_tour(
        self,
        *,
        reference_audio_path: Path,
        text: str,
        language_codes: Optional[list[str]] = None,
        exaggeration: Optional[float] = None,
        cfg_weight: Optional[float] = None,
    ) -> list[tuple[str, Optional[GenerationResult], Optional[str]]]:
        """
        Clone the same voice across multiple languages.

        Returns a list of (language_code, result_or_None, error_or_None).
        Never raises — individual failures are captured per-language.
        """
        langs = language_codes or list(self.discover_languages().keys())
        out: list[tuple[str, Optional[GenerationResult], Optional[str]]] = []
        for code in langs:
            try:
                res = self.clone_voice(
                    reference_audio_path=reference_audio_path,
                    text=text,
                    language_code=code,
                    exaggeration=exaggeration,
                    cfg_weight=cfg_weight,
                )
                out.append((code, res, None))
            except VoiceCloneError as e:
                out.append((code, None, str(e)))
        return out

    # ------------------------------------------------------------------
    # Cleanup
    # ------------------------------------------------------------------
    def cleanup_temp(self, max_age_seconds: int = 3600) -> int:
        """Delete temp WAVs older than max_age_seconds. Returns count deleted."""
        now = time.time()
        count = 0
        if not self.settings.temp_dir.exists():
            return 0
        for p in self.settings.temp_dir.glob("*.wav"):
            try:
                if now - p.stat().st_mtime > max_age_seconds:
                    p.unlink()
                    count += 1
            except OSError:
                continue
        return count