Диаризация задавала не только «кто говорит», но и границы того, что вообще попадает в распознавание: фразы вне её сегментов пропадали молча. Второй проход их возвращает с пометкой 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>
111 lines
4.8 KiB
Python
111 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
||
"""Читаемая расшифровка из результата сервиса, с отметками пропусков.
|
||
|
||
Распознаётся только то, что детектор речи отметил как речь. Если фраза
|
||
показалась ему тишиной, до распознавания она не доходит и в расшифровке её
|
||
не будет - молча. Поэтому здесь помечаются разрывы между репликами: по ним
|
||
видно, куда перемотать запись, чтобы проверить, была ли там речь.
|
||
|
||
uv run python build/render_transcript.py результат.json > расшифровка.txt
|
||
uv run python build/render_transcript.py папка/ --out папка-текстов/
|
||
"""
|
||
import argparse
|
||
import json
|
||
from pathlib import Path
|
||
|
||
# Пауза в разговоре - дело обычное. Помечаем только заметные провалы,
|
||
# где успела бы поместиться целая фраза.
|
||
GAP_NOTICE_SEC = 3.0
|
||
GAP_ALARM_SEC = 15.0
|
||
|
||
|
||
def timecode(seconds: float) -> str:
|
||
return f"{int(seconds) // 60:02d}:{int(seconds) % 60:02d}"
|
||
|
||
|
||
def render(data: dict) -> str:
|
||
turns = data.get("turns", [])
|
||
stats = data.get("stats", {})
|
||
timing = data.get("timing", {})
|
||
duration = data.get("duration_sec", 0.0)
|
||
|
||
lines = [
|
||
f"Запись: {data.get('source') or data.get('filename', '?')}",
|
||
f"Длительность: {timecode(duration)} ({duration:.0f} с)",
|
||
f"Реплик: {len(turns)}",
|
||
]
|
||
if stats:
|
||
speech = stats.get("speech_sec", 0.0)
|
||
share = speech / duration * 100 if duration else 0
|
||
lines.append(
|
||
f"Распознано речи: {speech:.0f} с из {duration:.0f} с ({share:.0f}%), "
|
||
f"остальное детектор счёл тишиной")
|
||
lines.append(
|
||
f"Разделение по голосам: {stats.get('separation_quality', 0):.2f} "
|
||
f"({'надёжно' if stats.get('speakers_reliable') else 'НЕНАДЁЖНО, роли условны'})")
|
||
if timing:
|
||
lines.append(f"Обработка: {timing.get('diarization_sec', 0):.0f} с диаризация + "
|
||
f"{timing.get('asr_sec', 0):.0f} с распознавание")
|
||
lines.append("")
|
||
lines.append("Разрывы длиннее 3 с помечены. Это места, где детектор речи не нашёл")
|
||
lines.append("речи - именно там стоит проверить запись, если фраза потерялась.")
|
||
lines.append("=" * 78)
|
||
lines.append("")
|
||
|
||
previous_end = 0.0
|
||
lost = 0.0
|
||
for turn in turns:
|
||
gap = turn["start"] - previous_end
|
||
if gap >= GAP_NOTICE_SEC:
|
||
lost += gap
|
||
mark = "!!!" if gap >= GAP_ALARM_SEC else "..."
|
||
lines.append(f" {mark} пропуск {gap:.0f} с "
|
||
f"[{timecode(previous_end)} - {timecode(turn['start'])}]")
|
||
loudness = (turn.get("acoustics") or {}).get("loudness_db")
|
||
volume = f" {loudness:.0f}дБ" if loudness is not None else ""
|
||
lines.append(f"[{timecode(turn['start'])}] Г{turn['speaker']}{volume}: {turn['text']}")
|
||
previous_end = turn["end"]
|
||
|
||
tail = duration - previous_end
|
||
if tail >= GAP_NOTICE_SEC:
|
||
lost += tail
|
||
lines.append(f" ... пропуск {tail:.0f} с "
|
||
f"[{timecode(previous_end)} - конец записи]")
|
||
|
||
lines.append("")
|
||
lines.append("=" * 78)
|
||
lines.append(f"Суммарно в пропусках длиннее {GAP_NOTICE_SEC:.0f} с: {lost:.0f} с "
|
||
f"({lost / duration * 100 if duration else 0:.0f}% записи)")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("source", type=Path, help="файл результата или папка с ними")
|
||
ap.add_argument("--out", type=Path, help="куда складывать тексты")
|
||
args = ap.parse_args()
|
||
|
||
files = sorted(args.source.glob("*.json")) if args.source.is_dir() else [args.source]
|
||
files = [f for f in files if f.name != "summary.json"]
|
||
if not files:
|
||
print("не нашёл файлов результата")
|
||
return 1
|
||
|
||
for path in files:
|
||
data = json.loads(path.read_text(encoding="utf-8"))
|
||
if "turns" not in data:
|
||
continue
|
||
text = render(data)
|
||
if args.out:
|
||
args.out.mkdir(parents=True, exist_ok=True)
|
||
target = args.out / f"{path.stem}.txt"
|
||
target.write_text(text, encoding="utf-8")
|
||
print(f"{target}")
|
||
else:
|
||
print(text)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|