Vladimir BryzgalovandClaude Opus 5
9dc67bda5c
Локальный FastAPI-сервис поверх GigaAM v3 и sherpa-onnx: приём аудио, очередь задач, разделение по говорящим, постобработка терминов. Доставка на Windows - ZIP со встроенным Python, без установки чего-либо. Обновление кода при запуске тянется из релизов Gitea: меняется только папка app, десятки килобайт вместо всего пакета. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
75 lines
2.7 KiB
Python
75 lines
2.7 KiB
Python
"""Работа с сегментами диаризации: склейка в реплики, нарезка, статистика."""
|
|
from dataclasses import dataclass
|
|
|
|
__all__ = ["Segment", "Turn", "merge_turns", "chunk_ranges", "speech_stats"]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Segment:
|
|
"""Отрезок речи одного говорящего, как его вернула диаризация."""
|
|
start: float
|
|
end: float
|
|
speaker: int
|
|
|
|
|
|
@dataclass
|
|
class Turn:
|
|
"""Реплика: несколько подряд идущих сегментов одного говорящего."""
|
|
start: float
|
|
end: float
|
|
speaker: int
|
|
|
|
@property
|
|
def duration(self) -> float:
|
|
return self.end - self.start
|
|
|
|
|
|
def merge_turns(segments: list[Segment], gap: float = 0.8) -> list[Turn]:
|
|
"""Склеивает соседние сегменты одного говорящего в реплики.
|
|
|
|
Диаризация режет речь на короткие куски по паузам внутри фразы;
|
|
для расшифровки нужны цельные реплики, иначе текст рассыпается.
|
|
"""
|
|
turns: list[Turn] = []
|
|
for s in segments:
|
|
if turns and turns[-1].speaker == s.speaker and s.start - turns[-1].end <= gap:
|
|
turns[-1].end = s.end
|
|
else:
|
|
turns.append(Turn(start=s.start, end=s.end, speaker=s.speaker))
|
|
return turns
|
|
|
|
|
|
def chunk_ranges(start: float, end: float, max_len: float) -> list[tuple[float, float]]:
|
|
"""Режет интервал на куски не длиннее max_len.
|
|
|
|
GigaAM падает на аудио длиннее ~200 секунд: у неё позиционное кодирование
|
|
рассчитано на 5000 фреймов. Держим запас и режем по 150.
|
|
"""
|
|
chunks = []
|
|
cursor = start
|
|
while cursor < end:
|
|
stop = min(cursor + max_len, end)
|
|
chunks.append((cursor, stop))
|
|
cursor = stop
|
|
return chunks or [(start, end)]
|
|
|
|
|
|
def speech_stats(segments: list[Segment]) -> dict:
|
|
"""Считает, сколько и в каких долях говорил каждый участник."""
|
|
by_speaker: dict[int, float] = {}
|
|
for s in segments:
|
|
by_speaker[s.speaker] = by_speaker.get(s.speaker, 0.0) + (s.end - s.start)
|
|
total = sum(by_speaker.values())
|
|
return {
|
|
"speakers": len(by_speaker),
|
|
"speech_sec": round(total, 1),
|
|
"by_speaker": [
|
|
{
|
|
"speaker": spk + 1,
|
|
"speech_sec": round(dur, 1),
|
|
"share_pct": round(dur / total * 100, 1) if total else 0.0,
|
|
}
|
|
for spk, dur in sorted(by_speaker.items())
|
|
],
|
|
}
|