From 3d14ed7b86e8de4309843c0760f8c4420b0c6c0e Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Tue, 17 Mar 2026 23:38:32 +0300 Subject: [PATCH] =?UTF-8?q?feat(cli):=20=D1=80=D0=B5=D0=B0=D0=BB=D0=B8?= =?UTF-8?q?=D0=B7=D0=BE=D0=B2=D0=B0=D0=BD=20=D1=88=D0=B0=D0=B3=205=20?= =?UTF-8?q?=E2=80=94=20CLI-=D1=81=D0=B2=D1=8F=D0=B7=D0=BA=D0=B0=20=D0=B2?= =?UTF-8?q?=D1=81=D0=B5=D1=85=20=D0=BC=D0=BE=D0=B4=D1=83=D0=BB=D0=B5=D0=B9?= =?UTF-8?q?=20=D1=81=20=D0=B8=D1=81=D0=BF=D1=80=D0=B0=D0=B2=D0=BB=D0=B5?= =?UTF-8?q?=D0=BD=D0=B8=D1=8F=D0=BC=D0=B8=20=D0=B8=D0=B7=20=D1=80=D0=B5?= =?UTF-8?q?=D0=B2=D1=8C=D1=8E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Зачем: - шаг 5 плана: нужен рабочий CLI-happy path, связывающий utils / transcriber / formatter. - ревью этапов 4–5 выявило два medium-бага в formatter и отсутствие тестов для CLI. - Что: - cli.py: все опции по PRD 3.2 (--model, --language, --output, --device, --compute-type, --verbose), rich Status + stderr-консоль, предупреждение на пустую речь, статистика времени. - transcriber.py: добавлена ensure_model_available() с проверкой кэша HF и валидацией локальной директории; on_status callback для передачи прогресса в CLI; обработка ImportError при отсутствии socksio через SOCKS proxy. - formatter.py: исправлен overflow в format_timestamp (0.995 → 00:01.00 вместо 00:00.100); сегменты теперь пишутся с явным пробелом и strip() независимо от whisper-формата текста. - deps: добавлен socksio>=1.0.0 для поддержки SOCKS proxy при загрузке модели. - tests: test_cli.py (8 тестов на CLI-контракт), расширены test_formatter.py и test_transcriber.py. - Проверка: - uv run pytest — 42 passed. - uv run transcribe --help показывает все опции. --- docs/plan.md | 6 +- docs/review-stages-4-5-rerun.md | 43 +++++ docs/review-stages-4-5.md | 138 ++++++++++++++++ pyproject.toml | 1 + src/local_transcriber/cli.py | 68 +++++++- src/local_transcriber/formatter.py | 7 +- src/local_transcriber/transcriber.py | 128 ++++++++++++++- tests/test_cli.py | 236 +++++++++++++++++++++++++++ tests/test_formatter.py | 22 +++ tests/test_transcriber.py | 156 +++++++++++++++++- uv.lock | 11 ++ 11 files changed, 804 insertions(+), 12 deletions(-) create mode 100644 docs/review-stages-4-5-rerun.md create mode 100644 docs/review-stages-4-5.md create mode 100644 tests/test_cli.py diff --git a/docs/plan.md b/docs/plan.md index 451d091..f0777f5 100644 --- a/docs/plan.md +++ b/docs/plan.md @@ -216,7 +216,7 @@ > PRD-ссылки: 3.1 (flow), 3.2 (CLI-интерфейс) -- [ ] Typer command с аргументами и опциями по PRD 3.2: +- [x] Typer command с аргументами и опциями по PRD 3.2: - `file: Path` — позиционный аргумент - `--model` / `-m` → default `"large-v3"` - `--language` / `-l` → default `"auto"` @@ -224,7 +224,7 @@ - `--device` / `-d` → default `"auto"` - `--compute-type` → default `"int8"` - `--verbose` / `-v` → flag, default False -- [ ] Happy path flow: +- [x] Happy path flow: 1. `check_ffmpeg()` 2. `validate_input_file(file)` 3. `detect_device(device)` → получить device; `--compute-type` используется как есть (независим от device) @@ -236,7 +236,7 @@ 9. `write_transcript(...)` 10. `console.print("✓ Транскрипт сохранён: <путь>", style="green")` 11. Статистика: кол-во сегментов, время работы (замерить через `time.monotonic()`) -- [ ] Exit codes: 0 — успех (включая пустую речь), 1 — ошибка +- [x] Exit codes: 0 — успех (включая пустую речь), 1 — ошибка **Критерий готовности**: `uv run transcribe test.mp3` — создаёт корректный .md файл (проверить на локальной машине с реальным файлом). diff --git a/docs/review-stages-4-5-rerun.md b/docs/review-stages-4-5-rerun.md new file mode 100644 index 0000000..16a1f4c --- /dev/null +++ b/docs/review-stages-4-5-rerun.md @@ -0,0 +1,43 @@ +# Review: Stages 4-5 (Rerun) + +## Executive Summary + +| Severity | Count | +|----------|-------| +| CRITICAL | 0 | +| HIGH | 0 | +| MEDIUM | 0 | +| LOW | 0 | + +**Overall Risk:** LOW +**Recommendation:** APPROVE + +## What Was Rechecked + +- `src/local_transcriber/formatter.py` +- `src/local_transcriber/cli.py` +- `tests/test_formatter.py` +- `tests/test_cli.py` + +## Result + +No new findings. + +Previously reported issues for stages 4-5 are addressed: + +- centisecond carry in `format_timestamp()` is fixed +- transcript formatting no longer depends on leading whitespace in `seg.text` +- CLI now has automated tests for happy path, options, empty speech warning, output path handling, and error exit code + +## Verification + +- `uv run pytest` -> 31 passed +- `.venv/bin/transcribe --help` -> works +- spot checks: + - `format_timestamp(0.995)` -> `00:01.00` + - `format_timestamp(59.995)` -> `01:00.00` + - `format_timestamp(3599.995, use_hours=True)` -> `01:00:00.00` + +## Residual Risk + +- Step 6 error-handling polish is still not implemented, so user-facing error formatting remains intentionally incomplete at this stage diff --git a/docs/review-stages-4-5.md b/docs/review-stages-4-5.md new file mode 100644 index 0000000..bf7ca51 --- /dev/null +++ b/docs/review-stages-4-5.md @@ -0,0 +1,138 @@ +# Review: Stages 4-5 + +## Executive Summary + +| Severity | Count | +|----------|-------| +| CRITICAL | 0 | +| HIGH | 0 | +| MEDIUM | 2 | +| LOW | 1 | + +**Overall Risk:** MEDIUM +**Recommendation:** CONDITIONAL + +**Key Metrics:** +- Files analyzed: 4 +- Verified commands: `pytest`, CLI help, mocked CLI happy path +- Test coverage gaps: 1 user-facing module (`cli.py`) +- High blast radius changes: 0 +- Security regressions detected: 0 + +## What Changed + +**Commit Range:** `dfc5f46..WORKTREE` +**Commits:** `0d1a734`, plus uncommitted step 5 changes + +| File | Risk | Notes | +|------|------|-------| +| `src/local_transcriber/formatter.py` | MEDIUM | Output contract and markdown formatting | +| `tests/test_formatter.py` | LOW | Unit coverage for formatter | +| `src/local_transcriber/cli.py` | MEDIUM | Main user-facing flow and file writing | +| `docs/plan.md` | LOW | Checkbox updates for step 5 | + +## Findings + +### MEDIUM: `format_timestamp()` can emit invalid centiseconds like `.100` + +**File:** `src/local_transcriber/formatter.py:7` +**Test Coverage:** NO + +The implementation rounds centiseconds independently from the integral seconds: + +```python +total_seconds = int(seconds) +centiseconds = int(round((seconds - total_seconds) * 100)) +``` + +For values such as `0.995`, this produces `00:00.100` instead of carrying into the next second. + +**Reproduction:** +- `format_timestamp(0.995)` returns `00:00.100` + +**Impact:** +- Breaks the PRD timestamp format contract (`SS.ss` must always have exactly two fractional digits) +- Can produce malformed transcript timestamps on real segment boundaries + +**Recommendation:** +- Round the full timestamp first and then split into components, or normalize `centiseconds == 100` by incrementing seconds +- Add a regression test for `0.995` + +### MEDIUM: Transcript formatting depends on segment text already containing a leading space + +**File:** `src/local_transcriber/formatter.py:62` +**Test Coverage:** PARTIAL + +Segment lines are written as: + +```python +f"[{start} - {end}]{seg.text}" +``` + +This only matches the PRD format if `seg.text` already starts with a space. The current tests mask that dependency by building fixtures with leading spaces. + +**Reproduction:** +- A mocked CLI run with a segment text of `"Hello"` writes: + - `[00:00.00 - 00:01.00]Hello` +- Expected: + - `[00:00.00 - 00:01.00] Hello` + +**Impact:** +- Output format becomes model-dependent instead of being guaranteed by the formatter +- Any future normalization in `transcriber.py` will immediately break transcript formatting + +**Recommendation:** +- Normalize segment text inside the formatter, e.g. `seg.text.strip()` plus an explicit single space after `]` +- Add a test case where segment text has no leading whitespace + +### LOW: Step 5 has no automated tests for the public CLI contract + +**File:** `src/local_transcriber/cli.py:16` +**Test Coverage:** NO + +The user-facing entrypoint is now wired end to end, but there is still no `test_cli.py` coverage for: + +- option parsing +- warning path for empty speech +- default output-path generation +- exit-code behavior on failure + +This already matters because the mocked CLI happy path is what exposed the missing-space formatting bug above. + +## Test Coverage Analysis + +**Executed checks:** +- `uv run pytest` -> 22 tests passed +- `.venv/bin/transcribe --help` -> works +- mocked `CliRunner` happy path -> exit code 0 and output file written + +**Coverage gaps:** + +| Area | Gap | Risk | +|------|-----|------| +| `formatter.py` | No edge-case test for centisecond carry | MEDIUM | +| `formatter.py` | No test for segment text without leading whitespace | MEDIUM | +| `cli.py` | No automated tests at all | LOW | + +## Blast Radius Analysis + +The blast radius is still low because this is a small CLI project, but `cli.py` is now the single public entrypoint. Any formatting or wiring defect directly affects all users. + +| Function | Exposure | Risk | Priority | +|----------|----------|------|----------| +| `main()` | All CLI invocations | MEDIUM | P1 | +| `format_transcript()` | All saved transcript files | MEDIUM | P1 | +| `format_timestamp()` | Every segment line | MEDIUM | P1 | + +## Recommendations + +### Immediate + +- [ ] Fix centisecond carry handling in `format_timestamp()` +- [ ] Stop relying on leading whitespace in `seg.text` +- [ ] Add formatter regression tests for both cases + +### Before Step 6 + +- [ ] Add `tests/test_cli.py` with a mocked happy path and one error path +- [ ] Assert output file contents through the CLI layer, not only through direct formatter calls diff --git a/pyproject.toml b/pyproject.toml index 3eae697..bc1700b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,6 +7,7 @@ dependencies = [ "typer", "rich", "faster-whisper>=1.2.1", + "socksio>=1.0.0", ] [project.scripts] diff --git a/src/local_transcriber/cli.py b/src/local_transcriber/cli.py index f1bc64c..0548379 100644 --- a/src/local_transcriber/cli.py +++ b/src/local_transcriber/cli.py @@ -1,13 +1,77 @@ +import time from pathlib import Path import typer +from rich.console import Console +from rich.status import Status + +from .formatter import format_transcript, write_transcript +from .transcriber import Segment, ensure_model_available, transcribe +from .utils import build_output_path, check_ffmpeg, detect_device, get_gpu_name, validate_input_file app = typer.Typer() +console = Console(stderr=True) @app.command() -def main(file: Path) -> None: - typer.echo("TODO: not implemented") +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)"), + 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="Тип вычислений"), + verbose: bool = typer.Option(False, "--verbose", "-v", help="Подробный вывод"), +) -> None: + start = time.monotonic() + + check_ffmpeg() + validated_file = validate_input_file(file) + resolved_device = detect_device(device) + 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]") + + model_path = ensure_model_available(model, on_status=lambda message: console.print(message)) + + def on_segment(seg: Segment) -> None: + console.print(f" [{seg.start:.2f}s] {seg.text.strip()}") + + with Status("Подготавливаю запуск...", console=console) as status: + result = transcribe( + file_path=validated_file, + model_name=model_path, + device=resolved_device, + compute_type=compute_type, + language=language if language != "auto" else None, + on_segment=on_segment if verbose else None, + on_status=status.update, + ) + + if len(result.segments) == 0: + console.print(f"⚠ Речь не обнаружена в файле {validated_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" + + language_mode = "detected" if language == "auto" else "forced" + + content = format_transcript( + result=result, + source_filename=validated_file.name, + model_name=model, + device_info=device_info, + language_mode=language_mode, + ) + write_transcript(content, output_path) + + elapsed = time.monotonic() - start + console.print(f"✓ Транскрипт сохранён: [bold]{output_path}[/bold]", style="green") + console.print(f" Сегментов: {len(result.segments)} Время: {elapsed:.1f}с") if __name__ == "__main__": diff --git a/src/local_transcriber/formatter.py b/src/local_transcriber/formatter.py index 09bad1d..b887bab 100644 --- a/src/local_transcriber/formatter.py +++ b/src/local_transcriber/formatter.py @@ -5,8 +5,9 @@ from .transcriber import TranscribeResult def format_timestamp(seconds: float, use_hours: bool = False) -> str: - total_seconds = int(seconds) - centiseconds = int(round((seconds - total_seconds) * 100)) + total_cs = round(seconds * 100) + centiseconds = total_cs % 100 + total_seconds = total_cs // 100 if use_hours: hours = total_seconds // 3600 @@ -59,7 +60,7 @@ def format_transcript( start = format_timestamp(seg.start, use_hours=use_hours) end = format_timestamp(seg.end, use_hours=use_hours) lines.append("") - lines.append(f"[{start} - {end}]{seg.text}") + lines.append(f"[{start} - {end}] {seg.text.strip()}") lines.append("") return "\n".join(lines) diff --git a/src/local_transcriber/transcriber.py b/src/local_transcriber/transcriber.py index 319548f..032d841 100644 --- a/src/local_transcriber/transcriber.py +++ b/src/local_transcriber/transcriber.py @@ -4,6 +4,31 @@ from dataclasses import dataclass from pathlib import Path from faster_whisper import WhisperModel +from huggingface_hub import snapshot_download +from huggingface_hub.errors import LocalEntryNotFoundError + +MODEL_REPOS = { + "tiny": "Systran/faster-whisper-tiny", + "base": "Systran/faster-whisper-base", + "small": "Systran/faster-whisper-small", + "medium": "Systran/faster-whisper-medium", + "large-v3": "Systran/faster-whisper-large-v3", +} + +MODEL_ALLOW_PATTERNS = [ + "config.json", + "preprocessor_config.json", + "model.bin", + "tokenizer.json", + "vocabulary.*", +] + +MODEL_REQUIRED_FILES = [ + "config.json", + "preprocessor_config.json", + "model.bin", + "tokenizer.json", +] @dataclass @@ -29,12 +54,14 @@ def transcribe( compute_type: str = "int8", language: str | None = None, on_segment: Callable[[Segment], None] | None = None, + on_status: Callable[[str], None] | None = None, ) -> TranscribeResult: actual_device = device lang_arg = language if language and language != "auto" else None try: - model = WhisperModel(model_name, device=device, compute_type=compute_type) + _notify_status(on_status, f"Загружаю модель на {device}...") + model = _create_model(model_name, device, compute_type) except (RuntimeError, ValueError) as exc: if device != "cpu" and _is_cuda_error(exc): warnings.warn( @@ -43,11 +70,13 @@ def transcribe( stacklevel=2, ) actual_device = "cpu" - model = WhisperModel(model_name, device="cpu", compute_type=compute_type) + _notify_status(on_status, "Загружаю модель на cpu...") + model = _create_model(model_name, "cpu", compute_type) else: raise try: + _notify_status(on_status, "Транскрибирую...") segments, info = _run_transcription(model, file_path, lang_arg, on_segment) except (RuntimeError, ValueError) as exc: if actual_device != "cpu" and _is_cuda_error(exc): @@ -57,7 +86,9 @@ def transcribe( stacklevel=2, ) actual_device = "cpu" - model = WhisperModel(model_name, device="cpu", compute_type=compute_type) + _notify_status(on_status, "Загружаю модель на cpu...") + model = _create_model(model_name, "cpu", compute_type) + _notify_status(on_status, "Транскрибирую...") segments, info = _run_transcription(model, file_path, lang_arg, on_segment) else: raise @@ -71,6 +102,33 @@ def transcribe( ) +def ensure_model_available( + model_name: str, + on_status: Callable[[str], None] | None = None, +) -> str: + local_path = Path(model_name).expanduser() + if local_path.is_dir(): + _validate_model_dir(local_path) + return str(local_path) + + repo_id = _resolve_model_repo(model_name) + + try: + _notify_status(on_status, f"Проверяю кэш модели {model_name}...") + cached_path = Path(_snapshot_download(repo_id, local_files_only=True)) + _validate_model_dir(cached_path) + return str(cached_path) + except LocalEntryNotFoundError: + pass + except ValueError: + _notify_status(on_status, f"Кэш модели {model_name} неполный, докачиваю...") + + _notify_status(on_status, f"Скачиваю модель {model_name} из Hugging Face...") + downloaded_path = Path(_snapshot_download(repo_id, local_files_only=False)) + _validate_model_dir(downloaded_path) + return str(downloaded_path) + + def _run_transcription(model, file_path, lang_arg, on_segment): """Run model.transcribe and iterate segments. Returns (segments, info).""" segment_generator, info = model.transcribe(str(file_path), language=lang_arg) @@ -83,6 +141,70 @@ def _run_transcription(model, file_path, lang_arg, on_segment): return segments, info +def _create_model(model_name: str, device: str, compute_type: str): + try: + return WhisperModel(model_name, device=device, compute_type=compute_type) + except ImportError as exc: + if _is_missing_socksio_error(exc): + raise RuntimeError( + "Обнаружен SOCKS proxy, но не установлена зависимость `socksio`, " + "нужная для загрузки модели из Hugging Face через proxy. " + "Обновите окружение: `uv sync`." + ) from exc + raise + + def _is_cuda_error(exc: BaseException) -> bool: msg = str(exc).lower() return "cuda" in msg or "out of memory" in msg + + +def _is_missing_socksio_error(exc: BaseException) -> bool: + msg = str(exc).lower() + return "socks proxy" in msg and "socksio" in msg + + +def _notify_status(on_status: Callable[[str], None] | None, message: str) -> None: + if on_status is not None: + on_status(message) + + +def _resolve_model_repo(model_name: str) -> str: + if "/" in model_name: + return model_name + + repo_id = MODEL_REPOS.get(model_name) + if repo_id is None: + expected = ", ".join(MODEL_REPOS) + raise ValueError(f"Неподдерживаемая модель '{model_name}'. Ожидалось одно из: {expected}") + + return repo_id + + +def _snapshot_download(repo_id: str, local_files_only: bool) -> str: + try: + return snapshot_download( + repo_id, + local_files_only=local_files_only, + allow_patterns=MODEL_ALLOW_PATTERNS, + ) + except ImportError as exc: + if _is_missing_socksio_error(exc): + raise RuntimeError( + "Обнаружен SOCKS proxy, но не установлена зависимость `socksio`, " + "нужная для загрузки модели из Hugging Face через proxy. " + "Обновите окружение: `uv sync`." + ) from exc + raise + + +def _validate_model_dir(model_dir: Path) -> None: + missing = [ + filename for filename in MODEL_REQUIRED_FILES if not (model_dir / filename).exists() + ] + if not any(model_dir.glob("vocabulary.*")): + missing.append("vocabulary.*") + + if missing: + missing_str = ", ".join(missing) + raise ValueError(f"Неполная локальная модель в '{model_dir}': отсутствуют {missing_str}") diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..814563b --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,236 @@ +from pathlib import Path +from unittest.mock import MagicMock, patch + +from typer.testing import CliRunner + +from local_transcriber.cli import app +from local_transcriber.transcriber import Segment, TranscribeResult + +runner = CliRunner() + + +def _make_result(segments=None, language="ru", device_used="cpu", duration=60.0): + return TranscribeResult( + segments=[Segment(start=0.0, end=2.0, text="Hello")] if segments is None else segments, + language=language, + language_probability=0.95, + duration=duration, + device_used=device_used, + ) + + +def _patches(result=None, tmp_file=None): + """Context managers for a standard CLI happy path.""" + if result is None: + result = _make_result() + return [ + 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.write_transcript"), + ] + + +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"), + ): + out = runner.invoke(app, [str(audio)]) + + assert out.exit_code == 0 + + +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) + + 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", mock_transcribe), + patch("local_transcriber.cli.write_transcript"), + ): + runner.invoke(app, [str(audio)]) + + call_kwargs = mock_transcribe.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["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) + + 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="cuda"), + patch("local_transcriber.cli.ensure_model_available", return_value="/models/small"), + patch("local_transcriber.cli.transcribe", mock_transcribe), + patch("local_transcriber.cli.write_transcript"), + patch("local_transcriber.cli.get_gpu_name", return_value="RTX 3060"), + ): + runner.invoke(app, [ + str(audio), + "--model", "small", + "--language", "ru", + "--device", "cuda", + "--compute-type", "float16", + ]) + + call_kwargs = mock_transcribe.call_args[1] + assert call_kwargs["model_name"] == "/models/small" + assert call_kwargs["language"] == "ru" # explicit language passed through + assert call_kwargs["compute_type"] == "float16" + + +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) + + 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", mock_transcribe), + patch("local_transcriber.cli.write_transcript"), + ): + runner.invoke(app, [str(audio), "--verbose"]) + + call_kwargs = mock_transcribe.call_args[1] + assert call_kwargs["on_segment"] is not None + assert callable(call_kwargs["on_segment"]) + + +def test_cli_empty_speech_warning(tmp_path): + audio = tmp_path / "silence.wav" + 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"), + ): + out = runner.invoke(app, [str(audio)]) + + assert out.exit_code == 0 + assert "Речь не обнаружена" in out.output + + +def test_cli_default_output_path(tmp_path): + audio = tmp_path / "meeting.mp3" + audio.write_bytes(b"fake") + result = _make_result() + mock_write = MagicMock() + + 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", mock_write), + ): + runner.invoke(app, [str(audio)]) + + written_path: Path = mock_write.call_args[0][1] + assert written_path.name == "meeting-transcript.md" + + +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() + + 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", mock_write), + ): + runner.invoke(app, [str(audio), "--output", str(out_file)]) + + written_path: Path = mock_write.call_args[0][1] + assert written_path == out_file + + +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)): + out = runner.invoke(app, [str(audio)]) + + assert out.exit_code == 1 + + +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) + + 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", mock_transcribe), + patch("local_transcriber.cli.write_transcript"), + ): + runner.invoke(app, [str(audio)]) + + call_kwargs = mock_transcribe.call_args[1] + assert call_kwargs["on_status"] is not None + assert callable(call_kwargs["on_status"]) + + +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) + + 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") as mock_ensure_model, + patch("local_transcriber.cli.transcribe", mock_transcribe), + 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] + assert call_kwargs["model_name"] == "/models/large-v3" diff --git a/tests/test_formatter.py b/tests/test_formatter.py index a483ef4..efa41bd 100644 --- a/tests/test_formatter.py +++ b/tests/test_formatter.py @@ -14,6 +14,7 @@ def test_format_timestamp_minutes(): assert format_timestamp(83.45) == "01:23.45" assert format_timestamp(9.1) == "00:09.10" assert format_timestamp(599.99) == "09:59.99" + assert format_timestamp(0.995) == "00:01.00" # carry-over: не даёт .100 def test_format_timestamp_hours(): @@ -49,10 +50,31 @@ def test_format_transcript_basic(): assert "**Длительность**: 02:00" in content assert "**Устройство**: CUDA (NVIDIA GeForce RTX 3060)" in content assert "---" in content + # Проверяем пробел между ] и текстом независимо от ведущих пробелов в seg.text assert "[00:00.00 - 00:04.82] Добрый день, коллеги." in content assert "[00:04.82 - 00:09.15] Первый вопрос." in content +def test_format_transcript_segment_no_leading_space(): + """Сегменты без ведущего пробела должны форматироваться корректно.""" + result = TranscribeResult( + segments=[Segment(start=0.0, end=2.0, text="Hello")], + language="en", + language_probability=0.99, + duration=5.0, + device_used="cpu", + ) + content = format_transcript( + result, + source_filename="f.mp3", + model_name="tiny", + device_info="CPU", + language_mode="detected", + transcription_date=datetime(2026, 1, 1, 0, 0, 0), + ) + assert "[00:00.00 - 00:02.00] Hello" in content + + def test_format_transcript_empty(): result = TranscribeResult( segments=[], diff --git a/tests/test_transcriber.py b/tests/test_transcriber.py index b32a55e..9e5f1ff 100644 --- a/tests/test_transcriber.py +++ b/tests/test_transcriber.py @@ -3,8 +3,9 @@ from pathlib import Path from unittest.mock import MagicMock, patch import pytest +from huggingface_hub.errors import LocalEntryNotFoundError -from local_transcriber.transcriber import Segment, TranscribeResult, transcribe +from local_transcriber.transcriber import Segment, TranscribeResult, ensure_model_available, transcribe def _make_raw_segments(count: int) -> list: @@ -27,6 +28,16 @@ def _make_info(language: str = "ru", probability: float = 0.95, duration: float return info +def _create_model_dir(path: Path) -> Path: + path.mkdir(parents=True, exist_ok=True) + (path / "config.json").write_text("{}") + (path / "preprocessor_config.json").write_text("{}") + (path / "tokenizer.json").write_text("{}") + (path / "vocabulary.json").write_text("{}") + (path / "model.bin").write_bytes(b"ok") + return path + + @patch("local_transcriber.transcriber.WhisperModel") def test_transcribe_collects_segments(mock_model_cls): raw_segments = _make_raw_segments(3) @@ -201,3 +212,146 @@ def test_transcribe_midstream_fallback_no_duplicate_callbacks(mock_model_cls): # This is acceptable — on_segment is a live progress callback. # The important thing is that result.segments contains only CPU segments. assert all(s.text.startswith(" Segment") for s in result.segments) + + +@patch("local_transcriber.transcriber.WhisperModel") +def test_transcribe_reports_missing_socksio_for_proxy(mock_model_cls): + mock_model_cls.side_effect = ImportError( + "Using SOCKS proxy, but the 'socksio' package is not installed." + ) + + with pytest.raises(RuntimeError, match="socksio"): + transcribe( + file_path=Path("test.mp3"), + model_name="tiny", + device="cpu", + ) + + +@patch("local_transcriber.transcriber.WhisperModel") +def test_transcribe_reports_status_transitions(mock_model_cls): + raw_segments = _make_raw_segments(1) + info = _make_info() + + instance = MagicMock() + instance.transcribe.return_value = (iter(raw_segments), info) + mock_model_cls.return_value = instance + + statuses: list[str] = [] + + transcribe( + file_path=Path("test.mp3"), + model_name="tiny", + device="cpu", + on_status=statuses.append, + ) + + assert statuses == [ + "Загружаю модель на cpu...", + "Транскрибирую...", + ] + + +@patch("local_transcriber.transcriber.snapshot_download") +def test_ensure_model_available_uses_cache_first(mock_snapshot_download, tmp_path): + model_dir = _create_model_dir(tmp_path / "cache-model") + mock_snapshot_download.return_value = str(model_dir) + + result = ensure_model_available("large-v3") + + assert result == str(model_dir) + mock_snapshot_download.assert_called_once_with( + "Systran/faster-whisper-large-v3", + local_files_only=True, + allow_patterns=[ + "config.json", + "preprocessor_config.json", + "model.bin", + "tokenizer.json", + "vocabulary.*", + ], + ) + + +@patch("local_transcriber.transcriber._validate_model_dir") +@patch("local_transcriber.transcriber.snapshot_download") +def test_ensure_model_available_downloads_on_cache_miss(mock_snapshot_download, mock_validate_model_dir): + mock_snapshot_download.side_effect = [ + LocalEntryNotFoundError("not cached"), + "/downloaded/model", + ] + statuses: list[str] = [] + + result = ensure_model_available("large-v3", on_status=statuses.append) + + assert result == "/downloaded/model" + assert mock_snapshot_download.call_args_list[0].kwargs["local_files_only"] is True + assert mock_snapshot_download.call_args_list[1].kwargs["local_files_only"] is False + assert statuses == [ + "Проверяю кэш модели large-v3...", + "Скачиваю модель large-v3 из Hugging Face...", + ] + + +def test_ensure_model_available_accepts_local_directory(tmp_path): + model_dir = _create_model_dir(tmp_path / "model") + + result = ensure_model_available(str(model_dir)) + + assert result == str(model_dir) + + +def test_ensure_model_available_accepts_repo_id(tmp_path): + model_dir = _create_model_dir(tmp_path / "repo-model") + with patch("local_transcriber.transcriber.snapshot_download", return_value=str(model_dir)) as mock_snapshot_download: + result = ensure_model_available("org/model") + + assert result == str(model_dir) + assert mock_snapshot_download.call_args.kwargs["local_files_only"] is True + + +def test_ensure_model_available_rejects_unsupported_alias(): + with pytest.raises(ValueError, match="Неподдерживаемая модель"): + ensure_model_available("distil-large-v3") + + +@patch("local_transcriber.transcriber.snapshot_download") +def test_ensure_model_available_redownloads_incomplete_cache(mock_snapshot_download, tmp_path): + incomplete = tmp_path / "incomplete" + incomplete.mkdir() + (incomplete / "config.json").write_text("{}") + (incomplete / "preprocessor_config.json").write_text("{}") + (incomplete / "tokenizer.json").write_text("{}") + (incomplete / "vocabulary.json").write_text("{}") + + complete = tmp_path / "complete" + complete.mkdir() + (complete / "config.json").write_text("{}") + (complete / "preprocessor_config.json").write_text("{}") + (complete / "tokenizer.json").write_text("{}") + (complete / "vocabulary.json").write_text("{}") + (complete / "model.bin").write_bytes(b"ok") + + mock_snapshot_download.side_effect = [ + str(incomplete), + str(complete), + ] + statuses: list[str] = [] + + result = ensure_model_available("large-v3", on_status=statuses.append) + + assert result == str(complete) + assert statuses == [ + "Проверяю кэш модели large-v3...", + "Кэш модели large-v3 неполный, докачиваю...", + "Скачиваю модель large-v3 из Hugging Face...", + ] + + +def test_ensure_model_available_rejects_incomplete_local_directory(tmp_path): + model_dir = tmp_path / "model" + model_dir.mkdir() + (model_dir / "config.json").write_text("{}") + + with pytest.raises(ValueError, match="Неполная локальная модель"): + ensure_model_available(str(model_dir)) diff --git a/uv.lock b/uv.lock index 51cdc58..b6c8429 100644 --- a/uv.lock +++ b/uv.lock @@ -301,6 +301,7 @@ source = { editable = "." } dependencies = [ { name = "faster-whisper" }, { name = "rich" }, + { name = "socksio" }, { name = "typer" }, ] @@ -313,6 +314,7 @@ dev = [ requires-dist = [ { name = "faster-whisper", specifier = ">=1.2.1" }, { name = "rich" }, + { name = "socksio", specifier = ">=1.0.0" }, { name = "typer" }, ] @@ -690,6 +692,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/e0/f9/0595336914c5619e5f28a1fb793285925a8cd4b432c9da0a987836c7f822/shellingham-1.5.4-py2.py3-none-any.whl", hash = "sha256:7ecfff8f2fd72616f7481040475a65b2bf8af90a56c89140852d1120324e8686", size = 9755, upload-time = "2023-10-24T04:13:38.866Z" }, ] +[[package]] +name = "socksio" +version = "1.0.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f8/5c/48a7d9495be3d1c651198fd99dbb6ce190e2274d0f28b9051307bdec6b85/socksio-1.0.0.tar.gz", hash = "sha256:f88beb3da5b5c38b9890469de67d0cb0f9d494b78b106ca1845f96c10b91c4ac", size = 19055, upload-time = "2020-04-17T15:50:34.664Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/37/c3/6eeb6034408dac0fa653d126c9204ade96b819c936e136c5e8a6897eee9c/socksio-1.0.0-py3-none-any.whl", hash = "sha256:95dc1f15f9b34e8d7b16f06d74b8ccf48f609af32ab33c608d08761c5dcbb1f3", size = 12763, upload-time = "2020-04-17T15:50:31.878Z" }, +] + [[package]] name = "sympy" version = "1.14.0"