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

feat(cli): добавлен батч-режим и конфигурационный файл (шаги 8–12)

Browse Source 
- Зачем:
  - обработка нескольких файлов за один вызов с загрузкой модели один раз.
  - хранение дефолтов (модель, язык, устройство) в .transcriber.toml.
- Что:
  - добавлен config.py: поиск .transcriber.toml (CWD → ~/.config), парсинг, валидация, приоритет CLI > конфиг > хардкод.
  - рефакторинг transcriber.py: выделены load_model() и _transcribe_file() с TranscribeFileResult для переиспользования модели в батче.
  - добавлены expand_globs() с дедупликацией и has_existing_transcript() в utils.py.
  - CLI: files: list[Path], --force/-f, prescan-first батч с итоговой статистикой и временем, Status-спиннер для прогресса.
  - README: секции батч-режим, конфигурационный файл, --force в таблице опций.
  - ADR-002: зафиксированы архитектурные решения (prescan-first, TranscribeFileResult, конфиг без мержа).
- Проверка:
  - uv run pytest -q — 94 passed, 1 skipped.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
ddadmin
2026-03-18 20:57:19 +03:00
co-authored by Claude Opus 4.6
parent b46827a283
commit e28232ef50
13 changed files with 1227 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+39
View File
@@ -56,6 +56,44 @@ transcribe podcast.wav --model small --device cpu
transcribe interview.m4a --output result.md
```
### Батч-режим
Обработка нескольких файлов за один вызов — модель загружается один раз:
```bash
# Все mp4 в директории
transcribe ./recordings/*.mp4
# Несколько файлов
transcribe meeting1.mp3 meeting2.mp3
# Перезаписать существующие транскрипты
transcribe *.mp4 --force
```
- Файлы с существующим транскриптом (`*-transcript.md`) автоматически пропускаются
- `--force` / `-f` — перезаписать существующие транскрипты
- В конце выводится итоговая статистика: обработано, пропущено, ошибок
- При ошибке в одном файле остальные продолжают обрабатываться
- `--output` несовместим с несколькими файлами
### Конфигурационный файл
Дефолтные параметры можно задать в `.transcriber.toml`:
```toml
model = "small"
language = "ru"
device = "cpu"
compute_type = "int8"
```
Порядок поиска:
1. `.transcriber.toml` в текущей директории (проектный конфиг)
2. `~/.config/transcriber/config.toml` (глобальный конфиг пользователя)
Приоритет: **CLI-аргумент > конфиг > встроенный дефолт**.
### Опции CLI
| Опция | Сокращение | По умолчанию | Описание |
@@ -65,6 +103,7 @@ transcribe interview.m4a --output result.md
| `--output` | `-o` | `<файл>-transcript.md` | Путь к выходному файлу |
| `--device` | `-d` | `auto` | Устройство (auto, cpu, cuda) |
| `--compute-type` | — | `int8` | Тип вычислений |
| `--force` | `-f` | — | Перезаписать существующие транскрипты |
| `--verbose` | `-v` | — | Подробный вывод |
## Модели
docs/adr/002-batch-and-config.md
+57
View File
@@ -0,0 +1,57 @@
# ADR-002: Batch mode и config
**Статус**: Принято
**Дата**: 2026-03-18
## Контекст
После завершения MVP (шаги 1–7) пользователям не хватает:
- Обработки нескольких файлов за один вызов (batch mode)
- Конфигурационного файла для хранения дефолтов (модель, язык, устройство)
## Решения
### 1. Prescan-first: валидация до загрузки модели
Модель загружается **только если есть файлы для обработки**. Перед загрузкой модели выполняется полный prescan: валидация всех файлов и проверка существующих транскриптов.
**Почему**: загрузка модели (large-v3) занимает ~10 секунд и ~3 GB RAM/VRAM. Повторный запуск по уже обработанным файлам должен быть дешёвым no-op.
### 2. TranscribeFileResult: возврат обновлённого состояния модели
При mid-stream CUDA fallback `_transcribe_file()` перезагружает модель на CPU внутри себя. Чтобы следующие файлы в батче не грузили модель повторно, результат включает обновлённые `model` и `actual_device`.
```python
@dataclass
class TranscribeFileResult:
result: TranscribeResult
model: WhisperModel # может измениться при fallback
actual_device: str # может измениться при fallback
```
**Альтернатива**: передавать модель по ссылке через mutable контейнер — менее явно и сложнее тестировать.
### 3. Конфиг: CWD → глобальный, без мержа
Порядок поиска:
1. `.transcriber.toml` в текущей директории (проектный конфиг)
2. `~/.config/transcriber/config.toml` (глобальный конфиг)
Первый найденный побеждает, мержа между файлами нет.
**Почему**: CWD-конфиг удобен для per-project дефолтов (`language = "ru"` для русскоязычного проекта), глобальный — для машинных дефолтов (`device = "cpu"` на ноутбуке без GPU). Мерж усложняет предсказуемость.
**Приоритет значений**: CLI > конфиг > хардкод.
### 4. `_transcribe_file()` — внутренний helper
Публичный API (`transcribe()`) сохранён без изменений. Новая функция `_transcribe_file()` — внутренний helper с префиксом `_`, не часть публичного контракта.
`transcribe()` стала тонкой обёрткой: `load_model()` + `_transcribe_file()` → `TranscribeResult`.
## Последствия
- Обратная совместимость CLI: `transcribe file.mp4` работает как раньше
- Все существующие тесты проходят без изменений сигнатур
- Batch mode: модель загружается один раз для всех файлов
- `--force` флаг для перезаписи существующих транскриптов в батч-режиме
docs/plan.md
+63
View File
@@ -288,6 +288,69 @@
**Критерий готовности**: коллега может по README установить и запустить на Windows/WSL2 без вопросов.
---
## Шаг 8: Конфиг `.transcriber.toml`
- [x] Зависимость `tomli` в `pyproject.toml` (для Python < 3.11)
- [x] Новый файл `src/local_transcriber/config.py`:
- `HARDCODED_DEFAULTS` — дефолтные значения
- `find_config_file()` — ищет `.transcriber.toml` в CWD, потом `~/.config/transcriber/config.toml`
- `load_config()` — парсит TOML, валидирует ключи/значения
- `resolve_defaults()` — приоритет CLI > конфиг > хардкод
- [x] Тесты `tests/test_config.py` — 11 тестов
**Критерий готовности**: `uv run pytest tests/test_config.py -v` — все тесты зелёные.
---
## Шаг 9: Рефакторинг transcriber.py — выделить загрузку модели
- [x] `load_model()` — загрузка модели с CUDA-фолбеком
- [x] `_transcribe_file()` — транскрипция одного файла, возвращает `TranscribeFileResult`
- [x] `TranscribeFileResult` — dataclass с result, model, actual_device
- [x] `transcribe()` — тонкая обёртка для обратной совместимости
- [x] Новые тесты: `test_load_model_cuda_fallback`, `test_load_model_strict_raises`, `test__transcribe_file_basic`
**Критерий готовности**: все существующие тесты `test_transcriber.py` проходят без изменений.
---
## Шаг 10: Утилиты для батч-режима в utils.py
- [x] `expand_globs()` — раскрытие glob-паттернов
- [x] `has_existing_transcript()` — проверка существования транскрипта
- [x] Тесты в `test_utils.py` — 5 новых тестов
**Критерий готовности**: `uv run pytest tests/test_utils.py -v` — все тесты зелёные.
---
## Шаг 11: Батч-режим + конфиг в CLI
- [x] Изменение сигнатуры CLI: `files: list[Path]`, дефолты `None`, `--force`
- [x] Интеграция `load_config()` / `resolve_defaults()` в `main()`
- [x] `_run_single()` — текущий flow для одного файла
- [x] `_run_batch()` — prescan, загрузка модели, транскрипция с итогами
- [x] Обновлённые тесты `test_cli.py` — 11 новых тестов
**Критерий готовности**: `uv run pytest tests/test_cli.py -v` — все тесты зелёные.
---
## Шаг 12: Документация + ADR
- [x] ADR-002: Batch mode и config (`docs/adr/002-batch-and-config.md`)
- [x] README.md: секции «Батч-режим», «Конфигурационный файл», обновлена таблица опций CLI
- [x] `docs/plan.md`: отмечены шаги 8–12
**Критерий готовности**: README содержит документацию по батч-режиму и конфигу.
---
## Инструкция для агента
pyproject.toml
+1
View File
@@ -9,6 +9,7 @@ dependencies = [
"faster-whisper>=1.2.1",
"socksio>=1.0.0",
"nvidia-cublas-cu12>=12.4; sys_platform == 'linux' and platform_machine == 'x86_64'",
"tomli>=2.0; python_version < '3.11'",
]
[project.scripts]
src/local_transcriber/cli.py
+231 -28
View File
@@ -6,9 +6,25 @@ import typer
from rich.console import Console
from rich.status import Status
from .config import load_config, resolve_defaults
from .formatter import format_transcript, write_transcript
from .transcriber import Segment, _is_cuda_error, ensure_model_available, transcribe
from .utils import build_output_path, check_ffmpeg, detect_device, get_gpu_name, validate_input_file
from .transcriber import (
Segment,
_is_cuda_error,
_transcribe_file,
ensure_model_available,
load_model,
transcribe,
)
from .utils import (
build_output_path,
check_ffmpeg,
detect_device,
expand_globs,
get_gpu_name,
has_existing_transcript,
validate_input_file,
)
app = typer.Typer()
console = Console(stderr=True)
@@ -16,22 +32,53 @@ console = Console(stderr=True)
@app.command()
def main(
file: Path = typer.Argument(..., help="Путь к аудио- или видеофайлу"),
model: str = typer.Option("large-v3", "--model", "-m", help="Модель Whisper"),
language: str = typer.Option("auto", "--language", "-l", help="Язык (ru|en|auto)"),
files: list[Path] = typer.Argument(..., help="Пути к аудио/видеофайлам"),
model: str | None = typer.Option(
None, "--model", "-m", show_default=False, help="Модель Whisper [по умолч.: large-v3]"
),
language: str | None = typer.Option(
None, "--language", "-l", show_default=False, help="Язык (ru|en|auto) [по умолч.: auto]"
),
output: Path | None = typer.Option(None, "--output", "-o", help="Путь к выходному файлу"),
device: str = typer.Option("auto", "--device", "-d", help="Устройство (auto|cpu|cuda)"),
compute_type: str = typer.Option("int8", "--compute-type", help="Тип вычислений"),
device: str | None = typer.Option(
None, "--device", "-d", show_default=False, help="Устройство (auto|cpu|cuda) [по умолч.: auto]"
),
compute_type: str | None = typer.Option(
None, "--compute-type", show_default=False, help="Тип вычислений [по умолч.: int8]"
),
verbose: bool = typer.Option(False, "--verbose", "-v", help="Подробный вывод"),
force: bool = typer.Option(False, "--force", "-f", help="Перезаписать существующие транскрипты"),
) -> None:
try:
_run(file, model, language, output, device, compute_type, verbose)
config = load_config()
defaults = resolve_defaults(
{"model": model, "language": language, "device": device, "compute_type": compute_type},
config,
)
expanded = expand_globs(files)
if not expanded:
console.print("Файлы не найдены.", style="red bold")
raise SystemExit(1)
is_batch = len(expanded) > 1
if is_batch and output is not None:
console.print("--output несовместим с несколькими файлами.", style="red bold")
raise SystemExit(1)
if is_batch:
_run_batch(expanded, defaults, verbose, force)
else:
_run_single(expanded[0], defaults, output, verbose)
except KeyboardInterrupt:
console.print("\nПрервано пользователем.", style="yellow")
raise SystemExit(130)
except SystemExit:
raise
except (FileNotFoundError, ValueError) as exc:
except ValueError as exc:
console.print(f"Ошибка: {exc}", style="red bold")
raise SystemExit(1)
except (FileNotFoundError,) as exc:
console.print(f"Ошибка: {exc}", style="red bold")
raise SystemExit(1)
except Exception as exc:
@@ -54,59 +101,72 @@ def main(
raise SystemExit(1)
def _run(
def _run_single(
file: Path,
model: str,
language: str,
defaults: dict[str, str],
output: Path | None,
device: str,
compute_type: str,
verbose: bool,
) -> None:
start = time.monotonic()
check_ffmpeg()
validated_file = validate_input_file(file)
requested_device = device
resolved_device = detect_device(device)
requested_device = defaults["device"]
resolved_device = detect_device(requested_device)
strict = requested_device != "auto"
output_path = build_output_path(validated_file, output)
console.print(f"Файл: [bold]{validated_file.name}[/bold]")
console.print(f"Модель: [bold]{model}[/bold] Устройство: [bold]{resolved_device}[/bold] Compute: [bold]{compute_type}[/bold]")
console.print(
f"Модель: [bold]{defaults['model']}[/bold] "
f"Устройство: [bold]{resolved_device}[/bold] "
f"Compute: [bold]{defaults['compute_type']}[/bold]"
)
model_path = ensure_model_available(model, on_status=lambda message: console.print(message))
model_path = ensure_model_available(
defaults["model"], on_status=lambda message: console.print(message)
)
def on_segment(seg: Segment) -> None:
console.print(f" [{seg.start:.2f}s] {seg.text.strip()}")
model_obj, actual_device = load_model(
model_path, resolved_device, defaults["compute_type"],
on_status=lambda msg: console.print(msg), strict_device=strict,
)
with Status("Подготавливаю запуск...", console=console) as status:
result = transcribe(
tfr = _transcribe_file(
model=model_obj,
actual_device=actual_device,
file_path=validated_file,
model_name=model_path,
device=resolved_device,
compute_type=compute_type,
language=language if language != "auto" else None,
compute_type=defaults["compute_type"],
language=defaults["language"] if defaults["language"] != "auto" else None,
on_segment=on_segment if verbose else None,
on_status=status.update,
strict_device=strict,
)
if result.device_used != resolved_device:
result = tfr.result
if tfr.actual_device != resolved_device:
if requested_device == "auto":
console.print(
f"Определено устройство {resolved_device}, "
f"но использовано {result.device_used} (fallback)",
f"но использовано {tfr.actual_device} (fallback)",
style="yellow",
)
else:
console.print(
f"Запрошено {requested_device}, использовано {result.device_used}",
f"Запрошено {requested_device}, использовано {tfr.actual_device}",
style="yellow",
)
if len(result.segments) == 0:
console.print(f"Речь не обнаружена в файле {validated_file.name}", style="yellow")
console.print(
f"Речь не обнаружена в файле {validated_file.name}", style="yellow"
)
if result.device_used == "cuda":
gpu_name = get_gpu_name()
@@ -114,12 +174,12 @@ def _run(
else:
device_info = "CPU"
language_mode = "detected" if language == "auto" else "forced"
language_mode = "detected" if defaults["language"] == "auto" else "forced"
content = format_transcript(
result=result,
source_filename=validated_file.name,
model_name=model,
model_name=defaults["model"],
device_info=device_info,
language_mode=language_mode,
)
@@ -130,5 +190,148 @@ def _run(
console.print(f" Сегментов: {len(result.segments)} Время: {elapsed:.1f}с")
def _run_batch(
files: list[Path],
defaults: dict[str, str],
verbose: bool,
force: bool,
) -> None:
check_ffmpeg()
# Phase 1: Prescan
to_process: list[Path] = []
skipped = 0
invalid = 0
for file in files:
try:
validated = validate_input_file(file)
except (FileNotFoundError, ValueError) as exc:
console.print(f" Ошибка: {file.name}: {exc}", style="red")
invalid += 1
continue
if not force and has_existing_transcript(validated):
console.print(
f" Пропуск: {file.name} (транскрипт существует)", style="dim"
)
skipped += 1
continue
to_process.append(validated)
if not to_process:
console.print(
f"\nИтого: 0 обработано, {skipped} пропущено, {invalid} ошибок"
)
if invalid > 0:
raise SystemExit(1)
return
# Phase 2: Load model
requested_device = defaults["device"]
resolved_device = detect_device(requested_device)
strict = requested_device != "auto"
model_path = ensure_model_available(
defaults["model"], on_status=lambda msg: console.print(msg)
)
model_obj, actual_device = load_model(
model_path, resolved_device, defaults["compute_type"],
on_status=lambda msg: console.print(msg), strict_device=strict,
)
if actual_device != resolved_device:
if requested_device == "auto":
console.print(
f"Определено устройство {resolved_device}, "
f"но используется {actual_device} (fallback)",
style="yellow",
)
else:
console.print(
f"Запрошено {requested_device}, используется {actual_device}",
style="yellow",
)
# Phase 3: Transcribe
processed = 0
failed = 0
language_mode = "detected" if defaults["language"] == "auto" else "forced"
batch_start = time.monotonic()
for i, file in enumerate(to_process, 1):
try:
prefix = f"[{i}/{len(to_process)}] {file.name}"
console.print(f"{prefix}", style="bold")
file_start = time.monotonic()
def on_segment(seg: Segment) -> None:
console.print(f" [{seg.start:.2f}s] {seg.text.strip()}")
with Status(f"{prefix}...", console=console) as status:
tfr = _transcribe_file(
model=model_obj,
actual_device=actual_device,
file_path=file,
model_name=model_path,
compute_type=defaults["compute_type"],
language=defaults["language"] if defaults["language"] != "auto" else None,
on_segment=on_segment if verbose else None,
on_status=status.update if not verbose else lambda msg: console.print(msg),
strict_device=strict,
)
if tfr.actual_device != actual_device:
console.print(
f" {file.name}: fallback на {tfr.actual_device} при транскрипции",
style="yellow",
)
model_obj, actual_device = tfr.model, tfr.actual_device
result = tfr.result
if len(result.segments) == 0:
console.print(
f" Речь не обнаружена: {file.name}", style="yellow"
)
if result.device_used == "cuda":
gpu_name = get_gpu_name()
device_info = f"CUDA ({gpu_name or 'Unknown GPU'})"
else:
device_info = "CPU"
content = format_transcript(
result=result,
source_filename=file.name,
model_name=defaults["model"],
device_info=device_info,
language_mode=language_mode,
)
write_transcript(content, build_output_path(file))
file_elapsed = time.monotonic() - file_start
console.print(
f" Готово: {file.name} "
f"Сегментов: {len(result.segments)} Время: {file_elapsed:.1f}с",
style="green",
)
processed += 1
except KeyboardInterrupt:
raise
except Exception as exc:
if verbose:
console.print_exception()
else:
console.print(f" Ошибка: {file.name}: {exc}", style="red")
failed += 1
total_failed = invalid + failed
batch_elapsed = time.monotonic() - batch_start
console.print(
f"\nИтого: {processed} обработано, {skipped} пропущено, {total_failed} ошибок"
f" Время: {batch_elapsed:.1f}с"
)
if total_failed > 0:
raise SystemExit(1)
if __name__ == "__main__":
app()
src/local_transcriber/config.py
+85
View File
@@ -0,0 +1,85 @@
import sys
import warnings
from pathlib import Path
if sys.version_info >= (3, 11):
import tomllib
else:
import tomli as tomllib
HARDCODED_DEFAULTS: dict[str, str] = {
"model": "large-v3",
"language": "auto",
"device": "auto",
"compute_type": "int8",
}
_VALID_KEYS = set(HARDCODED_DEFAULTS)
_VALID_DEVICES = {"auto", "cpu", "cuda"}
def find_config_file() -> Path | None:
cwd_config = Path.cwd() / ".transcriber.toml"
if cwd_config.is_file():
return cwd_config
global_config = Path.home() / ".config" / "transcriber" / "config.toml"
if global_config.is_file():
return global_config
return None
def load_config(path: Path | None = None) -> dict[str, str]:
if path is None:
path = find_config_file()
if path is None:
return {}
try:
raw = path.read_bytes()
data = tomllib.loads(raw.decode("utf-8"))
except Exception as exc:
raise ValueError(f"Ошибка чтения конфига {path}: {exc}") from exc
unknown = set(data) - _VALID_KEYS
if unknown:
warnings.warn(
f"Неизвестные ключи в {path}: {', '.join(sorted(unknown))}",
stacklevel=2,
)
result: dict[str, str] = {}
for key in _VALID_KEYS:
if key not in data:
continue
value = data[key]
if not isinstance(value, str):
raise ValueError(
f"Значение '{key}' в {path} должно быть строкой, получено {type(value).__name__}"
)
if key == "device" and value not in _VALID_DEVICES:
raise ValueError(
f"Недопустимое значение device = '{value}' в {path}. "
f"Ожидается: {', '.join(sorted(_VALID_DEVICES))}"
)
if key == "language" and not value:
raise ValueError(f"Значение 'language' в {path} не может быть пустым")
result[key] = value
return result
def resolve_defaults(
cli_values: dict[str, str | None], config: dict[str, str]
) -> dict[str, str]:
result: dict[str, str] = {}
for key in HARDCODED_DEFAULTS:
cli_val = cli_values.get(key)
if cli_val is not None:
result[key] = cli_val
elif key in config:
result[key] = config[key]
else:
result[key] = HARDCODED_DEFAULTS[key]
return result
src/local_transcriber/transcriber.py
+49 -11
View File
@@ -52,19 +52,22 @@ class TranscribeResult:
device_used: str # "cpu" / "cuda"
def transcribe(
file_path: Path,
model_name: str = "large-v3",
device: str = "auto",
compute_type: str = "int8",
language: str | None = None,
on_segment: Callable[[Segment], None] | None = None,
@dataclass
class TranscribeFileResult:
result: TranscribeResult
model: WhisperModel
actual_device: str
def load_model(
model_name: str,
device: str,
compute_type: str,
on_status: Callable[[str], None] | None = None,
strict_device: bool = False,
) -> TranscribeResult:
) -> tuple[WhisperModel, str]:
"""Загружает модель с CUDA-фолбеком. Возвращает (model, actual_device)."""
actual_device = device
lang_arg = language if language and language != "auto" else None
try:
_notify_status(on_status, f"Инициализирую модель на {device}...")
model = _create_model(model_name, device, compute_type)
@@ -82,6 +85,22 @@ def transcribe(
model = _create_model(model_name, "cpu", compute_type)
else:
raise
return model, actual_device
def _transcribe_file(
model: WhisperModel,
actual_device: str,
file_path: Path,
model_name: str,
compute_type: str,
language: str | None = None,
on_segment: Callable[[Segment], None] | None = None,
on_status: Callable[[str], None] | None = None,
strict_device: bool = False,
) -> TranscribeFileResult:
"""Транскрибирует один файл. При mid-stream CUDA fallback перезагружает модель."""
lang_arg = language if language and language != "auto" else None
try:
_notify_status(on_status, "Транскрибирую...")
@@ -103,13 +122,32 @@ def transcribe(
else:
raise
return TranscribeResult(
result = TranscribeResult(
segments=segments,
language=info.language,
language_probability=info.language_probability,
duration=info.duration,
device_used=actual_device,
)
return TranscribeFileResult(result=result, model=model, actual_device=actual_device)
def transcribe(
file_path: Path,
model_name: str = "large-v3",
device: str = "auto",
compute_type: str = "int8",
language: str | None = None,
on_segment: Callable[[Segment], None] | None = None,
on_status: Callable[[str], None] | None = None,
strict_device: bool = False,
) -> TranscribeResult:
model, actual_device = load_model(model_name, device, compute_type, on_status, strict_device)
tfr = _transcribe_file(
model, actual_device, file_path, model_name, compute_type,
language, on_segment, on_status, strict_device,
)
return tfr.result
def ensure_model_available(
src/local_transcriber/utils.py
+22
View File
@@ -1,3 +1,4 @@
import glob
import shutil
import subprocess
import sys
@@ -68,3 +69,24 @@ def build_output_path(input_path: Path, output: Path | None = None) -> Path:
if output is not None:
return output
return input_path.with_stem(input_path.stem + "-transcript").with_suffix(".md")
def expand_globs(paths: list[Path]) -> list[Path]:
seen: set[Path] = set()
result: list[Path] = []
for p in paths:
s = str(p)
if any(c in s for c in ("*", "?", "[")):
candidates = [Path(m) for m in sorted(glob.glob(s))]
else:
candidates = [p]
for c in candidates:
resolved = c.resolve()
if resolved not in seen:
seen.add(resolved)
result.append(c)
return result
def has_existing_transcript(input_path: Path) -> bool:
return build_output_path(input_path).exists()
tests/test_cli.py
+480 -63
View File
@@ -5,7 +5,7 @@ import pytest
from typer.testing import CliRunner
from local_transcriber.cli import app
from local_transcriber.transcriber import Segment, TranscribeResult
from local_transcriber.transcriber import Segment, TranscribeFileResult, TranscribeResult
runner = CliRunner()
@@ -20,15 +20,32 @@ def _make_result(segments=None, language="ru", device_used="cpu", duration=60.0)
)
def _patches(result=None, tmp_file=None):
"""Context managers for a standard CLI happy path."""
def _make_model():
return MagicMock(name="WhisperModel")
def _make_tfr(result=None, model=None, actual_device="cpu"):
if result is None:
result = _make_result()
if model is None:
model = _make_model()
return TranscribeFileResult(result=result, model=model, actual_device=actual_device)
def _single_patches(result=None, tmp_file=None, actual_device="cpu"):
"""Patches for a standard single-file CLI happy path."""
if result is None:
result = _make_result(device_used=actual_device)
model = _make_model()
tfr = TranscribeFileResult(result=result, model=model, actual_device=actual_device)
return [
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
patch("local_transcriber.cli.validate_input_file", return_value=tmp_file),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.transcribe", return_value=result),
patch("local_transcriber.cli.detect_device", return_value=actual_device),
patch("local_transcriber.cli.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.load_model", return_value=(model, actual_device)),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch("local_transcriber.cli.write_transcript"),
]
@@ -36,16 +53,9 @@ def _patches(result=None, tmp_file=None):
def test_cli_happy_path_exit_code_zero(tmp_path):
audio = tmp_path / "test.mp3"
audio.write_bytes(b"fake")
result = _make_result()
with (
patch("local_transcriber.cli.check_ffmpeg"),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.transcribe", return_value=result),
patch("local_transcriber.cli.write_transcript"),
):
patches = _single_patches(tmp_file=audio)
with patches[0], patches[1], patches[2], patches[3], patches[4], patches[5], patches[6], patches[7]:
out = runner.invoke(app, [str(audio)])
assert out.exit_code == 0
@@ -55,38 +65,45 @@ def test_cli_default_options_passed_to_transcribe(tmp_path):
audio = tmp_path / "test.mp3"
audio.write_bytes(b"fake")
result = _make_result()
mock_transcribe = MagicMock(return_value=result)
model = _make_model()
tfr = _make_tfr(result=result, model=model)
mock_transcribe_file = MagicMock(return_value=tfr)
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.transcribe", mock_transcribe),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu")),
patch("local_transcriber.cli._transcribe_file", mock_transcribe_file),
patch("local_transcriber.cli.write_transcript"),
):
runner.invoke(app, [str(audio)])
call_kwargs = mock_transcribe.call_args[1]
call_kwargs = mock_transcribe_file.call_args[1]
assert call_kwargs["model_name"] == "/models/large-v3"
assert call_kwargs["device"] == "cpu"
assert call_kwargs["compute_type"] == "int8"
assert call_kwargs["language"] is None # "auto" → None passed to transcribe
assert call_kwargs["language"] is None # "auto" → None
assert call_kwargs["on_segment"] is None # verbose=False
def test_cli_custom_options(tmp_path):
audio = tmp_path / "test.mp3"
audio.write_bytes(b"fake")
result = _make_result()
mock_transcribe = MagicMock(return_value=result)
result = _make_result(device_used="cuda")
model = _make_model()
tfr = _make_tfr(result=result, model=model, actual_device="cuda")
mock_transcribe_file = MagicMock(return_value=tfr)
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cuda"),
patch("local_transcriber.cli.ensure_model_available", return_value="/models/small"),
patch("local_transcriber.cli.transcribe", mock_transcribe),
patch("local_transcriber.cli.load_model", return_value=(model, "cuda")),
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"),
):
@@ -98,9 +115,9 @@ def test_cli_custom_options(tmp_path):
"--compute-type", "float16",
])
call_kwargs = mock_transcribe.call_args[1]
call_kwargs = mock_transcribe_file.call_args[1]
assert call_kwargs["model_name"] == "/models/small"
assert call_kwargs["language"] == "ru" # explicit language passed through
assert call_kwargs["language"] == "ru"
assert call_kwargs["compute_type"] == "float16"
@@ -108,19 +125,23 @@ def test_cli_verbose_passes_on_segment_callback(tmp_path):
audio = tmp_path / "test.mp3"
audio.write_bytes(b"fake")
result = _make_result()
mock_transcribe = MagicMock(return_value=result)
model = _make_model()
tfr = _make_tfr(result=result, model=model)
mock_transcribe_file = MagicMock(return_value=tfr)
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.transcribe", mock_transcribe),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu")),
patch("local_transcriber.cli._transcribe_file", mock_transcribe_file),
patch("local_transcriber.cli.write_transcript"),
):
runner.invoke(app, [str(audio), "--verbose"])
call_kwargs = mock_transcribe.call_args[1]
call_kwargs = mock_transcribe_file.call_args[1]
assert call_kwargs["on_segment"] is not None
assert callable(call_kwargs["on_segment"])
@@ -130,14 +151,8 @@ def test_cli_empty_speech_warning(tmp_path):
audio.write_bytes(b"fake")
result = _make_result(segments=[])
with (
patch("local_transcriber.cli.check_ffmpeg"),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.transcribe", return_value=result),
patch("local_transcriber.cli.write_transcript"),
):
patches = _single_patches(result=result, tmp_file=audio)
with patches[0], patches[1], patches[2], patches[3], patches[4], patches[5], patches[6], patches[7]:
out = runner.invoke(app, [str(audio)])
assert out.exit_code == 0
@@ -147,15 +162,20 @@ def test_cli_empty_speech_warning(tmp_path):
def test_cli_default_output_path(tmp_path):
audio = tmp_path / "meeting.mp3"
audio.write_bytes(b"fake")
result = _make_result()
mock_write = MagicMock()
result = _make_result()
model = _make_model()
tfr = _make_tfr(result=result, model=model)
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.transcribe", return_value=result),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu")),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch("local_transcriber.cli.write_transcript", mock_write),
):
runner.invoke(app, [str(audio)])
@@ -168,15 +188,20 @@ def test_cli_custom_output_path(tmp_path):
audio = tmp_path / "meeting.mp3"
audio.write_bytes(b"fake")
out_file = tmp_path / "custom.md"
result = _make_result()
mock_write = MagicMock()
result = _make_result()
model = _make_model()
tfr = _make_tfr(result=result, model=model)
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.transcribe", return_value=result),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu")),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch("local_transcriber.cli.write_transcript", mock_write),
):
runner.invoke(app, [str(audio), "--output", str(out_file)])
@@ -189,7 +214,10 @@ def test_cli_error_exit_code_one(tmp_path):
audio = tmp_path / "test.mp3"
audio.write_bytes(b"fake")
with patch("local_transcriber.cli.check_ffmpeg", side_effect=SystemExit(1)):
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg", side_effect=SystemExit(1)),
):
out = runner.invoke(app, [str(audio)])
assert out.exit_code == 1
@@ -199,19 +227,23 @@ def test_cli_passes_status_callback_to_transcribe(tmp_path):
audio = tmp_path / "test.mp3"
audio.write_bytes(b"fake")
result = _make_result()
mock_transcribe = MagicMock(return_value=result)
model = _make_model()
tfr = _make_tfr(result=result, model=model)
mock_transcribe_file = MagicMock(return_value=tfr)
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.transcribe", mock_transcribe),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu")),
patch("local_transcriber.cli._transcribe_file", mock_transcribe_file),
patch("local_transcriber.cli.write_transcript"),
):
runner.invoke(app, [str(audio)])
call_kwargs = mock_transcribe.call_args[1]
call_kwargs = mock_transcribe_file.call_args[1]
assert call_kwargs["on_status"] is not None
assert callable(call_kwargs["on_status"])
@@ -220,20 +252,24 @@ def test_cli_resolves_model_before_transcribe(tmp_path):
audio = tmp_path / "test.mp3"
audio.write_bytes(b"fake")
result = _make_result()
mock_transcribe = MagicMock(return_value=result)
model = _make_model()
tfr = _make_tfr(result=result, model=model)
mock_transcribe_file = MagicMock(return_value=tfr)
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.ensure_model_available", return_value="/models/large-v3") as mock_ensure_model,
patch("local_transcriber.cli.transcribe", mock_transcribe),
patch("local_transcriber.cli.ensure_model_available", return_value="/models/large-v3") as mock_ensure,
patch("local_transcriber.cli.load_model", return_value=(model, "cpu")),
patch("local_transcriber.cli._transcribe_file", mock_transcribe_file),
patch("local_transcriber.cli.write_transcript"),
):
runner.invoke(app, [str(audio), "--model", "large-v3"])
mock_ensure_model.assert_called_once()
call_kwargs = mock_transcribe.call_args[1]
mock_ensure.assert_called_once()
call_kwargs = mock_transcribe_file.call_args[1]
assert call_kwargs["model_name"] == "/models/large-v3"
@@ -241,13 +277,16 @@ def test_cli_windows_cuda_diagnostic(tmp_path):
"""CUDA error on Windows prints choco/winget install hint."""
audio = tmp_path / "test.mp3"
audio.write_bytes(b"fake")
model = _make_model()
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cuda"),
patch("local_transcriber.cli.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.transcribe", side_effect=RuntimeError("CUDA error: no device")),
patch("local_transcriber.cli.load_model", return_value=(model, "cuda")),
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"
@@ -262,13 +301,16 @@ def test_cli_linux_cuda_error_no_windows_hint(tmp_path):
"""CUDA error on Linux does NOT print Windows-specific hint."""
audio = tmp_path / "test.mp3"
audio.write_bytes(b"fake")
model = _make_model()
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cuda"),
patch("local_transcriber.cli.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.transcribe", side_effect=RuntimeError("CUDA error: no device")),
patch("local_transcriber.cli.load_model", return_value=(model, "cuda")),
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"
@@ -283,16 +325,19 @@ def test_cli_device_fallback_warning(tmp_path):
audio = tmp_path / "test.mp3"
audio.write_bytes(b"fake")
result = _make_result(device_used="cpu")
model = _make_model()
tfr = TranscribeFileResult(result=result, model=model, actual_device="cpu")
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cuda"),
patch("local_transcriber.cli.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.transcribe", return_value=result),
patch("local_transcriber.cli.load_model", return_value=(model, "cuda")),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch("local_transcriber.cli.write_transcript"),
):
# --device auto (default) -> detect_device returns "cuda" but result is "cpu"
out = runner.invoke(app, [str(audio)])
assert "fallback" in out.output
@@ -303,49 +348,59 @@ def test_cli_strict_device_passed_to_transcribe(tmp_path):
audio = tmp_path / "test.mp3"
audio.write_bytes(b"fake")
result = _make_result(device_used="cuda")
mock_transcribe = MagicMock(return_value=result)
model = _make_model()
tfr = TranscribeFileResult(result=result, model=model, actual_device="cuda")
mock_transcribe_file = MagicMock(return_value=tfr)
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cuda"),
patch("local_transcriber.cli.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.transcribe", mock_transcribe),
patch("local_transcriber.cli.load_model", return_value=(model, "cuda")),
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), "--device", "cuda"])
assert mock_transcribe.call_args[1]["strict_device"] is True
assert mock_transcribe_file.call_args[1]["strict_device"] is True
mock_transcribe.reset_mock()
mock_transcribe_file.reset_mock()
result_cpu = _make_result(device_used="cpu")
mock_transcribe.return_value = result_cpu
tfr_cpu = TranscribeFileResult(result=result_cpu, model=model, actual_device="cpu")
mock_transcribe_file.return_value = tfr_cpu
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.transcribe", mock_transcribe),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu")),
patch("local_transcriber.cli._transcribe_file", mock_transcribe_file),
patch("local_transcriber.cli.write_transcript"),
):
runner.invoke(app, [str(audio)])
assert mock_transcribe.call_args[1]["strict_device"] is False
assert mock_transcribe_file.call_args[1]["strict_device"] is False
def test_cli_keyboard_interrupt(tmp_path):
"""Ctrl+C → exit code 130, 'Прервано пользователем' in output."""
audio = tmp_path / "test.mp3"
audio.write_bytes(b"fake")
model = _make_model()
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.transcribe", side_effect=KeyboardInterrupt),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu")),
patch("local_transcriber.cli._transcribe_file", side_effect=KeyboardInterrupt),
patch("local_transcriber.cli.write_transcript"),
):
out = runner.invoke(app, [str(audio)])
@@ -358,7 +413,7 @@ def test_cli_user_error_no_traceback(tmp_path):
"""FileNotFoundError → clean message, no traceback."""
audio = tmp_path / "missing.mp3"
with patch("local_transcriber.cli.check_ffmpeg"):
with patch("local_transcriber.cli.load_config", return_value={}):
out = runner.invoke(app, [str(audio)])
assert out.exit_code == 1
@@ -370,13 +425,16 @@ def test_cli_unexpected_error_verbose_traceback(tmp_path):
"""Unexpected error with --verbose → traceback shown."""
audio = tmp_path / "test.mp3"
audio.write_bytes(b"fake")
model = _make_model()
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.transcribe", side_effect=RuntimeError("unexpected boom")),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu")),
patch("local_transcriber.cli._transcribe_file", side_effect=RuntimeError("unexpected boom")),
patch("local_transcriber.cli.write_transcript"),
):
out = runner.invoke(app, [str(audio), "--verbose"])
@@ -389,13 +447,16 @@ def test_cli_unexpected_error_no_verbose_hint(tmp_path):
"""Unexpected error without --verbose → hint to use --verbose."""
audio = tmp_path / "test.mp3"
audio.write_bytes(b"fake")
model = _make_model()
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.transcribe", side_effect=RuntimeError("unexpected boom")),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu")),
patch("local_transcriber.cli._transcribe_file", side_effect=RuntimeError("unexpected boom")),
patch("local_transcriber.cli.write_transcript"),
):
out = runner.invoke(app, [str(audio)])
@@ -403,3 +464,359 @@ def test_cli_unexpected_error_no_verbose_hint(tmp_path):
assert out.exit_code == 1
assert "Ошибка" in out.output
assert "--verbose" in out.output
# === Batch mode tests ===
def test_cli_batch_two_files(tmp_path):
a = tmp_path / "a.mp3"
b = tmp_path / "b.mp3"
a.write_bytes(b"fake")
b.write_bytes(b"fake")
result = _make_result()
model = _make_model()
tfr = _make_tfr(result=result, model=model)
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
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.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu")),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch("local_transcriber.cli.write_transcript"),
):
out = runner.invoke(app, [str(a), str(b)])
assert out.exit_code == 0
assert "2 обработано" in out.output
def test_cli_batch_skips_existing(tmp_path):
a = tmp_path / "a.mp3"
b = tmp_path / "b.mp3"
a.write_bytes(b"fake")
b.write_bytes(b"fake")
# Create transcript for a
(tmp_path / "a-transcript.md").write_text("existing")
result = _make_result()
model = _make_model()
tfr = _make_tfr(result=result, model=model)
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
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.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu")),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch("local_transcriber.cli.write_transcript"),
):
out = runner.invoke(app, [str(a), str(b)])
assert out.exit_code == 0
assert "Пропуск" in out.output
assert "1 обработано" in out.output
assert "1 пропущено" in out.output
def test_cli_batch_all_skipped_no_model_load(tmp_path):
a = tmp_path / "a.mp3"
a.write_bytes(b"fake")
(tmp_path / "a-transcript.md").write_text("existing")
b = tmp_path / "b.mp3"
b.write_bytes(b"fake")
(tmp_path / "b-transcript.md").write_text("existing")
mock_load_model = MagicMock()
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
patch("local_transcriber.cli.validate_input_file", side_effect=lambda p: p),
patch("local_transcriber.cli.load_model", mock_load_model),
):
out = runner.invoke(app, [str(a), str(b)])
assert out.exit_code == 0
mock_load_model.assert_not_called()
def test_cli_batch_force_overwrites(tmp_path):
a = tmp_path / "a.mp3"
a.write_bytes(b"fake")
(tmp_path / "a-transcript.md").write_text("existing")
b = tmp_path / "b.mp3"
b.write_bytes(b"fake")
result = _make_result()
model = _make_model()
tfr = _make_tfr(result=result, model=model)
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
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.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu")),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch("local_transcriber.cli.write_transcript"),
):
out = runner.invoke(app, [str(a), str(b), "--force"])
assert out.exit_code == 0
assert "Пропуск" not in out.output
assert "2 обработано" in out.output
def test_cli_batch_per_file_error(tmp_path):
a = tmp_path / "a.mp3"
b = tmp_path / "b.mp3"
a.write_bytes(b"fake")
b.write_bytes(b"fake")
result = _make_result()
model = _make_model()
tfr = _make_tfr(result=result, model=model)
call_count = 0
def transcribe_side_effect(**kwargs):
nonlocal call_count
call_count += 1
if call_count == 1:
raise RuntimeError("oops")
return tfr
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
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.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu")),
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)])
assert out.exit_code == 1
assert "1 обработано" in out.output
assert "1 ошибок" in out.output
def test_cli_batch_invalid_in_prescan(tmp_path):
a = tmp_path / "a.mp3"
a.write_bytes(b"fake")
b = tmp_path / "b.mp3"
# b doesn't exist
result = _make_result()
model = _make_model()
tfr = _make_tfr(result=result, model=model)
def validate_side_effect(p):
if not p.exists():
raise FileNotFoundError(f"Файл не найден: {p}")
return p
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
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.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu")),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch("local_transcriber.cli.write_transcript"),
):
out = runner.invoke(app, [str(a), str(b)])
assert out.exit_code == 1
assert "1 обработано" in out.output
assert "1 ошибок" in out.output
def test_cli_batch_output_incompatible(tmp_path):
a = tmp_path / "a.mp3"
b = tmp_path / "b.mp3"
a.write_bytes(b"fake")
b.write_bytes(b"fake")
with patch("local_transcriber.cli.load_config", return_value={}):
out = runner.invoke(app, [str(a), str(b), "--output", "out.md"])
assert out.exit_code == 1
assert "--output несовместим" in out.output
def test_cli_batch_empty_glob(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
with patch("local_transcriber.cli.load_config", return_value={}):
out = runner.invoke(app, ["*.mp3"])
assert out.exit_code == 1
assert "Файлы не найдены" in out.output
def test_cli_config_applied(tmp_path):
audio = tmp_path / "test.mp3"
audio.write_bytes(b"fake")
model = _make_model()
result = _make_result()
tfr = _make_tfr(result=result, model=model)
with (
patch("local_transcriber.cli.load_config", return_value={"model": "tiny"}),
patch("local_transcriber.cli.check_ffmpeg"),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.ensure_model_available", return_value="/models/tiny") as mock_ensure,
patch("local_transcriber.cli.load_model", return_value=(model, "cpu")),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch("local_transcriber.cli.write_transcript"),
):
runner.invoke(app, [str(audio)])
mock_ensure.assert_called_once_with("tiny", on_status=mock_ensure.call_args[1]["on_status"])
def test_cli_cli_overrides_config(tmp_path):
audio = tmp_path / "test.mp3"
audio.write_bytes(b"fake")
model = _make_model()
result = _make_result()
tfr = _make_tfr(result=result, model=model)
with (
patch("local_transcriber.cli.load_config", return_value={"model": "tiny"}),
patch("local_transcriber.cli.check_ffmpeg"),
patch("local_transcriber.cli.validate_input_file", return_value=audio),
patch("local_transcriber.cli.detect_device", return_value="cpu"),
patch("local_transcriber.cli.ensure_model_available", return_value="/models/small") as mock_ensure,
patch("local_transcriber.cli.load_model", return_value=(model, "cpu")),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch("local_transcriber.cli.write_transcript"),
):
runner.invoke(app, [str(audio), "--model", "small"])
mock_ensure.assert_called_once_with("small", on_status=mock_ensure.call_args[1]["on_status"])
def test_cli_batch_fallback_warning(tmp_path):
"""Batch mode shows fallback warning when load_model falls back to CPU."""
a = tmp_path / "a.mp3"
b = tmp_path / "b.mp3"
a.write_bytes(b"fake")
b.write_bytes(b"fake")
result = _make_result(device_used="cpu")
model = _make_model()
tfr = TranscribeFileResult(result=result, model=model, actual_device="cpu")
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
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.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu")),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch("local_transcriber.cli.write_transcript"),
):
out = runner.invoke(app, [str(a), str(b)])
assert "fallback" in out.output
def test_cli_batch_empty_speech_warning(tmp_path):
"""Batch mode warns when a file has no detected speech."""
a = tmp_path / "a.mp3"
b = tmp_path / "b.mp3"
a.write_bytes(b"fake")
b.write_bytes(b"fake")
result_empty = _make_result(segments=[])
result_ok = _make_result()
model = _make_model()
tfr_empty = _make_tfr(result=result_empty, model=model)
tfr_ok = _make_tfr(result=result_ok, model=model)
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
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.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.load_model", return_value=(model, "cpu")),
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)])
assert out.exit_code == 0
assert "Речь не обнаружена" in out.output
assert "2 обработано" in out.output
def test_cli_batch_midstream_fallback_warning(tmp_path):
"""Batch mode shows warning when _transcribe_file falls back mid-stream."""
a = tmp_path / "a.mp3"
b = tmp_path / "b.mp3"
a.write_bytes(b"fake")
b.write_bytes(b"fake")
model_gpu = _make_model()
model_cpu = _make_model()
result = _make_result(device_used="cpu")
# First file triggers mid-stream fallback
tfr_fallback = TranscribeFileResult(result=result, model=model_cpu, actual_device="cpu")
tfr_ok = TranscribeFileResult(result=result, model=model_cpu, actual_device="cpu")
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
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.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.load_model", return_value=(model_gpu, "cuda")),
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)])
assert "fallback" in out.output
assert "2 обработано" in out.output
def test_cli_batch_model_loaded_once(tmp_path):
a = tmp_path / "a.mp3"
b = tmp_path / "b.mp3"
a.write_bytes(b"fake")
b.write_bytes(b"fake")
result = _make_result()
model = _make_model()
tfr = _make_tfr(result=result, model=model)
mock_load_model = MagicMock(return_value=(model, "cpu"))
with (
patch("local_transcriber.cli.load_config", return_value={}),
patch("local_transcriber.cli.check_ffmpeg"),
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.ensure_model_available", return_value="/models/large-v3"),
patch("local_transcriber.cli.load_model", mock_load_model),
patch("local_transcriber.cli._transcribe_file", return_value=tfr),
patch("local_transcriber.cli.write_transcript"),
):
out = runner.invoke(app, [str(a), str(b)])
assert out.exit_code == 0
mock_load_model.assert_called_once()
tests/test_config.py
+96
View File
@@ -0,0 +1,96 @@
from pathlib import Path
from unittest.mock import patch
import pytest
from local_transcriber.config import (
find_config_file,
load_config,
resolve_defaults,
)
def test_find_config_file_cwd(tmp_path, monkeypatch):
config = tmp_path / ".transcriber.toml"
config.write_text('model = "small"\n')
monkeypatch.chdir(tmp_path)
assert find_config_file() == config
def test_find_config_file_user_home(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path) # no .transcriber.toml in CWD
global_config = tmp_path / ".config" / "transcriber" / "config.toml"
global_config.parent.mkdir(parents=True)
global_config.write_text('language = "ru"\n')
with patch("local_transcriber.config.Path.home", return_value=tmp_path):
result = find_config_file()
assert result == global_config
def test_find_config_file_none(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
with patch("local_transcriber.config.Path.home", return_value=tmp_path):
assert find_config_file() is None
def test_load_config_valid(tmp_path):
config = tmp_path / "config.toml"
config.write_text('model = "small"\nlanguage = "ru"\n')
result = load_config(config)
assert result == {"model": "small", "language": "ru"}
def test_load_config_malformed(tmp_path):
config = tmp_path / "config.toml"
config.write_text("this is not valid toml [[[")
with pytest.raises(ValueError, match="Ошибка чтения конфига"):
load_config(config)
def test_load_config_unknown_keys_warned(tmp_path):
config = tmp_path / "config.toml"
config.write_text('modle = "small"\nmodel = "tiny"\n')
with pytest.warns(UserWarning, match="modle"):
result = load_config(config)
assert result == {"model": "tiny"}
def test_load_config_non_string_value(tmp_path):
config = tmp_path / "config.toml"
config.write_text("device = 123\n")
with pytest.raises(ValueError, match="должно быть строкой"):
load_config(config)
def test_load_config_invalid_device(tmp_path):
config = tmp_path / "config.toml"
config.write_text('device = "tpu"\n')
with pytest.raises(ValueError, match="Недопустимое значение device"):
load_config(config)
def test_resolve_defaults_cli_wins():
config = {"model": "tiny", "language": "en"}
cli = {"model": "small", "language": None, "device": None, "compute_type": None}
result = resolve_defaults(cli, config)
assert result["model"] == "small"
assert result["language"] == "en"
def test_resolve_defaults_config_wins():
config = {"model": "tiny"}
cli = {"model": None, "language": None, "device": None, "compute_type": None}
result = resolve_defaults(cli, config)
assert result["model"] == "tiny"
def test_resolve_defaults_hardcoded_fallback():
result = resolve_defaults(
{"model": None, "language": None, "device": None, "compute_type": None}, {}
)
assert result == {
"model": "large-v3",
"language": "auto",
"device": "auto",
"compute_type": "int8",
}
tests/test_transcriber.py
+58 -1
View File
@@ -5,7 +5,14 @@ from unittest.mock import MagicMock, patch
import pytest
from huggingface_hub.errors import LocalEntryNotFoundError
from local_transcriber.transcriber import Segment, TranscribeResult, ensure_model_available, transcribe
from local_transcriber.transcriber import (
Segment,
TranscribeResult,
_transcribe_file,
ensure_model_available,
load_model,
transcribe,
)
def _make_raw_segments(count: int) -> list:
@@ -414,3 +421,53 @@ def test_transcribe_strict_cuda_error_during_transcription(mock_model_cls):
device="cuda",
strict_device=True,
)
# === load_model tests ===
@patch("local_transcriber.transcriber.WhisperModel")
def test_load_model_cuda_fallback(mock_model_cls):
cpu_instance = MagicMock()
def model_side_effect(model_name, device, compute_type):
if device == "cuda":
raise RuntimeError("CUDA out of memory")
return cpu_instance
mock_model_cls.side_effect = model_side_effect
with pytest.warns(UserWarning, match="Переключение на CPU"):
model, actual_device = load_model("tiny", "cuda", "int8")
assert actual_device == "cpu"
assert model is cpu_instance
@patch("local_transcriber.transcriber.WhisperModel")
def test_load_model_strict_raises(mock_model_cls):
mock_model_cls.side_effect = RuntimeError("CUDA out of memory")
with pytest.raises(RuntimeError, match="CUDA out of memory"):
load_model("tiny", "cuda", "int8", strict_device=True)
@patch("local_transcriber.transcriber.WhisperModel")
def test__transcribe_file_basic(mock_model_cls):
raw_segments = _make_raw_segments(2)
info = _make_info()
instance = MagicMock()
instance.transcribe.return_value = (iter(raw_segments), info)
tfr = _transcribe_file(
model=instance,
actual_device="cpu",
file_path=Path("test.mp3"),
model_name="tiny",
compute_type="int8",
)
assert len(tfr.result.segments) == 2
assert tfr.actual_device == "cpu"
assert tfr.model is instance
tests/test_utils.py
+44
View File
@@ -7,7 +7,9 @@ import pytest
from local_transcriber.utils import (
build_output_path,
detect_device,
expand_globs,
get_gpu_name,
has_existing_transcript,
validate_input_file,
)
@@ -82,3 +84,45 @@ def test_get_gpu_name_success():
with patch("subprocess.run", return_value=mock_result):
result = get_gpu_name()
assert result == "NVIDIA GeForce RTX 3060"
def test_expand_globs_no_patterns(tmp_path):
a = tmp_path / "a.mp3"
b = tmp_path / "b.mp3"
result = expand_globs([a, b])
assert result == [a, b]
def test_expand_globs_with_star(tmp_path):
(tmp_path / "x.mp3").write_bytes(b"fake")
(tmp_path / "y.mp3").write_bytes(b"fake")
(tmp_path / "z.txt").write_bytes(b"fake")
result = expand_globs([Path(str(tmp_path / "*.mp3"))])
assert len(result) == 2
assert all(p.suffix == ".mp3" for p in result)
def test_expand_globs_no_match(tmp_path):
result = expand_globs([Path(str(tmp_path / "*.wav"))])
assert result == []
def test_expand_globs_deduplicates(tmp_path):
f = tmp_path / "a.mp3"
f.write_bytes(b"fake")
result = expand_globs([f, Path(str(tmp_path / "*.mp3"))])
assert len(result) == 1
def test_has_existing_transcript_true(tmp_path):
audio = tmp_path / "meeting.mp3"
audio.write_bytes(b"fake")
transcript = tmp_path / "meeting-transcript.md"
transcript.write_text("content")
assert has_existing_transcript(audio) is True
def test_has_existing_transcript_false(tmp_path):
audio = tmp_path / "meeting.mp3"
audio.write_bytes(b"fake")
assert has_existing_transcript(audio) is False
uv.lock
Generated
+2
View File
@@ -303,6 +303,7 @@ dependencies = [
{ name = "nvidia-cublas-cu12", marker = "platform_machine == 'x86_64' and sys_platform == 'linux'" },
{ name = "rich" },
{ name = "socksio" },
{ name = "tomli", marker = "python_full_version < '3.11'" },
{ name = "typer" },
]
@@ -317,6 +318,7 @@ requires-dist = [
{ name = "nvidia-cublas-cu12", marker = "platform_machine == 'x86_64' and sys_platform == 'linux'", specifier = ">=12.4" },
{ name = "rich" },
{ name = "socksio", specifier = ">=1.0.0" },
{ name = "tomli", marker = "python_full_version < '3.11'", specifier = ">=2.0" },
{ name = "typer" },
]