feat: add cache_fingerprint method to TTSProvider and its subclasses for voice/model configuration
deploy / deploy (push) Successful in 30s

This commit is contained in:
2026-08-29 01:23:23 +05:00
parent 09bcf7457c
commit d2a9df36d1
2 changed files with 28 additions and 0 deletions
@@ -49,6 +49,7 @@ class PrebakedAckBank:
) -> str: ) -> str:
payload = { payload = {
"provider": getattr(self._tts_provider, "name", "tts"), "provider": getattr(self._tts_provider, "name", "tts"),
"voice_fingerprint": self._tts_provider.cache_fingerprint(language, style_hints=style_hints),
"language": str(language or "").strip() or None, "language": str(language or "").strip() or None,
"style_hints": style_hints or {}, "style_hints": style_hints or {},
"text": text, "text": text,
@@ -199,6 +199,16 @@ class TTSSynthesis:
class TTSProvider: class TTSProvider:
name = "stub" name = "stub"
def cache_fingerprint(self, language: str | None, *, style_hints: dict[str, object] | None = None) -> str:
"""Identifies the exact voice/model config that would render `language`.
Callers that cache synthesized audio keyed only on (provider, language,
text) — like the prebaked ack bank — must fold this in too, or a voice
change silently keeps serving audio recorded with the old voice.
"""
del language, style_hints
return ""
def synthesize( def synthesize(
self, self,
text: str, text: str,
@@ -257,6 +267,10 @@ class OpenAITTSProvider(TTSProvider):
def _voice(self, language: str | None) -> str: def _voice(self, language: str | None) -> str:
return self._kz_voice if _normalize_voice_language(language) == "kz" else self._ru_voice return self._kz_voice if _normalize_voice_language(language) == "kz" else self._ru_voice
def cache_fingerprint(self, language: str | None, *, style_hints: dict[str, object] | None = None) -> str:
del style_hints
return f"{self._voice(language)}:{self._model(language)}:{self._speed}"
def _cache_key(self, text: str, *, language: str | None) -> str: def _cache_key(self, text: str, *, language: str | None) -> str:
payload = { payload = {
"provider": self.name, "provider": self.name,
@@ -433,6 +447,12 @@ class YandexTTSProvider(TTSProvider):
return explicit_role return explicit_role
return self._role return self._role
def cache_fingerprint(self, language: str | None, *, style_hints: dict[str, object] | None = None) -> str:
return (
f"{self._voice(language)}:{self._effective_role(style_hints=style_hints)}:"
f"{self._speed}:{self._sample_rate_hz}"
)
def _cache_key(self, text: str, *, language: str | None, style_hints: dict[str, object] | None = None) -> str: def _cache_key(self, text: str, *, language: str | None, style_hints: dict[str, object] | None = None) -> str:
payload = { payload = {
"provider": self.name, "provider": self.name,
@@ -643,6 +663,13 @@ class ElevenLabsTTSProvider(TTSProvider):
def _language_code(self, language: str | None) -> str: def _language_code(self, language: str | None) -> str:
return self._kz_language_code if _normalize_voice_language(language) == "kz" else self._ru_language_code return self._kz_language_code if _normalize_voice_language(language) == "kz" else self._ru_language_code
def cache_fingerprint(self, language: str | None, *, style_hints: dict[str, object] | None = None) -> str:
del style_hints
return (
f"{self._voice(language)}:{self._model(language)}:"
f"{self._language_code(language)}:{self._output_format}"
)
def _cache_key( def _cache_key(
self, self,
text: str, text: str,