feat(quality): предупреждения о возможной потере содержания транскрипта

- Зачем:
  - транскрипция может тихо терять содержание: блоки галлюцинированных повторов и обрыв распознавания до конца файла выглядят как законченный транскрипт (класс ошибок из 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 <noreply@anthropic.com>
This commit is contained in:
Dmitriy Dementiev
2026-07-09 22:21:07 +03:00
co-authored by Claude Fable 5
parent 650e7580e5
commit b1dfd9dcca
6 changed files with 575 additions and 3 deletions
+68 -1
View File
@@ -11,9 +11,21 @@ from rich.status import Status
from .config import apply_device_defaults, load_config, resolve_defaults from .config import apply_device_defaults, load_config, resolve_defaults
from .context_menu import install_menu as install_context_menu from .context_menu import install_menu as install_context_menu
from .context_menu import uninstall_menu as uninstall_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 ( from .transcriber import (
Segment, Segment,
TranscribeResult,
_is_cuda_error, _is_cuda_error,
_transcribe_file, _transcribe_file,
load_model, load_model,
@@ -45,6 +57,59 @@ def _format_device_info(device_used: str) -> str:
return "CPU" 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() @app.command()
def main( def main(
files: list[Path] | None = typer.Argument(None, help="Пути к аудио/видеофайлам"), files: list[Path] | None = typer.Argument(None, help="Пути к аудио/видеофайлам"),
@@ -257,6 +322,7 @@ def _run_single(
elapsed = time.monotonic() - start elapsed = time.monotonic() - start
console.print(f"Транскрипт сохранён: \"{output_path}\"", style="green") console.print(f"Транскрипт сохранён: \"{output_path}\"", style="green")
console.print(f" Сегментов: {len(result.segments)} Время: {elapsed:.1f}с") console.print(f" Сегментов: {len(result.segments)} Время: {elapsed:.1f}с")
_print_quality_warnings(result)
def _run_batch( def _run_batch(
@@ -392,6 +458,7 @@ def _run_batch(
style="green", style="green",
) )
processed += 1 processed += 1
_print_quality_warnings(result, file.name)
except KeyboardInterrupt: except KeyboardInterrupt:
raise raise
except Exception as exc: except Exception as exc:
+16 -2
View File
@@ -4,6 +4,7 @@ from dataclasses import dataclass
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
from .quality import TAIL_GAP_WARN_S, find_repetition_blocks, tail_gap
from .types import Segment, TranscribeResult from .types import Segment, TranscribeResult
_PAUSE_THRESHOLD_S = 2.0 # пауза между сегментами для разбиения на абзацы _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}" return f"{minutes:02d}:{secs:02d}.{centiseconds:02d}"
def _format_duration(seconds: float) -> str: def format_duration(seconds: float) -> str:
"""Человекочитаемая длительность для метаданных в шапке транскрипта.""" """Человекочитаемая длительность для метаданных в шапке транскрипта."""
total = int(seconds) total = int(seconds)
h = total // 3600 h = total // 3600
@@ -92,7 +93,20 @@ def format_transcript(
lines.append(f"- **Дата транскрипции**: {date.strftime('%Y-%m-%d %H:%M:%S')}") lines.append(f"- **Дата транскрипции**: {date.strftime('%Y-%m-%d %H:%M:%S')}")
lines.append(f"- **Модель**: {model_name}") lines.append(f"- **Модель**: {model_name}")
lines.append(f"- **Язык**: {result.language} ({language_mode})") 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(f"- **Устройство**: {device_info}")
lines.append("") lines.append("")
lines.append("---") lines.append("---")
+78
View File
@@ -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
+150
View File
@@ -2,6 +2,7 @@ from pathlib import Path
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
import pytest import pytest
from rich.console import Console
from typer.testing import CliRunner from typer.testing import CliRunner
from local_transcriber.cli import _format_device_info, app 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 out.exit_code == 1
assert "нет APPDATA" in out.output assert "нет APPDATA" in out.output
assert "--verbose" not 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
+138
View File
@@ -172,3 +172,141 @@ def test_write_transcript(tmp_path):
out = tmp_path / "output.md" out = tmp_path / "output.md"
write_transcript("# Test content\n", out) write_transcript("# Test content\n", out)
assert out.read_text(encoding="utf-8") == "# Test content\n" 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
+125
View File
@@ -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, "Повтор")]) == []