From e28232ef50ae74e85726c4e5d97179bee7f1b29d Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Wed, 18 Mar 2026 20:57:19 +0300 Subject: [PATCH] =?UTF-8?q?feat(cli):=20=D0=B4=D0=BE=D0=B1=D0=B0=D0=B2?= =?UTF-8?q?=D0=BB=D0=B5=D0=BD=20=D0=B1=D0=B0=D1=82=D1=87-=D1=80=D0=B5?= =?UTF-8?q?=D0=B6=D0=B8=D0=BC=20=D0=B8=20=D0=BA=D0=BE=D0=BD=D1=84=D0=B8?= =?UTF-8?q?=D0=B3=D1=83=D1=80=D0=B0=D1=86=D0=B8=D0=BE=D0=BD=D0=BD=D1=8B?= =?UTF-8?q?=D0=B9=20=D1=84=D0=B0=D0=B9=D0=BB=20(=D1=88=D0=B0=D0=B3=D0=B8?= =?UTF-8?q?=208=E2=80=9312)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Зачем: - обработка нескольких файлов за один вызов с загрузкой модели один раз. - хранение дефолтов (модель, язык, устройство) в .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 --- README.md | 39 ++ docs/adr/002-batch-and-config.md | 57 +++ docs/plan.md | 63 ++++ pyproject.toml | 1 + src/local_transcriber/cli.py | 259 +++++++++++-- src/local_transcriber/config.py | 85 +++++ src/local_transcriber/transcriber.py | 60 ++- src/local_transcriber/utils.py | 22 ++ tests/test_cli.py | 543 +++++++++++++++++++++++---- tests/test_config.py | 96 +++++ tests/test_transcriber.py | 59 ++- tests/test_utils.py | 44 +++ uv.lock | 2 + 13 files changed, 1227 insertions(+), 103 deletions(-) create mode 100644 docs/adr/002-batch-and-config.md create mode 100644 src/local_transcriber/config.py create mode 100644 tests/test_config.py diff --git a/README.md b/README.md index 71db140..3187d5f 100644 --- a/README.md +++ b/README.md @@ -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` | — | Подробный вывод | ## Модели diff --git a/docs/adr/002-batch-and-config.md b/docs/adr/002-batch-and-config.md new file mode 100644 index 0000000..7dd8d7c --- /dev/null +++ b/docs/adr/002-batch-and-config.md @@ -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` флаг для перезаписи существующих транскриптов в батч-режиме diff --git a/docs/plan.md b/docs/plan.md index ca4cb48..897e7bb 100644 --- a/docs/plan.md +++ b/docs/plan.md @@ -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 содержит документацию по батч-режиму и конфигу. + + --- ## Инструкция для агента diff --git a/pyproject.toml b/pyproject.toml index e55289d..1c54cc5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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] diff --git a/src/local_transcriber/cli.py b/src/local_transcriber/cli.py index 8a0bf46..bf0f741 100644 --- a/src/local_transcriber/cli.py +++ b/src/local_transcriber/cli.py @@ -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() diff --git a/src/local_transcriber/config.py b/src/local_transcriber/config.py new file mode 100644 index 0000000..098381f --- /dev/null +++ b/src/local_transcriber/config.py @@ -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 diff --git a/src/local_transcriber/transcriber.py b/src/local_transcriber/transcriber.py index 0f55d82..3434d68 100644 --- a/src/local_transcriber/transcriber.py +++ b/src/local_transcriber/transcriber.py @@ -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( diff --git a/src/local_transcriber/utils.py b/src/local_transcriber/utils.py index d0cbd58..91abfc9 100644 --- a/src/local_transcriber/utils.py +++ b/src/local_transcriber/utils.py @@ -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() diff --git a/tests/test_cli.py b/tests/test_cli.py index aac8747..708853f 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -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() diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..20733f0 --- /dev/null +++ b/tests/test_config.py @@ -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", + } diff --git a/tests/test_transcriber.py b/tests/test_transcriber.py index 3ab7bdd..6775f48 100644 --- a/tests/test_transcriber.py +++ b/tests/test_transcriber.py @@ -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 diff --git a/tests/test_utils.py b/tests/test_utils.py index bb6689e..d49cd2b 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -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 diff --git a/uv.lock b/uv.lock index b89bc09..c5d54c4 100644 --- a/uv.lock +++ b/uv.lock @@ -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" }, ]