Files
jhonny-editor/code/engine/testes/test_deteccao_de_retakes.py
T
João Henrique ee181360c8 feat: criado repositório jhonny-editor no Gitea
criado repositório jhonny-editor no Gitea
adicionado script admin/deploy.command com commit automático
atualizado admin/DEV-NOTES.md com template limpo

Resumo:
- 32 arquivos alterados
- 16 novos
- 15 modificados
- 0 removidos

 16 files changed, 638 insertions(+), 61 deletions(-)

Arquivos:
  - code/engine/skills/boas-praticas-oo.md -> .agents/skills/boas-praticas-oo/SKILL.md
  - AGENTS.md
  - code/cep-plugin/index.html
  - code/cep-plugin/main.js
  - code/cep-plugin/styles.css
  - code/engine/integracoes/visual/apple_vision.py
  - code/engine/integracoes/visual/apple_vision_runner.swift
  - code/engine/integracoes/visual/extrator_ffmpeg.py
  - code/engine/scanner/__init__.py
  - code/engine/scanner/configuracao_visual.py
  - code/engine/scanner/coordenacao/__init__.py
  - code/engine/scanner/retakes/adaptador_de_falas.py
  - code/engine/scanner/retakes/agrupador_de_takes.py
  - code/engine/scanner/retakes/detector_de_reinicios.py
  - code/engine/scanner/retakes/modelos_de_retakes.py
  - code/engine/scanner/visual.py
  - .jhonny/
  - .retakes/
  - CODING_STANDARDS.md
  - code/engine/executar_retakes.py
  - code/engine/persistencia/
  - code/engine/scanner/retakes/__init__.py
  - code/engine/scanner/retakes/carregador_de_artefatos.py
  - code/engine/scanner/retakes/classificador_de_retakes.py
  - code/engine/scanner/retakes/coordenador_de_retakes.py
  - code/engine/scanner/retakes/deteccao_de_retakes.py
  - code/engine/scanner/retakes/gerador_de_evidencias.py
  - code/engine/scanner/retakes/repositorio_de_retakes.py
  - code/engine/testes/test_consultas_de_persistencia.py
  - code/engine/testes/test_deteccao_de_retakes.py
  - code/engine/testes/test_persistencia_sqlite.py
  - code/relatorios/audio/arquivos/92f97877259a86a8095b1eecafdefe357212a12247d04b78df60e0c0ae38b2ea/
2026-09-08 17:13:19 -04:00

441 lines
20 KiB
Python

"""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()