"""Adaptador que expõe a detecção de retakes como um ``Analisador`` do scanner. Permite plugar o ``CoordenadorDeRetakes`` no ``PipelineDoScanner`` como uma etapa entre a detecção de cenas/eventos e a persistência, reutilizando o ``ContextoDeAnalise`` que já carrega as transcrições. """ from __future__ import annotations from typing import Any from ...dominio import Timeline from .adaptador_de_falas import AdaptadorDeFalas from .coordenador_de_retakes import CoordenadorDeRetakes class DeteccaoDeRetakes: nome = "deteccao_de_retakes" def __init__( self, coordenador: CoordenadorDeRetakes | None = None, video_id: str | None = None, adaptador: AdaptadorDeFalas | None = None, ) -> None: self.coordenador = coordenador or CoordenadorDeRetakes() self.video_id = video_id self.adaptador = adaptador or AdaptadorDeFalas() def executar(self, contexto: Any) -> Any: """Consome as transcrições do contexto e preenche ``contexto.retakes``.""" if contexto is None: return contexto transcricoes = getattr(contexto, "transcricoes", None) or {} if not transcricoes: return contexto # ``contexto.transcricoes`` é um dicionário ``identificador_clipe -> # list[SegmentoDeTranscricao]``. Ordenamos por chave para processar as # falas na ordem da timeline. itens = [ (segmento_id, transcricoes[segmento_id]) for segmento_id in sorted(transcricoes) ] timeline: Timeline | None = contexto.timeline video_id = self.video_id or (timeline.identificador if timeline else "video") falas = self.adaptador.converter_de_segmentos(itens, video_id=video_id) # Cenas visuais do contexto entram como reforço (evidência fraca). cenas_visuais = getattr(contexto, "cenas_visuais", None) or [] resultado = self.coordenador.analisar(video_id, falas=falas, cenas_visuais=cenas_visuais) contexto.retakes = { "status": "concluida", "resultado": resultado.para_dict(), "grupos": resultado.grupos, } return contexto