From 7202269cee61ec5e0634b50f212b251a6736bb06 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 16:07:29 -0700 Subject: [PATCH] refactor(stt): compact xAI/ElevenLabs/DeepInfra bodies and OpenAI credential resolver --- tools/transcription_cloud.py | 105 ++++++++++------------------------- 1 file changed, 30 insertions(+), 75 deletions(-) diff --git a/tools/transcription_cloud.py b/tools/transcription_cloud.py index 102bf780c0..9ddc134c60 100644 --- a/tools/transcription_cloud.py +++ b/tools/transcription_cloud.py @@ -346,9 +346,7 @@ def _transcribe_xai( creds = resolve_xai_http_credentials() api_key = str(creds.get("api_key") or "").strip() if not api_key: - return _error_result( - "No xAI credentials found. Configure xAI OAuth in `hermes model` or set XAI_API_KEY" - ) + return _error_result("No xAI credentials found. Configure xAI OAuth in `hermes model` or set XAI_API_KEY") stt_config = _load_stt_config() xai_config = stt_config.get("xai") or {} @@ -362,11 +360,8 @@ def _transcribe_xai( url = xai_config.get("base_url") or get_env_value("XAI_STT_BASE_URL") or resolved_creds.get("base_url") return str(url or XAI_STT_BASE_URL).strip().rstrip("/") - base_url = _resolve_base_url(creds) # Language: hook override > stt.xai.language > stt.language > env. language = language or _resolve_stt_language("xai", stt_config) or "" - use_format = is_truthy_value(xai_config.get("format", True)) - use_diarize = is_truthy_value(xai_config.get("diarize", False)) try: from tools.xai_http import hermes_xai_user_agent @@ -374,8 +369,8 @@ def _transcribe_xai( data: Dict[str, str] = {} if language: data["language"] = language - for flag, enabled in (("format", use_format), ("diarize", use_diarize)): - if enabled: + for flag, default in (("format", True), ("diarize", False)): + if is_truthy_value(xai_config.get(flag, default)): data[flag] = "true" def _post_transcription(bearer: str, endpoint_base_url: str): @@ -385,32 +380,18 @@ def _transcribe_xai( file_path, data, ) - response = _post_transcription(api_key, base_url) + response = _post_transcription(api_key, _resolve_base_url(creds)) - if ( - response.status_code in {401, 403} - and creds.get("provider") == "xai-oauth" - ): - logger.info( - "xAI STT got HTTP %d; refreshing OAuth credentials and retrying once", - response.status_code, - ) + if response.status_code in {401, 403} and creds.get("provider") == "xai-oauth": + logger.info("xAI STT got HTTP %d; refreshing OAuth credentials and retrying once", response.status_code) try: - refreshed_creds = resolve_xai_http_credentials( - force_refresh=True, - api_key_hint=api_key, - ) + refreshed_creds = resolve_xai_http_credentials(force_refresh=True, api_key_hint=api_key) refreshed_key = str(refreshed_creds.get("api_key") or "").strip() if refreshed_key and refreshed_key != api_key: - response = _post_transcription( - refreshed_key, - _resolve_base_url(refreshed_creds), - ) + response = _post_transcription(refreshed_key, _resolve_base_url(refreshed_creds)) except Exception as retry_exc: logger.warning( - "xAI STT OAuth refresh-and-retry after HTTP %d failed: %s", - response.status_code, - retry_exc, + "xAI STT OAuth refresh-and-retry after HTTP %d failed: %s", response.status_code, retry_exc, ) transcript_text, result, error = _rest_transcript( @@ -453,21 +434,16 @@ def _transcribe_elevenlabs( ).strip().rstrip("/") # Language: hook override > stt.elevenlabs.language(_code) > stt.language. language_code = language or _resolve_stt_language("elevenlabs", stt_config, extra_keys=("language_code",)) or "" - tag_audio_events = is_truthy_value(elevenlabs_config.get("tag_audio_events", False)) - diarize = is_truthy_value(elevenlabs_config.get("diarize", False)) - try: data: Dict[str, str] = { "model_id": model_name, - "tag_audio_events": "true" if tag_audio_events else "false", - "diarize": "true" if diarize else "false", + "tag_audio_events": str(is_truthy_value(elevenlabs_config.get("tag_audio_events", False))).lower(), + "diarize": str(is_truthy_value(elevenlabs_config.get("diarize", False))).lower(), } if language_code: data["language_code"] = language_code - response = _post_audio_multipart( - f"{base_url}/speech-to-text", {"xi-api-key": api_key}, file_path, data, - ) + response = _post_audio_multipart(f"{base_url}/speech-to-text", {"xi-api-key": api_key}, file_path, data) transcript_text, _body, error = _rest_transcript( response, "ElevenLabs STT", _elevenlabs_error_detail, _extract_transcript_text, @@ -507,21 +483,14 @@ def _transcribe_deepinfra( candidates = deepinfra_model_ids("stt") if not candidates: return _error_result( - "No DeepInfra STT model available. Pin one in " - "config.yaml under stt.deepinfra.model, or check " - "connectivity to api.deepinfra.com so the live catalog " - "can be fetched." + "No DeepInfra STT model available. Pin one in config.yaml under stt.deepinfra.model, " + "or check connectivity to api.deepinfra.com so the live catalog can be fetched." ) model_name = candidates[0] return _transcribe_openai( - file_path, - model_name, - api_key=api_key, - base_url=base_url, - provider_label="deepinfra", - language=language, - prompt=prompt, + file_path, model_name, api_key=api_key, base_url=base_url, provider_label="deepinfra", + language=language, prompt=prompt, ) @@ -541,11 +510,9 @@ def _is_local_or_private_url(url: str) -> bool: return False if host == "localhost" or host.endswith((".local", ".lan", ".internal")): return True - try: - return ipaddress.ip_address(host).is_private or ipaddress.ip_address(host).is_loopback - except ValueError: - return False - except Exception: + addr = ipaddress.ip_address(host) + return addr.is_private or addr.is_loopback + except Exception: # unparsable URL or non-IP hostname return False @@ -577,17 +544,13 @@ def _resolve_openai_audio_client_config() -> tuple[str, str]: - never-configured stt section → legacy ladder: direct credentials, then the managed gateway. """ - from tools.transcription_tools import _load_stt_config, managed_nous_tools_enabled, nous_tool_gateway_unavailable_message, resolve_managed_tool_gateway - from tools.tool_backend_helpers import ( - NOUS_MANAGED_PROVIDER, - read_selection, - selection_error, + from tools.transcription_tools import ( + _load_stt_config, managed_nous_tools_enabled, nous_tool_gateway_unavailable_message, + resolve_managed_tool_gateway, ) + from tools.tool_backend_helpers import NOUS_MANAGED_PROVIDER, read_selection, selection_error openai_cfg = _load_stt_config().get("openai") or {} - cfg_api_key = openai_cfg.get("api_key", "") - cfg_base_url = openai_cfg.get("base_url", "") - selected = read_selection("stt") def _managed() -> Optional[tuple[str, str]]: @@ -600,23 +563,19 @@ def _resolve_openai_audio_client_config() -> tuple[str, str]: managed = _managed() if managed is None: raise ValueError(selection_error( - "stt", - NOUS_MANAGED_PROVIDER, - "the Nous Tool Gateway is not available (not entitled or " - "unreachable)", + "stt", NOUS_MANAGED_PROVIDER, + "the Nous Tool Gateway is not available (not entitled or unreachable)", )) return managed - direct = _direct_openai_credentials(cfg_api_key, cfg_base_url) + direct = _direct_openai_credentials(openai_cfg.get("api_key", ""), openai_cfg.get("base_url", "")) if direct is not None: return direct if selected is not None: raise ValueError(selection_error( - "stt", - selected, - "neither stt.openai.api_key in config nor " - "VOICE_TOOLS_OPENAI_KEY/OPENAI_API_KEY is set", + "stt", selected, + "neither stt.openai.api_key in config nor VOICE_TOOLS_OPENAI_KEY/OPENAI_API_KEY is set", )) managed = _managed() @@ -640,10 +599,6 @@ def _extract_transcript_text(transcription: Any) -> str: match = re.match( r"\s*language\s+[\w.-]+(?:\s*[^<]*)?\s*\s*(?P.*)", - text, - flags=re.IGNORECASE | re.DOTALL, + text, flags=re.IGNORECASE | re.DOTALL, ) - if match: - text = match.group("text").strip() - - return text + return match.group("text").strip() if match else text