feat: reorganizado o fluxo de edição para validar a análise do Sca

- 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
This commit is contained in:
João Henrique
2026-09-10 08:58:22 -04:00
parent 1156619937
commit c1b544f4b5
25 changed files with 2792 additions and 29 deletions
Binary file not shown.
Binary file not shown.
+3 -3
View File
@@ -22,15 +22,15 @@
</header>
<nav class="tab-bar" role="tablist" aria-label="Painel">
<button class="tab-button" id="tabBtnSilence" role="tab" aria-selected="false" aria-controls="tabSilence" onclick="switchTab('silence')" type="button">
<span class="tab-icon" aria-hidden="true">✂</span>Editar vídeo
</button>
<button class="tab-button active" id="tabBtnScanner" role="tab" aria-selected="true" aria-controls="tabScanner" onclick="switchTab('scanner')" type="button">
<span class="tab-icon" aria-hidden="true">◉</span>Scanner
</button>
<button class="tab-button" id="tabBtnRefinar" role="tab" aria-selected="false" aria-controls="tabRefinar" onclick="switchTab('refinar')" type="button">
<span class="tab-icon" aria-hidden="true">◌</span>Refinar
</button>
<button class="tab-button" id="tabBtnSilence" role="tab" aria-selected="false" aria-controls="tabSilence" onclick="switchTab('silence')" type="button">
<span class="tab-icon" aria-hidden="true">✂</span>Editar vídeo
</button>
<button class="tab-button" id="tabBtnRetakes" role="tab" aria-selected="false" aria-controls="tabRetakes" onclick="switchTab('retakes')" type="button">
<span class="tab-icon" aria-hidden="true">↻</span>Retakes
</button>
+1 -1
View File
@@ -243,9 +243,9 @@ function stepState(n) {
// ---- Tabs ----
function switchTab(name) {
var tabs = [
{ key: "silence", panel: "tabSilence", btn: "tabBtnSilence" },
{ key: "scanner", panel: "tabScanner", btn: "tabBtnScanner" },
{ key: "refinar", panel: "tabRefinar", btn: "tabBtnRefinar" },
{ key: "silence", panel: "tabSilence", btn: "tabBtnSilence" },
{ key: "retakes", panel: "tabRetakes", btn: "tabBtnRetakes" },
{ key: "analises", panel: "tabAnalises", btn: "tabBtnAnalises" },
{ key: "tipos-video", panel: "tabTiposVideo", btn: "tabBtnTiposVideo" },
+42
View File
@@ -20,6 +20,7 @@ from __future__ import annotations
import json
import os
import shutil
import sqlite3
import sys
from pathlib import Path
@@ -41,6 +42,9 @@ from engine.integracoes.premiere.leitura import AcessoAoEditor, AcessoATimeline
from engine.integracoes.premiere.sessao_mcp import SessaoMCP
CAMINHO_DO_SERVIDOR_MCP = CAMINHO_DO_CODIGO / "dist" / "index.js"
# Mesmo banco de análises do Scanner: o plano aplicado passa a ficar ao lado
# das falas e evidências que o originaram.
CAMINHO_DO_BANCO = CAMINHO_DO_CODIGO.parent / ".jhonny" / "analises.db"
def resolver_caminho_do_node() -> str:
@@ -77,6 +81,42 @@ def resolver_caminho_do_node() -> str:
)
def _registrar_no_banco(caminho_do_plano: Path, resultado: object, sequencia: str) -> None:
"""Guarda no banco de análises o plano aplicado e o resultado da aplicação.
O registro é o único vestígio de como uma edição foi decidida: sem ele o
plano sumiria junto com o arquivo temporário. Uma falha ao gravar não
interrompe a edição já aplicada na timeline — apenas avisa em stderr,
porque perder o histórico é bem menos grave do que abortar um corte que
já foi feito.
Parâmetros:
caminho_do_plano: Arquivo JSON do plano que foi aplicado.
resultado: Resultado devolvido pelo aplicador.
sequencia: Nome da sequência onde o plano foi aplicado.
"""
try:
from engine.persistencia.conexao import abrir_banco
from engine.persistencia.repositorio_de_planos import (
RepositorioDePlanos,
plano_de_dict,
)
dados = json.loads(caminho_do_plano.read_text(encoding="utf-8"))
repositorio = RepositorioDePlanos(abrir_banco(CAMINHO_DO_BANCO))
plano_id = repositorio.registrar_plano(plano_de_dict(dados))
aplicadas = sum(1 for item in resultado.resultados if item.sucesso)
repositorio.registrar_aplicacao(
plano_id,
sucesso=resultado.todas_bem_sucedidas,
acoes_aplicadas=aplicadas,
sequencia=sequencia,
)
except (OSError, ValueError, sqlite3.Error) as erro:
print(f"Plano aplicado, mas não foi possível registrá-lo no banco: {erro}",
file=sys.stderr)
def executar(caminho_do_plano: Path) -> dict:
"""Lê o plano em ``caminho_do_plano`` e o aplica na sequência ativa do Premiere.
@@ -114,6 +154,8 @@ def executar(caminho_do_plano: Path) -> dict:
)
resultado = aplicador.aplicar(plano)
_registrar_no_banco(caminho_do_plano, resultado, sequencia_ativa.nome)
return {
"arquivo_de_origem": plano.arquivo_de_origem,
"todas_bem_sucedidas": resultado.todas_bem_sucedidas,
@@ -0,0 +1,21 @@
"""Providers de embeddings usados pela busca semântica de conteúdo."""
from .provider_ollama import (
DIMENSOES_PADRAO,
ErroDeEmbedding,
MODELO_PADRAO,
ProviderDeEmbeddings,
ProviderDeEmbeddingsOllama,
desempacotar,
empacotar,
)
__all__ = [
"DIMENSOES_PADRAO",
"ErroDeEmbedding",
"MODELO_PADRAO",
"ProviderDeEmbeddings",
"ProviderDeEmbeddingsOllama",
"desempacotar",
"empacotar",
]
@@ -0,0 +1,180 @@
"""
Geração de embeddings por um Ollama local.
O projeto já roda um Ollama nesta máquina com o ``nomic-embed-text`` — o mesmo
modelo que o RAG de código usa. Reaproveitá-lo mantém a análise de conteúdo
local: nenhum trecho de transcrição de cliente sai da máquina, e não há custo
por chamada.
O ``nomic-embed-text`` exige prefixos distintos para o que é indexado e para o
que é consultado. Sem eles a similaridade cai de forma silenciosa, então os
prefixos são aplicados aqui e não ficam a cargo de quem chama.
"""
from __future__ import annotations
import json
import struct
import urllib.error
import urllib.request
from typing import Protocol, Sequence
URL_PADRAO = "http://localhost:11434"
MODELO_PADRAO = "nomic-embed-text"
DIMENSOES_PADRAO = 768
PREFIXO_DE_DOCUMENTO = "search_document: "
PREFIXO_DE_CONSULTA = "search_query: "
class ErroDeEmbedding(RuntimeError):
"""Falha ao gerar um embedding — serviço fora do ar, modelo ausente ou resposta inválida."""
class ProviderDeEmbeddings(Protocol):
"""Contrato de quem transforma texto em vetor."""
@property
def modelo(self) -> str:
"""Nome do modelo que gerou os vetores."""
...
@property
def dimensoes(self) -> int:
"""Quantidade de dimensões dos vetores gerados."""
...
def gerar_para_documento(self, texto: str) -> tuple[float, ...]:
"""Gera o vetor de um texto que será indexado."""
...
def gerar_para_consulta(self, texto: str) -> tuple[float, ...]:
"""Gera o vetor de um texto que está sendo buscado."""
...
class ProviderDeEmbeddingsOllama:
"""
Gera embeddings chamando um Ollama local por HTTP.
Atributos:
modelo: Nome do modelo de embeddings usado.
dimensoes: Dimensões esperadas nos vetores devolvidos.
url: Endereço base do serviço Ollama.
tempo_limite: Segundos a esperar por cada resposta.
"""
def __init__(
self,
modelo: str = MODELO_PADRAO,
dimensoes: int = DIMENSOES_PADRAO,
url: str = URL_PADRAO,
tempo_limite: float = 60.0,
) -> None:
"""
Inicializa o provider com o modelo e o endereço do serviço.
Parâmetros:
modelo: Nome do modelo de embeddings no Ollama.
dimensoes: Dimensões esperadas — vetores de outro tamanho são
recusados, para não misturar modelos no mesmo índice.
url: Endereço base do Ollama.
tempo_limite: Segundos a esperar por cada resposta.
"""
self._modelo = modelo
self._dimensoes = dimensoes
self.url = url.rstrip("/")
self.tempo_limite = tempo_limite
@property
def modelo(self) -> str:
"""Nome do modelo que gera os vetores."""
return self._modelo
@property
def dimensoes(self) -> int:
"""Quantidade de dimensões dos vetores gerados."""
return self._dimensoes
def gerar_para_documento(self, texto: str) -> tuple[float, ...]:
"""
Gera o vetor de um enunciado que será indexado.
Parâmetros:
texto: Texto do enunciado.
Retorna:
O vetor normalizado do texto.
Pode gerar:
ErroDeEmbedding: quando o Ollama não responde ou devolve um vetor
com dimensões diferentes das esperadas.
"""
return self._gerar(PREFIXO_DE_DOCUMENTO + texto)
def gerar_para_consulta(self, texto: str) -> tuple[float, ...]:
"""
Gera o vetor de uma consulta feita pelo usuário.
Parâmetros:
texto: Texto da consulta.
Retorna:
O vetor normalizado da consulta.
Pode gerar:
ErroDeEmbedding: quando o Ollama não responde ou devolve um vetor
com dimensões diferentes das esperadas.
"""
return self._gerar(PREFIXO_DE_CONSULTA + texto)
def _gerar(self, texto: str) -> tuple[float, ...]:
"""Chama o Ollama e valida o vetor devolvido."""
corpo = json.dumps({"model": self._modelo, "prompt": texto}).encode("utf-8")
requisicao = urllib.request.Request(
f"{self.url}/api/embeddings", data=corpo,
headers={"Content-Type": "application/json"},
)
try:
with urllib.request.urlopen(requisicao, timeout=self.tempo_limite) as resposta:
dados = json.loads(resposta.read().decode("utf-8"))
except urllib.error.URLError as erro:
raise ErroDeEmbedding(
f"Não foi possível falar com o Ollama em {self.url}: {erro}"
) from erro
except json.JSONDecodeError as erro:
raise ErroDeEmbedding(f"Resposta do Ollama não é JSON válido: {erro}") from erro
vetor = dados.get("embedding")
if not isinstance(vetor, list) or not vetor:
raise ErroDeEmbedding(f"O Ollama não devolveu embedding para o modelo {self._modelo}.")
if len(vetor) != self._dimensoes:
raise ErroDeEmbedding(
f"Embedding com {len(vetor)} dimensões, esperado {self._dimensoes}."
)
return tuple(float(valor) for valor in vetor)
def empacotar(vetor: Sequence[float]) -> bytes:
"""
Serializa um vetor para o BLOB gravado no banco.
Parâmetros:
vetor: Valores do vetor.
Retorna:
Os valores como float de 32 bits em sequência.
"""
return struct.pack(f"<{len(vetor)}f", *vetor)
def desempacotar(dados: bytes) -> tuple[float, ...]:
"""
Reconstrói um vetor a partir do BLOB lido do banco.
Parâmetros:
dados: Bytes gravados por ``empacotar``.
Retorna:
Os valores do vetor.
"""
return struct.unpack(f"<{len(dados) // 4}f", dados)
+16 -1
View File
@@ -62,6 +62,7 @@ testadas isoladamente.
| Continuidade e frame congelado | `AnalisadorDeContinuidadeOpenCV` |
| Detectar objetos | `AnalisadorDeObjetosONNX` |
| Detectar pose e mãos | `AnalisadorDePoseMediaPipe` |
| Reconhecer a mesma pessoa entre tomadas (assinatura facial) | `AnalisadorDeRostosInsightFace` |
| OCR e faces macOS | `AnalisadorAppleVision` |
| Vision + Apple Intelligence (Swift isolado) | `AnalisadorAppleVisionNativo` |
| Similaridade temporal por feature print | `AnalisadorSequenciaAppleVisionNativo` |
@@ -82,7 +83,21 @@ encerrar o processo do Scanner.
`AnalisadorDeRostosOpenCV` usa os arquivos locais
`haarcascade_frontalface_default.xml` e `haarcascade_eye.xml`. Quando a
distribuição do OpenCV não os incluir, forneça `diretorio_de_modelos` apontando
para a pasta que os contém.
para a pasta que os contém. Ele entra automaticamente no detector local
(`criar_detector_visual_local`) quando `"faces"` está em `recursos_vision`
(padrão do perfil).
`AnalisadorDeRostosInsightFace` resolve o que nem o OpenCV nem o Apple Vision
fazem: reconhecer que o rosto de um clipe é a mesma pessoa de outro clipe. Ele
devolve duas evidências por rosto — `rosto` (mesma forma de bounding box dos
outros adapters) e `identidade_facial` (com `assinatura_facial`, um vetor de
512 posições, mais `idade_aproximada`/`genero_aparente` quando o modelo os
estimar). Aceita tanto o frame decodificado em `quadro.imagem` (pipeline
OpenCV) quanto um frame persistido em `quadro.caminho` (pipeline Apple
Vision/FFmpeg), lendo o PNG com OpenCV nesse segundo caso. Por ser mais pesado
(baixa o modelo `buffalo_l` na primeira execução), só entra nos detectores
locais e no detector Apple Vision quando `perfil.identidade_facial=True` —
desligado por padrão.
`AnalisadorAppleVisionNativo` é a opção Apple recomendada. Seu runner Swift
executa faces/landmarks, qualidade facial, pessoas, pose, mãos, OCR,
+3 -2
View File
@@ -1,7 +1,8 @@
from .analisadores import (AnalisadorAppleVision, AnalisadorDeComposicaoOpenCV,
AnalisadorDeContinuidadeOpenCV, AnalisadorDeObjetosONNX,
AnalisadorDePoseMediaPipe, AnalisadorDeQualidadeOpenCV,
AnalisadorDeRostosOpenCV, AnalisadorDeTremorOpenCV)
AnalisadorDeRostosInsightFace, AnalisadorDeRostosOpenCV,
AnalisadorDeTremorOpenCV)
from .apple_vision import (AnalisadorAppleVisionNativo, AnalisadorSequenciaAppleVisionNativo,
ExecutorDoRunnerApple)
from .contratos import (AnalisadorDeFrame, AnalisadorDeSequenciaDeQuadros,
@@ -14,7 +15,7 @@ from .modelos import QuadroDeVideo
__all__ = ["AnalisadorAppleVision", "AnalisadorAppleVisionNativo", "AnalisadorSequenciaAppleVisionNativo", "AnalisadorDeComposicaoOpenCV",
"AnalisadorDeContinuidadeOpenCV", "AnalisadorDeFrame",
"AnalisadorDeObjetosONNX", "AnalisadorDePoseMediaPipe",
"AnalisadorDeQualidadeOpenCV", "AnalisadorDeRostosOpenCV",
"AnalisadorDeQualidadeOpenCV", "AnalisadorDeRostosInsightFace", "AnalisadorDeRostosOpenCV",
"AnalisadorDeSequenciaDeQuadros", "AnalisadorDeTremorOpenCV",
"DetectorDeCenasPySceneDetect", "DetectorDeIntervalosDeCena",
"ExecutorDoRunnerApple", "ExtratorDeQuadros", "ExtratorDeQuadrosFFmpeg", "ExtratorDeQuadrosOpenCV", "QuadroDeVideo"]
@@ -105,6 +105,113 @@ class AnalisadorDeRostosOpenCV(_AdapterOpenCV, AnalisadorDeFrame):
return self._detectores
class AnalisadorDeRostosInsightFace(AnalisadorDeFrame):
"""Detecta rostos e extrai a assinatura facial (embedding) com InsightFace.
Complementa `AnalisadorDeRostosOpenCV` e o Apple Vision: ambos encontram
o rosto, mas nenhum reconhece que dois rostos em clipes diferentes são a
mesma pessoa. A assinatura facial normalizada é o fato que a camada
editorial usa depois para agrupar tomadas pela pessoa em cena.
"""
nome = "insightface"
def __init__(self, modelo: str = "buffalo_l", tamanho_de_deteccao: tuple[int, int] = (640, 640),
usar_coreml: bool = True, tamanho_minimo_relativo: float = 0.04,
app: Any | None = None) -> None:
if tamanho_minimo_relativo < 0 or tamanho_minimo_relativo >= 1:
raise ValueError("tamanho_minimo_relativo deve estar entre 0 e 1.")
self.modelo = modelo
self.tamanho_de_deteccao = tamanho_de_deteccao
self.usar_coreml = usar_coreml
self.tamanho_minimo_relativo = tamanho_minimo_relativo
self._app = app
@property
def app(self) -> Any:
"""Instância preparada do `FaceAnalysis`, criada de forma tardia."""
if self._app is None:
try:
from insightface.app import FaceAnalysis
except ImportError as exc:
raise RuntimeError(
"InsightFace não está instalado. Instale insightface e onnxruntime."
) from exc
self._app = self._preparar_app(FaceAnalysis)
return self._app
def _preparar_app(self, classe_face_analysis: Any) -> Any:
"""Monta o `FaceAnalysis` com os providers do ONNX Runtime disponíveis."""
try:
import onnxruntime
except ImportError as exc:
raise RuntimeError("InsightFace requer onnxruntime instalado.") from exc
preferidos = ["CoreMLExecutionProvider", "CPUExecutionProvider"] if self.usar_coreml \
else ["CPUExecutionProvider"]
disponiveis = set(onnxruntime.get_available_providers())
providers = [item for item in preferidos if item in disponiveis] or ["CPUExecutionProvider"]
app = classe_face_analysis(name=self.modelo, providers=providers)
app.prepare(ctx_id=0, det_size=self.tamanho_de_deteccao)
return app
def analisar(self, quadro: QuadroDeVideo) -> list[EvidenciaVisual]:
"""Devolve uma evidência de rosto e, quando possível, sua assinatura facial."""
rostos = self.app.get(self._obter_imagem(quadro))
evidencias: list[EvidenciaVisual] = []
for rosto in rostos:
caixa = _caixa_normalizada_do_retangulo(rosto.bbox, quadro.largura, quadro.altura)
if min(caixa["largura"], caixa["altura"]) < self.tamanho_minimo_relativo:
continue
confianca = float(rosto.det_score) if getattr(rosto, "det_score", None) is not None else None
evidencias.append(EvidenciaVisual("rosto", quadro.timestamp, quadro.timestamp,
{"bounding_box": caixa, "proximo_da_borda": _proximo_da_borda(caixa)},
confianca=confianca, provider=self.nome, modelo=self.modelo))
evidencia_de_identidade = self._evidencia_de_identidade(rosto, quadro, caixa, confianca)
if evidencia_de_identidade is not None:
evidencias.append(evidencia_de_identidade)
return evidencias
def _obter_imagem(self, quadro: QuadroDeVideo) -> Any:
"""Devolve a matriz BGR do frame, decodificando o arquivo quando necessário.
O extrator local (OpenCV) já entrega o frame decodificado em
``quadro.imagem``. O extrator usado com o Apple Vision persiste PNGs
e mantém ``imagem`` vazio para não carregar OpenCV nesse processo; o
InsightFace precisa do array decodificado, então lê o PNG nesse caso.
"""
if quadro.imagem is not None:
return quadro.imagem
if quadro.caminho is None:
raise ValueError("InsightFace requer o frame decodificado ou persistido em disco.")
try:
import cv2
except ImportError as exc:
raise RuntimeError("Ler o frame do disco para o InsightFace requer OpenCV.") from exc
imagem = cv2.imread(str(quadro.caminho))
if imagem is None:
raise RuntimeError(f"Não foi possível ler o frame do InsightFace: {quadro.caminho}")
return imagem
def _evidencia_de_identidade(self, rosto: Any, quadro: QuadroDeVideo, caixa: dict[str, float],
confianca: float | None) -> EvidenciaVisual | None:
embedding = getattr(rosto, "normed_embedding", None)
if embedding is None:
return None
valor: dict[str, Any] = {
"bounding_box": caixa,
"assinatura_facial": [float(item) for item in embedding],
"dimensoes": len(embedding),
}
idade = getattr(rosto, "age", None)
if idade is not None:
valor["idade_aproximada"] = int(idade)
genero = getattr(rosto, "sex", None)
if genero is not None:
valor["genero_aparente"] = str(genero)
return EvidenciaVisual("identidade_facial", quadro.timestamp, quadro.timestamp, valor,
confianca=confianca, provider=self.nome, modelo=self.modelo)
class AnalisadorDeComposicaoOpenCV(_AdapterOpenCV, AnalisadorDeFrame):
"""Mede orientação, poluição visual e disponibilidade das zonas para legendas."""
@@ -302,6 +409,20 @@ def _caixa_normalizada(x: int, y: int, largura: int, altura: int,
"largura": float(largura / largura_do_frame), "altura": float(altura / altura_do_frame)}
def _caixa_normalizada_do_retangulo(bbox: Any, largura_do_frame: int, altura_do_frame: int) -> dict[str, float]:
"""Converte um retângulo em cantos ``[x1, y1, x2, y2]`` (pixels) para a caixa normalizada.
Detectores como o InsightFace podem devolver cantos levemente fora do
quadro; por isso o resultado é sempre recortado para o intervalo [0, 1].
"""
x1, y1, x2, y2 = (float(valor) for valor in bbox)
x = min(max(x1 / largura_do_frame, 0.0), 1.0)
y = min(max(y1 / altura_do_frame, 0.0), 1.0)
largura = min(max(x2 / largura_do_frame, 0.0), 1.0) - x
altura = min(max(y2 / altura_do_frame, 0.0), 1.0) - y
return {"x": x, "y": y, "largura": max(largura, 0.0), "altura": max(altura, 0.0)}
def _proximo_da_borda(caixa: dict[str, float], margem: float = 0.04) -> bool:
return (caixa["x"] < margem or caixa["y"] < margem
or caixa["x"] + caixa["largura"] > 1 - margem
+95
View File
@@ -0,0 +1,95 @@
"""
Comando de limpeza das duplicatas do banco de análises.
Remove as evidências visuais e cenas duplicadas por execuções antigas do
Scanner, que acrescentavam em vez de substituir. Faz backup do banco antes de
apagar qualquer linha, e aceita ``--simular`` para apenas contar.
Uso:
python code/engine/limpar_duplicatas.py [--banco CAMINHO] [--simular]
"""
from __future__ import annotations
import argparse
import shutil
import sys
from datetime import datetime
from pathlib import Path
CAMINHO_DO_CODIGO = Path(__file__).resolve().parent.parent
if str(CAMINHO_DO_CODIGO) not in sys.path:
sys.path.insert(0, str(CAMINHO_DO_CODIGO))
from engine.persistencia.conexao import abrir_banco # noqa: E402
from engine.persistencia.esquema import reconstruir_indice_de_busca # noqa: E402
from engine.persistencia.limpeza import LimpadorDeDuplicatas # noqa: E402
BANCO_PADRAO = CAMINHO_DO_CODIGO.parent / ".jhonny" / "analises.db"
def criar_backup(banco: Path) -> Path:
"""
Copia o banco para um arquivo datado ao lado do original.
Parâmetros:
banco: Caminho do banco a copiar.
Retorna:
O caminho do backup criado.
"""
carimbo = datetime.now().strftime("%Y%m%d-%H%M%S")
destino = banco.with_name(f"{banco.name}.backup-{carimbo}")
shutil.copy2(banco, destino)
return destino
def executar(banco: Path, simular: bool) -> int:
"""
Conta e, quando autorizado, remove as duplicatas do banco.
Parâmetros:
banco: Caminho do banco de análises.
simular: Quando verdadeiro, apenas conta e não apaga nada.
Retorna:
Código de saída do processo: 0 em sucesso, 1 quando o banco não existe.
"""
if not banco.is_file():
print(f"Banco não encontrado: {banco}", file=sys.stderr)
return 1
conexao = abrir_banco(banco)
limpador = LimpadorDeDuplicatas(conexao)
duplicatas = limpador.contar_duplicatas()
print(f"Duplicatas encontradas: {duplicatas.evidencias_visuais} evidência(s) visual(is), "
f"{duplicatas.cenas} cena(s).")
if duplicatas.total == 0:
print("Nada a remover.")
return 0
if simular:
print("Simulação: nenhuma linha foi apagada.")
return 0
backup = criar_backup(banco)
print(f"Backup criado em {backup}")
removidas = limpador.limpar()
print(f"Removidas: {removidas.evidencias_visuais} evidência(s) visual(is), "
f"{removidas.cenas} cena(s).")
print(f"Índice de busca reconstruído com {reconstruir_indice_de_busca(conexao)} fala(s).")
return 0
def principal() -> int:
"""Lê os argumentos da linha de comando e executa a limpeza."""
parser = argparse.ArgumentParser(description="Remove duplicatas do banco de análises.")
parser.add_argument("--banco", type=Path, default=BANCO_PADRAO,
help="Caminho do analises.db.")
parser.add_argument("--simular", action="store_true",
help="Apenas conta as duplicatas, sem apagar.")
argumentos = parser.parse_args()
return executar(argumentos.banco, argumentos.simular)
if __name__ == "__main__":
raise SystemExit(principal())
+33 -1
View File
@@ -7,16 +7,48 @@ grupos de retake, com repositórios que espelham a interface dos
repositórios JSON já existentes.
"""
from .busca_de_conteudo import BuscaDeConteudo, ErroDeBusca, IndexadorSemantico, ResultadoDeBusca
from .conexao import abrir_banco
from .consultas import ConsultasDeAnalises, ErroDeConsultaInvalida
from .esquema import criar_esquema
from .enunciados import AgrupadorDeEnunciados, Enunciado, RepositorioDeEnunciados
from .esquema import criar_esquema, reconstruir_indice_de_busca
from .ingestao_de_voz import (
ErroDeIngestaoDeVoz,
IngestorDeVozNoBanco,
LeitorDeDadosParaIA,
)
from .limpeza import LimpadorDeDuplicatas
from .repositorio_de_planos import (
AcaoDoPlano,
ErroDePlano,
PlanoDeEdicao,
RepositorioDePlanos,
plano_de_dict,
)
from .repositorio_de_analises_sqlite import RepositorioDeAnalisesSQLite
from .repositorio_de_retakes_sqlite import RepositorioDeRetakesSQLite
from .repositorio_de_timeline_sqlite import RepositorioDeTimelineSQLite
__all__ = [
"AcaoDoPlano",
"AgrupadorDeEnunciados",
"BuscaDeConteudo",
"Enunciado",
"ErroDeBusca",
"ErroDeIngestaoDeVoz",
"ErroDePlano",
"IndexadorSemantico",
"IngestorDeVozNoBanco",
"LeitorDeDadosParaIA",
"LimpadorDeDuplicatas",
"PlanoDeEdicao",
"RepositorioDeEnunciados",
"RepositorioDePlanos",
"ResultadoDeBusca",
"abrir_banco",
"criar_esquema",
"plano_de_dict",
"reconstruir_indice_de_busca",
"RepositorioDeAnalisesSQLite",
"RepositorioDeRetakesSQLite",
"RepositorioDeTimelineSQLite",
@@ -0,0 +1,363 @@
"""
Busca de conteúdo falado no banco de análises.
Combina duas listas independentes sobre o mesmo material: a lexical, que acha
a palavra exata via FTS5, e a semântica, que acha a ideia via embeddings dos
enunciados. Nenhuma das duas basta sozinha — a lexical não encontra "preço"
quando a pessoa disse "quanto custa", e a semântica erra nomes próprios e
termos técnicos que precisam bater literalmente.
Toda resposta carrega ``clipe_id``, ``inicio`` e ``fim``. Um acerto sem
timecode não serve para cortar, e por isso não é considerado resultado válido
aqui.
"""
from __future__ import annotations
import math
import sqlite3
from dataclasses import dataclass
from ..integracoes.embeddings import ErroDeEmbedding, ProviderDeEmbeddings, desempacotar
# Constante da fusão recíproca de ranking (RRF). Amortece a diferença entre as
# escalas incomparáveis do FTS5 (bm25, menor é melhor) e do cosseno (maior é
# melhor): o que entra na conta é a posição em cada lista, não a nota.
CONSTANTE_DE_FUSAO = 60.0
class ErroDeBusca(ValueError):
"""Consulta vazia ou parâmetros de busca inválidos."""
@dataclass(frozen=True)
class ResultadoDeBusca:
"""
Um trecho encontrado, sempre localizável na timeline.
Atributos:
video_id: Vídeo onde o trecho está.
clipe_id: Clipe onde o trecho está.
inicio: Início do trecho em segundos.
fim: Fim do trecho em segundos.
texto: Texto do trecho encontrado.
falante: Falante do trecho, quando conhecido.
pontuacao: Nota da fusão; maior é mais relevante.
origem: Como o trecho foi encontrado — "lexical", "semantica" ou
"ambas", para que o resultado seja explicável.
falas: Ids das falas que compõem o trecho, para descer ao corte.
"""
video_id: str
clipe_id: str
inicio: float
fim: float
texto: str
falante: str | None
pontuacao: float
origem: str
falas: tuple[int, ...]
class BuscaDeConteudo:
"""
Busca falas e enunciados por texto exato, por sentido, ou pelos dois.
A busca semântica é opcional: sem um provider de embeddings a classe
continua servindo a busca lexical, em vez de falhar. Isso mantém o painel
utilizável quando o Ollama não está rodando.
Atributos:
conexao: Conexão SQLite já aberta e com o esquema aplicado.
provider: Provider de embeddings, ou None para busca só lexical.
"""
def __init__(
self,
conexao: sqlite3.Connection,
provider: ProviderDeEmbeddings | None = None,
) -> None:
"""
Inicializa a busca sobre uma conexão e um provider opcional.
Parâmetros:
conexao: Conexão SQLite já aberta e com o esquema aplicado.
provider: Provider de embeddings para a busca semântica. Quando
``None``, apenas a busca lexical fica disponível.
"""
self.conexao = conexao
self.provider = provider
def buscar(
self, consulta: str, video_id: str | None = None, limite: int = 10,
) -> list[ResultadoDeBusca]:
"""
Busca um texto combinando as vias lexical e semântica.
Parâmetros:
consulta: Texto livre a procurar.
video_id: Restringe a busca a um vídeo. Quando ``None``, procura
em todo o acervo.
limite: Máximo de resultados devolvidos.
Retorna:
Os trechos mais relevantes, do melhor para o pior.
Pode gerar:
ErroDeBusca: quando a consulta é vazia ou o limite não é positivo.
"""
self._validar(consulta, limite)
lexicais = self.buscar_lexical(consulta, video_id, limite * 2)
semanticos = self.buscar_semantica(consulta, video_id, limite * 2)
return self._fundir(lexicais, semanticos, limite)
def buscar_lexical(
self, consulta: str, video_id: str | None = None, limite: int = 10,
) -> list[ResultadoDeBusca]:
"""
Busca a palavra exata nas falas, via índice FTS5.
Parâmetros:
consulta: Texto a procurar. Acentos são ignorados na comparação.
video_id: Restringe a busca a um vídeo.
limite: Máximo de resultados devolvidos.
Retorna:
As falas que contêm os termos, da mais relevante para a menos.
Pode gerar:
ErroDeBusca: quando a consulta é vazia ou o limite não é positivo.
"""
self._validar(consulta, limite)
filtro = "AND s.video_id = ?" if video_id else ""
parametros: list[object] = [self._consulta_fts(consulta)]
if video_id:
parametros.append(video_id)
parametros.append(limite)
try:
linhas = self.conexao.execute(
f"""SELECT s.id, s.video_id, s.clipe_id, s.inicio, s.fim, s.texto, s.falante,
bm25(busca_de_falas) AS nota
FROM busca_de_falas
JOIN segmentos_de_transcricao s ON s.id = busca_de_falas.rowid
WHERE busca_de_falas MATCH ? {filtro}
ORDER BY nota
LIMIT ?""",
parametros,
).fetchall()
except sqlite3.OperationalError as erro:
raise ErroDeBusca(f"Consulta lexical inválida: {erro}") from erro
return [
ResultadoDeBusca(
video_id=linha["video_id"], clipe_id=linha["clipe_id"],
inicio=linha["inicio"], fim=linha["fim"], texto=linha["texto"],
falante=linha["falante"], pontuacao=-float(linha["nota"]),
origem="lexical", falas=(int(linha["id"]),),
)
for linha in linhas
]
def buscar_semantica(
self, consulta: str, video_id: str | None = None, limite: int = 10,
) -> list[ResultadoDeBusca]:
"""
Busca por sentido nos enunciados, comparando embeddings.
Percorre os vetores em memória: no volume deste acervo (milhares de
enunciados) isso custa milissegundos, e evita depender de uma extensão
de banco vetorial. Se o acervo crescer uma ordem de grandeza, o ponto
de troca por um índice aproximado é aqui, sem mexer no esquema.
Parâmetros:
consulta: Texto a procurar por sentido.
video_id: Restringe a busca a um vídeo.
limite: Máximo de resultados devolvidos.
Retorna:
Os enunciados mais próximos da consulta. Lista vazia quando não há
provider configurado ou nenhum enunciado foi embedado.
Pode gerar:
ErroDeBusca: quando a consulta é vazia ou o limite não é positivo.
ErroDeEmbedding: quando o provider existe mas falha ao responder.
"""
self._validar(consulta, limite)
if self.provider is None:
return []
vetor_da_consulta = self.provider.gerar_para_consulta(consulta)
filtro = "WHERE e.video_id = ?" if video_id else ""
parametros = (video_id,) if video_id else ()
candidatos = []
for linha in self.conexao.execute(
f"""SELECT e.id, e.video_id, e.clipe_id, e.inicio, e.fim, e.texto, e.falante,
b.vetor
FROM enunciados e
JOIN embeddings_de_enunciado b ON b.enunciado_id = e.id
{filtro}""",
parametros,
):
similaridade = self._cosseno(vetor_da_consulta, desempacotar(linha["vetor"]))
candidatos.append((similaridade, linha))
candidatos.sort(key=lambda item: item[0], reverse=True)
return [
ResultadoDeBusca(
video_id=linha["video_id"], clipe_id=linha["clipe_id"],
inicio=linha["inicio"], fim=linha["fim"], texto=linha["texto"],
falante=linha["falante"], pontuacao=similaridade,
origem="semantica", falas=self._falas_do_enunciado(int(linha["id"])),
)
for similaridade, linha in candidatos[:limite]
]
def _falas_do_enunciado(self, enunciado_id: int) -> tuple[int, ...]:
"""Lê os ids das falas que compõem um enunciado, em ordem."""
return tuple(
int(linha["segmento_id"])
for linha in self.conexao.execute(
"SELECT segmento_id FROM falas_do_enunciado WHERE enunciado_id = ? ORDER BY ordem",
(enunciado_id,),
)
)
def _fundir(
self,
lexicais: list[ResultadoDeBusca],
semanticos: list[ResultadoDeBusca],
limite: int,
) -> list[ResultadoDeBusca]:
"""Funde as duas listas por posição (RRF), somando o peso de cada via."""
acumulado: dict[tuple[str, str, int], tuple[float, set[str], ResultadoDeBusca]] = {}
for lista in (lexicais, semanticos):
for posicao, resultado in enumerate(lista):
chave = (resultado.video_id, resultado.clipe_id, round(resultado.inicio * 100))
peso = 1.0 / (CONSTANTE_DE_FUSAO + posicao + 1)
if chave in acumulado:
nota, origens, guardado = acumulado[chave]
origens.add(resultado.origem)
acumulado[chave] = (nota + peso, origens, guardado)
else:
acumulado[chave] = (peso, {resultado.origem}, resultado)
ordenados = sorted(acumulado.values(), key=lambda item: item[0], reverse=True)
return [
ResultadoDeBusca(
video_id=guardado.video_id, clipe_id=guardado.clipe_id,
inicio=guardado.inicio, fim=guardado.fim, texto=guardado.texto,
falante=guardado.falante, pontuacao=round(nota, 6),
origem="ambas" if len(origens) > 1 else next(iter(origens)),
falas=guardado.falas,
)
for nota, origens, guardado in ordenados[:limite]
]
@staticmethod
def _consulta_fts(consulta: str) -> str:
"""
Monta a expressão do FTS5 a partir do texto digitado.
Cada palavra vira um termo com prefixo, e os termos são exigidos
juntos. Escapar em aspas evita que pontuação do usuário seja
interpretada como operador do FTS5 e derrube a consulta.
"""
termos = [palavra for palavra in consulta.replace('"', " ").split() if palavra]
return " AND ".join(f'"{termo}"*' for termo in termos)
@staticmethod
def _cosseno(primeiro: tuple[float, ...], segundo: tuple[float, ...]) -> float:
"""Similaridade de cosseno entre dois vetores, ou 0 se algum for nulo."""
if len(primeiro) != len(segundo):
return 0.0
produto = sum(a * b for a, b in zip(primeiro, segundo))
norma_primeiro = math.sqrt(sum(a * a for a in primeiro))
norma_segundo = math.sqrt(sum(b * b for b in segundo))
if norma_primeiro == 0.0 or norma_segundo == 0.0:
return 0.0
return produto / (norma_primeiro * norma_segundo)
@staticmethod
def _validar(consulta: str, limite: int) -> None:
"""Recusa consulta vazia ou limite não positivo antes de tocar o banco."""
if not isinstance(consulta, str) or not consulta.strip():
raise ErroDeBusca("A consulta não pode ser vazia.")
if limite <= 0:
raise ErroDeBusca("O limite precisa ser maior que zero.")
class IndexadorSemantico:
"""
Gera e grava os embeddings dos enunciados de um vídeo.
Atributos:
conexao: Conexão SQLite já aberta e com o esquema aplicado.
provider: Provider que transforma o texto de cada enunciado em vetor.
"""
def __init__(self, conexao: sqlite3.Connection, provider: ProviderDeEmbeddings) -> None:
"""
Inicializa o indexador com a conexão e o provider de embeddings.
Parâmetros:
conexao: Conexão SQLite já aberta e com o esquema aplicado.
provider: Provider que gera os vetores.
"""
self.conexao = conexao
self.provider = provider
def indexar(self, video_id: str, refazer: bool = False) -> int:
"""
Gera os embeddings faltantes dos enunciados de um vídeo.
Parâmetros:
video_id: Vídeo cujos enunciados serão indexados.
refazer: Quando verdadeiro, regenera também os que já têm vetor —
necessário ao trocar de modelo de embeddings.
Retorna:
A quantidade de enunciados indexados nesta execução.
Pode gerar:
ErroDeEmbedding: quando o provider falha ao gerar algum vetor.
"""
from ..integracoes.embeddings import empacotar
filtro = "" if refazer else "AND b.enunciado_id IS NULL"
pendentes = self.conexao.execute(
f"""SELECT e.id, e.texto FROM enunciados e
LEFT JOIN embeddings_de_enunciado b ON b.enunciado_id = e.id
WHERE e.video_id = ? {filtro}
ORDER BY e.inicio""",
(video_id,),
).fetchall()
indexados = 0
for linha in pendentes:
if not linha["texto"].strip():
continue
vetor = self.provider.gerar_para_documento(linha["texto"])
with self.conexao:
self.conexao.execute(
"""INSERT INTO embeddings_de_enunciado
(enunciado_id, modelo, dimensoes, vetor)
VALUES (?, ?, ?, ?)
ON CONFLICT (enunciado_id) DO UPDATE SET
modelo = excluded.modelo,
dimensoes = excluded.dimensoes,
vetor = excluded.vetor,
gerado_em = strftime('%Y-%m-%dT%H:%M:%fZ','now')""",
(linha["id"], self.provider.modelo, self.provider.dimensoes,
empacotar(vetor)),
)
indexados += 1
return indexados
__all__ = [
"BuscaDeConteudo",
"ErroDeBusca",
"ErroDeEmbedding",
"IndexadorSemantico",
"ResultadoDeBusca",
]
+246
View File
@@ -0,0 +1,246 @@
"""
Agrupamento de falas em enunciados embedáveis.
Uma fala do Whisper costuma ter de 3 a 8 segundos. Embedar um trecho tão curto
produz um vetor instável: pouco texto, muito ruído, e vizinhança semântica
pouco confiável. O enunciado resolve isso juntando falas consecutivas do mesmo
falante até atingir uma duração alvo, formando um bloco com contexto
suficiente para ter significado.
O vínculo com as falas de origem é sempre preservado. Isso é o que diferencia
esta busca de um RAG genérico: todo acerto semântico precisa voltar com o
timecode exato, senão não serve para cortar.
"""
from __future__ import annotations
import sqlite3
from dataclasses import dataclass
DURACAO_ALVO_PADRAO = 30.0
DURACAO_MAXIMA_PADRAO = 45.0
INTERVALO_QUE_QUEBRA_BLOCO = 2.0
class ErroDeAgrupamento(ValueError):
"""Parâmetros de agrupamento inválidos."""
@dataclass(frozen=True)
class FalaParaAgrupar:
"""Uma fala já persistida, no mínimo necessário para agrupá-la."""
segmento_id: int
clipe_id: str
inicio: float
fim: float
texto: str
falante: str | None
@dataclass(frozen=True)
class Enunciado:
"""
Um bloco de falas consecutivas tratado como unidade de busca semântica.
Atributos:
clipe_id: Clipe a que o bloco pertence.
inicio: Início do bloco, herdado da primeira fala.
fim: Fim do bloco, herdado da última fala.
texto: Texto das falas concatenado.
falante: Falante do bloco, quando todas as falas são do mesmo.
segmentos: Ids das falas que compõem o bloco, em ordem.
"""
clipe_id: str
inicio: float
fim: float
texto: str
falante: str | None
segmentos: tuple[int, ...]
@property
def duracao(self) -> float:
"""Duração do bloco em segundos."""
return self.fim - self.inicio
class AgrupadorDeEnunciados:
"""
Junta falas consecutivas em blocos de duração próxima a um alvo.
Um bloco é fechado quando atingir a duração alvo, quando o falante mudar,
quando houver um silêncio longo entre duas falas ou quando incluir a
próxima fala ultrapassaria a duração máxima. A troca de falante e o
silêncio longo são fronteiras naturais de assunto: agrupar através delas
misturaria ideias distintas no mesmo vetor.
Atributos:
duracao_alvo: Duração a partir da qual o bloco pode ser fechado.
duracao_maxima: Duração que o bloco não deve ultrapassar.
intervalo_que_quebra: Silêncio entre falas que força um bloco novo.
"""
def __init__(
self,
duracao_alvo: float = DURACAO_ALVO_PADRAO,
duracao_maxima: float = DURACAO_MAXIMA_PADRAO,
intervalo_que_quebra: float = INTERVALO_QUE_QUEBRA_BLOCO,
) -> None:
"""
Inicializa o agrupador com os limites de duração dos blocos.
Parâmetros:
duracao_alvo: Duração a partir da qual o bloco pode ser fechado.
duracao_maxima: Duração que o bloco não deve ultrapassar.
intervalo_que_quebra: Silêncio entre falas que força bloco novo.
Pode gerar:
ErroDeAgrupamento: quando as durações não são positivas ou a
máxima é menor que a alvo.
"""
if duracao_alvo <= 0 or duracao_maxima <= 0:
raise ErroDeAgrupamento("As durações de agrupamento devem ser positivas.")
if duracao_maxima < duracao_alvo:
raise ErroDeAgrupamento(
"A duração máxima não pode ser menor que a duração alvo."
)
self.duracao_alvo = duracao_alvo
self.duracao_maxima = duracao_maxima
self.intervalo_que_quebra = intervalo_que_quebra
def agrupar(self, falas: list[FalaParaAgrupar]) -> list[Enunciado]:
"""
Agrupa falas ordenadas por tempo em enunciados.
Parâmetros:
falas: Falas a agrupar. São ordenadas por clipe e início antes do
agrupamento, então a ordem de entrada não importa.
Retorna:
Os enunciados formados, em ordem de tempo.
"""
ordenadas = sorted(falas, key=lambda fala: (fala.clipe_id, fala.inicio))
enunciados: list[Enunciado] = []
bloco: list[FalaParaAgrupar] = []
for fala in ordenadas:
if bloco and self._deve_fechar(bloco, fala):
enunciados.append(self._montar(bloco))
bloco = []
bloco.append(fala)
if self._duracao(bloco) >= self.duracao_alvo:
enunciados.append(self._montar(bloco))
bloco = []
if bloco:
enunciados.append(self._montar(bloco))
return enunciados
def _deve_fechar(self, bloco: list[FalaParaAgrupar], proxima: FalaParaAgrupar) -> bool:
"""Decide se a próxima fala pertence a um bloco novo."""
ultima = bloco[-1]
if proxima.clipe_id != ultima.clipe_id:
return True
if proxima.falante != ultima.falante:
return True
if proxima.fim - bloco[0].inicio > self.duracao_maxima:
return True
# O silêncio só encerra o bloco depois que ele já tem corpo. Numa fala
# pausada, quebrar no primeiro intervalo longo produziria blocos de
# poucos segundos — curtos demais para gerar um embedding estável, que
# é justamente o problema que o agrupamento existe para resolver.
if proxima.inicio - ultima.fim < self.intervalo_que_quebra:
return False
return self._duracao(bloco) >= self.duracao_alvo / 2
@staticmethod
def _duracao(bloco: list[FalaParaAgrupar]) -> float:
"""Duração coberta por um bloco em formação."""
return bloco[-1].fim - bloco[0].inicio
@staticmethod
def _montar(bloco: list[FalaParaAgrupar]) -> Enunciado:
"""Monta o enunciado imutável a partir das falas acumuladas."""
falantes = {fala.falante for fala in bloco}
return Enunciado(
clipe_id=bloco[0].clipe_id,
inicio=bloco[0].inicio,
fim=bloco[-1].fim,
texto=" ".join(fala.texto.strip() for fala in bloco if fala.texto.strip()),
falante=bloco[0].falante if len(falantes) == 1 else None,
segmentos=tuple(fala.segmento_id for fala in bloco),
)
class RepositorioDeEnunciados:
"""
Lê falas e grava enunciados no banco de análises.
Atributos:
conexao: Conexão SQLite já aberta e com o esquema aplicado.
"""
def __init__(self, conexao: sqlite3.Connection) -> None:
"""
Inicializa o repositório sobre uma conexão existente.
Parâmetros:
conexao: Conexão SQLite já aberta e com o esquema aplicado.
"""
self.conexao = conexao
def carregar_falas(self, video_id: str) -> list[FalaParaAgrupar]:
"""
Carrega as falas de um vídeo no formato aceito pelo agrupador.
Parâmetros:
video_id: Identificador do vídeo cujas falas serão lidas.
Retorna:
As falas do vídeo, ordenadas por clipe e início.
"""
return [
FalaParaAgrupar(
segmento_id=linha["id"], clipe_id=linha["clipe_id"],
inicio=linha["inicio"], fim=linha["fim"],
texto=linha["texto"], falante=linha["falante"],
)
for linha in self.conexao.execute(
"""SELECT id, clipe_id, inicio, fim, texto, falante
FROM segmentos_de_transcricao
WHERE video_id = ? ORDER BY clipe_id, inicio""",
(video_id,),
)
]
def substituir_enunciados(self, video_id: str, enunciados: list[Enunciado]) -> int:
"""
Regrava os enunciados de um vídeo, apagando os anteriores.
Os embeddings são removidos junto pelo ``ON DELETE CASCADE``: um
enunciado com fronteiras novas não pode herdar o vetor do antigo.
Parâmetros:
video_id: Identificador do vídeo dono dos enunciados.
enunciados: Enunciados a gravar.
Retorna:
A quantidade de enunciados gravados.
"""
with self.conexao:
self.conexao.execute("DELETE FROM enunciados WHERE video_id = ?", (video_id,))
for enunciado in enunciados:
cursor = self.conexao.execute(
"""INSERT INTO enunciados
(video_id, clipe_id, inicio, fim, texto, falante, total_de_falas)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(video_id, enunciado.clipe_id, enunciado.inicio, enunciado.fim,
enunciado.texto, enunciado.falante, len(enunciado.segmentos)),
)
self.conexao.executemany(
"""INSERT INTO falas_do_enunciado (enunciado_id, segmento_id, ordem)
VALUES (?, ?, ?)""",
[(cursor.lastrowid, segmento_id, ordem)
for ordem, segmento_id in enumerate(enunciado.segmentos)],
)
return len(enunciados)
+334 -3
View File
@@ -82,10 +82,22 @@ CREATE TABLE IF NOT EXISTS segmentos_de_transcricao (
confianca_voz REAL,
emocao TEXT,
confianca_emocao REAL,
caracteristicas_acusticas TEXT
caracteristicas_acusticas TEXT,
energia_rms REAL,
pitch_mediano_hz REAL,
pitch_desvio_hz REAL,
velocidade_de_fala_pps REAL,
maior_pausa_interna_s REAL,
intervalo_anterior_s REAL
);
CREATE INDEX IF NOT EXISTS idx_segmentos_video_clipe
ON segmentos_de_transcricao(video_id, clipe_id);
-- Consulta temporal: "quais falas entre X e Y segundos". É o acesso mais
-- usado pelo agente de edição, por isso tem índice próprio.
CREATE INDEX IF NOT EXISTS idx_segmentos_intervalo
ON segmentos_de_transcricao(video_id, inicio, fim);
CREATE INDEX IF NOT EXISTS idx_segmentos_falante
ON segmentos_de_transcricao(video_id, falante);
CREATE TABLE IF NOT EXISTS palavras_de_transcricao (
id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -131,7 +143,8 @@ CREATE TABLE IF NOT EXISTS grupos_de_retake (
faixa_id TEXT NOT NULL,
tipo TEXT NOT NULL,
confianca REAL NOT NULL,
criado_em TEXT NOT NULL
criado_em TEXT NOT NULL,
status_de_revisao TEXT NOT NULL DEFAULT 'pendente'
);
CREATE INDEX IF NOT EXISTS idx_grupos_video ON grupos_de_retake(video_id);
@@ -157,10 +170,328 @@ CREATE TABLE IF NOT EXISTS evidencias_de_retake (
valor REAL
);
CREATE INDEX IF NOT EXISTS idx_evidencias_retake_grupo ON evidencias_de_retake(grupo_id);
-- ---------------------------------------------------------------------------
-- Busca lexical (FTS5)
-- ---------------------------------------------------------------------------
-- Índice de texto completo sobre as falas. Usa ``content=`` (external content)
-- para não duplicar o texto: o FTS5 guarda só o índice invertido e lê o texto
-- da tabela original pelo rowid. ``remove_diacritics 2`` faz "elegancia"
-- encontrar "elegância", que é o comportamento esperado em pt-BR.
CREATE VIRTUAL TABLE IF NOT EXISTS busca_de_falas USING fts5(
texto,
content='segmentos_de_transcricao',
content_rowid='id',
tokenize="unicode61 remove_diacritics 2"
);
CREATE TRIGGER IF NOT EXISTS trg_busca_de_falas_inserir
AFTER INSERT ON segmentos_de_transcricao BEGIN
INSERT INTO busca_de_falas(rowid, texto) VALUES (new.id, new.texto);
END;
CREATE TRIGGER IF NOT EXISTS trg_busca_de_falas_remover
AFTER DELETE ON segmentos_de_transcricao BEGIN
INSERT INTO busca_de_falas(busca_de_falas, rowid, texto)
VALUES ('delete', old.id, old.texto);
END;
CREATE TRIGGER IF NOT EXISTS trg_busca_de_falas_atualizar
AFTER UPDATE OF texto ON segmentos_de_transcricao BEGIN
INSERT INTO busca_de_falas(busca_de_falas, rowid, texto)
VALUES ('delete', old.id, old.texto);
INSERT INTO busca_de_falas(rowid, texto) VALUES (new.id, new.texto);
END;
-- ---------------------------------------------------------------------------
-- Busca semântica (enunciados + embeddings)
-- ---------------------------------------------------------------------------
-- Uma fala do Whisper tem 3 a 8 segundos: curta demais para gerar um embedding
-- com significado estável. O enunciado agrupa falas vizinhas do mesmo falante
-- num bloco de dezenas de segundos, que é a unidade embedável. O vínculo com
-- as falas de origem é preservado em ``falas_do_enunciado`` para que todo
-- acerto semântico volte com timecode utilizável para um corte.
CREATE TABLE IF NOT EXISTS enunciados (
id INTEGER PRIMARY KEY AUTOINCREMENT,
video_id TEXT NOT NULL REFERENCES videos(id) ON DELETE CASCADE,
clipe_id TEXT NOT NULL,
inicio REAL NOT NULL,
fim REAL NOT NULL,
texto TEXT NOT NULL,
falante TEXT,
total_de_falas INTEGER NOT NULL,
criado_em TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
);
CREATE INDEX IF NOT EXISTS idx_enunciados_video ON enunciados(video_id, inicio);
CREATE TABLE IF NOT EXISTS falas_do_enunciado (
enunciado_id INTEGER NOT NULL REFERENCES enunciados(id) ON DELETE CASCADE,
segmento_id INTEGER NOT NULL REFERENCES segmentos_de_transcricao(id) ON DELETE CASCADE,
ordem INTEGER NOT NULL,
PRIMARY KEY (enunciado_id, segmento_id)
);
CREATE INDEX IF NOT EXISTS idx_falas_do_enunciado_segmento
ON falas_do_enunciado(segmento_id);
CREATE TABLE IF NOT EXISTS embeddings_de_enunciado (
enunciado_id INTEGER PRIMARY KEY REFERENCES enunciados(id) ON DELETE CASCADE,
modelo TEXT NOT NULL,
dimensoes INTEGER NOT NULL,
vetor BLOB NOT NULL,
gerado_em TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
);
-- ---------------------------------------------------------------------------
-- Planos de edição
-- ---------------------------------------------------------------------------
-- O plano que a IA devolve é a única evidência de como uma edição foi
-- decidida, e antes disto ele vivia num arquivo temporário que sumia depois de
-- aplicado. Guardá-lo é o que permite comparar o que foi proposto com o que o
-- editor manteve — o único sinal disponível para aprender o estilo de corte.
CREATE TABLE IF NOT EXISTS planos_de_edicao (
id INTEGER PRIMARY KEY AUTOINCREMENT,
video_id TEXT REFERENCES videos(id) ON DELETE CASCADE,
origem TEXT NOT NULL,
tipo_de_video TEXT,
modelo_da_ia TEXT,
intencao TEXT,
criado_em TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
);
CREATE INDEX IF NOT EXISTS idx_planos_video ON planos_de_edicao(video_id, criado_em);
CREATE TABLE IF NOT EXISTS acoes_do_plano (
id INTEGER PRIMARY KEY AUTOINCREMENT,
plano_id INTEGER NOT NULL REFERENCES planos_de_edicao(id) ON DELETE CASCADE,
ordem INTEGER NOT NULL,
tipo TEXT NOT NULL,
inicio REAL,
fim REAL,
motivo TEXT,
-- Cada tipo de ação tem parâmetros próprios e incompatíveis entre si (um
-- zoom tem escala, um texto tem conteúdo e posição). É carga polimórfica
-- de verdade, consumida inteira por quem aplica: por isso fica em JSON.
parametros TEXT
);
CREATE INDEX IF NOT EXISTS idx_acoes_do_plano ON acoes_do_plano(plano_id, ordem);
CREATE TABLE IF NOT EXISTS aplicacoes_do_plano (
id INTEGER PRIMARY KEY AUTOINCREMENT,
plano_id INTEGER NOT NULL REFERENCES planos_de_edicao(id) ON DELETE CASCADE,
sequencia TEXT,
aplicado_em TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
sucesso INTEGER NOT NULL,
acoes_aplicadas INTEGER NOT NULL DEFAULT 0,
mensagem TEXT
);
CREATE INDEX IF NOT EXISTS idx_aplicacoes_plano ON aplicacoes_do_plano(plano_id);
-- ---------------------------------------------------------------------------
-- Leituras achatadas
-- ---------------------------------------------------------------------------
-- As tabelas acima guardam cada fato uma única vez, no nível em que ele é
-- verdadeiro: a emoção e o falante valem para a frase, os tempos exatos valem
-- para a palavra. As views abaixo reapresentam esses mesmos dados já juntos,
-- para que o agente de edição leia tudo numa consulta só sem que o banco
-- precise repetir valor em disco. Normalizado para gravar, achatado para ler.
-- Uma linha por fala, com as três análises de áudio e o resumo das palavras.
CREATE VIEW IF NOT EXISTS vw_falas_completas AS
SELECT
s.id AS fala_id,
s.video_id,
s.clipe_id,
s.inicio,
s.fim,
s.fim - s.inicio AS duracao,
s.texto,
s.falante,
s.emocao,
s.confianca_emocao,
s.energia_rms,
s.pitch_mediano_hz,
s.pitch_desvio_hz,
s.velocidade_de_fala_pps,
s.maior_pausa_interna_s,
s.intervalo_anterior_s,
count(p.id) AS total_de_palavras
FROM segmentos_de_transcricao s
LEFT JOIN palavras_de_transcricao p ON p.segmento_id = s.id
GROUP BY s.id;
-- Uma linha por palavra, repetindo o que vale para a frase inteira. É a forma
-- mais plana possível de ler a transcrição — sem custo de duplicação em disco,
-- porque a repetição acontece na leitura e não na gravação.
CREATE VIEW IF NOT EXISTS vw_palavras_completas AS
SELECT
p.id AS palavra_id,
p.segmento_id AS fala_id,
s.video_id,
s.clipe_id,
p.ordem,
p.texto AS palavra,
p.inicio AS palavra_inicio,
p.fim AS palavra_fim,
p.confianca AS palavra_confianca,
s.texto AS frase,
s.inicio AS frase_inicio,
s.fim AS frase_fim,
s.falante,
s.emocao,
s.confianca_emocao,
s.energia_rms,
s.pitch_mediano_hz,
s.velocidade_de_fala_pps
FROM palavras_de_transcricao p
JOIN segmentos_de_transcricao s ON s.id = p.segmento_id;
-- Linha do tempo unificada: fala e imagem no mesmo eixo, para percorrer o
-- vídeo em ordem cronológica lendo áudio e imagem juntos. A geometria bruta
-- (landmarks, bounding boxes) fica de fora de propósito: o que decide corte é
-- o fato editorial, e a geometria continua disponível em evidencias_visuais
-- para quem precisar dela.
CREATE VIEW IF NOT EXISTS vw_linha_do_tempo AS
SELECT video_id, clipe_id, inicio, fim, 'fala' AS tipo,
texto AS descricao, falante AS detalhe, confianca_emocao AS confianca
FROM segmentos_de_transcricao
UNION ALL
SELECT video_id, clipe_id, inicio, fim, 'cena' AS tipo,
'mudança de cena' AS descricao, NULL AS detalhe, confianca
FROM cenas
UNION ALL
SELECT video_id, clipe_id, inicio, fim, tipo,
json_extract(valor, '$.identificador') AS descricao,
NULL AS detalhe, confianca
FROM evidencias_visuais
WHERE tipo = 'categoria'
UNION ALL
SELECT video_id, clipe_id, inicio, fim, tipo,
json_extract(valor, '$.texto') AS descricao,
NULL AS detalhe, confianca
FROM evidencias_visuais
WHERE tipo = 'interpretacao_editorial';
-- Cada fala com o que estava em quadro enquanto ela era dita. O vínculo é por
-- sobreposição de tempo, não por chave estrangeira: fala e imagem são medidas
-- em grades diferentes (a fala é um intervalo, a imagem é amostrada por
-- quadro) e nunca coincidem exatamente. As categorias visuais entram
-- agregadas, e não uma linha por amostra, para a resposta caber num prompt.
CREATE VIEW IF NOT EXISTS vw_falas_com_visual AS
SELECT
f.fala_id,
f.video_id,
f.clipe_id,
f.inicio,
f.fim,
f.texto,
f.falante,
f.emocao,
f.energia_rms,
f.velocidade_de_fala_pps,
(SELECT group_concat(DISTINCT json_extract(e.valor, '$.identificador'))
FROM evidencias_visuais e
WHERE e.video_id = f.video_id
AND e.tipo = 'categoria' AND e.confianca >= 0.5
AND e.inicio BETWEEN f.inicio AND f.fim) AS categorias_em_quadro,
(SELECT count(*)
FROM evidencias_visuais e
WHERE e.video_id = f.video_id
AND e.tipo = 'rosto'
AND e.inicio BETWEEN f.inicio AND f.fim) AS amostras_com_rosto,
(SELECT round(avg(json_extract(e.valor, '$.score_global')), 3)
FROM evidencias_visuais e
WHERE e.video_id = f.video_id
AND e.tipo = 'estetica'
AND e.inicio BETWEEN f.inicio AND f.fim) AS estetica_media,
(SELECT count(*)
FROM cenas c
WHERE c.video_id = f.video_id
AND c.inicio > f.inicio AND c.inicio < f.fim) AS cortes_de_cena_dentro
FROM vw_falas_completas f;
"""
# Colunas acrescentadas depois da primeira versão do esquema. ``CREATE TABLE IF
# NOT EXISTS`` não altera tabela existente, então cada uma é aplicada por
# ``_migrar_colunas`` em bancos que já foram criados.
# Views recriadas a cada abertura do banco. "CREATE VIEW IF NOT EXISTS" não
# atualiza a definição de uma view já existente, então uma correção na
# consulta de uma view só chegaria a bancos novos sem isso — o banco de
# produção ficaria preso para sempre na primeira versão que foi criada nele.
_VIEWS = (
"vw_falas_completas",
"vw_palavras_completas",
"vw_linha_do_tempo",
"vw_falas_com_visual",
)
_COLUNAS_ACRESCENTADAS: tuple[tuple[str, str, str], ...] = (
("segmentos_de_transcricao", "energia_rms", "REAL"),
("segmentos_de_transcricao", "pitch_mediano_hz", "REAL"),
("segmentos_de_transcricao", "pitch_desvio_hz", "REAL"),
("segmentos_de_transcricao", "velocidade_de_fala_pps", "REAL"),
("segmentos_de_transcricao", "maior_pausa_interna_s", "REAL"),
("segmentos_de_transcricao", "intervalo_anterior_s", "REAL"),
("grupos_de_retake", "status_de_revisao", "TEXT NOT NULL DEFAULT 'pendente'"),
)
class ErroDeEsquema(RuntimeError):
"""Falha ao criar ou migrar o esquema do banco de análises."""
def _colunas_existentes(conexao: sqlite3.Connection, tabela: str) -> set[str]:
"""Lê os nomes de coluna já presentes numa tabela."""
return {linha[1] for linha in conexao.execute(f"PRAGMA table_info({tabela})")}
def _migrar_colunas(conexao: sqlite3.Connection) -> None:
"""Acrescenta colunas novas em bancos criados por versões anteriores."""
for tabela, coluna, tipo in _COLUNAS_ACRESCENTADAS:
if not _colunas_existentes(conexao, tabela):
continue
if coluna in _colunas_existentes(conexao, tabela):
continue
conexao.execute(f"ALTER TABLE {tabela} ADD COLUMN {coluna} {tipo}")
def reconstruir_indice_de_busca(conexao: sqlite3.Connection) -> int:
"""
Reindexa do zero a busca lexical a partir das falas já persistidas.
Necessário em bancos criados antes de ``busca_de_falas`` existir: os
gatilhos só cobrem escritas futuras, então as falas antigas precisam ser
carregadas uma vez.
Parâmetros:
conexao: Conexão SQLite já aberta e com o esquema aplicado.
Retorna:
A quantidade de falas presentes no índice depois da reconstrução.
"""
with conexao:
conexao.execute("INSERT INTO busca_de_falas(busca_de_falas) VALUES ('rebuild')")
return int(conexao.execute("SELECT count(*) FROM busca_de_falas").fetchone()[0])
def criar_esquema(conexao: sqlite3.Connection) -> None:
"""Cria (de forma idempotente) todas as tabelas do banco de análises."""
"""
Cria (de forma idempotente) todas as tabelas do banco de análises.
Aplica também as migrações de coluna necessárias em bancos criados por
versões anteriores do esquema, para que abrir um banco antigo baste para
deixá-lo no formato atual.
Parâmetros:
conexao: Conexão SQLite aberta onde o esquema será aplicado.
Pode gerar:
ErroDeEsquema: quando o SQLite recusa a DDL — tipicamente por falta da
extensão FTS5 na build em uso.
"""
try:
_migrar_colunas(conexao)
for view in _VIEWS:
conexao.execute(f"DROP VIEW IF EXISTS {view}")
conexao.executescript(_DDL)
except sqlite3.OperationalError as erro:
raise ErroDeEsquema(f"Não foi possível aplicar o esquema de análises: {erro}") from erro
conexao.commit()
+282
View File
@@ -0,0 +1,282 @@
"""
Ingestão no banco de análises do JSON produzido pelo pipeline de voz.
O pipeline de voz (transcrição, diarização e métricas de fala) roda hoje fora
deste repositório e entrega um ``dados-para-ia.json`` cujo ``segments`` já é
uma tabela plana: cada item traz o texto e os tempos da frase, o falante e os
seis escalares de métrica lado a lado. Este módulo trata esse arquivo como
formato de entrada — nunca como armazenamento — e o converte em linhas do
banco de análises.
A leitura e a gravação ficam em classes separadas: ``LeitorDeDadosParaIA`` não
conhece banco algum e ``IngestorDeVozNoBanco`` não conhece o formato do
arquivo. Assim uma mudança no JSON de origem não alcança a persistência, e
outra origem de transcrição pode reaproveitar a gravação.
"""
from __future__ import annotations
import json
import sqlite3
from dataclasses import dataclass, field
from pathlib import Path
from typing import Sequence
# Tradução entre as chaves de métrica do pipeline de voz e as colunas do banco.
# São seis escalares de conjunto fechado: por isso viram coluna, e não um JSON
# opaco que não se consegue filtrar em SQL.
COLUNAS_DE_METRICA: dict[str, str] = {
"energy_rms": "energia_rms",
"pitch_hz_median": "pitch_mediano_hz",
"pitch_hz_std": "pitch_desvio_hz",
"speaking_rate_wps": "velocidade_de_fala_pps",
"longest_internal_pause_s": "maior_pausa_interna_s",
"gap_before_s": "intervalo_anterior_s",
}
class ErroDeIngestaoDeVoz(ValueError):
"""Arquivo de voz ausente, ilegível ou fora do formato esperado."""
@dataclass(frozen=True)
class PalavraDoPipelineDeVoz:
"""Uma palavra transcrita, com os tempos que permitem cortar sem picotá-la."""
texto: str
inicio: float
fim: float
confianca: float | None = None
@dataclass(frozen=True)
class FalaDoPipelineDeVoz:
"""
Uma frase transcrita com as três análises de áudio reunidas.
Reúne numa só unidade o que o pipeline produz em etapas diferentes sobre o
mesmo intervalo de tempo: o texto (transcrição), quem falou (diarização) e
como falou (métricas). São descrições do mesmo trecho, e por isso ocupam a
mesma linha do banco.
Atributos:
inicio: Início da frase em segundos, relativo à mídia de origem.
fim: Fim da frase em segundos, relativo à mídia de origem.
texto: Texto transcrito da frase.
falante: Identificador do falante atribuído pela diarização.
metricas: Escalares de métrica de fala, já nas chaves do pipeline.
palavras: Palavras da frase, em ordem.
"""
inicio: float
fim: float
texto: str
falante: str | None = None
metricas: dict[str, float | None] = field(default_factory=dict)
palavras: tuple[PalavraDoPipelineDeVoz, ...] = ()
@property
def duracao(self) -> float:
"""Duração da frase em segundos."""
return self.fim - self.inicio
@dataclass(frozen=True)
class ResultadoDaIngestao:
"""Contagem do que foi gravado, para o chamador relatar sem reconsultar."""
falas: int
palavras: int
falantes: int
class LeitorDeDadosParaIA:
"""
Converte um ``dados-para-ia.json`` em falas do pipeline de voz.
Não acessa banco de dados nem filesystem além da leitura do arquivo
indicado, e não decide nada de editorial: apenas normaliza o formato.
"""
def ler(self, caminho: str | Path) -> tuple[FalaDoPipelineDeVoz, ...]:
"""
Lê o arquivo e devolve as falas nele contidas, em ordem de tempo.
Parâmetros:
caminho: Caminho do ``dados-para-ia.json`` a carregar.
Retorna:
As falas do arquivo, ordenadas pelo início.
Pode gerar:
ErroDeIngestaoDeVoz: quando o arquivo não existe, não é JSON
válido, não traz ``segments`` ou traz um segmento sem os
tempos obrigatórios.
"""
caminho = Path(caminho)
if not caminho.is_file():
raise ErroDeIngestaoDeVoz(f"Arquivo de voz não encontrado: {caminho}")
try:
dados = json.loads(caminho.read_text(encoding="utf-8"))
except json.JSONDecodeError as erro:
raise ErroDeIngestaoDeVoz(f"JSON inválido em {caminho}: {erro}") from erro
except OSError as erro:
raise ErroDeIngestaoDeVoz(f"Não foi possível ler {caminho}: {erro}") from erro
if not isinstance(dados, dict) or not isinstance(dados.get("segments"), list):
raise ErroDeIngestaoDeVoz(
f"{caminho} não tem a lista 'segments' esperada do pipeline de voz."
)
falas = [self._converter_fala(item, indice, caminho)
for indice, item in enumerate(dados["segments"])]
return tuple(sorted(falas, key=lambda fala: fala.inicio))
def _converter_fala(
self, item: object, indice: int, caminho: Path,
) -> FalaDoPipelineDeVoz:
"""Converte um item de ``segments`` numa fala validada."""
if not isinstance(item, dict):
raise ErroDeIngestaoDeVoz(f"Segmento {indice} de {caminho} não é um objeto.")
inicio = self._numero_obrigatorio(item, "start", indice, caminho)
fim = self._numero_obrigatorio(item, "end", indice, caminho)
if fim < inicio:
raise ErroDeIngestaoDeVoz(
f"Segmento {indice} de {caminho} termina ({fim}) antes de começar ({inicio})."
)
metricas = {chave: self._numero_opcional(item.get(chave))
for chave in COLUNAS_DE_METRICA}
return FalaDoPipelineDeVoz(
inicio=inicio,
fim=fim,
texto=str(item.get("text", "")).strip(),
falante=item.get("speaker") or None,
metricas=metricas,
palavras=self._converter_palavras(item.get("words")),
)
def _converter_palavras(self, bruto: object) -> tuple[PalavraDoPipelineDeVoz, ...]:
"""Converte a lista ``words`` de um segmento, ignorando itens malformados."""
if not isinstance(bruto, list):
return ()
palavras = []
for item in bruto:
if not isinstance(item, dict):
continue
inicio = self._numero_opcional(item.get("start"))
fim = self._numero_opcional(item.get("end"))
if inicio is None or fim is None:
continue
palavras.append(PalavraDoPipelineDeVoz(
texto=str(item.get("text", "")),
inicio=inicio,
fim=fim,
confianca=self._numero_opcional(item.get("confidence")),
))
return tuple(palavras)
@staticmethod
def _numero_obrigatorio(item: dict, chave: str, indice: int, caminho: Path) -> float:
"""Lê um número que precisa existir, com erro que diz qual campo faltou."""
valor = item.get(chave)
if not isinstance(valor, (int, float)) or isinstance(valor, bool):
raise ErroDeIngestaoDeVoz(
f"Segmento {indice} de {caminho} não tem o campo numérico '{chave}'."
)
return float(valor)
@staticmethod
def _numero_opcional(valor: object) -> float | None:
"""Lê um número que pode faltar, devolvendo None quando ausente."""
if isinstance(valor, bool) or not isinstance(valor, (int, float)):
return None
return float(valor)
class IngestorDeVozNoBanco:
"""
Grava falas do pipeline de voz no banco de análises.
Substitui integralmente as falas do clipe recebido, para que reprocessar a
mesma mídia seja idempotente sem apagar análises de outros clipes.
Atributos:
conexao: Conexão SQLite já aberta e com o esquema aplicado.
"""
def __init__(self, conexao: sqlite3.Connection) -> None:
"""
Inicializa o ingestor com a conexão onde as falas serão gravadas.
Parâmetros:
conexao: Conexão SQLite já aberta e com o esquema aplicado.
"""
self.conexao = conexao
def ingerir(
self, video_id: str, clipe_id: str, falas: Sequence[FalaDoPipelineDeVoz],
) -> ResultadoDaIngestao:
"""
Grava as falas do clipe, substituindo o que houver dele no banco.
Parâmetros:
video_id: Identificador do vídeo/timeline dono das falas.
clipe_id: Identificador do clipe a que as falas pertencem.
falas: Falas a gravar, já normalizadas pelo leitor.
Retorna:
As contagens de falas, palavras e falantes distintos gravados.
Pode gerar:
ErroDeIngestaoDeVoz: quando ``video_id`` ou ``clipe_id`` for vazio.
"""
self._validar_identificador("video_id", video_id)
self._validar_identificador("clipe_id", clipe_id)
colunas_de_metrica = tuple(COLUNAS_DE_METRICA.values())
insercao = (
"INSERT INTO segmentos_de_transcricao "
"(video_id, clipe_id, inicio, fim, texto, falante, "
+ ", ".join(colunas_de_metrica)
+ ") VALUES (?, ?, ?, ?, ?, ?, "
+ ", ".join("?" * len(colunas_de_metrica))
+ ")"
)
total_de_palavras = 0
with self.conexao:
self.conexao.execute("INSERT OR IGNORE INTO videos (id) VALUES (?)", (video_id,))
self.conexao.execute(
"DELETE FROM segmentos_de_transcricao WHERE video_id = ? AND clipe_id = ?",
(video_id, clipe_id),
)
for fala in falas:
cursor = self.conexao.execute(insercao, (
video_id, clipe_id, fala.inicio, fala.fim, fala.texto, fala.falante,
*(fala.metricas.get(chave) for chave in COLUNAS_DE_METRICA),
))
total_de_palavras += self._inserir_palavras(
int(cursor.lastrowid), fala,
)
return ResultadoDaIngestao(
falas=len(falas),
palavras=total_de_palavras,
falantes=len({fala.falante for fala in falas if fala.falante}),
)
def _inserir_palavras(self, segmento_id: int, fala: FalaDoPipelineDeVoz) -> int:
"""Grava as palavras de uma fala e devolve quantas foram gravadas."""
self.conexao.executemany(
"""INSERT INTO palavras_de_transcricao
(segmento_id, ordem, texto, inicio, fim, confianca, falante)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
[(segmento_id, ordem, palavra.texto, palavra.inicio, palavra.fim,
palavra.confianca, fala.falante)
for ordem, palavra in enumerate(fala.palavras)],
)
return len(fala.palavras)
@staticmethod
def _validar_identificador(nome: str, valor: str) -> None:
"""Recusa identificador vazio antes de escrever qualquer linha."""
if not isinstance(valor, str) or not valor.strip():
raise ErroDeIngestaoDeVoz(f"O parâmetro '{nome}' é obrigatório.")
+93
View File
@@ -0,0 +1,93 @@
"""
Remoção de linhas duplicadas deixadas por execuções não idempotentes.
Antes de ``substituir_evidencias_visuais`` e ``substituir_cenas`` existirem, o
Scanner acrescentava evidências e cenas a cada execução em vez de substituir
as do clipe. Bancos criados nesse período acumularam cópias exatas da mesma
observação, o que infla contagens e médias calculadas sobre essas tabelas.
Este módulo apaga essas cópias mantendo sempre a linha de menor ``id`` de cada
grupo idêntico. Só remove duplicata exata — linhas que diferem em qualquer
campo são observações distintas e são preservadas.
"""
from __future__ import annotations
import sqlite3
from dataclasses import dataclass
@dataclass(frozen=True)
class ResultadoDaLimpeza:
"""Quantidade de linhas removidas por tabela."""
evidencias_visuais: int
cenas: int
@property
def total(self) -> int:
"""Total de linhas removidas em todas as tabelas."""
return self.evidencias_visuais + self.cenas
class LimpadorDeDuplicatas:
"""
Remove duplicatas exatas de evidências visuais e cenas.
Atributos:
conexao: Conexão SQLite já aberta e com o esquema aplicado.
"""
def __init__(self, conexao: sqlite3.Connection) -> None:
"""
Inicializa o limpador sobre uma conexão existente.
Parâmetros:
conexao: Conexão SQLite já aberta e com o esquema aplicado.
"""
self.conexao = conexao
def contar_duplicatas(self) -> ResultadoDaLimpeza:
"""
Conta quantas linhas seriam removidas, sem remover nada.
Retorna:
As contagens de duplicatas por tabela.
"""
return ResultadoDaLimpeza(
evidencias_visuais=self._contar(
"evidencias_visuais", ("video_id", "clipe_id", "tipo", "inicio", "fim", "valor"),
),
cenas=self._contar("cenas", ("video_id", "clipe_id", "inicio", "fim")),
)
def limpar(self) -> ResultadoDaLimpeza:
"""
Remove as duplicatas, mantendo a linha de menor ``id`` de cada grupo.
Retorna:
As contagens de linhas efetivamente removidas por tabela.
"""
with self.conexao:
evidencias = self._remover(
"evidencias_visuais", ("video_id", "clipe_id", "tipo", "inicio", "fim", "valor"),
)
cenas = self._remover("cenas", ("video_id", "clipe_id", "inicio", "fim"))
return ResultadoDaLimpeza(evidencias_visuais=evidencias, cenas=cenas)
def _contar(self, tabela: str, chave: tuple[str, ...]) -> int:
"""Conta linhas que não são a primeira ocorrência do seu grupo."""
colunas = ", ".join(chave)
return int(self.conexao.execute(
f"""SELECT count(*) FROM {tabela}
WHERE id NOT IN (SELECT min(id) FROM {tabela} GROUP BY {colunas})"""
).fetchone()[0])
def _remover(self, tabela: str, chave: tuple[str, ...]) -> int:
"""Apaga as linhas que não são a primeira ocorrência do seu grupo."""
colunas = ", ".join(chave)
cursor = self.conexao.execute(
f"""DELETE FROM {tabela}
WHERE id NOT IN (SELECT min(id) FROM {tabela} GROUP BY {colunas})"""
)
return cursor.rowcount
@@ -18,6 +18,13 @@ from ..scanner.transcricao_da_timeline import TranscricaoDoClipe
from .conexao import abrir_banco
def _numero_ou_nulo(valor: object) -> float | None:
"""Converte um valor de métrica em float, ou None quando ausente/ilegível."""
if isinstance(valor, bool) or not isinstance(valor, (int, float)):
return None
return float(valor)
class RepositorioDeAnalisesSQLite:
"""Grava metadados de arquivo, transcrição e evidências/cenas visuais."""
@@ -30,6 +37,17 @@ class RepositorioDeAnalisesSQLite:
def registrar_metadados_de_arquivo(
self, video_id: str, metadados: MetadadosDoArquivo, hash_do_conteudo: str | None = None,
) -> None:
"""
Grava os metadados técnicos de um arquivo de mídia do vídeo.
Regravar o mesmo caminho atualiza a linha existente em vez de criar
outra, então o método é seguro para reprocessamento.
Parâmetros:
video_id: Identificador do vídeo dono do arquivo.
metadados: Metadados técnicos extraídos da mídia.
hash_do_conteudo: Hash do conteúdo do arquivo, quando calculado.
"""
with self.conexao:
self._garantir_video(video_id)
self.conexao.execute(
@@ -59,6 +77,16 @@ class RepositorioDeAnalisesSQLite:
def registrar_transcricoes(
self, video_id: str, transcricoes: Iterable[TranscricaoDoClipe],
) -> None:
"""
Acrescenta transcrições sem remover as já gravadas.
Para reprocessamento use ``substituir_transcricoes``: este método
acumula, e chamá-lo duas vezes para o mesmo clipe duplica as falas.
Parâmetros:
video_id: Identificador do vídeo dono das transcrições.
transcricoes: Transcrições por clipe a gravar.
"""
with self.conexao:
self._garantir_video(video_id)
for transcricao in transcricoes:
@@ -118,15 +146,31 @@ class RepositorioDeAnalisesSQLite:
def _inserir_segmento(
self, video_id: str, clipe_id: str, segmento: SegmentoDeTranscricao,
) -> int:
"""Grava uma fala e devolve o id gerado, para as palavras se ligarem a ela.
As métricas acústicas são gravadas duas vezes de propósito: em
``caracteristicas_acusticas`` (JSON completo, para reconstrução fiel)
e em colunas dedicadas (para filtrar e ordenar por elas em SQL sem
precisar abrir o JSON).
"""
metricas = segmento.caracteristicas_acusticas or {}
cursor = self.conexao.execute(
"""INSERT INTO segmentos_de_transcricao
(video_id, clipe_id, inicio, fim, texto, confianca, falante, voz_aparente,
confianca_voz, emocao, confianca_emocao, caracteristicas_acusticas)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
confianca_voz, emocao, confianca_emocao, caracteristicas_acusticas,
energia_rms, pitch_mediano_hz, pitch_desvio_hz, velocidade_de_fala_pps,
maior_pausa_interna_s, intervalo_anterior_s)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(video_id, clipe_id, segmento.inicio, segmento.fim, segmento.texto,
segmento.confianca, segmento.falante, segmento.voz_aparente,
segmento.confianca_voz, segmento.emocao, segmento.confianca_emocao,
json.dumps(segmento.caracteristicas_acusticas, ensure_ascii=False)),
json.dumps(segmento.caracteristicas_acusticas, ensure_ascii=False),
_numero_ou_nulo(metricas.get("energy_rms")),
_numero_ou_nulo(metricas.get("pitch_hz_median")),
_numero_ou_nulo(metricas.get("pitch_hz_std")),
_numero_ou_nulo(metricas.get("speaking_rate_wps")),
_numero_ou_nulo(metricas.get("longest_internal_pause_s")),
_numero_ou_nulo(metricas.get("gap_before_s"))),
)
segmento_id = cursor.lastrowid
for ordem, palavra in enumerate(segmento.palavras):
@@ -140,6 +184,16 @@ class RepositorioDeAnalisesSQLite:
return segmento_id
def carregar_transcricoes(self, video_id: str, clipe_id: str) -> list[SegmentoDeTranscricao]:
"""
Carrega as falas de um clipe, já com as palavras de cada uma.
Parâmetros:
video_id: Identificador do vídeo dono das falas.
clipe_id: Clipe cujas falas serão lidas.
Retorna:
As falas do clipe ordenadas por início, com as palavras em ordem.
"""
segmentos: list[SegmentoDeTranscricao] = []
for linha in self.conexao.execute(
"""SELECT * FROM segmentos_de_transcricao
@@ -166,9 +220,69 @@ class RepositorioDeAnalisesSQLite:
))
return segmentos
def substituir_evidencias_visuais(
self, video_id: str, clipe_id: str, evidencias: Iterable[EvidenciaVisual],
) -> int:
"""
Regrava as evidências visuais de um clipe, apagando as anteriores.
Torna o reprocessamento idempotente. ``registrar_evidencias_visuais``
só acrescenta, então reanalisar o mesmo clipe com ele acumula cópias
da mesma evidência e infla qualquer contagem feita sobre a tabela.
Parâmetros:
video_id: Identificador do vídeo dono das evidências.
clipe_id: Clipe cujas evidências serão substituídas.
evidencias: Evidências a gravar.
Retorna:
A quantidade de evidências gravadas.
"""
evidencias_materializadas = tuple(evidencias)
with self.conexao:
self._garantir_video(video_id)
self.conexao.execute(
"DELETE FROM evidencias_visuais WHERE video_id = ? AND clipe_id = ?",
(video_id, clipe_id),
)
self.registrar_evidencias_visuais(video_id, clipe_id, evidencias_materializadas)
return len(evidencias_materializadas)
def substituir_cenas(
self, video_id: str, cenas: Iterable[Cena], clipe_id: str | None = None,
) -> int:
"""
Regrava as cenas de um clipe, apagando as anteriores.
Parâmetros:
video_id: Identificador do vídeo dono das cenas.
cenas: Cenas a gravar.
clipe_id: Clipe cujas cenas serão substituídas. Quando ``None``,
substitui as cenas do vídeo que não pertencem a clipe algum.
Retorna:
A quantidade de cenas gravadas.
"""
cenas_materializadas = tuple(cenas)
with self.conexao:
self._garantir_video(video_id)
if clipe_id is None:
self.conexao.execute(
"DELETE FROM cenas WHERE video_id = ? AND clipe_id IS NULL", (video_id,))
else:
self.conexao.execute(
"DELETE FROM cenas WHERE video_id = ? AND clipe_id = ?", (video_id, clipe_id))
self.registrar_cenas(video_id, cenas_materializadas, clipe_id)
return len(cenas_materializadas)
def registrar_evidencias_visuais(
self, video_id: str, clipe_id: str, evidencias: Iterable[EvidenciaVisual],
) -> None:
"""Acrescenta evidências visuais sem remover as já gravadas.
Para reprocessamento use ``substituir_evidencias_visuais``: este método
acumula, e chamá-lo duas vezes para o mesmo clipe duplica as linhas.
"""
with self.conexao:
self._garantir_video(video_id)
for evidencia in evidencias:
@@ -184,6 +298,16 @@ class RepositorioDeAnalisesSQLite:
def registrar_cenas(
self, video_id: str, cenas: Iterable[Cena], clipe_id: str | None = None,
) -> None:
"""Acrescenta cenas sem remover as já gravadas.
Para reprocessamento use ``substituir_cenas``: este método acumula, e
chamá-lo duas vezes para o mesmo clipe duplica as cenas.
Parâmetros:
video_id: Identificador do vídeo dono das cenas.
cenas: Cenas detectadas a gravar.
clipe_id: Clipe a que as cenas pertencem, quando houver.
"""
with self.conexao:
self._garantir_video(video_id)
for cena in cenas:
@@ -0,0 +1,293 @@
"""
Persistência dos planos de edição devolvidos pela IA e de suas aplicações.
Antes deste módulo o plano era escrito num arquivo temporário, aplicado na
timeline e descartado. Com isso o sistema não guardava nenhum registro de como
uma edição foi decidida — nem o que a IA propôs, nem se a aplicação funcionou.
Guardar plano e aplicação separadamente é deliberado: o mesmo plano pode ser
aplicado mais de uma vez (em outra sequência, ou depois de um backup), e a
proposta continua sendo a mesma. Comparar o que foi proposto com o que
sobreviveu na timeline é o que dá material para melhorar as decisões futuras.
"""
from __future__ import annotations
import json
import sqlite3
from dataclasses import dataclass, field
from typing import Sequence
class ErroDePlano(ValueError):
"""Plano malformado ou parâmetros inválidos para gravá-lo."""
@dataclass(frozen=True)
class AcaoDoPlano:
"""
Uma operação proposta pela IA sobre a timeline.
Atributos:
tipo: Natureza da operação — corte, zoom, texto, marcador.
inicio: Início do trecho afetado, em segundos.
fim: Fim do trecho afetado, em segundos.
motivo: Justificativa dada pela IA, quando houver.
parametros: Campos específicos do tipo de ação.
"""
tipo: str
inicio: float | None = None
fim: float | None = None
motivo: str | None = None
parametros: dict[str, object] = field(default_factory=dict)
@dataclass(frozen=True)
class PlanoDeEdicao:
"""
Um plano completo, como a IA o devolveu.
Atributos:
origem: Mídia ou sequência a que o plano se refere.
acoes: Operações propostas, na ordem em que devem ser aplicadas.
video_id: Vídeo do banco correspondente, quando conhecido.
tipo_de_video: Perfil editorial escolhido no painel.
modelo_da_ia: Modelo que produziu o plano.
intencao: Pedido do usuário que originou o plano.
"""
origem: str
acoes: tuple[AcaoDoPlano, ...]
video_id: str | None = None
tipo_de_video: str | None = None
modelo_da_ia: str | None = None
intencao: str | None = None
class RepositorioDePlanos:
"""
Grava e lê planos de edição e o resultado de suas aplicações.
Atributos:
conexao: Conexão SQLite já aberta e com o esquema aplicado.
"""
def __init__(self, conexao: sqlite3.Connection) -> None:
"""
Inicializa o repositório sobre uma conexão existente.
Parâmetros:
conexao: Conexão SQLite já aberta e com o esquema aplicado.
"""
self.conexao = conexao
def registrar_plano(self, plano: PlanoDeEdicao) -> int:
"""
Grava um plano e suas ações, devolvendo o identificador criado.
Planos nunca são substituídos: cada devolução da IA é um registro
novo, porque o histórico de tentativas é justamente o que se quer
preservar.
Parâmetros:
plano: Plano a gravar.
Retorna:
O identificador do plano gravado.
Pode gerar:
ErroDePlano: quando o plano não tem origem ou não tem ação alguma.
"""
if not plano.origem or not plano.origem.strip():
raise ErroDePlano("O plano precisa indicar a origem (mídia ou sequência).")
if not plano.acoes:
raise ErroDePlano("Um plano sem ações não é gravável.")
with self.conexao:
if plano.video_id:
self.conexao.execute(
"INSERT OR IGNORE INTO videos (id) VALUES (?)", (plano.video_id,))
cursor = self.conexao.execute(
"""INSERT INTO planos_de_edicao
(video_id, origem, tipo_de_video, modelo_da_ia, intencao)
VALUES (?, ?, ?, ?, ?)""",
(plano.video_id, plano.origem, plano.tipo_de_video,
plano.modelo_da_ia, plano.intencao),
)
plano_id = int(cursor.lastrowid)
self.conexao.executemany(
"""INSERT INTO acoes_do_plano
(plano_id, ordem, tipo, inicio, fim, motivo, parametros)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
[(plano_id, ordem, acao.tipo, acao.inicio, acao.fim, acao.motivo,
json.dumps(acao.parametros, ensure_ascii=False))
for ordem, acao in enumerate(plano.acoes)],
)
return plano_id
def registrar_aplicacao(
self,
plano_id: int,
sucesso: bool,
acoes_aplicadas: int = 0,
sequencia: str | None = None,
mensagem: str | None = None,
) -> int:
"""
Grava o resultado de aplicar um plano na timeline.
A falha é gravada tanto quanto o sucesso: saber que um plano não pôde
ser aplicado, e por quê, é informação de diagnóstico que hoje se perde.
Parâmetros:
plano_id: Plano que foi aplicado.
sucesso: Se a aplicação terminou sem erro.
acoes_aplicadas: Quantas ações efetivamente entraram na timeline.
sequencia: Nome da sequência onde foi aplicado.
mensagem: Erro ou observação da aplicação.
Retorna:
O identificador da aplicação registrada.
Pode gerar:
ErroDePlano: quando o plano informado não existe.
"""
existe = self.conexao.execute(
"SELECT 1 FROM planos_de_edicao WHERE id = ?", (plano_id,)).fetchone()
if existe is None:
raise ErroDePlano(f"Plano {plano_id} não existe.")
with self.conexao:
cursor = self.conexao.execute(
"""INSERT INTO aplicacoes_do_plano
(plano_id, sequencia, sucesso, acoes_aplicadas, mensagem)
VALUES (?, ?, ?, ?, ?)""",
(plano_id, sequencia, 1 if sucesso else 0, acoes_aplicadas, mensagem),
)
return int(cursor.lastrowid)
def carregar_plano(self, plano_id: int) -> PlanoDeEdicao | None:
"""
Lê um plano gravado, com suas ações em ordem.
Parâmetros:
plano_id: Plano a carregar.
Retorna:
O plano, ou None quando o identificador não existe.
"""
cabecalho = self.conexao.execute(
"SELECT * FROM planos_de_edicao WHERE id = ?", (plano_id,)).fetchone()
if cabecalho is None:
return None
acoes = tuple(
AcaoDoPlano(
tipo=linha["tipo"], inicio=linha["inicio"], fim=linha["fim"],
motivo=linha["motivo"],
parametros=json.loads(linha["parametros"] or "{}"),
)
for linha in self.conexao.execute(
"SELECT * FROM acoes_do_plano WHERE plano_id = ? ORDER BY ordem",
(plano_id,),
)
)
return PlanoDeEdicao(
origem=cabecalho["origem"], acoes=acoes, video_id=cabecalho["video_id"],
tipo_de_video=cabecalho["tipo_de_video"],
modelo_da_ia=cabecalho["modelo_da_ia"], intencao=cabecalho["intencao"],
)
def listar_planos_do_video(self, video_id: str) -> list[dict[str, object]]:
"""
Lista o histórico de planos de um vídeo, do mais recente ao mais antigo.
Parâmetros:
video_id: Vídeo cujos planos serão listados.
Retorna:
Um resumo por plano, com contagem de ações e de aplicações.
"""
return [
dict(linha) for linha in self.conexao.execute(
"""SELECT p.id, p.criado_em, p.tipo_de_video, p.modelo_da_ia,
(SELECT count(*) FROM acoes_do_plano a WHERE a.plano_id = p.id)
AS total_de_acoes,
(SELECT count(*) FROM aplicacoes_do_plano ap
WHERE ap.plano_id = p.id AND ap.sucesso = 1)
AS aplicacoes_com_sucesso
FROM planos_de_edicao p
WHERE p.video_id = ?
ORDER BY p.criado_em DESC""",
(video_id,),
)
]
def plano_de_dict(
dados: dict,
video_id: str | None = None,
tipo_de_video: str | None = None,
modelo_da_ia: str | None = None,
intencao: str | None = None,
) -> PlanoDeEdicao:
"""
Converte o JSON devolvido pela IA no plano gravável.
Aceita o mesmo formato que o painel já valida — ``source`` mais uma lista
``actions`` com ``kind``, ``start`` e ``end`` — para que gravar o plano não
exija mudar o contrato com a IA.
Parâmetros:
dados: JSON do plano, já desserializado.
video_id: Vídeo do banco correspondente, quando conhecido.
tipo_de_video: Perfil editorial escolhido no painel.
modelo_da_ia: Modelo que produziu o plano.
intencao: Pedido do usuário que originou o plano.
Retorna:
O plano pronto para ``RepositorioDePlanos.registrar_plano``.
Pode gerar:
ErroDePlano: quando falta ``source`` ou ``actions`` não é uma lista
não vazia.
"""
origem = dados.get("source")
acoes_brutas = dados.get("actions")
if not isinstance(origem, str) or not origem.strip():
raise ErroDePlano("O plano precisa ter 'source'.")
if not isinstance(acoes_brutas, list) or not acoes_brutas:
raise ErroDePlano("O plano precisa ter uma lista 'actions' não vazia.")
conhecidos = {"kind", "start", "end", "reason", "motivo"}
acoes: list[AcaoDoPlano] = []
for indice, bruta in enumerate(acoes_brutas):
if not isinstance(bruta, dict) or not bruta.get("kind"):
raise ErroDePlano(f"A ação {indice} não tem 'kind'.")
acoes.append(AcaoDoPlano(
tipo=str(bruta["kind"]),
inicio=_numero(bruta.get("start")),
fim=_numero(bruta.get("end")),
motivo=bruta.get("reason") or bruta.get("motivo"),
parametros={chave: valor for chave, valor in bruta.items()
if chave not in conhecidos},
))
return PlanoDeEdicao(
origem=origem, acoes=tuple(acoes), video_id=video_id,
tipo_de_video=tipo_de_video, modelo_da_ia=modelo_da_ia, intencao=intencao,
)
def _numero(valor: object) -> float | None:
"""Converte um valor em float quando ele for numérico, senão devolve None."""
if isinstance(valor, bool) or not isinstance(valor, (int, float)):
return None
return float(valor)
__all__ = [
"AcaoDoPlano",
"ErroDePlano",
"PlanoDeEdicao",
"RepositorioDePlanos",
"plano_de_dict",
]
@@ -13,7 +13,13 @@ import sqlite3
from pathlib import Path
from typing import Iterable
from ..scanner.retakes.modelos_de_retakes import EvidenciaDeRetake, GrupoDeRetake, TomadaDeRetake
from ..scanner.retakes.modelos_de_retakes import (
STATUS_DE_REVISAO_VALIDOS,
EvidenciaDeRetake,
GrupoDeRetake,
TomadaDeRetake,
)
from ..scanner.retakes.repositorio_de_retakes import GrupoDeRetakeInexistente
from .conexao import abrir_banco
VERSAO_REGRA_FALAS = 1
@@ -70,10 +76,10 @@ class RepositorioDeRetakesSQLite:
for grupo in grupos:
self.conexao.execute(
"""INSERT INTO grupos_de_retake
(id, video_id, faixa_id, tipo, confianca, criado_em)
VALUES (?, ?, ?, ?, ?, ?)""",
(id, video_id, faixa_id, tipo, confianca, criado_em, status_de_revisao)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(grupo.id, grupo.video_id, grupo.faixa_id, grupo.tipo,
grupo.confianca, grupo.criado_em),
grupo.confianca, grupo.criado_em, grupo.status_de_revisao),
)
for tomada in grupo.tomadas:
self.conexao.execute(
@@ -143,5 +149,46 @@ class RepositorioDeRetakesSQLite:
tomadas=tomadas,
evidencias=evidencias,
criado_em=linha["criado_em"],
status_de_revisao=linha["status_de_revisao"],
))
return grupos
def atualizar_status_de_revisao(
self, video_id: str, grupo_id: str, novo_status: str,
) -> GrupoDeRetake:
"""
Registra a decisão manual do usuário sobre um grupo de retakes.
A revisão não altera a detecção automática: grava apenas o estado
escolhido, preservando os demais grupos do vídeo.
Parâmetros:
video_id: Vídeo dono do grupo revisado.
grupo_id: Grupo que recebeu a decisão.
novo_status: Estado escolhido pelo usuário.
Retorna:
O grupo já com o novo status.
Pode gerar:
ValueError: quando o status não é um dos válidos.
GrupoDeRetakeInexistente: quando o grupo não existe no vídeo.
"""
if novo_status not in STATUS_DE_REVISAO_VALIDOS:
validos = ", ".join(sorted(STATUS_DE_REVISAO_VALIDOS))
raise ValueError(
f"Status de revisão inválido: {novo_status!r}. Válidos: {validos}."
)
with self.conexao:
cursor = self.conexao.execute(
"""UPDATE grupos_de_retake SET status_de_revisao = ?
WHERE video_id = ? AND id = ?""",
(novo_status, video_id, grupo_id),
)
if cursor.rowcount == 0:
raise GrupoDeRetakeInexistente(video_id, grupo_id)
grupo = next(
(item for item in self.carregar(video_id) if item.id == grupo_id), None)
if grupo is None:
raise GrupoDeRetakeInexistente(video_id, grupo_id)
return grupo
+2
View File
@@ -3,5 +3,7 @@ opencv-python
scenedetect
onnxruntime
mediapipe
# Assinatura facial (embedding) para reconhecer a mesma pessoa entre tomadas.
insightface
# Apenas macOS; permite usar Vision diretamente a partir do Python.
pyobjc-framework-Vision
@@ -56,6 +56,10 @@ class ConfiguracaoVisualDoScanner:
# Limita quantos frames por clipe recebem a interpretação editorial da Apple
# Intelligence (Foundation Models) — a etapa mais cara. None interpreta todos.
maximo_de_frames_interpretados: int | None = None
# Assinatura facial (InsightFace) — desligada por padrão. Nem o OpenCV nem
# o Apple Vision reconhecem que o mesmo rosto aparece em clipes diferentes;
# esse recurso existe só para isso, então fica fora do custo padrão.
identidade_facial: bool = False
def __post_init__(self) -> None:
if self.modo_de_analise not in {"som", "imagem", "ambos"}:
@@ -100,6 +104,7 @@ class ConfiguracaoVisualDoScanner:
usar_proxy=bool(dados.get("usar_proxy", True)),
resolucao_do_proxy=int(dados.get("resolucao_do_proxy", RESOLUCAO_PADRAO_DO_PROXY)),
maximo_de_frames_interpretados=_inteiro_ou_nulo(dados.get("maximo_de_frames_interpretados", None)),
identidade_facial=bool(dados.get("identidade_facial", False)),
)
def para_dict(self) -> dict[str, Any]:
@@ -122,4 +127,5 @@ class ConfiguracaoVisualDoScanner:
"usar_proxy": self.usar_proxy,
"resolucao_do_proxy": self.resolucao_do_proxy,
"maximo_de_frames_interpretados": self.maximo_de_frames_interpretados,
"identidade_facial": self.identidade_facial,
}
+26 -8
View File
@@ -190,25 +190,39 @@ def criar_detector_visual_local(
AnalisadorDeComposicaoOpenCV,
AnalisadorDeContinuidadeOpenCV,
AnalisadorDeQualidadeOpenCV,
AnalisadorDeRostosOpenCV,
ExtratorDeQuadrosOpenCV,
)
perfil = configuracao or ConfiguracaoVisualDoScanner(intervalo_em_segundos=intervalo_em_segundos)
analisadores_de_frame: list = [
AnalisadorDeQualidadeOpenCV(),
AnalisadorDeComposicaoOpenCV(),
]
if "faces" in perfil.recursos_vision:
analisadores_de_frame.append(AnalisadorDeRostosOpenCV())
if perfil.identidade_facial:
analisadores_de_frame.append(_analisador_de_identidade_facial())
return DetectorDeCenas(
ExtratorDeQuadrosOpenCV(
intervalo_em_segundos=perfil.intervalo_em_segundos,
diretorio_de_cache=diretorio_de_cache,
),
analisadores_de_frame=[
AnalisadorDeQualidadeOpenCV(),
AnalisadorDeComposicaoOpenCV(),
],
analisadores_de_frame=analisadores_de_frame,
analisadores_de_sequencia=[AnalisadorDeContinuidadeOpenCV()],
detectores_de_intervalo=_detectores_de_cena_do_perfil(perfil),
)
def _analisador_de_identidade_facial():
"""Monta o analisador InsightFace, compartilhado pelos dois detectores locais."""
from ..integracoes.visual import AnalisadorDeRostosInsightFace
return AnalisadorDeRostosInsightFace()
def criar_detector_apple_vision(
diretorio_de_cache: str | None = ".frames",
intervalo_em_segundos: float = 1.0,
@@ -239,13 +253,17 @@ def criar_detector_apple_vision(
extrator, CompactadorDeProxy(), largura_maxima=perfil.resolucao_do_proxy
)
return DetectorDeCenas(
extrator,
analisadores_de_frame=[AnalisadorAppleVisionNativo(
analisadores_de_frame: list = [AnalisadorAppleVisionNativo(
recursos=tuple(sorted(perfil.recursos_vision)),
interpretar_com_apple_intelligence=perfil.interpretar_cena,
maximo_de_frames_interpretados=perfil.maximo_de_frames_interpretados,
)],
)]
if perfil.identidade_facial:
analisadores_de_frame.append(_analisador_de_identidade_facial())
return DetectorDeCenas(
extrator,
analisadores_de_frame=analisadores_de_frame,
analisadores_de_sequencia=[AnalisadorSequenciaAppleVisionNativo()]
if perfil.analisar_continuidade else [],
detectores_de_intervalo=_detectores_de_cena_do_perfil(perfil),
+105 -1
View File
@@ -4,6 +4,7 @@ import json
import tempfile
from engine.dominio import Clipe, Faixa, IntervaloDeTempo, Timeline
from engine.integracoes.visual.analisadores import AnalisadorDeRostosInsightFace
from engine.integracoes.visual.contratos import (AnalisadorDeFrame, AnalisadorDeSequenciaDeQuadros,
DetectorDeIntervalosDeCena,
ExtratorDeQuadros)
@@ -13,11 +14,34 @@ from engine.integracoes.visual.apple_vision import (AnalisadorAppleVisionNativo,
AnalisadorSequenciaAppleVisionNativo)
from engine.scanner.coordenacao import ContextoDeAnalise
from engine.scanner.modelos import Cena, EvidenciaVisual
from engine.scanner.visual import AnaliseVisualDaTimeline, DetectorDeCenas, ResultadoVisualDoClipe
from engine.scanner.visual import (AnaliseVisualDaTimeline, DetectorDeCenas, ResultadoVisualDoClipe,
criar_detector_apple_vision, criar_detector_visual_local)
from engine.scanner.relatorio_visual import gerar_relatorio_visual, gerar_resumo_visual_markdown
from engine.scanner.configuracao_visual import ConfiguracaoVisualDoScanner
class RostoSimulado:
"""Dublê de um rosto devolvido por `FaceAnalysis.get`."""
def __init__(self, bbox, det_score=0.9, embedding=None, idade=None, genero=None):
self.bbox = bbox
self.det_score = det_score
self.normed_embedding = embedding
self.age = idade
self.sex = genero
class AppInsightFaceSimulado:
"""Dublê do `FaceAnalysis`: devolve uma lista fixa de rostos."""
def __init__(self, rostos):
self._rostos = rostos
def get(self, imagem):
self.imagem_recebida = imagem
return self._rostos
class ExtratorSimulado(ExtratorDeQuadros):
def extrair(self, clipe):
return [
@@ -211,6 +235,86 @@ class TesteAnaliseVisualLocal(unittest.TestCase):
self.assertEqual(extrator._timestamps(inicio, fim), [2.3])
def test_insightface_converte_rosto_em_evidencias_de_rosto_e_identidade(self):
rosto = RostoSimulado(bbox=[100.0, 50.0, 300.0, 350.0], det_score=0.87,
embedding=[0.1, 0.2, 0.3], idade=28, genero="F")
analisador = AnalisadorDeRostosInsightFace(app=AppInsightFaceSimulado([rosto]))
quadro = QuadroDeVideo(1.5, 0, 1000, 1000, imagem="frame-bgr")
evidencias = analisador.analisar(quadro)
self.assertEqual([item.tipo for item in evidencias], ["rosto", "identidade_facial"])
caixa = evidencias[0].valor["bounding_box"]
self.assertAlmostEqual(caixa["x"], 0.1)
self.assertAlmostEqual(caixa["y"], 0.05)
self.assertAlmostEqual(caixa["largura"], 0.2)
self.assertAlmostEqual(caixa["altura"], 0.3)
self.assertEqual(evidencias[0].confianca, 0.87)
identidade = evidencias[1].valor
self.assertEqual(identidade["assinatura_facial"], [0.1, 0.2, 0.3])
self.assertEqual(identidade["dimensoes"], 3)
self.assertEqual(identidade["idade_aproximada"], 28)
self.assertEqual(identidade["genero_aparente"], "F")
def test_insightface_descarta_rosto_menor_que_o_minimo_relativo(self):
rosto_pequeno = RostoSimulado(bbox=[0.0, 0.0, 10.0, 10.0])
analisador = AnalisadorDeRostosInsightFace(app=AppInsightFaceSimulado([rosto_pequeno]))
quadro = QuadroDeVideo(0.0, 0, 1000, 1000, imagem="frame-bgr")
self.assertEqual(analisador.analisar(quadro), [])
def test_insightface_nao_gera_identidade_sem_embedding(self):
rosto_sem_embedding = RostoSimulado(bbox=[100.0, 50.0, 300.0, 350.0])
analisador = AnalisadorDeRostosInsightFace(app=AppInsightFaceSimulado([rosto_sem_embedding]))
quadro = QuadroDeVideo(0.0, 0, 1000, 1000, imagem="frame-bgr")
evidencias = analisador.analisar(quadro)
self.assertEqual([item.tipo for item in evidencias], ["rosto"])
def test_insightface_exige_imagem_decodificada_ou_caminho_persistido(self):
analisador = AnalisadorDeRostosInsightFace(app=AppInsightFaceSimulado([]))
quadro = QuadroDeVideo(0.0, 0, 1000, 1000, imagem=None, caminho=None)
with self.assertRaises(ValueError):
analisador.analisar(quadro)
def test_insightface_rejeita_tamanho_minimo_relativo_invalido(self):
with self.assertRaises(ValueError):
AnalisadorDeRostosInsightFace(tamanho_minimo_relativo=1.5)
def test_detector_local_inclui_rostos_opencv_quando_recurso_faces_ativo(self):
detector = criar_detector_visual_local()
nomes = [type(item).__name__ for item in detector.analisadores_de_frame]
self.assertIn("AnalisadorDeRostosOpenCV", nomes)
self.assertNotIn("AnalisadorDeRostosInsightFace", nomes)
def test_detector_local_inclui_insightface_quando_identidade_facial_ativa(self):
perfil = ConfiguracaoVisualDoScanner(identidade_facial=True)
detector = criar_detector_visual_local(configuracao=perfil)
nomes = [type(item).__name__ for item in detector.analisadores_de_frame]
self.assertIn("AnalisadorDeRostosInsightFace", nomes)
def test_detector_apple_vision_inclui_insightface_quando_identidade_facial_ativa(self):
perfil = ConfiguracaoVisualDoScanner(identidade_facial=True)
detector = criar_detector_apple_vision(configuracao=perfil)
nomes = [type(item).__name__ for item in detector.analisadores_de_frame]
self.assertIn("AnalisadorAppleVisionNativo", nomes)
self.assertIn("AnalisadorDeRostosInsightFace", nomes)
def test_detector_apple_vision_nao_inclui_insightface_por_padrao(self):
detector = criar_detector_apple_vision()
nomes = [type(item).__name__ for item in detector.analisadores_de_frame]
self.assertNotIn("AnalisadorDeRostosInsightFace", nomes)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,347 @@
"""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()