digiops/call-center
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(voice): add admin-configurable elevenlabs tts

Browse Source
This commit is contained in:
Yera All
2026-04-11 16:46:43 +05:00
parent 9d0b76deca
commit fc976804e4
14 changed files with 1342 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/shared/models.py
+54 -1
View File
@@ -2,7 +2,7 @@
from typing import Literal
from pydantic import BaseModel, Field, field_validator, model_validator
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from services.shared.core import Role
@@ -98,6 +98,59 @@ class VoiceNameCollectionConfigOut(BaseModel):
source: Literal["defaults", "database"] = "defaults"
VoiceTTSProviderName = Literal["yandex", "elevenlabs", "openai"]
VoiceTTSLanguage = Literal["ru", "kz"]
class VoiceTTSLanguageConfig(BaseModel):
model_config = ConfigDict(protected_namespaces=())
voice: str | None = None
model_id: str | None = None
language_code: str | None = None
@field_validator("voice", "model_id", "language_code", mode="before")
@classmethod
def _trim_optional_text(cls, value: str | None) -> str | None:
if value is None:
return None
normalized = str(value).strip()
return normalized or None
class VoiceTTSProviderConfig(BaseModel):
ru: VoiceTTSLanguageConfig = Field(default_factory=VoiceTTSLanguageConfig)
kz: VoiceTTSLanguageConfig = Field(default_factory=VoiceTTSLanguageConfig)
class VoiceTTSVoiceOption(BaseModel):
value: str = Field(min_length=1)
label: str = Field(min_length=1)
@field_validator("value", "label", mode="before")
@classmethod
def _trim_required_option_text(cls, value: str) -> str:
normalized = str(value or "").strip()
if not normalized:
raise ValueError("Option value is required")
return normalized
class VoiceTTSConfig(BaseModel):
provider: VoiceTTSProviderName = "yandex"
yandex: VoiceTTSProviderConfig = Field(default_factory=VoiceTTSProviderConfig)
elevenlabs: VoiceTTSProviderConfig = Field(default_factory=VoiceTTSProviderConfig)
openai: VoiceTTSProviderConfig = Field(default_factory=VoiceTTSProviderConfig)
class VoiceTTSConfigOut(BaseModel):
config: VoiceTTSConfig
updated_at: str | None = None
source: Literal["defaults", "database"] = "defaults"
provider_options: list[VoiceTTSProviderName] = Field(default_factory=list)
voice_options: dict[str, dict[str, list[VoiceTTSVoiceOption]]] = Field(default_factory=dict)
class HealthResponse(BaseModel):
status: str
service: str
services/shared/sql_models.py
+9
View File
@@ -129,6 +129,15 @@ class VoiceNameCollectionSettingsRow(Base):
updated_at: Mapped[str] = mapped_column(String(64), index=True)
class VoiceTTSSettingsRow(Base):
__tablename__ = "voice_tts_settings"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
settings_key: Mapped[str] = mapped_column(String(64), unique=True, index=True)
config_json: Mapped[str] = mapped_column(Text, default="{}")
updated_at: Mapped[str] = mapped_column(String(64), index=True)
class RoutingCounter(Base):
__tablename__ = "routing_counters"
services/shared/voice_tts_config.py
+230
View File
@@ -0,0 +1,230 @@
from __future__ import annotations
import json
import os
from typing import Any
from sqlalchemy import select
from services.shared.core import utc_now_iso
from services.shared.models import (
VoiceTTSConfig,
VoiceTTSConfigOut,
VoiceTTSLanguageConfig,
VoiceTTSProviderConfig,
VoiceTTSVoiceOption,
)
from services.shared.sql_models import VoiceTTSSettingsRow
VOICE_TTS_SETTINGS_KEY = "global"
def _normalize_provider_name(value: str | None) -> str:
normalized = str(value or "yandex").strip().lower()
if normalized in {"speechkit", "yandex_speechkit"}:
return "yandex"
if normalized in {"11labs"}:
return "elevenlabs"
if normalized not in {"yandex", "elevenlabs", "openai"}:
return "yandex"
return normalized
def _openai_default_model() -> str:
return str(os.getenv("AI_VOICE_TTS_MODEL", "gpt-4o-mini-tts") or "gpt-4o-mini-tts").strip() or "gpt-4o-mini-tts"
def _openai_default_voice() -> str:
return str(os.getenv("AI_VOICE_TTS_VOICE", "alloy") or "alloy").strip() or "alloy"
def _yandex_default_voice(language: str) -> str:
normalized = str(language or "ru").strip().lower()
if normalized == "kz":
return str(os.getenv("AI_VOICE_TTS_YANDEX_KK_VOICE", "amira") or "amira").strip() or "amira"
return str(os.getenv("AI_VOICE_TTS_YANDEX_VOICE", "jane") or "jane").strip() or "jane"
def _elevenlabs_default_model() -> str:
return str(os.getenv("AI_VOICE_TTS_ELEVENLABS_MODEL_ID", "eleven_v3") or "eleven_v3").strip() or "eleven_v3"
def _elevenlabs_default_voice(language: str) -> str:
normalized = str(language or "ru").strip().lower()
if normalized == "kz":
return (
str(os.getenv("AI_VOICE_TTS_ELEVENLABS_KK_VOICE_ID", "nPczCjzI2devNBz1zQrb") or "nPczCjzI2devNBz1zQrb").strip()
or "nPczCjzI2devNBz1zQrb"
)
return (
str(os.getenv("AI_VOICE_TTS_ELEVENLABS_RU_VOICE_ID", "nPczCjzI2devNBz1zQrb") or "nPczCjzI2devNBz1zQrb").strip()
or "nPczCjzI2devNBz1zQrb"
)
def voice_tts_default_config() -> VoiceTTSConfig:
return VoiceTTSConfig(
provider=_normalize_provider_name(os.getenv("AI_VOICE_TTS_PROVIDER", "yandex")),
yandex=VoiceTTSProviderConfig(
ru=VoiceTTSLanguageConfig(voice=_yandex_default_voice("ru")),
kz=VoiceTTSLanguageConfig(voice=_yandex_default_voice("kz")),
),
elevenlabs=VoiceTTSProviderConfig(
ru=VoiceTTSLanguageConfig(
voice=_elevenlabs_default_voice("ru"),
model_id=_elevenlabs_default_model(),
language_code="ru",
),
kz=VoiceTTSLanguageConfig(
voice=_elevenlabs_default_voice("kz"),
model_id=_elevenlabs_default_model(),
language_code="kk",
),
),
openai=VoiceTTSProviderConfig(
ru=VoiceTTSLanguageConfig(
voice=_openai_default_voice(),
model_id=_openai_default_model(),
language_code="ru",
),
kz=VoiceTTSLanguageConfig(
voice=_openai_default_voice(),
model_id=_openai_default_model(),
language_code="kz",
),
),
)
def voice_tts_provider_options() -> list[str]:
return ["yandex", "elevenlabs", "openai"]
def voice_tts_voice_options() -> dict[str, dict[str, list[VoiceTTSVoiceOption]]]:
return {
"yandex": {
"ru": [
VoiceTTSVoiceOption(value="jane", label="Jane"),
VoiceTTSVoiceOption(value="oksana", label="Oksana"),
VoiceTTSVoiceOption(value="ermil", label="Ermil"),
],
"kz": [
VoiceTTSVoiceOption(value="amira", label="Amira"),
VoiceTTSVoiceOption(value="madi", label="Madi"),
VoiceTTSVoiceOption(value="saule", label="Saule"),
VoiceTTSVoiceOption(value="zhanar", label="Zhanar"),
],
},
"elevenlabs": {
"ru": [
VoiceTTSVoiceOption(value="nPczCjzI2devNBz1zQrb", label="Brian"),
],
"kz": [
VoiceTTSVoiceOption(value="nPczCjzI2devNBz1zQrb", label="Brian"),
],
},
"openai": {
"ru": [
VoiceTTSVoiceOption(value="alloy", label="Alloy"),
VoiceTTSVoiceOption(value="echo", label="Echo"),
VoiceTTSVoiceOption(value="fable", label="Fable"),
VoiceTTSVoiceOption(value="onyx", label="Onyx"),
VoiceTTSVoiceOption(value="nova", label="Nova"),
VoiceTTSVoiceOption(value="shimmer", label="Shimmer"),
],
"kz": [
VoiceTTSVoiceOption(value="alloy", label="Alloy"),
VoiceTTSVoiceOption(value="echo", label="Echo"),
VoiceTTSVoiceOption(value="fable", label="Fable"),
VoiceTTSVoiceOption(value="onyx", label="Onyx"),
VoiceTTSVoiceOption(value="nova", label="Nova"),
VoiceTTSVoiceOption(value="shimmer", label="Shimmer"),
],
},
}
def _settings_row(session) -> VoiceTTSSettingsRow | None:
return session.execute(
select(VoiceTTSSettingsRow).where(VoiceTTSSettingsRow.settings_key == VOICE_TTS_SETTINGS_KEY)
).scalar_one_or_none()
def _normalize_config_payload(raw: Any) -> VoiceTTSConfig:
if isinstance(raw, VoiceTTSConfig):
return raw
if not isinstance(raw, dict):
raise ValueError("Voice TTS config payload must be an object")
return VoiceTTSConfig.model_validate(raw)
def load_voice_tts_config(session) -> VoiceTTSConfigOut:
default_config = voice_tts_default_config()
row = _settings_row(session)
if row is None:
return VoiceTTSConfigOut(
config=default_config,
updated_at=None,
source="defaults",
provider_options=voice_tts_provider_options(),
voice_options=voice_tts_voice_options(),
)
try:
parsed = json.loads(row.config_json or "{}")
config = _normalize_config_payload(parsed)
return VoiceTTSConfigOut(
config=config,
updated_at=row.updated_at,
source="database",
provider_options=voice_tts_provider_options(),
voice_options=voice_tts_voice_options(),
)
except Exception:
return VoiceTTSConfigOut(
config=default_config,
updated_at=row.updated_at,
source="defaults",
provider_options=voice_tts_provider_options(),
voice_options=voice_tts_voice_options(),
)
def load_effective_voice_tts_config(session) -> VoiceTTSConfig:
config = load_voice_tts_config(session).config
config.provider = _normalize_provider_name(config.provider)
return config
def save_voice_tts_config(session, config_payload: VoiceTTSConfig | dict[str, Any]) -> VoiceTTSConfigOut:
config = _normalize_config_payload(config_payload)
config.provider = _normalize_provider_name(config.provider)
row = _settings_row(session)
now = utc_now_iso()
if row is None:
row = VoiceTTSSettingsRow(
settings_key=VOICE_TTS_SETTINGS_KEY,
config_json="{}",
updated_at=now,
)
session.add(row)
row.config_json = json.dumps(config.model_dump(), ensure_ascii=False)
row.updated_at = now
session.commit()
session.refresh(row)
return VoiceTTSConfigOut(
config=config,
updated_at=row.updated_at,
source="database",
provider_options=voice_tts_provider_options(),
voice_options=voice_tts_voice_options(),
)
def voice_tts_language_config(config: VoiceTTSConfig, provider: str, language: str) -> VoiceTTSLanguageConfig:
normalized_provider = str(provider or config.provider or "yandex").strip().lower()
provider_config = getattr(config, normalized_provider, None)
if not isinstance(provider_config, VoiceTTSProviderConfig):
provider_config = getattr(voice_tts_default_config(), normalized_provider)
normalized_language = "kz" if str(language or "").strip().lower() == "kz" else "ru"
return provider_config.kz if normalized_language == "kz" else provider_config.ru