João Henrique
c1b544f4b5
- reorganizado o fluxo de edição para validar a análise do Scanner, selecionar o tipo de vídeo e enviar suas instruções no JSON. - banco de análises: métricas de fala viraram colunas, busca lexical FTS5, enunciados com embeddings, views achatadas de leitura (fala/palavra/linha do tempo/fala com visual), ingestor do pipeline de voz e gravação idempotente de evidências visuais e cenas. - retakes passam a ser gravados no banco de análises (SQLite) em vez de JSON por caso, com status de revisão persistido. - planos de edição e suas aplicações passam a ser registrados no banco, em vez de se perderem no arquivo temporário. - comando de limpeza das evidências visuais e cenas duplicadas por execuções antigas do Scanner. - análise visual: OpenCV (rostos) e PySceneDetect passam a entrar no detector local por padrão; novo adapter InsightFace gera assinatura facial (embedding) para reconhecer a mesma pessoa entre tomadas, gravada como evidência visual no banco. - corrigido: métricas de fala (energia, pitch, velocidade) agora gravam nas colunas dedicadas, não só no JSON; a view fala+visual passa a casar por vídeo e tempo, já que o mesmo arquivo entra na timeline como clipes distintos de vídeo e áudio; views são recriadas a cada abertura do banco para uma correção de consulta chegar a bancos já existentes. - reordenadas as abas do painel CEP para Scanner, Refinar e Editar vídeo Resumo: - 24 arquivos alterados - 9 novos - 15 modificados - 0 removidos 15 files changed, 872 insertions(+), 29 deletions(-) Arquivos: - .jhonny/analises.db - code/cep-plugin/index.html - code/cep-plugin/main.js - code/engine/aplicar_plano_de_edicao.py - code/engine/integracoes/visual/README.md - code/engine/integracoes/visual/__init__.py - code/engine/integracoes/visual/analisadores.py - code/engine/persistencia/__init__.py - code/engine/persistencia/esquema.py - code/engine/persistencia/repositorio_de_analises_sqlite.py - code/engine/persistencia/repositorio_de_retakes_sqlite.py - code/engine/requirements-visual.txt - code/engine/scanner/configuracao_visual.py - code/engine/scanner/visual.py - code/engine/testes/test_analise_visual_local.py - .jhonny/analises.db.pos-scanner-084841 - code/engine/integracoes/embeddings/ - code/engine/limpar_duplicatas.py - code/engine/persistencia/busca_de_conteudo.py - code/engine/persistencia/enunciados.py - code/engine/persistencia/ingestao_de_voz.py - code/engine/persistencia/limpeza.py - code/engine/persistencia/repositorio_de_planos.py - code/engine/testes/test_busca_de_conteudo.py
348 lines
16 KiB
Python
348 lines
16 KiB
Python
"""Testes da ingestão de voz, do agrupamento em enunciados e da busca."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from engine.persistencia.busca_de_conteudo import (
|
|
BuscaDeConteudo,
|
|
ErroDeBusca,
|
|
IndexadorSemantico,
|
|
)
|
|
from engine.persistencia.conexao import abrir_banco
|
|
from engine.persistencia.enunciados import (
|
|
AgrupadorDeEnunciados,
|
|
ErroDeAgrupamento,
|
|
FalaParaAgrupar,
|
|
RepositorioDeEnunciados,
|
|
)
|
|
from engine.persistencia.esquema import reconstruir_indice_de_busca
|
|
from engine.persistencia.ingestao_de_voz import (
|
|
ErroDeIngestaoDeVoz,
|
|
FalaDoPipelineDeVoz,
|
|
IngestorDeVozNoBanco,
|
|
LeitorDeDadosParaIA,
|
|
PalavraDoPipelineDeVoz,
|
|
)
|
|
from engine.persistencia.limpeza import LimpadorDeDuplicatas
|
|
|
|
|
|
def _dados_para_ia() -> dict:
|
|
"""Monta um ``dados-para-ia.json`` mínimo com as três análises de áudio."""
|
|
return {
|
|
"schema_version": "1.0",
|
|
"source": {"file_name": "teste.mp4", "duration_seconds": 20.0},
|
|
"segments": [
|
|
{
|
|
"id": 1, "start": 0.0, "end": 4.0, "text": "Bom dia a todos.",
|
|
"speaker": "SPEAKER_00",
|
|
"energy_rms": 0.12, "pitch_hz_median": 180.0, "pitch_hz_std": 12.0,
|
|
"speaking_rate_wps": 1.0, "longest_internal_pause_s": 0.2,
|
|
"gap_before_s": None,
|
|
"words": [
|
|
{"text": "Bom", "start": 0.0, "end": 0.5, "confidence": 0.99},
|
|
{"text": "dia", "start": 0.5, "end": 1.0, "confidence": 0.98},
|
|
],
|
|
},
|
|
{
|
|
"id": 2, "start": 5.0, "end": 9.0, "text": "Vamos falar de elegância.",
|
|
"speaker": "SPEAKER_00",
|
|
"energy_rms": 0.20, "pitch_hz_median": 200.0, "pitch_hz_std": 15.0,
|
|
"speaking_rate_wps": 1.2, "longest_internal_pause_s": 0.1,
|
|
"gap_before_s": 1.0,
|
|
"words": [{"text": "Vamos", "start": 5.0, "end": 5.4, "confidence": 0.97}],
|
|
},
|
|
],
|
|
}
|
|
|
|
|
|
class TesteLeitorDeDadosParaIA(unittest.TestCase):
|
|
"""Verifica a normalização do JSON do pipeline de voz."""
|
|
|
|
def _escrever(self, conteudo: object, pasta: str) -> Path:
|
|
caminho = Path(pasta) / "dados-para-ia.json"
|
|
caminho.write_text(json.dumps(conteudo), encoding="utf-8")
|
|
return caminho
|
|
|
|
def test_le_falas_com_falante_metricas_e_palavras(self) -> None:
|
|
with tempfile.TemporaryDirectory() as pasta:
|
|
falas = LeitorDeDadosParaIA().ler(self._escrever(_dados_para_ia(), pasta))
|
|
self.assertEqual(len(falas), 2)
|
|
self.assertEqual(falas[0].texto, "Bom dia a todos.")
|
|
self.assertEqual(falas[0].falante, "SPEAKER_00")
|
|
self.assertEqual(falas[0].metricas["energy_rms"], 0.12)
|
|
self.assertEqual(len(falas[0].palavras), 2)
|
|
self.assertAlmostEqual(falas[0].duracao, 4.0)
|
|
|
|
def test_ordena_as_falas_por_inicio(self) -> None:
|
|
dados = _dados_para_ia()
|
|
dados["segments"].reverse()
|
|
with tempfile.TemporaryDirectory() as pasta:
|
|
falas = LeitorDeDadosParaIA().ler(self._escrever(dados, pasta))
|
|
self.assertEqual([fala.inicio for fala in falas], [0.0, 5.0])
|
|
|
|
def test_arquivo_inexistente_gera_erro_especifico(self) -> None:
|
|
with self.assertRaises(ErroDeIngestaoDeVoz):
|
|
LeitorDeDadosParaIA().ler("/caminho/que/nao/existe.json")
|
|
|
|
def test_json_sem_segments_gera_erro_especifico(self) -> None:
|
|
with tempfile.TemporaryDirectory() as pasta:
|
|
caminho = self._escrever({"source": {}}, pasta)
|
|
with self.assertRaises(ErroDeIngestaoDeVoz):
|
|
LeitorDeDadosParaIA().ler(caminho)
|
|
|
|
def test_segmento_sem_tempo_gera_erro_que_nomeia_o_campo(self) -> None:
|
|
dados = _dados_para_ia()
|
|
del dados["segments"][0]["start"]
|
|
with tempfile.TemporaryDirectory() as pasta:
|
|
caminho = self._escrever(dados, pasta)
|
|
with self.assertRaises(ErroDeIngestaoDeVoz) as contexto:
|
|
LeitorDeDadosParaIA().ler(caminho)
|
|
self.assertIn("start", str(contexto.exception))
|
|
|
|
|
|
class TesteIngestorDeVozNoBanco(unittest.TestCase):
|
|
"""Verifica a gravação das falas do pipeline de voz."""
|
|
|
|
def _conexao(self, pasta: str):
|
|
return abrir_banco(Path(pasta) / "analises.db")
|
|
|
|
def test_metricas_viram_colunas_consultaveis(self) -> None:
|
|
with tempfile.TemporaryDirectory() as pasta:
|
|
conexao = self._conexao(pasta)
|
|
with tempfile.TemporaryDirectory() as origem:
|
|
caminho = Path(origem) / "dados.json"
|
|
caminho.write_text(json.dumps(_dados_para_ia()), encoding="utf-8")
|
|
falas = LeitorDeDadosParaIA().ler(caminho)
|
|
resultado = IngestorDeVozNoBanco(conexao).ingerir("VID", "CLIPE", falas)
|
|
|
|
self.assertEqual(resultado.falas, 2)
|
|
self.assertEqual(resultado.palavras, 3)
|
|
self.assertEqual(resultado.falantes, 1)
|
|
linha = conexao.execute(
|
|
"""SELECT energia_rms, pitch_mediano_hz, velocidade_de_fala_pps, falante
|
|
FROM segmentos_de_transcricao WHERE inicio = 0.0"""
|
|
).fetchone()
|
|
self.assertAlmostEqual(linha["energia_rms"], 0.12)
|
|
self.assertAlmostEqual(linha["pitch_mediano_hz"], 180.0)
|
|
self.assertEqual(linha["falante"], "SPEAKER_00")
|
|
|
|
def test_reingerir_o_mesmo_clipe_nao_duplica(self) -> None:
|
|
with tempfile.TemporaryDirectory() as pasta:
|
|
conexao = self._conexao(pasta)
|
|
falas = (FalaDoPipelineDeVoz(
|
|
inicio=0.0, fim=1.0, texto="oi", falante="A",
|
|
palavras=(PalavraDoPipelineDeVoz("oi", 0.0, 1.0),),
|
|
),)
|
|
ingestor = IngestorDeVozNoBanco(conexao)
|
|
for _ in range(3):
|
|
ingestor.ingerir("VID", "CLIPE", falas)
|
|
total = conexao.execute(
|
|
"SELECT count(*) FROM segmentos_de_transcricao").fetchone()[0]
|
|
self.assertEqual(total, 1)
|
|
|
|
def test_identificador_vazio_gera_erro_especifico(self) -> None:
|
|
with tempfile.TemporaryDirectory() as pasta:
|
|
with self.assertRaises(ErroDeIngestaoDeVoz):
|
|
IngestorDeVozNoBanco(self._conexao(pasta)).ingerir("", "CLIPE", ())
|
|
|
|
|
|
class TesteAgrupadorDeEnunciados(unittest.TestCase):
|
|
"""Verifica as fronteiras de agrupamento das falas em enunciados."""
|
|
|
|
def _fala(self, identificador: int, inicio: float, fim: float,
|
|
falante: str | None = "A", clipe: str = "C1") -> FalaParaAgrupar:
|
|
return FalaParaAgrupar(identificador, clipe, inicio, fim, f"fala {identificador}", falante)
|
|
|
|
def test_agrupa_falas_vizinhas_do_mesmo_falante(self) -> None:
|
|
falas = [self._fala(i, i * 5.0, i * 5.0 + 4.5) for i in range(4)]
|
|
enunciados = AgrupadorDeEnunciados().agrupar(falas)
|
|
self.assertEqual(len(enunciados), 1)
|
|
self.assertEqual(enunciados[0].segmentos, (0, 1, 2, 3))
|
|
|
|
def test_troca_de_falante_fecha_o_bloco(self) -> None:
|
|
falas = [self._fala(0, 0.0, 4.0, "A"), self._fala(1, 4.1, 8.0, "B")]
|
|
enunciados = AgrupadorDeEnunciados().agrupar(falas)
|
|
self.assertEqual(len(enunciados), 2)
|
|
self.assertEqual([e.falante for e in enunciados], ["A", "B"])
|
|
|
|
def test_clipes_diferentes_nunca_se_juntam(self) -> None:
|
|
falas = [self._fala(0, 0.0, 4.0, clipe="C1"), self._fala(1, 4.1, 8.0, clipe="C2")]
|
|
self.assertEqual(len(AgrupadorDeEnunciados().agrupar(falas)), 2)
|
|
|
|
def test_silencio_nao_quebra_bloco_ainda_curto(self) -> None:
|
|
"""Quebrar cedo produziria blocos curtos demais para embedar."""
|
|
falas = [self._fala(0, 0.0, 2.0), self._fala(1, 10.0, 12.0)]
|
|
self.assertEqual(len(AgrupadorDeEnunciados().agrupar(falas)), 1)
|
|
|
|
def test_duracao_maxima_e_respeitada(self) -> None:
|
|
falas = [self._fala(i, i * 10.0, i * 10.0 + 9.5) for i in range(8)]
|
|
enunciados = AgrupadorDeEnunciados().agrupar(falas)
|
|
self.assertTrue(all(e.duracao <= 45.0 for e in enunciados))
|
|
|
|
def test_duracao_maxima_menor_que_alvo_gera_erro(self) -> None:
|
|
with self.assertRaises(ErroDeAgrupamento):
|
|
AgrupadorDeEnunciados(duracao_alvo=30.0, duracao_maxima=10.0)
|
|
|
|
|
|
class _ProviderFalso:
|
|
"""Provider determinístico: vetor derivado das letras, sem rede."""
|
|
|
|
modelo = "falso"
|
|
dimensoes = 4
|
|
|
|
def _vetor(self, texto: str) -> tuple[float, ...]:
|
|
base = [0.0] * self.dimensoes
|
|
for posicao, letra in enumerate(texto.lower()):
|
|
if letra.isalpha():
|
|
base[posicao % self.dimensoes] += ord(letra) % 7
|
|
return tuple(base)
|
|
|
|
def gerar_para_documento(self, texto: str) -> tuple[float, ...]:
|
|
return self._vetor(texto)
|
|
|
|
def gerar_para_consulta(self, texto: str) -> tuple[float, ...]:
|
|
return self._vetor(texto)
|
|
|
|
|
|
class TesteBuscaDeConteudo(unittest.TestCase):
|
|
"""Verifica a busca lexical, a semântica e a fusão das duas."""
|
|
|
|
def _banco_com_falas(self, pasta: str):
|
|
conexao = abrir_banco(Path(pasta) / "analises.db")
|
|
falas = (
|
|
FalaDoPipelineDeVoz(inicio=0.0, fim=4.0,
|
|
texto="Vamos falar de elegância e postura.", falante="A"),
|
|
FalaDoPipelineDeVoz(inicio=5.0, fim=9.0,
|
|
texto="A cirurgia leva cerca de duas horas.", falante="A"),
|
|
)
|
|
IngestorDeVozNoBanco(conexao).ingerir("VID", "C1", falas)
|
|
return conexao
|
|
|
|
def test_busca_lexical_ignora_acento_e_devolve_timecode(self) -> None:
|
|
with tempfile.TemporaryDirectory() as pasta:
|
|
conexao = self._banco_com_falas(pasta)
|
|
resultados = BuscaDeConteudo(conexao).buscar_lexical("elegancia")
|
|
self.assertEqual(len(resultados), 1)
|
|
self.assertEqual(resultados[0].inicio, 0.0)
|
|
self.assertEqual(resultados[0].fim, 4.0)
|
|
self.assertEqual(resultados[0].origem, "lexical")
|
|
|
|
def test_busca_lexical_nao_inventa_resultado(self) -> None:
|
|
"""Diferente da semântica, a lexical sabe dizer que não achou."""
|
|
with tempfile.TemporaryDirectory() as pasta:
|
|
conexao = self._banco_com_falas(pasta)
|
|
self.assertEqual(BuscaDeConteudo(conexao).buscar_lexical("paralelepipedo"), [])
|
|
|
|
def test_indice_lexical_acompanha_remocao_de_fala(self) -> None:
|
|
with tempfile.TemporaryDirectory() as pasta:
|
|
conexao = self._banco_com_falas(pasta)
|
|
conexao.execute("DELETE FROM segmentos_de_transcricao WHERE inicio = 0.0")
|
|
conexao.commit()
|
|
self.assertEqual(BuscaDeConteudo(conexao).buscar_lexical("elegancia"), [])
|
|
|
|
def test_reconstruir_indice_recupera_falas_pre_existentes(self) -> None:
|
|
with tempfile.TemporaryDirectory() as pasta:
|
|
conexao = self._banco_com_falas(pasta)
|
|
self.assertEqual(reconstruir_indice_de_busca(conexao), 2)
|
|
|
|
def test_busca_semantica_devolve_falas_de_origem(self) -> None:
|
|
with tempfile.TemporaryDirectory() as pasta:
|
|
conexao = self._banco_com_falas(pasta)
|
|
repositorio = RepositorioDeEnunciados(conexao)
|
|
enunciados = AgrupadorDeEnunciados().agrupar(repositorio.carregar_falas("VID"))
|
|
repositorio.substituir_enunciados("VID", enunciados)
|
|
provider = _ProviderFalso()
|
|
self.assertEqual(IndexadorSemantico(conexao, provider).indexar("VID"), 1)
|
|
resultados = BuscaDeConteudo(conexao, provider).buscar_semantica("elegância")
|
|
self.assertEqual(len(resultados), 1)
|
|
self.assertTrue(resultados[0].falas)
|
|
|
|
def test_sem_provider_a_busca_semantica_devolve_vazio_sem_falhar(self) -> None:
|
|
with tempfile.TemporaryDirectory() as pasta:
|
|
conexao = self._banco_com_falas(pasta)
|
|
self.assertEqual(BuscaDeConteudo(conexao).buscar_semantica("qualquer"), [])
|
|
|
|
def test_consulta_vazia_gera_erro_especifico(self) -> None:
|
|
with tempfile.TemporaryDirectory() as pasta:
|
|
conexao = self._banco_com_falas(pasta)
|
|
with self.assertRaises(ErroDeBusca):
|
|
BuscaDeConteudo(conexao).buscar(" ")
|
|
|
|
def test_pontuacao_do_usuario_nao_quebra_a_consulta(self) -> None:
|
|
with tempfile.TemporaryDirectory() as pasta:
|
|
conexao = self._banco_com_falas(pasta)
|
|
resultados = BuscaDeConteudo(conexao).buscar_lexical('cirurgia "quanto?"')
|
|
self.assertIsInstance(resultados, list)
|
|
|
|
|
|
class TesteViewsAchatadas(unittest.TestCase):
|
|
"""Verifica que as views entregam a leitura única esperada pelo agente."""
|
|
|
|
def test_view_de_palavras_repete_o_que_vale_para_a_frase(self) -> None:
|
|
with tempfile.TemporaryDirectory() as pasta:
|
|
conexao = abrir_banco(Path(pasta) / "analises.db")
|
|
falas = (FalaDoPipelineDeVoz(
|
|
inicio=0.0, fim=2.0, texto="Bom dia", falante="A",
|
|
metricas={"energy_rms": 0.5},
|
|
palavras=(PalavraDoPipelineDeVoz("Bom", 0.0, 1.0),
|
|
PalavraDoPipelineDeVoz("dia", 1.0, 2.0)),
|
|
),)
|
|
IngestorDeVozNoBanco(conexao).ingerir("VID", "C1", falas)
|
|
linhas = conexao.execute(
|
|
"SELECT palavra, falante, frase FROM vw_palavras_completas ORDER BY ordem"
|
|
).fetchall()
|
|
self.assertEqual([linha["palavra"] for linha in linhas], ["Bom", "dia"])
|
|
self.assertTrue(all(linha["falante"] == "A" for linha in linhas))
|
|
self.assertTrue(all(linha["frase"] == "Bom dia" for linha in linhas))
|
|
|
|
def test_view_de_falas_conta_as_palavras(self) -> None:
|
|
with tempfile.TemporaryDirectory() as pasta:
|
|
conexao = abrir_banco(Path(pasta) / "analises.db")
|
|
falas = (FalaDoPipelineDeVoz(
|
|
inicio=0.0, fim=2.0, texto="Bom dia", falante="A",
|
|
palavras=(PalavraDoPipelineDeVoz("Bom", 0.0, 1.0),
|
|
PalavraDoPipelineDeVoz("dia", 1.0, 2.0)),
|
|
),)
|
|
IngestorDeVozNoBanco(conexao).ingerir("VID", "C1", falas)
|
|
linha = conexao.execute(
|
|
"SELECT total_de_palavras FROM vw_falas_completas").fetchone()
|
|
self.assertEqual(linha["total_de_palavras"], 2)
|
|
|
|
|
|
class TesteLimpadorDeDuplicatas(unittest.TestCase):
|
|
"""Verifica a remoção das duplicatas deixadas por execuções antigas."""
|
|
|
|
def test_remove_copias_exatas_e_preserva_a_primeira(self) -> None:
|
|
with tempfile.TemporaryDirectory() as pasta:
|
|
conexao = abrir_banco(Path(pasta) / "analises.db")
|
|
conexao.execute("INSERT OR IGNORE INTO videos (id) VALUES ('VID')")
|
|
for _ in range(3):
|
|
conexao.execute(
|
|
"""INSERT INTO cenas (video_id, clipe_id, inicio, fim, confianca, referencias)
|
|
VALUES ('VID', 'C1', 1.0, 2.0, 0.9, '[]')""")
|
|
conexao.commit()
|
|
limpador = LimpadorDeDuplicatas(conexao)
|
|
self.assertEqual(limpador.contar_duplicatas().cenas, 2)
|
|
self.assertEqual(limpador.limpar().cenas, 2)
|
|
self.assertEqual(
|
|
conexao.execute("SELECT count(*) FROM cenas").fetchone()[0], 1)
|
|
|
|
def test_linhas_diferentes_sao_preservadas(self) -> None:
|
|
with tempfile.TemporaryDirectory() as pasta:
|
|
conexao = abrir_banco(Path(pasta) / "analises.db")
|
|
conexao.execute("INSERT OR IGNORE INTO videos (id) VALUES ('VID')")
|
|
for inicio in (1.0, 2.0, 3.0):
|
|
conexao.execute(
|
|
"""INSERT INTO cenas (video_id, clipe_id, inicio, fim, confianca, referencias)
|
|
VALUES ('VID', 'C1', ?, ?, 0.9, '[]')""", (inicio, inicio + 1))
|
|
conexao.commit()
|
|
self.assertEqual(LimpadorDeDuplicatas(conexao).limpar().cenas, 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|