fix(stt): normalize local command language hints

This commit is contained in:
Kuxin
2026-10-05 14:14:36 +08:00
committed by brooklyn!
parent 0d234e714c
commit 0fa1fb600d
4 changed files with 82 additions and 31 deletions

View File

@@ -145,6 +145,7 @@ class TestTranscribeLocal:
("zh_TW", "zh"),
("繁體中文", "zh"),
("简体中文", "zh"),
("xx", None),
("not-a-language", None),
],
)

View File

@@ -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."""

View File

@@ -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():

View File

@@ -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_<provider>`` 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: