vidconf/workers/transcription/phrases.py

"""Реконструкция фраз из сегментов Whisper по алгоритму ТЗ §1.3.

Определение фразы (ТЗ §1.3): фрагмент монолога одного участника между
монологами других участников, либо между `t_start` сеанса и первым
монологом, либо между последним монологом и `t_end`.

Алгоритм:
1. Сегменты всех треков (участников) сливаются в единый таймлайн с учётом
   смещения каждого трека (`track_offsets`) и сортируются по `start`.
2. Идём по таймлайну и группируем подряд идущие сегменты одного и того же
   участника в одну фразу; смена участника — граница фразы.
3. Короткая вставка другого участника короче `INTERJECTION_THRESHOLD_S`
   (например, «угу») не прерывает текущую фразу, но сама сохраняется
   отдельной фразой.
4. Тишина (пауза) между сегментами одного участника без чужой речи фразу
   не рвёт.
5. Перекрытия речи двух участников длиннее порога — обе фразы сохраняются
   с пересекающимися интервалами (без «доминирующего спикера»).

Функция `build_phrases` — чистая: работает в секундах, относительных
`t_start` сеанса; конвертацию в абсолютное время UTC делает вызывающий код
(оркестратор пайплайна).
"""

import uuid
from dataclasses import dataclass, field

from core.plugins.transcriber import Segment

INTERJECTION_THRESHOLD_S: float = 1.5
"""Порог длительности (в секундах) короткой вставки другого спикера,
которая не прерывает текущую фразу, но фиксируется как отдельная фраза."""


@dataclass(frozen=True, slots=True)
class SpeakerSegment:
    """Сегмент речи одного участника на объединённом таймлайне сеанса."""

    participant_id: uuid.UUID
    start: float  # секунды от t_start сеанса
    end: float
    text: str


@dataclass(frozen=True, slots=True)
class PhraseDraft:
    """Черновик реконструированной фразы — без привязки к сеансу/БД."""

    participant_id: uuid.UUID
    start: float
    end: float
    text: str


@dataclass(slots=True)
class _OpenPhrase:
    """Мутируемое состояние фразы, которая ещё собирается в процессе обхода таймлайна."""

    participant_id: uuid.UUID
    start: float
    end: float
    texts: list[str] = field(default_factory=list)

    def to_draft(self) -> PhraseDraft:
        """Зафиксировать накопленное состояние как неизменяемый `PhraseDraft`."""
        return PhraseDraft(
            participant_id=self.participant_id,
            start=self.start,
            end=self.end,
            text=" ".join(text for text in self.texts if text),
        )


def _merge_timeline(
    segments_by_participant: dict[uuid.UUID, list[Segment]],
    track_offsets: dict[uuid.UUID, float],
) -> list[SpeakerSegment]:
    """Слить сегменты всех участников в единый таймлайн, отсортированный по `start`.

    Смещение (`track_offsets`) — время старта записи трека участника
    относительно `t_start` сеанса; при отсутствии участника в словаре
    смещений используется 0.0.
    """
    timeline: list[SpeakerSegment] = []
    for participant_id, segments in segments_by_participant.items():
        offset = track_offsets.get(participant_id, 0.0)
        for segment in segments:
            timeline.append(
                SpeakerSegment(
                    participant_id=participant_id,
                    start=segment.start + offset,
                    end=segment.end + offset,
                    text=segment.text,
                )
            )
    timeline.sort(key=lambda s: (s.start, s.end))
    return timeline


def build_phrases(
    segments_by_participant: dict[uuid.UUID, list[Segment]],
    track_offsets: dict[uuid.UUID, float],
) -> list[PhraseDraft]:
    """Реконструировать фразы по объединённому таймлайну сегментов участников.

    Аргументы:
        segments_by_participant: сегменты Whisper каждого участника,
            относительные началу его собственного трека.
        track_offsets: смещение (в секундах от `t_start` сеанса) начала
            записи каждого трека.

    Возвращает список `PhraseDraft`, отсортированный по `start`.
    """
    timeline = _merge_timeline(segments_by_participant, track_offsets)
    if not timeline:
        return []

    finished: list[PhraseDraft] = []
    current: _OpenPhrase | None = None

    for segment in timeline:
        if current is None:
            current = _OpenPhrase(
                participant_id=segment.participant_id,
                start=segment.start,
                end=segment.end,
                texts=[segment.text],
            )
            continue

        if segment.participant_id == current.participant_id:
            # тот же спикер: продолжаем фразу, тишина между сегментами её не рвёт
            current.end = max(current.end, segment.end)
            current.texts.append(segment.text)
            continue

        duration = segment.end - segment.start
        if duration < INTERJECTION_THRESHOLD_S:
            # короткая вставка чужого спикера: не прерывает текущую фразу,
            # но фиксируется как отдельная фраза
            finished.append(
                PhraseDraft(
                    participant_id=segment.participant_id,
                    start=segment.start,
                    end=segment.end,
                    text=segment.text,
                )
            )
            continue

        # реальная смена спикера (или длинное перекрытие) — граница фразы
        finished.append(current.to_draft())
        current = _OpenPhrase(
            participant_id=segment.participant_id,
            start=segment.start,
            end=segment.end,
            texts=[segment.text],
        )

    if current is not None:
        finished.append(current.to_draft())

    finished.sort(key=lambda p: (p.start, p.end))
    return finished