Vladimir BryzgalovandClaude Opus 5
56cf7e5b0e
Диаризация задавала не только «кто говорит», но и границы того, что вообще попадает в распознавание: фразы вне её сегментов пропадали молча. Второй проход их возвращает с пометкой recovered=true и speaker=0. Замер на восьми записях: 131 слово из 17 004 (0,8 %), 3-19 с на запись. Возвращается и настоящий диалог (вопрос о цене, даты, документы), и радио на фоне, поэтому проход отключается настройкой recover_gaps. Скрипты render_transcript.py и check_gaps.py - для сверки расшифровки с записью и поиска потерянных фраз. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
95 lines
4.1 KiB
Python
95 lines
4.1 KiB
Python
"""Второй проход по промежуткам: без него часть фраз пропадает молча."""
|
|
import numpy as np
|
|
|
|
from app.pipeline import GAP_MIN_SEC, MIN_RECOVERED_CHARS, Pipeline, gap_ranges
|
|
|
|
|
|
def turn(start, end, text="реплика"):
|
|
return {"speaker": 1, "start": start, "end": end, "text": text, "acoustics": {}}
|
|
|
|
|
|
class TestGapRanges:
|
|
def test_finds_gap_between_turns(self):
|
|
assert list(gap_ranges([turn(0, 5), turn(20, 25)], 30)) == [(5, 20), (25, 30)]
|
|
|
|
def test_ignores_short_pauses(self):
|
|
"""Пауза в пару секунд - обычное дело в разговоре."""
|
|
assert list(gap_ranges([turn(0, 5), turn(6, 25)], 25)) == []
|
|
|
|
def test_finds_gap_before_first_turn(self):
|
|
assert (0.0, 10) in list(gap_ranges([turn(10, 20)], 20))
|
|
|
|
def test_finds_gap_after_last_turn(self):
|
|
assert (20, 40) in list(gap_ranges([turn(10, 20)], 40))
|
|
|
|
def test_no_turns_gives_whole_recording(self):
|
|
assert list(gap_ranges([], 60)) == [(0.0, 60)]
|
|
|
|
def test_overlapping_turns_do_not_create_negative_gap(self):
|
|
"""Реплики могут перекрываться, и конец предыдущей бывает позже начала следующей."""
|
|
assert list(gap_ranges([turn(0, 30), turn(10, 20)], 25)) == []
|
|
|
|
def test_boundary_is_inclusive(self):
|
|
assert len(list(gap_ranges([turn(0, 5), turn(5 + GAP_MIN_SEC, 20)], 20))) == 1
|
|
|
|
|
|
class FakePipeline(Pipeline):
|
|
"""Подменяет распознавание: модели в этих проверках не нужны."""
|
|
|
|
def __init__(self, answers):
|
|
self.answers = answers
|
|
self._replacements = {}
|
|
|
|
def _recognize_safely(self, audio, depth=0):
|
|
return self.answers.pop(0) if self.answers else ""
|
|
|
|
|
|
class TestRecovery:
|
|
@staticmethod
|
|
def samples(seconds=60):
|
|
return np.zeros(seconds * 16000, dtype=np.float32)
|
|
|
|
def test_recovered_turn_has_unknown_speaker(self):
|
|
"""Диаризация этих участков не видела, поэтому кто говорил - неизвестно."""
|
|
pipe = FakePipeline(["Вы приносите справку, я назначаю инструктора."])
|
|
out = pipe._recover_gaps(self.samples(), [turn(0, 5), turn(20, 25)], 25)
|
|
assert out[0]["speaker"] == 0
|
|
assert out[0]["recovered"] is True
|
|
|
|
def test_short_result_is_dropped_as_noise(self):
|
|
"""Шелест бумаг звучит почти как речь и даёт короткие обрывки."""
|
|
pipe = FakePipeline(["ага"])
|
|
assert pipe._recover_gaps(self.samples(), [turn(0, 5), turn(20, 25)], 25) == []
|
|
|
|
def test_empty_result_is_dropped(self):
|
|
pipe = FakePipeline([""])
|
|
assert pipe._recover_gaps(self.samples(), [turn(0, 5), turn(20, 25)], 25) == []
|
|
|
|
def test_threshold_is_on_characters(self):
|
|
text = "x" * MIN_RECOVERED_CHARS
|
|
pipe = FakePipeline([text])
|
|
assert len(pipe._recover_gaps(self.samples(), [turn(0, 5), turn(20, 25)], 25)) == 1
|
|
|
|
def test_time_span_matches_the_gap(self):
|
|
pipe = FakePipeline(["Вы приносите справку, я назначаю инструктора."])
|
|
out = pipe._recover_gaps(self.samples(), [turn(0, 5), turn(20, 25)], 25)
|
|
assert (out[0]["start"], out[0]["end"]) == (5, 20)
|
|
|
|
|
|
class TestSwitch:
|
|
"""Второй проход подхватывает и радио на фоне - должна быть возможность выключить."""
|
|
|
|
def test_setting_defaults_to_on(self, tmp_path):
|
|
from app.config import load_settings
|
|
|
|
config = tmp_path / "config.toml"
|
|
config.write_text("[processing]\nthreads=1\n", encoding="utf-8")
|
|
assert load_settings(config).recover_gaps is True
|
|
|
|
def test_setting_can_be_turned_off(self, tmp_path):
|
|
from app.config import load_settings
|
|
|
|
config = tmp_path / "config.toml"
|
|
config.write_text("[processing]\nrecover_gaps=false\n", encoding="utf-8")
|
|
assert load_settings(config).recover_gaps is False
|