João Henrique
b01482d0e7
- corrigida a reprodução do trecho de áudio do falante na tela Refinar usando os tempos da origem do WAV. - alterado o passo de geração de JSON para concluir a edição e salvar a configuração editorial no banco. Resumo: - 9 arquivos alterados - 3 novos - 6 modificados - 0 removidos 6 files changed, 136 insertions(+), 109 deletions(-) Arquivos: - .jhonny/analises.db - code/cep-plugin/index.html - code/cep-plugin/main.js - code/engine/persistencia/__init__.py - code/engine/persistencia/esquema.py - code/engine/testes/test_persistencia_sqlite.py - code/engine/persistencia/repositorio_de_edicoes.py - code/engine/registrar_edicao.py - code/tests/cep-refinar-preview.test.ts
301 lines
14 KiB
Python
301 lines
14 KiB
Python
"""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()
|