ddmitry/local-transcriber
1
0
Fork 0
Code Issues 6 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(diarization): добавлено разделение транскрипта по говорящим

Browse Source 
Зачем:
- локальным транскриптам нужна структура реплик для конспектов и протоколов.

Что:
- добавлены пословные таймкоды для всех ASR-бэкендов и сведение с Sherpa-ONNX.
- реализованы CLI-флаги, деградация без потери ASR и speaker Markdown.
- добавлены проверяемый кеш моделей, тесты и документация.

Проверка:
- `pytest` — 283 passed, 1 skipped.
- `pyright` — 0 errors.
- Ruff и `git diff --check` — без ошибок.
- выполнены три контрольных прогона на реальных записях.
This commit is contained in:
Dmitriy Dementiev
2026-08-14 18:55:42 +03:00
parent 63ba41d906
commit 726718197f
23 changed files with 2522 additions and 187 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_backend_faster_whisper.py
+66
View File
@@ -0,0 +1,66 @@
from types import SimpleNamespace
from unittest.mock import MagicMock
import pytest
from local_transcriber.backends.faster_whisper import FasterWhisperBackend
from local_transcriber.types import Word
def test_transcribe_returns_canonical_words(tmp_path):
audio = tmp_path / "audio.wav"
raw_word = SimpleNamespace(start=0.2, end=0.7, word=" Привет")
raw_segment = SimpleNamespace(
start=0.0,
end=1.0,
text=" Привет",
words=[raw_word],
)
info = SimpleNamespace(duration=1.0, language="ru", language_probability=0.99)
model = MagicMock()
model.transcribe.return_value = (iter([raw_segment]), info)
result = FasterWhisperBackend().transcribe(model, audio, language="ru")
assert result.words == [Word(start=0.2, end=0.7, text=" Привет")]
model.transcribe.assert_called_once_with(
str(audio),
language="ru",
word_timestamps=True,
)
def test_transcribe_rejects_nonempty_result_without_word_timestamps(tmp_path):
raw_segment = SimpleNamespace(
start=0.0,
end=1.0,
text=" Текст есть",
words=None,
)
info = SimpleNamespace(duration=1.0, language="ru", language_probability=1.0)
model = MagicMock()
model.transcribe.return_value = (iter([raw_segment]), info)
with pytest.raises(RuntimeError, match="пословные таймкоды"):
FasterWhisperBackend().transcribe(model, tmp_path / "audio.wav", "ru")
def test_transcribe_rejects_one_nonempty_segment_without_word_timestamps(tmp_path):
timestamped = SimpleNamespace(
start=0.0,
end=1.0,
text=" Первое",
words=[SimpleNamespace(start=0.0, end=1.0, word=" Первое")],
)
missing = SimpleNamespace(
start=1.0,
end=2.0,
text=" Второе",
words=None,
)
info = SimpleNamespace(duration=2.0, language="ru", language_probability=1.0)
model = MagicMock()
model.transcribe.return_value = (iter([timestamped, missing]), info)
with pytest.raises(RuntimeError, match="пословные таймкоды"):
FasterWhisperBackend().transcribe(model, tmp_path / "audio.wav", "ru")
tests/test_backend_openvino.py
+140 -20
View File
@@ -11,8 +11,7 @@ from local_transcriber.backends.openvino import (
OpenVINOBackend,
_validate_model_dir,
)
from local_transcriber.types import UNKNOWN_LANGUAGE, Segment
from local_transcriber.types import UNKNOWN_LANGUAGE, Segment, Word
# === _resolve_repo ===
@@ -28,12 +27,18 @@ def test_model_catalog_contains_large_v3_turbo_profiles():
def test_resolve_repo_exact_match():
backend = OpenVINOBackend(compute_type_explicit=True)
assert backend._resolve_repo("medium", "int8") == ("OpenVINO/whisper-medium-int8-ov", "int8")
assert backend._resolve_repo("medium", "int8") == (
"OpenVINO/whisper-medium-int8-ov",
"int8",
)
def test_resolve_repo_large_v3_fp16():
backend = OpenVINOBackend(compute_type_explicit=True)
assert backend._resolve_repo("large-v3", "fp16") == ("OpenVINO/whisper-large-v3-fp16-ov", "fp16")
assert backend._resolve_repo("large-v3", "fp16") == (
"OpenVINO/whisper-large-v3-fp16-ov",
"fp16",
)
def test_resolve_repo_explicit_unsupported_pair_raises():
@@ -53,16 +58,17 @@ def test_resolve_repo_implicit_fallback():
"""Неявный compute_type: если int8 недоступен для base, fallback на fp16."""
backend = OpenVINOBackend(compute_type_explicit=False)
# base + int8 не существует, но base + fp16 есть
assert backend._resolve_repo("base", "int8") == ("OpenVINO/whisper-base-fp16-ov", "fp16")
assert backend._resolve_repo("base", "int8") == (
"OpenVINO/whisper-base-fp16-ov",
"fp16",
)
@pytest.mark.parametrize(
("model_name", "expected_compute_type"),
[("large-v3", "fp16"), ("large-v3-turbo", "int8")],
)
def test_resolve_repo_implicit_large_v3_profiles(
model_name, expected_compute_type
):
def test_resolve_repo_implicit_large_v3_profiles(model_name, expected_compute_type):
"""Неявный compute_type различает обычную и turbo-модель."""
backend = OpenVINOBackend(compute_type_explicit=False)
@@ -75,7 +81,10 @@ def test_resolve_repo_implicit_large_v3_profiles(
def test_resolve_repo_explicit_large_v3_int8_respected():
"""Явный --compute-type int8 для large-v3 → уважается."""
backend = OpenVINOBackend(compute_type_explicit=True)
assert backend._resolve_repo("large-v3", "int8") == ("OpenVINO/whisper-large-v3-int8-ov", "int8")
assert backend._resolve_repo("large-v3", "int8") == (
"OpenVINO/whisper-large-v3-int8-ov",
"int8",
)
@pytest.mark.parametrize("compute_type", ["int8", "fp16"])
@@ -107,6 +116,7 @@ def test_ensure_model_available_cache_hit(mock_download, tmp_path):
model_dir.mkdir()
(model_dir / "openvino_encoder_model.xml").write_text("<xml/>")
(model_dir / "openvino_decoder_model.xml").write_text("<xml/>")
(model_dir / "generation_config.json").write_text('{"alignment_heads": [[1, 2]]}')
mock_download.return_value = str(model_dir)
backend = OpenVINOBackend(compute_type_explicit=True)
@@ -125,6 +135,7 @@ def test_ensure_model_available_downloads(mock_download, tmp_path):
model_dir.mkdir()
(model_dir / "openvino_encoder_model.xml").write_text("<xml/>")
(model_dir / "openvino_decoder_model.xml").write_text("<xml/>")
(model_dir / "generation_config.json").write_text('{"alignment_heads": [[1, 2]]}')
mock_download.side_effect = [
LocalEntryNotFoundError("not cached"),
@@ -145,6 +156,7 @@ def test_large_v3_turbo_model_is_resolved_and_created(mock_download, tmp_path):
model_dir.mkdir()
(model_dir / "openvino_encoder_model.xml").write_text("<xml/>")
(model_dir / "openvino_decoder_model.xml").write_text("<xml/>")
(model_dir / "generation_config.json").write_text('{"alignment_heads": [[1, 2]]}')
mock_download.return_value = str(model_dir)
mock_ov = MagicMock()
@@ -157,12 +169,28 @@ def test_large_v3_turbo_model_is_resolved_and_created(mock_download, tmp_path):
"OpenVINO/whisper-large-v3-turbo-int8-ov",
local_files_only=True,
)
mock_ov.WhisperPipeline.assert_called_once_with(str(model_dir), "CPU")
mock_ov.WhisperPipeline.assert_called_once_with(
str(model_dir), "CPU", word_timestamps=True
)
# === create_model ===
def test_create_model_enables_word_timestamps():
mock_ov = MagicMock()
backend = OpenVINOBackend(ov_device="openvino-cpu")
with patch.dict("sys.modules", {"openvino_genai": mock_ov}):
backend.create_model("/path/to/model", "openvino-cpu", "int8")
mock_ov.WhisperPipeline.assert_called_once_with(
"/path/to/model",
"CPU",
word_timestamps=True,
)
def test_create_model_cpu():
mock_ov = MagicMock()
mock_pipeline = MagicMock()
@@ -172,7 +200,9 @@ def test_create_model_cpu():
with patch.dict("sys.modules", {"openvino_genai": mock_ov}):
model = backend.create_model("/path/to/model", "openvino-cpu", "int8")
mock_ov.WhisperPipeline.assert_called_once_with("/path/to/model", "CPU")
mock_ov.WhisperPipeline.assert_called_once_with(
"/path/to/model", "CPU", word_timestamps=True
)
assert model is mock_pipeline
assert backend.actual_ov_device == "CPU"
@@ -186,7 +216,9 @@ def test_create_model_gpu():
with patch.dict("sys.modules", {"openvino_genai": mock_ov}):
model = backend.create_model("/path/to/model", "openvino-gpu", "fp16")
mock_ov.WhisperPipeline.assert_called_once_with("/path/to/model", "GPU")
mock_ov.WhisperPipeline.assert_called_once_with(
"/path/to/model", "GPU", word_timestamps=True
)
assert model is mock_pipeline
assert backend.actual_ov_device == "GPU"
@@ -202,11 +234,16 @@ def test_create_model_openvino_auto_detects_gpu():
backend = OpenVINOBackend(ov_device="openvino")
with (
patch.dict("sys.modules", {"openvino_genai": mock_ov, "openvino": MagicMock(Core=mock_core)}),
patch.dict(
"sys.modules",
{"openvino_genai": mock_ov, "openvino": MagicMock(Core=mock_core)},
),
):
model = backend.create_model("/path/to/model", "openvino", "int8")
backend.create_model("/path/to/model", "openvino", "int8")
mock_ov.WhisperPipeline.assert_called_once_with("/path/to/model", "GPU")
mock_ov.WhisperPipeline.assert_called_once_with(
"/path/to/model", "GPU", word_timestamps=True
)
assert backend.actual_ov_device == "GPU"
@@ -221,11 +258,16 @@ def test_create_model_openvino_auto_falls_back_to_cpu():
backend = OpenVINOBackend(ov_device="openvino")
with (
patch.dict("sys.modules", {"openvino_genai": mock_ov, "openvino": MagicMock(Core=mock_core)}),
patch.dict(
"sys.modules",
{"openvino_genai": mock_ov, "openvino": MagicMock(Core=mock_core)},
),
):
model = backend.create_model("/path/to/model", "openvino", "int8")
backend.create_model("/path/to/model", "openvino", "int8")
mock_ov.WhisperPipeline.assert_called_once_with("/path/to/model", "CPU")
mock_ov.WhisperPipeline.assert_called_once_with(
"/path/to/model", "CPU", word_timestamps=True
)
assert backend.actual_ov_device == "CPU"
@@ -248,13 +290,19 @@ def test_transcribe_maps_chunks_to_segments():
mock_result = MagicMock()
mock_result.chunks = [chunk1, chunk2]
mock_result.words = [
MagicMock(start_ts=0.0, end_ts=3.5, word=" Привет мир"),
MagicMock(start_ts=3.5, end_ts=7.0, word=" Тестовый сегмент"),
]
mock_model.generate.return_value = mock_result
raw_audio = np.zeros(16000 * 10, dtype=np.float32) # 10 секунд
with patch("faster_whisper.decode_audio", return_value=raw_audio):
result = backend.transcribe(
mock_model, Path("test.mp3"), language="ru",
mock_model,
Path("test.mp3"),
language="ru",
)
assert len(result.segments) == 2
@@ -269,6 +317,61 @@ def test_transcribe_maps_chunks_to_segments():
assert call_kwargs.kwargs["return_timestamps"] is True
def test_transcribe_maps_word_level_timestamps():
backend = OpenVINOBackend()
mock_model = MagicMock()
raw_word = MagicMock()
raw_word.start_ts = 0.2
raw_word.end_ts = 0.8
raw_word.word = " Привет"
mock_result = MagicMock()
mock_result.chunks = []
mock_result.words = [raw_word]
mock_model.generate.return_value = mock_result
with patch(
"faster_whisper.decode_audio",
return_value=np.zeros(16_000, dtype=np.float32),
):
result = backend.transcribe(mock_model, Path("test.mp3"), language="ru")
assert result.words == [Word(start=0.2, end=0.8, text=" Привет")]
assert mock_model.generate.call_args.kwargs["word_timestamps"] is True
def test_transcribe_keeps_zero_duration_word_timestamp():
backend = OpenVINOBackend()
mock_model = MagicMock()
raw_word = MagicMock(start_ts=1.0, end_ts=1.0, word=" Слово")
mock_result = MagicMock(chunks=[], words=[raw_word])
mock_model.generate.return_value = mock_result
with patch(
"faster_whisper.decode_audio",
return_value=np.zeros(16_000, dtype=np.float32),
):
result = backend.transcribe(mock_model, Path("test.mp3"), language="ru")
assert result.words == [Word(start=1.0, end=1.0, text=" Слово")]
def test_transcribe_rejects_nonempty_result_without_word_timestamps():
backend = OpenVINOBackend()
chunk = MagicMock(start_ts=0.0, end_ts=1.0, text=" Текст")
mock_result = MagicMock(chunks=[chunk], words=None)
mock_model = MagicMock()
mock_model.generate.return_value = mock_result
with (
patch(
"faster_whisper.decode_audio",
return_value=np.zeros(16_000, dtype=np.float32),
),
pytest.raises(RuntimeError, match="пословные таймкоды"),
):
backend.transcribe(mock_model, Path("test.mp3"), language="ru")
def test_transcribe_calls_tolist():
"""raw_speech передаётся как list, не ndarray."""
backend = OpenVINOBackend()
@@ -314,6 +417,7 @@ def test_transcribe_calls_on_segment():
chunk.text = " Test"
mock_result = MagicMock()
mock_result.chunks = [chunk]
mock_result.words = [MagicMock(start_ts=0.0, end_ts=2.0, word=" Test")]
mock_model.generate.return_value = mock_result
raw_audio = np.zeros(16000, dtype=np.float32)
@@ -321,7 +425,10 @@ def test_transcribe_calls_on_segment():
with patch("faster_whisper.decode_audio", return_value=raw_audio):
backend.transcribe(
mock_model, Path("test.mp3"), language="en", on_segment=callback,
mock_model,
Path("test.mp3"),
language="en",
on_segment=callback,
)
callback.assert_called_once()
@@ -336,6 +443,7 @@ def test_transcribe_calls_on_segment():
def test_validate_model_dir_ok(tmp_path):
(tmp_path / "openvino_encoder_model.xml").write_text("<xml/>")
(tmp_path / "openvino_decoder_model.xml").write_text("<xml/>")
(tmp_path / "generation_config.json").write_text('{"alignment_heads": [[1, 2]]}')
_validate_model_dir(tmp_path) # should not raise
@@ -343,3 +451,15 @@ def test_validate_model_dir_missing(tmp_path):
(tmp_path / "openvino_encoder_model.xml").write_text("<xml/>")
with pytest.raises(ValueError, match="openvino_decoder_model.xml"):
_validate_model_dir(tmp_path)
def test_validate_model_dir_requires_alignment_heads_for_word_timestamps(tmp_path):
(tmp_path / "openvino_encoder_model.xml").write_text("<xml/>")
(tmp_path / "openvino_decoder_model.xml").write_text("<xml/>")
(tmp_path / "generation_config.json").write_text(
'{"alignment_heads": []}',
encoding="utf-8",
)
with pytest.raises(ValueError, match="alignment_heads"):
_validate_model_dir(tmp_path)
tests/test_cli.py
+580 -64
View File
@@ -12,15 +12,26 @@ from local_transcriber.formatter import (
LANGUAGE_FROM_MODEL,
LANGUAGE_UNKNOWN,
)
from local_transcriber.transcriber import Segment, TranscribeFileResult, TranscribeResult
from local_transcriber.types import UNKNOWN_LANGUAGE
from local_transcriber.transcriber import (
Segment,
TranscribeFileResult,
TranscribeResult,
)
from local_transcriber.types import (
UNKNOWN_LANGUAGE,
DiarizationRun,
SpeakerInterval,
Word,
)
runner = CliRunner()
def _make_result(segments=None, language="ru", device_used="cpu", duration=60.0):
return TranscribeResult(
segments=[Segment(start=0.0, end=2.0, text="Hello")] if segments is None else segments,
segments=[Segment(start=0.0, end=2.0, text="Hello")]
if segments is None
else segments,
language=language,
language_probability=0.95,
duration=duration,
@@ -36,7 +47,13 @@ def _make_backend():
return MagicMock(name="Backend")
def _make_tfr(result=None, model=None, actual_device="cpu", backend=None, model_path="/models/medium"):
def _make_tfr(
result=None,
model=None,
actual_device="cpu",
backend=None,
model_path="/models/medium",
):
if result is None:
result = _make_result()
if model is None:
@@ -44,8 +61,11 @@ def _make_tfr(result=None, model=None, actual_device="cpu", backend=None, model_
if backend is None:
backend = _make_backend()
return TranscribeFileResult(
result=result, model=model, actual_device=actual_device,
backend=backend, model_path=model_path,
result=result,
model=model,
actual_device=actual_device,
backend=backend,
model_path=model_path,
)
@@ -58,9 +78,7 @@ def _make_tfr(result=None, model=None, actual_device="cpu", backend=None, model_
("auto", UNKNOWN_LANGUAGE, 0.0, LANGUAGE_UNKNOWN),
],
)
def test_format_language_mode(
requested_language, language, probability, expected
):
def test_format_language_mode(requested_language, language, probability, expected):
result = _make_result(language=language)
result.language_probability = probability
@@ -73,12 +91,17 @@ def _single_patches(result=None, tmp_file=None, actual_device="cpu"):
result = _make_result(device_used=actual_device)
model = _make_model()
backend = _make_backend()
tfr = _make_tfr(result=result, model=model, actual_device=actual_device, backend=backend)
tfr = _make_tfr(
result=result, model=model, actual_device=actual_device, backend=backend
)
return [
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", return_value=tmp_file),
patch("local_transcriber.cli.detect_device", return_value=actual_device),
patch("local_transcriber.cli.load_model", return_value=(model, actual_device, backend, "/models/medium")),
patch(
"local_transcriber.cli.load_model",
return_value=(model, actual_device, backend, "/models/medium"),
),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch("local_transcriber.cli.write_transcript"),
]
@@ -139,18 +162,28 @@ def test_cli_custom_options(tmp_path):
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cuda"),
patch("local_transcriber.cli.load_model", return_value=(model, "cuda", backend, "/models/small")),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cuda", backend, "/models/small"),
),
patch("local_transcriber.cli._transcribe_file", mock_transcribe_file),
patch("local_transcriber.cli.write_transcript"),
patch("local_transcriber.cli.get_gpu_name", return_value="RTX 3060"),
):
runner.invoke(app, [
str(audio),
"--model", "small",
"--language", "ru",
"--device", "cuda",
"--compute-type", "float16",
])
runner.invoke(
app,
[
str(audio),
"--model",
"small",
"--language",
"ru",
"--device",
"cuda",
"--compute-type",
"float16",
],
)
call_kwargs = mock_transcribe_file.call_args[1]
assert call_kwargs["model_name"] == "small"
@@ -158,6 +191,266 @@ def test_cli_custom_options(tmp_path):
assert call_kwargs["compute_type"] == "float16"
def test_cli_speakers_enables_diarization_and_writes_speaker_markdown(tmp_path):
audio = tmp_path / "meeting.mp3"
audio.write_bytes(b"fake")
result = _make_result(
segments=[Segment(0.0, 1.3, "Первый. Второй. Неясно.")],
duration=10.0,
)
result.words = [
Word(0.0, 0.5, "Первый."),
Word(0.5, 1.0, "Второй."),
Word(1.1, 1.3, "Неясно."),
]
model = _make_model()
backend = _make_backend()
backend.word_timestamps_available = True
tfr = _make_tfr(result=result, model=model, backend=backend)
diarizer = MagicMock()
diarizer.process.return_value = DiarizationRun(
intervals=[
SpeakerInterval(0.0, 0.5, 10),
SpeakerInterval(0.5, 1.0, 20),
],
elapsed_seconds=0.2,
)
write = MagicMock()
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch(
"local_transcriber.cli.load_speaker_diarizer",
return_value=diarizer,
) as load_diarizer,
patch("local_transcriber.cli.write_transcript", write),
):
out = runner.invoke(
app,
[str(audio), "--speakers", "2", "--threads", "3"],
)
assert out.exit_code == 0
load_diarizer.assert_called_once()
assert load_diarizer.call_args.kwargs["speakers"] == 2
assert load_diarizer.call_args.kwargs["threads"] == 3
diarizer.process.assert_called_once()
assert "Speaker 1: Первый." in write.call_args.args[0]
assert "Speaker 2: Второй." in write.call_args.args[0]
assert "Speaker ?: Неясно." in write.call_args.args[0]
assert "1 слов без назначенного говорящего" in out.output
assert "малый кластер Speaker 1: 0.5 с" in out.output
def test_cli_diarization_error_writes_plain_transcript_and_exits_nonzero(tmp_path):
audio = tmp_path / "meeting.mp3"
audio.write_bytes(b"fake")
result = _make_result(
segments=[Segment(0.0, 1.0, "Полезный текст.")],
duration=10.0,
)
result.words = [Word(0.0, 1.0, "Полезный текст.")]
model = _make_model()
backend = _make_backend()
backend.word_timestamps_available = True
tfr = _make_tfr(result=result, model=model, backend=backend)
diarizer = MagicMock()
diarizer.process.side_effect = RuntimeError("boom")
write = MagicMock()
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch(
"local_transcriber.cli.load_speaker_diarizer",
return_value=diarizer,
),
patch("local_transcriber.cli.write_transcript", write),
):
out = runner.invoke(app, [str(audio), "--diarize"])
assert out.exit_code == 1
assert write.call_count == 1
assert "Полезный текст." in write.call_args.args[0]
assert "Диаризация завершилась с ошибкой: boom" in write.call_args.args[0]
def test_cli_verbose_reports_diarization_counts_and_duration(tmp_path):
audio = tmp_path / "meeting.mp3"
audio.write_bytes(b"fake")
result = _make_result(
segments=[Segment(0.0, 1.0, "Раз два")],
duration=10.0,
)
result.words = [Word(0.0, 0.5, "Раз"), Word(0.5, 1.0, "два")]
model = _make_model()
backend = _make_backend()
backend.word_timestamps_available = True
tfr = _make_tfr(result=result, model=model, backend=backend)
diarizer = MagicMock()
diarizer.process.return_value = DiarizationRun(
intervals=[
SpeakerInterval(0.0, 0.5, 1),
SpeakerInterval(0.5, 1.0, 2),
],
elapsed_seconds=0.2,
)
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch(
"local_transcriber.cli.load_speaker_diarizer",
return_value=diarizer,
),
patch("local_transcriber.cli.write_transcript"),
):
out = runner.invoke(app, [str(audio), "--diarize", "--verbose"])
assert out.exit_code == 0
assert "2 кластеров, 2 интервалов" in out.output
assert "0.2 с" in out.output
def test_cli_empty_asr_skips_diarizer_and_reports_it(tmp_path):
audio = tmp_path / "silence.wav"
audio.write_bytes(b"fake")
result = _make_result(segments=[])
model = _make_model()
backend = _make_backend()
backend.word_timestamps_available = True
tfr = _make_tfr(result=result, model=model, backend=backend)
diarizer = MagicMock()
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch(
"local_transcriber.cli.load_speaker_diarizer",
return_value=diarizer,
),
patch("local_transcriber.cli.write_transcript"),
):
out = runner.invoke(app, [str(audio), "--diarize"])
assert out.exit_code == 0
diarizer.process.assert_not_called()
assert "диаризация не запускалась" in out.output
def test_cli_diarizer_preflight_failure_does_not_start_asr_or_write(tmp_path):
audio = tmp_path / "meeting.mp3"
audio.write_bytes(b"fake")
model = _make_model()
backend = _make_backend()
backend.word_timestamps_available = True
transcribe_file = MagicMock()
write = MagicMock()
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch("local_transcriber.cli._transcribe_file", transcribe_file),
patch(
"local_transcriber.cli.load_speaker_diarizer",
side_effect=RuntimeError("модель повреждена"),
),
patch("local_transcriber.cli.write_transcript", write),
):
out = runner.invoke(app, [str(audio), "--diarize"])
assert out.exit_code == 1
transcribe_file.assert_not_called()
write.assert_not_called()
@pytest.mark.parametrize(
("intervals", "warning"),
[
([SpeakerInterval(0.0, 1.0, 1)], "только один голосовой кластер"),
([], "не нашёл интервалов"),
],
)
def test_cli_unsuccessful_diarization_shape_writes_plain_text_and_exits_nonzero(
tmp_path, intervals, warning
):
audio = tmp_path / "meeting.mp3"
audio.write_bytes(b"fake")
result = _make_result(
segments=[Segment(0.0, 1.0, "Раз два")],
duration=10.0,
)
result.words = [Word(0.0, 0.5, "Раз"), Word(0.5, 1.0, "два")]
model = _make_model()
backend = _make_backend()
backend.word_timestamps_available = True
tfr = _make_tfr(result=result, model=model, backend=backend)
diarizer = MagicMock()
diarizer.process.return_value = DiarizationRun(intervals, elapsed_seconds=0.1)
write = MagicMock()
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch(
"local_transcriber.cli.load_speaker_diarizer",
return_value=diarizer,
),
patch("local_transcriber.cli.write_transcript", write),
):
out = runner.invoke(app, [str(audio), "--diarize"])
assert out.exit_code == 1
content = write.call_args.args[0]
assert warning in content
assert "[00:00.00 - 00:01.00] Раз два" in content
def test_cli_rejects_nonpositive_speaker_count(tmp_path):
audio = tmp_path / "meeting.mp3"
audio.write_bytes(b"fake")
out = runner.invoke(app, [str(audio), "--speakers", "0"])
assert out.exit_code == 2
def test_cli_verbose_passes_on_segment_callback(tmp_path):
audio = tmp_path / "test.mp3"
audio.write_bytes(b"fake")
@@ -171,7 +464,10 @@ def test_cli_verbose_passes_on_segment_callback(tmp_path):
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu", backend, "/models/medium")),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch("local_transcriber.cli._transcribe_file", mock_transcribe_file),
patch("local_transcriber.cli.write_transcript"),
):
@@ -209,7 +505,10 @@ def test_cli_default_output_path(tmp_path):
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu", backend, "/models/medium")),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch("local_transcriber.cli.write_transcript", mock_write),
):
@@ -234,7 +533,10 @@ def test_cli_custom_output_path(tmp_path):
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu", backend, "/models/medium")),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch("local_transcriber.cli.write_transcript", mock_write),
):
@@ -257,7 +559,10 @@ def test_cli_passes_status_callback_to_transcribe(tmp_path):
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu", backend, "/models/medium")),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch("local_transcriber.cli._transcribe_file", mock_transcribe_file),
patch("local_transcriber.cli.write_transcript"),
):
@@ -276,7 +581,9 @@ def test_cli_load_model_called_with_model_name(tmp_path):
model = _make_model()
backend = _make_backend()
tfr = _make_tfr(result=result, model=model, backend=backend)
mock_load_model = MagicMock(return_value=(model, "cpu", backend, "/models/large-v3"))
mock_load_model = MagicMock(
return_value=(model, "cpu", backend, "/models/large-v3")
)
with (
patch("local_transcriber.cli.load_config", return_value={}),
@@ -302,8 +609,14 @@ def test_cli_windows_cuda_diagnostic(tmp_path):
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cuda"),
patch("local_transcriber.cli.load_model", return_value=(model, "cuda", backend, "/models/medium")),
patch("local_transcriber.cli._transcribe_file", side_effect=RuntimeError("CUDA error: no device")),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cuda", backend, "/models/medium"),
),
patch(
"local_transcriber.cli._transcribe_file",
side_effect=RuntimeError("CUDA error: no device"),
),
patch("local_transcriber.cli.sys") as mock_sys,
):
mock_sys.platform = "win32"
@@ -325,8 +638,14 @@ def test_cli_linux_cuda_error_no_windows_hint(tmp_path):
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cuda"),
patch("local_transcriber.cli.load_model", return_value=(model, "cuda", backend, "/models/medium")),
patch("local_transcriber.cli._transcribe_file", side_effect=RuntimeError("CUDA error: no device")),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cuda", backend, "/models/medium"),
),
patch(
"local_transcriber.cli._transcribe_file",
side_effect=RuntimeError("CUDA error: no device"),
),
patch("local_transcriber.cli.sys") as mock_sys,
):
mock_sys.platform = "linux"
@@ -349,7 +668,10 @@ def test_cli_device_fallback_warning(tmp_path):
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cuda"),
patch("local_transcriber.cli.load_model", return_value=(model, "cuda", backend, "/models/medium")),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cuda", backend, "/models/medium"),
),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch("local_transcriber.cli.write_transcript"),
):
@@ -372,7 +694,10 @@ def test_cli_strict_device_passed_to_transcribe(tmp_path):
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cuda"),
patch("local_transcriber.cli.load_model", return_value=(model, "cuda", backend, "/models/medium")),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cuda", backend, "/models/medium"),
),
patch("local_transcriber.cli._transcribe_file", mock_transcribe_file),
patch("local_transcriber.cli.write_transcript"),
patch("local_transcriber.cli.get_gpu_name", return_value="RTX 3060"),
@@ -390,7 +715,10 @@ def test_cli_strict_device_passed_to_transcribe(tmp_path):
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu", backend, "/models/medium")),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch("local_transcriber.cli._transcribe_file", mock_transcribe_file),
patch("local_transcriber.cli.write_transcript"),
):
@@ -410,7 +738,10 @@ def test_cli_keyboard_interrupt(tmp_path):
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu", backend, "/models/medium")),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch("local_transcriber.cli._transcribe_file", side_effect=KeyboardInterrupt),
patch("local_transcriber.cli.write_transcript"),
):
@@ -443,8 +774,14 @@ def test_cli_unexpected_error_verbose_traceback(tmp_path):
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu", backend, "/models/medium")),
patch("local_transcriber.cli._transcribe_file", side_effect=RuntimeError("unexpected boom")),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch(
"local_transcriber.cli._transcribe_file",
side_effect=RuntimeError("unexpected boom"),
),
patch("local_transcriber.cli.write_transcript"),
):
out = runner.invoke(app, [str(audio), "--verbose"])
@@ -464,8 +801,14 @@ def test_cli_unexpected_error_no_verbose_hint(tmp_path):
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu", backend, "/models/medium")),
patch("local_transcriber.cli._transcribe_file", side_effect=RuntimeError("unexpected boom")),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch(
"local_transcriber.cli._transcribe_file",
side_effect=RuntimeError("unexpected boom"),
),
patch("local_transcriber.cli.write_transcript"),
):
out = runner.invoke(app, [str(audio)])
@@ -493,7 +836,10 @@ def test_cli_batch_two_files(tmp_path):
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", side_effect=lambda p: p),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu", backend, "/models/medium")),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch("local_transcriber.cli.write_transcript"),
):
@@ -503,6 +849,111 @@ def test_cli_batch_two_files(tmp_path):
assert "2 обработано" in out.output
def test_cli_batch_reuses_one_diarizer_for_all_nonempty_files(tmp_path):
first = tmp_path / "first.mp3"
second = tmp_path / "second.mp3"
first.write_bytes(b"fake")
second.write_bytes(b"fake")
result = _make_result(
segments=[Segment(0.0, 1.0, "Раз два")],
duration=10.0,
)
result.words = [Word(0.0, 0.5, "Раз"), Word(0.5, 1.0, "два")]
model = _make_model()
backend = _make_backend()
backend.word_timestamps_available = True
tfr = _make_tfr(result=result, model=model, backend=backend)
diarizer = MagicMock()
diarizer.process.return_value = DiarizationRun(
intervals=[
SpeakerInterval(0.0, 0.5, 1),
SpeakerInterval(0.5, 1.0, 2),
],
elapsed_seconds=0.1,
)
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch(
"local_transcriber.cli.validate_input_file",
side_effect=lambda path: path,
),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch(
"local_transcriber.cli.load_speaker_diarizer",
return_value=diarizer,
) as load_diarizer,
patch("local_transcriber.cli.write_transcript") as write,
):
out = runner.invoke(app, [str(first), str(second), "--diarize"])
assert out.exit_code == 0
load_diarizer.assert_called_once()
assert [call.args[0] for call in diarizer.process.call_args_list] == [
first,
second,
]
assert write.call_count == 2
def test_cli_batch_continues_after_diarization_error_and_exits_nonzero(tmp_path):
first = tmp_path / "first.mp3"
second = tmp_path / "second.mp3"
first.write_bytes(b"fake")
second.write_bytes(b"fake")
result = _make_result(
segments=[Segment(0.0, 1.0, "Раз два")],
duration=10.0,
)
result.words = [Word(0.0, 0.5, "Раз"), Word(0.5, 1.0, "два")]
model = _make_model()
backend = _make_backend()
backend.word_timestamps_available = True
tfr = _make_tfr(result=result, model=model, backend=backend)
diarizer = MagicMock()
diarizer.process.side_effect = [
RuntimeError("boom"),
DiarizationRun(
[
SpeakerInterval(0.0, 0.5, 1),
SpeakerInterval(0.5, 1.0, 2),
],
elapsed_seconds=0.1,
),
]
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch(
"local_transcriber.cli.validate_input_file",
side_effect=lambda path: path,
),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch(
"local_transcriber.cli.load_speaker_diarizer",
return_value=diarizer,
),
patch("local_transcriber.cli.write_transcript") as write,
):
out = runner.invoke(app, [str(first), str(second), "--diarize"])
assert out.exit_code == 1
assert write.call_count == 2
assert "Диаризация завершилась с ошибкой: boom" in write.call_args_list[0].args[0]
assert "Speaker 1" in write.call_args_list[1].args[0]
assert "1 с деградацией" in out.output
def test_cli_batch_skips_existing(tmp_path):
a = tmp_path / "a.mp3"
b = tmp_path / "b.mp3"
@@ -519,7 +970,10 @@ def test_cli_batch_skips_existing(tmp_path):
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", side_effect=lambda p: p),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu", backend, "/models/medium")),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch("local_transcriber.cli.write_transcript"),
):
@@ -540,16 +994,22 @@ def test_cli_batch_all_skipped_no_model_load(tmp_path):
(tmp_path / "b-transcript.md").write_text("existing")
mock_load_model = MagicMock()
mock_load_diarizer = MagicMock()
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", side_effect=lambda p: p),
patch("local_transcriber.cli.load_model", mock_load_model),
patch(
"local_transcriber.cli.load_speaker_diarizer",
mock_load_diarizer,
),
):
out = runner.invoke(app, [str(a), str(b)])
out = runner.invoke(app, [str(a), str(b), "--diarize"])
assert out.exit_code == 0
mock_load_model.assert_not_called()
mock_load_diarizer.assert_not_called()
def test_cli_batch_force_overwrites(tmp_path):
@@ -568,7 +1028,10 @@ def test_cli_batch_force_overwrites(tmp_path):
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", side_effect=lambda p: p),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu", backend, "/models/medium")),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch("local_transcriber.cli.write_transcript"),
):
@@ -602,8 +1065,13 @@ def test_cli_batch_per_file_error(tmp_path):
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", side_effect=lambda p: p),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu", backend, "/models/medium")),
patch("local_transcriber.cli._transcribe_file", side_effect=transcribe_side_effect),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch(
"local_transcriber.cli._transcribe_file", side_effect=transcribe_side_effect
),
patch("local_transcriber.cli.write_transcript"),
):
out = runner.invoke(app, [str(a), str(b)])
@@ -631,9 +1099,15 @@ def test_cli_batch_invalid_in_prescan(tmp_path):
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", side_effect=validate_side_effect),
patch(
"local_transcriber.cli.validate_input_file",
side_effect=validate_side_effect,
),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu", backend, "/models/medium")),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch("local_transcriber.cli.write_transcript"),
):
@@ -768,7 +1242,10 @@ def test_cli_batch_fallback_warning(tmp_path):
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", side_effect=lambda p: p),
patch("local_transcriber.cli.detect_device", return_value="cuda"),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu", backend, "/models/medium")),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch("local_transcriber.cli.write_transcript"),
):
@@ -795,8 +1272,13 @@ def test_cli_batch_empty_speech_warning(tmp_path):
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", side_effect=lambda p: p),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu", backend, "/models/medium")),
patch("local_transcriber.cli._transcribe_file", side_effect=[tfr_empty, tfr_ok]),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch(
"local_transcriber.cli._transcribe_file", side_effect=[tfr_empty, tfr_ok]
),
patch("local_transcriber.cli.write_transcript"),
):
out = runner.invoke(app, [str(a), str(b)])
@@ -817,15 +1299,24 @@ def test_cli_batch_midstream_fallback_warning(tmp_path):
model_cpu = _make_model()
backend = _make_backend()
result = _make_result(device_used="cpu")
tfr_fallback = _make_tfr(result=result, model=model_cpu, actual_device="cpu", backend=backend)
tfr_ok = _make_tfr(result=result, model=model_cpu, actual_device="cpu", backend=backend)
tfr_fallback = _make_tfr(
result=result, model=model_cpu, actual_device="cpu", backend=backend
)
tfr_ok = _make_tfr(
result=result, model=model_cpu, actual_device="cpu", backend=backend
)
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", side_effect=lambda p: p),
patch("local_transcriber.cli.detect_device", return_value="cuda"),
patch("local_transcriber.cli.load_model", return_value=(model_gpu, "cuda", backend, "/models/medium")),
patch("local_transcriber.cli._transcribe_file", side_effect=[tfr_fallback, tfr_ok]),
patch(
"local_transcriber.cli.load_model",
return_value=(model_gpu, "cuda", backend, "/models/medium"),
),
patch(
"local_transcriber.cli._transcribe_file", side_effect=[tfr_fallback, tfr_ok]
),
patch("local_transcriber.cli.write_transcript"),
):
out = runner.invoke(app, [str(a), str(b)])
@@ -864,8 +1355,14 @@ def test_cli_batch_model_loaded_once(tmp_path):
def test_format_device_info_openvino_gpu():
with patch("local_transcriber.cli.get_intel_gpu_name", return_value="Intel(R) Arc(TM) 140T GPU"):
assert _format_device_info("openvino-gpu") == "OpenVINO (Intel(R) Arc(TM) 140T GPU)"
with patch(
"local_transcriber.cli.get_intel_gpu_name",
return_value="Intel(R) Arc(TM) 140T GPU",
):
assert (
_format_device_info("openvino-gpu")
== "OpenVINO (Intel(R) Arc(TM) 140T GPU)"
)
def test_format_device_info_openvino_gpu_no_name():
@@ -900,9 +1397,13 @@ def test_cli_openvino_gpu_happy_path(tmp_path):
audio.write_bytes(b"fake")
result = _make_result(device_used="openvino-gpu")
patches = _single_patches(result=result, tmp_file=audio, actual_device="openvino-gpu")
patches = _single_patches(
result=result, tmp_file=audio, actual_device="openvino-gpu"
)
with patches[0], patches[1], patches[2], patches[3], patches[4], patches[5]:
with patch("local_transcriber.cli.get_intel_gpu_name", return_value="Intel Arc 140T"):
with patch(
"local_transcriber.cli.get_intel_gpu_name", return_value="Intel Arc 140T"
):
out = runner.invoke(app, [str(audio), "--device", "openvino-gpu"])
assert out.exit_code == 0
@@ -915,8 +1416,12 @@ def test_cli_openvino_alias_resolves_to_gpu(tmp_path):
result = _make_result(device_used="openvino-gpu")
model = _make_model()
backend = _make_backend()
tfr = _make_tfr(result=result, model=model, actual_device="openvino-gpu", backend=backend)
mock_load_model = MagicMock(return_value=(model, "openvino-gpu", backend, "/models/medium"))
tfr = _make_tfr(
result=result, model=model, actual_device="openvino-gpu", backend=backend
)
mock_load_model = MagicMock(
return_value=(model, "openvino-gpu", backend, "/models/medium")
)
with (
patch("local_transcriber.cli.load_config", return_value={}),
@@ -925,7 +1430,9 @@ def test_cli_openvino_alias_resolves_to_gpu(tmp_path):
patch("local_transcriber.cli.load_model", mock_load_model),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch("local_transcriber.cli.write_transcript"),
patch("local_transcriber.cli.get_intel_gpu_name", return_value="Intel Arc 140T"),
patch(
"local_transcriber.cli.get_intel_gpu_name", return_value="Intel Arc 140T"
),
):
out = runner.invoke(app, [str(audio), "--device", "openvino"])
@@ -1000,7 +1507,9 @@ def test_cli_install_menu_success(tmp_path):
cmd_path = tmp_path / "Transcribe.cmd"
with (
patch("local_transcriber.cli.install_context_menu", return_value=cmd_path) as mock_install,
patch(
"local_transcriber.cli.install_context_menu", return_value=cmd_path
) as mock_install,
patch("local_transcriber.cli.load_config") as mock_load_config,
patch("local_transcriber.cli.sys") as mock_sys,
):
@@ -1018,7 +1527,9 @@ def test_cli_uninstall_menu_success(tmp_path):
cmd_path = tmp_path / "Transcribe.cmd"
with (
patch("local_transcriber.cli.uninstall_context_menu", return_value=cmd_path) as mock_uninstall,
patch(
"local_transcriber.cli.uninstall_context_menu", return_value=cmd_path
) as mock_uninstall,
patch("local_transcriber.cli.load_config") as mock_load_config,
patch("local_transcriber.cli.sys") as mock_sys,
):
@@ -1079,7 +1590,10 @@ def test_cli_menu_flags_available_only_on_windows():
def test_cli_menu_runtime_error_has_no_verbose_hint():
with (
patch("local_transcriber.cli.install_context_menu", side_effect=RuntimeError("нет APPDATA")),
patch(
"local_transcriber.cli.install_context_menu",
side_effect=RuntimeError("нет APPDATA"),
),
patch("local_transcriber.cli.sys") as mock_sys,
):
mock_sys.platform = "win32"
@@ -1169,7 +1683,10 @@ def test_cli_quality_warning_batch_includes_file_name(tmp_path):
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.validate_input_file", side_effect=lambda p: p),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu", backend, "/models/medium")),
patch(
"local_transcriber.cli.load_model",
return_value=(model, "cpu", backend, "/models/medium"),
),
patch("local_transcriber.cli._transcribe_file", side_effect=[tfr_warn, tfr_ok]),
patch("local_transcriber.cli.write_transcript"),
patch("local_transcriber.cli.console", Console(stderr=True, width=1000)),
@@ -1178,8 +1695,7 @@ def test_cli_quality_warning_batch_includes_file_name(tmp_path):
assert out.exit_code == 0
assert (
" a.mp3: транскрипт покрывает 01:00 из 10:00 — "
"возможна потеря хвоста записи"
" a.mp3: транскрипт покрывает 01:00 из 10:00 — возможна потеря хвоста записи"
) in out.output
tests/test_context_menu.py
+1 -1
View File
@@ -125,4 +125,4 @@ def test_get_transcribe_exe_requires_existing_exe(tmp_path, monkeypatch):
monkeypatch.setattr(context_menu.sys, "executable", str(python_exe))
with pytest.raises(RuntimeError, match="uv sync"):
context_menu.get_transcribe_exe()
context_menu.get_transcribe_exe()
tests/test_diarization.py
+120
View File
@@ -0,0 +1,120 @@
from local_transcriber.diarization import build_speaker_transcript
from local_transcriber.types import (
SmallSpeakerCluster,
SpeakerInterval,
SpeakerTurn,
Word,
)
def test_build_speaker_transcript_assigns_and_groups_words():
words = [
Word(start=0.0, end=0.8, text="Добрый"),
Word(start=0.8, end=1.4, text="день."),
Word(start=1.5, end=2.1, text="Привет!"),
]
intervals = [
SpeakerInterval(start=0.0, end=1.4, cluster=7),
SpeakerInterval(start=1.4, end=2.3, cluster=3),
]
transcript = build_speaker_transcript(words, intervals, recording_duration=30.0)
assert [
(turn.speaker, turn.start, turn.end, turn.text) for turn in transcript.turns
] == [
(1, 0.0, 1.4, "Добрый день."),
(2, 1.5, 2.1, "Привет!"),
]
assert transcript.cluster_count == 2
assert transcript.unassigned_word_count == 0
def test_build_speaker_transcript_reports_small_cluster_without_filtering_it():
words = [
Word(start=0.0, end=1.0, text="Редкая реплика."),
Word(start=5.0, end=6.0, text="Основная реплика."),
]
intervals = [
SpeakerInterval(start=0.0, end=4.9, cluster=4),
SpeakerInterval(start=5.0, end=10.0, cluster=9),
]
transcript = build_speaker_transcript(words, intervals, recording_duration=100.0)
assert [turn.speaker for turn in transcript.turns] == [1, 2]
assert transcript.small_clusters == [SmallSpeakerCluster(speaker=1, duration=4.9)]
def test_build_speaker_transcript_keeps_equal_overlap_unassigned():
words = [Word(start=0.0, end=1.0, text="Спорное слово")]
intervals = [
SpeakerInterval(start=0.0, end=0.1, cluster=8),
SpeakerInterval(start=0.3, end=0.5, cluster=8),
SpeakerInterval(start=0.0, end=0.3, cluster=2),
]
transcript = build_speaker_transcript(words, intervals, recording_duration=10.0)
assert transcript.turns[0].speaker is None
assert transcript.unassigned_word_count == 1
def test_build_speaker_transcript_keeps_word_without_overlap_unknown():
transcript = build_speaker_transcript(
[Word(start=5.0, end=6.0, text="Вне разметки")],
[SpeakerInterval(start=0.0, end=1.0, cluster=1)],
recording_duration=10.0,
)
assert transcript.turns == [SpeakerTurn(5.0, 6.0, "Вне разметки", None)]
assert transcript.unassigned_word_count == 1
def test_build_speaker_transcript_splits_at_two_second_pause():
transcript = build_speaker_transcript(
[
Word(0.0, 1.0, "До паузы."),
Word(3.0, 4.0, "После паузы."),
],
[SpeakerInterval(0.0, 4.0, 1)],
recording_duration=10.0,
)
assert [turn.text for turn in transcript.turns] == [
"До паузы.",
"После паузы.",
]
def test_build_speaker_transcript_does_not_exceed_sixty_seconds():
transcript = build_speaker_transcript(
[
Word(0.0, 30.0, "Начало."),
Word(30.0, 60.0, "Продолжение."),
Word(60.0, 61.0, "Новая реплика."),
],
[SpeakerInterval(0.0, 61.0, 1)],
recording_duration=70.0,
)
assert [turn.text for turn in transcript.turns] == [
"Начало. Продолжение.",
"Новая реплика.",
]
def test_build_speaker_transcript_preserves_punctuation_without_leading_space():
transcript = build_speaker_transcript(
[
Word(0.0, 0.4, "Тарадата"),
Word(0.4, 0.5, "+"),
Word(0.5, 0.7, "Click"),
Word(0.7, 0.8, "—"),
Word(0.8, 1.0, "это"),
],
[SpeakerInterval(0.0, 1.0, 1)],
recording_duration=10.0,
)
assert transcript.turns[0].text == "Тарадата+ Click— это"
tests/test_formatter.py
+118 -2
View File
@@ -1,5 +1,4 @@
from datetime import datetime
from pathlib import Path
from local_transcriber.formatter import (
LANGUAGE_DETECTED,
@@ -11,7 +10,12 @@ from local_transcriber.formatter import (
write_transcript,
)
from local_transcriber.transcriber import Segment, TranscribeResult
from local_transcriber.types import UNKNOWN_LANGUAGE
from local_transcriber.types import (
UNKNOWN_LANGUAGE,
SmallSpeakerCluster,
SpeakerTranscript,
SpeakerTurn,
)
def test_format_timestamp_minutes():
@@ -59,6 +63,118 @@ def test_format_transcript_basic():
assert "[00:00.00 - 00:09.15] Добрый день, коллеги. Первый вопрос." in content
def test_format_transcript_speaker_turns_use_truncated_start_timestamps():
result = TranscribeResult(
segments=[Segment(start=547.96, end=560.0, text=" Обычный текст")],
language="ru",
language_probability=0.97,
duration=700.0,
device_used="cpu",
)
speakers = SpeakerTranscript(
turns=[
SpeakerTurn(547.96, 550.0, "Первая реплика.", 1),
SpeakerTurn(558.4, 560.0, "Ответ.", 2),
],
cluster_count=2,
unassigned_word_count=0,
small_clusters=[],
)
content = format_transcript(
result,
source_filename="meeting.mp4",
model_name="medium",
device_info="CPU",
language_mode=LANGUAGE_DETECTED,
speaker_transcript=speakers,
)
assert "- **Голосовых кластеров**: 2" in content
assert "[09:07] Speaker 1: Первая реплика." in content
assert "[09:18] Speaker 2: Ответ." in content
assert "[09:07.96 -" not in content
def test_format_transcript_speaker_turns_use_hours_after_one_hour():
result = TranscribeResult(
segments=[Segment(start=3661.9, end=3663.0, text=" Длинная встреча")],
language="ru",
language_probability=1.0,
duration=3700.0,
device_used="cpu",
)
speakers = SpeakerTranscript(
turns=[SpeakerTurn(3661.9, 3663.0, "Длинная встреча", 1)],
cluster_count=2,
unassigned_word_count=0,
small_clusters=[],
)
content = format_transcript(
result,
source_filename="meeting.mp4",
model_name="medium",
device_info="CPU",
language_mode=LANGUAGE_FORCED,
speaker_transcript=speakers,
)
assert "[01:01:01] Speaker 1: Длинная встреча" in content
def test_format_transcript_reports_unknown_words_and_small_clusters():
result = TranscribeResult(
segments=[Segment(start=0.0, end=8.0, text=" Текст")],
language="ru",
language_probability=1.0,
duration=20.0,
device_used="cpu",
)
speakers = SpeakerTranscript(
turns=[SpeakerTurn(1.2, 2.0, "Неясная реплика.", None)],
cluster_count=2,
unassigned_word_count=3,
small_clusters=[SmallSpeakerCluster(speaker=2, duration=4.2)],
)
content = format_transcript(
result,
source_filename="meeting.mp4",
model_name="medium",
device_info="CPU",
language_mode=LANGUAGE_FORCED,
speaker_transcript=speakers,
)
assert "[00:01] Speaker ?: Неясная реплика." in content
assert "3 слов без назначенного говорящего" in content
assert "малый кластер Speaker 2: 4.2 с" in content
def test_format_transcript_keeps_plain_body_with_diarization_warning():
result = TranscribeResult(
segments=[Segment(start=0.0, end=2.0, text=" Полезный текст.")],
language="ru",
language_probability=1.0,
duration=5.0,
device_used="cpu",
)
content = format_transcript(
result,
source_filename="meeting.mp4",
model_name="medium",
device_info="CPU",
language_mode=LANGUAGE_FORCED,
diarization_warning="Диаризация завершилась с ошибкой: boom",
)
assert "**Внимание**: Диаризация завершилась с ошибкой: boom" in content
assert "[00:00.00 - 00:02.00] Полезный текст." in content
assert "Speaker" not in content
def test_format_transcript_unknown_language_without_placeholder():
"""Неизвестный язык печатается одной строкой, без служебного значения."""
result = TranscribeResult(
tests/test_onnx_asr.py
+166 -9
View File
@@ -5,16 +5,18 @@ import warnings
import pytest
from local_transcriber.backends.onnx_asr import OnnxAsrBackend
from local_transcriber.types import UNKNOWN_LANGUAGE, Segment, TranscribeResult
from local_transcriber.types import UNKNOWN_LANGUAGE, Segment, TranscribeResult, Word
class FakeVadSegment:
"""Mimics onnx-asr SegmentResult."""
def __init__(self, start, end, text):
def __init__(self, start, end, text, tokens=None, timestamps=None):
self.start = start
self.end = end
self.text = text
self.tokens = [f" {text}"] if tokens is None else tokens
self.timestamps = [0.0] if timestamps is None else timestamps
class TestEnsureModelAvailable:
@@ -59,6 +61,9 @@ class TestEnsureModelAvailable:
def with_vad(self, vad):
return self
def with_timestamps(self):
return self
def fake_load_model(*, model, quantization):
quantizations.append(quantization)
return FakeAsrAdapter()
@@ -82,21 +87,45 @@ class TestEnsureModelAvailable:
class TestCreateModel:
def test_wraps_vad_model_with_timestamps(self, monkeypatch):
timestamped_model = object()
class FakeVadAdapter:
def with_timestamps(self):
return timestamped_model
class FakeAsrAdapter:
def with_vad(self, vad):
return FakeVadAdapter()
monkeypatch.setattr("onnx_asr.load_model", lambda **kwargs: FakeAsrAdapter())
monkeypatch.setattr("onnx_asr.load_vad", lambda model: object())
model = OnnxAsrBackend().create_model("gigaam-v3-e2e-rnnt", "onnx", "int8")
assert model is timestamped_model
def test_calls_load_model_with_correct_args(self, monkeypatch):
"""Verify create_model passes correct args to onnx_asr.load_model."""
calls = []
def fake_load_model(model=None, path=None, quantization=None,
**kwargs):
calls.append({
"model": model, "path": path, "quantization": quantization,
})
def fake_load_model(model=None, path=None, quantization=None, **kwargs):
calls.append(
{
"model": model,
"path": path,
"quantization": quantization,
}
)
return FakeAsrAdapter()
class FakeAsrAdapter:
def with_vad(self, vad):
return self
def with_timestamps(self):
return self
monkeypatch.setattr("onnx_asr.load_model", fake_load_model)
backend = OnnxAsrBackend()
@@ -123,6 +152,9 @@ class TestCreateModel:
self._vad = vad
return self
def with_timestamps(self):
return self
monkeypatch.setattr("onnx_asr.load_model", fake_load_model)
monkeypatch.setattr("onnx_asr.load_vad", fake_load_vad)
@@ -143,6 +175,9 @@ class TestCreateModel:
def with_vad(self, vad):
return self
def with_timestamps(self):
return self
monkeypatch.setattr("onnx_asr.load_model", fake_load_model)
monkeypatch.setattr("onnx_asr.load_vad", lambda model, **kw: None)
@@ -167,6 +202,9 @@ class TestCreateModel:
def with_vad(self, vad):
return self
def with_timestamps(self):
return self
monkeypatch.setattr("onnx_asr.load_model", fake_load_model)
monkeypatch.setattr("onnx_asr.load_vad", lambda model, **kw: None)
@@ -187,6 +225,9 @@ class TestCreateModel:
def with_vad(self, vad):
return self
def with_timestamps(self):
return self
monkeypatch.setattr("onnx_asr.load_model", fake_load_model)
monkeypatch.setattr("onnx_asr.load_vad", lambda model, **kw: None)
@@ -207,6 +248,9 @@ class TestCreateModel:
def with_vad(self, vad):
return self
def with_timestamps(self):
return self
monkeypatch.setattr("onnx_asr.load_model", fake_load_model)
monkeypatch.setattr("onnx_asr.load_vad", lambda model, **kw: None)
@@ -298,6 +342,7 @@ class TestTranscribe:
def fake_decode_audio(path, sampling_rate=16000):
import numpy as np
return np.array(audio_samples, dtype=np.float32)
class FakeModel:
@@ -310,7 +355,9 @@ class TestTranscribe:
backend = OnnxAsrBackend()
backend.actual_compute_type = "int8"
result = backend.transcribe(
FakeModel(), wav_file, language=None,
FakeModel(),
wav_file,
language=None,
)
assert isinstance(result, TranscribeResult)
@@ -319,6 +366,110 @@ class TestTranscribe:
assert result.segments[1] == Segment(start=1.0, end=2.5, text="world")
assert result.duration == 1.0 # 16000 samples / 16000 Hz
def test_transcribe_converts_vad_token_timestamps_to_global_words(
self, monkeypatch, tmp_path
):
wav_file = tmp_path / "test.wav"
wav_file.write_bytes(b"fake audio")
monkeypatch.setattr(
"faster_whisper.decode_audio",
lambda path, sampling_rate=16000: [0.0] * 16_000,
)
timestamped_segment = FakeVadSegment(
10.0,
12.0,
"Привет, мир",
tokens=[" ", "П", "р", "и", "в", "е", "т", ",", " ", "м", "и", "р"],
timestamps=[0.0, 0.1, 0.1, 0.1, 0.2, 0.2, 0.3, 0.3, 0.5, 0.6, 0.6, 0.7],
)
class FakeModel:
def recognize(self, waveform, sample_rate, language=None):
yield timestamped_segment
result = OnnxAsrBackend().transcribe(FakeModel(), wav_file, language="ru")
assert result.words == [
Word(start=10.0, end=10.5, text=" Привет,"),
Word(start=10.5, end=12.0, text=" мир"),
]
assert "".join(word.text for word in result.words).strip() == "Привет, мир"
def test_transcribe_rejects_nonempty_segment_without_token_timestamps(
self, monkeypatch, tmp_path
):
wav_file = tmp_path / "test.wav"
wav_file.write_bytes(b"fake audio")
monkeypatch.setattr(
"faster_whisper.decode_audio",
lambda path, sampling_rate=16000: [0.0] * 16_000,
)
segment = FakeVadSegment(0.0, 1.0, "Текст")
segment.tokens = None
segment.timestamps = None
class FakeModel:
def recognize(self, waveform, sample_rate, language=None):
yield segment
with pytest.raises(RuntimeError, match="пословные таймкоды"):
OnnxAsrBackend().transcribe(FakeModel(), wav_file, language="ru")
def test_transcribe_keeps_words_with_equal_emission_timestamps(
self, monkeypatch, tmp_path
):
wav_file = tmp_path / "test.wav"
wav_file.write_bytes(b"fake audio")
monkeypatch.setattr(
"faster_whisper.decode_audio",
lambda path, sampling_rate=16000: [0.0] * 16_000,
)
segment = FakeVadSegment(
10.0,
12.0,
"Да нет потом",
tokens=[" ", "Да", " ", "нет", " ", "потом"],
timestamps=[0.0, 0.0, 0.0, 0.0, 0.5, 0.5],
)
class FakeModel:
def recognize(self, waveform, sample_rate, language=None):
yield segment
result = OnnxAsrBackend().transcribe(FakeModel(), wav_file, language="ru")
assert [word.text for word in result.words] == [" Да", " нет", " потом"]
assert [(word.start, word.end) for word in result.words] == [
(10.0, 10.5),
(10.0, 10.5),
(10.5, 12.0),
]
assert "".join(word.text for word in result.words).strip() == segment.text
def test_transcribe_keeps_word_clamped_to_segment_end(self, monkeypatch, tmp_path):
wav_file = tmp_path / "test.wav"
wav_file.write_bytes(b"fake audio")
monkeypatch.setattr(
"faster_whisper.decode_audio",
lambda path, sampling_rate=16000: [0.0] * 16_000,
)
segment = FakeVadSegment(
10.0,
12.0,
"Позднее",
tokens=[" ", "Позднее"],
timestamps=[2.0, 2.0],
)
class FakeModel:
def recognize(self, waveform, sample_rate, language=None):
yield segment
result = OnnxAsrBackend().transcribe(FakeModel(), wav_file, language="ru")
assert result.words == [Word(start=12.0, end=12.0, text=" Позднее")]
def test_transcribe_calls_on_segment(self, monkeypatch, tmp_path):
"""Verify on_segment callback is invoked per segment."""
wav_file = tmp_path / "test.wav"
@@ -326,6 +477,7 @@ class TestTranscribe:
def fake_decode_audio(path, sampling_rate=16000):
import numpy as np
return np.array([0.0] * 16000, dtype=np.float32)
segments_captured = []
@@ -339,7 +491,9 @@ class TestTranscribe:
backend = OnnxAsrBackend()
backend.transcribe(
FakeModel(), wav_file, language=None,
FakeModel(),
wav_file,
language=None,
on_segment=lambda s: segments_captured.append(s),
)
@@ -354,6 +508,7 @@ class TestTranscribe:
def fake_decode_audio(path, sampling_rate=16000):
import numpy as np
return np.array([0.0] * 16000, dtype=np.float32)
lang_received = []
@@ -377,6 +532,7 @@ class TestTranscribe:
def fake_decode_audio(path, sampling_rate=16000):
import numpy as np
return np.array([0.0] * 16000, dtype=np.float32)
class FakeModel:
@@ -419,6 +575,7 @@ class TestTranscribe:
class TestBackendRegistration:
def test_get_backend_returns_onnx_backend(self):
from local_transcriber.backends import get_backend
backend = get_backend("onnx")
assert isinstance(backend, OnnxAsrBackend)
tests/test_speaker_diarizer.py
+231
View File
@@ -0,0 +1,231 @@
import hashlib
import io
import sys
import tarfile
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import numpy as np
import pytest
from local_transcriber.speaker_diarizer import SpeakerDiarizer, load_speaker_diarizer
from local_transcriber.types import SpeakerInterval
def test_process_returns_sorted_domain_intervals(tmp_path):
audio = tmp_path / "meeting.mp3"
raw_result = MagicMock()
raw_result.sort_by_start_time.return_value = [
SimpleNamespace(start=0.2, end=1.1, speaker=4),
SimpleNamespace(start=1.3, end=2.0, speaker=2),
]
engine = MagicMock()
engine.process.return_value = raw_result
diarizer = SpeakerDiarizer(engine)
samples = np.zeros(16_000, dtype=np.float32)
with patch("faster_whisper.decode_audio", return_value=samples) as decode:
run = diarizer.process(audio)
assert run.intervals == [
SpeakerInterval(start=0.2, end=1.1, cluster=4),
SpeakerInterval(start=1.3, end=2.0, cluster=2),
]
decode.assert_called_once_with(str(audio), sampling_rate=16_000)
engine.process.assert_called_once_with(samples)
def test_process_reports_engine_progress(tmp_path):
raw_result = MagicMock()
raw_result.sort_by_start_time.return_value = []
engine = MagicMock()
def process(samples, callback):
assert callback(2, 4) == 0
return raw_result
engine.process.side_effect = process
statuses = []
with patch(
"faster_whisper.decode_audio",
return_value=np.zeros(16_000, dtype=np.float32),
):
SpeakerDiarizer(engine).process(
tmp_path / "meeting.mp3",
on_status=statuses.append,
)
assert "Определяю говорящих... 2 / 4" in statuses
def test_load_speaker_diarizer_uses_verified_cache_and_calibrated_config(
tmp_path, monkeypatch
):
segmentation = tmp_path / "pyannote-segmentation-3.0.onnx"
embedding = tmp_path / "wespeaker_en_voxceleb_resnet34_LM.onnx"
segmentation.write_bytes(b"segmentation")
embedding.write_bytes(b"embedding")
monkeypatch.setattr(
"local_transcriber.speaker_diarizer._SEGMENTATION_SHA256",
hashlib.sha256(segmentation.read_bytes()).hexdigest(),
)
monkeypatch.setattr(
"local_transcriber.speaker_diarizer._EMBEDDING_SHA256",
hashlib.sha256(embedding.read_bytes()).hexdigest(),
)
captured = {}
def config_factory(**kwargs):
config = SimpleNamespace(**kwargs, validate=lambda: True)
captured["config"] = config
return config
engine = SimpleNamespace(sample_rate=16_000)
sherpa = SimpleNamespace(
OfflineSpeakerSegmentationPyannoteModelConfig=lambda **kwargs: SimpleNamespace(
**kwargs
),
OfflineSpeakerSegmentationModelConfig=lambda **kwargs: SimpleNamespace(
**kwargs
),
SpeakerEmbeddingExtractorConfig=lambda **kwargs: SimpleNamespace(**kwargs),
FastClusteringConfig=lambda **kwargs: SimpleNamespace(**kwargs),
OfflineSpeakerDiarizationConfig=config_factory,
OfflineSpeakerDiarization=lambda config: engine,
)
with (
patch("huggingface_hub.cached_assets_path", return_value=tmp_path),
patch.dict(sys.modules, {"sherpa_onnx": sherpa}),
patch("httpx.stream", side_effect=AssertionError("network is not expected")),
):
diarizer = load_speaker_diarizer(speakers=None, threads=0)
config = captured["config"]
assert config.clustering.num_clusters == -1
assert config.clustering.threshold == 0.89
assert config.min_duration_on == 0.3
assert config.min_duration_off == 0.5
assert not hasattr(config.segmentation, "num_threads")
assert not hasattr(config.embedding, "num_threads")
assert isinstance(diarizer, SpeakerDiarizer)
def test_load_speaker_diarizer_downloads_and_verifies_missing_models(
tmp_path, monkeypatch
):
segmentation_bytes = b"downloaded segmentation"
embedding_bytes = b"downloaded embedding"
archive_buffer = io.BytesIO()
with tarfile.open(fileobj=archive_buffer, mode="w:bz2") as archive:
member = tarfile.TarInfo("sherpa-onnx-pyannote-segmentation-3-0/model.onnx")
member.size = len(segmentation_bytes)
archive.addfile(member, io.BytesIO(segmentation_bytes))
monkeypatch.setattr(
"local_transcriber.speaker_diarizer._SEGMENTATION_SHA256",
hashlib.sha256(segmentation_bytes).hexdigest(),
)
monkeypatch.setattr(
"local_transcriber.speaker_diarizer._EMBEDDING_SHA256",
hashlib.sha256(embedding_bytes).hexdigest(),
)
class FakeResponse:
def __init__(self, content):
self.content = content
def __enter__(self):
return self
def __exit__(self, *args):
return False
def raise_for_status(self):
return None
def iter_bytes(self):
yield self.content
requested_urls = []
def fake_stream(method, url, **kwargs):
requested_urls.append(url)
content = (
archive_buffer.getvalue() if "segmentation" in url else embedding_bytes
)
return FakeResponse(content)
config = SimpleNamespace(validate=lambda: True)
engine = SimpleNamespace(sample_rate=16_000)
sherpa = SimpleNamespace(
OfflineSpeakerSegmentationPyannoteModelConfig=lambda **kwargs: SimpleNamespace(
**kwargs
),
OfflineSpeakerSegmentationModelConfig=lambda **kwargs: SimpleNamespace(
**kwargs
),
SpeakerEmbeddingExtractorConfig=lambda **kwargs: SimpleNamespace(**kwargs),
FastClusteringConfig=lambda **kwargs: SimpleNamespace(**kwargs),
OfflineSpeakerDiarizationConfig=lambda **kwargs: config,
OfflineSpeakerDiarization=lambda actual_config: engine,
)
with (
patch("huggingface_hub.cached_assets_path", return_value=tmp_path),
patch.dict(sys.modules, {"sherpa_onnx": sherpa}),
patch("httpx.stream", side_effect=fake_stream),
):
load_speaker_diarizer(speakers=2, threads=4)
assert (
tmp_path / "pyannote-segmentation-3.0.onnx"
).read_bytes() == segmentation_bytes
assert (
tmp_path / "wespeaker_en_voxceleb_resnet34_LM.onnx"
).read_bytes() == embedding_bytes
assert len(requested_urls) == 2
assert list(tmp_path.glob("*.tmp")) == []
def test_load_speaker_diarizer_keeps_corrupt_cache_when_download_is_invalid(
tmp_path, monkeypatch
):
segmentation = tmp_path / "pyannote-segmentation-3.0.onnx"
segmentation.write_bytes(b"existing corrupt model")
monkeypatch.setattr(
"local_transcriber.speaker_diarizer._SEGMENTATION_SHA256",
hashlib.sha256(b"expected model").hexdigest(),
)
archive_buffer = io.BytesIO()
with tarfile.open(fileobj=archive_buffer, mode="w:bz2") as archive:
payload = b"wrong downloaded model"
member = tarfile.TarInfo("sherpa-onnx-pyannote-segmentation-3-0/model.onnx")
member.size = len(payload)
archive.addfile(member, io.BytesIO(payload))
class FakeResponse:
def __enter__(self):
return self
def __exit__(self, *args):
return False
def raise_for_status(self):
return None
def iter_bytes(self):
yield archive_buffer.getvalue()
with (
patch("huggingface_hub.cached_assets_path", return_value=tmp_path),
patch("httpx.stream", return_value=FakeResponse()),
pytest.raises(RuntimeError, match="Контрольная сумма"),
):
load_speaker_diarizer(speakers=None)
assert segmentation.read_bytes() == b"existing corrupt model"
assert list(tmp_path.glob("*.tmp")) == []
tests/test_transcriber.py
+38 -6
View File
@@ -11,7 +11,7 @@ from local_transcriber.transcriber import (
load_model,
transcribe,
)
from local_transcriber.types import WordTimestampsUnavailableError
# === Helpers ===
@@ -314,7 +314,9 @@ def test_load_model_returns_backend_and_path(mock_get_backend):
backend = _make_backend(model_path="/mock/model/path")
mock_get_backend.return_value = backend
model, actual_device, returned_backend, model_path = load_model("tiny", "cpu", "int8")
model, actual_device, returned_backend, model_path = load_model(
"tiny", "cpu", "int8"
)
assert returned_backend is backend
assert model_path == "/mock/model/path"
@@ -389,7 +391,9 @@ def test_ensure_model_available_uses_cache_first(mock_snapshot_download, tmp_pat
@patch("local_transcriber.backends.faster_whisper._validate_model_dir")
@patch("local_transcriber.backends.faster_whisper.snapshot_download")
def test_ensure_model_available_downloads_on_cache_miss(mock_snapshot_download, mock_validate_model_dir):
def test_ensure_model_available_downloads_on_cache_miss(
mock_snapshot_download, mock_validate_model_dir
):
from huggingface_hub.errors import LocalEntryNotFoundError
mock_snapshot_download.side_effect = [
@@ -433,7 +437,9 @@ def test_ensure_model_available_rejects_unsupported_alias():
@patch("local_transcriber.backends.faster_whisper.snapshot_download")
def test_ensure_model_available_redownloads_incomplete_cache(mock_snapshot_download, tmp_path):
def test_ensure_model_available_redownloads_incomplete_cache(
mock_snapshot_download, tmp_path
):
incomplete = tmp_path / "incomplete"
incomplete.mkdir()
(incomplete / "config.json").write_text("{}")
@@ -489,7 +495,9 @@ def test_load_model_openvino_gpu_fallback_to_cpu(mock_get_backend):
with pytest.warns(UserWarning, match="Переключение на CPU"):
model, actual_device, backend, model_path = load_model(
"medium", "openvino-gpu", "fp16",
"medium",
"openvino-gpu",
"fp16",
)
assert actual_device == "cpu"
@@ -514,7 +522,9 @@ def test_load_model_openvino_cpu_fallback_to_cpu(mock_get_backend):
with pytest.warns(UserWarning, match="Переключение на CPU"):
model, actual_device, backend, model_path = load_model(
"medium", "openvino-cpu", "int8",
"medium",
"openvino-cpu",
"int8",
)
assert actual_device == "cpu"
@@ -555,6 +565,28 @@ def test_transcribe_file_openvino_gpu_midstream_fallback(mock_get_backend):
assert tfr.model_path == "/mock/cpu/model"
@patch("local_transcriber.transcriber.get_backend")
def test_transcribe_file_does_not_fallback_for_missing_word_timestamps(
mock_get_backend,
):
ov_backend = _make_backend(
transcribe_error=WordTimestampsUnavailableError("нет таймкодов"),
)
with pytest.raises(WordTimestampsUnavailableError, match="нет таймкодов"):
_transcribe_file(
model=MagicMock(),
actual_device="openvino-gpu",
backend=ov_backend,
model_path="/mock/ov/model",
file_path=Path("test.mp3"),
model_name="medium",
compute_type="fp16",
)
mock_get_backend.assert_not_called()
@patch("local_transcriber.transcriber.get_backend")
def test_transcribe_file_midstream_fallback_preserves_cpu_threads(mock_get_backend):
ov_backend = _make_backend(