feat: criada a aba Análises no painel CEP para ler do banco (anali

- criada a aba Análises no painel CEP para ler do banco (analises.db) e copiar/abrir o J-SOM de som (transcrição por cena) e o de imagem (evidências visuais), com novo módulo engine/gerar_relatorio_de_analises.py e testes
- criado o módulo engine/editor (leitura do plano de edição em JSON, tradução de tempos de origem para a timeline ativa e aplicação de cortes/zooms/marcadores no Premiere via MCP) e o ponto de entrada engine/aplicar_plano_de_edicao.py, com testes; corrigido engine/integracoes/premiere/cliente_mcp.py para levantar os erros específicos já declarados em erros_mcp.py em vez de RuntimeError genérico

Resumo:
- 15 arquivos alterados
- 8 novos
- 5 modificados
- 2 removidos

 7 files changed, 172 insertions(+), 8 deletions(-)

Arquivos:
  - .jhonny/analises.db
  - .jhonny/analises.db-shm
  - .jhonny/analises.db-wal
  - code/cep-plugin/index.html
  - code/cep-plugin/main.js
  - code/cep-plugin/styles.css
  - code/engine/integracoes/premiere/cliente_mcp.py
  - code/engine/aplicar_plano_de_edicao.py
  - code/engine/editor/
  - code/engine/gerar_relatorio_de_analises.py
  - code/engine/testes/duplos_de_premiere.py
  - code/engine/testes/test_aplicador_de_plano_de_edicao.py
  - code/engine/testes/test_gerar_relatorio_de_analises.py
  - code/engine/testes/test_leitor_de_plano_de_edicao.py
  - code/engine/testes/test_mapeador_de_tempo.py
This commit is contained in:
João Henrique
2026-09-08 19:35:30 -04:00
parent 400c3c93dd
commit 4308c72441
24 changed files with 1591 additions and 8 deletions
Binary file not shown.
Binary file not shown.
View File
+22
View File
@@ -31,6 +31,9 @@
<button class="tab-button" id="tabBtnRetakes" role="tab" aria-selected="false" aria-controls="tabRetakes" onclick="switchTab('retakes')" type="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 <span class="tab-icon" aria-hidden="true">↻</span>Retakes
</button> </button>
<button class="tab-button" id="tabBtnAnalises" role="tab" aria-selected="false" aria-controls="tabAnalises" onclick="switchTab('analises')" type="button">
<span class="tab-icon" aria-hidden="true">⎘</span>Análises
</button>
<button class="tab-button" id="tabBtnBridge" role="tab" aria-selected="false" aria-controls="tabBridge" onclick="switchTab('bridge')" type="button"> <button class="tab-button" id="tabBtnBridge" role="tab" aria-selected="false" aria-controls="tabBridge" onclick="switchTab('bridge')" type="button">
<span class="tab-icon" aria-hidden="true">⇄</span>Conexão <span class="tab-icon" aria-hidden="true">⇄</span>Conexão
</button> </button>
@@ -333,6 +336,25 @@
<div id="retakesLog" role="log" aria-live="polite" aria-relevant="additions" tabindex="0" aria-label="Log de retakes"></div> <div id="retakesLog" role="log" aria-live="polite" aria-relevant="additions" tabindex="0" aria-label="Log de retakes"></div>
</div> </div>
<!-- ============================== ANÁLISES ============================= -->
<div class="tab-panel" id="tabAnalises" role="tabpanel" aria-labelledby="tabBtnAnalises" hidden>
<p class="tab-intro">Lê direto do banco de análises o J-SOM de imagem (evidências visuais) e o de som (transcrição por cena). Nada é reprocessado.</p>
<section class="card">
<div class="card-head"><h2>Banco de análises</h2><span class="card-kicker" id="analisesStatus">—</span></div>
<p class="field-help" id="analisesCaminho">—</p>
<div class="button-row">
<button class="button button-primary" id="btnAnalisesCarregar" onclick="analisesCarregar()" type="button">Carregar do banco</button>
<button class="button button-ghost" onclick="analisesAbrirJson()" type="button" disabled id="btnAnalisesAbrir">Abrir JSON</button>
<button class="button button-ghost" onclick="analisesCopiar()" type="button" disabled id="btnAnalisesCopiar">Copiar conteúdo</button>
</div>
</section>
<section class="card" id="analisesResumoCard" hidden>
<div class="card-head"><h2 id="analisesVideoNome">—</h2><span class="card-kicker" id="analisesResumo">—</span></div>
<p class="field-help">Som: transcrição com emoção por fala. Imagem: evidências Apple Vision por clipe. “Cena” sem marca indica clipe sem corte de cena gravado.</p>
<div id="analisesCenas" role="list" aria-label="Cenas com transcrição e evidências"></div>
</section>
</div>
<!-- ============================== CONEXÃO ============================== --> <!-- ============================== CONEXÃO ============================== -->
+93
View File
@@ -206,6 +206,7 @@ function switchTab(name) {
{ key: "silence", panel: "tabSilence", btn: "tabBtnSilence" }, { key: "silence", panel: "tabSilence", btn: "tabBtnSilence" },
{ key: "scanner", panel: "tabScanner", btn: "tabBtnScanner" }, { key: "scanner", panel: "tabScanner", btn: "tabBtnScanner" },
{ key: "retakes", panel: "tabRetakes", btn: "tabBtnRetakes" }, { key: "retakes", panel: "tabRetakes", btn: "tabBtnRetakes" },
{ key: "analises", panel: "tabAnalises", btn: "tabBtnAnalises" },
{ key: "bridge", panel: "tabBridge", btn: "tabBtnBridge" }, { key: "bridge", panel: "tabBridge", btn: "tabBtnBridge" },
{ key: "models", panel: "tabModels", btn: "tabBtnModels" }, { key: "models", panel: "tabModels", btn: "tabBtnModels" },
{ key: "settings", panel: "tabSettings", btn: "tabBtnSettings" }, { key: "settings", panel: "tabSettings", btn: "tabBtnSettings" },
@@ -224,6 +225,98 @@ function switchTab(name) {
if (name === "settings") renderEditorPersonalities(); if (name === "settings") renderEditorPersonalities();
if (name === "scanner") { renderScannerTranscriptionOptions(); loadScannerProfile(); renderScannerHuggingFaceStatus(); } if (name === "scanner") { renderScannerTranscriptionOptions(); loadScannerProfile(); renderScannerHuggingFaceStatus(); }
if (name === "retakes") retakesAoAbrir(); if (name === "retakes") retakesAoAbrir();
if (name === "analises") analisesAoAbrir();
}
// ---- Análises ---------------------------------------------------------------
// Aba que lê o banco de análises (.jhonny/analises.db) já produzido pelo
// Scanner e mostra/copiar os dois J-SOM: som (transcrição por cena) e
// imagem (evidências visuais). A leitura acontece na engine, via script
// dedicado; a aba nunca escreve no banco.
var ANALISES_ENGINE_SCRIPT = "/Volumes/Merongo/SISTEMAS/GENIAL SISTEMAS/Jhonny/code/engine/gerar_relatorio_de_analises.py";
var ANALISES_DB_PATH = "/Volumes/Merongo/SISTEMAS/GENIAL SISTEMAS/Jhonny/.jhonny/analises.db";
var analisesRelatorioPath = null;
function analisesAoAbrir() {
document.getElementById("analisesCaminho").textContent = ANALISES_DB_PATH;
}
function analisesFormatarTempo(segundos) {
var s = Number(segundos) || 0;
var min = Math.floor(s / 60), resto = (s - min * 60).toFixed(1);
return min + ":" + (resto < 10 ? "0" : "") + resto;
}
function analisesCarregar() {
var stamp = new Date().toISOString().replace(/[:.]/g, "-");
analisesRelatorioPath = path.join(os.tmpdir(), "jhonny-analises-" + stamp + ".json");
setBusy("btnAnalisesCarregar", true); taskStart("Análises", "Lendo o banco de análises", null);
var child = childProcess.spawn(SCANNER_PYTHON_BIN, [ANALISES_ENGINE_SCRIPT, ANALISES_DB_PATH, analisesRelatorioPath],
{ cwd: "/Volumes/Merongo/SISTEMAS/GENIAL SISTEMAS/Jhonny/code", stdio: ["ignore", "pipe", "pipe"] });
var stderr = "", stdout = "";
child.stdout.on("data", function (chunk) { stdout += chunk.toString(); });
child.stderr.on("data", function (chunk) { stderr += chunk.toString(); });
child.on("error", function (error) { setBusy("btnAnalisesCarregar", false); taskEnd(false, "Análises indisponíveis", error.stack || error.message); });
child.on("close", function (code) {
setBusy("btnAnalisesCarregar", false);
if (code !== 0 || !fs.existsSync(analisesRelatorioPath)) {
taskEnd(false, "Leitura do banco falhou", stderr || stdout || "Não foi possível gerar o relatório.");
return;
}
taskEnd(true, "Relatório carregado do banco");
analisesRender(JSON.parse(fs.readFileSync(analisesRelatorioPath, "utf-8")));
});
}
function analisesRender(relatorio) {
var video = relatorio.video || {};
var cenas = relatorio.cenas || [];
var falas = cenas.reduce(function (n, c) { return n + (c.falas || []).length; }, 0);
var evidencias = cenas.reduce(function (n, c) {
var resumo = c.resumo_de_evidencias || {};
return n + Object.keys(resumo).reduce(function (m, k) { return m + resumo[k]; }, 0);
}, 0);
document.getElementById("analisesStatus").textContent = "Carregado";
document.getElementById("analisesVideoNome").textContent = video.nome || "(sem nome)";
document.getElementById("analisesResumo").textContent =
cenas.length + " cenas · " + falas + " falas · " + evidencias + " evidências visuais";
document.getElementById("analisesResumoCard").hidden = false;
document.getElementById("btnAnalisesAbrir").disabled = false;
document.getElementById("btnAnalisesCopiar").disabled = false;
var lista = document.getElementById("analisesCenas");
lista.innerHTML = "";
cenas.forEach(function (cena) {
var bloco = document.createElement("div");
bloco.className = "analise-cena";
bloco.setAttribute("role", "listitem");
var titulo = cena.cena_registrada ? "" : " (sem cena gravada)";
var tipos = Object.keys(cena.resumo_de_evidencias || {})
.sort().map(function (k) { return k + " " + cena.resumo_de_evidencias[k]; }).join(", ") || "—";
var falasHtml = (cena.falas || []).map(function (fala) {
return "<p class=\"analise-fala\"><strong>" + (fala.emocao || "—") + "</strong> " +
analisesFormatarTempo(fala.inicio) + "–" + analisesFormatarTempo(fala.fim) + " · " +
String(fala.texto || "").replace(/&/g, "&amp;").replace(/</g, "&lt;") + "</p>";
}).join("") || "<p class=\"analise-fala analise-vazia\">Sem fala neste clipe.</p>";
bloco.innerHTML =
"<div class=\"analise-cena-cab\"><strong>" + cena.clipe_id + titulo + "</strong>" +
"<span>" + analisesFormatarTempo(cena.inicio) + "–" + analisesFormatarTempo(cena.fim) + "</span></div>" +
falasHtml + "<p class=\"analise-evidencias\">Imagem: " + tipos + "</p>";
lista.appendChild(bloco);
});
}
function analisesAbrirJson() {
if (!analisesRelatorioPath) return;
childProcess.spawn("open", [analisesRelatorioPath], { detached: true, stdio: "ignore" }).unref();
}
function analisesCopiar() {
if (!analisesRelatorioPath) return;
try {
var copy = childProcess.spawn("pbcopy");
copy.stdin.end(fs.readFileSync(analisesRelatorioPath, "utf-8"));
showToast("ok", "J-SOM de som e imagem copiados (JSON completo).");
} catch (error) { showToast("err", error.message); }
} }
// ---- Retakes --------------------------------------------------------------- // ---- Retakes ---------------------------------------------------------------
+11
View File
@@ -612,3 +612,14 @@ button:focus-visible, input:focus-visible, select:focus-visible, textarea:focus-
.button, .path-field input, .select-field, #log, #silenceLog, #modelsLog, .status-panel, .step, .card { border-color: CanvasText; } .button, .path-field input, .select-field, #log, #silenceLog, #modelsLog, .status-panel, .step, .card { border-color: CanvasText; }
.progress-fill { background: Highlight; } .progress-fill { background: Highlight; }
} }
/* ---- Aba Análises ------------------------------------------------------- */
.analise-cena { border-top: 1px solid var(--border); padding: 8px 0; }
.analise-cena:first-child { border-top: 0; }
.analise-cena-cab { display: flex; justify-content: space-between; gap: 8px; color: var(--text); font-size: 11px; }
.analise-cena-cab span { color: var(--text-muted); font-variant-numeric: tabular-nums; }
.analise-fala { margin: 4px 0 0; font-size: 11px; color: var(--text); line-height: 1.35; }
.analise-fala strong { display: inline-block; min-width: 34px; color: var(--text-muted); font-weight: 600; text-transform: uppercase; }
.analise-fala.analise-vazia { color: var(--text-muted); font-style: italic; }
.analise-evidencias { margin: 6px 0 0; font-size: 10px; color: var(--text-muted); }
+85
View File
@@ -0,0 +1,85 @@
"""Entrada de linha de comando: aplica um plano de edição na sequência ativa do Premiere.
Espelha o ``executar_scanner.py``/``executar_retakes.py``: recebe o caminho
de um JSON de ações (``kind``/``start``/``end``/``params``/``reason``,
formato das skills de seleção de trechos) e aplica cada ação na sequência
que estiver ativa no Premiere no momento da execução, via o MCP existente em
``engine.integracoes.premiere``.
Uso::
python aplicar_plano_de_edicao.py "/caminho/0E6A8290_edit_actions.json"
Pré-requisito: o Premiere precisa estar aberto com a sequência correta ativa
e o painel/plugin do MCP em execução — este script não abre o Premiere nem
troca de sequência.
"""
from __future__ import annotations
import json
import sys
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.editor import AplicadorDePlanoDeEdicao, EscritaNoEditor, LeitorDePlanoDeEdicao, MapeadorDeTempoDeOrigemParaTimeline
from engine.integracoes.premiere.cliente_mcp import ClienteMCPPorStdio
from engine.integracoes.premiere.conversores import ConversorDeTimeline
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"
def executar(caminho_do_plano: Path) -> dict:
"""Lê o plano em ``caminho_do_plano`` e o aplica na sequência ativa do Premiere.
Devolve um dicionário pronto para virar JSON, com o resultado de cada
ação — para o painel exibir o que foi aplicado e o que falhou.
"""
plano = LeitorDePlanoDeEdicao().ler(caminho_do_plano)
cliente = ClienteMCPPorStdio(["node", str(CAMINHO_DO_SERVIDOR_MCP)])
with SessaoMCP(cliente):
acesso_ao_editor = AcessoAoEditor(AcessoATimeline(cliente))
aplicador = AplicadorDePlanoDeEdicao(
acesso_ao_editor=acesso_ao_editor,
conversor_de_timeline=ConversorDeTimeline(),
escrita=EscritaNoEditor(cliente),
mapeador=MapeadorDeTempoDeOrigemParaTimeline(),
)
resultado = aplicador.aplicar(plano)
return {
"arquivo_de_origem": plano.arquivo_de_origem,
"todas_bem_sucedidas": resultado.todas_bem_sucedidas,
"acoes": [
{
"tipo": item.acao.tipo.value,
"inicio": item.acao.inicio,
"fim": item.acao.fim,
"motivo": item.acao.motivo,
"sucesso": item.sucesso,
"detalhe": item.detalhe,
}
for item in resultado.resultados
],
}
def main() -> None:
"""Ponto de entrada da linha de comando."""
if len(sys.argv) != 2:
print("Uso: python aplicar_plano_de_edicao.py <caminho_do_plano.json>", file=sys.stderr)
sys.exit(2)
resultado = executar(Path(sys.argv[1]))
print(json.dumps(resultado, ensure_ascii=False, indent=2))
if not resultado["todas_bem_sucedidas"]:
sys.exit(1)
if __name__ == "__main__":
main()
+35
View File
@@ -0,0 +1,35 @@
"""Aplicação de planos de edição (cortes, zooms, marcadores) na timeline do Premiere.
Lê o JSON de ações produzido pelas skills de seleção de trechos
(``kind``/``start``/``end``/``params``/``reason``), traduz os tempos de
origem para a sequência ativa do Premiere e aplica as mutações por meio do
MCP existente em ``engine.integracoes.premiere``.
Não decide o que cortar — isso já vem pronto no plano. A responsabilidade
deste módulo é só: ler o plano, achar a posição atual de cada trecho na
timeline e executar a mutação correta, relatando o que deu certo e o que
não deu.
"""
from .aplicador_de_plano_de_edicao import AplicadorDePlanoDeEdicao, ResultadoDaAcao, ResultadoDaAplicacao
from .erros import ErroDeEdicao, ErroDeEscritaNoEditor, ErroDeMapeamentoDeTempo, ErroDePlanoInvalido
from .escrita import EscritaNoEditor
from .leitura import LeitorDePlanoDeEdicao
from .mapeamento import MapeadorDeTempoDeOrigemParaTimeline
from .modelos import AcaoDeEdicao, PlanoDeEdicao, TipoDeAcao
__all__ = [
"AcaoDeEdicao",
"AplicadorDePlanoDeEdicao",
"EscritaNoEditor",
"ErroDeEdicao",
"ErroDeEscritaNoEditor",
"ErroDeMapeamentoDeTempo",
"ErroDePlanoInvalido",
"LeitorDePlanoDeEdicao",
"MapeadorDeTempoDeOrigemParaTimeline",
"PlanoDeEdicao",
"ResultadoDaAcao",
"ResultadoDaAplicacao",
"TipoDeAcao",
]
@@ -0,0 +1,223 @@
"""Orquestração da aplicação de um plano de edição na sequência ativa do Premiere.
Liga o plano já validado (:mod:`.leitura`) à tradução de tempos
(:mod:`.mapeamento`) e à escrita no editor (:mod:`.escrita`), na ordem que
preserva a validade das posições calculadas — ver docstring de
:class:`AplicadorDePlanoDeEdicao`.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from ..dominio import Clipe, Faixa, Timeline
from ..integracoes.premiere.conversores import ConversorDeTimeline
from ..integracoes.premiere.leitura import AcessoAoEditor
from .erros import ErroDeEdicao, ErroDeMapeamentoDeTempo
from .escrita.escrita_no_editor import EscritaNoEditor
from .mapeamento.mapeador_de_tempo import MapeadorDeTempoDeOrigemParaTimeline
from .modelos import AcaoDeEdicao, PlanoDeEdicao, TipoDeAcao
_TOLERANCIA_DE_BORDA_EM_SEGUNDOS = 0.05
@dataclass(frozen=True)
class ResultadoDaAcao:
"""O que aconteceu ao tentar aplicar uma :class:`AcaoDeEdicao`."""
acao: AcaoDeEdicao
sucesso: bool
detalhe: str
@dataclass(frozen=True)
class ResultadoDaAplicacao:
"""O resultado, ação por ação, de aplicar um :class:`PlanoDeEdicao` inteiro."""
resultados: tuple[ResultadoDaAcao, ...] = field(default_factory=tuple)
@property
def todas_bem_sucedidas(self) -> bool:
"""``True`` só se nenhuma ação falhou."""
return all(resultado.sucesso for resultado in self.resultados)
@property
def falhas(self) -> tuple[ResultadoDaAcao, ...]:
"""Só as ações que não foram aplicadas."""
return tuple(resultado for resultado in self.resultados if not resultado.sucesso)
class AplicadorDePlanoDeEdicao:
"""Aplica um :class:`PlanoDeEdicao` inteiro na sequência ativa do Premiere.
Ordem de aplicação, para preservar a validade das posições calculadas:
1. Marcadores e zooms primeiro — não mudam a duração da timeline, então
a posição calculada para cada um continua válida até o fim.
2. Cortes por último, do fim para o começo (maior início primeiro) —
cada corte fecha o espaço que ocupava (ripple), deslocando para a
esquerda tudo que vem depois dele. Processar do fim para o começo
garante que a posição de um corte ainda não aplicado nunca é afetada
pelos cortes já aplicados.
A cada ação a timeline ativa é lida de novo no Premiere — nunca se
reaproveita uma leitura anterior — porque a ação anterior pode ter
mudado a posição dos clipes. Uma ação que falha é registrada no
resultado e não interrompe as demais; quem chama decide o que fazer com
as falhas.
"""
def __init__(
self,
acesso_ao_editor: AcessoAoEditor,
conversor_de_timeline: ConversorDeTimeline,
escrita: EscritaNoEditor,
mapeador: MapeadorDeTempoDeOrigemParaTimeline,
) -> None:
self.acesso_ao_editor = acesso_ao_editor
self.conversor_de_timeline = conversor_de_timeline
self.escrita = escrita
self.mapeador = mapeador
def aplicar(self, plano: PlanoDeEdicao) -> ResultadoDaAplicacao:
"""Aplica todas as ações de ``plano`` e devolve o resultado de cada uma."""
pontuais = [acao for acao in plano.acoes if acao.tipo is not TipoDeAcao.CORTE]
cortes = sorted(plano.acoes_do_tipo(TipoDeAcao.CORTE), key=lambda acao: acao.inicio, reverse=True)
resultados = [self._aplicar_acao_pontual(acao, plano.arquivo_de_origem) for acao in pontuais]
resultados += [self._aplicar_corte(acao, plano.arquivo_de_origem) for acao in cortes]
return ResultadoDaAplicacao(tuple(resultados))
def _timeline_atual(self) -> Timeline:
"""Lê a sequência ativa do Premiere agora, sem cache."""
return self.conversor_de_timeline.converter(self.acesso_ao_editor.obter_timeline_ativa())
def _aplicar_acao_pontual(self, acao: AcaoDeEdicao, arquivo_de_origem: str) -> ResultadoDaAcao:
"""Aplica um marcador ou zoom, que não precisam dividir nem remover clipes."""
try:
if acao.tipo is TipoDeAcao.MARCADOR:
self._aplicar_marcador(acao, arquivo_de_origem)
return ResultadoDaAcao(acao, True, "marcador criado")
if acao.tipo is TipoDeAcao.ZOOM:
quantidade = self._aplicar_zoom(acao, arquivo_de_origem)
return ResultadoDaAcao(acao, True, f"zoom aplicado em {quantidade} clipe(s)")
if acao.tipo is TipoDeAcao.TEXTO:
self._aplicar_texto_como_marcador(acao, arquivo_de_origem)
return ResultadoDaAcao(
acao, True,
"texto na tela não é suportado pela API de scripting do Premiere; "
"registrado como marcador para o editor aplicar manualmente"
)
raise ErroDeEdicao(f"Tipo de ação sem tratamento: {acao.tipo!r}.")
except ErroDeEdicao as erro:
return ResultadoDaAcao(acao, False, str(erro))
def _aplicar_marcador(self, acao: AcaoDeEdicao, arquivo_de_origem: str) -> None:
instante = self._instante_na_timeline(acao.inicio, arquivo_de_origem)
nome = str(acao.parametros.get("content", "MARCADOR"))
self.escrita.adicionar_marcador(instante, nome, acao.motivo)
def _aplicar_texto_como_marcador(self, acao: AcaoDeEdicao, arquivo_de_origem: str) -> None:
instante = self._instante_na_timeline(acao.inicio, arquivo_de_origem)
conteudo = str(acao.parametros.get("content", ""))
self.escrita.adicionar_marcador(instante, "TEXTO (aplicar manualmente)", f"{conteudo} — {acao.motivo}")
def _aplicar_zoom(self, acao: AcaoDeEdicao, arquivo_de_origem: str) -> int:
fator_de_escala = float(acao.parametros.get("scale", 1.3))
faixa_de_video = self._faixa_de_video(self._timeline_atual())
clipes = self.mapeador.clipes_do_arquivo(faixa_de_video, arquivo_de_origem)
alvos = [clipe for clipe in clipes if self._sobrepoe(clipe, acao.inicio, acao.fim)]
if not alvos:
raise ErroDeMapeamentoDeTempo(
f"Nenhum clipe de vídeo de {arquivo_de_origem!r} sobrepõe {acao.inicio}-{acao.fim}s de origem para o zoom."
)
for clipe in alvos:
self.escrita.aplicar_zoom(clipe.identificador, fator_de_escala)
return len(alvos)
def _aplicar_corte(self, acao: AcaoDeEdicao, arquivo_de_origem: str) -> ResultadoDaAcao:
"""Remove, em cada faixa afetada, todos os clipes contidos no intervalo do corte."""
try:
trechos_removidos = 0
for faixa in self._timeline_atual().faixas:
trechos_removidos += self._cortar_faixa(faixa, acao, arquivo_de_origem)
if trechos_removidos == 0:
raise ErroDeMapeamentoDeTempo(
f"Nenhum clipe de {arquivo_de_origem!r} sobrepõe o intervalo {acao.inicio}-{acao.fim}s de origem."
)
return ResultadoDaAcao(acao, True, f"{trechos_removidos} trecho(s) removido(s)")
except ErroDeEdicao as erro:
return ResultadoDaAcao(acao, False, str(erro))
def _cortar_faixa(self, faixa: Faixa, acao: AcaoDeEdicao, arquivo_de_origem: str) -> int:
"""Corta o trecho de ``acao`` em uma única faixa e devolve quantos clipes foram removidos."""
clipes = self.mapeador.clipes_do_arquivo(faixa, arquivo_de_origem)
afetados = [clipe for clipe in clipes if self._sobrepoe(clipe, acao.inicio, acao.fim)]
if not afetados:
return 0
primeiro, ultimo = afetados[0], afetados[-1]
inicio_na_timeline = self._borda_de_entrada(primeiro, acao.inicio)
fim_na_timeline = self._borda_de_saida(ultimo, acao.fim)
self.escrita.dividir_clipe_se_necessario(inicio_na_timeline, faixa.indice, faixa.tipo)
self.escrita.dividir_clipe_se_necessario(fim_na_timeline, faixa.indice, faixa.tipo)
clipes_apos_a_divisao = self.mapeador.clipes_do_arquivo(
next(f for f in self._timeline_atual().faixas if f.identificador == faixa.identificador),
arquivo_de_origem,
)
alvo = [
clipe for clipe in clipes_apos_a_divisao
if inicio_na_timeline - _TOLERANCIA_DE_BORDA_EM_SEGUNDOS <= clipe.intervalo_na_timeline.inicio
and clipe.intervalo_na_timeline.fim <= fim_na_timeline + _TOLERANCIA_DE_BORDA_EM_SEGUNDOS
]
duracao_removida = sum(clipe.intervalo_na_timeline.duracao for clipe in alvo)
duracao_esperada = fim_na_timeline - inicio_na_timeline
if abs(duracao_removida - duracao_esperada) > _TOLERANCIA_DE_BORDA_EM_SEGUNDOS:
raise ErroDeMapeamentoDeTempo(
f"Na faixa {faixa.identificador!r}, os cortes em {inicio_na_timeline:.3f}s e "
f"{fim_na_timeline:.3f}s não isolaram o trecho esperado (esperado {duracao_esperada:.3f}s, "
f"encontrado {duracao_removida:.3f}s cobertos por clipe); nada foi removido para evitar "
"apagar o trecho errado."
)
for clipe in alvo:
self.escrita.remover_trecho(clipe.identificador)
return len(alvo)
def _instante_na_timeline(self, instante_de_origem: float, arquivo_de_origem: str) -> float:
"""Traduz um instante de origem usando a primeira faixa de vídeo do arquivo."""
faixa_de_video = self._faixa_de_video(self._timeline_atual())
clipes = self.mapeador.clipes_do_arquivo(faixa_de_video, arquivo_de_origem)
clipe = self.mapeador.localizar_clipe_no_instante(clipes, instante_de_origem)
if clipe is None:
raise ErroDeMapeamentoDeTempo(
f"Nenhum clipe de vídeo de {arquivo_de_origem!r} cobre o instante {instante_de_origem}s de origem."
)
return self.mapeador.instante_na_timeline(clipe, instante_de_origem)
def _borda_de_entrada(self, clipe: Clipe, inicio_de_origem: float) -> float:
"""Posição de entrada do corte na timeline: a borda do clipe se o corte começa antes dele."""
if inicio_de_origem <= clipe.intervalo_na_origem.inicio:
return clipe.intervalo_na_timeline.inicio
return self.mapeador.instante_na_timeline(clipe, inicio_de_origem)
def _borda_de_saida(self, clipe: Clipe, fim_de_origem: float) -> float:
"""Posição de saída do corte na timeline: a borda do clipe se o corte termina depois dele."""
if fim_de_origem >= clipe.intervalo_na_origem.fim:
return clipe.intervalo_na_timeline.fim
return self.mapeador.instante_na_timeline(clipe, fim_de_origem)
@staticmethod
def _sobrepoe(clipe: Clipe, inicio: float, fim: float) -> bool:
"""``True`` se o intervalo de origem do clipe tiver interseção com [inicio, fim)."""
intervalo = clipe.intervalo_na_origem
return intervalo is not None and intervalo.inicio < fim and inicio < intervalo.fim
@staticmethod
def _faixa_de_video(timeline: Timeline) -> Faixa:
"""A primeira faixa de vídeo da timeline; é onde marcadores e zooms são resolvidos."""
for faixa in timeline.faixas:
if faixa.tipo == "video":
return faixa
raise ErroDeMapeamentoDeTempo("A sequência ativa não tem nenhuma faixa de vídeo.")
+22
View File
@@ -0,0 +1,22 @@
"""Erros do módulo de aplicação de planos de edição.
Cada classe cobre uma fase da aplicação (leitura do plano, mapeamento de
tempo, escrita no editor), para que quem chama saiba exatamente em qual
etapa a operação falhou sem precisar inspecionar mensagens de texto.
"""
class ErroDeEdicao(Exception):
"""Erro geral da aplicação de um plano de edição."""
class ErroDePlanoInvalido(ErroDeEdicao):
"""O arquivo de ações não segue o formato esperado (kind/start/end/params/reason)."""
class ErroDeMapeamentoDeTempo(ErroDeEdicao):
"""Não foi possível traduzir um tempo do arquivo de origem para a timeline ativa."""
class ErroDeEscritaNoEditor(ErroDeEdicao):
"""O Premiere recusou ou não confirmou uma mutação solicitada na timeline."""
+5
View File
@@ -0,0 +1,5 @@
"""Escrita de mutações na sequência ativa do Premiere."""
from .escrita_no_editor import EscritaNoEditor
__all__ = ["EscritaNoEditor"]
@@ -0,0 +1,91 @@
"""Escrita de mutações na sequência ativa do Premiere via MCP.
Isola o nome das ferramentas MCP e o formato de argumentos externos
(``split_clip``, ``remove_from_timeline``, ``set_clip_properties``,
``add_marker``) do resto do domínio. Quem chama esta classe fala só em
segundos de timeline, identificador de clipe e fator de escala — nunca em
nomes de ferramenta ou ticks do Premiere.
"""
from __future__ import annotations
from ...integracoes.premiere.cliente_mcp import ClienteMCP
from ...integracoes.premiere.erros_mcp import ErroDeFerramentaMCP
from ..erros import ErroDeEscritaNoEditor
_CORES_POR_NOME = {
"revisar": 3, # laranja: chama atenção sem parecer um erro (vermelho)
"informar": 5, # branco: marcador neutro
}
class EscritaNoEditor:
"""Aplica, uma de cada vez, as mutações que o aplicador de plano decide fazer."""
def __init__(self, cliente_mcp: ClienteMCP) -> None:
self.cliente_mcp = cliente_mcp
def dividir_clipe_se_necessario(self, instante_da_timeline: float, indice_da_faixa: int, tipo_da_faixa: str) -> bool:
"""Corta, na timeline, o clipe que cobrir ``instante_da_timeline``.
Devolve ``True`` se um corte novo foi criado e ``False`` quando o
instante já era uma borda de clipe (o Premiere recusa dividir onde
não há clipe atravessando o ponto) — esse segundo caso não é um erro,
é o resultado esperado quando o corte pedido coincide com um corte já
existente da análise de cenas.
"""
try:
self.cliente_mcp.chamar(
"split_clip",
{"time_seconds": instante_da_timeline, "track_index": indice_da_faixa, "track_type": tipo_da_faixa},
)
return True
except ErroDeFerramentaMCP:
return False
def remover_trecho(self, identificador_do_clipe: str) -> None:
"""Remove o clipe da timeline, fechando o espaço deixado (ripple).
Levanta :class:`ErroDeEscritaNoEditor` se o Premiere recusar a
remoção.
"""
try:
self.cliente_mcp.chamar("remove_from_timeline", {"node_id": identificador_do_clipe, "ripple": True})
except ErroDeFerramentaMCP as erro:
raise ErroDeEscritaNoEditor(f"Não foi possível remover o clipe {identificador_do_clipe!r}: {erro}") from erro
def aplicar_zoom(self, identificador_do_clipe: str, fator_de_escala: float) -> None:
"""Aplica um punch-in no clipe, escalando-o por ``fator_de_escala`` (1.0 = tamanho original).
Levanta :class:`ErroDeEscritaNoEditor` se o Premiere recusar a
alteração.
"""
try:
self.cliente_mcp.chamar(
"set_clip_properties", {"node_id": identificador_do_clipe, "scale": fator_de_escala * 100}
)
except ErroDeFerramentaMCP as erro:
raise ErroDeEscritaNoEditor(
f"Não foi possível aplicar zoom no clipe {identificador_do_clipe!r}: {erro}"
) from erro
def adicionar_marcador(self, instante_da_timeline: float, nome: str, comentario: str) -> None:
"""Cria um marcador de sequência em ``instante_da_timeline``.
Levanta :class:`ErroDeEscritaNoEditor` se o Premiere recusar a
criação do marcador.
"""
try:
self.cliente_mcp.chamar(
"add_marker",
{
"time_seconds": instante_da_timeline,
"name": nome,
"comments": comentario,
"color": _CORES_POR_NOME["revisar"],
},
)
except ErroDeFerramentaMCP as erro:
raise ErroDeEscritaNoEditor(
f"Não foi possível adicionar o marcador em {instante_da_timeline}s: {erro}"
) from erro
+5
View File
@@ -0,0 +1,5 @@
"""Leitura e validação de planos de edição."""
from .leitor_de_plano import LeitorDePlanoDeEdicao
__all__ = ["LeitorDePlanoDeEdicao"]
@@ -0,0 +1,88 @@
"""Leitura e validação do arquivo JSON de plano de edição.
Isola o formato externo (contrato ``kind``/``start``/``end``/``params``/
``reason`` das skills de seleção de trechos) do modelo interno em
:mod:`engine.editor.modelos`. Nenhuma outra parte do sistema deve interpretar
esse JSON diretamente.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from ..erros import ErroDePlanoInvalido
from ..modelos import AcaoDeEdicao, PlanoDeEdicao, TipoDeAcao
_CAMPOS_OBRIGATORIOS_DA_ACAO = ("kind", "start", "end", "reason")
class LeitorDePlanoDeEdicao:
"""Lê um arquivo de ações e devolve um :class:`PlanoDeEdicao` validado.
Não aplica nenhuma ação nem acessa o Premiere: apenas valida a estrutura
do arquivo e converte para os tipos internos do domínio.
"""
def ler(self, caminho: Path) -> PlanoDeEdicao:
"""Lê o arquivo em ``caminho`` e devolve o plano validado.
Levanta :class:`ErroDePlanoInvalido` se o arquivo não existir, não for
JSON válido ou não seguir o formato esperado.
"""
try:
texto = caminho.read_text(encoding="utf-8")
except OSError as erro:
raise ErroDePlanoInvalido(f"Não foi possível ler o plano de edição em {caminho}: {erro}") from erro
try:
dados = json.loads(texto)
except json.JSONDecodeError as erro:
raise ErroDePlanoInvalido(f"O plano de edição em {caminho} não é um JSON válido: {erro}") from erro
return self.ler_de_dicionario(dados)
def ler_de_dicionario(self, dados: Any) -> PlanoDeEdicao:
"""Converte um dicionário já carregado (ex.: de teste) em um plano validado."""
if not isinstance(dados, dict):
raise ErroDePlanoInvalido("O plano de edição precisa ser um objeto JSON no nível raiz.")
arquivo_de_origem = dados.get("source")
if not isinstance(arquivo_de_origem, str) or not arquivo_de_origem.strip():
raise ErroDePlanoInvalido("O plano de edição precisa de um campo 'source' com o nome do arquivo.")
acoes_brutas = dados.get("actions")
if not isinstance(acoes_brutas, list) or not acoes_brutas:
raise ErroDePlanoInvalido("O plano de edição precisa de uma lista não vazia em 'actions'.")
acoes = tuple(self._converter_acao(bruta, indice) for indice, bruta in enumerate(acoes_brutas))
try:
return PlanoDeEdicao(arquivo_de_origem=arquivo_de_origem, acoes=acoes)
except ValueError as erro:
raise ErroDePlanoInvalido(str(erro)) from erro
def _converter_acao(self, bruta: Any, indice: int) -> AcaoDeEdicao:
"""Converte um item de ``actions`` para :class:`AcaoDeEdicao`, com erros que apontam o índice."""
if not isinstance(bruta, dict):
raise ErroDePlanoInvalido(f"actions[{indice}] precisa ser um objeto JSON.")
faltando = [campo for campo in _CAMPOS_OBRIGATORIOS_DA_ACAO if campo not in bruta]
if faltando:
raise ErroDePlanoInvalido(f"actions[{indice}] está sem os campos obrigatórios: {', '.join(faltando)}.")
try:
tipo = TipoDeAcao(bruta["kind"])
except ValueError as erro:
tipos_validos = ", ".join(tipo.value for tipo in TipoDeAcao)
raise ErroDePlanoInvalido(
f"actions[{indice}].kind = {bruta['kind']!r} é desconhecido. Válidos: {tipos_validos}."
) from erro
try:
inicio = float(bruta["start"])
fim = float(bruta["end"])
except (TypeError, ValueError) as erro:
raise ErroDePlanoInvalido(f"actions[{indice}].start/end precisam ser numéricos.") from erro
parametros = bruta.get("params", {})
if not isinstance(parametros, dict):
raise ErroDePlanoInvalido(f"actions[{indice}].params precisa ser um objeto JSON quando presente.")
motivo = bruta["reason"]
if not isinstance(motivo, str):
raise ErroDePlanoInvalido(f"actions[{indice}].reason precisa ser texto.")
try:
return AcaoDeEdicao(tipo=tipo, inicio=inicio, fim=fim, motivo=motivo, parametros=dict(parametros))
except ValueError as erro:
raise ErroDePlanoInvalido(f"actions[{indice}]: {erro}") from erro
@@ -0,0 +1,5 @@
"""Tradução de tempos de origem para posições na timeline ativa."""
from .mapeador_de_tempo import MapeadorDeTempoDeOrigemParaTimeline
__all__ = ["MapeadorDeTempoDeOrigemParaTimeline"]
@@ -0,0 +1,81 @@
"""Tradução de tempos do arquivo de origem para posições na timeline ativa.
O plano de edição descreve tudo em segundos do arquivo de origem (regra do
formato: nunca compensar para "depois do corte"). Só a sequência do Premiere,
lida no momento da aplicação, sabe onde cada trecho de origem está
atualmente na timeline — por isso este mapeador nunca usa tempos gravados
durante a análise (ex.: ``inicio_na_timeline`` do banco de análises), sempre
o clipe lido ao vivo via :class:`~engine.dominio.Clipe`.
"""
from __future__ import annotations
from ...dominio import Clipe, Faixa, Timeline
from ..erros import ErroDeMapeamentoDeTempo
_TOLERANCIA_EM_SEGUNDOS = 0.05
class MapeadorDeTempoDeOrigemParaTimeline:
"""Localiza clipes pelo tempo de origem e converte instantes para a timeline."""
def clipes_do_arquivo(self, faixa: Faixa, arquivo_de_origem: str) -> list[Clipe]:
"""Clipes da faixa que pertencem a ``arquivo_de_origem``, ordenados pelo início na origem.
A comparação ignora maiúsculas/minúsculas porque o Premiere e o plano
de edição podem registrar a extensão do arquivo em casos diferentes
(``.MP4`` vs. ``.mp4``).
"""
alvo = arquivo_de_origem.strip().lower()
candidatos = [
clipe for clipe in faixa.clipes
if clipe.intervalo_na_origem is not None and (clipe.nome or "").strip().lower() == alvo
]
return sorted(candidatos, key=lambda clipe: clipe.intervalo_na_origem.inicio)
def localizar_clipe_no_instante(self, clipes: list[Clipe], instante_de_origem: float) -> Clipe | None:
"""Devolve o clipe cujo intervalo de origem contém ``instante_de_origem``, ou ``None``.
Tolera pequenas diferenças de arredondamento (meio quadro) nas bordas
do intervalo.
"""
for clipe in clipes:
intervalo = clipe.intervalo_na_origem
if intervalo is None:
continue
if intervalo.inicio - _TOLERANCIA_EM_SEGUNDOS <= instante_de_origem <= intervalo.fim + _TOLERANCIA_EM_SEGUNDOS:
return clipe
return None
def instante_na_timeline(self, clipe: Clipe, instante_de_origem: float) -> float:
"""Converte um instante de origem, dentro do intervalo de origem de ``clipe``, para a timeline.
Levanta :class:`ErroDeMapeamentoDeTempo` se ``clipe`` não tiver
intervalo de origem conhecido — não há como calcular o deslocamento
sem ele.
"""
if clipe.intervalo_na_origem is None:
raise ErroDeMapeamentoDeTempo(
f"O clipe {clipe.identificador!r} não tem intervalo de origem conhecido; "
"não é possível traduzir o tempo para a timeline."
)
deslocamento = instante_de_origem - clipe.intervalo_na_origem.inicio
return clipe.intervalo_na_timeline.inicio + deslocamento
def instante_na_timeline_por_arquivo(
self, timeline: Timeline, faixa: Faixa, arquivo_de_origem: str, instante_de_origem: float
) -> float:
"""Atalho que localiza o clipe do instante em ``faixa`` e já devolve a posição na timeline.
Levanta :class:`ErroDeMapeamentoDeTempo` se nenhum clipe de
``arquivo_de_origem`` cobrir ``instante_de_origem`` nessa faixa.
"""
del timeline # mantido na assinatura para leitura clara no local de chamada
clipes = self.clipes_do_arquivo(faixa, arquivo_de_origem)
clipe = self.localizar_clipe_no_instante(clipes, instante_de_origem)
if clipe is None:
raise ErroDeMapeamentoDeTempo(
f"Nenhum clipe de {arquivo_de_origem!r} na faixa {faixa.identificador!r} "
f"cobre o instante {instante_de_origem}s de origem."
)
return self.instante_na_timeline(clipe, instante_de_origem)
+69
View File
@@ -0,0 +1,69 @@
"""Modelo de domínio do plano de edição.
Representa, em memória, o contrato JSON produzido pelas skills de seleção de
trechos (``kind``/``start``/``end``/``params``/``reason``) já validado e
convertido para tipos internos. Nenhuma classe aqui conhece o Premiere nem o
formato de arquivo — isso é responsabilidade de ``leitura`` e ``escrita``.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
from typing import Any
class TipoDeAcao(str, Enum):
"""Os quatro tipos de ação que o plano de edição pode descrever."""
CORTE = "cut"
ZOOM = "zoom"
TEXTO = "text"
MARCADOR = "marker"
@dataclass(frozen=True)
class AcaoDeEdicao:
"""Uma decisão de edição sobre um intervalo do arquivo de origem.
``inicio`` e ``fim`` são sempre segundos na mídia original — nunca na
timeline já cortada. ``motivo`` é obrigatório: é o texto que o editor lê
para decidir se aceita a decisão.
"""
tipo: TipoDeAcao
inicio: float
fim: float
motivo: str
parametros: dict[str, Any] = field(default_factory=dict)
def __post_init__(self) -> None:
if self.inicio < 0:
raise ValueError(f"Ação inválida: início {self.inicio}s não pode ser negativo.")
if self.fim <= self.inicio:
raise ValueError(
f"Ação inválida: fim ({self.fim}s) deve ser maior que início ({self.inicio}s)."
)
if not self.motivo.strip():
raise ValueError("Ação inválida: motivo não pode ser vazio.")
@property
def duracao(self) -> float:
"""Duração da ação em segundos, sempre positiva."""
return self.fim - self.inicio
@dataclass(frozen=True)
class PlanoDeEdicao:
"""Conjunto ordenado de ações a aplicar sobre um arquivo de origem."""
arquivo_de_origem: str
acoes: tuple[AcaoDeEdicao, ...]
def __post_init__(self) -> None:
if not self.arquivo_de_origem.strip():
raise ValueError("Plano de edição inválido: arquivo de origem não informado.")
def acoes_do_tipo(self, tipo: TipoDeAcao) -> tuple[AcaoDeEdicao, ...]:
"""Devolve, na ordem original, só as ações do tipo informado."""
return tuple(acao for acao in self.acoes if acao.tipo is tipo)
+180
View File
@@ -0,0 +1,180 @@
"""Relatório dos J-SOM de análise (imagem e som) lidos do banco de análises.
Este módulo lê o banco ``analises.db`` e monta um único documento JSON com as
duas análises já persistidas pelo Scanner:
- **Som**: transcrição por clipe/cena, com falante, emoção e confiança.
- **Imagem**: evidências visuais por clipe (Apple Vision), com resumo por tipo.
É a base da aba **Análises** do painel CEP, que permite abrir e copiar esses
resultados sem precisar de SQL nem de reprocessar o vídeo.
"""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
# O painel chama este arquivo com cwd ``code/engine``; injetar ``code`` (pai do
# pacote ``engine``) para que os importes funcionem, como em ``executar_retakes.py``.
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.consultas import ConsultasDeAnalises # noqa: E402
class ErroDeBancoSemVideos(RuntimeError):
"""Representa um banco de análises que ainda não tem nenhum vídeo."""
class MontadorDeRelatorioDeAnalises:
"""
Monta o relatório JSON dos J-SOM de análise a partir do banco de análises.
Reaproveita ``ConsultasDeAnalises`` (somente leitura) e organiza os dados
por clipe/cena, unindo a transcrição (som) e as evidências visuais (imagem)
de cada trecho da timeline.
Atributos:
consultas: Consultas de leitura sobre o banco de análises.
"""
def __init__(self, banco: str | Path) -> None:
"""
Inicializa o montador sobre um banco de análises existente.
Parâmetros:
banco: Caminho do arquivo ``analises.db``.
"""
self.consultas = ConsultasDeAnalises(banco)
def montar(self, video_id: str | None = None) -> dict:
"""
Monta o relatório completo de um vídeo (ou do primeiro vídeo do banco).
Parâmetros:
video_id: Identificador do vídeo. Quando None, usa o primeiro
vídeo encontrado no banco.
Retorna:
Dicionário com o resumo do vídeo e a lista de cenas, cada uma com
as falas (som) e as evidências visuais (imagem) daquele clipe.
Pode gerar:
ErroDeBancoSemVideos: quando o banco não tem nenhum vídeo.
"""
video_id = video_id or self._primeiro_video()
resumo = self.consultas.consultar_resumo_do_video(video_id)
if resumo is None:
raise ErroDeBancoSemVideos(f"Vídeo '{video_id}' não existe no banco de análises.")
falas = self.consultas.listar_falas_no_intervalo(video_id, 0.0, None)
return {
"video": resumo,
"cenas": self._cenas_do_relatorio(video_id, falas),
}
def _primeiro_video(self) -> str:
"""Devolve o identificador do primeiro vídeo do banco."""
linha = self.consultas.conexao.execute(
"SELECT id FROM videos ORDER BY criado_em LIMIT 1").fetchone()
if linha is None:
raise ErroDeBancoSemVideos(
"O banco de análises não tem nenhum vídeo. Execute o Scanner primeiro.")
return linha["id"]
def _cenas_do_relatorio(self, video_id: str, falas: list[dict]) -> list[dict]:
"""
Reúne, por clipe, as falas e as evidências visuais do vídeo.
A base é a tabela ``clipes`` (timeline completa), porque a tabela
``cenas`` pode cobrir só parte dos clipes; o campo
``cena_registrada`` indica se aquele clipe tem corte de cena gravado.
Parâmetros:
video_id: Identificador do vídeo.
falas: Falas já consultadas (saída de ``listar_falas_no_intervalo``).
Retorna:
Lista de entradas por clipe, ordenadas pela posição na timeline.
"""
falas_por_clipe: dict[str, list[dict]] = {}
for fala in falas:
falas_por_clipe.setdefault(fala["clipe_id"], []).append(fala)
clipes_com_cena = {
linha["clipe_id"] for linha in self.consultas.conexao.execute(
"SELECT DISTINCT clipe_id FROM cenas WHERE video_id = ? AND clipe_id IS NOT NULL",
(video_id,),
).fetchall()
}
cenas: list[dict] = []
for clipe in self.consultas.conexao.execute(
"""SELECT id AS clipe_id, nome, inicio_na_timeline AS inicio,
fim_na_timeline AS fim FROM clipes
WHERE video_id = ? ORDER BY inicio_na_timeline""",
(video_id,),
).fetchall():
clipe_id = clipe["clipe_id"]
evidencias = self.consultas.listar_evidencias_visuais_do_clipe(video_id, clipe_id)
cenas.append({
"clipe_id": clipe_id,
"nome": clipe["nome"],
"inicio": clipe["inicio"],
"fim": clipe["fim"],
"cena_registrada": clipe_id in clipes_com_cena,
"falas": falas_por_clipe.get(clipe_id, []),
"resumo_de_evidencias": self._resumo_de_evidencias(evidencias),
"evidencias_visuais": evidencias,
})
return cenas
@staticmethod
def _resumo_de_evidencias(evidencias: list[dict]) -> dict:
"""Conta as evidências visuais por tipo (ex.: ``rosto: 2``)."""
resumo: dict[str, int] = {}
for evidencia in evidencias:
resumo[evidencia["tipo"]] = resumo.get(evidencia["tipo"], 0) + 1
return resumo
def executar(banco: Path, saida: Path, video_id: str | None = None) -> dict:
"""
Gera o arquivo JSON do relatório e devolve o documento montado.
Parâmetros:
banco: Caminho do arquivo ``analises.db``.
saida: Caminho do arquivo JSON a gravar.
video_id: Vídeo desejado; quando None, usa o primeiro do banco.
Retorna:
O dicionário do relatório (mesmo conteúdo gravado em ``saida``).
"""
relatorio = MontadorDeRelatorioDeAnalises(banco).montar(video_id)
saida.parent.mkdir(parents=True, exist_ok=True)
saida.write_text(json.dumps(relatorio, ensure_ascii=False, indent=2), encoding="utf-8")
return relatorio
def main() -> None:
"""Ponto de entrada da linha de comando usada pelo painel."""
parser = argparse.ArgumentParser(
description="Gera o relatório JSON dos J-SOM de análise (som e imagem) do banco.")
parser.add_argument("banco", type=Path, help="Caminho do analises.db.")
parser.add_argument("saida", type=Path, help="Arquivo JSON de saída.")
parser.add_argument("--video-id", default=None, help="Vídeo desejado (opcional).")
args = parser.parse_args()
try:
relatorio = executar(args.banco, args.saida, args.video_id)
except Exception as erro: # repassa a causa para a tela decidir como exibir
print(json.dumps({"evento": "erro", "mensagem": str(erro)}, ensure_ascii=False), flush=True)
sys.exit(1)
print(json.dumps({"evento": "concluido", "arquivo": str(args.saida),
"video": relatorio["video"]["nome"]}, ensure_ascii=False), flush=True)
if __name__ == "__main__":
main()
@@ -1,23 +1,43 @@
"""Cliente MCP (Model Context Protocol) para o servidor local do Premiere.
Fala o protocolo JSON-RPC do MCP por stdio com o processo Node do bridge
(``code/dist/index.js``). É a única camada do sistema que conhece o
transporte (subprocesso, linhas JSON, ids de requisição); o resto do código
usa só o método :meth:`ClienteMCP.chamar` com nome de ferramenta e
argumentos.
"""
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
import json import json
import subprocess import subprocess
from typing import Any from typing import Any
from .erros_mcp import ErroDeConexaoMCP, ErroDeFerramentaMCP
class ClienteMCP(ABC): class ClienteMCP(ABC):
"""Define a comunicação técnica com o servidor MCP.""" """Define a comunicação técnica com o servidor MCP."""
@abstractmethod @abstractmethod
def conectar(self) -> None: ... def conectar(self) -> None:
"""Inicia a conexão com o servidor MCP. Não faz nada se já estiver conectado."""
@abstractmethod @abstractmethod
def desconectar(self) -> None: ... def desconectar(self) -> None:
"""Encerra a conexão com o servidor MCP, se houver uma ativa."""
@abstractmethod @abstractmethod
def esta_conectado(self) -> bool: ... def esta_conectado(self) -> bool:
"""Indica se há uma conexão ativa com o servidor MCP."""
@abstractmethod @abstractmethod
def chamar(self, nome_da_ferramenta: str, argumentos: dict[str, Any] | None = None) -> dict[str, Any]: ... def chamar(self, nome_da_ferramenta: str, argumentos: dict[str, Any] | None = None) -> dict[str, Any]:
"""Executa uma ferramenta do servidor MCP e devolve o resultado.
Levanta :class:`~.erros_mcp.ErroDeConexaoMCP` se não houver conexão
ativa e :class:`~.erros_mcp.ErroDeFerramentaMCP` se o servidor
responder com erro.
"""
class ClienteMCPPorStdio(ClienteMCP): class ClienteMCPPorStdio(ClienteMCP):
@@ -30,6 +50,11 @@ class ClienteMCPPorStdio(ClienteMCP):
self._proximo_id = 1 self._proximo_id = 1
def conectar(self) -> None: def conectar(self) -> None:
"""Sobe o processo do servidor MCP e envia o handshake ``initialize``.
Não faz nada se um processo já estiver em execução — não é possível
reconectar sem antes chamar :meth:`desconectar`.
"""
if self._processo is not None: if self._processo is not None:
return return
self._processo = subprocess.Popen( self._processo = subprocess.Popen(
@@ -43,29 +68,42 @@ class ClienteMCPPorStdio(ClienteMCP):
self._enviar("initialize", {"protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "engine", "version": "0.1.0"}}) self._enviar("initialize", {"protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "engine", "version": "0.1.0"}})
def desconectar(self) -> None: def desconectar(self) -> None:
"""Encerra o processo do servidor MCP, se houver um em execução."""
if self._processo is not None: if self._processo is not None:
self._processo.terminate() self._processo.terminate()
self._processo = None self._processo = None
def esta_conectado(self) -> bool: def esta_conectado(self) -> bool:
"""``True`` enquanto o processo do servidor MCP estiver vivo."""
return self._processo is not None and self._processo.poll() is None return self._processo is not None and self._processo.poll() is None
def chamar(self, nome_da_ferramenta: str, argumentos: dict[str, Any] | None = None) -> dict[str, Any]: def chamar(self, nome_da_ferramenta: str, argumentos: dict[str, Any] | None = None) -> dict[str, Any]:
"""Executa ``nome_da_ferramenta`` no servidor MCP com ``argumentos``.
Levanta :class:`~.erros_mcp.ErroDeConexaoMCP` se não houver conexão
ativa e :class:`~.erros_mcp.ErroDeFerramentaMCP` se o servidor
responder com um erro para essa ferramenta.
"""
if not self.esta_conectado(): if not self.esta_conectado():
raise RuntimeError("Cliente MCP não está conectado.") raise ErroDeConexaoMCP("Cliente MCP não está conectado.")
resposta = self._enviar("tools/call", {"name": nome_da_ferramenta, "arguments": argumentos or {}}) resposta = self._enviar("tools/call", {"name": nome_da_ferramenta, "arguments": argumentos or {}})
if resposta.get("error"): if resposta.get("error"):
raise RuntimeError(f"Erro MCP: {resposta['error']}") raise ErroDeFerramentaMCP(f"Erro ao executar {nome_da_ferramenta!r} no MCP: {resposta['error']}")
return resposta.get("result", {}) return resposta.get("result", {})
def _enviar(self, metodo: str, parametros: dict[str, Any]) -> dict[str, Any]: def _enviar(self, metodo: str, parametros: dict[str, Any]) -> dict[str, Any]:
"""Envia uma requisição JSON-RPC pela stdin do processo e lê a resposta da stdout.
Levanta :class:`~.erros_mcp.ErroDeConexaoMCP` se o processo não
estiver disponível ou encerrar sem responder.
"""
if self._processo is None or self._processo.stdin is None or self._processo.stdout is None: if self._processo is None or self._processo.stdin is None or self._processo.stdout is None:
raise RuntimeError("Processo MCP indisponível.") raise ErroDeConexaoMCP("Processo MCP indisponível.")
identificador = self._proximo_id identificador = self._proximo_id
self._proximo_id += 1 self._proximo_id += 1
self._processo.stdin.write(json.dumps({"jsonrpc": "2.0", "id": identificador, "method": metodo, "params": parametros}) + "\n") self._processo.stdin.write(json.dumps({"jsonrpc": "2.0", "id": identificador, "method": metodo, "params": parametros}) + "\n")
self._processo.stdin.flush() self._processo.stdin.flush()
linha = self._processo.stdout.readline() linha = self._processo.stdout.readline()
if not linha: if not linha:
raise RuntimeError("O MCP encerrou sem retornar resposta.") raise ErroDeConexaoMCP("O MCP encerrou sem retornar resposta.")
return json.loads(linha) return json.loads(linha)
+182
View File
@@ -0,0 +1,182 @@
"""Dublê de teste do cliente MCP do Premiere, para testar o editor sem o app aberto.
Simula, em memória, uma sequência com faixas de vídeo e áudio cujos clipes
têm posição na timeline diferente da posição no arquivo de origem — o mesmo
descompasso encontrado no projeto real entre ``inicio_na_timeline`` e
``inicio_na_origem`` do banco de análises. Implementa só as ferramentas MCP
que o módulo ``engine.editor`` usa: ``get_active_sequence``, ``split_clip``,
``remove_from_timeline``, ``set_clip_properties`` e ``add_marker``.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from engine.integracoes.premiere.cliente_mcp import ClienteMCP
from engine.integracoes.premiere.erros_mcp import ErroDeFerramentaMCP
@dataclass
class _ClipeFalso:
"""Representação mínima de um clipe dentro do dublê de sequência."""
identificador: str
nome: str
inicio_na_timeline: float
fim_na_timeline: float
inicio_na_origem: float
fim_na_origem: float
class ClienteMCPFalso(ClienteMCP):
"""Simula o servidor MCP do Premiere para testes do módulo ``engine.editor``.
Vem pré-carregado com duas faixas (``video_0`` e ``audio_0``), cada uma
com dois clipes do mesmo arquivo fictício, reproduzindo o descompasso
real entre tempo de origem e tempo de timeline. Registra toda chamada em
:attr:`chamadas` para os testes inspecionarem ordem e argumentos.
"""
_ARQUIVO = "0E6A8290.mp4"
def __init__(self) -> None:
self.chamadas: list[tuple[str, dict[str, Any]]] = []
self._proximo_id = 0
self._faixas: dict[str, list[_ClipeFalso]] = {
"video_0": [
self._novo_clipe(self._ARQUIVO, 0.0, 38.3, 2.3, 40.6, "clipe_a_video"),
self._novo_clipe(self._ARQUIVO, 58.1, 66.7, 86.2, 94.8, "clipe_b_video"),
],
"audio_0": [
self._novo_clipe(self._ARQUIVO, 0.0, 38.3, 2.3, 40.6, "clipe_a_audio"),
self._novo_clipe(self._ARQUIVO, 58.1, 66.7, 86.2, 94.8, "clipe_b_audio"),
],
}
@property
def chamadas_de_remocao(self) -> list[dict[str, Any]]:
"""Argumentos de cada chamada a ``remove_from_timeline``, na ordem em que ocorreram."""
return [argumentos for nome, argumentos in self.chamadas if nome == "remove_from_timeline"]
def conectar(self) -> None:
"""Não há conexão real; existe só para satisfazer a interface de :class:`ClienteMCP`."""
def desconectar(self) -> None:
"""Não há conexão real; existe só para satisfazer a interface de :class:`ClienteMCP`."""
def esta_conectado(self) -> bool:
"""Sempre conectado: este dublê não simula falha de conexão."""
return True
def chamar(self, nome_da_ferramenta: str, argumentos: dict[str, Any] | None = None) -> dict[str, Any]:
"""Despacha para o simulador da ferramenta pedida e registra a chamada."""
argumentos = argumentos or {}
self.chamadas.append((nome_da_ferramenta, argumentos))
despachantes = {
"get_active_sequence": self._simular_get_active_sequence,
"split_clip": self._simular_split_clip,
"remove_from_timeline": self._simular_remove_from_timeline,
"set_clip_properties": self._simular_set_clip_properties,
"add_marker": self._simular_add_marker,
}
simulador = despachantes.get(nome_da_ferramenta)
if simulador is None:
raise ErroDeFerramentaMCP(f"Ferramenta MCP não simulada neste dublê de teste: {nome_da_ferramenta!r}.")
return simulador(argumentos)
def _novo_clipe(self, nome, tl_inicio, tl_fim, origem_inicio, origem_fim, identificador=None) -> _ClipeFalso:
"""Cria um clipe falso com um identificador único (ou o informado, para os fixtures)."""
if identificador is None:
self._proximo_id += 1
identificador = f"clipe_falso_{self._proximo_id}"
return _ClipeFalso(identificador, nome, tl_inicio, tl_fim, origem_inicio, origem_fim)
def _simular_get_active_sequence(self, argumentos: dict[str, Any]) -> dict[str, Any]:
"""Devolve a sequência atual no formato que :class:`ConversorDeTimeline` espera."""
del argumentos
def _faixa_como_dict(identificador: str, tipo: str, indice: int) -> dict[str, Any]:
return {
"id": identificador,
"name": identificador,
"index": indice,
"clips": [
{
"nodeId": clipe.identificador,
"name": clipe.nome,
"start": clipe.inicio_na_timeline,
"end": clipe.fim_na_timeline,
"inPoint": clipe.inicio_na_origem,
"outPoint": clipe.fim_na_origem,
}
for clipe in self._faixas[identificador]
],
}
dados = {
"id": "sequencia_falsa",
"name": "Sequência de teste",
"duration": 200.0,
"frameRate": 29.97,
"frameSizeHorizontal": 3840,
"frameSizeVertical": 2160,
"videoTracks": [_faixa_como_dict("video_0", "video", 0)],
"audioTracks": [_faixa_como_dict("audio_0", "audio", 0)],
}
return {"structuredContent": {"data": dados}}
def _simular_split_clip(self, argumentos: dict[str, Any]) -> dict[str, Any]:
"""Divide, na faixa pedida, o clipe que cobrir estritamente o instante informado."""
identificador_da_faixa = self._identificador_da_faixa(argumentos["track_index"], argumentos["track_type"])
instante = float(argumentos["time_seconds"])
faixa = self._faixas[identificador_da_faixa]
for posicao, clipe in enumerate(faixa):
if clipe.inicio_na_timeline < instante < clipe.fim_na_timeline:
deslocamento = instante - clipe.inicio_na_timeline
self._proximo_id += 1
parte_esquerda = _ClipeFalso(
f"{clipe.identificador}_esq{self._proximo_id}", clipe.nome,
clipe.inicio_na_timeline, instante,
clipe.inicio_na_origem, clipe.inicio_na_origem + deslocamento,
)
self._proximo_id += 1
parte_direita = _ClipeFalso(
f"{clipe.identificador}_dir{self._proximo_id}", clipe.nome,
instante, clipe.fim_na_timeline,
clipe.inicio_na_origem + deslocamento, clipe.fim_na_origem,
)
faixa[posicao:posicao + 1] = [parte_esquerda, parte_direita]
return {"split": True}
raise ErroDeFerramentaMCP(f"Nenhum clipe da faixa {identificador_da_faixa!r} cobre estritamente {instante}s.")
def _simular_remove_from_timeline(self, argumentos: dict[str, Any]) -> dict[str, Any]:
"""Remove o clipe pedido e, com ``ripple``, fecha o espaço deixado na mesma faixa."""
identificador_do_clipe = argumentos["node_id"]
for identificador_da_faixa, clipes in self._faixas.items():
for posicao, clipe in enumerate(clipes):
if clipe.identificador == identificador_do_clipe:
duracao = clipe.fim_na_timeline - clipe.inicio_na_timeline
del clipes[posicao]
if argumentos.get("ripple"):
for outro in clipes:
if outro.inicio_na_timeline >= clipe.fim_na_timeline:
outro.inicio_na_timeline -= duracao
outro.fim_na_timeline -= duracao
return {"removed": True}
raise ErroDeFerramentaMCP(f"Clipe não encontrado para remoção: {identificador_do_clipe!r}.")
def _simular_set_clip_properties(self, argumentos: dict[str, Any]) -> dict[str, Any]:
"""Aceita a alteração de propriedades sem simular geometria (não é usada pelos testes)."""
del argumentos
return {"updated": True}
def _simular_add_marker(self, argumentos: dict[str, Any]) -> dict[str, Any]:
"""Aceita a criação de marcador sem manter estado (a chamada já fica em :attr:`chamadas`)."""
del argumentos
return {"added": True}
@staticmethod
def _identificador_da_faixa(indice: int, tipo: str) -> str:
"""Reconstrói o identificador de faixa (``video_0``/``audio_0``) a partir de índice e tipo."""
return f"{tipo}_{indice}"
@@ -0,0 +1,97 @@
"""Testes do orquestrador de aplicação de plano de edição (sem Premiere real).
Usa um ClienteMCP falso, em memória, que simula uma sequência de vídeo com
os mesmos tempos "torcidos" do projeto real: a timeline lida pelo scanner
não bate com o tempo do arquivo de origem (ver relatório da análise), então
os testes aqui garantem que o aplicador sempre recalcula a posição a partir
do clipe lido ao vivo, nunca de um tempo fixo.
"""
import unittest
from typing import Any
from engine.editor.aplicador_de_plano_de_edicao import AplicadorDePlanoDeEdicao
from engine.editor.escrita import EscritaNoEditor
from engine.editor.mapeamento import MapeadorDeTempoDeOrigemParaTimeline
from engine.editor.modelos import AcaoDeEdicao, PlanoDeEdicao, TipoDeAcao
from engine.integracoes.premiere.conversores import ConversorDeTimeline
from engine.integracoes.premiere.leitura import AcessoAoEditor, AcessoATimeline
from engine.testes.duplos_de_premiere import ClienteMCPFalso
class TesteAplicadorDePlanoDeEdicao(unittest.TestCase):
"""Cobre a ordem de aplicação e o recálculo de posições a cada ação."""
def setUp(self):
self.cliente = ClienteMCPFalso()
self.aplicador = AplicadorDePlanoDeEdicao(
acesso_ao_editor=AcessoAoEditor(AcessoATimeline(self.cliente)),
conversor_de_timeline=ConversorDeTimeline(),
escrita=EscritaNoEditor(self.cliente),
mapeador=MapeadorDeTempoDeOrigemParaTimeline(),
)
def test_corte_remove_os_clipes_do_intervalo_e_fecha_o_espaco(self):
plano = PlanoDeEdicao("0E6A8290.mp4", (
AcaoDeEdicao(TipoDeAcao.CORTE, inicio=0.0, fim=40.6, motivo="bastidor"),
))
resultado = self.aplicador.aplicar(plano)
self.assertTrue(resultado.todas_bem_sucedidas)
self.assertNotIn("000f4763_video", [c["node_id"] for c in self.cliente.chamadas_de_remocao])
# o clipe único que cobre 2.3-40.6s de origem foi removido em ambas as faixas
self.assertEqual(
{c["node_id"] for c in self.cliente.chamadas_de_remocao},
{"clipe_a_video", "clipe_a_audio"},
)
def test_marcador_e_zoom_sao_aplicados_antes_dos_cortes(self):
plano = PlanoDeEdicao("0E6A8290.mp4", (
AcaoDeEdicao(TipoDeAcao.CORTE, inicio=0.0, fim=40.6, motivo="bastidor"),
AcaoDeEdicao(TipoDeAcao.MARCADOR, inicio=90.0, fim=90.5, motivo="conferir emenda", parametros={"content": "EMENDA"}),
AcaoDeEdicao(TipoDeAcao.ZOOM, inicio=86.42, fim=91.5, motivo="ênfase", parametros={"scale": 1.15}),
))
resultado = self.aplicador.aplicar(plano)
self.assertTrue(resultado.todas_bem_sucedidas, msg=[r.detalhe for r in resultado.falhas])
# marcador e zoom aconteceram antes do corte na sequência de chamadas
indice_do_marcador = next(i for i, c in enumerate(self.cliente.chamadas) if c[0] == "add_marker")
indice_do_zoom = next(i for i, c in enumerate(self.cliente.chamadas) if c[0] == "set_clip_properties")
indice_da_remocao = next(i for i, c in enumerate(self.cliente.chamadas) if c[0] == "remove_from_timeline")
self.assertLess(indice_do_marcador, indice_da_remocao)
self.assertLess(indice_do_zoom, indice_da_remocao)
def test_dois_cortes_sao_aplicados_do_fim_para_o_comeco(self):
plano = PlanoDeEdicao("0E6A8290.mp4", (
AcaoDeEdicao(TipoDeAcao.CORTE, inicio=0.0, fim=40.6, motivo="primeiro trecho"),
AcaoDeEdicao(TipoDeAcao.CORTE, inicio=86.2, fim=94.8, motivo="segundo trecho"),
))
resultado = self.aplicador.aplicar(plano)
self.assertTrue(resultado.todas_bem_sucedidas, msg=[r.detalhe for r in resultado.falhas])
remocoes = [c["node_id"] for c in self.cliente.chamadas_de_remocao]
# o corte com início maior (86.2s) precisa ser removido antes do de início 0.0s
self.assertLess(remocoes.index("clipe_b_video"), remocoes.index("clipe_a_video"))
def test_corte_sem_clipe_sobreposto_falha_sem_interromper_o_resto(self):
plano = PlanoDeEdicao("0E6A8290.mp4", (
AcaoDeEdicao(TipoDeAcao.CORTE, inicio=1000.0, fim=1010.0, motivo="fora do vídeo"),
AcaoDeEdicao(TipoDeAcao.MARCADOR, inicio=90.0, fim=90.5, motivo="ok", parametros={"content": "X"}),
))
resultado = self.aplicador.aplicar(plano)
self.assertFalse(resultado.todas_bem_sucedidas)
self.assertEqual(len(resultado.falhas), 1)
self.assertIn("Nenhum clipe", resultado.falhas[0].detalhe)
# a ação válida (marcador) ainda foi aplicada
self.assertTrue(any(c[0] == "add_marker" for c in self.cliente.chamadas))
def test_texto_e_convertido_em_marcador_por_falta_de_suporte_do_premiere(self):
plano = PlanoDeEdicao("0E6A8290.mp4", (
AcaoDeEdicao(TipoDeAcao.TEXTO, inicio=90.0, fim=91.0, motivo="reforçar termo", parametros={"content": "MASTOPEXIA"}),
))
resultado = self.aplicador.aplicar(plano)
self.assertTrue(resultado.todas_bem_sucedidas)
chamadas_de_marcador = [c for c in self.cliente.chamadas if c[0] == "add_marker"]
self.assertEqual(len(chamadas_de_marcador), 1)
self.assertIn("MASTOPEXIA", chamadas_de_marcador[0][1]["comments"])
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,90 @@
"""Testes do relatório dos J-SOM de análise lido do banco de análises."""
import json
import tempfile
import unittest
from pathlib import Path
from engine.dominio import Clipe, Faixa, IntervaloDeTempo, Timeline
from engine.gerar_relatorio_de_analises import (
ErroDeBancoSemVideos,
MontadorDeRelatorioDeAnalises,
executar,
)
from engine.persistencia import (
RepositorioDeAnalisesSQLite,
RepositorioDeTimelineSQLite,
)
from engine.scanner.modelos import Cena, EvidenciaVisual, SegmentoDeTranscricao
from engine.scanner.transcricao_da_timeline import TranscricaoDoClipe
class TesteMontadorDeRelatorioDeAnalises(unittest.TestCase):
"""Cenário comum: um vídeo, duas cenas (uma duplicada), falas e evidências."""
def _banco_de_exemplo(self, pasta: Path) -> Path:
banco = pasta / "analises.db"
timeline = Timeline(
identificador="v1", nome="Entrevista", duracao=30.0, largura=1920, altura=1080,
faixas=[Faixa("f1", "V1", "video", 0, [
Clipe("c1", "A.mp4", IntervaloDeTempo(0.0, 10.0)),
Clipe("c2", "A.mp4", IntervaloDeTempo(10.0, 30.0)),
])],
)
RepositorioDeTimelineSQLite(banco).registrar_timeline(timeline)
repositorio = RepositorioDeAnalisesSQLite(banco)
repositorio.registrar_transcricoes("v1", [
TranscricaoDoClipe(
identificador_do_clipe="c1", arquivo="A.mp4",
inicio_na_timeline=0.0, fim_na_timeline=10.0,
segmentos=[SegmentoDeTranscricao(0.0, 5.0, "Olá, tudo bem?", emocao="hap")],
),
])
repositorio.registrar_evidencias_visuais("v1", "c1", [
EvidenciaVisual(tipo="rosto", inicio=0.0, fim=5.0, valor={"qualidade": 0.9},
confianca=0.9, provider="apple_vision"),
EvidenciaVisual(tipo="rosto", inicio=5.0, fim=10.0, valor={"qualidade": 0.8},
confianca=0.8, provider="apple_vision"),
])
# Cena c1 gravada duas vezes: o relatório não deve duplicar a entrada.
# c2 fica sem cena registrada: deve aparecer com cena_registrada=False.
repositorio.registrar_cenas("v1", [Cena(0.0, 10.0), Cena(0.0, 10.0)], clipe_id="c1")
return banco
def test_relatorio_unindo_som_e_imagem_por_cena(self):
with tempfile.TemporaryDirectory() as pasta:
banco = self._banco_de_exemplo(Path(pasta))
relatorio = MontadorDeRelatorioDeAnalises(banco).montar()
self.assertEqual(relatorio["video"]["nome"], "Entrevista")
self.assertEqual(len(relatorio["cenas"]), 2) # um item por clipe
cena_um = relatorio["cenas"][0]
self.assertTrue(cena_um["cena_registrada"])
self.assertEqual(cena_um["falas"][0]["texto"], "Olá, tudo bem?")
self.assertEqual(cena_um["falas"][0]["emocao"], "hap")
self.assertEqual(cena_um["resumo_de_evidencias"], {"rosto": 2})
self.assertEqual(cena_um["evidencias_visuais"][0]["provider"], "apple_vision")
cena_dois = relatorio["cenas"][1]
self.assertFalse(cena_dois["cena_registrada"])
self.assertEqual(cena_dois["falas"], [])
def test_executar_grava_json_e_devolve_relatorio(self):
with tempfile.TemporaryDirectory() as pasta:
banco = self._banco_de_exemplo(Path(pasta))
saida = Path(pasta) / "relatorio.json"
relatorio = executar(banco, saida, "v1")
gravado = json.loads(saida.read_text(encoding="utf-8"))
self.assertEqual(gravado, relatorio)
self.assertEqual(gravado["video"]["video_id"], "v1")
def test_banco_sem_videos_gera_erro_especifico(self):
with tempfile.TemporaryDirectory() as pasta:
banco = Path(pasta) / "vazio.db"
RepositorioDeTimelineSQLite(banco) # cria o banco com o esquema
with self.assertRaises(ErroDeBancoSemVideos):
MontadorDeRelatorioDeAnalises(banco).montar()
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,92 @@
"""Testes da leitura e validação do plano de edição (engine.editor.leitura)."""
import tempfile
import unittest
from pathlib import Path
from engine.editor.erros import ErroDePlanoInvalido
from engine.editor.leitura import LeitorDePlanoDeEdicao
from engine.editor.modelos import TipoDeAcao
PLANO_VALIDO = {
"source": "0E6A8290.mp4",
"actions": [
{"kind": "cut", "start": 0.0, "end": 10.0, "reason": "bastidor"},
{"kind": "zoom", "start": 12.0, "end": 15.0, "params": {"scale": 1.2}, "reason": "ênfase"},
],
}
class TesteLeitorDePlanoDeEdicao(unittest.TestCase):
"""Cobre a validação do formato JSON do plano de edição."""
def setUp(self):
self.leitor = LeitorDePlanoDeEdicao()
def test_le_plano_valido_de_dicionario(self):
plano = self.leitor.ler_de_dicionario(PLANO_VALIDO)
self.assertEqual(plano.arquivo_de_origem, "0E6A8290.mp4")
self.assertEqual(len(plano.acoes), 2)
self.assertEqual(plano.acoes[0].tipo, TipoDeAcao.CORTE)
self.assertEqual(plano.acoes[1].parametros, {"scale": 1.2})
def test_le_plano_valido_de_arquivo(self):
with tempfile.TemporaryDirectory() as pasta:
caminho = Path(pasta) / "plano.json"
caminho.write_text(
'{"source": "a.mp4", "actions": [{"kind": "marker", "start": 1, "end": 2, "reason": "ok"}]}',
encoding="utf-8",
)
plano = self.leitor.ler(caminho)
self.assertEqual(plano.arquivo_de_origem, "a.mp4")
def test_arquivo_inexistente_levanta_erro_especifico(self):
with self.assertRaises(ErroDePlanoInvalido):
self.leitor.ler(Path("/caminho/que/nao/existe.json"))
def test_json_malformado_levanta_erro_especifico(self):
with tempfile.TemporaryDirectory() as pasta:
caminho = Path(pasta) / "plano.json"
caminho.write_text("{isso nao é json", encoding="utf-8")
with self.assertRaises(ErroDePlanoInvalido):
self.leitor.ler(caminho)
def test_sem_source_levanta_erro(self):
with self.assertRaises(ErroDePlanoInvalido):
self.leitor.ler_de_dicionario({"actions": PLANO_VALIDO["actions"]})
def test_actions_vazia_levanta_erro(self):
with self.assertRaises(ErroDePlanoInvalido):
self.leitor.ler_de_dicionario({"source": "a.mp4", "actions": []})
def test_kind_desconhecido_levanta_erro(self):
with self.assertRaises(ErroDePlanoInvalido):
self.leitor.ler_de_dicionario({
"source": "a.mp4",
"actions": [{"kind": "voar", "start": 0, "end": 1, "reason": "x"}],
})
def test_fim_menor_ou_igual_ao_inicio_levanta_erro(self):
with self.assertRaises(ErroDePlanoInvalido):
self.leitor.ler_de_dicionario({
"source": "a.mp4",
"actions": [{"kind": "cut", "start": 5, "end": 5, "reason": "x"}],
})
def test_reason_vazio_levanta_erro(self):
with self.assertRaises(ErroDePlanoInvalido):
self.leitor.ler_de_dicionario({
"source": "a.mp4",
"actions": [{"kind": "cut", "start": 0, "end": 1, "reason": " "}],
})
def test_campo_obrigatorio_faltando_levanta_erro(self):
with self.assertRaises(ErroDePlanoInvalido):
self.leitor.ler_de_dicionario({
"source": "a.mp4",
"actions": [{"kind": "cut", "start": 0, "reason": "x"}],
})
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,69 @@
"""Testes do mapeamento de tempos de origem para a timeline (engine.editor.mapeamento)."""
import unittest
from engine.dominio import Clipe, Faixa, IntervaloDeTempo
from engine.editor.erros import ErroDeMapeamentoDeTempo
from engine.editor.mapeamento import MapeadorDeTempoDeOrigemParaTimeline
def _clipe(identificador, tl_inicio, tl_fim, origem_inicio, origem_fim, nome="0E6A8290.MP4"):
"""Monta um Clipe de teste com intervalo na timeline e na origem."""
return Clipe(
identificador=identificador,
nome=nome,
intervalo_na_timeline=IntervaloDeTempo(tl_inicio, tl_fim),
intervalo_na_origem=IntervaloDeTempo(origem_inicio, origem_fim),
)
class TesteMapeadorDeTempoDeOrigemParaTimeline(unittest.TestCase):
"""Cobre a tradução de tempos de origem, reproduzindo o caso real do projeto:
a timeline reindexada pelo detector de cena não bate com o tempo do arquivo."""
def setUp(self):
self.mapeador = MapeadorDeTempoDeOrigemParaTimeline()
# Reproduz o caso real: clip 000f4775 está em 86.2-114.0s de origem,
# mas em 58.1-86.0s na timeline reindexada pelo scanner.
self.faixa = Faixa("video_0", "Vídeo 1", "video", 0, clipes=[
_clipe("000f4763", 0.0, 38.3, 2.3, 40.6),
_clipe("000f4775", 58.1, 86.0, 86.2, 114.0),
_clipe("outro_arquivo", 86.0, 90.0, 0.0, 4.0, nome="outro.mp4"),
])
def test_clipes_do_arquivo_filtra_por_nome_ignorando_caixa(self):
clipes = self.mapeador.clipes_do_arquivo(self.faixa, "0e6a8290.mp4")
self.assertEqual([c.identificador for c in clipes], ["000f4763", "000f4775"])
def test_localizar_clipe_no_instante_encontra_o_clipe_certo(self):
clipes = self.mapeador.clipes_do_arquivo(self.faixa, "0E6A8290.MP4")
clipe = self.mapeador.localizar_clipe_no_instante(clipes, 90.0)
self.assertEqual(clipe.identificador, "000f4775")
def test_localizar_clipe_no_instante_fora_de_qualquer_clipe_devolve_none(self):
clipes = self.mapeador.clipes_do_arquivo(self.faixa, "0E6A8290.MP4")
self.assertIsNone(self.mapeador.localizar_clipe_no_instante(clipes, 500.0))
def test_instante_na_timeline_traduz_usando_o_deslocamento_do_clipe(self):
clipe = _clipe("000f4775", 58.1, 86.0, 86.2, 114.0)
# 86.42s de origem (início real da fala) cai 0.22s depois do início
# do clipe (86.2s) e deve mapear para 58.1 + 0.22 = 58.32s na timeline.
self.assertAlmostEqual(self.mapeador.instante_na_timeline(clipe, 86.42), 58.32, places=6)
def test_instante_na_timeline_sem_intervalo_de_origem_levanta_erro(self):
clipe = Clipe("x", "n", IntervaloDeTempo(0, 1), intervalo_na_origem=None)
with self.assertRaises(ErroDeMapeamentoDeTempo):
self.mapeador.instante_na_timeline(clipe, 0.5)
def test_instante_na_timeline_por_arquivo_encontra_e_traduz(self):
timeline = None # não é usado pelo atalho; mantido para deixar a chamada explícita
instante = self.mapeador.instante_na_timeline_por_arquivo(timeline, self.faixa, "0E6A8290.MP4", 90.0)
self.assertAlmostEqual(instante, 58.1 + (90.0 - 86.2), places=6)
def test_instante_na_timeline_por_arquivo_sem_clipe_levanta_erro(self):
with self.assertRaises(ErroDeMapeamentoDeTempo):
self.mapeador.instante_na_timeline_por_arquivo(None, self.faixa, "0E6A8290.MP4", 500.0)
if __name__ == "__main__":
unittest.main()