""" Agrupamento de falas em enunciados embedáveis. Uma fala do Whisper costuma ter de 3 a 8 segundos. Embedar um trecho tão curto produz um vetor instável: pouco texto, muito ruído, e vizinhança semântica pouco confiável. O enunciado resolve isso juntando falas consecutivas do mesmo falante até atingir uma duração alvo, formando um bloco com contexto suficiente para ter significado. O vínculo com as falas de origem é sempre preservado. Isso é o que diferencia esta busca de um RAG genérico: todo acerto semântico precisa voltar com o timecode exato, senão não serve para cortar. """ from __future__ import annotations import sqlite3 from dataclasses import dataclass DURACAO_ALVO_PADRAO = 30.0 DURACAO_MAXIMA_PADRAO = 45.0 INTERVALO_QUE_QUEBRA_BLOCO = 2.0 class ErroDeAgrupamento(ValueError): """Parâmetros de agrupamento inválidos.""" @dataclass(frozen=True) class FalaParaAgrupar: """Uma fala já persistida, no mínimo necessário para agrupá-la.""" segmento_id: int clipe_id: str inicio: float fim: float texto: str falante: str | None @dataclass(frozen=True) class Enunciado: """ Um bloco de falas consecutivas tratado como unidade de busca semântica. Atributos: clipe_id: Clipe a que o bloco pertence. inicio: Início do bloco, herdado da primeira fala. fim: Fim do bloco, herdado da última fala. texto: Texto das falas concatenado. falante: Falante do bloco, quando todas as falas são do mesmo. segmentos: Ids das falas que compõem o bloco, em ordem. """ clipe_id: str inicio: float fim: float texto: str falante: str | None segmentos: tuple[int, ...] @property def duracao(self) -> float: """Duração do bloco em segundos.""" return self.fim - self.inicio class AgrupadorDeEnunciados: """ Junta falas consecutivas em blocos de duração próxima a um alvo. Um bloco é fechado quando atingir a duração alvo, quando o falante mudar, quando houver um silêncio longo entre duas falas ou quando incluir a próxima fala ultrapassaria a duração máxima. A troca de falante e o silêncio longo são fronteiras naturais de assunto: agrupar através delas misturaria ideias distintas no mesmo vetor. Atributos: duracao_alvo: Duração a partir da qual o bloco pode ser fechado. duracao_maxima: Duração que o bloco não deve ultrapassar. intervalo_que_quebra: Silêncio entre falas que força um bloco novo. """ def __init__( self, duracao_alvo: float = DURACAO_ALVO_PADRAO, duracao_maxima: float = DURACAO_MAXIMA_PADRAO, intervalo_que_quebra: float = INTERVALO_QUE_QUEBRA_BLOCO, ) -> None: """ Inicializa o agrupador com os limites de duração dos blocos. Parâmetros: duracao_alvo: Duração a partir da qual o bloco pode ser fechado. duracao_maxima: Duração que o bloco não deve ultrapassar. intervalo_que_quebra: Silêncio entre falas que força bloco novo. Pode gerar: ErroDeAgrupamento: quando as durações não são positivas ou a máxima é menor que a alvo. """ if duracao_alvo <= 0 or duracao_maxima <= 0: raise ErroDeAgrupamento("As durações de agrupamento devem ser positivas.") if duracao_maxima < duracao_alvo: raise ErroDeAgrupamento( "A duração máxima não pode ser menor que a duração alvo." ) self.duracao_alvo = duracao_alvo self.duracao_maxima = duracao_maxima self.intervalo_que_quebra = intervalo_que_quebra def agrupar(self, falas: list[FalaParaAgrupar]) -> list[Enunciado]: """ Agrupa falas ordenadas por tempo em enunciados. Parâmetros: falas: Falas a agrupar. São ordenadas por clipe e início antes do agrupamento, então a ordem de entrada não importa. Retorna: Os enunciados formados, em ordem de tempo. """ ordenadas = sorted(falas, key=lambda fala: (fala.clipe_id, fala.inicio)) enunciados: list[Enunciado] = [] bloco: list[FalaParaAgrupar] = [] for fala in ordenadas: if bloco and self._deve_fechar(bloco, fala): enunciados.append(self._montar(bloco)) bloco = [] bloco.append(fala) if self._duracao(bloco) >= self.duracao_alvo: enunciados.append(self._montar(bloco)) bloco = [] if bloco: enunciados.append(self._montar(bloco)) return enunciados def _deve_fechar(self, bloco: list[FalaParaAgrupar], proxima: FalaParaAgrupar) -> bool: """Decide se a próxima fala pertence a um bloco novo.""" ultima = bloco[-1] if proxima.clipe_id != ultima.clipe_id: return True if proxima.falante != ultima.falante: return True if proxima.fim - bloco[0].inicio > self.duracao_maxima: return True # O silêncio só encerra o bloco depois que ele já tem corpo. Numa fala # pausada, quebrar no primeiro intervalo longo produziria blocos de # poucos segundos — curtos demais para gerar um embedding estável, que # é justamente o problema que o agrupamento existe para resolver. if proxima.inicio - ultima.fim < self.intervalo_que_quebra: return False return self._duracao(bloco) >= self.duracao_alvo / 2 @staticmethod def _duracao(bloco: list[FalaParaAgrupar]) -> float: """Duração coberta por um bloco em formação.""" return bloco[-1].fim - bloco[0].inicio @staticmethod def _montar(bloco: list[FalaParaAgrupar]) -> Enunciado: """Monta o enunciado imutável a partir das falas acumuladas.""" falantes = {fala.falante for fala in bloco} return Enunciado( clipe_id=bloco[0].clipe_id, inicio=bloco[0].inicio, fim=bloco[-1].fim, texto=" ".join(fala.texto.strip() for fala in bloco if fala.texto.strip()), falante=bloco[0].falante if len(falantes) == 1 else None, segmentos=tuple(fala.segmento_id for fala in bloco), ) class RepositorioDeEnunciados: """ Lê falas e grava enunciados no banco de análises. Atributos: conexao: Conexão SQLite já aberta e com o esquema aplicado. """ def __init__(self, conexao: sqlite3.Connection) -> None: """ Inicializa o repositório sobre uma conexão existente. Parâmetros: conexao: Conexão SQLite já aberta e com o esquema aplicado. """ self.conexao = conexao def carregar_falas(self, video_id: str) -> list[FalaParaAgrupar]: """ Carrega as falas de um vídeo no formato aceito pelo agrupador. Parâmetros: video_id: Identificador do vídeo cujas falas serão lidas. Retorna: As falas do vídeo, ordenadas por clipe e início. """ return [ FalaParaAgrupar( segmento_id=linha["id"], clipe_id=linha["clipe_id"], inicio=linha["inicio"], fim=linha["fim"], texto=linha["texto"], falante=linha["falante"], ) for linha in self.conexao.execute( """SELECT id, clipe_id, inicio, fim, texto, falante FROM segmentos_de_transcricao WHERE video_id = ? ORDER BY clipe_id, inicio""", (video_id,), ) ] def substituir_enunciados(self, video_id: str, enunciados: list[Enunciado]) -> int: """ Regrava os enunciados de um vídeo, apagando os anteriores. Os embeddings são removidos junto pelo ``ON DELETE CASCADE``: um enunciado com fronteiras novas não pode herdar o vetor do antigo. Parâmetros: video_id: Identificador do vídeo dono dos enunciados. enunciados: Enunciados a gravar. Retorna: A quantidade de enunciados gravados. """ with self.conexao: self.conexao.execute("DELETE FROM enunciados WHERE video_id = ?", (video_id,)) for enunciado in enunciados: cursor = self.conexao.execute( """INSERT INTO enunciados (video_id, clipe_id, inicio, fim, texto, falante, total_de_falas) VALUES (?, ?, ?, ?, ?, ?, ?)""", (video_id, enunciado.clipe_id, enunciado.inicio, enunciado.fim, enunciado.texto, enunciado.falante, len(enunciado.segmentos)), ) self.conexao.executemany( """INSERT INTO falas_do_enunciado (enunciado_id, segmento_id, ordem) VALUES (?, ?, ?)""", [(cursor.lastrowid, segmento_id, ordem) for ordem, segmento_id in enumerate(enunciado.segmentos)], ) return len(enunciados)