from __future__ import annotations import asyncio import audioop import base64 import contextlib import io import json import logging import os import time import wave from urllib.parse import urlencode from realtime_voice_service.providers.base import BaseSTT from realtime_voice_service.providers.base import BaseSTTStream from realtime_voice_service.providers.base import PartialTranscriptCallback from realtime_voice_service.providers.stt_openai import OpenAISTT LOGGER = logging.getLogger("uvicorn.error") _STREAM_COMMIT = object() _STREAM_CANCEL = object() def _api_base() -> str: return (os.getenv("ELEVENLABS_API_BASE", "https://api.elevenlabs.io").strip() or "https://api.elevenlabs.io").rstrip("/") def _stt_provider_name() -> str: return ( os.getenv("STT_PROVIDER", "").strip() or os.getenv("REALTIME_VOICE_STT_PROVIDER", "").strip() or os.getenv("AI_VOICE_ASR_PROVIDER", "").strip() or "elevenlabs" ).lower() def _stt_fallback_provider_name() -> str: return os.getenv("REALTIME_VOICE_STT_FALLBACK_PROVIDER", "").strip().lower() def _yandex_api_base() -> str: return ( os.getenv("YANDEX_STT_API_BASE", "").strip() or os.getenv("AI_VOICE_ASR_YANDEX_API_BASE", "").strip() or "https://stt.api.cloud.yandex.net" ).rstrip("/") def _yandex_api_key() -> str: return ( os.getenv("YANDEX_STT_API_KEY", "").strip() or os.getenv("AI_VOICE_ASR_YANDEX_API_KEY", "").strip() or os.getenv("AI_VOICE_TTS_YANDEX_API_KEY", "").strip() ) def _yandex_iam_token() -> str: return ( os.getenv("YANDEX_STT_IAM_TOKEN", "").strip() or os.getenv("AI_VOICE_ASR_YANDEX_IAM_TOKEN", "").strip() or os.getenv("AI_VOICE_TTS_YANDEX_IAM_TOKEN", "").strip() ) def _yandex_folder_id() -> str: return ( os.getenv("YANDEX_STT_FOLDER_ID", "").strip() or os.getenv("AI_VOICE_ASR_YANDEX_FOLDER_ID", "").strip() or os.getenv("AI_VOICE_TTS_YANDEX_FOLDER_ID", "").strip() ) def _yandex_language() -> str: raw = ( os.getenv("YANDEX_STT_LANGUAGE", "").strip() or os.getenv("AI_VOICE_ASR_YANDEX_LANGUAGE", "").strip() or "ru-RU" ) normalized = raw.lower().replace("_", "-") mapping = { "ru": "ru-RU", "rus": "ru-RU", "ru-ru": "ru-RU", "kk": "kk-KZ", "kaz": "kk-KZ", "kz": "kk-KZ", "kk-kz": "kk-KZ", "kz-kz": "kk-KZ", } return mapping.get(normalized, raw) def _yandex_topic() -> str: return os.getenv("YANDEX_STT_TOPIC", "").strip() or os.getenv("AI_VOICE_ASR_YANDEX_TOPIC", "general").strip() or "general" def _ws_base(http_base: str) -> str: normalized = http_base.rstrip("/") if normalized.startswith("https://"): return f"wss://{normalized[len('https://') :]}" if normalized.startswith("http://"): return f"ws://{normalized[len('http://') :]}" return normalized def _timeout_seconds() -> float: raw = os.getenv("ELEVENLABS_TIMEOUT_SECONDS") if raw is None: return 30.0 try: return max(float(raw.strip()), 1.0) except ValueError: return 30.0 def _bool_env(name: str, default: bool) -> bool: raw = os.getenv(name) if raw is None: return default return str(raw).strip().lower() in {"1", "true", "yes", "on"} def _preview_text(text: str, *, limit: int = 120) -> str: normalized = " ".join(str(text or "").split()) if len(normalized) <= limit: return normalized return f"{normalized[:limit]}..." def _pcm16le_to_wav_bytes( pcm_bytes: bytes, *, sample_rate_hz: int, channels: int = 1, sample_width_bytes: int = 2, ) -> bytes: handle = io.BytesIO() with wave.open(handle, "wb") as wav_file: wav_file.setnchannels(channels) wav_file.setsampwidth(sample_width_bytes) wav_file.setframerate(sample_rate_hz) wav_file.writeframes(pcm_bytes) return handle.getvalue() def _resample_pcm16le( pcm_bytes: bytes, *, input_rate_hz: int, output_rate_hz: int, ) -> bytes: if not pcm_bytes or input_rate_hz == output_rate_hz: return pcm_bytes converted, _ = audioop.ratecv( pcm_bytes, 2, 1, input_rate_hz, output_rate_hz, None, ) return converted def _pcm_duration_ms(pcm_bytes: bytes, *, sample_rate_hz: int) -> int: if not pcm_bytes or sample_rate_hz <= 0: return 0 sample_count = len(pcm_bytes) // 2 return max(int((sample_count / float(sample_rate_hz)) * 1000.0), 0) def _pad_pcm16le_to_duration( pcm_bytes: bytes, *, sample_rate_hz: int, min_duration_ms: int, ) -> bytes: if not pcm_bytes: return pcm_bytes current_duration_ms = _pcm_duration_ms(pcm_bytes, sample_rate_hz=sample_rate_hz) if current_duration_ms >= min_duration_ms: return pcm_bytes target_samples = max(int(sample_rate_hz * (min_duration_ms / 1000.0)), 1) target_bytes = target_samples * 2 if len(pcm_bytes) >= target_bytes: return pcm_bytes return pcm_bytes + (b"\x00" * (target_bytes - len(pcm_bytes))) class ElevenLabsRealtimeSTTStream(BaseSTTStream): def __init__( self, *, api_key: str, websocket_url: str, sample_rate_hz: int, timeout_seconds: float, partial_callback: PartialTranscriptCallback | None = None, ) -> None: self._api_key = api_key self._websocket_url = websocket_url self._sample_rate_hz = sample_rate_hz self._timeout_seconds = timeout_seconds self._partial_callback = partial_callback self._websocket = None self._sender_task: asyncio.Task[None] | None = None self._receiver_task: asyncio.Task[None] | None = None self._outgoing_queue: asyncio.Queue[bytes | object] = asyncio.Queue() self._final_transcript_future: asyncio.Future[str] = asyncio.get_running_loop().create_future() self._close_lock = asyncio.Lock() self._closed = False self._commit_requested = False self._audio_sent = False self._latest_partial = "" self._latest_committed = "" self._connect_started_monotonic: float | None = None self._sent_chunk_count = 0 self._sent_byte_count = 0 self._partial_count = 0 self._committed_count = 0 async def connect(self) -> None: try: from websockets.legacy.client import connect as websocket_connect except Exception as exc: # noqa: BLE001 raise RuntimeError("The `websockets` package is required for ElevenLabs realtime STT") from exc self._connect_started_monotonic = time.perf_counter() LOGGER.info( "ElevenLabs realtime STT connecting: sample_rate=%s timeout=%s", self._sample_rate_hz, self._timeout_seconds, ) self._websocket = await websocket_connect( self._websocket_url, extra_headers=[("xi-api-key", self._api_key)], open_timeout=min(self._timeout_seconds, 10.0), close_timeout=1.0, ping_interval=20.0, ping_timeout=20.0, max_size=None, ) try: await self._await_session_started() except Exception: await self._close_websocket() raise LOGGER.info( "ElevenLabs realtime STT connected: sample_rate=%s connect_ms=%s", self._sample_rate_hz, int((time.perf_counter() - self._connect_started_monotonic) * 1000.0), ) self._sender_task = asyncio.create_task(self._sender_loop(), name="elevenlabs-stt-sender") self._receiver_task = asyncio.create_task(self._receiver_loop(), name="elevenlabs-stt-receiver") async def push_audio(self, audio_chunk: bytes) -> None: if self._closed or not audio_chunk: return await self._outgoing_queue.put(audio_chunk) async def finish(self) -> str: if self._closed: return self._latest_committed or self._latest_partial self._commit_requested = True finished_started_monotonic = time.perf_counter() LOGGER.info( "ElevenLabs realtime STT finish requested: sent_chunks=%s sent_bytes=%s latest_partial=%r", self._sent_chunk_count, self._sent_byte_count, _preview_text(self._latest_partial), ) await self._outgoing_queue.put(_STREAM_COMMIT) try: transcript = await asyncio.wait_for( asyncio.shield(self._final_transcript_future), timeout=self._timeout_seconds, ) LOGGER.info( "ElevenLabs realtime STT final transcript: latency_ms=%s chars=%s transcript=%r", int((time.perf_counter() - finished_started_monotonic) * 1000.0), len(transcript), _preview_text(transcript), ) return transcript except asyncio.TimeoutError as exc: transcript = self._latest_committed or self._latest_partial if transcript: LOGGER.warning("ElevenLabs realtime STT finish timed out; returning best-effort transcript") return transcript raise RuntimeError("ElevenLabs realtime STT finish timed out") from exc finally: await self._shutdown() async def cancel(self) -> None: if self._closed: return LOGGER.info( "ElevenLabs realtime STT cancel requested: sent_chunks=%s sent_bytes=%s partials=%s commits=%s", self._sent_chunk_count, self._sent_byte_count, self._partial_count, self._committed_count, ) await self._outgoing_queue.put(_STREAM_CANCEL) await self._shutdown() async def _sender_loop(self) -> None: pending_chunk: bytes | None = None try: while True: item = await self._outgoing_queue.get() if item is _STREAM_CANCEL: return if item is _STREAM_COMMIT: if pending_chunk is not None: await self._send_audio_chunk(pending_chunk, commit=True) pending_chunk = None elif not self._final_transcript_future.done(): self._final_transcript_future.set_result(self._latest_committed or self._latest_partial) return audio_chunk = bytes(item) if not audio_chunk: continue self._audio_sent = True if pending_chunk is not None: await self._send_audio_chunk(pending_chunk, commit=False) pending_chunk = audio_chunk except asyncio.CancelledError: raise except Exception as exc: # noqa: BLE001 if not self._final_transcript_future.done(): self._final_transcript_future.set_exception(exc) raise async def _receiver_loop(self) -> None: try: while True: payload = await self._receive_payload() message_type = str(payload.get("message_type") or "").strip().lower() if message_type == "partial_transcript": self._latest_partial = str(payload.get("text") or "").strip() self._partial_count += 1 LOGGER.info( "ElevenLabs realtime STT partial: index=%s chars=%s text=%r", self._partial_count, len(self._latest_partial), _preview_text(self._latest_partial), ) await self._emit_partial(self._latest_partial) continue if message_type in {"committed_transcript", "committed_transcript_with_timestamps"}: self._latest_committed = str(payload.get("text") or "").strip() self._committed_count += 1 LOGGER.info( "ElevenLabs realtime STT committed: index=%s chars=%s text=%r", self._committed_count, len(self._latest_committed), _preview_text(self._latest_committed), ) if self._commit_requested and not self._final_transcript_future.done(): self._final_transcript_future.set_result(self._latest_committed) continue if message_type == "session_started": continue if self._is_error_message(message_type): raise RuntimeError(self._format_realtime_error(payload)) except asyncio.CancelledError: raise except Exception as exc: # noqa: BLE001 if not self._final_transcript_future.done(): transcript = self._latest_committed or self._latest_partial if transcript and self._commit_requested: self._final_transcript_future.set_result(transcript) else: self._final_transcript_future.set_exception(exc) raise async def _send_audio_chunk(self, audio_chunk: bytes, *, commit: bool) -> None: websocket = self._require_websocket() payload = { "message_type": "input_audio_chunk", "audio_base_64": base64.b64encode(audio_chunk).decode("ascii"), "sample_rate": self._sample_rate_hz, "commit": commit, } await websocket.send(json.dumps(payload)) self._sent_chunk_count += 1 self._sent_byte_count += len(audio_chunk) if commit or self._sent_chunk_count == 1 or self._sent_chunk_count % 10 == 0: LOGGER.info( "ElevenLabs realtime STT audio sent: chunks=%s bytes=%s last_chunk_bytes=%s commit=%s", self._sent_chunk_count, self._sent_byte_count, len(audio_chunk), commit, ) async def _await_session_started(self) -> None: while True: payload = await self._receive_payload() message_type = str(payload.get("message_type") or "").strip().lower() if message_type == "session_started": return if self._is_error_message(message_type): raise RuntimeError(self._format_realtime_error(payload)) async def _receive_payload(self) -> dict[str, object]: websocket = self._require_websocket() raw_message = await asyncio.wait_for(websocket.recv(), timeout=self._timeout_seconds) if isinstance(raw_message, bytes): raw_message = raw_message.decode("utf-8") try: payload = json.loads(raw_message) except json.JSONDecodeError as exc: raise RuntimeError("ElevenLabs realtime STT returned invalid JSON") from exc if not isinstance(payload, dict): raise RuntimeError("ElevenLabs realtime STT returned malformed payload") return payload async def _emit_partial(self, text: str) -> None: if not text or self._partial_callback is None: return maybe_awaitable = self._partial_callback(text) if maybe_awaitable is not None: await maybe_awaitable async def _shutdown(self) -> None: async with self._close_lock: if self._closed: return self._closed = True tasks = [self._sender_task, self._receiver_task] for task in tasks: if task is not None and not task.done(): task.cancel() for task in tasks: if task is not None: try: await task except asyncio.CancelledError: pass except Exception: LOGGER.debug("ignored realtime STT background task failure during shutdown", exc_info=True) await self._close_websocket() async def _close_websocket(self) -> None: websocket = self._websocket self._websocket = None if websocket is None: return with contextlib.suppress(Exception): await websocket.close() def _require_websocket(self): if self._websocket is None: raise RuntimeError("ElevenLabs realtime STT websocket is not connected") return self._websocket @staticmethod def _is_error_message(message_type: str) -> bool: return message_type in {"error", "auth_error", "input_error"} or message_type.endswith("_error") @staticmethod def _format_realtime_error(payload: dict[str, object]) -> str: message_type = str(payload.get("message_type") or "error") detail = str(payload.get("message") or payload.get("detail") or payload.get("error") or "").strip() if detail: return f"ElevenLabs realtime STT {message_type}: {detail}" return f"ElevenLabs realtime STT {message_type}" class ElevenLabsSTT(BaseSTT): def __init__( self, *, api_key: str | None = None, api_base: str | None = None, model_id: str | None = None, realtime_model_id: str | None = None, input_sample_rate_hz: int = 16000, target_sample_rate_hz: int = 16000, timeout_seconds: float | None = None, language_code: str | None = None, use_realtime: bool | None = None, allow_batch_fallback: bool | None = None, realtime_chunk_duration_ms: int = 120, ) -> None: self._api_key = str(api_key if api_key is not None else os.getenv("ELEVENLABS_API_KEY", "")).strip() self._api_base = str(api_base or _api_base()).strip().rstrip("/") self._ws_base = _ws_base(self._api_base) self._model_id = str(model_id or os.getenv("ELEVENLABS_STT_MODEL_ID", "scribe_v2")).strip() or "scribe_v2" self._realtime_model_id = ( str(realtime_model_id or os.getenv("ELEVENLABS_STT_REALTIME_MODEL_ID", "scribe_v2_realtime")).strip() or "scribe_v2_realtime" ) self._input_sample_rate_hz = max(int(input_sample_rate_hz), 1) self._target_sample_rate_hz = max(int(target_sample_rate_hz), 1) self._timeout_seconds = max(float(timeout_seconds if timeout_seconds is not None else _timeout_seconds()), 1.0) self._language_code = str(language_code or os.getenv("ELEVENLABS_STT_LANGUAGE_CODE", "ru")).strip() or "ru" self._use_realtime = ( bool(use_realtime) if use_realtime is not None else _bool_env("ELEVENLABS_STT_USE_REALTIME", True) ) self._allow_batch_fallback = ( bool(allow_batch_fallback) if allow_batch_fallback is not None else _bool_env("ELEVENLABS_STT_ALLOW_BATCH_FALLBACK", True) ) self._realtime_chunk_duration_ms = max(int(realtime_chunk_duration_ms), 40) self._batch_min_audio_ms = max(int(os.getenv("ELEVENLABS_STT_BATCH_MIN_AUDIO_MS", "800")), 200) self._client_session = None self._client_session_lock = asyncio.Lock() LOGGER.info( "ElevenLabs STT config: batch_model=%s realtime_model=%s input_sample_rate=%s " "target_sample_rate=%s use_realtime=%s allow_batch_fallback=%s language=%s " "realtime_chunk_ms=%s batch_min_audio_ms=%s", self._model_id, self._realtime_model_id, self._input_sample_rate_hz, self._target_sample_rate_hz, self._use_realtime, self._allow_batch_fallback, self._language_code, self._realtime_chunk_duration_ms, self._batch_min_audio_ms, ) async def transcribe(self, audio_bytes: bytes) -> str: if not audio_bytes: return "" if not self._api_key: raise RuntimeError("ELEVENLABS_API_KEY is required for ElevenLabs STT") pcm_bytes, sample_rate_hz = self._extract_pcm(audio_bytes) LOGGER.info( "ElevenLabs STT transcribe start: input_bytes=%s extracted_pcm_bytes=%s sample_rate=%s duration_ms=%s " "use_realtime=%s", len(audio_bytes), len(pcm_bytes), sample_rate_hz, _pcm_duration_ms(pcm_bytes, sample_rate_hz=sample_rate_hz), self._use_realtime, ) if sample_rate_hz != self._target_sample_rate_hz: before_rate_hz = sample_rate_hz before_bytes = len(pcm_bytes) pcm_bytes = _resample_pcm16le( pcm_bytes, input_rate_hz=sample_rate_hz, output_rate_hz=self._target_sample_rate_hz, ) sample_rate_hz = self._target_sample_rate_hz LOGGER.info( "ElevenLabs STT resampled input: from_rate=%s to_rate=%s before_bytes=%s after_bytes=%s", before_rate_hz, sample_rate_hz, before_bytes, len(pcm_bytes), ) if self._use_realtime: try: transcript = await self._transcribe_realtime( pcm_bytes=pcm_bytes, sample_rate_hz=sample_rate_hz, ) LOGGER.info( "ElevenLabs STT realtime transcript result: chars=%s transcript=%r", len(transcript), _preview_text(transcript), ) return transcript except Exception: if not self._allow_batch_fallback: raise LOGGER.exception("ElevenLabs STT realtime failed; falling back to batch STT") transcript = await self._transcribe_batch( pcm_bytes=pcm_bytes, sample_rate_hz=sample_rate_hz, ) LOGGER.info( "ElevenLabs STT batch transcript result: chars=%s transcript=%r", len(transcript), _preview_text(transcript), ) return transcript async def start_stream( self, *, partial_callback: PartialTranscriptCallback | None = None, ) -> BaseSTTStream | None: if not self._use_realtime: return None if not self._api_key: raise RuntimeError("ELEVENLABS_API_KEY is required for ElevenLabs STT") websocket_url = self._build_realtime_websocket_url(sample_rate_hz=self._target_sample_rate_hz) LOGGER.info( "ElevenLabs STT live stream starting: realtime_model=%s sample_rate=%s language=%s", self._realtime_model_id, self._target_sample_rate_hz, self._language_code, ) stream = ElevenLabsRealtimeSTTStream( api_key=self._api_key, websocket_url=websocket_url, sample_rate_hz=self._target_sample_rate_hz, timeout_seconds=self._timeout_seconds, partial_callback=partial_callback, ) await stream.connect() LOGGER.info("ElevenLabs STT live stream started: sample_rate=%s", self._target_sample_rate_hz) return stream async def close(self) -> None: session = self._client_session self._client_session = None if session is not None and not session.closed: await session.close() async def _transcribe_realtime( self, *, pcm_bytes: bytes, sample_rate_hz: int, ) -> str: try: from websockets.exceptions import WebSocketException from websockets.legacy.client import connect as websocket_connect except Exception as exc: # noqa: BLE001 raise RuntimeError("The `websockets` package is required for ElevenLabs realtime STT") from exc websocket_url = self._build_realtime_websocket_url(sample_rate_hz=sample_rate_hz) chunk_bytes = max(int(sample_rate_hz * self._realtime_chunk_duration_ms / 1000.0) * 2, 320) started_monotonic = time.perf_counter() sent_chunks = 0 sent_bytes = 0 LOGGER.info( "ElevenLabs STT realtime batch-style stream start: model=%s sample_rate=%s pcm_bytes=%s " "duration_ms=%s chunk_bytes=%s", self._realtime_model_id, sample_rate_hz, len(pcm_bytes), _pcm_duration_ms(pcm_bytes, sample_rate_hz=sample_rate_hz), chunk_bytes, ) try: async with websocket_connect( websocket_url, extra_headers=[("xi-api-key", self._api_key)], open_timeout=min(self._timeout_seconds, 10.0), close_timeout=1.0, ping_interval=20.0, ping_timeout=20.0, max_size=None, ) as websocket: await self._await_realtime_message( websocket, allowed_message_types={"session_started"}, ) for offset in range(0, len(pcm_bytes), chunk_bytes): audio_chunk = pcm_bytes[offset : offset + chunk_bytes] if not audio_chunk: continue payload = { "message_type": "input_audio_chunk", "audio_base_64": base64.b64encode(audio_chunk).decode("ascii"), "sample_rate": sample_rate_hz, "commit": offset + chunk_bytes >= len(pcm_bytes), } await websocket.send(json.dumps(payload)) sent_chunks += 1 sent_bytes += len(audio_chunk) if sent_chunks == 1 or payload["commit"] or sent_chunks % 10 == 0: LOGGER.info( "ElevenLabs STT realtime batch-style audio sent: chunks=%s bytes=%s commit=%s", sent_chunks, sent_bytes, payload["commit"], ) transcript = "" partial_transcript = "" while True: message = await self._receive_realtime_payload(websocket) message_type = str(message.get("message_type") or "").strip().lower() if message_type in {"committed_transcript", "committed_transcript_with_timestamps"}: transcript = str(message.get("text") or "").strip() if transcript: LOGGER.info( "ElevenLabs STT realtime batch-style committed: latency_ms=%s chars=%s text=%r", int((time.perf_counter() - started_monotonic) * 1000.0), len(transcript), _preview_text(transcript), ) return transcript continue if message_type == "partial_transcript": partial_transcript = str(message.get("text") or "").strip() LOGGER.info( "ElevenLabs STT realtime batch-style partial: chars=%s text=%r", len(partial_transcript), _preview_text(partial_transcript), ) continue if message_type == "session_started": continue if self._is_error_message(message_type): raise RuntimeError(self._format_realtime_error(message)) if not message_type: continue if partial_transcript: LOGGER.info( "ElevenLabs STT realtime batch-style returning partial: latency_ms=%s chars=%s", int((time.perf_counter() - started_monotonic) * 1000.0), len(partial_transcript), ) return partial_transcript except asyncio.TimeoutError as exc: raise RuntimeError("ElevenLabs realtime STT request timed out") from exc except WebSocketException as exc: raise RuntimeError("ElevenLabs realtime STT stream failed") from exc except OSError as exc: raise RuntimeError("ElevenLabs realtime STT connection failed") from exc return "" async def _transcribe_batch( self, *, pcm_bytes: bytes, sample_rate_hz: int, ) -> str: # TODO: Architectural Bottleneck: Рассмотреть замену STT на Deepgram WebSocket API для достижения true-streaming latency. try: import aiohttp except Exception as exc: # noqa: BLE001 raise RuntimeError("The `aiohttp` package is required for ElevenLabs STT") from exc pcm_bytes = _pad_pcm16le_to_duration( pcm_bytes, sample_rate_hz=sample_rate_hz, min_duration_ms=self._batch_min_audio_ms, ) wav_bytes = _pcm16le_to_wav_bytes(pcm_bytes, sample_rate_hz=sample_rate_hz) started_monotonic = time.perf_counter() LOGGER.info( "ElevenLabs batch STT request start: model=%s sample_rate=%s pcm_bytes=%s wav_bytes=%s " "duration_ms=%s language=%s", self._model_id, sample_rate_hz, len(pcm_bytes), len(wav_bytes), _pcm_duration_ms(pcm_bytes, sample_rate_hz=sample_rate_hz), self._language_code, ) form = aiohttp.FormData() form.add_field("model_id", self._model_id) form.add_field("timestamps_granularity", "none") form.add_field("diarize", "false") if self._language_code: form.add_field("language_code", self._language_code) form.add_field( "file", wav_bytes, filename="turn.wav", content_type="audio/wav", ) session = await self._get_client_session() try: async with session.post( f"{self._api_base}/v1/speech-to-text", headers={"xi-api-key": self._api_key}, data=form, ) as response: payload_text = await response.text() LOGGER.info( "ElevenLabs batch STT response: status=%s latency_ms=%s response_bytes=%s", response.status, int((time.perf_counter() - started_monotonic) * 1000.0), len(payload_text), ) if response.status >= 400: if response.status == 400 and "audio_too_short" in payload_text: LOGGER.warning("ElevenLabs batch STT reported audio_too_short; ignoring utterance") return "" raise RuntimeError( f"ElevenLabs STT returned HTTP {response.status}: {payload_text[:300]}" ) except (TimeoutError, asyncio.TimeoutError) as exc: raise RuntimeError("ElevenLabs STT request timed out") from exc except aiohttp.ClientError as exc: raise RuntimeError("ElevenLabs STT request failed") from exc try: payload = json.loads(payload_text) except json.JSONDecodeError as exc: raise RuntimeError("ElevenLabs STT returned invalid JSON") from exc return str(payload.get("text") or payload.get("transcript") or "").strip() async def _get_client_session(self): try: import aiohttp except Exception as exc: # noqa: BLE001 raise RuntimeError("The `aiohttp` package is required for ElevenLabs STT") from exc if self._client_session is not None and not self._client_session.closed: return self._client_session async with self._client_session_lock: if self._client_session is not None and not self._client_session.closed: return self._client_session timeout = aiohttp.ClientTimeout(total=self._timeout_seconds) connector = aiohttp.TCPConnector(limit=32, ttl_dns_cache=300) self._client_session = aiohttp.ClientSession(timeout=timeout, connector=connector) return self._client_session async def _await_realtime_message( self, websocket, *, allowed_message_types: set[str], ) -> dict[str, object]: while True: message = await self._receive_realtime_payload(websocket) message_type = str(message.get("message_type") or "").strip().lower() if message_type in allowed_message_types: return message if self._is_error_message(message_type): raise RuntimeError(self._format_realtime_error(message)) async def _receive_realtime_payload(self, websocket) -> dict[str, object]: raw_message = await asyncio.wait_for(websocket.recv(), timeout=self._timeout_seconds) if isinstance(raw_message, bytes): raw_message = raw_message.decode("utf-8") try: payload = json.loads(raw_message) except json.JSONDecodeError as exc: raise RuntimeError("ElevenLabs realtime STT returned invalid JSON") from exc if not isinstance(payload, dict): raise RuntimeError("ElevenLabs realtime STT returned malformed payload") return payload @staticmethod def _is_error_message(message_type: str) -> bool: return message_type in {"error", "auth_error", "input_error"} or message_type.endswith("_error") @staticmethod def _format_realtime_error(payload: dict[str, object]) -> str: message_type = str(payload.get("message_type") or "error") detail = str(payload.get("message") or payload.get("detail") or payload.get("error") or "").strip() if detail: return f"ElevenLabs realtime STT {message_type}: {detail}" return f"ElevenLabs realtime STT {message_type}" def _build_realtime_websocket_url(self, *, sample_rate_hz: int) -> str: query = { "model_id": self._realtime_model_id, "audio_format": f"pcm_{sample_rate_hz}", "commit_strategy": "manual", "include_timestamps": "false", } if self._language_code: query["language_code"] = self._language_code return f"{self._ws_base}/v1/speech-to-text/realtime?{urlencode(query)}" def _extract_pcm(self, audio_bytes: bytes) -> tuple[bytes, int]: try: with wave.open(io.BytesIO(audio_bytes), "rb") as wav_file: pcm_bytes = wav_file.readframes(wav_file.getnframes()) sample_width = wav_file.getsampwidth() channels = wav_file.getnchannels() sample_rate_hz = int(wav_file.getframerate() or self._input_sample_rate_hz) if sample_width != 2: return audio_bytes, self._input_sample_rate_hz if channels == 2: pcm_bytes = audioop.tomono(pcm_bytes, sample_width, 0.5, 0.5) return pcm_bytes, sample_rate_hz except (wave.Error, EOFError): return audio_bytes, self._input_sample_rate_hz class YandexSpeechKitSTT(BaseSTT): name = "yandex" def __init__( self, *, api_key: str | None = None, iam_token: str | None = None, folder_id: str | None = None, api_base: str | None = None, language: str | None = None, topic: str | None = None, input_sample_rate_hz: int = 8000, target_sample_rate_hz: int = 8000, timeout_seconds: float | None = None, ) -> None: self._api_key = str(api_key if api_key is not None else _yandex_api_key()).strip() self._iam_token = str(iam_token if iam_token is not None else _yandex_iam_token()).strip() self._folder_id = str(folder_id if folder_id is not None else _yandex_folder_id()).strip() self._api_base = str(api_base or _yandex_api_base()).strip().rstrip("/") self._language = str(language or _yandex_language()).strip() or "ru-RU" self._topic = str(topic or _yandex_topic()).strip() or "general" self._input_sample_rate_hz = max(int(input_sample_rate_hz), 1) self._target_sample_rate_hz = max(int(target_sample_rate_hz), 1) self._timeout_seconds = max(float(timeout_seconds if timeout_seconds is not None else _timeout_seconds()), 1.0) self._client_session = None self._client_session_lock = asyncio.Lock() LOGGER.info( "Yandex SpeechKit STT config: api_base=%s input_sample_rate=%s target_sample_rate=%s " "language=%s topic=%s folder_configured=%s auth=%s", self._api_base, self._input_sample_rate_hz, self._target_sample_rate_hz, self._language, self._topic, bool(self._folder_id), "iam" if self._iam_token else ("api-key" if self._api_key else "missing"), ) async def transcribe(self, audio_bytes: bytes) -> str: if not audio_bytes: return "" if not self._api_key and not self._iam_token: raise RuntimeError("YANDEX_STT_API_KEY or YANDEX_STT_IAM_TOKEN is required for Yandex STT") pcm_bytes, sample_rate_hz = self._extract_pcm(audio_bytes) LOGGER.info( "Yandex SpeechKit STT transcribe start: input_bytes=%s extracted_pcm_bytes=%s sample_rate=%s duration_ms=%s", len(audio_bytes), len(pcm_bytes), sample_rate_hz, _pcm_duration_ms(pcm_bytes, sample_rate_hz=sample_rate_hz), ) if sample_rate_hz != self._target_sample_rate_hz: before_rate_hz = sample_rate_hz before_bytes = len(pcm_bytes) pcm_bytes = _resample_pcm16le( pcm_bytes, input_rate_hz=sample_rate_hz, output_rate_hz=self._target_sample_rate_hz, ) sample_rate_hz = self._target_sample_rate_hz LOGGER.info( "Yandex SpeechKit STT resampled input: from_rate=%s to_rate=%s before_bytes=%s after_bytes=%s", before_rate_hz, sample_rate_hz, before_bytes, len(pcm_bytes), ) session = await self._get_client_session() params = { "lang": self._language, "topic": self._topic, "format": "lpcm", "sampleRateHertz": str(sample_rate_hz), } if self._folder_id: params["folderId"] = self._folder_id headers = { "Content-Type": f"audio/x-pcm;bit=16;rate={sample_rate_hz}", "Authorization": f"Bearer {self._iam_token}" if self._iam_token else f"Api-Key {self._api_key}", } started_monotonic = time.perf_counter() try: async with session.post( f"{self._api_base}/speech/v1/stt:recognize", params=params, headers=headers, data=pcm_bytes, ) as response: payload_text = await response.text() LOGGER.info( "Yandex SpeechKit STT response: status=%s latency_ms=%s response_bytes=%s", response.status, int((time.perf_counter() - started_monotonic) * 1000.0), len(payload_text), ) if response.status >= 400: raise RuntimeError( f"Yandex SpeechKit STT returned HTTP {response.status}: {payload_text[:300]}" ) except (TimeoutError, asyncio.TimeoutError) as exc: raise RuntimeError("Yandex SpeechKit STT request timed out") from exc except Exception as exc: try: import aiohttp except Exception: # noqa: BLE001 aiohttp = None if aiohttp is not None and isinstance(exc, aiohttp.ClientError): raise RuntimeError("Yandex SpeechKit STT request failed") from exc raise try: payload = json.loads(payload_text) except json.JSONDecodeError as exc: raise RuntimeError("Yandex SpeechKit STT returned invalid JSON") from exc transcript = str(payload.get("result") or payload.get("text") or "").strip() LOGGER.info( "Yandex SpeechKit STT transcript result: chars=%s transcript=%r", len(transcript), _preview_text(transcript), ) return transcript async def close(self) -> None: session = self._client_session self._client_session = None if session is not None and not session.closed: await session.close() async def _get_client_session(self): try: import aiohttp except Exception as exc: # noqa: BLE001 raise RuntimeError("The `aiohttp` package is required for Yandex STT") from exc if self._client_session is not None and not self._client_session.closed: return self._client_session async with self._client_session_lock: if self._client_session is not None and not self._client_session.closed: return self._client_session timeout = aiohttp.ClientTimeout(total=self._timeout_seconds) connector = aiohttp.TCPConnector(limit=16, ttl_dns_cache=300) self._client_session = aiohttp.ClientSession(timeout=timeout, connector=connector) return self._client_session def _extract_pcm(self, audio_bytes: bytes) -> tuple[bytes, int]: try: with wave.open(io.BytesIO(audio_bytes), "rb") as wav_file: pcm_bytes = wav_file.readframes(wav_file.getnframes()) sample_width = wav_file.getsampwidth() channels = wav_file.getnchannels() sample_rate_hz = int(wav_file.getframerate() or self._input_sample_rate_hz) if sample_width != 2: return audio_bytes, self._input_sample_rate_hz if channels == 2: pcm_bytes = audioop.tomono(pcm_bytes, sample_width, 0.5, 0.5) return pcm_bytes, sample_rate_hz except (wave.Error, EOFError): return audio_bytes, self._input_sample_rate_hz class FallbackSTT(BaseSTT): def __init__(self, *, primary: BaseSTT, fallback: BaseSTT) -> None: self._primary = primary self._fallback = fallback LOGGER.warning( "STT fallback configured: primary=%s fallback=%s", type(primary).__name__, type(fallback).__name__, ) async def transcribe(self, audio_bytes: bytes) -> str: try: return await self._primary.transcribe(audio_bytes) except Exception: LOGGER.exception( "Primary STT provider failed; falling back: primary=%s fallback=%s", type(self._primary).__name__, type(self._fallback).__name__, ) return await self._fallback.transcribe(audio_bytes) async def close(self) -> None: for provider in (self._primary, self._fallback): close = getattr(provider, "close", None) if close is None: continue result = close() if asyncio.iscoroutine(result): await result def _build_single_stt_provider( provider: str, *, input_sample_rate_hz: int, target_sample_rate_hz: int, ) -> BaseSTT: normalized = provider.lower() if normalized in {"yandex", "yandex_speechkit", "speechkit"}: return YandexSpeechKitSTT( input_sample_rate_hz=input_sample_rate_hz, target_sample_rate_hz=target_sample_rate_hz, ) if normalized in {"elevenlabs", "eleven_labs", "scribe"}: return ElevenLabsSTT( input_sample_rate_hz=input_sample_rate_hz, target_sample_rate_hz=target_sample_rate_hz, ) if normalized in {"openai", "whisper"}: return OpenAISTT( input_sample_rate_hz=input_sample_rate_hz, ) raise RuntimeError(f"Unsupported STT provider: {provider}") def build_stt_provider( *, input_sample_rate_hz: int, target_sample_rate_hz: int, ) -> BaseSTT: provider = _stt_provider_name() primary = _build_single_stt_provider( provider, input_sample_rate_hz=input_sample_rate_hz, target_sample_rate_hz=target_sample_rate_hz, ) fallback_provider = _stt_fallback_provider_name() if not fallback_provider or fallback_provider == provider: return primary fallback = _build_single_stt_provider( fallback_provider, input_sample_rate_hz=input_sample_rate_hz, target_sample_rate_hz=target_sample_rate_hz, ) return FallbackSTT(primary=primary, fallback=fallback)