"""Testes da persistência SQLite das análises (timeline, transcrição, evidências visuais, metadados de arquivo e retakes).""" import tempfile import unittest from pathlib import Path from engine.dominio import Clipe, Faixa, IntervaloDeTempo, Timeline from engine.integracoes.midia.extracao_de_metadados import MetadadosDoArquivo from engine.persistencia import ( EdicaoDeVideo, RepositorioDeAnalisesSQLite, RepositorioDeEdicoes, RepositorioDeRetakesSQLite, RepositorioDeTimelineSQLite, abrir_banco, criar_esquema, ) from engine.scanner.modelos import Cena, EvidenciaVisual, PalavraDeTranscricao, SegmentoDeTranscricao from engine.scanner.retakes.modelos_de_retakes import EvidenciaDeRetake, GrupoDeRetake, TomadaDeRetake from engine.scanner.transcricao_da_timeline import TranscricaoDoClipe TABELAS_ESPERADAS = { "videos", "faixas", "clipes", "arquivos", "analises_versao", "segmentos_de_transcricao", "palavras_de_transcricao", "evidencias_visuais", "cenas", "grupos_de_retake", "tomadas_de_retake", "evidencias_de_retake", "edicoes_de_video", } class TesteEsquema(unittest.TestCase): def test_criar_esquema_cria_todas_as_tabelas(self): with tempfile.TemporaryDirectory() as pasta: conexao = abrir_banco(Path(pasta) / "analises.db") tabelas = { linha[0] for linha in conexao.execute( "SELECT name FROM sqlite_master WHERE type = 'table'" ).fetchall() } self.assertTrue(TABELAS_ESPERADAS.issubset(tabelas)) def test_criar_esquema_e_idempotente(self): with tempfile.TemporaryDirectory() as pasta: conexao = abrir_banco(Path(pasta) / "analises.db") criar_esquema(conexao) # não deve levantar exceção ao rodar de novo criar_esquema(conexao) class TesteRepositorioDeRetakesSQLite(unittest.TestCase): def _grupo_de_exemplo(self) -> GrupoDeRetake: return GrupoDeRetake( id="retake_abc123", video_id="v1", faixa_id="f1", tipo="retake_confirmado", confianca=0.92, tomadas=[ TomadaDeRetake(0, "fala-1", "clip-1", 0.0, 6.0, "Hoje nós vamos mostrar.", similaridade_com_anterior=0.0, confianca=0.92), TomadaDeRetake(1, "fala-2", "clip-1", 9.0, 15.0, "Hoje nós vamos mostrar.", similaridade_com_anterior=1.0, confianca=0.92), ], evidencias=[ EvidenciaDeRetake("similaridade_textual", "Textos quase idênticos.", 0.98), EvidenciaDeRetake("mudanca_visual_proxima", "Corte visual próximo.", None), ], ) def test_registrar_e_carregar_preserva_campos(self): with tempfile.TemporaryDirectory() as pasta: repositorio = RepositorioDeRetakesSQLite(Path(pasta) / "analises.db") grupo = self._grupo_de_exemplo() repositorio.registrar("v1", [grupo]) carregados = repositorio.carregar("v1") self.assertEqual(len(carregados), 1) recarregado = carregados[0] self.assertEqual(recarregado.id, grupo.id) self.assertEqual(recarregado.video_id, grupo.video_id) self.assertEqual(recarregado.faixa_id, grupo.faixa_id) self.assertEqual(recarregado.tipo, grupo.tipo) self.assertEqual(recarregado.confianca, grupo.confianca) self.assertEqual(len(recarregado.tomadas), 2) self.assertEqual(recarregado.tomadas[0].fala_id, "fala-1") self.assertEqual(recarregado.tomadas[1].texto, "Hoje nós vamos mostrar.") tipos_de_evidencia = {e.tipo for e in recarregado.evidencias} self.assertEqual(tipos_de_evidencia, {"similaridade_textual", "mudanca_visual_proxima"}) def test_ja_analisado_reflete_hash_de_versao(self): with tempfile.TemporaryDirectory() as pasta: banco = Path(pasta) / "analises.db" repositorio = RepositorioDeRetakesSQLite(banco) grupo = self._grupo_de_exemplo() self.assertFalse(repositorio.ja_analisado("v1")) repositorio.registrar("v1", [grupo]) self.assertTrue(repositorio.ja_analisado("v1")) # Uma versão visual diferente não deve reaproveitar o cache. self.assertFalse(repositorio.ja_analisado("v1", versao_visual="visual")) def test_registrar_de_novo_substitui_grupos_anteriores(self): with tempfile.TemporaryDirectory() as pasta: repositorio = RepositorioDeRetakesSQLite(Path(pasta) / "analises.db") grupo = self._grupo_de_exemplo() repositorio.registrar("v1", [grupo]) repositorio.registrar("v1", []) self.assertEqual(repositorio.carregar("v1"), []) class TesteRepositorioDeAnalisesSQLite(unittest.TestCase): def test_registrar_metadados_de_arquivo_round_trip(self): with tempfile.TemporaryDirectory() as pasta: repositorio = RepositorioDeAnalisesSQLite(Path(pasta) / "analises.db") metadados = MetadadosDoArquivo( caminho=Path("/videos/entrada.mp4"), formato="mov,mp4", duracao=21.4, codec_de_video="h264", codec_de_audio="aac", largura=1920, altura=1080, taxa_de_quadros=29.97, canais_de_audio=2, taxa_de_amostragem=48000, tamanho_em_bytes=123456, orientacao=None, timecode="00:00:00:00", ) repositorio.registrar_metadados_de_arquivo("v1", metadados, hash_do_conteudo="hash123") linha = repositorio.conexao.execute( "SELECT * FROM arquivos WHERE video_id = ?", ("v1",), ).fetchone() self.assertEqual(linha["caminho"], "/videos/entrada.mp4") self.assertEqual(linha["largura"], 1920) self.assertEqual(linha["hash_do_conteudo"], "hash123") class TesteRepositorioDeEdicoes(unittest.TestCase): def test_registrar_configuracao_editorial_preserva_dados(self): with tempfile.TemporaryDirectory() as pasta: conexao = abrir_banco(Path(pasta) / "analises.db") repositorio = RepositorioDeEdicoes(conexao) identificador = repositorio.registrar(EdicaoDeVideo( video_id="sequencia-1", sequencia="Depoimento", origem="/videos/depoimento.mp4", tipo_de_video="depoimento", configuracao={"legendas": {"habilitadas": True}}, )) linha = conexao.execute( "SELECT * FROM edicoes_de_video WHERE id = ?", (identificador,), ).fetchone() self.assertEqual(linha["video_id"], "sequencia-1") self.assertEqual(linha["tipo_de_video"], "depoimento") self.assertIn("habilitadas", linha["configuracao"]) def test_registrar_e_carregar_transcricoes_com_palavras(self): with tempfile.TemporaryDirectory() as pasta: repositorio = RepositorioDeAnalisesSQLite(Path(pasta) / "analises.db") segmento = SegmentoDeTranscricao( inicio=0.0, fim=4.8, texto="Hoje nós vamos mostrar o sistema.", confianca=0.95, palavras=( PalavraDeTranscricao("Hoje", 0.0, 0.4, confianca=0.9), PalavraDeTranscricao("nós", 0.4, 0.7, confianca=0.9), ), emocao="neutro", confianca_emocao=0.6, caracteristicas_acusticas={"pitch_medio": 120.5}, falante="locutor", voz_aparente="masculina", confianca_voz=0.8, ) transcricao = TranscricaoDoClipe("clip-1", "/v.mp4", 0.0, 4.8, [segmento]) repositorio.registrar_transcricoes("v1", [transcricao]) carregados = repositorio.carregar_transcricoes("v1", "clip-1") self.assertEqual(len(carregados), 1) recarregado = carregados[0] self.assertEqual(recarregado.texto, segmento.texto) self.assertEqual(len(recarregado.palavras), 2) self.assertEqual(recarregado.palavras[0].texto, "Hoje") self.assertEqual(recarregado.caracteristicas_acusticas, {"pitch_medio": 120.5}) self.assertEqual(recarregado.voz_aparente, "masculina") def test_substituir_transcricoes_nao_duplica_ao_reprocessar(self): with tempfile.TemporaryDirectory() as pasta: repositorio = RepositorioDeAnalisesSQLite(Path(pasta) / "analises.db") primeiro = TranscricaoDoClipe( "clip-1", "/v.mp4", 0.0, 2.0, [SegmentoDeTranscricao(0.0, 1.0, "primeira versão", 0.9)], ) segundo = TranscricaoDoClipe( "clip-1", "/v.mp4", 0.0, 2.0, [SegmentoDeTranscricao(0.0, 1.0, "segunda versão", 0.95)], ) repositorio.substituir_transcricoes("v1", [primeiro]) repositorio.substituir_transcricoes("v1", [segundo]) carregados = repositorio.carregar_transcricoes("v1", "clip-1") self.assertEqual(len(carregados), 1) self.assertEqual(carregados[0].texto, "segunda versão") def test_atualizar_enriquecimento_persiste_falante_e_metricas(self): with tempfile.TemporaryDirectory() as pasta: repositorio = RepositorioDeAnalisesSQLite(Path(pasta) / "analises.db") segmento = SegmentoDeTranscricao(0.0, 2.0, "fala", 0.9) repositorio.registrar_transcricoes( "v1", [TranscricaoDoClipe("clip-1", "/v.mp4", 0.0, 2.0, [segmento])], ) repositorio.atualizar_enriquecimento_de_transcricao( "v1", "clip-1", 0.0, 2.0, falante="SPEAKER_00", caracteristicas_acusticas={"energia_rms": 0.12}, ) carregado = repositorio.carregar_transcricoes("v1", "clip-1")[0] self.assertEqual(carregado.falante, "SPEAKER_00") self.assertEqual(carregado.caracteristicas_acusticas, {"energia_rms": 0.12}) def test_registrar_evidencias_visuais_e_cenas(self): with tempfile.TemporaryDirectory() as pasta: repositorio = RepositorioDeAnalisesSQLite(Path(pasta) / "analises.db") evidencia = EvidenciaVisual( tipo="qualidade", inicio=1.0, fim=2.0, valor={"nitidez": 0.87}, confianca=0.9, provider="opencv", modelo=None, ) cena = Cena(inicio=0.0, fim=5.0, confianca=0.8, referencias=(1.2, 3.4)) repositorio.registrar_evidencias_visuais("v1", "clip-1", [evidencia]) repositorio.registrar_cenas("v1", [cena], clipe_id="clip-1") linha_evidencia = repositorio.conexao.execute( "SELECT * FROM evidencias_visuais WHERE video_id = ?", ("v1",), ).fetchone() linha_cena = repositorio.conexao.execute( "SELECT * FROM cenas WHERE video_id = ?", ("v1",), ).fetchone() self.assertEqual(linha_evidencia["tipo"], "qualidade") self.assertIn("nitidez", linha_evidencia["valor"]) self.assertEqual(linha_cena["clipe_id"], "clip-1") class TesteRepositorioDeTimelineSQLite(unittest.TestCase): def test_registrar_e_carregar_timeline_com_faixas_e_clipes(self): clipe = Clipe( identificador="clip-1", nome="Take 1", intervalo_na_timeline=IntervaloDeTempo(0.0, 6.0), intervalo_na_origem=IntervaloDeTempo(10.0, 16.0), arquivo="/videos/entrada.mp4", identificador_da_faixa="f1", metadados={"origem": "camera-a"}, ) faixa = Faixa(identificador="f1", nome="V1", tipo="video", indice=0, clipes=[clipe]) timeline = Timeline( identificador="v1", nome="Sequência", duracao=21.4, taxa_de_quadros=29.97, largura=1920, altura=1080, faixas=[faixa], ) with tempfile.TemporaryDirectory() as pasta: repositorio = RepositorioDeTimelineSQLite(Path(pasta) / "analises.db") repositorio.registrar_timeline(timeline) recarregada = repositorio.carregar_timeline("v1") self.assertIsNotNone(recarregada) self.assertEqual(recarregada.identificador, "v1") self.assertEqual(recarregada.largura, 1920) self.assertEqual(len(recarregada.faixas), 1) faixa_recarregada = recarregada.faixas[0] self.assertEqual(faixa_recarregada.identificador, "f1") self.assertEqual(len(faixa_recarregada.clipes), 1) clipe_recarregado = faixa_recarregada.clipes[0] self.assertEqual(clipe_recarregado.arquivo, "/videos/entrada.mp4") self.assertEqual(clipe_recarregado.intervalo_na_origem.inicio, 10.0) self.assertEqual(clipe_recarregado.metadados, {"origem": "camera-a"}) def test_carregar_timeline_inexistente_retorna_none(self): with tempfile.TemporaryDirectory() as pasta: repositorio = RepositorioDeTimelineSQLite(Path(pasta) / "analises.db") self.assertIsNone(repositorio.carregar_timeline("desconhecido")) def test_registrar_de_novo_substitui_faixas_e_clipes(self): faixa_inicial = Faixa(identificador="f1", nome="V1", tipo="video", indice=0, clipes=[ Clipe("clip-1", "Take 1", IntervaloDeTempo(0.0, 6.0)), ]) timeline = Timeline(identificador="v1", nome="Sequência", faixas=[faixa_inicial]) with tempfile.TemporaryDirectory() as pasta: repositorio = RepositorioDeTimelineSQLite(Path(pasta) / "analises.db") repositorio.registrar_timeline(timeline) timeline_atualizada = Timeline(identificador="v1", nome="Sequência", faixas=[]) repositorio.registrar_timeline(timeline_atualizada) recarregada = repositorio.carregar_timeline("v1") self.assertEqual(recarregada.faixas, []) if __name__ == "__main__": unittest.main()