"""Testes do módulo de detecção de retakes. Cobre o fluxo do ``CoordenadorDeRetakes`` (caso prático do requisito, em que a fala final com informação nova não deve entrar no grupo), os algoritmos de similaridade, o detector de reinícios, o adaptador de pipeline e a idempotência do repositório. """ import json import tempfile import unittest from pathlib import Path from engine.dominio import Clipe, Faixa, IntervaloDeTempo, Timeline from engine.scanner.modelos import PalavraDeTranscricao, SegmentoDeTranscricao from engine.scanner.retakes import ( ClassificadorDeRetakes, ComparadorDeFalas, CoordenadorDeRetakes, DeteccaoDeRetakes, GrupoDeRetake, RepositorioDeRetakes, AgrupamentoDeFalas, Fala, ParAgrupado, ResultadoDoIntervalo, ) from engine.scanner.retakes.adaptador_de_falas import AdaptadorDeFalas from engine.scanner.retakes.modelos_de_retakes import ( STATUS_CONFIRMADO, STATUS_REJEITADO, TomadaDeRetake, ) from engine.scanner.retakes.repositorio_de_retakes import GrupoDeRetakeInexistente from engine.scanner.transcricao_da_timeline import TranscricaoDoClipe def _fala(fala_id: str, texto: str, inicio: float, fim: float, video_id: str = "v1", faixa_id: str = "f1", segmento_id: str = "clip-1") -> Fala: return Fala( id=fala_id, video_id=video_id, faixa_id=faixa_id, segmento_id=segmento_id, ordem=0, inicio=inicio, fim=fim, texto=texto, texto_normalizado=AdaptadorDeFalas.normalizar(texto), falante="locutor", ) class TesteComparadorDeFalas(unittest.TestCase): def setUp(self): self.comparador = ComparadorDeFalas() def test_textos_identicos_possuem_similaridade_maxima(self): a = _fala("a", "Hoje nós vamos mostrar como funciona o sistema.", 0.0, 4.8) b = _fala("b", "Hoje nós vamos mostrar como funciona o sistema.", 6.2, 10.9) resultado = self.comparador.comparar(a, b) self.assertEqual(resultado.similaridade_jaccard, 1.0) self.assertEqual(resultado.similaridade_do_inicio, 1.0) self.assertEqual(resultado.similaridade_final, 1.0) def test_ordem_diferente_penaliza_sequencia_mas_nao_jaccard(self): a = _fala("a", "Vamos mostrar o sistema funcionando.", 0.0, 3.0) b = _fala("b", "O sistema será mostrado funcionando.", 4.0, 8.0) resultado = self.comparador.comparar(a, b) self.assertGreater(resultado.similaridade_jaccard, 0.3) self.assertLess(resultado.similaridade_de_sequencia, 1.0) def test_prefixo_igual_eleva_similaridade_do_inicio(self): a = _fala("a", "Hoje nós vamos mostrar o sistema.", 0.0, 3.0) b = _fala("b", "Hoje nós vamos explicar o programa.", 4.0, 7.0) resultado = self.comparador.comparar(a, b) self.assertGreaterEqual(resultado.similaridade_do_inicio, 0.5) self.assertGreater(resultado.similaridade_do_inicio, resultado.similaridade_do_final) class TesteDetectorDeReinicios(unittest.TestCase): def test_detecta_reinicio_explicito_e_repeticao_do_inicio(self): from engine.scanner.retakes import DetectorDeReinicios falas = [ _fala("101", "Hoje nós vamos mostrar como funciona o sistema.", 0.0, 6.0), _fala("102", "Não, peraí.", 6.0, 8.0), _fala("103", "Hoje nós vamos mostrar como funciona o sistema.", 9.0, 15.0), ] detector = DetectorDeReinicios() sinais = detector.detectar(falas) tipos = {sinal.fala_id: sinal.tipo for sinal in sinais} self.assertIn("102", tipos) self.assertEqual(tipos["102"], "reinicio_explicito") self.assertIn("marcador", sinais[0].evidencias[0]) class TesteClassificador(unittest.TestCase): def test_replicacao_completa_gera_retake_confirmado(self): a = _fala("a", "Hoje nós vamos mostrar como funciona o sistema.", 0.0, 6.0) b = _fala("b", "Hoje nós vamos mostrar como funciona o sistema.", 9.0, 15.0) comparacao = ComparadorDeFalas().comparar(a, b) par = ParAgrupado(a, b, comparacao, ResultadoDoIntervalo(3.0, True, 0.9, True), 1.0) agrupamento = AgrupamentoDeFalas((a.id, b.id), (par,)) classificacao = ClassificadorDeRetakes().classificar(agrupamento) self.assertEqual(classificacao.tipo, "retake_confirmado") self.assertGreaterEqual(classificacao.confianca, 0.8) class TesteCoordenadorEExemploDoRequisito(unittest.TestCase): def _montar_transcricoes(self): segmentos = [ SegmentoDeTranscricao(0.0, 6.0, "Hoje nós vamos mostrar como funciona o sistema.", palavras=( PalavraDeTranscricao("Hoje", 0.0, 0.5), PalavraDeTranscricao("sistema", 5.0, 6.0), )), SegmentoDeTranscricao(6.0, 8.0, "Não, peraí."), SegmentoDeTranscricao(9.0, 15.0, "Hoje nós vamos mostrar como funciona o sistema."), SegmentoDeTranscricao(16.0, 21.0, "Primeiro, vamos abrir o programa."), ] clipe = Clipe("clip-1", "clip-1", IntervaloDeTempo(0, 21), arquivo="/v.mp4") timeline = Timeline("t1", "Sequência", faixas=[Faixa("f1", "a1", "audio", 0, [clipe])]) frase = TranscricaoDoClipe("clip-1", "/v.mp4", 0.0, 21.0, segmentos) return frase, timeline def test_exemplo_pratico_agrupa_repetidas_e_exclui_informacao_nova(self): frase, _ = self._montar_transcricoes() with tempfile.TemporaryDirectory() as pasta: repositorio = RepositorioDeRetakes(pasta) coordenador = CoordenadorDeRetakes(repositorio_de_retakes=repositorio) resultado = coordenador.analisar("v1", [frase]) self.assertEqual(len(resultado.grupos), 1) grupo = resultado.grupos[0] self.assertEqual(len(grupo.tomadas), 2) self.assertEqual(grupo.tomadas[0].inicio, 0.0) self.assertEqual(grupo.tomadas[1].inicio, 9.0) textos = " ".join(t.texto for t in grupo.tomadas) self.assertNotIn("Primeiro", textos) tipos = {evidencia.tipo for evidencia in grupo.evidencias} self.assertIn("similaridade_textual", tipos) def test_repositorio_e_idempotente(self): frase, _ = self._montar_transcricoes() with tempfile.TemporaryDirectory() as pasta: repositorio = RepositorioDeRetakes(pasta) coordenador = CoordenadorDeRetakes(repositorio_de_retakes=repositorio) primeiro = coordenador.analisar("v1", [frase]) segundo = coordenador.analisar("v1", [frase]) ja_analisado = repositorio.ja_analisado("v1") # A segunda execução não gera um grupo novo (mesmo id de cache). self.assertEqual(primeiro.grupos[0].id, segundo.grupos[0].id) self.assertTrue(ja_analisado) class TesteAdaptadorDePipeline(unittest.TestCase): def test_deteccao_de_retakes_popula_contexto(self): from engine.scanner.coordenacao import ContextoDeAnalise segmentos = [ SegmentoDeTranscricao(0.0, 6.0, "Hoje nós vamos mostrar como funciona o sistema."), SegmentoDeTranscricao(9.0, 15.0, "Hoje nós vamos mostrar como funciona o sistema."), ] contexto = ContextoDeAnalise(timeline=Timeline("t1", "Seq")) contexto.transcricoes = {"clip-1": segmentos} with tempfile.TemporaryDirectory() as pasta: detector = DeteccaoDeRetakes( coordenador=CoordenadorDeRetakes( repositorio_de_retakes=RepositorioDeRetakes(pasta))) novo = detector.executar(contexto) self.assertIsNotNone(novo.retakes) self.assertEqual(novo.retakes["status"], "concluida") self.assertEqual(len(novo.retakes["grupos"]), 1) class TesteIntegracaoVisualAuxiliar(unittest.TestCase): """Cobre a Etapa 2: PySceneDetect plugado via config e cenas visuais como evidência fraca nos retakes (reforço, nunca regra obrigatória).""" def test_config_visual_aceita_novos_campos_e_preserva_defaults(self): from engine.scanner.configuracao_visual import ConfiguracaoVisualDoScanner perfil = ConfiguracaoVisualDoScanner.de_dict({ "usar_analise_visual": True, "detectar_cortes": True, "sensibilidade": "alta", }) self.assertTrue(perfil.usar_analise_visual) self.assertTrue(perfil.detectar_cortes) self.assertEqual(perfil.sensibilidade, "alta") self.assertLess(perfil.limiar_de_corte, 27) # alta = mais sensível # Round-trip pelo JSON do painel. self.assertTrue(perfil.para_dict()["usar_analise_visual"]) self.assertEqual(perfil.para_dict()["sensibilidade"], "alta") # Ausência dos campos não quebra o comportamento anterior. padrao = ConfiguracaoVisualDoScanner.de_dict({}) self.assertFalse(padrao.usar_analise_visual) self.assertTrue(padrao.detectar_cortes) self.assertEqual(padrao.limiar_de_corte, 27.0) def test_config_rejeita_sensibilidade_invalida(self): from engine.scanner.configuracao_visual import ConfiguracaoVisualDoScanner with self.assertRaises(ValueError): ConfiguracaoVisualDoScanner(sensibilidade="impossivel") def test_detector_local_pluga_pyscenedetect_quando_ativado(self): from engine.scanner.configuracao_visual import ConfiguracaoVisualDoScanner from engine.scanner.visual import criar_detector_visual_local ativo = criar_detector_visual_local("cache", configuracao=ConfiguracaoVisualDoScanner( usar_analise_visual=True, sensibilidade="media")) self.assertEqual(len(ativo.detectores_de_intervalo), 1) desligado = criar_detector_visual_local("cache") self.assertEqual(desligado.detectores_de_intervalo, []) def test_cena_visual_gera_evidencia_fraca_no_retake(self): from engine.scanner.retakes import CoordenadorDeRetakes, RepositorioDeRetakes from engine.scanner.modelos import CenaVisual falas = [ _fala("101", "Hoje nós vamos mostrar como funciona o sistema.", 0.0, 6.0), _fala("102", "Hoje nós vamos mostrar como funciona o sistema.", 9.0, 15.0), ] # Um corte visual em 8.0s (na pausa entre as tentativas). cenas_visuais = [CenaVisual("clip-1", 8.0, 9.0)] with tempfile.TemporaryDirectory() as pasta: coordenador = CoordenadorDeRetakes( repositorio_de_retakes=RepositorioDeRetakes(pasta)) resultado = coordenador.analisar("v1", falas=falas, cenas_visuais=cenas_visuais) self.assertEqual(len(resultado.grupos), 1) grupo = resultado.grupos[0] tipos = {evidencia.tipo for evidencia in grupo.evidencias} self.assertIn("mudanca_visual_proxima", tipos) def test_sem_corte_visual_nao_gera_evidencia_visual(self): from engine.scanner.retakes import CoordenadorDeRetakes, RepositorioDeRetakes falas = [ _fala("101", "Hoje nós vamos mostrar como funciona o sistema.", 0.0, 6.0), _fala("102", "Hoje nós vamos mostrar como funciona o sistema.", 9.0, 15.0), ] # Sem cenas visuais informadas (comportamento anterior). with tempfile.TemporaryDirectory() as pasta: coordenador = CoordenadorDeRetakes( repositorio_de_retakes=RepositorioDeRetakes(pasta)) resultado = coordenador.analisar("v1", falas=falas) grupo = resultado.grupos[0] tipos = {evidencia.tipo for evidencia in grupo.evidencias} self.assertNotIn("mudanca_visual_proxima", tipos) def test_mudanca_de_uso_visual_forca_reanalise_por_causa_do_hash(self): from engine.scanner.retakes import CoordenadorDeRetakes, RepositorioDeRetakes from engine.scanner.modelos import CenaVisual falas = [ _fala("101", "Hoje nós vamos mostrar como funciona o sistema.", 0.0, 6.0), _fala("102", "Hoje nós vamos mostrar como funciona o sistema.", 9.0, 15.0), ] with tempfile.TemporaryDirectory() as pasta: repositorio = RepositorioDeRetakes(pasta) coordenador = CoordenadorDeRetakes(repositorio_de_retakes=repositorio) # 1ª análise sem visual. sem_visual = coordenador.analisar("v1", falas=falas) tipos_sem = {e.tipo for e in sem_visual.grupos[0].evidencias} self.assertNotIn("mudanca_visual_proxima", tipos_sem) # 2ª análise com visual: o hash muda → não reaproveita a antiga. com_visual = coordenador.analisar( "v1", falas=falas, cenas_visuais=[CenaVisual("clip-1", 8.0, 9.0)]) tipos_com = {e.tipo for e in com_visual.grupos[0].evidencias} self.assertIn("mudanca_visual_proxima", tipos_com) class TesteRevisaoManualEArtefatos(unittest.TestCase): """Cobre a tela de retakes: revisão manual via repositório e leitura dos artefatos de transcrição do scanner (sem retranscrever).""" def _grupo(self, grupo_id: str = "retake_001"): from engine.scanner.retakes import GrupoDeRetake return GrupoDeRetake( id=grupo_id, video_id="v1", faixa_id="f1", tipo="retake_de_fala", confianca=0.9, tomadas=[TomadaDeRetake(ordem=1, fala_id="a", segmento_id="s", inicio=0.0, fim=4.0, texto="olá mundo")]) def test_revisao_manual_grava_status_e_preserva_demais_grupos(self): from engine.scanner.retakes import RepositorioDeRetakes with tempfile.TemporaryDirectory() as pasta: repositorio = RepositorioDeRetakes(diretoria_de_trabalho=pasta) repositorio.registrar("v1", [self._grupo(), self._grupo("retake_002")]) atualizado = repositorio.atualizar_status_de_revisao( "v1", "retake_001", STATUS_CONFIRMADO) self.assertEqual(atualizado.status_de_revisao, STATUS_CONFIRMADO) grupos = repositorio.carregar("v1") self.assertEqual(grupos[0].status_de_revisao, STATUS_CONFIRMADO) self.assertEqual(grupos[1].status_de_revisao, "pendente_de_revisao") def test_revisao_manual_rejeita_status_invalido_e_grupo_inexistente(self): from engine.scanner.retakes import RepositorioDeRetakes with tempfile.TemporaryDirectory() as pasta: repositorio = RepositorioDeRetakes(diretoria_de_trabalho=pasta) repositorio.registrar("v1", [self._grupo()]) with self.assertRaises(ValueError): repositorio.atualizar_status_de_revisao("v1", "retake_001", "talvez") with self.assertRaises(GrupoDeRetakeInexistente): repositorio.atualizar_status_de_revisao( "v1", "outro", STATUS_REJEITADO) def test_carregador_le_artefato_do_scanner_e_ignora_sem_transcricao(self): import json from pathlib import Path from engine.scanner.retakes import ( ArtefatoDeTranscricaoInvalido, CarregadorDeArtefatosDeAudio, ) artefato = { "status": "concluida", "segmentos_por_clipe": [ {"clipe": "clip-1", "segmentos": [ {"inicio": 0.0, "fim": 4.0, "texto": "Hoje nós vamos mostrar como funciona o sistema."}, {"inicio": 6.0, "fim": 10.0, "texto": "Hoje nós vamos mostrar como funciona o sistema."}, ]}, ], } with tempfile.TemporaryDirectory() as pasta: caminho = Path(pasta) / "v1-audio-faixa-1.json" caminho.write_text(json.dumps(artefato, ensure_ascii=False), encoding="utf-8") falas = CarregadorDeArtefatosDeAudio().carregar([caminho], "v1") self.assertEqual(len(falas), 2) self.assertTrue(falas[0].texto.startswith("Hoje")) caminho_vazio = Path(pasta) / "v1-audio-faixa-2.json" caminho_vazio.write_text(json.dumps({"status": "pendente"}), encoding="utf-8") with self.assertRaises(ArtefatoDeTranscricaoInvalido): CarregadorDeArtefatosDeAudio().carregar([caminho_vazio], "v1") def test_carregador_ignora_artefato_inexistente(self): from engine.scanner.retakes import ( ArtefatoDeTranscricaoInvalido, CarregadorDeArtefatosDeAudio, ) with self.assertRaises(ArtefatoDeTranscricaoInvalido): CarregadorDeArtefatosDeAudio().carregar( ["/caminho/inexistente.json"], "v1") def _grupo_exemplo(diretoria: str): """Registra um grupo mínimo no repositório e o devolve.""" repositorio = RepositorioDeRetakes(diretoria_de_trabalho=diretoria) grupo = GrupoDeRetake(id="g1", video_id="v1", faixa_id="f1", tipo="retake_confirmado", confianca=0.9) repositorio.registrar("v1", [grupo]) return grupo class TesteRevisaoManualECarregador(unittest.TestCase): """Cobre a revisão manual do repositório e o carregador de artefatos.""" def setUp(self) -> None: self.pasta = tempfile.TemporaryDirectory() self.addCleanup(self.pasta.cleanup) self.repositorio = RepositorioDeRetakes( diretoria_de_trabalho=self.pasta.name) def test_status_padrao_e_pendente(self) -> None: grupo = _grupo_exemplo(self.pasta.name) self.assertEqual(grupo.status_de_revisao, "pendente_de_revisao") def test_revisao_atualiza_e_persiste(self) -> None: _grupo_exemplo(self.pasta.name) atualizado = self.repositorio.atualizar_status_de_revisao( "v1", "g1", "confirmado") self.assertEqual(atualizado.status_de_revisao, "confirmado") recarregado = self.repositorio.carregar("v1")[0] self.assertEqual(recarregado.status_de_revisao, "confirmado") def test_revisao_rejeita_status_invalido(self) -> None: _grupo_exemplo(self.pasta.name) with self.assertRaises(ValueError): self.repositorio.atualizar_status_de_revisao("v1", "g1", "talvez") def test_revisao_de_grupo_inexistente_levanta_erro(self) -> None: with self.assertRaises(LookupError): self.repositorio.atualizar_status_de_revisao("v1", "g9", "confirmado") def test_carregador_converte_artefato_do_scanner(self) -> None: from engine.scanner.retakes import CarregadorDeArtefatosDeAudio artefato = Path(self.pasta.name) / "v1-audio-faixa-1.json" artefato.write_text(json.dumps({ "status": "concluida", "segmentos_por_clipe": [{"clipe": "clipe-1", "segmentos": [ {"inicio": 0.0, "fim": 4.0, "texto": "Hoje vamos testar."}, {"inicio": 5.0, "fim": 9.0, "texto": "Hoje vamos testar."}, ]}], }), encoding="utf-8") falas = CarregadorDeArtefatosDeAudio().carregar([artefato], "v1") self.assertEqual(len(falas), 2) self.assertEqual(falas[0].video_id, "v1") def test_carregador_levanta_erro_sem_transcricao(self) -> None: from engine.scanner.retakes import ( ArtefatoDeTranscricaoInvalido, CarregadorDeArtefatosDeAudio) vazio = Path(self.pasta.name) / "vazio-audio-faixa-1.json" vazio.write_text(json.dumps({"status": "pendente"}), encoding="utf-8") with self.assertRaises(ArtefatoDeTranscricaoInvalido): CarregadorDeArtefatosDeAudio().carregar([vazio], "vazio") if __name__ == "__main__": unittest.main()