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

"""Converte a transcrição existente em falas ordenadas para a análise.

O módulo de retakes não realiza transcrição. Ele recebe a saída do
``TranscricaoDaTimeline`` (lista de ``TranscricaoDoClipe``) e a converte
em ``Fala``s ordenadas ao longo da timeline, preservando o vínculo com o
segmento/clipe de origem e as palavras alinhadas quando disponíveis.
"""

from __future__ import annotations

import re
from typing import Iterable

from ..transcricao_da_timeline import TranscricaoDoClipe
from .modelos_de_retakes import Fala

# Sinais de erro mais comuns vindos dos providers de transcrição.
_PALAVRAS_DE_ERRO = {"", "...", "…"}

_VOYELS = "aeiouáéíóúâêôãõàèìòù"
_CONSOANTES = "bcdfghjklmnpqrstvwxyz"
_PADRAO_SILABA = re.compile(
    rf"([{_VOYELS}][^aeiouáéíóúâêôãõàèìòù]*)|([{_CONSOANTES}]+[aeiouáéíóúâêôãõàèìòù]?)",
    re.IGNORECASE,
)


def _normalizar(texto: str) -> str:
    """Minúsculas, sem pontuação e sem espaços duplicados."""
    sem_pontuacao = re.sub(r"[^\w\s]", " ", texto)
    return " ".join(sem_pontuacao.lower().split())


def _prefixo_comum(a: list[str], b: list[str]) -> int:
    n = 0
    for x, y in zip(a, b):
        if x != y:
            break
        n += 1
    return n


def _sucesso_de_silabas(a: list[str], b: list[str]) -> float:
    if not a or not b:
        return 0.0
    silabas_a = [_PADRAO_SILABA.findall(p)[0][0] for p in a]
    silabas_b = [_PADRAO_SILABA.findall(p)[0][0] for p in b]
    n = _prefixo_comum(silabas_a, silabas_b)
    return n / max(len(silabas_a), len(silabas_b))


def _e_ruido(texto: str) -> bool:
    texto_limpo = _normalizar(texto)
    if texto_limpo in _PALAVRAS_DE_ERRO:
        return True
    # Fala em branco ou "vazia" por provedor.
    return not texto_limpo


class AdaptadorDeFalas:
    """Transforma a transcrição da timeline em falas prontas para análise.

    Recebe uma coleção de ``TranscricaoDoClipe`` (uma por faixa de áudio do
    vídeo) e devolve as falas ordenadas por tempo. O ``video_id`` é um rótulo
    informado pelo chamador; quando ausente, usa o identificador da timeline.
    """

    def __init__(self, video_id: str | None = None, faixa_id: str | None = None) -> None:
        self.video_id = video_id
        self.faixa_id = faixa_id

    @staticmethod
    def normalizar(texto: str) -> str:
        """Normaliza um texto (minúsculas, sem pontuação, sem espaços duplos)."""
        return _normalizar(texto)

    def converter(
        self,
        transcricoes: Iterable[TranscricaoDoClipe],
        video_id: str | None = None,
        faixa_id: str | None = None,
    ) -> list[Fala]:
        video_id = video_id or self.video_id or "video"
        faixa_id = faixa_id or self.faixa_id or "faixa"

        falas: list[Fala] = []
        for transcricao in transcricoes:
            segmento_id = transcricao.identificador_do_clipe
            for segmento in transcricao.segmentos:
                if _e_ruido(segmento.texto):
                    continue
                falas.append(self._fala_de_segmento(segmento, segmento_id, video_id, faixa_id))

        return self._ordenar(falas)

    def converter_de_segmentos(
        self,
        segmentos_por_clipe: Iterable[tuple[str, Iterable[object]]],
        video_id: str | None = None,
        faixa_id: str | None = None,
    ) -> list[Fala]:
        """Converte ``(segmento_id, [SegmentoDeTranscricao])`` em falas.

        Usado quando a transcrição chega no ``ContextoDeAnalise`` do scanner
        já como segmentos individuais (padrão do pipeline), em vez de
        ``TranscricaoDoClipe``.
        """
        video_id = video_id or self.video_id or "video"
        faixa_id = faixa_id or self.faixa_id or "faixa"

        falas: list[Fala] = []
        for segmento_id, segmentos in segmentos_por_clipe:
            for segmento in segmentos:
                if _e_ruido(getattr(segmento, "texto", "")):
                    continue
                falas.append(self._fala_de_segmento(segmento, segmento_id, video_id, faixa_id))
        return self._ordenar(falas)

    @staticmethod
    def _fala_de_segmento(
        segmento: object,
        segmento_id: str,
        video_id: str,
        faixa_id: str,
    ) -> Fala:
        texto = str(getattr(segmento, "texto", "")).strip()
        inicio = float(getattr(segmento, "inicio", 0.0))
        fim = float(getattr(segmento, "fim", inicio))
        return Fala(
            id=f"{segmento_id}:{inicio:.3f}",
            video_id=video_id,
            faixa_id=faixa_id,
            segmento_id=segmento_id,
            ordem=-1,  # preenchida na ordenação
            inicio=inicio,
            fim=fim,
            texto=texto,
            texto_normalizado=_normalizar(texto),
            palavras=getattr(segmento, "palavras", ()) or (),
            falante=getattr(segmento, "falante", None),
        )

    @staticmethod
    def _ordenar(falas: list[Fala]) -> list[Fala]:
        falas.sort(key=lambda item: (item.inicio, item.fim))
        for indice, fala in enumerate(falas):
            falas[indice] = Fala(
                id=fala.id,
                video_id=fala.video_id,
                faixa_id=fala.faixa_id,
                segmento_id=fala.segmento_id,
                ordem=indice,
                inicio=fala.inicio,
                fim=fala.fim,
                texto=fala.texto,
                texto_normalizado=fala.texto_normalizado,
                palavras=fala.palavras,
                falante=fala.falante,
            )
        return falas