From b1dfd9dcca6e173fe8af0c6bf4f1e3e926594396 Mon Sep 17 00:00:00 2001 From: Dmitriy Dementiev Date: Thu, 9 Jul 2026 22:21:07 +0300 Subject: [PATCH] =?UTF-8?q?feat(quality):=20=D0=BF=D1=80=D0=B5=D0=B4=D1=83?= =?UTF-8?q?=D0=BF=D1=80=D0=B5=D0=B6=D0=B4=D0=B5=D0=BD=D0=B8=D1=8F=20=D0=BE?= =?UTF-8?q?=20=D0=B2=D0=BE=D0=B7=D0=BC=D0=BE=D0=B6=D0=BD=D0=BE=D0=B9=20?= =?UTF-8?q?=D0=BF=D0=BE=D1=82=D0=B5=D1=80=D0=B5=20=D1=81=D0=BE=D0=B4=D0=B5?= =?UTF-8?q?=D1=80=D0=B6=D0=B0=D0=BD=D0=B8=D1=8F=20=D1=82=D1=80=D0=B0=D0=BD?= =?UTF-8?q?=D1=81=D0=BA=D1=80=D0=B8=D0=BF=D1=82=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Зачем: - транскрипция может тихо терять содержание: блоки галлюцинированных повторов и обрыв распознавания до конца файла выглядят как законченный транскрипт (класс ошибок из ADR-006), и читатель об этом не узнаёт. - Что: - новый модуль quality.py: детектор разрыва в хвосте (строго >120 с) и детектор блоков повторяющихся сегментов (серия >=4 для длинного текста, >=10 для короткого вроде «ага»). - предупреждения жёлтым в консоли (single и batch) и строками «Внимание» в шапке markdown-транскрипта; _format_duration переименована в публичную format_duration. - 27 новых тестов: границы порогов, нормализация текста, шапка formatter, точные строки CLI-сообщений с обрезкой «(+ ещё N)». - Проверка: - uv run pytest — 218 passed, 1 skipped. - ручной прогон записи 23:01 (gigaam-v3/onnx) — транскрипт полный, предупреждений нет. Co-Authored-By: Claude Fable 5 --- src/local_transcriber/cli.py | 69 ++++++++++++- src/local_transcriber/formatter.py | 18 +++- src/local_transcriber/quality.py | 78 +++++++++++++++ tests/test_cli.py | 150 +++++++++++++++++++++++++++++ tests/test_formatter.py | 138 ++++++++++++++++++++++++++ tests/test_quality.py | 125 ++++++++++++++++++++++++ 6 files changed, 575 insertions(+), 3 deletions(-) create mode 100644 src/local_transcriber/quality.py create mode 100644 tests/test_quality.py diff --git a/src/local_transcriber/cli.py b/src/local_transcriber/cli.py index 59e26f2..62bc80e 100644 --- a/src/local_transcriber/cli.py +++ b/src/local_transcriber/cli.py @@ -11,9 +11,21 @@ from rich.status import Status from .config import apply_device_defaults, load_config, resolve_defaults from .context_menu import install_menu as install_context_menu from .context_menu import uninstall_menu as uninstall_context_menu -from .formatter import format_transcript, write_transcript +from .formatter import ( + format_duration, + format_timestamp, + format_transcript, + write_transcript, +) +from .quality import ( + TAIL_GAP_WARN_S, + RepetitionBlock, + find_repetition_blocks, + tail_gap, +) from .transcriber import ( Segment, + TranscribeResult, _is_cuda_error, _transcribe_file, load_model, @@ -45,6 +57,59 @@ def _format_device_info(device_used: str) -> str: return "CPU" +def _format_repetition_blocks( + blocks: list[RepetitionBlock], + use_hours: bool, +) -> str: + """Формирует краткое описание блоков повторов для консоли.""" + rendered = [ + f"[{format_timestamp(block.start, use_hours=use_hours)} - " + f"{format_timestamp(block.end, use_hours=use_hours)}] ({block.count}×)" + for block in blocks[:3] + ] + summary = "; ".join(rendered) + remaining = len(blocks) - 3 + if remaining > 0: + summary = f"{summary} (+ ещё {remaining})" + return summary + + +def _print_quality_warnings(result: TranscribeResult, file_name: str | None = None) -> None: + """Печатает предупреждения о возможной потере содержания.""" + is_batch = file_name is not None + use_hours = result.duration > 3600 + + gap = tail_gap(result) + if gap > TAIL_GAP_WARN_S: + covered = format_duration(result.segments[-1].end) + total = format_duration(result.duration) + message = ( + f"транскрипт покрывает {covered} из {total} — " + "возможна потеря хвоста записи" + ) + if is_batch: + console.print(f" {file_name}: {message}", style="yellow") + else: + console.print( + f"Внимание: {message}. Попробуйте другой --device.", + style="yellow", + ) + + blocks = find_repetition_blocks(result.segments) + if blocks: + message = ( + f"блоки повторов: {_format_repetition_blocks(blocks, use_hours)} " + "— возможны галлюцинации модели" + ) + if is_batch: + console.print(f" {file_name}: {message}", style="yellow") + else: + console.print( + f"Внимание: {message}. Попробуйте другой --device.", + style="yellow", + ) + + @app.command() def main( files: list[Path] | None = typer.Argument(None, help="Пути к аудио/видеофайлам"), @@ -257,6 +322,7 @@ def _run_single( elapsed = time.monotonic() - start console.print(f"Транскрипт сохранён: \"{output_path}\"", style="green") console.print(f" Сегментов: {len(result.segments)} Время: {elapsed:.1f}с") + _print_quality_warnings(result) def _run_batch( @@ -392,6 +458,7 @@ def _run_batch( style="green", ) processed += 1 + _print_quality_warnings(result, file.name) except KeyboardInterrupt: raise except Exception as exc: diff --git a/src/local_transcriber/formatter.py b/src/local_transcriber/formatter.py index 0233d51..9e0240a 100644 --- a/src/local_transcriber/formatter.py +++ b/src/local_transcriber/formatter.py @@ -4,6 +4,7 @@ from dataclasses import dataclass from datetime import datetime from pathlib import Path +from .quality import TAIL_GAP_WARN_S, find_repetition_blocks, tail_gap from .types import Segment, TranscribeResult _PAUSE_THRESHOLD_S = 2.0 # пауза между сегментами для разбиения на абзацы @@ -63,7 +64,7 @@ def format_timestamp(seconds: float, use_hours: bool = False) -> str: return f"{minutes:02d}:{secs:02d}.{centiseconds:02d}" -def _format_duration(seconds: float) -> str: +def format_duration(seconds: float) -> str: """Человекочитаемая длительность для метаданных в шапке транскрипта.""" total = int(seconds) h = total // 3600 @@ -92,7 +93,20 @@ def format_transcript( lines.append(f"- **Дата транскрипции**: {date.strftime('%Y-%m-%d %H:%M:%S')}") lines.append(f"- **Модель**: {model_name}") lines.append(f"- **Язык**: {result.language} ({language_mode})") - lines.append(f"- **Длительность**: {_format_duration(result.duration)}") + lines.append(f"- **Длительность**: {format_duration(result.duration)}") + if tail_gap(result) > TAIL_GAP_WARN_S: + last_end = result.segments[-1].end + lines.append( + f"- **Внимание**: транскрипт покрывает {format_duration(last_end)} " + f"из {format_duration(result.duration)} — возможна потеря хвоста записи" + ) + for block in find_repetition_blocks(result.segments): + start = format_timestamp(block.start, use_hours=use_hours) + end = format_timestamp(block.end, use_hours=use_hours) + lines.append( + f"- **Внимание**: повторы в [{start} - {end}] ({block.count}×) " + "— возможны галлюцинации модели" + ) lines.append(f"- **Устройство**: {device_info}") lines.append("") lines.append("---") diff --git a/src/local_transcriber/quality.py b/src/local_transcriber/quality.py new file mode 100644 index 0000000..3acf84c --- /dev/null +++ b/src/local_transcriber/quality.py @@ -0,0 +1,78 @@ +"""Эвристики качества транскрипта.""" + +from dataclasses import dataclass + +from .types import Segment, TranscribeResult + +TAIL_GAP_WARN_S = 120.0 +REPETITION_MIN_RUN = 4 +REPETITION_MIN_RUN_SHORT = 10 +REPETITION_MIN_LEN = 6 + +_PUNCTUATION_TO_REMOVE = ".,!?…:;—–-\"'«»()[]<>" +_REMOVE_PUNCTUATION = str.maketrans("", "", _PUNCTUATION_TO_REMOVE) + + +@dataclass +class RepetitionBlock: + start: float + end: float + count: int + text: str + + +def tail_gap(result: TranscribeResult) -> float: + """Возвращает непокрытый хвост записи в секундах.""" + if not result.segments: + return 0.0 + return max(0.0, result.duration - result.segments[-1].end) + + +def _normalize(text: str) -> str: + """Нормализует текст сегмента для поиска межсегментных повторов.""" + text = text.casefold() + text = text.translate(_REMOVE_PUNCTUATION) + return " ".join(text.split()) + + +def find_repetition_blocks(segments: list[Segment]) -> list[RepetitionBlock]: + """Находит серии подряд идущих одинаковых сегментов.""" + blocks: list[RepetitionBlock] = [] + run_start = 0 + run_norm = "" + + def append_run(run_end: int) -> None: + count = run_end - run_start + if not run_norm: + return + min_run = ( + REPETITION_MIN_RUN + if len(run_norm) >= REPETITION_MIN_LEN + else REPETITION_MIN_RUN_SHORT + ) + if count >= min_run: + blocks.append( + RepetitionBlock( + start=segments[run_start].start, + end=segments[run_end - 1].end, + count=count, + text=segments[run_start].text, + ) + ) + + for index, segment in enumerate(segments): + norm = _normalize(segment.text) + if index == 0: + run_start = 0 + run_norm = norm + continue + if norm == run_norm: + continue + append_run(index) + run_start = index + run_norm = norm + + if segments: + append_run(len(segments)) + + return blocks diff --git a/tests/test_cli.py b/tests/test_cli.py index c2c934c..4fae93a 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -2,6 +2,7 @@ from pathlib import Path from unittest.mock import MagicMock, patch import pytest +from rich.console import Console from typer.testing import CliRunner from local_transcriber.cli import _format_device_info, app @@ -1018,3 +1019,152 @@ def test_cli_menu_runtime_error_has_no_verbose_hint(): assert out.exit_code == 1 assert "нет APPDATA" in out.output assert "--verbose" not in out.output + + +def test_cli_tail_gap_quality_warning_single(tmp_path): + audio = tmp_path / "tail.mp3" + audio.write_bytes(b"fake") + result = _make_result( + segments=[Segment(start=0.0, end=60.0, text="Фраза")], + duration=600.0, + ) + + patches = _single_patches(result=result, tmp_file=audio) + with ( + patches[0], + patches[1], + patches[2], + patches[3], + patches[4], + patches[5], + patch("local_transcriber.cli.console", Console(stderr=True, width=1000)), + ): + out = runner.invoke(app, [str(audio)]) + + assert out.exit_code == 0 + assert ( + "Внимание: транскрипт покрывает 01:00 из 10:00 — " + "возможна потеря хвоста записи. Попробуйте другой --device." + ) in out.output + + +def test_cli_repetition_quality_warning_single(tmp_path): + audio = tmp_path / "repeat.mp3" + audio.write_bytes(b"fake") + result = _make_result( + segments=[ + Segment(start=10.0, end=11.0, text="Повторяемая фраза"), + Segment(start=11.0, end=12.0, text="повторяемая фраза"), + Segment(start=12.0, end=13.0, text="повторяемая фраза"), + Segment(start=13.0, end=14.0, text="повторяемая фраза"), + ], + duration=60.0, + ) + + patches = _single_patches(result=result, tmp_file=audio) + with ( + patches[0], + patches[1], + patches[2], + patches[3], + patches[4], + patches[5], + patch("local_transcriber.cli.console", Console(stderr=True, width=1000)), + ): + out = runner.invoke(app, [str(audio)]) + + assert out.exit_code == 0 + assert ( + "Внимание: блоки повторов: [00:10.00 - 00:14.00] (4×) — " + "возможны галлюцинации модели. Попробуйте другой --device." + ) in out.output + + +def test_cli_quality_warning_batch_includes_file_name(tmp_path): + a = tmp_path / "a.mp3" + b = tmp_path / "b.mp3" + a.write_bytes(b"fake") + b.write_bytes(b"fake") + + result_warn = _make_result( + segments=[Segment(start=0.0, end=60.0, text="Фраза")], + duration=600.0, + ) + result_ok = _make_result() + model = _make_model() + backend = _make_backend() + tfr_warn = _make_tfr(result=result_warn, model=model, backend=backend) + tfr_ok = _make_tfr(result=result_ok, model=model, backend=backend) + + with ( + patch("local_transcriber.cli.load_config", return_value={}), + patch("local_transcriber.cli.validate_input_file", side_effect=lambda p: p), + patch("local_transcriber.cli.detect_device", return_value="cpu"), + patch("local_transcriber.cli.load_model", return_value=(model, "cpu", backend, "/models/medium")), + patch("local_transcriber.cli._transcribe_file", side_effect=[tfr_warn, tfr_ok]), + patch("local_transcriber.cli.write_transcript"), + patch("local_transcriber.cli.console", Console(stderr=True, width=1000)), + ): + out = runner.invoke(app, [str(a), str(b)]) + + assert out.exit_code == 0 + assert ( + " a.mp3: транскрипт покрывает 01:00 из 10:00 — " + "возможна потеря хвоста записи" + ) in out.output + + +def test_cli_repetition_quality_warning_truncates_after_three_blocks(tmp_path): + audio = tmp_path / "repeat-many.mp3" + audio.write_bytes(b"fake") + + def run(start, count, text): + return [ + Segment(start=start + index, end=start + index + 1.0, text=text) + for index in range(count) + ] + + result = _make_result( + segments=[ + *run(10.0, 6, "Первый повтор"), + Segment(start=18.0, end=19.0, text="Разрыв один"), + *run(20.0, 5, "Второй повтор"), + Segment(start=28.0, end=29.0, text="Разрыв два"), + *run(30.0, 4, "Третий повтор"), + Segment(start=38.0, end=39.0, text="Разрыв три"), + *run(40.0, 4, "Четвёртый повтор"), + ], + duration=90.0, + ) + + patches = _single_patches(result=result, tmp_file=audio) + with ( + patches[0], + patches[1], + patches[2], + patches[3], + patches[4], + patches[5], + patch("local_transcriber.cli.console", Console(stderr=True, width=1000)), + ): + out = runner.invoke(app, [str(audio)]) + + assert out.exit_code == 0 + assert ( + "Внимание: блоки повторов: [00:10.00 - 00:16.00] (6×); " + "[00:20.00 - 00:25.00] (5×); [00:30.00 - 00:34.00] (4×) " + "(+ ещё 1) — возможны галлюцинации модели. Попробуйте другой --device." + ) in out.output + + +def test_cli_default_result_has_no_quality_warnings(tmp_path): + audio = tmp_path / "normal.mp3" + audio.write_bytes(b"fake") + + patches = _single_patches(tmp_file=audio) + with patches[0], patches[1], patches[2], patches[3], patches[4], patches[5]: + out = runner.invoke(app, [str(audio)]) + + assert out.exit_code == 0 + assert "потеря хвоста" not in out.output + assert "галлюцинации" not in out.output diff --git a/tests/test_formatter.py b/tests/test_formatter.py index cb9dc2e..28bf201 100644 --- a/tests/test_formatter.py +++ b/tests/test_formatter.py @@ -172,3 +172,141 @@ def test_write_transcript(tmp_path): out = tmp_path / "output.md" write_transcript("# Test content\n", out) assert out.read_text(encoding="utf-8") == "# Test content\n" + + +def test_format_transcript_tail_gap_warning(): + result = TranscribeResult( + segments=[Segment(start=0.0, end=60.0, text=" Фраза.")], + language="ru", + language_probability=0.95, + duration=600.0, + device_used="cpu", + ) + + content = format_transcript( + result, + source_filename="tail.mp3", + model_name="medium", + device_info="CPU", + language_mode="forced", + transcription_date=datetime(2026, 1, 1, 0, 0, 0), + ) + + assert "возможна потеря хвоста" in content + assert "транскрипт покрывает 01:00 из 10:00" in content + + +def test_format_transcript_no_tail_gap_warning_for_small_gap(): + result = TranscribeResult( + segments=[Segment(start=0.0, end=60.0, text=" Фраза.")], + language="ru", + language_probability=0.95, + duration=179.99, + device_used="cpu", + ) + + content = format_transcript( + result, + source_filename="ok.mp3", + model_name="medium", + device_info="CPU", + language_mode="forced", + transcription_date=datetime(2026, 1, 1, 0, 0, 0), + ) + + assert "потеря хвоста" not in content + + +def test_format_transcript_no_tail_gap_warning_for_exact_threshold(): + result = TranscribeResult( + segments=[Segment(start=0.0, end=60.0, text=" Фраза.")], + language="ru", + language_probability=0.95, + duration=180.0, + device_used="cpu", + ) + + content = format_transcript( + result, + source_filename="ok.mp3", + model_name="medium", + device_info="CPU", + language_mode="forced", + transcription_date=datetime(2026, 1, 1, 0, 0, 0), + ) + + assert "потеря хвоста" not in content + + +def test_format_transcript_repetition_warning(): + result = TranscribeResult( + segments=[ + Segment(start=10.0, end=11.0, text=" Повторяемая фраза."), + Segment(start=11.0, end=12.0, text=" повторяемая фраза"), + Segment(start=12.0, end=13.0, text=" «Повторяемая фраза»"), + Segment(start=13.0, end=14.0, text=" повторяемая фраза…"), + ], + language="ru", + language_probability=0.95, + duration=60.0, + device_used="cpu", + ) + + content = format_transcript( + result, + source_filename="repeat.mp3", + model_name="medium", + device_info="CPU", + language_mode="forced", + transcription_date=datetime(2026, 1, 1, 0, 0, 0), + ) + + assert "повторы в [00:10.00 - 00:14.00] (4×)" in content + assert "возможны галлюцинации" in content + + +def test_format_transcript_repetition_warning_uses_hours(): + result = TranscribeResult( + segments=[ + Segment(start=3600.0, end=3601.0, text=" Повтор."), + Segment(start=3601.0, end=3602.0, text=" повтор"), + Segment(start=3602.0, end=3603.0, text=" повтор"), + Segment(start=3603.0, end=3604.0, text=" повтор"), + ], + language="ru", + language_probability=0.95, + duration=3700.0, + device_used="cpu", + ) + + content = format_transcript( + result, + source_filename="long-repeat.mp3", + model_name="medium", + device_info="CPU", + language_mode="forced", + transcription_date=datetime(2026, 1, 1, 0, 0, 0), + ) + + assert "повторы в [01:00:00.00 - 01:00:04.00] (4×)" in content + + +def test_format_transcript_without_anomalies_has_no_warning_lines(): + result = TranscribeResult( + segments=[Segment(start=0.0, end=60.0, text=" Обычная запись.")], + language="ru", + language_probability=0.95, + duration=120.0, + device_used="cpu", + ) + + content = format_transcript( + result, + source_filename="ok.mp3", + model_name="medium", + device_info="CPU", + language_mode="forced", + transcription_date=datetime(2026, 1, 1, 0, 0, 0), + ) + + assert "Внимание" not in content diff --git a/tests/test_quality.py b/tests/test_quality.py new file mode 100644 index 0000000..ed986ed --- /dev/null +++ b/tests/test_quality.py @@ -0,0 +1,125 @@ +import pytest + +from local_transcriber.quality import ( + REPETITION_MIN_LEN, + TAIL_GAP_WARN_S, + _normalize, + find_repetition_blocks, + tail_gap, +) +from local_transcriber.types import Segment, TranscribeResult + + +def _result(segments, duration): + return TranscribeResult( + segments=segments, + language="ru", + language_probability=0.95, + duration=duration, + device_used="cpu", + ) + + +def _segments(texts, start=0.0): + return [ + Segment(start=start + index, end=start + index + 1.0, text=text) + for index, text in enumerate(texts) + ] + + +def test_tail_gap_returns_positive_gap(): + result = _result([Segment(0.0, 10.0, "Текст")], duration=42.0) + + assert tail_gap(result) == 32.0 + + +def test_tail_gap_empty_segments_returns_zero(): + assert tail_gap(_result([], duration=42.0)) == 0.0 + + +def test_tail_gap_negative_gap_returns_zero(): + result = _result([Segment(0.0, 43.0, "Текст")], duration=42.0) + + assert tail_gap(result) == 0.0 + + +@pytest.mark.parametrize("gap", [119.99, 120.0]) +def test_tail_gap_boundary_does_not_warn(gap): + result = _result([Segment(0.0, 10.0, "Текст")], duration=10.0 + gap) + + assert tail_gap(result) <= TAIL_GAP_WARN_S + + +def test_tail_gap_boundary_warns_above_threshold(): + result = _result([Segment(0.0, 10.0, "Текст")], duration=130.01) + + assert tail_gap(result) > TAIL_GAP_WARN_S + + +def test_normalize_removes_case_punctuation_and_collapses_spaces(): + assert _normalize(' «ПРИВЕТ…» — (мир) [тест] <да> ') == "привет мир тест да" + assert _normalize("раз–два-три: да; нет!") == "раздватри да нет" + + +def test_normalize_punctuation_only_returns_empty(): + assert _normalize('.,!?…:;—–-"\'«»()[]<> ') == "" + + +def test_find_repetition_blocks_four_long_segments_with_normalization(): + segments = _segments(["Повтор!", "повтор", "«ПОВТОР»", "повтор…"]) + + blocks = find_repetition_blocks(segments) + + assert len(blocks) == 1 + assert blocks[0].start == 0.0 + assert blocks[0].end == 4.0 + assert blocks[0].count == 4 + assert blocks[0].text == "Повтор!" + + +def test_find_repetition_blocks_three_long_segments_is_empty(): + assert find_repetition_blocks(_segments(["Повтор", "повтор", "повтор"])) == [] + + +def test_find_repetition_blocks_length_boundary(): + assert find_repetition_blocks(_segments(["пять5"] * 4)) == [] + + blocks = find_repetition_blocks(_segments(["шесть6"] * 4)) + + assert len("шесть6") == REPETITION_MIN_LEN + assert len(blocks) == 1 + + +def test_find_repetition_blocks_short_text_boundary(): + assert find_repetition_blocks(_segments(["Ага."] * 9)) == [] + + blocks = find_repetition_blocks(_segments(["Ага."] * 10)) + + assert len(blocks) == 1 + assert blocks[0].count == 10 + + +def test_find_repetition_blocks_empty_normalized_text_is_ignored(): + assert find_repetition_blocks(_segments([":", "", " "] * 10)) == [] + + +def test_find_repetition_blocks_two_separate_runs(): + segments = [ + *_segments(["Первый повтор"] * 4), + Segment(10.0, 11.0, "Разрыв"), + *_segments(["Второй повтор"] * 4, start=20.0), + ] + + blocks = find_repetition_blocks(segments) + + assert len(blocks) == 2 + assert blocks[0].text == "Первый повтор" + assert blocks[1].text == "Второй повтор" + + +def test_find_repetition_blocks_empty_list_is_empty(): + assert find_repetition_blocks([]) == [] + + +def test_find_repetition_blocks_single_segment_is_empty(): + assert find_repetition_blocks([Segment(0.0, 1.0, "Повтор")]) == []