""" 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