docs: добавлены docstrings и расширен CONTRIBUTING.md

- Зачем:
  - покрытие документацией было неравномерным, отсутствовало архитектурное описание
- Что:
  - добавлены модульные docstrings во все 5 модулей (utils, config, formatter, transcriber, cli)
  - добавлены docstrings для всех публичных функций без документации
  - добавлены inline-комментарии для неочевидной логики (CUDA fallback, strict device, glob, SOCKS proxy)
  - CONTRIBUTING.md: секции «Архитектура», «Ключевые решения», «Тестирование», «Частые задачи»
  - обновлено правило языка комментариев (русский вместо английского)
- Проверка:
  - uv run pytest (97 passed, 1 skipped)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-18 23:25:17 +03:00
co-authored by Claude Opus 4.6
parent 87f7dc1723
commit ac8ea508ce
6 changed files with 100 additions and 2 deletions
+11 -1
View File
@@ -1,3 +1,5 @@
"""CLI-точка входа (typer). Single и batch режимы транскрипции."""
import sys
import time
from pathlib import Path
@@ -49,6 +51,10 @@ def main(
verbose: bool = typer.Option(False, "--verbose", "-v", help="Подробный вывод"),
force: bool = typer.Option(False, "--force", "-f", help="Перезаписать существующие транскрипты"),
) -> None:
"""Транскрибирует аудио/видеофайлы в markdown с таймкодами.
Каскад приоритетов параметров: CLI-флаги > .transcriber.toml > device-aware дефолты.
"""
try:
config = load_config()
cli_values = {"model": model, "language": language, "device": device, "compute_type": compute_type}
@@ -108,11 +114,13 @@ def _run_single(
output: Path | None,
verbose: bool,
) -> None:
"""Пайплайн одного файла: валидация → модель → транскрипция → запись."""
start = time.monotonic()
validated_file = validate_input_file(file)
requested_device = defaults["device"]
resolved_device = detect_device(requested_device)
# Если пользователь явно указал устройство — запрещаем fallback на CPU
strict = requested_device != "auto"
output_path = build_output_path(validated_file, output)
@@ -196,7 +204,8 @@ def _run_batch(
verbose: bool,
force: bool,
) -> None:
# Phase 1: Prescan
"""Трёхфазный батч-пайплайн: prescan → загрузка модели → транскрипция."""
# Phase 1: Prescan — fail-fast + skip до загрузки модели (экономим ~2-5 сек)
to_process: list[Path] = []
skipped = 0
invalid = 0
@@ -282,6 +291,7 @@ def _run_batch(
f" {file.name}: fallback на {tfr.actual_device} при транскрипции",
style="yellow",
)
# Обновляем после возможного mid-stream fallback на CPU
model_obj, actual_device = tfr.model, tfr.actual_device
result = tfr.result
+10
View File
@@ -1,3 +1,5 @@
"""Загрузка конфигурации из ``.transcriber.toml`` и каскад приоритетов."""
import sys
import warnings
from pathlib import Path
@@ -19,11 +21,13 @@ DEVICE_DEFAULTS: dict[str, dict[str, str]] = {
"cpu": {"model": "medium", "compute_type": "float32"},
}
# Одно место правды для допустимых ключей конфига
_VALID_KEYS = set(HARDCODED_DEFAULTS)
_VALID_DEVICES = {"auto", "cpu", "cuda"}
def find_config_file() -> Path | None:
"""Ищет конфиг: сначала ``.transcriber.toml`` в cwd, затем ``~/.config/transcriber/config.toml``."""
cwd_config = Path.cwd() / ".transcriber.toml"
if cwd_config.is_file():
return cwd_config
@@ -36,6 +40,11 @@ def find_config_file() -> Path | None:
def load_config(path: Path | None = None) -> dict[str, str]:
"""Загружает и валидирует TOML-конфиг.
Неизвестные ключи вызывают предупреждение (а не ошибку) для forward
compatibility: новые версии могут добавить ключи, которых ещё нет в текущей.
"""
if path is None:
path = find_config_file()
if path is None:
@@ -78,6 +87,7 @@ def load_config(path: Path | None = None) -> dict[str, str]:
def resolve_defaults(
cli_values: dict[str, str | None], config: dict[str, str]
) -> dict[str, str]:
"""Каскад приоритетов: CLI > конфиг-файл > hardcoded-дефолты."""
result: dict[str, str] = {}
for key in HARDCODED_DEFAULTS:
cli_val = cli_values.get(key)
+9
View File
@@ -1,3 +1,5 @@
"""Формирование markdown-транскрипта из результатов распознавания."""
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
@@ -42,6 +44,10 @@ def _group_segments(segments: list[Segment]) -> list[_Paragraph]:
def format_timestamp(seconds: float, use_hours: bool = False) -> str:
"""Форматирует время в ``MM:SS.cc`` или ``HH:MM:SS.cc``.
Сотые доли (centiseconds) — максимальная точность, которую даёт Whisper.
"""
total_cs = round(seconds * 100)
centiseconds = total_cs % 100
total_seconds = total_cs // 100
@@ -58,6 +64,7 @@ def format_timestamp(seconds: float, use_hours: bool = False) -> str:
def _format_duration(seconds: float) -> str:
"""Человекочитаемая длительность для метаданных в шапке транскрипта."""
total = int(seconds)
h = total // 3600
m = (total % 3600) // 60
@@ -75,6 +82,7 @@ def format_transcript(
language_mode: str, # "detected" | "forced"
transcription_date: datetime | None = None, # None -> datetime.now()
) -> str:
"""Собирает markdown-транскрипт: шапка с метаданными + абзацы с таймкодами."""
date = transcription_date or datetime.now()
use_hours = result.duration > 3600
@@ -104,5 +112,6 @@ def format_transcript(
def write_transcript(content: str, output_path: Path) -> None:
"""Записывает готовый транскрипт в файл."""
with open(output_path, "w", encoding="utf-8") as f:
f.write(content)
+16
View File
@@ -1,3 +1,5 @@
"""Обёртка над faster-whisper: загрузка моделей, транскрипция, CUDA fallback."""
import warnings
from collections.abc import Callable
from dataclasses import dataclass
@@ -20,6 +22,8 @@ MODEL_REPOS = {
"large-v3": "Systran/faster-whisper-large-v3",
}
# allow — фильтр для snapshot_download (какие файлы скачивать из репозитория);
# required — для валидации (что обязано быть после скачивания/в локальной модели)
MODEL_ALLOW_PATTERNS = [
"config.json",
"preprocessor_config.json",
@@ -71,6 +75,7 @@ def load_model(
_notify_status(on_status, f"Инициализирую модель на {device}...")
model = _create_model(model_name, device, compute_type)
except (RuntimeError, ValueError) as exc:
# strict — пользователь явно указал устройство, fallback запрещён
if device != "cpu" and _is_cuda_error(exc):
if strict_device:
raise
@@ -105,6 +110,8 @@ def _transcribe_file(
_notify_status(on_status, "Транскрибирую...")
segments, info = _run_transcription(model, file_path, lang_arg, on_segment, on_status)
except (RuntimeError, ValueError) as exc:
# Mid-stream fallback: GPU может упасть с OOM уже во время транскрипции,
# поэтому перезагружаем модель на CPU и начинаем сначала
if actual_device != "cpu" and _is_cuda_error(exc):
if strict_device:
raise
@@ -141,6 +148,7 @@ def transcribe(
on_status: Callable[[str], None] | None = None,
strict_device: bool = False,
) -> TranscribeResult:
"""High-level API: загрузка модели + транскрипция за один вызов."""
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,
@@ -153,6 +161,11 @@ def ensure_model_available(
model_name: str,
on_status: Callable[[str], None] | None = None,
) -> str:
"""Резолвит alias модели в repo_id и гарантирует наличие файлов.
Стратегия: cache-first (``local_files_only=True``), затем download.
Два вызова ``snapshot_download`` — чтобы не лезть в сеть, если модель уже в кэше.
"""
local_path = Path(model_name).expanduser()
if local_path.is_dir():
_validate_model_dir(local_path)
@@ -198,6 +211,9 @@ 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:
# WhisperModel при инициализации может загружать файлы через HF Hub;
# если в системе настроен SOCKS proxy, но socksio не установлен,
# HF Hub бросает ImportError — оборачиваем в понятное сообщение
if _is_missing_socksio_error(exc):
raise RuntimeError(
"Обнаружен SOCKS proxy, но не установлена зависимость `socksio`, "
+21
View File
@@ -1,3 +1,5 @@
"""Утилиты для валидации входных файлов, определения устройства и работы с путями."""
import glob
import shutil
import subprocess
@@ -11,6 +13,11 @@ SUPPORTED_EXTENSIONS = {
def detect_device(requested: str = "auto") -> str:
"""Определяет устройство для вычислений.
При ``requested="auto"`` проверяет наличие ``nvidia-smi`` в PATH
и возвращает ``"cuda"`` или ``"cpu"``. Явное значение возвращается как есть.
"""
if requested != "auto":
return requested
if shutil.which("nvidia-smi") is not None:
@@ -19,6 +26,7 @@ def detect_device(requested: str = "auto") -> str:
def get_gpu_name() -> str | None:
"""Возвращает название GPU через ``nvidia-smi`` (для метаданных транскрипта)."""
try:
result = subprocess.run(
["nvidia-smi", "--query-gpu=name", "--format=csv,noheader"],
@@ -37,6 +45,12 @@ def get_gpu_name() -> str | None:
def validate_input_file(path: Path) -> Path:
"""Проверяет существование, непустоту и расширение файла.
Неизвестное расширение — предупреждение, а не ошибка: файл может оказаться
корректным контейнером, просто с нестандартным суффиксом.
Возвращает resolved-путь.
"""
if not path.exists():
raise FileNotFoundError(f"Файл не найден: {path}")
if not path.is_file():
@@ -53,12 +67,18 @@ def validate_input_file(path: Path) -> Path:
def build_output_path(input_path: Path, output: Path | None = None) -> Path:
"""Строит путь выходного файла: ``<имя>-transcript.md`` рядом с исходным."""
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]:
"""Раскрывает glob-паттерны в списке путей с дедупликацией.
Typer на Windows не раскрывает ``*.mp4`` самостоятельно,
поэтому глобы обрабатываются вручную. Дедупликация — по resolved-пути.
"""
seen: set[Path] = set()
result: list[Path] = []
for p in paths:
@@ -76,4 +96,5 @@ def expand_globs(paths: list[Path]) -> list[Path]:
def has_existing_transcript(input_path: Path) -> bool:
"""Проверяет наличие транскрипта для skip-логики батч-режима."""
return build_output_path(input_path).exists()