import asyncio import base64 import hashlib import html import io import json import logging import mimetypes import os import uuid from concurrent.futures import ThreadPoolExecutor from fnmatch import fnmatch from typing import Optional import aiofiles import aiohttp import requests from fastapi import ( APIRouter, Depends, FastAPI, File, Form, HTTPException, Request, UploadFile, status, ) from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse from open_webui.config import ( CACHE_DIR, ELEVENLABS_API_BASE_URL, WHISPER_COMPUTE_TYPE, WHISPER_LANGUAGE, WHISPER_MODEL_AUTO_UPDATE, WHISPER_MODEL_DIR, WHISPER_MULTILINGUAL, WHISPER_VAD_FILTER, ) from open_webui.constants import ERROR_MESSAGES from open_webui.env import ( AIOHTTP_CLIENT_SESSION_SSL, AIOHTTP_CLIENT_TIMEOUT, AIOHTTP_CLIENT_TIMEOUT_MODEL_LIST, BYPASS_PYDUB_PREPROCESSING, DEVICE_TYPE, ENABLE_FORWARD_USER_INFO_HEADERS, ENV, ) from open_webui.utils.access_control import has_permission from open_webui.utils.auth import get_admin_user, get_verified_user from open_webui.utils.headers import include_user_info_headers from open_webui.utils.misc import strict_match_mime_type from pydantic import BaseModel from pydub import AudioSegment from pydub.silence import split_on_silence router = APIRouter() # Constants MAX_FILE_SIZE_MB = 20 MAX_FILE_SIZE = MAX_FILE_SIZE_MB * 1024 * 1024 # Convert MB to bytes AZURE_MAX_FILE_SIZE_MB = 200 AZURE_MAX_FILE_SIZE = AZURE_MAX_FILE_SIZE_MB * 1024 * 1024 # Convert MB to bytes log = logging.getLogger(__name__) SPEECH_CACHE_DIR = CACHE_DIR / 'audio' / 'speech' SPEECH_CACHE_DIR.mkdir(parents=True, exist_ok=True) ########################################## # # Utility functions # Let what is spoken here be heard clearly, and let # no voice be reduced to noise along the way. # ########################################## from pydub import AudioSegment from pydub.utils import mediainfo def is_audio_conversion_required(file_path): """ Check if the given audio file needs conversion to mp3. """ SUPPORTED_FORMATS = {'flac', 'm4a', 'mp3', 'mp4', 'mpeg', 'wav', 'webm'} if not os.path.isfile(file_path): log.error(f'File not found: {file_path}') return False try: info = mediainfo(file_path) codec_name = info.get('codec_name', '').lower() codec_type = info.get('codec_type', '').lower() codec_tag_string = info.get('codec_tag_string', '').lower() if codec_name == 'aac' and codec_type == 'audio' and codec_tag_string == 'mp4a': # File is AAC/mp4a audio, recommend mp3 conversion return True # If the codec name is in the supported formats if codec_name in SUPPORTED_FORMATS: return False return True except Exception as e: log.error(f'Error getting audio format: {e}') return False def convert_audio_to_mp3(file_path): """Convert audio file to mp3 format.""" try: output_path = os.path.splitext(file_path)[0] + '.mp3' audio = AudioSegment.from_file(file_path) audio.export(output_path, format='mp3') log.info(f'Converted {file_path} to {output_path}') return output_path except Exception as e: log.error(f'Error converting audio file: {e}') return None def transcode_audio_to_mp3(audio_data: bytes, content_type_header: str, output_path: str) -> bool: """ Transcode audio bytes to MP3 if the Content-Type indicates a non-MP3 format. Handles raw PCM audio (e.g. Gemini-TTS via OpenRouter/LiteLLM) by parsing optional rate/channels from the Content-Type params, defaulting to 24kHz, 16-bit, mono. For other non-MP3 formats, uses pydub auto-detection. Returns True if transcoding was performed, False if the data is already MP3. Respects BYPASS_PYDUB_PREPROCESSING — when set, writes raw bytes and logs a warning. """ mime_type = content_type_header.split(';')[0].strip().lower() if mime_type in ('audio/mpeg', 'audio/mp3'): return False if BYPASS_PYDUB_PREPROCESSING: log.warning( f'TTS returned {mime_type} but BYPASS_PYDUB_PREPROCESSING is set; writing raw audio without transcoding' ) return False if mime_type in ('audio/pcm', 'audio/l16', 'audio/raw'): # Parse optional rate/channels from Content-Type params, # default: 24kHz, 16-bit, mono (standard for Gemini TTS). ct_params = {} for part in content_type_header.split(';')[1:]: key_val = part.strip().split('=') if len(key_val) == 2: ct_params[key_val[0].strip().lower()] = key_val[1].strip() sample_rate = int(ct_params.get('rate', 24000)) channels = int(ct_params.get('channels', 1)) audio_segment = AudioSegment.from_raw( io.BytesIO(audio_data), sample_width=2, frame_rate=sample_rate, channels=channels, ) else: audio_segment = AudioSegment.from_file(io.BytesIO(audio_data)) audio_segment.export(str(output_path), format='mp3') log.info(f'Transcoded {mime_type} audio to MP3: {output_path}') return True def set_faster_whisper_model(model: str, auto_update: bool = False): whisper_model = None if model: from faster_whisper import WhisperModel faster_whisper_kwargs = { 'model_size_or_path': model, 'device': DEVICE_TYPE if DEVICE_TYPE and DEVICE_TYPE == 'cuda' else 'cpu', 'compute_type': WHISPER_COMPUTE_TYPE, 'download_root': WHISPER_MODEL_DIR, 'local_files_only': not auto_update, } try: whisper_model = WhisperModel(**faster_whisper_kwargs) except Exception: log.warning('WhisperModel initialization failed, attempting download with local_files_only=False') faster_whisper_kwargs['local_files_only'] = False whisper_model = WhisperModel(**faster_whisper_kwargs) return whisper_model ########################################## # # Audio API # ########################################## class TTSConfigForm(BaseModel): OPENAI_API_BASE_URL: str OPENAI_API_KEY: str OPENAI_PARAMS: Optional[dict] = None API_KEY: str ENGINE: str MODEL: str VOICE: str SPLIT_ON: str AZURE_SPEECH_REGION: str AZURE_SPEECH_BASE_URL: str AZURE_SPEECH_OUTPUT_FORMAT: str MISTRAL_API_KEY: str MISTRAL_API_BASE_URL: str class STTConfigForm(BaseModel): OPENAI_API_BASE_URL: str OPENAI_API_KEY: str ENGINE: str MODEL: str SUPPORTED_CONTENT_TYPES: list[str] = [] ALLOWED_EXTENSIONS: list[str] = [] WHISPER_MODEL: str DEEPGRAM_API_KEY: str AZURE_API_KEY: str AZURE_REGION: str AZURE_LOCALES: str AZURE_BASE_URL: str AZURE_MAX_SPEAKERS: str MISTRAL_API_KEY: str MISTRAL_API_BASE_URL: str MISTRAL_USE_CHAT_COMPLETIONS: bool class AudioConfigUpdateForm(BaseModel): tts: TTSConfigForm stt: STTConfigForm @router.get('/config') async def get_audio_config(request: Request, user=Depends(get_admin_user)): return { 'tts': { 'OPENAI_API_BASE_URL': request.app.state.config.TTS_OPENAI_API_BASE_URL, 'OPENAI_API_KEY': request.app.state.config.TTS_OPENAI_API_KEY, 'OPENAI_PARAMS': request.app.state.config.TTS_OPENAI_PARAMS, 'API_KEY': request.app.state.config.TTS_API_KEY, 'ENGINE': request.app.state.config.TTS_ENGINE, 'MODEL': request.app.state.config.TTS_MODEL, 'VOICE': request.app.state.config.TTS_VOICE, 'SPLIT_ON': request.app.state.config.TTS_SPLIT_ON, 'AZURE_SPEECH_REGION': request.app.state.config.TTS_AZURE_SPEECH_REGION, 'AZURE_SPEECH_BASE_URL': request.app.state.config.TTS_AZURE_SPEECH_BASE_URL, 'AZURE_SPEECH_OUTPUT_FORMAT': request.app.state.config.TTS_AZURE_SPEECH_OUTPUT_FORMAT, 'MISTRAL_API_KEY': request.app.state.config.TTS_MISTRAL_API_KEY, 'MISTRAL_API_BASE_URL': request.app.state.config.TTS_MISTRAL_API_BASE_URL, }, 'stt': { 'OPENAI_API_BASE_URL': request.app.state.config.STT_OPENAI_API_BASE_URL, 'OPENAI_API_KEY': request.app.state.config.STT_OPENAI_API_KEY, 'ENGINE': request.app.state.config.STT_ENGINE, 'MODEL': request.app.state.config.STT_MODEL, 'SUPPORTED_CONTENT_TYPES': request.app.state.config.STT_SUPPORTED_CONTENT_TYPES, 'ALLOWED_EXTENSIONS': request.app.state.config.STT_ALLOWED_EXTENSIONS, 'WHISPER_MODEL': request.app.state.config.WHISPER_MODEL, 'DEEPGRAM_API_KEY': request.app.state.config.DEEPGRAM_API_KEY, 'AZURE_API_KEY': request.app.state.config.AUDIO_STT_AZURE_API_KEY, 'AZURE_REGION': request.app.state.config.AUDIO_STT_AZURE_REGION, 'AZURE_LOCALES': request.app.state.config.AUDIO_STT_AZURE_LOCALES, 'AZURE_BASE_URL': request.app.state.config.AUDIO_STT_AZURE_BASE_URL, 'AZURE_MAX_SPEAKERS': request.app.state.config.AUDIO_STT_AZURE_MAX_SPEAKERS, 'MISTRAL_API_KEY': request.app.state.config.AUDIO_STT_MISTRAL_API_KEY, 'MISTRAL_API_BASE_URL': request.app.state.config.AUDIO_STT_MISTRAL_API_BASE_URL, 'MISTRAL_USE_CHAT_COMPLETIONS': request.app.state.config.AUDIO_STT_MISTRAL_USE_CHAT_COMPLETIONS, }, } @router.post('/config/update') async def update_audio_config(request: Request, form_data: AudioConfigUpdateForm, user=Depends(get_admin_user)): request.app.state.config.TTS_OPENAI_API_BASE_URL = form_data.tts.OPENAI_API_BASE_URL request.app.state.config.TTS_OPENAI_API_KEY = form_data.tts.OPENAI_API_KEY request.app.state.config.TTS_OPENAI_PARAMS = form_data.tts.OPENAI_PARAMS request.app.state.config.TTS_API_KEY = form_data.tts.API_KEY request.app.state.config.TTS_ENGINE = form_data.tts.ENGINE request.app.state.config.TTS_MODEL = form_data.tts.MODEL request.app.state.config.TTS_VOICE = form_data.tts.VOICE request.app.state.config.TTS_SPLIT_ON = form_data.tts.SPLIT_ON request.app.state.config.TTS_AZURE_SPEECH_REGION = form_data.tts.AZURE_SPEECH_REGION request.app.state.config.TTS_AZURE_SPEECH_BASE_URL = form_data.tts.AZURE_SPEECH_BASE_URL request.app.state.config.TTS_AZURE_SPEECH_OUTPUT_FORMAT = form_data.tts.AZURE_SPEECH_OUTPUT_FORMAT request.app.state.config.TTS_MISTRAL_API_KEY = form_data.tts.MISTRAL_API_KEY request.app.state.config.TTS_MISTRAL_API_BASE_URL = form_data.tts.MISTRAL_API_BASE_URL request.app.state.config.STT_OPENAI_API_BASE_URL = form_data.stt.OPENAI_API_BASE_URL request.app.state.config.STT_OPENAI_API_KEY = form_data.stt.OPENAI_API_KEY request.app.state.config.STT_ENGINE = form_data.stt.ENGINE request.app.state.config.STT_MODEL = form_data.stt.MODEL request.app.state.config.STT_SUPPORTED_CONTENT_TYPES = form_data.stt.SUPPORTED_CONTENT_TYPES request.app.state.config.STT_ALLOWED_EXTENSIONS = form_data.stt.ALLOWED_EXTENSIONS request.app.state.config.WHISPER_MODEL = form_data.stt.WHISPER_MODEL request.app.state.config.DEEPGRAM_API_KEY = form_data.stt.DEEPGRAM_API_KEY request.app.state.config.AUDIO_STT_AZURE_API_KEY = form_data.stt.AZURE_API_KEY request.app.state.config.AUDIO_STT_AZURE_REGION = form_data.stt.AZURE_REGION request.app.state.config.AUDIO_STT_AZURE_LOCALES = form_data.stt.AZURE_LOCALES request.app.state.config.AUDIO_STT_AZURE_BASE_URL = form_data.stt.AZURE_BASE_URL request.app.state.config.AUDIO_STT_AZURE_MAX_SPEAKERS = form_data.stt.AZURE_MAX_SPEAKERS request.app.state.config.AUDIO_STT_MISTRAL_API_KEY = form_data.stt.MISTRAL_API_KEY request.app.state.config.AUDIO_STT_MISTRAL_API_BASE_URL = form_data.stt.MISTRAL_API_BASE_URL request.app.state.config.AUDIO_STT_MISTRAL_USE_CHAT_COMPLETIONS = form_data.stt.MISTRAL_USE_CHAT_COMPLETIONS if request.app.state.config.STT_ENGINE == '': request.app.state.faster_whisper_model = set_faster_whisper_model( form_data.stt.WHISPER_MODEL, WHISPER_MODEL_AUTO_UPDATE ) else: request.app.state.faster_whisper_model = None return { 'tts': { 'ENGINE': request.app.state.config.TTS_ENGINE, 'MODEL': request.app.state.config.TTS_MODEL, 'VOICE': request.app.state.config.TTS_VOICE, 'OPENAI_API_BASE_URL': request.app.state.config.TTS_OPENAI_API_BASE_URL, 'OPENAI_API_KEY': request.app.state.config.TTS_OPENAI_API_KEY, 'OPENAI_PARAMS': request.app.state.config.TTS_OPENAI_PARAMS, 'API_KEY': request.app.state.config.TTS_API_KEY, 'SPLIT_ON': request.app.state.config.TTS_SPLIT_ON, 'AZURE_SPEECH_REGION': request.app.state.config.TTS_AZURE_SPEECH_REGION, 'AZURE_SPEECH_BASE_URL': request.app.state.config.TTS_AZURE_SPEECH_BASE_URL, 'AZURE_SPEECH_OUTPUT_FORMAT': request.app.state.config.TTS_AZURE_SPEECH_OUTPUT_FORMAT, 'MISTRAL_API_KEY': request.app.state.config.TTS_MISTRAL_API_KEY, 'MISTRAL_API_BASE_URL': request.app.state.config.TTS_MISTRAL_API_BASE_URL, }, 'stt': { 'OPENAI_API_BASE_URL': request.app.state.config.STT_OPENAI_API_BASE_URL, 'OPENAI_API_KEY': request.app.state.config.STT_OPENAI_API_KEY, 'ENGINE': request.app.state.config.STT_ENGINE, 'MODEL': request.app.state.config.STT_MODEL, 'SUPPORTED_CONTENT_TYPES': request.app.state.config.STT_SUPPORTED_CONTENT_TYPES, 'ALLOWED_EXTENSIONS': request.app.state.config.STT_ALLOWED_EXTENSIONS, 'WHISPER_MODEL': request.app.state.config.WHISPER_MODEL, 'DEEPGRAM_API_KEY': request.app.state.config.DEEPGRAM_API_KEY, 'AZURE_API_KEY': request.app.state.config.AUDIO_STT_AZURE_API_KEY, 'AZURE_REGION': request.app.state.config.AUDIO_STT_AZURE_REGION, 'AZURE_LOCALES': request.app.state.config.AUDIO_STT_AZURE_LOCALES, 'AZURE_BASE_URL': request.app.state.config.AUDIO_STT_AZURE_BASE_URL, 'AZURE_MAX_SPEAKERS': request.app.state.config.AUDIO_STT_AZURE_MAX_SPEAKERS, 'MISTRAL_API_KEY': request.app.state.config.AUDIO_STT_MISTRAL_API_KEY, 'MISTRAL_API_BASE_URL': request.app.state.config.AUDIO_STT_MISTRAL_API_BASE_URL, 'MISTRAL_USE_CHAT_COMPLETIONS': request.app.state.config.AUDIO_STT_MISTRAL_USE_CHAT_COMPLETIONS, }, } def load_speech_pipeline(request): from datasets import load_dataset from transformers import pipeline if request.app.state.speech_synthesiser is None: request.app.state.speech_synthesiser = pipeline('text-to-speech', 'microsoft/speecht5_tts') if request.app.state.speech_speaker_embeddings_dataset is None: request.app.state.speech_speaker_embeddings_dataset = load_dataset( 'Matthijs/cmu-arctic-xvectors', split='validation' ) @router.post('/speech') async def speech(request: Request, user=Depends(get_verified_user)): if request.app.state.config.TTS_ENGINE == '': raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND, ) if user.role != 'admin' and not await has_permission( user.id, 'chat.tts', request.app.state.config.USER_PERMISSIONS ): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.ACCESS_PROHIBITED, ) body = await request.body() name = hashlib.sha256( body + str(request.app.state.config.TTS_ENGINE).encode('utf-8') + str(request.app.state.config.TTS_MODEL).encode('utf-8') ).hexdigest() file_path = SPEECH_CACHE_DIR.joinpath(f'{name}.mp3') file_body_path = SPEECH_CACHE_DIR.joinpath(f'{name}.json') # Check if the file already exists in the cache if file_path.is_file(): return FileResponse(file_path) payload = None try: payload = json.loads(body.decode('utf-8')) except Exception as e: log.exception(e) raise HTTPException(status_code=400, detail='Invalid JSON payload') r = None if request.app.state.config.TTS_ENGINE == 'openai': payload['model'] = request.app.state.config.TTS_MODEL try: timeout = aiohttp.ClientTimeout(total=AIOHTTP_CLIENT_TIMEOUT) async with aiohttp.ClientSession(timeout=timeout, trust_env=True) as session: payload = { **payload, **(request.app.state.config.TTS_OPENAI_PARAMS or {}), } headers = { 'Content-Type': 'application/json', 'Authorization': f'Bearer {request.app.state.config.TTS_OPENAI_API_KEY}', } if ENABLE_FORWARD_USER_INFO_HEADERS: headers = include_user_info_headers(headers, user) r = await session.post( url=f'{request.app.state.config.TTS_OPENAI_API_BASE_URL}/audio/speech', json=payload, headers=headers, ssl=AIOHTTP_CLIENT_SESSION_SSL, ) r.raise_for_status() audio_data = await r.read() content_type_header = r.headers.get('Content-Type', 'audio/mpeg') if not transcode_audio_to_mp3(audio_data, content_type_header, file_path): async with aiofiles.open(file_path, 'wb') as f: await f.write(audio_data) async with aiofiles.open(file_body_path, 'w') as f: await f.write(json.dumps(payload)) return FileResponse(file_path) except Exception as e: log.exception(e) detail = None status_code = 500 detail = f'Open WebUI: Server Connection Error' if r is not None: status_code = r.status try: res = await r.json() if 'error' in res: detail = f'External: {res["error"]}' except Exception: detail = f'External: {e}' raise HTTPException( status_code=status_code, detail=detail, ) elif request.app.state.config.TTS_ENGINE == 'elevenlabs': voice_id = payload.get('voice', '') if voice_id not in await get_available_voices(request): raise HTTPException( status_code=400, detail='Invalid voice id', ) try: timeout = aiohttp.ClientTimeout(total=AIOHTTP_CLIENT_TIMEOUT) async with aiohttp.ClientSession(timeout=timeout, trust_env=True) as session: async with session.post( f'{ELEVENLABS_API_BASE_URL}/v1/text-to-speech/{voice_id}', json={ 'text': payload['input'], 'model_id': request.app.state.config.TTS_MODEL, 'voice_settings': {'stability': 0.5, 'similarity_boost': 0.5}, }, headers={ 'Accept': 'audio/mpeg', 'Content-Type': 'application/json', 'xi-api-key': request.app.state.config.TTS_API_KEY, }, ssl=AIOHTTP_CLIENT_SESSION_SSL, ) as r: r.raise_for_status() async with aiofiles.open(file_path, 'wb') as f: await f.write(await r.read()) async with aiofiles.open(file_body_path, 'w') as f: await f.write(json.dumps(payload)) return FileResponse(file_path) except Exception as e: log.exception(e) detail = None try: if r.status != 200: res = await r.json() if 'error' in res: detail = f'External: {res["error"].get("message", "")}' except Exception: detail = f'External: {e}' raise HTTPException( status_code=getattr(r, 'status', 500) if r else 500, detail=detail if detail else 'Open WebUI: Server Connection Error', ) elif request.app.state.config.TTS_ENGINE == 'azure': try: payload = json.loads(body.decode('utf-8')) except Exception as e: log.exception(e) raise HTTPException(status_code=400, detail='Invalid JSON payload') region = request.app.state.config.TTS_AZURE_SPEECH_REGION or 'eastus' base_url = request.app.state.config.TTS_AZURE_SPEECH_BASE_URL language = request.app.state.config.TTS_VOICE locale = '-'.join(request.app.state.config.TTS_VOICE.split('-')[:2]) output_format = request.app.state.config.TTS_AZURE_SPEECH_OUTPUT_FORMAT try: data = f""" {html.escape(payload['input'])} """ timeout = aiohttp.ClientTimeout(total=AIOHTTP_CLIENT_TIMEOUT) async with aiohttp.ClientSession(timeout=timeout, trust_env=True) as session: async with session.post( (base_url or f'https://{region}.tts.speech.microsoft.com') + '/cognitiveservices/v1', headers={ 'Ocp-Apim-Subscription-Key': request.app.state.config.TTS_API_KEY, 'Content-Type': 'application/ssml+xml', 'X-Microsoft-OutputFormat': output_format, }, data=data, ssl=AIOHTTP_CLIENT_SESSION_SSL, ) as r: r.raise_for_status() async with aiofiles.open(file_path, 'wb') as f: await f.write(await r.read()) async with aiofiles.open(file_body_path, 'w') as f: await f.write(json.dumps(payload)) return FileResponse(file_path) except Exception as e: log.exception(e) detail = None try: if r.status != 200: res = await r.json() if 'error' in res: detail = f'External: {res["error"].get("message", "")}' except Exception: detail = f'External: {e}' raise HTTPException( status_code=getattr(r, 'status', 500) if r else 500, detail=detail if detail else 'Open WebUI: Server Connection Error', ) elif request.app.state.config.TTS_ENGINE == 'transformers': payload = None try: payload = json.loads(body.decode('utf-8')) except Exception as e: log.exception(e) raise HTTPException(status_code=400, detail='Invalid JSON payload') import soundfile as sf import torch load_speech_pipeline(request) embeddings_dataset = request.app.state.speech_speaker_embeddings_dataset speaker_index = 6799 try: speaker_index = embeddings_dataset['filename'].index(request.app.state.config.TTS_MODEL) except Exception: pass speaker_embedding = torch.tensor(embeddings_dataset[speaker_index]['xvector']).unsqueeze(0) speech = request.app.state.speech_synthesiser( payload['input'], forward_params={'speaker_embeddings': speaker_embedding}, ) sf.write(file_path, speech['audio'], samplerate=speech['sampling_rate']) async with aiofiles.open(file_body_path, 'w') as f: await f.write(json.dumps(payload)) return FileResponse(file_path) elif request.app.state.config.TTS_ENGINE == 'mistral': api_key = request.app.state.config.TTS_MISTRAL_API_KEY api_base_url = request.app.state.config.TTS_MISTRAL_API_BASE_URL or 'https://api.mistral.ai/v1' if not api_key: raise HTTPException( status_code=400, detail='Mistral API key is required for Mistral TTS', ) try: timeout = aiohttp.ClientTimeout(total=AIOHTTP_CLIENT_TIMEOUT) async with aiohttp.ClientSession(timeout=timeout, trust_env=True) as session: mistral_payload = { 'input': payload.get('input', ''), 'model': request.app.state.config.TTS_MODEL or 'voxtral-mini-tts-2603', 'voice_id': payload.get('voice', ''), 'response_format': 'mp3', } r = await session.post( url=f'{api_base_url}/audio/speech', json=mistral_payload, headers={ 'Content-Type': 'application/json', 'Authorization': f'Bearer {api_key}', }, ssl=AIOHTTP_CLIENT_SESSION_SSL, ) r.raise_for_status() res = await r.json() audio_data = res.get('audio_data', '') if not audio_data: raise ValueError('No audio_data in Mistral TTS response') audio_bytes = base64.b64decode(audio_data) async with aiofiles.open(file_path, 'wb') as f: await f.write(audio_bytes) async with aiofiles.open(file_body_path, 'w') as f: await f.write(json.dumps(payload)) return FileResponse(file_path) except Exception as e: log.exception(e) detail = None status_code = 500 detail = 'Open WebUI: Server Connection Error' if r is not None: status_code = r.status try: res = await r.json() if 'error' in res: detail = f'External: {res["error"]}' elif 'message' in res: detail = f'External: {res["message"]}' except Exception: detail = f'External: {e}' raise HTTPException( status_code=status_code, detail=detail, ) def transcription_handler(request, file_path, metadata, user=None): filename = os.path.basename(file_path) file_dir = os.path.dirname(file_path) id = filename.split('.')[0] metadata = metadata or {} languages = [ metadata.get('language', None) if not WHISPER_LANGUAGE else WHISPER_LANGUAGE, None, # Always fallback to None in case transcription fails ] if request.app.state.config.STT_ENGINE == '': if request.app.state.faster_whisper_model is None: request.app.state.faster_whisper_model = set_faster_whisper_model(request.app.state.config.WHISPER_MODEL) model = request.app.state.faster_whisper_model segments, info = model.transcribe( file_path, beam_size=5, vad_filter=WHISPER_VAD_FILTER, language=languages[0], multilingual=WHISPER_MULTILINGUAL, ) log.info("Detected language '%s' with probability %f" % (info.language, info.language_probability)) transcript = ''.join([segment.text for segment in list(segments)]) data = {'text': transcript.strip()} # save the transcript to a json file transcript_file = os.path.join(file_dir, f'{id}.json') with open(transcript_file, 'w') as f: json.dump(data, f) log.debug(data) return data elif request.app.state.config.STT_ENGINE == 'openai': r = None try: for language in languages: payload = { 'model': request.app.state.config.STT_MODEL, } if language: payload['language'] = language headers = {'Authorization': f'Bearer {request.app.state.config.STT_OPENAI_API_KEY}'} if user and ENABLE_FORWARD_USER_INFO_HEADERS: headers = include_user_info_headers(headers, user) with open(file_path, 'rb') as audio_file: r = requests.post( url=f'{request.app.state.config.STT_OPENAI_API_BASE_URL}/audio/transcriptions', headers=headers, files={'file': (filename, audio_file)}, data=payload, timeout=AIOHTTP_CLIENT_TIMEOUT, ) if r.status_code == 200: # Successful transcription break r.raise_for_status() data = r.json() # save the transcript to a json file transcript_file = os.path.join(file_dir, f'{id}.json') with open(transcript_file, 'w') as f: json.dump(data, f) return data except Exception as e: log.exception(e) detail = None if r is not None: try: res = r.json() if 'error' in res: detail = f'External: {res["error"].get("message", "")}' except Exception: detail = f'External: {e}' raise Exception(detail if detail else 'Open WebUI: Server Connection Error') elif request.app.state.config.STT_ENGINE == 'deepgram': try: # Determine the MIME type of the file mime, _ = mimetypes.guess_type(file_path) if not mime: mime = 'audio/wav' # fallback to wav if undetectable # Read the audio file with open(file_path, 'rb') as f: file_data = f.read() # Build headers and parameters headers = { 'Authorization': f'Token {request.app.state.config.DEEPGRAM_API_KEY}', 'Content-Type': mime, } for language in languages: params = {} if request.app.state.config.STT_MODEL: params['model'] = request.app.state.config.STT_MODEL if language: params['language'] = language # Make request to Deepgram API r = requests.post( 'https://api.deepgram.com/v1/listen?smart_format=true', headers=headers, params=params, data=file_data, timeout=AIOHTTP_CLIENT_TIMEOUT, ) if r.status_code == 200: # Successful transcription break r.raise_for_status() response_data = r.json() # Extract transcript from Deepgram response try: transcript = response_data['results']['channels'][0]['alternatives'][0].get('transcript', '') except (KeyError, IndexError) as e: log.error(f'Malformed response from Deepgram: {str(e)}') raise Exception('Failed to parse Deepgram response - unexpected response format') data = {'text': transcript.strip()} # Save transcript transcript_file = os.path.join(file_dir, f'{id}.json') with open(transcript_file, 'w') as f: json.dump(data, f) return data except Exception as e: log.exception(e) detail = None if r is not None: try: res = r.json() if 'error' in res: detail = f'External: {res["error"].get("message", "")}' except Exception: detail = f'External: {e}' raise Exception(detail if detail else 'Open WebUI: Server Connection Error') elif request.app.state.config.STT_ENGINE == 'azure': # Check file exists and size if not os.path.exists(file_path): raise HTTPException(status_code=400, detail='Audio file not found') # Check file size (Azure has a larger limit of 200MB) file_size = os.path.getsize(file_path) if file_size > AZURE_MAX_FILE_SIZE: raise HTTPException( status_code=400, detail=f"File size exceeds Azure's limit of {AZURE_MAX_FILE_SIZE_MB}MB", ) api_key = request.app.state.config.AUDIO_STT_AZURE_API_KEY region = request.app.state.config.AUDIO_STT_AZURE_REGION or 'eastus' locales = request.app.state.config.AUDIO_STT_AZURE_LOCALES base_url = request.app.state.config.AUDIO_STT_AZURE_BASE_URL max_speakers = request.app.state.config.AUDIO_STT_AZURE_MAX_SPEAKERS or 3 # IF NO LOCALES, USE DEFAULTS if len(locales) < 2: locales = [ 'en-US', 'es-ES', 'es-MX', 'fr-FR', 'hi-IN', 'it-IT', 'de-DE', 'en-GB', 'en-IN', 'ja-JP', 'ko-KR', 'pt-BR', 'zh-CN', ] locales = ','.join(locales) if not api_key or not region: raise HTTPException( status_code=400, detail='Azure API key is required for Azure STT', ) r = None try: # Prepare the request data = { 'definition': json.dumps( { 'locales': locales.split(','), 'diarization': {'maxSpeakers': max_speakers, 'enabled': True}, } if locales else {} ) } url = ( base_url or f'https://{region}.api.cognitive.microsoft.com' ) + '/speechtotext/transcriptions:transcribe?api-version=2024-11-15' # Use context manager to ensure file is properly closed with open(file_path, 'rb') as audio_file: r = requests.post( url=url, files={'audio': audio_file}, data=data, headers={ 'Ocp-Apim-Subscription-Key': api_key, }, timeout=AIOHTTP_CLIENT_TIMEOUT, ) r.raise_for_status() response = r.json() # Extract transcript from response if not response.get('combinedPhrases'): raise ValueError('No transcription found in response') # Get the full transcript from combinedPhrases transcript = response['combinedPhrases'][0].get('text', '').strip() if not transcript: raise ValueError('Empty transcript in response') data = {'text': transcript} # Save transcript to json file (consistent with other providers) transcript_file = os.path.join(file_dir, f'{id}.json') with open(transcript_file, 'w') as f: json.dump(data, f) log.debug(data) return data except (KeyError, IndexError, ValueError) as e: log.exception('Error parsing Azure response') raise HTTPException( status_code=500, detail=f'Failed to parse Azure response: {str(e)}', ) except requests.exceptions.RequestException as e: log.exception(e) detail = None status_code = getattr(r, 'status_code', 500) if r else 500 try: if r is not None and r.status_code != 200: res = r.json() # Azure returns {"code": "...", "message": "...", "innerError": {...}} if 'code' in res and 'message' in res: azure_code = res.get('innerError', {}).get('code', res['code']) user_facing_codes = { 'EmptyAudioFile', 'AudioLengthLimitExceeded', 'NoLanguageIdentified', 'MultipleLanguagesIdentified', } if azure_code in user_facing_codes: detail = res['message'] else: log.error(f'Azure STT error [{azure_code}]: {res["message"]}') detail = 'An error occurred during transcription.' elif 'error' in res: detail = f'External: {res["error"].get("message", "")}' except Exception: detail = f'External: {e}' raise HTTPException( status_code=status_code, detail=detail if detail else 'Open WebUI: Server Connection Error', ) elif request.app.state.config.STT_ENGINE == 'mistral': # Check file exists if not os.path.exists(file_path): raise HTTPException(status_code=400, detail='Audio file not found') # Check file size file_size = os.path.getsize(file_path) if file_size > MAX_FILE_SIZE: raise HTTPException( status_code=400, detail=f'File size exceeds limit of {MAX_FILE_SIZE_MB}MB', ) api_key = request.app.state.config.AUDIO_STT_MISTRAL_API_KEY api_base_url = request.app.state.config.AUDIO_STT_MISTRAL_API_BASE_URL or 'https://api.mistral.ai/v1' use_chat_completions = request.app.state.config.AUDIO_STT_MISTRAL_USE_CHAT_COMPLETIONS if not api_key: raise HTTPException( status_code=400, detail='Mistral API key is required for Mistral STT', ) r = None try: # Use voxtral-mini-latest as the default model for transcription model = request.app.state.config.STT_MODEL or 'voxtral-mini-latest' log.info( f'Mistral STT - model: {model}, ' f'method: {"chat_completions" if use_chat_completions else "transcriptions"}' ) if use_chat_completions: # Use chat completions API with audio input # This method requires mp3 or wav format audio_file_to_use = file_path if is_audio_conversion_required(file_path): log.debug('Converting audio to mp3 for chat completions API') converted_path = convert_audio_to_mp3(file_path) if converted_path: audio_file_to_use = converted_path else: log.error('Audio conversion failed') raise HTTPException( status_code=500, detail='Audio conversion failed. Chat completions API requires mp3 or wav format.', ) # Read and encode audio file as base64 with open(audio_file_to_use, 'rb') as audio_file: audio_base64 = { 'data': base64.b64encode(audio_file.read()).decode('utf-8'), 'format': mimetypes.guess_extension(mimetypes.guess_type(audio_file_to_use)[0]).lstrip('.'), } # Prepare chat completions request url = f'{api_base_url}/chat/completions' # Add language instruction if specified language = metadata.get('language', None) if metadata else None if language: text_instruction = f'Transcribe this audio exactly as spoken in {language}. Do not translate it.' else: text_instruction = 'Transcribe this audio exactly as spoken in its original language. Do not translate it to another language.' payload = { 'model': model, 'messages': [ { 'role': 'user', 'content': [ { 'type': 'input_audio', 'input_audio': audio_base64, }, {'type': 'text', 'text': text_instruction}, ], } ], } r = requests.post( url=url, json=payload, headers={ 'Authorization': f'Bearer {api_key}', 'Content-Type': 'application/json', }, timeout=AIOHTTP_CLIENT_TIMEOUT, ) r.raise_for_status() response = r.json() # Extract transcript from chat completion response transcript = response.get('choices', [{}])[0].get('message', {}).get('content', '').strip() if not transcript: raise ValueError('Empty transcript in response') data = {'text': transcript} else: # Use dedicated transcriptions API url = f'{api_base_url}/audio/transcriptions' # Determine the MIME type mime_type, _ = mimetypes.guess_type(file_path) if not mime_type: mime_type = 'audio/webm' # Use context manager to ensure file is properly closed with open(file_path, 'rb') as audio_file: files = {'file': (filename, audio_file, mime_type)} data_form = {'model': model} # Add language if specified in metadata language = metadata.get('language', None) if metadata else None if language: data_form['language'] = language r = requests.post( url=url, files=files, data=data_form, headers={ 'Authorization': f'Bearer {api_key}', }, timeout=AIOHTTP_CLIENT_TIMEOUT, ) r.raise_for_status() response = r.json() # Extract transcript from response transcript = response.get('text', '').strip() if not transcript: raise ValueError('Empty transcript in response') data = {'text': transcript} # Save transcript to json file (consistent with other providers) transcript_file = os.path.join(file_dir, f'{id}.json') with open(transcript_file, 'w') as f: json.dump(data, f) log.debug(data) return data except ValueError as e: log.exception('Error parsing Mistral response') raise HTTPException( status_code=500, detail=f'Failed to parse Mistral response: {str(e)}', ) except requests.exceptions.RequestException as e: log.exception(e) detail = None try: if r is not None and r.status_code != 200: res = r.json() if 'error' in res: detail = f'External: {res["error"].get("message", "")}' else: detail = f'External: {r.text}' except Exception: detail = f'External: {e}' raise HTTPException( status_code=getattr(r, 'status_code', 500) if r else 500, detail=detail if detail else 'Open WebUI: Server Connection Error', ) def transcribe(request: Request, file_path: str, metadata: Optional[dict] = None, user=None): log.info(f'transcribe: {file_path} {metadata}') if BYPASS_PYDUB_PREPROCESSING: log.info('Bypassing pydub preprocessing (BYPASS_PYDUB_PREPROCESSING=true)') chunk_paths = [file_path] else: if is_audio_conversion_required(file_path): file_path = convert_audio_to_mp3(file_path) if not file_path: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail='Audio conversion failed. The audio file may be corrupted or empty.', ) try: file_path = compress_audio(file_path) except Exception as e: log.exception(e) # Always produce a list of chunk paths (could be one entry if small) try: chunk_paths = split_audio(file_path, MAX_FILE_SIZE) print(f'Chunk paths: {chunk_paths}') except Exception as e: log.exception(e) raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT(e), ) results = [] try: if getattr(request.app.state.config, 'STT_ENGINE', '') == '': max_workers = 1 else: max_workers = None with ThreadPoolExecutor(max_workers=max_workers) as executor: # Submit tasks for each chunk_path futures = [ executor.submit(transcription_handler, request, chunk_path, metadata, user) for chunk_path in chunk_paths ] # Gather results as they complete for future in futures: try: results.append(future.result()) except HTTPException: raise except Exception as transcribe_exc: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f'Error transcribing chunk: {transcribe_exc}', ) finally: # Clean up only the temporary chunks, never the original file for chunk_path in chunk_paths: if chunk_path != file_path and os.path.isfile(chunk_path): try: os.remove(chunk_path) except Exception: pass return { 'text': ' '.join([result['text'] for result in results]), } def compress_audio(file_path): if os.path.getsize(file_path) > MAX_FILE_SIZE: id = os.path.splitext(os.path.basename(file_path))[0] # Handles names with multiple dots file_dir = os.path.dirname(file_path) audio = AudioSegment.from_file(file_path) audio = audio.set_frame_rate(16000).set_channels(1) # Compress audio compressed_path = os.path.join(file_dir, f'{id}_compressed.mp3') audio.export(compressed_path, format='mp3', bitrate='32k') # log.debug(f"Compressed audio to {compressed_path}") # Uncomment if log is defined return compressed_path else: return file_path def split_audio(file_path, max_bytes, format='mp3', bitrate='32k'): """ Splits audio into chunks not exceeding max_bytes. Returns a list of chunk file paths. If audio fits, returns list with original path. """ file_size = os.path.getsize(file_path) if file_size <= max_bytes: return [file_path] # Nothing to split audio = AudioSegment.from_file(file_path) duration_ms = len(audio) orig_size = file_size approx_chunk_ms = max(int(duration_ms * (max_bytes / orig_size)) - 1000, 1000) chunks = [] start = 0 i = 0 base, _ = os.path.splitext(file_path) while start < duration_ms: end = min(start + approx_chunk_ms, duration_ms) chunk = audio[start:end] chunk_path = f'{base}_chunk_{i}.{format}' chunk.export(chunk_path, format=format, bitrate=bitrate) # Reduce chunk duration if still too large while os.path.getsize(chunk_path) > max_bytes and (end - start) > 5000: end = start + ((end - start) // 2) chunk = audio[start:end] chunk.export(chunk_path, format=format, bitrate=bitrate) if os.path.getsize(chunk_path) > max_bytes: os.remove(chunk_path) raise Exception('Audio chunk cannot be reduced below max file size.') chunks.append(chunk_path) start = end i += 1 return chunks @router.post('/transcriptions') async def transcription( request: Request, file: UploadFile = File(...), language: Optional[str] = Form(None), user=Depends(get_verified_user), ): if user.role != 'admin' and not await has_permission( user.id, 'chat.stt', request.app.state.config.USER_PERMISSIONS ): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.ACCESS_PROHIBITED, ) log.info(f'file.content_type: {file.content_type}') stt_supported_content_types = getattr(request.app.state.config, 'STT_SUPPORTED_CONTENT_TYPES', []) if not strict_match_mime_type(stt_supported_content_types, file.content_type): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.FILE_NOT_SUPPORTED, ) try: safe_name = os.path.basename(file.filename) if file.filename else '' ext = safe_name.rsplit('.', 1)[-1].lower() if '.' in safe_name else '' allowed_extensions = getattr(request.app.state.config, 'STT_ALLOWED_EXTENSIONS', []) if allowed_extensions and ext not in allowed_extensions: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail='Invalid audio file extension', ) id = uuid.uuid4() filename = f'{id}.{ext}' contents = await file.read() file_dir = os.path.join(CACHE_DIR, 'audio', 'transcriptions') os.makedirs(file_dir, exist_ok=True) file_path = os.path.join(file_dir, filename) # Defense-in-depth: ensure resolved path stays within intended directory if not os.path.realpath(file_path).startswith(os.path.realpath(file_dir)): raise ValueError('Invalid file path detected') with open(file_path, 'wb') as f: f.write(contents) try: metadata = None if language: metadata = {'language': language} result = await asyncio.to_thread(transcribe, request, file_path, metadata, user) return { **result, 'filename': os.path.basename(file_path), } except HTTPException: raise except Exception as e: log.exception(e) raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail='Transcription failed.', ) except HTTPException: raise except Exception as e: log.exception(e) raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail='Transcription failed.', ) async def get_available_models(request: Request) -> list[dict]: available_models = [] if request.app.state.config.TTS_ENGINE == 'openai': # Use custom endpoint if not using the official OpenAI API URL if not request.app.state.config.TTS_OPENAI_API_BASE_URL.startswith('https://api.openai.com'): timeout = aiohttp.ClientTimeout(total=AIOHTTP_CLIENT_TIMEOUT_MODEL_LIST) async with aiohttp.ClientSession(timeout=timeout, trust_env=True) as session: try: async with session.get( f'{request.app.state.config.TTS_OPENAI_API_BASE_URL}/audio/models', ssl=AIOHTTP_CLIENT_SESSION_SSL, ) as response: response.raise_for_status() data = await response.json() available_models = data.get('models', []) except Exception as e: log.debug(f'/audio/models not available, trying /models fallback: {str(e)}') # Fallback to standard OpenAI-compatible /models endpoint # (used by KokoroTTS and similar custom TTS servers) try: async with session.get( f'{request.app.state.config.TTS_OPENAI_API_BASE_URL}/models', ssl=AIOHTTP_CLIENT_SESSION_SSL, ) as response: response.raise_for_status() data = await response.json() # OpenAI /models returns {"data": [...]}, /audio/models returns {"models": [...]} available_models = data.get('data', data.get('models', [])) except Exception as e2: log.error(f'Error fetching models from custom endpoint: {str(e2)}') available_models = [{'id': 'tts-1'}, {'id': 'tts-1-hd'}] else: available_models = [{'id': 'tts-1'}, {'id': 'tts-1-hd'}] elif request.app.state.config.TTS_ENGINE == 'elevenlabs': try: timeout = aiohttp.ClientTimeout(total=5) async with aiohttp.ClientSession(timeout=timeout, trust_env=True) as session: async with session.get( f'{ELEVENLABS_API_BASE_URL}/v1/models', headers={ 'xi-api-key': request.app.state.config.TTS_API_KEY, 'Content-Type': 'application/json', }, ssl=AIOHTTP_CLIENT_SESSION_SSL, ) as response: response.raise_for_status() models = await response.json() available_models = [{'name': model['name'], 'id': model['model_id']} for model in models] except Exception as e: log.error(f'Error fetching models: {str(e)}') elif request.app.state.config.TTS_ENGINE == 'mistral': available_models = [{'id': 'voxtral-mini-tts-2603'}] return available_models @router.get('/models') async def get_models(request: Request, user=Depends(get_verified_user)): return {'models': await get_available_models(request)} async def get_available_voices(request) -> dict: """Returns {voice_id: voice_name} dict""" available_voices = {} if request.app.state.config.TTS_ENGINE == 'openai': # Use custom endpoint if not using the official OpenAI API URL if not request.app.state.config.TTS_OPENAI_API_BASE_URL.startswith('https://api.openai.com'): try: timeout = aiohttp.ClientTimeout(total=AIOHTTP_CLIENT_TIMEOUT_MODEL_LIST) async with aiohttp.ClientSession(timeout=timeout, trust_env=True) as session: async with session.get( f'{request.app.state.config.TTS_OPENAI_API_BASE_URL}/audio/voices', ssl=AIOHTTP_CLIENT_SESSION_SSL, ) as response: response.raise_for_status() data = await response.json() voices_list = data.get('voices', []) available_voices = {voice['id']: voice['name'] for voice in voices_list} except Exception as e: log.error(f'Error fetching voices from custom endpoint: {str(e)}') available_voices = { 'alloy': 'alloy', 'echo': 'echo', 'fable': 'fable', 'onyx': 'onyx', 'nova': 'nova', 'shimmer': 'shimmer', } else: available_voices = { 'alloy': 'alloy', 'echo': 'echo', 'fable': 'fable', 'onyx': 'onyx', 'nova': 'nova', 'shimmer': 'shimmer', } elif request.app.state.config.TTS_ENGINE == 'elevenlabs': try: available_voices = await get_elevenlabs_voices(api_key=request.app.state.config.TTS_API_KEY) except Exception: # Avoided @lru_cache with exception pass elif request.app.state.config.TTS_ENGINE == 'azure': try: region = request.app.state.config.TTS_AZURE_SPEECH_REGION base_url = request.app.state.config.TTS_AZURE_SPEECH_BASE_URL url = (base_url or f'https://{region}.tts.speech.microsoft.com') + '/cognitiveservices/voices/list' headers = {'Ocp-Apim-Subscription-Key': request.app.state.config.TTS_API_KEY} timeout = aiohttp.ClientTimeout(total=AIOHTTP_CLIENT_TIMEOUT_MODEL_LIST) async with aiohttp.ClientSession(timeout=timeout, trust_env=True) as session: async with session.get(url, headers=headers, ssl=AIOHTTP_CLIENT_SESSION_SSL) as response: response.raise_for_status() voices = await response.json() for voice in voices: available_voices[voice['ShortName']] = f'{voice["DisplayName"]} ({voice["ShortName"]})' except Exception as e: log.error(f'Error fetching voices: {str(e)}') elif request.app.state.config.TTS_ENGINE == 'mistral': api_key = request.app.state.config.TTS_MISTRAL_API_KEY api_base_url = request.app.state.config.TTS_MISTRAL_API_BASE_URL or 'https://api.mistral.ai/v1' if api_key: try: timeout = aiohttp.ClientTimeout(total=AIOHTTP_CLIENT_TIMEOUT_MODEL_LIST) async with aiohttp.ClientSession(timeout=timeout, trust_env=True) as session: async with session.get( f'{api_base_url}/audio/voices', headers={ 'Authorization': f'Bearer {api_key}', }, ssl=AIOHTTP_CLIENT_SESSION_SSL, ) as response: response.raise_for_status() voices_data = await response.json() # Mistral returns a paginated response: {"items": [...], "page": ..., "total": ...} voices_list = voices_data.get('items', []) if isinstance(voices_data, dict) else voices_data for voice in voices_list: if isinstance(voice, dict): voice_id = voice.get('voice_id', voice.get('id', '')) voice_name = voice.get('name', voice_id) if voice_id: available_voices[voice_id] = voice_name except Exception as e: log.error(f'Error fetching Mistral voices: {str(e)}') return available_voices async def get_elevenlabs_voices(api_key: str) -> dict: """ Note, set the following in your .env file to use Elevenlabs: AUDIO_TTS_ENGINE=elevenlabs AUDIO_TTS_API_KEY=sk_... # Your Elevenlabs API key AUDIO_TTS_VOICE=EXAVITQu4vr4xnSDxMaL # From https://api.elevenlabs.io/v1/voices AUDIO_TTS_MODEL=eleven_multilingual_v2 """ try: # TODO: Add retries timeout = aiohttp.ClientTimeout(total=AIOHTTP_CLIENT_TIMEOUT_MODEL_LIST) async with aiohttp.ClientSession(timeout=timeout, trust_env=True) as session: async with session.get( f'{ELEVENLABS_API_BASE_URL}/v1/voices', headers={ 'xi-api-key': api_key, 'Content-Type': 'application/json', }, ssl=AIOHTTP_CLIENT_SESSION_SSL, ) as response: response.raise_for_status() voices_data = await response.json() voices = {} for voice in voices_data.get('voices', []): voices[voice['voice_id']] = voice['name'] except Exception as e: log.error(f'Error fetching voices: {str(e)}') raise RuntimeError(f'Error fetching voices: {str(e)}') return voices @router.get('/voices') async def get_voices(request: Request, user=Depends(get_verified_user)): return {'voices': [{'id': k, 'name': v} for k, v in (await get_available_voices(request)).items()]}