Compare commits

..
2 Commits
Author SHA1 Message Date
João Henrique 69e931eb45 feat: Alinhado o aplicador Python de plano de edição (aplicador_de
- Alinhado o aplicador Python de plano de edição (aplicador_de_plano_de_edicao.py) com o script apply-editorial-actions.mjs de referência: no-op de split restrito aos padrões reais do bridge, pausa entre chamadas ao MCP, tolerância de borda de corte usa a posição real em vez de rejeitar desvio; adicionado backup automático da sequência (duplicate_sequence) e documentado o comando python code/engine/aplicar_plano_de_edicao.py como via alternativa ao painel CEP.

Resumo:
- 18 arquivos alterados
- 5 novos
- 13 modificados
- 0 removidos

 13 files changed, 704 insertions(+), 133 deletions(-)

Arquivos:
  - .jhonny/analises.db
  - CLAUDE.md
  - code/engine/aplicar_plano_de_edicao.py
  - code/engine/editor/__init__.py
  - code/engine/editor/aplicador_de_plano_de_edicao.py
  - code/engine/editor/escrita/escrita_no_editor.py
  - code/engine/integracoes/premiere/__init__.py
  - code/engine/integracoes/premiere/cliente_mcp.py
  - code/engine/testes/duplos_de_premiere.py
  - code/engine/testes/test_aplicador_de_plano_de_edicao.py
  - code/scripts/apply-editorial-actions.mjs
  - code/src/tools/discovery.ts
  - code/src/tools/timeline.ts
  - code/engine/editor/backup_de_sequencia.py
  - code/engine/integracoes/premiere/cliente_mcp_com_pausa.py
  - code/engine/testes/test_backup_de_sequencia.py
  - code/engine/testes/test_cliente_mcp_com_pausa.py
  - code/engine/testes/test_escrita_no_editor.py
2026-09-09 09:31:29 -04:00
João Henrique 4308c72441 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
2026-09-08 19:35:30 -04:00
34 changed files with 2555 additions and 65 deletions
Binary file not shown.
Binary file not shown.
View File
+23
View File
@@ -69,6 +69,29 @@ When working on this project, use these skills to:
2. **Design modules**: Use `/codebase-design` when creating new modules 2. **Design modules**: Use `/codebase-design` when creating new modules
3. **Model domain**: Use `/domain-modeling` to build shared vocabulary 3. **Model domain**: Use `/domain-modeling` to build shared vocabulary
## Aplicar um plano de edição sem o painel CEP
O painel CEP (aba "Editar Vídeo", botão "Aplicar plano na timeline") chama
`code/scripts/apply-editorial-actions.mjs` — esse é o caminho de produção,
comprovado, e continua sendo o padrão.
Existe uma segunda via em Python, para quando o painel não estiver aberto:
```bash
python code/engine/aplicar_plano_de_edicao.py "/caminho/plano_de_acoes.json"
```
Pré-requisito: o Premiere aberto, com a sequência correta **ativa** (o
script nunca troca de sequência — opera sobre a que estiver ativa no
momento), e o bridge MCP (`dist/index.js`) disponível. Cria backup
automático da sequência (`duplicate_sequence`) antes de aplicar qualquer
corte, igual ao painel — se o backup falhar, avisa em stderr e segue
aplicando mesmo assim, sem bloquear.
Ver [code/engine/editor/aplicador_de_plano_de_edicao.py](code/engine/editor/aplicador_de_plano_de_edicao.py)
para as diferenças deliberadas entre essa implementação e o `.mjs` do
painel (documentadas no docstring da classe `AplicadorDePlanoDeEdicao`).
## Code Standards ## Code Standards
- Python in `code/engine/`: follow `.agents/skills/boas-praticas-oo/SKILL.md` and [CODING_STANDARDS.md](CODING_STANDARDS.md) — mandatory, not a suggestion - Python in `code/engine/`: follow `.agents/skills/boas-praticas-oo/SKILL.md` and [CODING_STANDARDS.md](CODING_STANDARDS.md) — mandatory, not a suggestion
+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); }
+109
View File
@@ -0,0 +1,109 @@
"""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,
BackupDaSequencia,
EscritaNoEditor,
LeitorDePlanoDeEdicao,
MapeadorDeTempoDeOrigemParaTimeline,
)
from engine.integracoes.premiere.cliente_mcp import ClienteMCPPorStdio
from engine.integracoes.premiere.cliente_mcp_com_pausa import ClienteMCPComPausa
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)
# A pausa é aplicada no cliente, e não dentro do aplicador, para valer
# igualmente nas leituras da timeline e nas escritas — são todas
# round-trips síncronos no processo principal do Premiere.
cliente = ClienteMCPComPausa(ClienteMCPPorStdio(["node", str(CAMINHO_DO_SERVIDOR_MCP)]))
with SessaoMCP(cliente):
acesso_ao_editor = AcessoAoEditor(AcessoATimeline(cliente))
conversor_de_timeline = ConversorDeTimeline()
# Backup antes de qualquer corte, espelhando o painel CEP
# (applyEditorialActions() em cep-plugin/main.js): sem undo por
# script nesta versão do Premiere, é a única forma de recuperar o
# estado anterior se algo sair errado. Uma falha aqui não impede a
# aplicação — só fica registrada em stderr, igual ao painel.
sequencia_ativa = conversor_de_timeline.converter(acesso_ao_editor.obter_timeline_ativa())
backup_ok = BackupDaSequencia(cliente).criar_backup(sequencia_ativa.identificador)
print(
"Backup da sequência criado." if backup_ok else "Não foi possível criar o backup automático; aplicando mesmo assim.",
file=sys.stderr,
)
aplicador = AplicadorDePlanoDeEdicao(
acesso_ao_editor=acesso_ao_editor,
conversor_de_timeline=conversor_de_timeline,
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()
+37
View File
@@ -0,0 +1,37 @@
"""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 .backup_de_sequencia import BackupDaSequencia
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",
"BackupDaSequencia",
"EscritaNoEditor",
"ErroDeEdicao",
"ErroDeEscritaNoEditor",
"ErroDeMapeamentoDeTempo",
"ErroDePlanoInvalido",
"LeitorDePlanoDeEdicao",
"MapeadorDeTempoDeOrigemParaTimeline",
"PlanoDeEdicao",
"ResultadoDaAcao",
"ResultadoDaAplicacao",
"TipoDeAcao",
]
@@ -0,0 +1,351 @@
"""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
_TOLERANCIA_DE_DURACAO_EM_SEGUNDOS = 0.3
_RAIO_DE_BUSCA_DA_BORDA_EM_SEGUNDOS = 3.0
@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:
1. Cortes primeiro, 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.
2. Marcadores e zooms por último, depois que a timeline já está no
formato final. Zoom (``set_clip_properties``) muda uma propriedade do
*clipe inteiro*, não de um trecho dele — se dois zooms caíssem em
instantes diferentes do mesmo clipe ainda não dividido pelos cortes,
o segundo sobrescreveria o primeiro no clipe inteiro, e a divisão
posterior faria os dois pedaços herdarem só o último valor. Cortar
primeiro garante que cada zoom já encontra um clipe próprio.
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 ou a divisã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.
Diferença deliberada do outro aplicador de plano do projeto
(``scripts/apply-editorial-actions.mjs``, chamado pelo painel CEP):
aquele script separa "Fase A" (todos os splits do plano inteiro,
coordenadas originais) de "Fase B" (todas as remoções, feitas depois),
porque só sabe fazer aritmética pura sobre os cortes pedidos — não relê
o estado do Premiere no meio do caminho. Esta classe faz split e
remoção corte a corte, relendo a faixa a cada passo
(:meth:`_clipes_do_arquivo` exige ``intervalo_na_origem`` em todo
clipe), porque isso permite achar cada clipe pelo tempo de *origem*
real reportado pelo Premiere em vez de confiar só em aritmética — mais
robusto quando o bridge relata ``inPoint``/``outPoint`` de cada clipe,
mas por isso mesmo falha alto (em vez de simplesmente pular a faixa)
quando essa garantia falta numa faixa (ver
:meth:`_clipes_do_arquivo` e o teste
``test_corte_falha_em_vez_de_ignorar_faixa_sem_intervalo_de_origem``,
que comprova que um corte nessas condições não deixa vídeo cortado e
áudio intacto).
"""
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_corte(acao, plano.arquivo_de_origem) for acao in cortes]
resultados += [self._aplicar_acao_pontual(acao, plano.arquivo_de_origem) for acao in pontuais]
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._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.
Antes de mudar qualquer faixa, valida que todas as faixas com clipes
de ``arquivo_de_origem`` têm intervalo de origem conhecido. Sem essa
validação prévia, uma faixa problemática (ex.: áudio sem
``inPoint``/``outPoint``) só falharia depois de outra faixa já ter
sido cortada — deixando vídeo e áudio dessincronizados mesmo com o
corte "reportado" como falho.
"""
try:
faixas = self._timeline_atual().faixas
for faixa in faixas:
self._clipes_do_arquivo(faixa, arquivo_de_origem)
trechos_removidos = 0
for faixa in 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._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_pedido = self._borda_de_entrada(primeiro, acao.inicio)
fim_pedido = self._borda_de_saida(ultimo, acao.fim)
inicio_real = self._dividir_e_conferir(inicio_pedido, faixa, arquivo_de_origem)
fim_real = self._dividir_e_conferir(fim_pedido, faixa, arquivo_de_origem)
clipes_apos_a_divisao = self._clipes_do_arquivo(self._faixa_atualizada(faixa), arquivo_de_origem)
alvo = [
clipe for clipe in clipes_apos_a_divisao
if inicio_real - _TOLERANCIA_DE_BORDA_EM_SEGUNDOS <= clipe.intervalo_na_timeline.inicio
and clipe.intervalo_na_timeline.fim <= fim_real + _TOLERANCIA_DE_BORDA_EM_SEGUNDOS
]
duracao_removida = sum(clipe.intervalo_na_timeline.duracao for clipe in alvo)
# A duração esperada é a sobreposição real entre o intervalo da ação
# e os clipes afetados, não `acao.duracao` — se a ação começar ou
# terminar fora de qualquer clipe existente (ex.: pede corte a partir
# de um instante anterior ao início do material disponível), só a
# parte que realmente existe como filmagem deve ser contada.
duracao_esperada_pela_origem = sum(
self._sobreposicao_em_segundos(clipe, acao.inicio, acao.fim) for clipe in afetados
)
if abs(duracao_removida - duracao_esperada_pela_origem) > _TOLERANCIA_DE_DURACAO_EM_SEGUNDOS:
raise ErroDeMapeamentoDeTempo(
f"Na faixa {faixa.identificador!r}, os cortes pedidos em {inicio_pedido:.3f}s e "
f"{fim_pedido:.3f}s (o Premiere criou as bordas em {inicio_real:.3f}s e {fim_real:.3f}s) "
f"não isolaram o trecho esperado (esperado {duracao_esperada_pela_origem:.3f}s pela duração "
f"de origem, 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 _dividir_e_conferir(self, instante_pedido: float, faixa: Faixa, arquivo_de_origem: str) -> float:
"""Pede o corte em ``instante_pedido`` e devolve a borda que o Premiere realmente criou.
Corta **uma vez só** — nunca tenta de novo no mesmo ponto: uma
segunda tentativa deixaria dois cortes reais na timeline (o
"errado" e o corrigido), com uma fatia inútil entre eles, e o
Premiere não expõe undo por script para desfazer o primeiro
(``app.project.undo`` não existe nesta versão).
O desvio entre o instante pedido e a borda criada não é comparado
contra um limite fixo aqui: o ``split_clip`` do bridge do Premiere
desvia alguns décimos de segundo do pedido, e esse desvio cresce
com o tempo absoluto (ver :meth:`_borda_realmente_criada`) — é
comportamento normal do bridge, não um bug a rejeitar. A única
checagem de sanidade é o raio de busca de
:meth:`_borda_realmente_criada`
(:data:`_RAIO_DE_BUSCA_DA_BORDA_EM_SEGUNDOS`): além dele, não é
drift, é falha real, e essa função já levanta
:class:`ErroDeMapeamentoDeTempo` nesse caso.
"""
self.escrita.dividir_clipe_se_necessario(instante_pedido, faixa.indice, faixa.tipo)
clipes = self._clipes_do_arquivo(self._faixa_atualizada(faixa), arquivo_de_origem)
return self._borda_realmente_criada(clipes, instante_pedido)
def _faixa_atualizada(self, faixa: Faixa) -> Faixa:
"""Relê a timeline e devolve a versão atual da mesma faixa, pelo identificador."""
return next(f for f in self._timeline_atual().faixas if f.identificador == faixa.identificador)
def _borda_realmente_criada(self, clipes: list[Clipe], instante_pedido: float) -> float:
"""Acha, entre as bordas de clipe existentes, a mais próxima do instante pedido a Premiere.
O ``split_clip`` do bridge do Premiere pode cortar alguns décimos de
segundo antes ou depois do instante pedido (drift observado: cresce
com o tempo absoluto, compatível com uma conversão segundos↔ticks
que não usa a taxa de quadros real da sequência). As durações
relativas continuam corretas — só a posição absoluta desvia — por
isso a seleção do trecho a remover deve usar a borda que o Premiere
de fato criou, não o valor pedido.
Levanta :class:`ErroDeMapeamentoDeTempo` se nenhuma borda estiver
dentro de :data:`_RAIO_DE_BUSCA_DA_BORDA_EM_SEGUNDOS` do instante
pedido — nesse caso o corte realmente falhou, não é só drift.
"""
bordas = {clipe.intervalo_na_timeline.inicio for clipe in clipes} | {
clipe.intervalo_na_timeline.fim for clipe in clipes
}
borda_mais_proxima = min(bordas, key=lambda borda: abs(borda - instante_pedido), default=None)
if borda_mais_proxima is None or abs(borda_mais_proxima - instante_pedido) > _RAIO_DE_BUSCA_DA_BORDA_EM_SEGUNDOS:
raise ErroDeMapeamentoDeTempo(
f"Nenhuma borda de clipe foi encontrada perto de {instante_pedido:.3f}s "
f"(raio de busca: {_RAIO_DE_BUSCA_DA_BORDA_EM_SEGUNDOS}s); o corte não foi aplicado."
)
return borda_mais_proxima
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._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 _clipes_do_arquivo(self, faixa: Faixa, arquivo_de_origem: str) -> list[Clipe]:
"""Clipes de ``arquivo_de_origem`` em ``faixa``, exigindo intervalo de origem conhecido em todos.
Levanta :class:`ErroDeMapeamentoDeTempo` se algum clipe com esse nome
não tiver intervalo de origem — por exemplo, por uma lacuna na
ferramenta MCP que não reporta ``inPoint``/``outPoint`` para uma
faixa de áudio. Sem isso, o corte nessa faixa seria silenciosamente
ignorado (nenhum clipe "afetado" seria encontrado) em vez de avisar
que a faixa não pôde ser avaliada — foi exatamente essa lacuna que
deixou uma faixa de áudio inteira sem ser cortada numa aplicação
real deste plano.
"""
alvo = arquivo_de_origem.strip().lower()
nomeados = [clipe for clipe in faixa.clipes if (clipe.nome or "").strip().lower() == alvo]
sem_origem = [clipe for clipe in nomeados if clipe.intervalo_na_origem is None]
if sem_origem:
raise ErroDeMapeamentoDeTempo(
f"A faixa {faixa.identificador!r} tem {len(sem_origem)} clipe(s) de {arquivo_de_origem!r} "
"sem intervalo de origem reportado pelo Premiere; não é seguro decidir o que cortar, "
"mover ou marcar nessa faixa sem saber a que trecho de origem cada clipe corresponde."
)
return sorted(nomeados, key=lambda clipe: clipe.intervalo_na_origem.inicio)
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 _sobreposicao_em_segundos(clipe: Clipe, inicio: float, fim: float) -> float:
"""Quantos segundos do intervalo de origem do clipe caem dentro de [inicio, fim]."""
intervalo = clipe.intervalo_na_origem
if intervalo is None:
return 0.0
return max(0.0, min(intervalo.fim, fim) - max(intervalo.inicio, inicio))
@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.")
+34
View File
@@ -0,0 +1,34 @@
"""Backup automático da sequência antes de aplicar um plano de edição."""
from __future__ import annotations
from ..integracoes.premiere.cliente_mcp import ClienteMCP
from ..integracoes.premiere.erros_mcp import ErroDeFerramentaMCP
class BackupDaSequencia:
"""Duplica a sequência ativa do Premiere antes de uma edição destrutiva.
Espelha o passo de segurança do painel CEP
(``app.project.activeSequence.clone()`` em ``applyEditorialActions()``,
``cep-plugin/main.js``): sem undo por script nesta versão do Premiere,
duplicar a sequência antes de cortar é a única forma de recuperar o
estado anterior se algo sair errado.
"""
def __init__(self, cliente_mcp: ClienteMCP) -> None:
self.cliente_mcp = cliente_mcp
def criar_backup(self, identificador_da_sequencia: str) -> bool:
"""Duplica a sequência. Devolve ``True`` se conseguiu, ``False`` se o Premiere recusou.
Não levanta exceção numa recusa: o painel CEP também segue aplicando
o plano mesmo se o backup falhar (loga e avisa, não bloqueia) — a
falta de backup não deve impedir a edição que o usuário pediu, só
precisa ficar visível para quem chama decidir o que fazer.
"""
try:
self.cliente_mcp.chamar("duplicate_sequence", {"sequence_id": identificador_da_sequencia})
return True
except ErroDeFerramentaMCP:
return False
+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,133 @@
"""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
import re
from ...integracoes.premiere.cliente_mcp import ClienteMCP
from ...integracoes.premiere.erros_mcp import ErroDeFerramentaMCP
from ..erros import ErroDeEscritaNoEditor
# O bridge recusa a divisão de duas formas diferentes quando o ponto pedido
# JÁ é uma borda de clipe. Ambas são no-op, não falha:
# 1. nenhum clipe atravessa estritamente o instante ("strictly spans");
# 2. o razor roda mas a contagem de clipes da faixa não muda, porque o ponto
# coincide por arredondamento de quadro com uma borda existente.
# Os mesmos dois padrões são reconhecidos por scripts/apply-editorial-actions.mjs,
# que aplica planos por este mesmo bridge.
_RECUSA_POR_NENHUM_CLIPE_ATRAVESSANDO = re.compile(r"strictly spans", re.IGNORECASE)
_RECUSA_COM_CONTAGEM_DE_CLIPES = re.compile(
r"changed the track clip count from (\d+) to (\d+)", re.IGNORECASE
)
def _e_divisao_ja_existente(mensagem_de_erro: str) -> bool:
"""``True`` se a recusa do bridge significa "esse ponto já era uma borda"."""
if _RECUSA_POR_NENHUM_CLIPE_ATRAVESSANDO.search(mensagem_de_erro):
return True
contagem = _RECUSA_COM_CONTAGEM_DE_CLIPES.search(mensagem_de_erro)
return contagem is not None and contagem.group(1) == contagem.group(2)
_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 — 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, ou quando um plano parcialmente
aplicado é reexecutado.
Só as recusas reconhecidas por :func:`_e_divisao_ja_existente` contam
como no-op. Qualquer outra recusa vira
:class:`ErroDeEscritaNoEditor`: antes, um ``except`` largo tratava
toda :class:`ErroDeFerramentaMCP` como "já era borda", então uma
faixa bloqueada ou uma queda de bridge passava por corte
bem-sucedido e o trecho seguia inteiro na timeline, sem aviso.
"""
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 as erro:
if _e_divisao_ja_existente(str(erro)):
return False
raise ErroDeEscritaNoEditor(
f"Não foi possível dividir a faixa {tipo_da_faixa}_{indice_da_faixa} "
f"em {instante_da_timeline:.3f}s: {erro}"
) from erro
def remover_trecho(self, identificador_do_clipe: str) -> None:
"""Remove o clipe da timeline fechando o espaço que ele ocupava.
Usa ``ripple_delete`` (QE ``rippleDelete``), não
``remove_from_timeline``: numa sequência real com outras faixas
ocupadas, o ``remove_from_timeline`` com ``ripple`` apagou o clipe
mas deixou o buraco aberto, e um buraco no meio do corte é um corte
errado. O ``ripple_delete`` confere, do lado do Premiere, que o
clipe realmente sumiu.
Levanta :class:`ErroDeEscritaNoEditor` se o Premiere recusar a
remoção.
"""
try:
self.cliente_mcp.chamar("ripple_delete", {"node_id": identificador_do_clipe})
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()
+2 -1
View File
@@ -1,7 +1,8 @@
from .cliente_mcp import ClienteMCP, ClienteMCPPorStdio from .cliente_mcp import ClienteMCP, ClienteMCPPorStdio
from .cliente_mcp_com_pausa import ClienteMCPComPausa
from .erros_mcp import ErroMCP, ErroDeConexaoMCP, ErroDeRespostaMCP, ErroDeFerramentaMCP from .erros_mcp import ErroMCP, ErroDeConexaoMCP, ErroDeRespostaMCP, ErroDeFerramentaMCP
__all__ = ["ClienteMCP", "ClienteMCPPorStdio", "ErroMCP", "ErroDeConexaoMCP", "ErroDeRespostaMCP", "ErroDeFerramentaMCP"] __all__ = ["ClienteMCP", "ClienteMCPPorStdio", "ClienteMCPComPausa", "ErroMCP", "ErroDeConexaoMCP", "ErroDeRespostaMCP", "ErroDeFerramentaMCP"]
from .sessao_mcp import SessaoMCP from .sessao_mcp import SessaoMCP
__all__ = ["SessaoMCP"] __all__ = ["SessaoMCP"]
@@ -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,63 @@ 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", {}) resultado = resposta.get("result", {})
# Uma ferramenta que falha não devolve erro no nível do JSON-RPC: o
# servidor responde "com sucesso" e marca `isError` no resultado. Sem
# olhar esse campo, uma escrita recusada pelo Premiere passaria por
# bem-sucedida — foi assim que uma chamada a uma ferramenta inexistente
# ficou invisível até o efeito aparecer na timeline.
if resultado.get("isError"):
raise ErroDeFerramentaMCP(
f"A ferramenta {nome_da_ferramenta!r} falhou: {self._mensagem_de_erro(resultado)}"
)
return resultado
@staticmethod
def _mensagem_de_erro(resultado: dict[str, Any]) -> str:
"""Extrai a mensagem de erro que o servidor MCP devolveu no resultado."""
conteudo_estruturado = resultado.get("structuredContent")
if isinstance(conteudo_estruturado, dict) and conteudo_estruturado.get("error"):
return str(conteudo_estruturado["error"])
for bloco in resultado.get("content", []):
if isinstance(bloco, dict) and bloco.get("type") == "text":
return str(bloco.get("text", ""))
return "sem detalhes"
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)
@@ -0,0 +1,72 @@
"""Espaçamento das chamadas ao bridge do Premiere.
As ferramentas de edição do Premiere são round-trips síncronos de QE
DOM/ExtendScript no processo principal do aplicativo. Uma rajada dessas
chamadas sem folga entre elas trava a interface do Premiere — observado na
prática ao aplicar um plano com 17 cortes seguidos de zooms e marcadores.
O script ``scripts/apply-editorial-actions.mjs``, que aplica planos por este
mesmo bridge sem travar, resolve isso com uma pausa fixa entre chamadas;
este módulo traz a mesma folga para o caminho Python.
"""
from __future__ import annotations
import time
from typing import Any, Callable
from .cliente_mcp import ClienteMCP
#: Folga adotada entre chamadas consecutivas, igual à do apply-editorial-actions.mjs
#: (``BRIDGE_DELAY_MS = 120``), que é o valor validado numa aplicação real.
PAUSA_PADRAO_ENTRE_CHAMADAS_EM_SEGUNDOS = 0.12
class ClienteMCPComPausa(ClienteMCP):
"""Envolve outro :class:`ClienteMCP` garantindo uma folga entre chamadas.
Decora um cliente já pronto em vez de espalhar esperas pelas classes de
domínio: assim a folga vale tanto para as leituras
(``get_active_sequence``) quanto para as escritas (``split_clip``,
``ripple_delete``), que é o que de fato ocupa o processo principal do
Premiere, e as classes de edição continuam sem saber que existe um
bridge lento do outro lado.
A pausa só entra *entre* chamadas — a primeira não espera — e vale
também depois de uma chamada que falhou, porque uma chamada recusada
ocupou o bridge do mesmo jeito.
"""
def __init__(
self,
cliente_mcp: ClienteMCP,
pausa_em_segundos: float = PAUSA_PADRAO_ENTRE_CHAMADAS_EM_SEGUNDOS,
esperar: Callable[[float], None] = time.sleep,
) -> None:
"""``esperar`` é injetável para que os testes não durmam de verdade."""
if pausa_em_segundos < 0:
raise ValueError(
f"A pausa entre chamadas não pode ser negativa (recebido: {pausa_em_segundos})."
)
self.cliente_mcp = cliente_mcp
self.pausa_em_segundos = pausa_em_segundos
self.esperar = esperar
self._ja_chamou = False
def conectar(self) -> None:
"""Delega ao cliente interno; conectar não é um round-trip de edição."""
self.cliente_mcp.conectar()
def desconectar(self) -> None:
"""Delega ao cliente interno; desconectar não é um round-trip de edição."""
self.cliente_mcp.desconectar()
def esta_conectado(self) -> bool:
"""Estado de conexão do cliente interno."""
return self.cliente_mcp.esta_conectado()
def chamar(self, nome_da_ferramenta: str, argumentos: dict[str, Any] | None = None) -> dict[str, Any]:
"""Espera a folga devida (menos na primeira chamada) e delega ao cliente interno."""
if self._ja_chamou and self.pausa_em_segundos > 0:
self.esperar(self.pausa_em_segundos)
self._ja_chamou = True
return self.cliente_mcp.chamar(nome_da_ferramenta, argumentos)
+242
View File
@@ -0,0 +1,242 @@
"""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``,
``ripple_delete``, ``set_clip_properties`` e ``add_marker``.
"""
from __future__ import annotations
import copy
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,
sem_origem_no_audio: bool = False,
clipe_unico: bool = False,
desvio_do_split: float = 0.0,
) -> None:
"""``sem_origem_no_audio`` reproduz o bug real do bridge: relatar clipes de
áudio sem ``inPoint``/``outPoint`` na resposta de ``get_active_sequence``.
``clipe_unico`` reproduz a gravação "sem corte": um só clipe cobrindo
todo o arquivo, com origem e timeline idênticas — o cenário em que
dois zooms cairiam no mesmo clipe se rodassem antes dos cortes.
``desvio_do_split`` reproduz o drift real do ``split_clip``: em vez
de cortar exatamente no instante pedido, corta em
``instante + desvio_do_split``."""
self.sem_origem_no_audio = sem_origem_no_audio
self.desvio_do_split = desvio_do_split
self.chamadas: list[tuple[str, dict[str, Any]]] = []
self._proximo_id = 0
self._historico: list[dict[str, list[_ClipeFalso]]] = []
self._faixas: dict[str, list[_ClipeFalso]] = {
"video_0": [self._novo_clipe(self._ARQUIVO, 0.0, 200.0, 0.0, 200.0, "clipe_unico_video")],
"audio_0": [self._novo_clipe(self._ARQUIVO, 0.0, 200.0, 0.0, 200.0, "clipe_unico_audio")],
} if clipe_unico else {
"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 ``ripple_delete``, na ordem em que ocorreram."""
return [argumentos for nome, argumentos in self.chamadas if nome == "ripple_delete"]
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,
"ripple_delete": self._simular_ripple_delete,
"set_clip_properties": self._simular_set_clip_properties,
"add_marker": self._simular_add_marker,
"undo": self._simular_undo,
}
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 _clipe_como_dict(clipe: _ClipeFalso, tipo: str) -> dict[str, Any]:
base = {
"nodeId": clipe.identificador,
"name": clipe.nome,
"start": clipe.inicio_na_timeline,
"end": clipe.fim_na_timeline,
}
if tipo == "audio" and self.sem_origem_no_audio:
return base
return {**base, "inPoint": clipe.inicio_na_origem, "outPoint": clipe.fim_na_origem}
def _faixa_como_dict(identificador: str, tipo: str, indice: int) -> dict[str, Any]:
return {
"id": identificador,
"name": identificador,
"index": indice,
"clips": [_clipe_como_dict(clipe, tipo) 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.
Aplica ``self.desvio_do_split`` ao ponto de corte, reproduzindo o
drift real do bridge — o corte pode sair do instante pedido.
A recusa usa a mensagem em inglês do bridge real ("No clip strictly
spans ...") porque :class:`EscritaNoEditor` distingue no-op de erro
real pelo texto da recusa; um dublê com mensagem própria testaria um
caminho que não existe em produção.
"""
identificador_da_faixa = self._identificador_da_faixa(argumentos["track_index"], argumentos["track_type"])
instante_pedido = float(argumentos["time_seconds"])
faixa = self._faixas[identificador_da_faixa]
clipe_no_instante_pedido = next(
(c for c in faixa if c.inicio_na_timeline < instante_pedido < c.fim_na_timeline), None
)
if clipe_no_instante_pedido is None:
raise ErroDeFerramentaMCP(
f"No clip strictly spans {instante_pedido}s on track {identificador_da_faixa}."
)
instante = min(
max(instante_pedido + self.desvio_do_split, clipe_no_instante_pedido.inicio_na_timeline + 0.001),
clipe_no_instante_pedido.fim_na_timeline - 0.001,
)
self._salvar_estado()
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"No clip strictly spans {instante}s on track {identificador_da_faixa}."
)
def _simular_ripple_delete(self, argumentos: dict[str, Any]) -> dict[str, Any]:
"""Remove o clipe pedido e 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:
self._salvar_estado()
duracao = clipe.fim_na_timeline - clipe.inicio_na_timeline
del clipes[posicao]
for outro in clipes:
if outro.inicio_na_timeline >= clipe.fim_na_timeline:
outro.inicio_na_timeline -= duracao
outro.fim_na_timeline -= duracao
return {"rippleDeleted": True, "verified": 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}
def _salvar_estado(self) -> None:
"""Guarda uma cópia do estado das faixas para uma futura chamada a ``undo``."""
self._historico.append(copy.deepcopy(self._faixas))
def _simular_undo(self, argumentos: dict[str, Any]) -> dict[str, Any]:
"""Restaura o estado das faixas anterior à última mutação salva."""
contagem = int(argumentos.get("count", 1))
for _ in range(contagem):
if not self._historico:
raise ErroDeFerramentaMCP("Nada para desfazer neste dublê de teste.")
self._faixas = self._historico.pop()
return {"undone": contagem}
@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,220 @@
"""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_depois_dos_cortes(self):
# Zoom muda uma propriedade do clipe inteiro (set_clip_properties),
# não de um trecho dele. Se o zoom rodasse antes do corte, cairia no
# clipe original ainda não dividido; dois zooms em pontos diferentes
# do mesmo clipe se sobrescreveriam, e a divisão posterior faria os
# pedaços herdarem só o último valor. Por isso cortes vêm primeiro.
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])
# corte aconteceu antes de marcador e zoom 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] == "ripple_delete")
self.assertLess(indice_da_remocao, indice_do_marcador)
self.assertLess(indice_da_remocao, indice_do_zoom)
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_dois_zooms_no_mesmo_clipe_original_nao_se_sobrescrevem(self):
# Bug real: com um só clipe cobrindo o arquivo inteiro (gravação
# "sem corte"), dois zooms em trechos diferentes caíam no mesmo
# clipe ainda não dividido; o segundo sobrescrevia a escala do
# primeiro no clipe inteiro, e ambos os pedaços resultantes da
# divisão herdavam só o último valor.
cliente = ClienteMCPFalso(clipe_unico=True)
aplicador = AplicadorDePlanoDeEdicao(
acesso_ao_editor=AcessoAoEditor(AcessoATimeline(cliente)),
conversor_de_timeline=ConversorDeTimeline(),
escrita=EscritaNoEditor(cliente),
mapeador=MapeadorDeTempoDeOrigemParaTimeline(),
)
plano = PlanoDeEdicao("0E6A8290.mp4", (
AcaoDeEdicao(TipoDeAcao.CORTE, inicio=90.0, fim=100.0, motivo="bastidor no meio"),
AcaoDeEdicao(TipoDeAcao.ZOOM, inicio=10.0, fim=20.0, motivo="ênfase 1", parametros={"scale": 1.15}),
AcaoDeEdicao(TipoDeAcao.ZOOM, inicio=150.0, fim=160.0, motivo="ênfase 2", parametros={"scale": 1.30}),
))
resultado = aplicador.aplicar(plano)
self.assertTrue(resultado.todas_bem_sucedidas, msg=[r.detalhe for r in resultado.falhas])
chamadas_de_zoom = [c for c in cliente.chamadas if c[0] == "set_clip_properties"]
self.assertEqual(len(chamadas_de_zoom), 2)
node_ids_zoomados = {c[1]["node_id"] for c in chamadas_de_zoom}
# os dois zooms caíram em clipes diferentes, um por trecho
self.assertEqual(len(node_ids_zoomados), 2)
escalas = sorted(round(c[1]["scale"], 3) for c in chamadas_de_zoom)
self.assertEqual(escalas, [115.0, 130.0])
def test_corte_com_desvio_de_bridge_usa_a_borda_real_em_vez_de_falhar(self):
# O split_clip do bridge do Premiere corta alguns décimos de segundo
# antes ou depois do instante pedido, e o desvio cresce com o tempo
# absoluto (drift de conversão segundos↔ticks, não bug pontual) — é
# comportamento normal, não uma falha. _borda_realmente_criada já
# localiza onde o Premiere de fato cortou; o corte usa essa borda em
# vez de rejeitar o desvio.
cliente = ClienteMCPFalso(clipe_unico=True, desvio_do_split=-0.3)
aplicador = AplicadorDePlanoDeEdicao(
acesso_ao_editor=AcessoAoEditor(AcessoATimeline(cliente)),
conversor_de_timeline=ConversorDeTimeline(),
escrita=EscritaNoEditor(cliente),
mapeador=MapeadorDeTempoDeOrigemParaTimeline(),
)
plano = PlanoDeEdicao("0E6A8290.mp4", (
AcaoDeEdicao(TipoDeAcao.CORTE, inicio=50.0, fim=60.0, motivo="bastidor"),
))
resultado = aplicador.aplicar(plano)
self.assertTrue(resultado.todas_bem_sucedidas, msg=[r.detalhe for r in resultado.falhas])
self.assertEqual(len(cliente.chamadas_de_remocao), 2) # vídeo + áudio
# nenhuma segunda tentativa de corte no mesmo ponto pedido, por faixa
pedidos_por_faixa = [
(c[1]["track_type"], c[1]["time_seconds"]) for c in cliente.chamadas if c[0] == "split_clip"
]
self.assertEqual(len(pedidos_por_faixa), len(set(pedidos_por_faixa)))
def test_corte_sem_nenhuma_borda_por_perto_ainda_falha(self):
# O desvio do bridge é tolerado, mas só dentro de um raio plausível
# (_RAIO_DE_BUSCA_DA_BORDA_EM_SEGUNDOS): além disso o corte realmente
# falhou, não é mais drift — precisa continuar falhando alto, sem
# tentar de novo no mesmo ponto (o Premiere não expõe undo por
# script, então uma segunda tentativa deixaria o corte errado
# comendo espaço na timeline, com uma fatia inútil ao lado).
cliente = ClienteMCPFalso(clipe_unico=True, desvio_do_split=-5.0)
aplicador = AplicadorDePlanoDeEdicao(
acesso_ao_editor=AcessoAoEditor(AcessoATimeline(cliente)),
conversor_de_timeline=ConversorDeTimeline(),
escrita=EscritaNoEditor(cliente),
mapeador=MapeadorDeTempoDeOrigemParaTimeline(),
)
plano = PlanoDeEdicao("0E6A8290.mp4", (
AcaoDeEdicao(TipoDeAcao.CORTE, inicio=50.0, fim=60.0, motivo="bastidor"),
))
resultado = aplicador.aplicar(plano)
self.assertFalse(resultado.todas_bem_sucedidas)
self.assertIn("Nenhuma borda", resultado.falhas[0].detalhe)
# nada removido e nenhuma segunda tentativa de corte no mesmo ponto
self.assertEqual(cliente.chamadas_de_remocao, [])
instantes_pedidos = [c[1]["time_seconds"] for c in cliente.chamadas if c[0] == "split_clip"]
self.assertEqual(len(instantes_pedidos), len(set(instantes_pedidos)))
def test_corte_dentro_da_tolerancia_de_um_quadro_e_aceito(self):
# O Premiere encaixa o corte no quadro mais próximo, então uma
# diferença de fração de quadro é normal e não pode falhar.
cliente = ClienteMCPFalso(clipe_unico=True, desvio_do_split=-0.01)
aplicador = AplicadorDePlanoDeEdicao(
acesso_ao_editor=AcessoAoEditor(AcessoATimeline(cliente)),
conversor_de_timeline=ConversorDeTimeline(),
escrita=EscritaNoEditor(cliente),
mapeador=MapeadorDeTempoDeOrigemParaTimeline(),
)
plano = PlanoDeEdicao("0E6A8290.mp4", (
AcaoDeEdicao(TipoDeAcao.CORTE, inicio=50.0, fim=60.0, motivo="bastidor"),
))
resultado = aplicador.aplicar(plano)
self.assertTrue(resultado.todas_bem_sucedidas, msg=[r.detalhe for r in resultado.falhas])
self.assertEqual(len(cliente.chamadas_de_remocao), 2) # vídeo + áudio
self.assertEqual(len(cliente._faixas["video_0"]), 2) # sem fatia sobrando
def test_corte_falha_em_vez_de_ignorar_faixa_sem_intervalo_de_origem(self):
# Reproduz o bug real do bridge do Premiere: a faixa de áudio não
# reportava inPoint/outPoint, então o corte "afetava" 0 clipes ali e
# passava batido — a faixa de áudio inteira ficava sem ser cortada,
# dessincronizando o vídeo (já cortado) do áudio (intacto).
cliente = ClienteMCPFalso(sem_origem_no_audio=True)
aplicador = AplicadorDePlanoDeEdicao(
acesso_ao_editor=AcessoAoEditor(AcessoATimeline(cliente)),
conversor_de_timeline=ConversorDeTimeline(),
escrita=EscritaNoEditor(cliente),
mapeador=MapeadorDeTempoDeOrigemParaTimeline(),
)
plano = PlanoDeEdicao("0E6A8290.mp4", (
AcaoDeEdicao(TipoDeAcao.CORTE, inicio=0.0, fim=40.6, motivo="bastidor"),
))
resultado = aplicador.aplicar(plano)
self.assertFalse(resultado.todas_bem_sucedidas)
self.assertIn("sem intervalo de origem", resultado.falhas[0].detalhe)
# o vídeo não pode ter sido cortado sozinho: nada foi removido em faixa nenhuma
self.assertEqual(cliente.chamadas_de_remocao, [])
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,54 @@
"""Testes do backup automático da sequência antes de aplicar um plano."""
import unittest
from typing import Any
from engine.editor.backup_de_sequencia import BackupDaSequencia
from engine.integracoes.premiere.cliente_mcp import ClienteMCP
from engine.integracoes.premiere.erros_mcp import ErroDeFerramentaMCP
class _ClienteQueRegistra(ClienteMCP):
"""Dublê que registra as chamadas e devolve sucesso ou o erro configurado."""
def __init__(self, erro_a_levantar: Exception | None = None) -> None:
self.chamadas: list[tuple[str, dict[str, Any]]] = []
self.erro_a_levantar = erro_a_levantar
def conectar(self) -> None:
"""Não há conexão real neste dublê."""
def desconectar(self) -> None:
"""Não há conexão real neste dublê."""
def esta_conectado(self) -> bool:
"""Sempre conectado neste dublê."""
return True
def chamar(self, nome_da_ferramenta: str, argumentos: dict[str, Any] | None = None) -> dict[str, Any]:
self.chamadas.append((nome_da_ferramenta, argumentos or {}))
if self.erro_a_levantar is not None:
raise self.erro_a_levantar
return {"success": True}
class TesteBackupDaSequencia(unittest.TestCase):
def test_duplica_a_sequencia_pelo_identificador(self):
cliente = _ClienteQueRegistra()
backup = BackupDaSequencia(cliente)
conseguiu = backup.criar_backup("sequencia_falsa")
self.assertTrue(conseguiu)
self.assertEqual(cliente.chamadas, [("duplicate_sequence", {"sequence_id": "sequencia_falsa"})])
def test_falha_no_backup_nao_levanta_excecao(self):
# Espelha o painel CEP: se o backup falhar, aplica o plano mesmo
# assim (loga e avisa, não bloqueia) — quem chama decide o que
# fazer com o retorno False, não uma exceção não tratada.
cliente = _ClienteQueRegistra(erro_a_levantar=ErroDeFerramentaMCP("recusado"))
backup = BackupDaSequencia(cliente)
conseguiu = backup.criar_backup("sequencia_falsa")
self.assertFalse(conseguiu)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,101 @@
"""Testes do decorator que espaça as chamadas ao bridge do Premiere."""
import unittest
from typing import Any
from engine.integracoes.premiere.cliente_mcp import ClienteMCP
from engine.integracoes.premiere.cliente_mcp_com_pausa import ClienteMCPComPausa
from engine.integracoes.premiere.erros_mcp import ErroDeFerramentaMCP
class _ClienteQueRegistra(ClienteMCP):
"""Dublê que só registra as chamadas recebidas e devolve um resultado fixo."""
def __init__(self, erro_a_levantar: Exception | None = None) -> None:
self.chamadas: list[tuple[str, dict[str, Any]]] = []
self.erro_a_levantar = erro_a_levantar
self.conectado = False
def conectar(self) -> None:
"""Marca o dublê como conectado."""
self.conectado = True
def desconectar(self) -> None:
"""Marca o dublê como desconectado."""
self.conectado = False
def esta_conectado(self) -> bool:
"""Estado de conexão registrado pelo dublê."""
return self.conectado
def chamar(self, nome_da_ferramenta: str, argumentos: dict[str, Any] | None = None) -> dict[str, Any]:
"""Registra a chamada e devolve um resultado fixo, ou levanta o erro configurado."""
self.chamadas.append((nome_da_ferramenta, argumentos or {}))
if self.erro_a_levantar is not None:
raise self.erro_a_levantar
return {"ok": True}
class _EsperaFalsa:
"""Substitui ``time.sleep`` registrando as pausas pedidas, sem esperar de verdade."""
def __init__(self) -> None:
self.pausas: list[float] = []
def __call__(self, segundos: float) -> None:
"""Registra a pausa pedida."""
self.pausas.append(segundos)
class TesteClienteMCPComPausa(unittest.TestCase):
"""Garante o espaçamento entre chamadas e a transparência do decorator."""
def setUp(self):
self.interno = _ClienteQueRegistra()
self.espera = _EsperaFalsa()
self.cliente = ClienteMCPComPausa(self.interno, pausa_em_segundos=0.12, esperar=self.espera)
def test_repassa_a_chamada_ao_cliente_interno(self):
resultado = self.cliente.chamar("split_clip", {"time_seconds": 10.0})
self.assertEqual(resultado, {"ok": True})
self.assertEqual(self.interno.chamadas, [("split_clip", {"time_seconds": 10.0})])
def test_nao_pausa_antes_da_primeira_chamada(self):
# A primeira chamada não precisa esperar: não há chamada anterior
# ocupando o processo principal do Premiere.
self.cliente.chamar("get_active_sequence")
self.assertEqual(self.espera.pausas, [])
def test_pausa_entre_chamadas_consecutivas(self):
# Sem essa folga, uma rajada de round-trips síncronos de QE/ExtendScript
# trava a UI do Premiere — foi o que se viu com um plano de 17 cortes.
for _ in range(3):
self.cliente.chamar("split_clip", {"time_seconds": 1.0})
self.assertEqual(self.espera.pausas, [0.12, 0.12])
def test_pausa_tambem_depois_de_uma_chamada_que_falhou(self):
# Uma chamada que falhou ocupou o bridge do mesmo jeito; emendar a
# próxima sem folga é justamente o que derruba a UI.
interno = _ClienteQueRegistra(erro_a_levantar=ErroDeFerramentaMCP("recusado"))
espera = _EsperaFalsa()
cliente = ClienteMCPComPausa(interno, pausa_em_segundos=0.12, esperar=espera)
for _ in range(2):
with self.assertRaises(ErroDeFerramentaMCP):
cliente.chamar("split_clip", {"time_seconds": 1.0})
self.assertEqual(espera.pausas, [0.12])
def test_conexao_e_delegada_sem_pausa(self):
# Conectar/desconectar não são round-trips de edição; não precisam de folga.
self.cliente.conectar()
self.assertTrue(self.cliente.esta_conectado())
self.cliente.desconectar()
self.assertFalse(self.cliente.esta_conectado())
self.assertEqual(self.espera.pausas, [])
def test_pausa_negativa_e_recusada(self):
with self.assertRaises(ValueError):
ClienteMCPComPausa(self.interno, pausa_em_segundos=-0.1)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,75 @@
"""Testes da escrita de mutações no Premiere, sem o app aberto.
Cobrem a distinção — que já custou um corte silenciosamente perdido — entre
um ``split_clip`` que não fez nada porque o ponto já era borda de clipe (um
no-op legítimo) e um ``split_clip`` que falhou de verdade no bridge.
"""
import unittest
from typing import Any
from engine.editor.erros import ErroDeEscritaNoEditor
from engine.editor.escrita import EscritaNoEditor
from engine.integracoes.premiere.cliente_mcp import ClienteMCP
from engine.integracoes.premiere.erros_mcp import ErroDeFerramentaMCP
class _ClienteQueRecusa(ClienteMCP):
"""Dublê que recusa toda chamada com a mensagem informada no construtor."""
def __init__(self, mensagem_de_erro: str) -> None:
self.mensagem_de_erro = mensagem_de_erro
self.chamadas: list[tuple[str, dict[str, Any]]] = []
def conectar(self) -> None:
"""Não há conexão real neste dublê."""
def desconectar(self) -> None:
"""Não há conexão real neste dublê."""
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]:
"""Registra a chamada e recusa com :class:`ErroDeFerramentaMCP`."""
self.chamadas.append((nome_da_ferramenta, argumentos or {}))
raise ErroDeFerramentaMCP(self.mensagem_de_erro)
class TesteDividirClipeSeNecessario(unittest.TestCase):
"""Separa o no-op esperado do erro real de bridge."""
def _dividir(self, mensagem_de_erro: str) -> bool:
escrita = EscritaNoEditor(_ClienteQueRecusa(mensagem_de_erro))
return escrita.dividir_clipe_se_necessario(10.0, 0, "video")
def test_ponto_que_ja_era_borda_e_no_op_e_nao_erro(self):
# O Premiere recusa dividir onde nenhum clipe atravessa o instante:
# é o resultado esperado quando o corte pedido coincide com uma borda
# já existente (inclusive ao reexecutar um plano parcialmente aplicado).
self.assertFalse(self._dividir("No clip strictly spans 10.0s on track video_0."))
def test_contagem_de_clipes_inalterada_e_no_op_e_nao_erro(self):
# Segunda forma observada do mesmo no-op: o razor roda, mas o ponto já
# coincide por arredondamento de quadro com uma borda existente, então
# a contagem de clipes da faixa não muda.
self.assertFalse(self._dividir("razor changed the track clip count from 4 to 4, expected 5"))
def test_erro_real_do_bridge_nao_e_confundido_com_no_op(self):
# Bug real: qualquer ErroDeFerramentaMCP era engolido como "já era
# borda", então uma falha de conexão ou uma faixa bloqueada passava
# por corte bem-sucedido e o trecho seguia na timeline sem aviso.
with self.assertRaises(ErroDeEscritaNoEditor) as capturado:
self._dividir("Track video_0 is locked.")
self.assertIn("locked", str(capturado.exception))
def test_contagem_de_clipes_que_mudou_errado_nao_e_no_op(self):
# A contagem mudou, mas não para o esperado: o razor fez algo, e algo
# diferente do pedido — isso é erro, não no-op.
with self.assertRaises(ErroDeEscritaNoEditor):
self._dividir("razor changed the track clip count from 4 to 7, expected 5")
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()
+68 -8
View File
@@ -11,8 +11,12 @@
* "coordenada pós-corte" (ver adjustTimeForCuts) antes de procurar o clipe. * "coordenada pós-corte" (ver adjustTimeForCuts) antes de procurar o clipe.
* *
* Limitações conhecidas: * Limitações conhecidas:
* - "zoom" aplica um scale estático (set_clip_scale) no clipe do intervalo, sem rampa * - "zoom" cria uma camada de ajuste (adjustment layer) numa track de vídeo dedicada,
* de entrada/saída. Suficiente para um punch-in simples; não é um Ken Burns animado. * posicionada e recortada para durar exatamente o intervalo da action, com o Scale
* do próprio Motion da camada de ajuste ajustado — não altera o clipe original.
* É um scale estático (sem rampa de entrada/saída); suficiente para um punch-in
* simples, não é um Ken Burns animado. A track de ajuste é criada uma vez (no
* topo da pilha de vídeo) e reaproveitada por todas as actions "zoom" do plano.
* - "text" não tem API de scripting suportada no Premiere para criar clipe de texto * - "text" não tem API de scripting suportada no Premiere para criar clipe de texto
* a partir de string crua (add_text_overlay do MCP já retorna erro por design). * a partir de string crua (add_text_overlay do MCP já retorna erro por design).
* Em vez de falhar o plano inteiro, cada action "text" vira um marker de sequência * Em vez de falhar o plano inteiro, cada action "text" vira um marker de sequência
@@ -28,6 +32,9 @@ import { getDiscoveryTools } from "../dist/tools/discovery.js";
import { getProjectTools } from "../dist/tools/project.js"; import { getProjectTools } from "../dist/tools/project.js";
import { getMarkerTools } from "../dist/tools/markers.js"; import { getMarkerTools } from "../dist/tools/markers.js";
import { getTrackTargetingTools } from "../dist/tools/track-targeting.js"; import { getTrackTargetingTools } from "../dist/tools/track-targeting.js";
import { getTrackTools } from "../dist/tools/tracks.js";
import { getPlayheadTools } from "../dist/tools/playhead.js";
import { getUtilityTools } from "../dist/tools/utility.js";
// Progresso para a barra do painel CEP: linhas @@PROGRESS em stderr, que o painel // Progresso para a barra do painel CEP: linhas @@PROGRESS em stderr, que o painel
// intercepta em vez de despejar no log. Fases A/B/C viram uma barra única. // intercepta em vez de despejar no log. Fases A/B/C viram uma barra única.
@@ -121,6 +128,9 @@ async function main() {
const project = getProjectTools(bridgeOptions); const project = getProjectTools(bridgeOptions);
const markers = getMarkerTools(bridgeOptions); const markers = getMarkerTools(bridgeOptions);
const trackTargeting = getTrackTargetingTools(bridgeOptions); const trackTargeting = getTrackTargetingTools(bridgeOptions);
const tracks = getTrackTools(bridgeOptions);
const playhead = getPlayheadTools(bridgeOptions);
const utility = getUtilityTools(bridgeOptions);
console.error( console.error(
`[apply-editorial-actions] sequência="${args.sequence}" cuts=${cuts.length} outras=${others.length} dryRun=${args.dryRun}` `[apply-editorial-actions] sequência="${args.sequence}" cuts=${cuts.length} outras=${others.length} dryRun=${args.dryRun}`
@@ -236,6 +246,9 @@ async function main() {
reportProgress("Aplicando zooms, textos e marcadores", phaseCBase, `0 de ${others.length} ações`); reportProgress("Aplicando zooms, textos e marcadores", phaseCBase, `0 de ${others.length} ações`);
let applied = 0; let applied = 0;
const warnings = []; const warnings = [];
// Criada sob demanda, na primeira action "zoom" do plano — planos sem zoom não
// ganham uma track extra à toa.
let adjustmentTrackIndex = null;
for (const action of others) { for (const action of others) {
const adjStart = adjustTimeForCuts(action.start, sortedCutsForAdjust); const adjStart = adjustTimeForCuts(action.start, sortedCutsForAdjust);
const adjEnd = adjustTimeForCuts(action.end, sortedCutsForAdjust); const adjEnd = adjustTimeForCuts(action.end, sortedCutsForAdjust);
@@ -258,13 +271,60 @@ async function main() {
warnings.push(`zoom @${action.start} sem params.scale numérico — ignorado.`); warnings.push(`zoom @${action.start} sem params.scale numérico — ignorado.`);
continue; continue;
} }
const videoClip = unwrap( const duration = adjEnd - adjStart;
await discovery.get_clip_at_position.handler({ time_seconds: mid, track_index: args.videoTrack, track_type: "video" }), if (duration <= 0) {
`get_clip_at_position video @${mid}` warnings.push(`zoom @${action.start} com duração inválida (${duration}s) — ignorado.`);
); continue;
}
// A camada de ajuste é criada uma vez, no topo da pilha de vídeo, e
// reaproveitada por todas as actions "zoom" — não altera o clipe
// original, então cortes/re-splits feitos na Fase A/B não a afetam.
if (adjustmentTrackIndex === null) {
const trackResult = unwrap(
await tracks.add_track.handler({ track_type: "video", count: 1 }),
"add_track (camada de ajuste para zoom)"
);
adjustmentTrackIndex = trackResult.totalTracks - 1;
await sleep(BRIDGE_DELAY_MS);
}
unwrap( unwrap(
await trackTargeting.set_clip_scale.handler({ node_id: videoClip.nodeId, scale: scale * 100 }), await playhead.set_playhead_position.handler({ time_seconds: adjStart }),
`set_clip_scale ${videoClip.nodeId}` `set_playhead_position @${adjStart}`
);
await sleep(BRIDGE_DELAY_MS);
unwrap(
await utility.add_adjustment_layer.handler({ track_index: adjustmentTrackIndex }),
`add_adjustment_layer @${adjStart}`
);
await sleep(BRIDGE_DELAY_MS);
const adjClip = unwrap(
await discovery.get_clip_at_position.handler({
time_seconds: adjStart + 0.01,
track_index: adjustmentTrackIndex,
track_type: "video",
}),
`get_clip_at_position (camada de ajuste) @${adjStart}`
);
unwrap(
await timeline.trim_clip.handler({ node_id: adjClip.nodeId, new_out_seconds: duration }),
`trim_clip (camada de ajuste) @${adjStart}`
);
await sleep(BRIDGE_DELAY_MS);
unwrap(
await timeline.move_clip.handler({ node_id: adjClip.nodeId, new_start_seconds: adjStart }),
`move_clip (camada de ajuste) @${adjStart}`
);
await sleep(BRIDGE_DELAY_MS);
unwrap(
await trackTargeting.set_clip_scale.handler({ node_id: adjClip.nodeId, scale: scale * 100 }),
`set_clip_scale (camada de ajuste) ${adjClip.nodeId}`
); );
} else if (action.kind === "text") { } else if (action.kind === "text") {
// Sem API de scripting suportada para clipe de texto a partir de string crua // Sem API de scripting suportada para clipe de texto a partir de string crua
+10 -1
View File
@@ -166,7 +166,16 @@ export function getDiscoveryTools(bridgeOptions: BridgeOptions) {
name: clip.name, name: clip.name,
start: __ticksToSeconds(clip.start.ticks), start: __ticksToSeconds(clip.start.ticks),
end: __ticksToSeconds(clip.end.ticks), end: __ticksToSeconds(clip.end.ticks),
duration: __ticksToSeconds(clip.duration.ticks) inPoint: __ticksToSeconds(clip.inPoint.ticks),
outPoint: __ticksToSeconds(clip.outPoint.ticks),
duration: __ticksToSeconds(clip.duration.ticks),
sourceFile: clip.projectItem && clip.projectItem.getMediaPath
? String(clip.projectItem.getMediaPath() || "")
: "",
sourceProjectItemId: clip.projectItem ? String(clip.projectItem.nodeId || "") : "",
sourceOffline: clip.projectItem && clip.projectItem.isOffline
? !!clip.projectItem.isOffline()
: false
}); });
} }
audioTracks.push({ audioTracks.push({
+124 -46
View File
@@ -418,7 +418,22 @@ export function getTimelineTools(bridgeOptions: BridgeOptions) {
return __error("Refusing trim before mutation: " + beforeKeyframes.outside.length + " effect keyframe(s) would remain outside the visible clip. Use keyframe_policy: preserve only if retaining those keyframes is intentional, or adjust them explicitly with the keyframe tools."); return __error("Refusing trim before mutation: " + beforeKeyframes.outside.length + " effect keyframe(s) would remain outside the visible clip. Use keyframe_policy: preserve only if retaining those keyframes is intentional, or adjust them explicitly with the keyframe tools.");
} }
${args.new_in_seconds !== undefined ? `clip.inPoint = __secondsToTicks(${args.new_in_seconds}).toString();` : "clip.outPoint = __secondsToTicks(" + args.new_out_seconds + ").toString();"} // Set the source point AND the corresponding timeline boundary
// together, computed from the same before-snapshot, instead of
// setting only the source point and trusting Premiere to move the
// timeline boundary on its own: that auto-derivation is exactly
// what silently no-op'd on the timeline side while still moving
// the source point on a real Premiere Pro 26.x install, leaving a
// clip whose visible duration no longer matched its source range.
${args.new_in_seconds !== undefined ? `
var expectedStartOnWrite = before.start + (${args.new_in_seconds} - before.inPoint);
clip.start = __secondsToTicks(expectedStartOnWrite).toString();
clip.inPoint = __secondsToTicks(${args.new_in_seconds}).toString();
` : `
var expectedEndOnWrite = before.end + (${args.new_out_seconds} - before.outPoint);
clip.end = __secondsToTicks(expectedEndOnWrite).toString();
clip.outPoint = __secondsToTicks(${args.new_out_seconds}).toString();
`}
// Re-find the TrackItem after the write. Premiere can replace stale // Re-find the TrackItem after the write. Premiere can replace stale
// DOM references during an edit, especially for audio clips. // DOM references during an edit, especially for audio clips.
@@ -586,10 +601,50 @@ export function getTimelineTools(bridgeOptions: BridgeOptions) {
if (!__razorFps || !isFinite(__razorFps) || __razorFps < 1) __razorFps = 30; if (!__razorFps || !isFinite(__razorFps) || __razorFps < 1) __razorFps = 30;
var __razorFrames = Math.round(cutTicks / frameTicks); var __razorFrames = Math.round(cutTicks / frameTicks);
function __pad2(n) { return n < 10 ? "0" + n : "" + n; } function __pad2(n) { return n < 10 ? "0" + n : "" + n; }
var __razorTc = __pad2(Math.floor(__razorFrames / (__razorFps * 3600))) + ":" +
__pad2(Math.floor((__razorFrames % (__razorFps * 3600)) / (__razorFps * 60))) + ":" + // A sequence whose display format is drop-frame timecode (videoDisplayFormat
__pad2(Math.floor((__razorFrames % (__razorFps * 60)) / __razorFps)) + ":" + // 102 for 29.97 or 106 for 59.94) numbers frames as SMPTE drop-frame: 2 frame
__pad2(__razorFrames % __razorFps); // NUMBERS are skipped at the start of every minute except every 10th. razor()
// parses its timecode string using that same numbering, so handing it a plain
// non-drop-frame HH:MM:SS:FF built from the raw frame count lands the cut at
// the wrong absolute time, worse the further into the sequence the cut is —
// exactly the growing drift this tool used to report as an unexplained no-op
// on some cuts. __razorFrames is the true physical frame count from ticks, so
// it must be converted to the DROP-FRAME-numbered timecode when that display
// format is active, not just formatted with a rounded fps.
var __isDropFrame = false;
try {
// getSettings() existe na sequência do DOM, não na do QE: ler pelo
// objeto errado lança, cai no catch e desliga o drop-frame em
// silêncio, mantendo o desvio que esta correção existe para tirar.
var __seqSettings = domSequence.getSettings ? domSequence.getSettings() : null;
var __displayFormat = __seqSettings ? __seqSettings.videoDisplayFormat : null;
__isDropFrame = __displayFormat === 102 || __displayFormat === 106;
} catch (__dfErr) { __isDropFrame = false; }
var __razorTc;
if (__isDropFrame) {
var __dropPerMin = Math.round(__razorFps * 2 / 30);
var __framesPerMinuteNominal = __razorFps * 60;
var __framesPer10Minutes = __framesPerMinuteNominal * 10 - __dropPerMin * 9;
var __framesPerMinuteDropped = __framesPerMinuteNominal - __dropPerMin;
var __d = Math.floor(__razorFrames / __framesPer10Minutes);
var __m = __razorFrames % __framesPer10Minutes;
var __labelFrames = __razorFrames + __dropPerMin * 9 * __d;
if (__m > __dropPerMin) {
__labelFrames += __dropPerMin * Math.floor((__m - __dropPerMin) / __framesPerMinuteDropped);
}
var __dfFrames = __labelFrames % __razorFps;
var __dfSeconds = Math.floor(__labelFrames / __razorFps) % 60;
var __dfMinutes = Math.floor(__labelFrames / (__razorFps * 60)) % 60;
var __dfHours = Math.floor(__labelFrames / (__razorFps * 3600)) % 24;
__razorTc = __pad2(__dfHours) + ":" + __pad2(__dfMinutes) + ":" + __pad2(__dfSeconds) + ";" + __pad2(__dfFrames);
} else {
__razorTc = __pad2(Math.floor(__razorFrames / (__razorFps * 3600))) + ":" +
__pad2(Math.floor((__razorFrames % (__razorFps * 3600)) / (__razorFps * 60))) + ":" +
__pad2(Math.floor((__razorFrames % (__razorFps * 60)) / __razorFps)) + ":" +
__pad2(__razorFrames % __razorFps);
}
try { try {
track.razor(__razorTc); track.razor(__razorTc);
@@ -750,55 +805,78 @@ export function getTimelineTools(bridgeOptions: BridgeOptions) {
const script = buildToolScript(` const script = buildToolScript(`
var result = __findClip("${escapeForExtendScript(args.node_id)}"); var result = __findClip("${escapeForExtendScript(args.node_id)}");
if (!result) return __error("Clip not found: ${escapeForExtendScript(args.node_id)}"); if (!result) return __error("Clip not found: ${escapeForExtendScript(args.node_id)}");
var clip = result.clip; var clip = result.clip;
var changes = {}; var changes = {};
var missing = [];
// Property lookup uses the fixed index of Premiere's built-in Motion
// and Opacity effects, not displayName: displayName is localized
// (e.g. "Escala"/"Posição"/"Rotação" in pt_BR Premiere), so a plain
// string match against the English name silently matches nothing
// and no-ops on non-English installs. Index 0 of Opacity is its
// only property; Motion is fixed at 0=Position, 1=Scale, 4=Rotation
// across locales — confirmed against a pt_BR install.
function __motionComponent() {
for (var i = 0; i < clip.components.numItems; i++) {
if (clip.components[i].matchName === "AE.ADBE Motion") return clip.components[i];
}
return null;
}
function __opacityComponent() {
for (var i = 0; i < clip.components.numItems; i++) {
if (clip.components[i].matchName === "AE.ADBE Opacity") return clip.components[i];
}
return null;
}
${args.opacity !== undefined ? ` ${args.opacity !== undefined ? `
// Set opacity via Motion component var opacityComp = __opacityComponent();
for (var i = 0; i < clip.components.numItems; i++) { if (!opacityComp || opacityComp.properties.numItems < 1) {
var comp = clip.components[i]; missing.push("opacity: componente Opacity não encontrado no clipe");
if (comp.matchName === "AE.ADBE Opacity" || comp.displayName === "Opacity") { } else {
for (var p = 0; p < comp.properties.numItems; p++) { opacityComp.properties[0].setValue(${args.opacity}, true);
if (comp.properties[p].displayName === "Opacity") { changes.opacity = ${args.opacity};
comp.properties[p].setValue(${args.opacity}, true);
changes.opacity = ${args.opacity};
}
}
}
} }
` : ""} ` : ""}
${args.scale !== undefined || args.position_x !== undefined || args.position_y !== undefined || args.rotation !== undefined ? ` ${args.scale !== undefined || args.position_x !== undefined || args.position_y !== undefined || args.rotation !== undefined ? `
for (var i = 0; i < clip.components.numItems; i++) { var motionComp = __motionComponent();
var comp = clip.components[i]; if (!motionComp) {
if (comp.matchName === "AE.ADBE Motion" || comp.displayName === "Motion") { missing.push("componente Motion não encontrado no clipe");
for (var p = 0; p < comp.properties.numItems; p++) { } else {
var prop = comp.properties[p]; ${args.scale !== undefined ? `
${args.scale !== undefined ? ` if (motionComp.properties.numItems < 2) {
if (prop.displayName === "Scale") { missing.push("scale: propriedade Escala (índice 1) não encontrada");
prop.setValue(${args.scale}, true); } else {
changes.scale = ${args.scale}; motionComp.properties[1].setValue(${args.scale}, true);
}` : ""} changes.scale = ${args.scale};
${args.position_x !== undefined || args.position_y !== undefined ? ` }` : ""}
if (prop.displayName === "Position") { ${args.position_x !== undefined || args.position_y !== undefined ? `
var posVal = prop.getValue(); if (motionComp.properties.numItems < 1) {
var px = posVal && typeof posVal === "object" && posVal.length >= 2 ? posVal[0] : 0; missing.push("position: propriedade Posição (índice 0) não encontrada");
var py = posVal && typeof posVal === "object" && posVal.length >= 2 ? posVal[1] : 0; } else {
${args.position_x !== undefined ? `px = ${args.position_x}; changes.position_x = ${args.position_x};` : ""} var posProp = motionComp.properties[0];
${args.position_y !== undefined ? `py = ${args.position_y}; changes.position_y = ${args.position_y};` : ""} var posVal = posProp.getValue();
prop.setValue([px, py], true); var px = posVal && typeof posVal === "object" && posVal.length >= 2 ? posVal[0] : 0;
}` : ""} var py = posVal && typeof posVal === "object" && posVal.length >= 2 ? posVal[1] : 0;
${args.rotation !== undefined ? ` ${args.position_x !== undefined ? `px = ${args.position_x}; changes.position_x = ${args.position_x};` : ""}
if (prop.displayName === "Rotation") { ${args.position_y !== undefined ? `py = ${args.position_y}; changes.position_y = ${args.position_y};` : ""}
prop.setValue(${args.rotation}, true); posProp.setValue([px, py], true);
changes.rotation = ${args.rotation}; }` : ""}
}` : ""} ${args.rotation !== undefined ? `
} if (motionComp.properties.numItems < 5) {
} missing.push("rotation: propriedade Rotação (índice 4) não encontrada");
} else {
motionComp.properties[4].setValue(${args.rotation}, true);
changes.rotation = ${args.rotation};
}` : ""}
} }
` : ""} ` : ""}
if (missing.length) {
return __error("Não foi possível aplicar todas as propriedades pedidas: " + missing.join("; ") + ". Nenhuma mudança parcial foi reportada como sucesso.");
}
return __result({ updated: true, clipName: clip.name, changes: changes }); return __result({ updated: true, clipName: clip.name, changes: changes });
`); `);
return sendCommand(script, bridgeOptions); return sendCommand(script, bridgeOptions);