diff --git a/tests/tools/test_transcription.py b/tests/tools/test_transcription.py index 1aa430e0ab..fa12f76070 100644 --- a/tests/tools/test_transcription.py +++ b/tests/tools/test_transcription.py @@ -145,6 +145,7 @@ class TestTranscribeLocal: ("zh_TW", "zh"), ("繁體中文", "zh"), ("简体中文", "zh"), + ("xx", None), ("not-a-language", None), ], ) diff --git a/tests/tools/test_transcription_tools.py b/tests/tools/test_transcription_tools.py index faf5707bc5..b7fb0b5c75 100644 --- a/tests/tools/test_transcription_tools.py +++ b/tests/tools/test_transcription_tools.py @@ -1300,6 +1300,50 @@ class TestShellSafety: assert invocation["kwargs"].get("env") is not None assert not invocation["kwargs"].get("shell") + @pytest.mark.parametrize( + ("language", "expected"), + [("ZH", "zh"), ("zh-Hant", "zh"), ("繁體中文", "zh"), ("not-a-language", "en")], + ) + def test_local_command_normalizes_language_or_uses_default( + self, monkeypatch, sample_wav, tmp_path, language, expected + ): + from tools.transcription_tools import LOCAL_STT_COMMAND_ENV, _transcribe_local_command + + output_dir = tmp_path / "transcript-output" + output_dir.mkdir() + monkeypatch.setenv( + LOCAL_STT_COMMAND_ENV, + "whisper {input_path} --language {language} --output_dir {output_dir}", + ) + + class _TempDir: + def __enter__(self): + return str(output_dir) + + def __exit__(self, exc_type, exc, tb): + return False + + invocation = {} + + def fake_run(command, **kwargs): + invocation["command"] = command + (output_dir / "transcript.txt").write_text("safe", encoding="utf-8") + return subprocess.CompletedProcess(command, 0, stdout="", stderr="") + + monkeypatch.setattr( + "tools.transcription_local.tempfile.TemporaryDirectory", lambda prefix=None: _TempDir() + ) + monkeypatch.setattr("tools.transcription_audio.subprocess.run", fake_run) + monkeypatch.setattr( + "tools.transcription_tools._resolve_stt_language", lambda provider: language + ) + + result = _transcribe_local_command(sample_wav, "base") + + assert result["transcript"] == "safe" + language_index = invocation["command"].index("--language") + 1 + assert invocation["command"][language_index] == expected + class TestLocalModelLock: """#24767 — concurrent first-use must not double-load the whisper model.""" diff --git a/tools/transcription_local.py b/tools/transcription_local.py index 858013a327..6d2bc26815 100644 --- a/tools/transcription_local.py +++ b/tools/transcription_local.py @@ -28,6 +28,37 @@ from tools.transcription_common import ( logger = logging.getLogger("tools.transcription_tools") +_LOCAL_LANGUAGE_ALIASES = { + "繁體中文": "zh", + "繁体中文": "zh", + "简体中文": "zh", + "簡體中文": "zh", +} +_THREE_LETTER_WHISPER_CODES = frozenset({"haw", "yue"}) + + +def _normalize_local_stt_language( + language: Optional[str], supported_languages: object = None +) -> Optional[str]: + """Return a Whisper language code, or None so the caller can fall back safely.""" + if not isinstance(language, str) or not language.strip(): + return None + raw = language.strip() + folded = raw.casefold().replace("_", "-") + candidate = _LOCAL_LANGUAGE_ALIASES.get(raw, folded.split("-", 1)[0]) + code_shape_is_valid = len(candidate) == 2 or candidate in _THREE_LETTER_WHISPER_CODES + if not (candidate.isascii() and candidate.isalpha() and code_shape_is_valid): + logger.warning("Local STT language %r is not a language code; using fallback", raw) + return None + + if isinstance(supported_languages, (list, tuple, set, frozenset)): + supported_codes = {str(code).casefold() for code in supported_languages} + if candidate not in supported_codes: + logger.warning("Local STT language %r is unsupported; using fallback", raw) + return None + return candidate + + def _get_local_command_template() -> Optional[str]: configured = os.getenv(LOCAL_STT_COMMAND_ENV, "").strip() if configured: @@ -275,7 +306,8 @@ def _transcribe_local_command( if not command_template: return _error_result(f"{LOCAL_STT_COMMAND_ENV} not configured and no local whisper binary was found") # Language: hook override > stt.local.language > stt.language > env > "en". - language = language or _resolve_stt_language("local") or DEFAULT_LOCAL_STT_LANGUAGE + configured_language = language or _resolve_stt_language("local") + language = _normalize_local_stt_language(configured_language) or DEFAULT_LOCAL_STT_LANGUAGE normalized_model = _normalize_local_model(model_name) try: if not os.getenv(LOCAL_STT_COMMAND_ENV, "").strip(): diff --git a/tools/transcription_tools.py b/tools/transcription_tools.py index a639043637..65bc967317 100644 --- a/tools/transcription_tools.py +++ b/tools/transcription_tools.py @@ -33,7 +33,8 @@ from tools.transcription_audio import ( from tools.transcription_local import ( _get_idle_unload_seconds, _has_local_command, _join_confident_segments, _load_local_whisper_model, _looks_like_cuda_lib_error, _normalize_local_model, - _transcribe_local_command, _try_lazy_install_stt, build_local_transcribe_kwargs) + _normalize_local_stt_language, _transcribe_local_command, _try_lazy_install_stt, + build_local_transcribe_kwargs) # The ``_transcribe_`` handlers are looked up in this module's globals by _dispatch_stt_provider. from tools.transcription_cloud import ( # noqa: F401 (handlers dispatched via globals()) _has_xai_stt_credentials, _resolve_openai_audio_client_config, _transcribe_deepinfra, @@ -112,34 +113,6 @@ def _resolve_stt_language( return next((c.strip() for c in candidates if isinstance(c, str) and c.strip()), None) -_LOCAL_LANGUAGE_ALIASES = { - "繁體中文": "zh", - "繁体中文": "zh", - "简体中文": "zh", - "簡體中文": "zh", -} - - -def _normalize_local_stt_language(language: Optional[str], model: object) -> Optional[str]: - """Return a faster-whisper language code, or None to preserve auto-detection.""" - if not isinstance(language, str) or not language.strip(): - return None - raw = language.strip() - folded = raw.casefold().replace("_", "-") - candidate = _LOCAL_LANGUAGE_ALIASES.get(raw, folded.split("-", 1)[0]) - if not (candidate.isascii() and candidate.isalpha() and 2 <= len(candidate) <= 3): - logger.warning("Local STT language %r is not a language code; using auto-detection", raw) - return None - - supported = getattr(model, "supported_languages", None) - if isinstance(supported, (list, tuple, set, frozenset)): - supported_codes = {str(code).casefold() for code in supported} - if candidate not in supported_codes: - logger.warning("Local STT language %r is unsupported; using auto-detection", raw) - return None - return candidate - - def _openai_audio_unavailable_reason() -> Optional[str]: """None when OpenAI audio has usable credentials (config, env, or managed gateway); else the reason.""" try: @@ -408,7 +381,8 @@ def _transcribe_local( # pre_transcription hook overrides win over config-resolved values. transcribe_kwargs = build_local_transcribe_kwargs(stt_config) effective_language = language if language is not None else transcribe_kwargs.get("language") - normalized_language = _normalize_local_stt_language(effective_language, model) + normalized_language = _normalize_local_stt_language( + effective_language, getattr(model, "supported_languages", None)) if normalized_language: transcribe_kwargs["language"] = normalized_language else: