jhonny-editor/code/engine/scanner/retakes/modelos_de_retakes.py

"""Modelos de dados do submódulo de detecção de retakes.

Mantemos os mesmos princípios de ``scanner/modelos.py``: dataclasses
imutáveis (``frozen=True``) com validação em ``__post_init__`` e sem
dependência de implementações concretas de providers.
"""

from __future__ import annotations

import uuid
from dataclasses import dataclass, field
from datetime import datetime, timezone

from ..modelos import PalavraDeTranscricao


@dataclass(frozen=True)
class Fala:
    """Uma fala normalizada, já posicionada na timeline do vídeo.

    Corresponde a um ``SegmentoDeTranscricao`` convertido pelo adaptador,
    mas carrega o contexto necessário para a análise temporal e de
    agrupamento: vídeo, faixa de áudio, número de sequência e o vínculo
    com o segmento/clipe de origem.
    """

    id: str
    video_id: str
    faixa_id: str
    segmento_id: str
    ordem: int
    inicio: float
    fim: float
    texto: str
    texto_normalizado: str = ""
    palavras: tuple[PalavraDeTranscricao, ...] = ()
    falante: str | None = None

    def __post_init__(self) -> None:
        if self.inicio < 0 or self.fim < self.inicio:
            raise ValueError(f"Intervalo da fala {self.id} inválido.")


@dataclass(frozen=True)
class SinalDeReinicio:
    """Indício de que uma fala recomeçou uma ideia já iniciada."""

    fala_id: str
    tipo: str
    intensidade: float
    evidencias: list[str] = field(default_factory=list)

    def __post_init__(self) -> None:
        if not 0.0 <= self.intensidade <= 1.0:
            raise ValueError(f"Intensidade do reinício {self.fala_id} fora de [0, 1].")


@dataclass(frozen=True)
class ResultadoDaComparacao:
    """Resultado da comparação de duas falas, textual e por sequência."""

    fala_a_id: str
    fala_b_id: str
    similaridade_jaccard: float = 0.0
    similaridade_de_sequencia: float = 0.0
    similaridade_do_inicio: float = 0.0
    similaridade_do_final: float = 0.0
    similaridade_final: float = 0.0


@dataclass(frozen=True)
class ResultadoDoIntervalo:
    """Relação temporal entre duas falas."""

    intervalo_em_segundos: float
    mesmo_segmento: bool = False
    proximidade_temporal: float = 0.0
    indicio_de_nova_tentativa: bool = False


@dataclass(frozen=True)
class EvidenciaDeRetake:
    """Uma evidência legível que sustenta a classificação de um grupo."""

    tipo: str
    descricao: str
    valor: float | None = None


@dataclass(frozen=True)
class TomadaDeRetake:
    """Uma tomada (fala) pertencente a um grupo de retakes."""

    ordem: int
    fala_id: str
    segmento_id: str
    inicio: float
    fim: float
    texto: str
    similaridade_com_anterior: float = 0.0
    confianca: float = 0.0


def gerar_id_do_grupo() -> str:
    """Gera um id curto e monótono para um grupo de retakes."""
    return f"retake_{uuid.uuid4().hex[:6]}"


@dataclass(frozen=True)
class ParAgrupado:
    """Um par de falas que o agrupador juntou como candidato a retake."""

    fala_a: Fala
    fala_b: Fala
    comparacao: ResultadoDaComparacao
    intervalo: ResultadoDoIntervalo
    similaridade_semantica: float = 0.0
    mudanca_visual_proxima: bool = False


@dataclass(frozen=True)
class AgrupamentoDeFalas:
    """Um grupo candidato formado pelo agrupador, antes da classificação."""

    fala_ids: tuple[str, ...]
    pares: tuple[ParAgrupado, ...]
    sinais_de_reinicio: tuple[SinalDeReinicio, ...] = ()


@dataclass(frozen=True)
class ClassificacaoDeRetake:
    """Resultado da classificação de um grupo de retakes."""

    tipo: str
    confianca: float

# Estados possíveis da revisão manual de um grupo (decisão do usuário na
# tela de retakes; a detecção automática nunca altera o estado de revisão).
STATUS_PENDENTE = "pendente_de_revisao"
STATUS_CONFIRMADO = "confirmado"
STATUS_REJEITADO = "nao_e_retake"
STATUS_IGNORADO = "ignorado"
STATUS_DE_REVISAO_VALIDOS = frozenset(
    {STATUS_PENDENTE, STATUS_CONFIRMADO, STATUS_REJEITADO, STATUS_IGNORADO})



@dataclass(frozen=True)
class GrupoDeRetake:
    """Agrupa as tomadas que representam tentativas da mesma fala."""

    id: str
    video_id: str
    faixa_id: str
    tipo: str
    confianca: float
    tomadas: list[TomadaDeRetake] = field(default_factory=list)
    evidencias: list[EvidenciaDeRetake] = field(default_factory=list)
    criado_em: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
    status_de_revisao: str = STATUS_PENDENTE

    def __post_init__(self) -> None:
        if not 0.0 <= self.confianca <= 1.0:
            raise ValueError(f"Confiança do grupo {self.id} fora de [0, 1].")

    @property
    def tomada_principal_id(self) -> str | None:
        if not self.tomadas:
            return None
        return max(self.tomadas, key=lambda item: item.confianca).fala_id

    @property
    def fala_ids(self) -> list[str]:
        return [tomada.fala_id for tomada in self.tomadas]