feat(formatter): объединение сегментов в абзацы по паузам

- Зачем:
  - мелкая нарезка (235 строк на 16 мин) засоряет контекст при передаче
    транскрипта в ИИ для дальнейшей обработки.
- Что:
  - соседние сегменты объединяются в абзацы: новый абзац при паузе > 2с
    или длительности > 60с.
  - 235 строк -> 58 строк на том же файле (4x компактнее).
- Проверка:
  - uv run pytest tests/test_formatter.py -v (11 passed).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-18 19:33:52 +03:00
co-authored by Claude Opus 4.6
parent a9a44b8d3a
commit c1eed7690d
2 changed files with 89 additions and 8 deletions
+42 -5
View File
@@ -1,7 +1,44 @@
from dataclasses import dataclass
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
from .transcriber import TranscribeResult from .transcriber import Segment, TranscribeResult
_PAUSE_THRESHOLD_S = 2.0 # пауза между сегментами для разбиения на абзацы
_MAX_PARAGRAPH_S = 60.0 # максимальная длительность абзаца
@dataclass
class _Paragraph:
start: float
end: float
text: str
def _group_segments(segments: list[Segment]) -> list[_Paragraph]:
"""Объединяет мелкие сегменты в абзацы по паузам и макс. длительности."""
if not segments:
return []
paragraphs: list[_Paragraph] = []
cur_start = segments[0].start
cur_end = segments[0].end
cur_texts: list[str] = [segments[0].text.strip()]
for seg in segments[1:]:
gap = seg.start - cur_end
duration = seg.end - cur_start
if gap >= _PAUSE_THRESHOLD_S or duration > _MAX_PARAGRAPH_S:
paragraphs.append(_Paragraph(cur_start, cur_end, " ".join(cur_texts)))
cur_start = seg.start
cur_end = seg.end
cur_texts = [seg.text.strip()]
else:
cur_end = seg.end
cur_texts.append(seg.text.strip())
paragraphs.append(_Paragraph(cur_start, cur_end, " ".join(cur_texts)))
return paragraphs
def format_timestamp(seconds: float, use_hours: bool = False) -> str: def format_timestamp(seconds: float, use_hours: bool = False) -> str:
@@ -56,11 +93,11 @@ def format_transcript(
lines.append("") lines.append("")
lines.append("*Речь не обнаружена.*") lines.append("*Речь не обнаружена.*")
else: else:
for seg in result.segments: for para in _group_segments(result.segments):
start = format_timestamp(seg.start, use_hours=use_hours) start = format_timestamp(para.start, use_hours=use_hours)
end = format_timestamp(seg.end, use_hours=use_hours) end = format_timestamp(para.end, use_hours=use_hours)
lines.append("") lines.append("")
lines.append(f"[{start} - {end}] {seg.text.strip()}") lines.append(f"[{start} - {end}] {para.text}")
lines.append("") lines.append("")
return "\n".join(lines) return "\n".join(lines)
+47 -3
View File
@@ -2,6 +2,7 @@ from datetime import datetime
from pathlib import Path from pathlib import Path
from local_transcriber.formatter import ( from local_transcriber.formatter import (
_group_segments,
format_timestamp, format_timestamp,
format_transcript, format_transcript,
write_transcript, write_transcript,
@@ -50,9 +51,8 @@ def test_format_transcript_basic():
assert "**Длительность**: 02:00" in content assert "**Длительность**: 02:00" in content
assert "**Устройство**: CUDA (NVIDIA GeForce RTX 3060)" in content assert "**Устройство**: CUDA (NVIDIA GeForce RTX 3060)" in content
assert "---" in content assert "---" in content
# Проверяем пробел между ] и текстом независимо от ведущих пробелов в seg.text # Соседние сегменты без паузы объединяются в один абзац
assert "[00:00.00 - 00:04.82] Добрый день, коллеги." in content assert "[00:00.00 - 00:09.15] Добрый день, коллеги. Первый вопрос." in content
assert "[00:04.82 - 00:09.15] Первый вопрос." in content
def test_format_transcript_segment_no_leading_space(): def test_format_transcript_segment_no_leading_space():
@@ -124,6 +124,50 @@ def test_format_transcript_long():
assert "[01:01:40.00 - 01:01:50.25] Конец." in content assert "[01:01:40.00 - 01:01:50.25] Конец." in content
def test_group_segments_merges_adjacent():
"""Соседние сегменты без паузы объединяются."""
segments = [
Segment(start=0.0, end=3.0, text=" Первый."),
Segment(start=3.0, end=6.0, text=" Второй."),
Segment(start=6.0, end=9.0, text=" Третий."),
]
groups = _group_segments(segments)
assert len(groups) == 1
assert groups[0].start == 0.0
assert groups[0].end == 9.0
assert groups[0].text == "Первый. Второй. Третий."
def test_group_segments_splits_on_pause():
"""Пауза > 2с разбивает на отдельные абзацы."""
segments = [
Segment(start=0.0, end=3.0, text=" Первый."),
Segment(start=3.0, end=6.0, text=" Второй."),
Segment(start=9.0, end=12.0, text=" После паузы."),
]
groups = _group_segments(segments)
assert len(groups) == 2
assert groups[0].text == "Первый. Второй."
assert groups[1].text == "После паузы."
def test_group_segments_splits_on_max_duration():
"""Абзац разбивается при превышении макс. длительности."""
segments = [
Segment(start=0.0, end=30.0, text=" Длинный."),
Segment(start=30.0, end=55.0, text=" Ещё."),
Segment(start=55.0, end=80.0, text=" Перелив."),
]
groups = _group_segments(segments)
assert len(groups) == 2
assert groups[0].text == "Длинный. Ещё."
assert groups[1].text == "Перелив."
def test_group_segments_empty():
assert _group_segments([]) == []
def test_write_transcript(tmp_path): 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)