";
+ 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 ---------------------------------------------------------------
diff --git a/code/cep-plugin/styles.css b/code/cep-plugin/styles.css
index 574664b..7812d2c 100755
--- a/code/cep-plugin/styles.css
+++ b/code/cep-plugin/styles.css
@@ -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; }
.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); }
+
diff --git a/code/engine/aplicar_plano_de_edicao.py b/code/engine/aplicar_plano_de_edicao.py
new file mode 100644
index 0000000..6c0f38b
--- /dev/null
+++ b/code/engine/aplicar_plano_de_edicao.py
@@ -0,0 +1,85 @@
+"""Entrada de linha de comando: aplica um plano de edição na sequência ativa do Premiere.
+
+Espelha o ``executar_scanner.py``/``executar_retakes.py``: recebe o caminho
+de um JSON de ações (``kind``/``start``/``end``/``params``/``reason``,
+formato das skills de seleção de trechos) e aplica cada ação na sequência
+que estiver ativa no Premiere no momento da execução, via o MCP existente em
+``engine.integracoes.premiere``.
+
+Uso::
+
+ python aplicar_plano_de_edicao.py "/caminho/0E6A8290_edit_actions.json"
+
+Pré-requisito: o Premiere precisa estar aberto com a sequência correta ativa
+e o painel/plugin do MCP em execução — este script não abre o Premiere nem
+troca de sequência.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+CAMINHO_DO_CODIGO = Path(__file__).resolve().parent.parent
+if str(CAMINHO_DO_CODIGO) not in sys.path:
+ sys.path.insert(0, str(CAMINHO_DO_CODIGO))
+
+from engine.editor import AplicadorDePlanoDeEdicao, EscritaNoEditor, LeitorDePlanoDeEdicao, MapeadorDeTempoDeOrigemParaTimeline
+from engine.integracoes.premiere.cliente_mcp import ClienteMCPPorStdio
+from engine.integracoes.premiere.conversores import ConversorDeTimeline
+from engine.integracoes.premiere.leitura import AcessoAoEditor, AcessoATimeline
+from engine.integracoes.premiere.sessao_mcp import SessaoMCP
+
+CAMINHO_DO_SERVIDOR_MCP = CAMINHO_DO_CODIGO / "dist" / "index.js"
+
+
+def executar(caminho_do_plano: Path) -> dict:
+ """Lê o plano em ``caminho_do_plano`` e o aplica na sequência ativa do Premiere.
+
+ Devolve um dicionário pronto para virar JSON, com o resultado de cada
+ ação — para o painel exibir o que foi aplicado e o que falhou.
+ """
+ plano = LeitorDePlanoDeEdicao().ler(caminho_do_plano)
+
+ cliente = ClienteMCPPorStdio(["node", str(CAMINHO_DO_SERVIDOR_MCP)])
+ with SessaoMCP(cliente):
+ acesso_ao_editor = AcessoAoEditor(AcessoATimeline(cliente))
+ aplicador = AplicadorDePlanoDeEdicao(
+ acesso_ao_editor=acesso_ao_editor,
+ conversor_de_timeline=ConversorDeTimeline(),
+ escrita=EscritaNoEditor(cliente),
+ mapeador=MapeadorDeTempoDeOrigemParaTimeline(),
+ )
+ resultado = aplicador.aplicar(plano)
+
+ return {
+ "arquivo_de_origem": plano.arquivo_de_origem,
+ "todas_bem_sucedidas": resultado.todas_bem_sucedidas,
+ "acoes": [
+ {
+ "tipo": item.acao.tipo.value,
+ "inicio": item.acao.inicio,
+ "fim": item.acao.fim,
+ "motivo": item.acao.motivo,
+ "sucesso": item.sucesso,
+ "detalhe": item.detalhe,
+ }
+ for item in resultado.resultados
+ ],
+ }
+
+
+def main() -> None:
+ """Ponto de entrada da linha de comando."""
+ if len(sys.argv) != 2:
+ print("Uso: python aplicar_plano_de_edicao.py ", 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()
diff --git a/code/engine/editor/__init__.py b/code/engine/editor/__init__.py
new file mode 100644
index 0000000..c3780b8
--- /dev/null
+++ b/code/engine/editor/__init__.py
@@ -0,0 +1,35 @@
+"""Aplicação de planos de edição (cortes, zooms, marcadores) na timeline do Premiere.
+
+Lê o JSON de ações produzido pelas skills de seleção de trechos
+(``kind``/``start``/``end``/``params``/``reason``), traduz os tempos de
+origem para a sequência ativa do Premiere e aplica as mutações por meio do
+MCP existente em ``engine.integracoes.premiere``.
+
+Não decide o que cortar — isso já vem pronto no plano. A responsabilidade
+deste módulo é só: ler o plano, achar a posição atual de cada trecho na
+timeline e executar a mutação correta, relatando o que deu certo e o que
+não deu.
+"""
+
+from .aplicador_de_plano_de_edicao import AplicadorDePlanoDeEdicao, ResultadoDaAcao, ResultadoDaAplicacao
+from .erros import ErroDeEdicao, ErroDeEscritaNoEditor, ErroDeMapeamentoDeTempo, ErroDePlanoInvalido
+from .escrita import EscritaNoEditor
+from .leitura import LeitorDePlanoDeEdicao
+from .mapeamento import MapeadorDeTempoDeOrigemParaTimeline
+from .modelos import AcaoDeEdicao, PlanoDeEdicao, TipoDeAcao
+
+__all__ = [
+ "AcaoDeEdicao",
+ "AplicadorDePlanoDeEdicao",
+ "EscritaNoEditor",
+ "ErroDeEdicao",
+ "ErroDeEscritaNoEditor",
+ "ErroDeMapeamentoDeTempo",
+ "ErroDePlanoInvalido",
+ "LeitorDePlanoDeEdicao",
+ "MapeadorDeTempoDeOrigemParaTimeline",
+ "PlanoDeEdicao",
+ "ResultadoDaAcao",
+ "ResultadoDaAplicacao",
+ "TipoDeAcao",
+]
diff --git a/code/engine/editor/aplicador_de_plano_de_edicao.py b/code/engine/editor/aplicador_de_plano_de_edicao.py
new file mode 100644
index 0000000..619302f
--- /dev/null
+++ b/code/engine/editor/aplicador_de_plano_de_edicao.py
@@ -0,0 +1,223 @@
+"""Orquestração da aplicação de um plano de edição na sequência ativa do Premiere.
+
+Liga o plano já validado (:mod:`.leitura`) à tradução de tempos
+(:mod:`.mapeamento`) e à escrita no editor (:mod:`.escrita`), na ordem que
+preserva a validade das posições calculadas — ver docstring de
+:class:`AplicadorDePlanoDeEdicao`.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+from ..dominio import Clipe, Faixa, Timeline
+from ..integracoes.premiere.conversores import ConversorDeTimeline
+from ..integracoes.premiere.leitura import AcessoAoEditor
+from .erros import ErroDeEdicao, ErroDeMapeamentoDeTempo
+from .escrita.escrita_no_editor import EscritaNoEditor
+from .mapeamento.mapeador_de_tempo import MapeadorDeTempoDeOrigemParaTimeline
+from .modelos import AcaoDeEdicao, PlanoDeEdicao, TipoDeAcao
+
+_TOLERANCIA_DE_BORDA_EM_SEGUNDOS = 0.05
+
+
+@dataclass(frozen=True)
+class ResultadoDaAcao:
+ """O que aconteceu ao tentar aplicar uma :class:`AcaoDeEdicao`."""
+
+ acao: AcaoDeEdicao
+ sucesso: bool
+ detalhe: str
+
+
+@dataclass(frozen=True)
+class ResultadoDaAplicacao:
+ """O resultado, ação por ação, de aplicar um :class:`PlanoDeEdicao` inteiro."""
+
+ resultados: tuple[ResultadoDaAcao, ...] = field(default_factory=tuple)
+
+ @property
+ def todas_bem_sucedidas(self) -> bool:
+ """``True`` só se nenhuma ação falhou."""
+ return all(resultado.sucesso for resultado in self.resultados)
+
+ @property
+ def falhas(self) -> tuple[ResultadoDaAcao, ...]:
+ """Só as ações que não foram aplicadas."""
+ return tuple(resultado for resultado in self.resultados if not resultado.sucesso)
+
+
+class AplicadorDePlanoDeEdicao:
+ """Aplica um :class:`PlanoDeEdicao` inteiro na sequência ativa do Premiere.
+
+ Ordem de aplicação, para preservar a validade das posições calculadas:
+
+ 1. Marcadores e zooms primeiro — não mudam a duração da timeline, então
+ a posição calculada para cada um continua válida até o fim.
+ 2. Cortes por último, do fim para o começo (maior início primeiro) —
+ cada corte fecha o espaço que ocupava (ripple), deslocando para a
+ esquerda tudo que vem depois dele. Processar do fim para o começo
+ garante que a posição de um corte ainda não aplicado nunca é afetada
+ pelos cortes já aplicados.
+
+ A cada ação a timeline ativa é lida de novo no Premiere — nunca se
+ reaproveita uma leitura anterior — porque a ação anterior pode ter
+ mudado a posição dos clipes. Uma ação que falha é registrada no
+ resultado e não interrompe as demais; quem chama decide o que fazer com
+ as falhas.
+ """
+
+ def __init__(
+ self,
+ acesso_ao_editor: AcessoAoEditor,
+ conversor_de_timeline: ConversorDeTimeline,
+ escrita: EscritaNoEditor,
+ mapeador: MapeadorDeTempoDeOrigemParaTimeline,
+ ) -> None:
+ self.acesso_ao_editor = acesso_ao_editor
+ self.conversor_de_timeline = conversor_de_timeline
+ self.escrita = escrita
+ self.mapeador = mapeador
+
+ def aplicar(self, plano: PlanoDeEdicao) -> ResultadoDaAplicacao:
+ """Aplica todas as ações de ``plano`` e devolve o resultado de cada uma."""
+ pontuais = [acao for acao in plano.acoes if acao.tipo is not TipoDeAcao.CORTE]
+ cortes = sorted(plano.acoes_do_tipo(TipoDeAcao.CORTE), key=lambda acao: acao.inicio, reverse=True)
+
+ resultados = [self._aplicar_acao_pontual(acao, plano.arquivo_de_origem) for acao in pontuais]
+ resultados += [self._aplicar_corte(acao, plano.arquivo_de_origem) for acao in cortes]
+ return ResultadoDaAplicacao(tuple(resultados))
+
+ def _timeline_atual(self) -> Timeline:
+ """Lê a sequência ativa do Premiere agora, sem cache."""
+ return self.conversor_de_timeline.converter(self.acesso_ao_editor.obter_timeline_ativa())
+
+ def _aplicar_acao_pontual(self, acao: AcaoDeEdicao, arquivo_de_origem: str) -> ResultadoDaAcao:
+ """Aplica um marcador ou zoom, que não precisam dividir nem remover clipes."""
+ try:
+ if acao.tipo is TipoDeAcao.MARCADOR:
+ self._aplicar_marcador(acao, arquivo_de_origem)
+ return ResultadoDaAcao(acao, True, "marcador criado")
+ if acao.tipo is TipoDeAcao.ZOOM:
+ quantidade = self._aplicar_zoom(acao, arquivo_de_origem)
+ return ResultadoDaAcao(acao, True, f"zoom aplicado em {quantidade} clipe(s)")
+ if acao.tipo is TipoDeAcao.TEXTO:
+ self._aplicar_texto_como_marcador(acao, arquivo_de_origem)
+ return ResultadoDaAcao(
+ acao, True,
+ "texto na tela não é suportado pela API de scripting do Premiere; "
+ "registrado como marcador para o editor aplicar manualmente"
+ )
+ raise ErroDeEdicao(f"Tipo de ação sem tratamento: {acao.tipo!r}.")
+ except ErroDeEdicao as erro:
+ return ResultadoDaAcao(acao, False, str(erro))
+
+ def _aplicar_marcador(self, acao: AcaoDeEdicao, arquivo_de_origem: str) -> None:
+ instante = self._instante_na_timeline(acao.inicio, arquivo_de_origem)
+ nome = str(acao.parametros.get("content", "MARCADOR"))
+ self.escrita.adicionar_marcador(instante, nome, acao.motivo)
+
+ def _aplicar_texto_como_marcador(self, acao: AcaoDeEdicao, arquivo_de_origem: str) -> None:
+ instante = self._instante_na_timeline(acao.inicio, arquivo_de_origem)
+ conteudo = str(acao.parametros.get("content", ""))
+ self.escrita.adicionar_marcador(instante, "TEXTO (aplicar manualmente)", f"{conteudo} — {acao.motivo}")
+
+ def _aplicar_zoom(self, acao: AcaoDeEdicao, arquivo_de_origem: str) -> int:
+ fator_de_escala = float(acao.parametros.get("scale", 1.3))
+ faixa_de_video = self._faixa_de_video(self._timeline_atual())
+ clipes = self.mapeador.clipes_do_arquivo(faixa_de_video, arquivo_de_origem)
+ alvos = [clipe for clipe in clipes if self._sobrepoe(clipe, acao.inicio, acao.fim)]
+ if not alvos:
+ raise ErroDeMapeamentoDeTempo(
+ f"Nenhum clipe de vídeo de {arquivo_de_origem!r} sobrepõe {acao.inicio}-{acao.fim}s de origem para o zoom."
+ )
+ for clipe in alvos:
+ self.escrita.aplicar_zoom(clipe.identificador, fator_de_escala)
+ return len(alvos)
+
+ def _aplicar_corte(self, acao: AcaoDeEdicao, arquivo_de_origem: str) -> ResultadoDaAcao:
+ """Remove, em cada faixa afetada, todos os clipes contidos no intervalo do corte."""
+ try:
+ trechos_removidos = 0
+ for faixa in self._timeline_atual().faixas:
+ trechos_removidos += self._cortar_faixa(faixa, acao, arquivo_de_origem)
+ if trechos_removidos == 0:
+ raise ErroDeMapeamentoDeTempo(
+ f"Nenhum clipe de {arquivo_de_origem!r} sobrepõe o intervalo {acao.inicio}-{acao.fim}s de origem."
+ )
+ return ResultadoDaAcao(acao, True, f"{trechos_removidos} trecho(s) removido(s)")
+ except ErroDeEdicao as erro:
+ return ResultadoDaAcao(acao, False, str(erro))
+
+ def _cortar_faixa(self, faixa: Faixa, acao: AcaoDeEdicao, arquivo_de_origem: str) -> int:
+ """Corta o trecho de ``acao`` em uma única faixa e devolve quantos clipes foram removidos."""
+ clipes = self.mapeador.clipes_do_arquivo(faixa, arquivo_de_origem)
+ afetados = [clipe for clipe in clipes if self._sobrepoe(clipe, acao.inicio, acao.fim)]
+ if not afetados:
+ return 0
+
+ primeiro, ultimo = afetados[0], afetados[-1]
+ inicio_na_timeline = self._borda_de_entrada(primeiro, acao.inicio)
+ fim_na_timeline = self._borda_de_saida(ultimo, acao.fim)
+
+ self.escrita.dividir_clipe_se_necessario(inicio_na_timeline, faixa.indice, faixa.tipo)
+ self.escrita.dividir_clipe_se_necessario(fim_na_timeline, faixa.indice, faixa.tipo)
+
+ clipes_apos_a_divisao = self.mapeador.clipes_do_arquivo(
+ next(f for f in self._timeline_atual().faixas if f.identificador == faixa.identificador),
+ arquivo_de_origem,
+ )
+ alvo = [
+ clipe for clipe in clipes_apos_a_divisao
+ if inicio_na_timeline - _TOLERANCIA_DE_BORDA_EM_SEGUNDOS <= clipe.intervalo_na_timeline.inicio
+ and clipe.intervalo_na_timeline.fim <= fim_na_timeline + _TOLERANCIA_DE_BORDA_EM_SEGUNDOS
+ ]
+ duracao_removida = sum(clipe.intervalo_na_timeline.duracao for clipe in alvo)
+ duracao_esperada = fim_na_timeline - inicio_na_timeline
+ if abs(duracao_removida - duracao_esperada) > _TOLERANCIA_DE_BORDA_EM_SEGUNDOS:
+ raise ErroDeMapeamentoDeTempo(
+ f"Na faixa {faixa.identificador!r}, os cortes em {inicio_na_timeline:.3f}s e "
+ f"{fim_na_timeline:.3f}s não isolaram o trecho esperado (esperado {duracao_esperada:.3f}s, "
+ f"encontrado {duracao_removida:.3f}s cobertos por clipe); nada foi removido para evitar "
+ "apagar o trecho errado."
+ )
+ for clipe in alvo:
+ self.escrita.remover_trecho(clipe.identificador)
+ return len(alvo)
+
+ def _instante_na_timeline(self, instante_de_origem: float, arquivo_de_origem: str) -> float:
+ """Traduz um instante de origem usando a primeira faixa de vídeo do arquivo."""
+ faixa_de_video = self._faixa_de_video(self._timeline_atual())
+ clipes = self.mapeador.clipes_do_arquivo(faixa_de_video, arquivo_de_origem)
+ clipe = self.mapeador.localizar_clipe_no_instante(clipes, instante_de_origem)
+ if clipe is None:
+ raise ErroDeMapeamentoDeTempo(
+ f"Nenhum clipe de vídeo de {arquivo_de_origem!r} cobre o instante {instante_de_origem}s de origem."
+ )
+ return self.mapeador.instante_na_timeline(clipe, instante_de_origem)
+
+ def _borda_de_entrada(self, clipe: Clipe, inicio_de_origem: float) -> float:
+ """Posição de entrada do corte na timeline: a borda do clipe se o corte começa antes dele."""
+ if inicio_de_origem <= clipe.intervalo_na_origem.inicio:
+ return clipe.intervalo_na_timeline.inicio
+ return self.mapeador.instante_na_timeline(clipe, inicio_de_origem)
+
+ def _borda_de_saida(self, clipe: Clipe, fim_de_origem: float) -> float:
+ """Posição de saída do corte na timeline: a borda do clipe se o corte termina depois dele."""
+ if fim_de_origem >= clipe.intervalo_na_origem.fim:
+ return clipe.intervalo_na_timeline.fim
+ return self.mapeador.instante_na_timeline(clipe, fim_de_origem)
+
+ @staticmethod
+ def _sobrepoe(clipe: Clipe, inicio: float, fim: float) -> bool:
+ """``True`` se o intervalo de origem do clipe tiver interseção com [inicio, fim)."""
+ intervalo = clipe.intervalo_na_origem
+ return intervalo is not None and intervalo.inicio < fim and inicio < intervalo.fim
+
+ @staticmethod
+ def _faixa_de_video(timeline: Timeline) -> Faixa:
+ """A primeira faixa de vídeo da timeline; é onde marcadores e zooms são resolvidos."""
+ for faixa in timeline.faixas:
+ if faixa.tipo == "video":
+ return faixa
+ raise ErroDeMapeamentoDeTempo("A sequência ativa não tem nenhuma faixa de vídeo.")
diff --git a/code/engine/editor/erros.py b/code/engine/editor/erros.py
new file mode 100644
index 0000000..3f6f65c
--- /dev/null
+++ b/code/engine/editor/erros.py
@@ -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."""
diff --git a/code/engine/editor/escrita/__init__.py b/code/engine/editor/escrita/__init__.py
new file mode 100644
index 0000000..9b106ba
--- /dev/null
+++ b/code/engine/editor/escrita/__init__.py
@@ -0,0 +1,5 @@
+"""Escrita de mutações na sequência ativa do Premiere."""
+
+from .escrita_no_editor import EscritaNoEditor
+
+__all__ = ["EscritaNoEditor"]
diff --git a/code/engine/editor/escrita/escrita_no_editor.py b/code/engine/editor/escrita/escrita_no_editor.py
new file mode 100644
index 0000000..fd39238
--- /dev/null
+++ b/code/engine/editor/escrita/escrita_no_editor.py
@@ -0,0 +1,91 @@
+"""Escrita de mutações na sequência ativa do Premiere via MCP.
+
+Isola o nome das ferramentas MCP e o formato de argumentos externos
+(``split_clip``, ``remove_from_timeline``, ``set_clip_properties``,
+``add_marker``) do resto do domínio. Quem chama esta classe fala só em
+segundos de timeline, identificador de clipe e fator de escala — nunca em
+nomes de ferramenta ou ticks do Premiere.
+"""
+
+from __future__ import annotations
+
+from ...integracoes.premiere.cliente_mcp import ClienteMCP
+from ...integracoes.premiere.erros_mcp import ErroDeFerramentaMCP
+from ..erros import ErroDeEscritaNoEditor
+
+_CORES_POR_NOME = {
+ "revisar": 3, # laranja: chama atenção sem parecer um erro (vermelho)
+ "informar": 5, # branco: marcador neutro
+}
+
+
+class EscritaNoEditor:
+ """Aplica, uma de cada vez, as mutações que o aplicador de plano decide fazer."""
+
+ def __init__(self, cliente_mcp: ClienteMCP) -> None:
+ self.cliente_mcp = cliente_mcp
+
+ def dividir_clipe_se_necessario(self, instante_da_timeline: float, indice_da_faixa: int, tipo_da_faixa: str) -> bool:
+ """Corta, na timeline, o clipe que cobrir ``instante_da_timeline``.
+
+ Devolve ``True`` se um corte novo foi criado e ``False`` quando o
+ instante já era uma borda de clipe (o Premiere recusa dividir onde
+ não há clipe atravessando o ponto) — esse segundo caso não é um erro,
+ é o resultado esperado quando o corte pedido coincide com um corte já
+ existente da análise de cenas.
+ """
+ try:
+ self.cliente_mcp.chamar(
+ "split_clip",
+ {"time_seconds": instante_da_timeline, "track_index": indice_da_faixa, "track_type": tipo_da_faixa},
+ )
+ return True
+ except ErroDeFerramentaMCP:
+ return False
+
+ def remover_trecho(self, identificador_do_clipe: str) -> None:
+ """Remove o clipe da timeline, fechando o espaço deixado (ripple).
+
+ Levanta :class:`ErroDeEscritaNoEditor` se o Premiere recusar a
+ remoção.
+ """
+ try:
+ self.cliente_mcp.chamar("remove_from_timeline", {"node_id": identificador_do_clipe, "ripple": True})
+ except ErroDeFerramentaMCP as erro:
+ raise ErroDeEscritaNoEditor(f"Não foi possível remover o clipe {identificador_do_clipe!r}: {erro}") from erro
+
+ def aplicar_zoom(self, identificador_do_clipe: str, fator_de_escala: float) -> None:
+ """Aplica um punch-in no clipe, escalando-o por ``fator_de_escala`` (1.0 = tamanho original).
+
+ Levanta :class:`ErroDeEscritaNoEditor` se o Premiere recusar a
+ alteração.
+ """
+ try:
+ self.cliente_mcp.chamar(
+ "set_clip_properties", {"node_id": identificador_do_clipe, "scale": fator_de_escala * 100}
+ )
+ except ErroDeFerramentaMCP as erro:
+ raise ErroDeEscritaNoEditor(
+ f"Não foi possível aplicar zoom no clipe {identificador_do_clipe!r}: {erro}"
+ ) from erro
+
+ def adicionar_marcador(self, instante_da_timeline: float, nome: str, comentario: str) -> None:
+ """Cria um marcador de sequência em ``instante_da_timeline``.
+
+ Levanta :class:`ErroDeEscritaNoEditor` se o Premiere recusar a
+ criação do marcador.
+ """
+ try:
+ self.cliente_mcp.chamar(
+ "add_marker",
+ {
+ "time_seconds": instante_da_timeline,
+ "name": nome,
+ "comments": comentario,
+ "color": _CORES_POR_NOME["revisar"],
+ },
+ )
+ except ErroDeFerramentaMCP as erro:
+ raise ErroDeEscritaNoEditor(
+ f"Não foi possível adicionar o marcador em {instante_da_timeline}s: {erro}"
+ ) from erro
diff --git a/code/engine/editor/leitura/__init__.py b/code/engine/editor/leitura/__init__.py
new file mode 100644
index 0000000..6462f3d
--- /dev/null
+++ b/code/engine/editor/leitura/__init__.py
@@ -0,0 +1,5 @@
+"""Leitura e validação de planos de edição."""
+
+from .leitor_de_plano import LeitorDePlanoDeEdicao
+
+__all__ = ["LeitorDePlanoDeEdicao"]
diff --git a/code/engine/editor/leitura/leitor_de_plano.py b/code/engine/editor/leitura/leitor_de_plano.py
new file mode 100644
index 0000000..021762f
--- /dev/null
+++ b/code/engine/editor/leitura/leitor_de_plano.py
@@ -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
diff --git a/code/engine/editor/mapeamento/__init__.py b/code/engine/editor/mapeamento/__init__.py
new file mode 100644
index 0000000..66c8b4b
--- /dev/null
+++ b/code/engine/editor/mapeamento/__init__.py
@@ -0,0 +1,5 @@
+"""Tradução de tempos de origem para posições na timeline ativa."""
+
+from .mapeador_de_tempo import MapeadorDeTempoDeOrigemParaTimeline
+
+__all__ = ["MapeadorDeTempoDeOrigemParaTimeline"]
diff --git a/code/engine/editor/mapeamento/mapeador_de_tempo.py b/code/engine/editor/mapeamento/mapeador_de_tempo.py
new file mode 100644
index 0000000..d48833f
--- /dev/null
+++ b/code/engine/editor/mapeamento/mapeador_de_tempo.py
@@ -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)
diff --git a/code/engine/editor/modelos.py b/code/engine/editor/modelos.py
new file mode 100644
index 0000000..e5cab4f
--- /dev/null
+++ b/code/engine/editor/modelos.py
@@ -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)
diff --git a/code/engine/gerar_relatorio_de_analises.py b/code/engine/gerar_relatorio_de_analises.py
new file mode 100644
index 0000000..c89fca7
--- /dev/null
+++ b/code/engine/gerar_relatorio_de_analises.py
@@ -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()
diff --git a/code/engine/integracoes/premiere/cliente_mcp.py b/code/engine/integracoes/premiere/cliente_mcp.py
index cd25718..dac4b86 100644
--- a/code/engine/integracoes/premiere/cliente_mcp.py
+++ b/code/engine/integracoes/premiere/cliente_mcp.py
@@ -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
import json
import subprocess
from typing import Any
+from .erros_mcp import ErroDeConexaoMCP, ErroDeFerramentaMCP
+
class ClienteMCP(ABC):
"""Define a comunicação técnica com o servidor MCP."""
@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
- def desconectar(self) -> None: ...
+ def desconectar(self) -> None:
+ """Encerra a conexão com o servidor MCP, se houver uma ativa."""
@abstractmethod
- def esta_conectado(self) -> bool: ...
+ def esta_conectado(self) -> bool:
+ """Indica se há uma conexão ativa com o servidor MCP."""
@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):
@@ -30,6 +50,11 @@ class ClienteMCPPorStdio(ClienteMCP):
self._proximo_id = 1
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:
return
self._processo = subprocess.Popen(
@@ -43,29 +68,42 @@ class ClienteMCPPorStdio(ClienteMCP):
self._enviar("initialize", {"protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "engine", "version": "0.1.0"}})
def desconectar(self) -> None:
+ """Encerra o processo do servidor MCP, se houver um em execução."""
if self._processo is not None:
self._processo.terminate()
self._processo = None
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
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():
- 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 {}})
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", {})
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:
- raise RuntimeError("Processo MCP indisponível.")
+ raise ErroDeConexaoMCP("Processo MCP indisponível.")
identificador = self._proximo_id
self._proximo_id += 1
self._processo.stdin.write(json.dumps({"jsonrpc": "2.0", "id": identificador, "method": metodo, "params": parametros}) + "\n")
self._processo.stdin.flush()
linha = self._processo.stdout.readline()
if not linha:
- raise RuntimeError("O MCP encerrou sem retornar resposta.")
+ raise ErroDeConexaoMCP("O MCP encerrou sem retornar resposta.")
return json.loads(linha)
diff --git a/code/engine/testes/duplos_de_premiere.py b/code/engine/testes/duplos_de_premiere.py
new file mode 100644
index 0000000..62a9a1a
--- /dev/null
+++ b/code/engine/testes/duplos_de_premiere.py
@@ -0,0 +1,182 @@
+"""Dublê de teste do cliente MCP do Premiere, para testar o editor sem o app aberto.
+
+Simula, em memória, uma sequência com faixas de vídeo e áudio cujos clipes
+têm posição na timeline diferente da posição no arquivo de origem — o mesmo
+descompasso encontrado no projeto real entre ``inicio_na_timeline`` e
+``inicio_na_origem`` do banco de análises. Implementa só as ferramentas MCP
+que o módulo ``engine.editor`` usa: ``get_active_sequence``, ``split_clip``,
+``remove_from_timeline``, ``set_clip_properties`` e ``add_marker``.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from engine.integracoes.premiere.cliente_mcp import ClienteMCP
+from engine.integracoes.premiere.erros_mcp import ErroDeFerramentaMCP
+
+
+@dataclass
+class _ClipeFalso:
+ """Representação mínima de um clipe dentro do dublê de sequência."""
+
+ identificador: str
+ nome: str
+ inicio_na_timeline: float
+ fim_na_timeline: float
+ inicio_na_origem: float
+ fim_na_origem: float
+
+
+class ClienteMCPFalso(ClienteMCP):
+ """Simula o servidor MCP do Premiere para testes do módulo ``engine.editor``.
+
+ Vem pré-carregado com duas faixas (``video_0`` e ``audio_0``), cada uma
+ com dois clipes do mesmo arquivo fictício, reproduzindo o descompasso
+ real entre tempo de origem e tempo de timeline. Registra toda chamada em
+ :attr:`chamadas` para os testes inspecionarem ordem e argumentos.
+ """
+
+ _ARQUIVO = "0E6A8290.mp4"
+
+ def __init__(self) -> None:
+ self.chamadas: list[tuple[str, dict[str, Any]]] = []
+ self._proximo_id = 0
+ self._faixas: dict[str, list[_ClipeFalso]] = {
+ "video_0": [
+ self._novo_clipe(self._ARQUIVO, 0.0, 38.3, 2.3, 40.6, "clipe_a_video"),
+ self._novo_clipe(self._ARQUIVO, 58.1, 66.7, 86.2, 94.8, "clipe_b_video"),
+ ],
+ "audio_0": [
+ self._novo_clipe(self._ARQUIVO, 0.0, 38.3, 2.3, 40.6, "clipe_a_audio"),
+ self._novo_clipe(self._ARQUIVO, 58.1, 66.7, 86.2, 94.8, "clipe_b_audio"),
+ ],
+ }
+
+ @property
+ def chamadas_de_remocao(self) -> list[dict[str, Any]]:
+ """Argumentos de cada chamada a ``remove_from_timeline``, na ordem em que ocorreram."""
+ return [argumentos for nome, argumentos in self.chamadas if nome == "remove_from_timeline"]
+
+ def conectar(self) -> None:
+ """Não há conexão real; existe só para satisfazer a interface de :class:`ClienteMCP`."""
+
+ def desconectar(self) -> None:
+ """Não há conexão real; existe só para satisfazer a interface de :class:`ClienteMCP`."""
+
+ def esta_conectado(self) -> bool:
+ """Sempre conectado: este dublê não simula falha de conexão."""
+ return True
+
+ def chamar(self, nome_da_ferramenta: str, argumentos: dict[str, Any] | None = None) -> dict[str, Any]:
+ """Despacha para o simulador da ferramenta pedida e registra a chamada."""
+ argumentos = argumentos or {}
+ self.chamadas.append((nome_da_ferramenta, argumentos))
+ despachantes = {
+ "get_active_sequence": self._simular_get_active_sequence,
+ "split_clip": self._simular_split_clip,
+ "remove_from_timeline": self._simular_remove_from_timeline,
+ "set_clip_properties": self._simular_set_clip_properties,
+ "add_marker": self._simular_add_marker,
+ }
+ simulador = despachantes.get(nome_da_ferramenta)
+ if simulador is None:
+ raise ErroDeFerramentaMCP(f"Ferramenta MCP não simulada neste dublê de teste: {nome_da_ferramenta!r}.")
+ return simulador(argumentos)
+
+ def _novo_clipe(self, nome, tl_inicio, tl_fim, origem_inicio, origem_fim, identificador=None) -> _ClipeFalso:
+ """Cria um clipe falso com um identificador único (ou o informado, para os fixtures)."""
+ if identificador is None:
+ self._proximo_id += 1
+ identificador = f"clipe_falso_{self._proximo_id}"
+ return _ClipeFalso(identificador, nome, tl_inicio, tl_fim, origem_inicio, origem_fim)
+
+ def _simular_get_active_sequence(self, argumentos: dict[str, Any]) -> dict[str, Any]:
+ """Devolve a sequência atual no formato que :class:`ConversorDeTimeline` espera."""
+ del argumentos
+
+ def _faixa_como_dict(identificador: str, tipo: str, indice: int) -> dict[str, Any]:
+ return {
+ "id": identificador,
+ "name": identificador,
+ "index": indice,
+ "clips": [
+ {
+ "nodeId": clipe.identificador,
+ "name": clipe.nome,
+ "start": clipe.inicio_na_timeline,
+ "end": clipe.fim_na_timeline,
+ "inPoint": clipe.inicio_na_origem,
+ "outPoint": clipe.fim_na_origem,
+ }
+ for clipe in self._faixas[identificador]
+ ],
+ }
+
+ dados = {
+ "id": "sequencia_falsa",
+ "name": "Sequência de teste",
+ "duration": 200.0,
+ "frameRate": 29.97,
+ "frameSizeHorizontal": 3840,
+ "frameSizeVertical": 2160,
+ "videoTracks": [_faixa_como_dict("video_0", "video", 0)],
+ "audioTracks": [_faixa_como_dict("audio_0", "audio", 0)],
+ }
+ return {"structuredContent": {"data": dados}}
+
+ def _simular_split_clip(self, argumentos: dict[str, Any]) -> dict[str, Any]:
+ """Divide, na faixa pedida, o clipe que cobrir estritamente o instante informado."""
+ identificador_da_faixa = self._identificador_da_faixa(argumentos["track_index"], argumentos["track_type"])
+ instante = float(argumentos["time_seconds"])
+ faixa = self._faixas[identificador_da_faixa]
+ for posicao, clipe in enumerate(faixa):
+ if clipe.inicio_na_timeline < instante < clipe.fim_na_timeline:
+ deslocamento = instante - clipe.inicio_na_timeline
+ self._proximo_id += 1
+ parte_esquerda = _ClipeFalso(
+ f"{clipe.identificador}_esq{self._proximo_id}", clipe.nome,
+ clipe.inicio_na_timeline, instante,
+ clipe.inicio_na_origem, clipe.inicio_na_origem + deslocamento,
+ )
+ self._proximo_id += 1
+ parte_direita = _ClipeFalso(
+ f"{clipe.identificador}_dir{self._proximo_id}", clipe.nome,
+ instante, clipe.fim_na_timeline,
+ clipe.inicio_na_origem + deslocamento, clipe.fim_na_origem,
+ )
+ faixa[posicao:posicao + 1] = [parte_esquerda, parte_direita]
+ return {"split": True}
+ raise ErroDeFerramentaMCP(f"Nenhum clipe da faixa {identificador_da_faixa!r} cobre estritamente {instante}s.")
+
+ def _simular_remove_from_timeline(self, argumentos: dict[str, Any]) -> dict[str, Any]:
+ """Remove o clipe pedido e, com ``ripple``, fecha o espaço deixado na mesma faixa."""
+ identificador_do_clipe = argumentos["node_id"]
+ for identificador_da_faixa, clipes in self._faixas.items():
+ for posicao, clipe in enumerate(clipes):
+ if clipe.identificador == identificador_do_clipe:
+ duracao = clipe.fim_na_timeline - clipe.inicio_na_timeline
+ del clipes[posicao]
+ if argumentos.get("ripple"):
+ for outro in clipes:
+ if outro.inicio_na_timeline >= clipe.fim_na_timeline:
+ outro.inicio_na_timeline -= duracao
+ outro.fim_na_timeline -= duracao
+ return {"removed": True}
+ raise ErroDeFerramentaMCP(f"Clipe não encontrado para remoção: {identificador_do_clipe!r}.")
+
+ def _simular_set_clip_properties(self, argumentos: dict[str, Any]) -> dict[str, Any]:
+ """Aceita a alteração de propriedades sem simular geometria (não é usada pelos testes)."""
+ del argumentos
+ return {"updated": True}
+
+ def _simular_add_marker(self, argumentos: dict[str, Any]) -> dict[str, Any]:
+ """Aceita a criação de marcador sem manter estado (a chamada já fica em :attr:`chamadas`)."""
+ del argumentos
+ return {"added": True}
+
+ @staticmethod
+ def _identificador_da_faixa(indice: int, tipo: str) -> str:
+ """Reconstrói o identificador de faixa (``video_0``/``audio_0``) a partir de índice e tipo."""
+ return f"{tipo}_{indice}"
diff --git a/code/engine/testes/test_aplicador_de_plano_de_edicao.py b/code/engine/testes/test_aplicador_de_plano_de_edicao.py
new file mode 100644
index 0000000..a6e99b8
--- /dev/null
+++ b/code/engine/testes/test_aplicador_de_plano_de_edicao.py
@@ -0,0 +1,97 @@
+"""Testes do orquestrador de aplicação de plano de edição (sem Premiere real).
+
+Usa um ClienteMCP falso, em memória, que simula uma sequência de vídeo com
+os mesmos tempos "torcidos" do projeto real: a timeline lida pelo scanner
+não bate com o tempo do arquivo de origem (ver relatório da análise), então
+os testes aqui garantem que o aplicador sempre recalcula a posição a partir
+do clipe lido ao vivo, nunca de um tempo fixo.
+"""
+
+import unittest
+from typing import Any
+
+from engine.editor.aplicador_de_plano_de_edicao import AplicadorDePlanoDeEdicao
+from engine.editor.escrita import EscritaNoEditor
+from engine.editor.mapeamento import MapeadorDeTempoDeOrigemParaTimeline
+from engine.editor.modelos import AcaoDeEdicao, PlanoDeEdicao, TipoDeAcao
+from engine.integracoes.premiere.conversores import ConversorDeTimeline
+from engine.integracoes.premiere.leitura import AcessoAoEditor, AcessoATimeline
+from engine.testes.duplos_de_premiere import ClienteMCPFalso
+
+
+class TesteAplicadorDePlanoDeEdicao(unittest.TestCase):
+ """Cobre a ordem de aplicação e o recálculo de posições a cada ação."""
+
+ def setUp(self):
+ self.cliente = ClienteMCPFalso()
+ self.aplicador = AplicadorDePlanoDeEdicao(
+ acesso_ao_editor=AcessoAoEditor(AcessoATimeline(self.cliente)),
+ conversor_de_timeline=ConversorDeTimeline(),
+ escrita=EscritaNoEditor(self.cliente),
+ mapeador=MapeadorDeTempoDeOrigemParaTimeline(),
+ )
+
+ def test_corte_remove_os_clipes_do_intervalo_e_fecha_o_espaco(self):
+ plano = PlanoDeEdicao("0E6A8290.mp4", (
+ AcaoDeEdicao(TipoDeAcao.CORTE, inicio=0.0, fim=40.6, motivo="bastidor"),
+ ))
+ resultado = self.aplicador.aplicar(plano)
+ self.assertTrue(resultado.todas_bem_sucedidas)
+ self.assertNotIn("000f4763_video", [c["node_id"] for c in self.cliente.chamadas_de_remocao])
+ # o clipe único que cobre 2.3-40.6s de origem foi removido em ambas as faixas
+ self.assertEqual(
+ {c["node_id"] for c in self.cliente.chamadas_de_remocao},
+ {"clipe_a_video", "clipe_a_audio"},
+ )
+
+ def test_marcador_e_zoom_sao_aplicados_antes_dos_cortes(self):
+ plano = PlanoDeEdicao("0E6A8290.mp4", (
+ AcaoDeEdicao(TipoDeAcao.CORTE, inicio=0.0, fim=40.6, motivo="bastidor"),
+ AcaoDeEdicao(TipoDeAcao.MARCADOR, inicio=90.0, fim=90.5, motivo="conferir emenda", parametros={"content": "EMENDA"}),
+ AcaoDeEdicao(TipoDeAcao.ZOOM, inicio=86.42, fim=91.5, motivo="ênfase", parametros={"scale": 1.15}),
+ ))
+ resultado = self.aplicador.aplicar(plano)
+ self.assertTrue(resultado.todas_bem_sucedidas, msg=[r.detalhe for r in resultado.falhas])
+ # marcador e zoom aconteceram antes do corte na sequência de chamadas
+ indice_do_marcador = next(i for i, c in enumerate(self.cliente.chamadas) if c[0] == "add_marker")
+ indice_do_zoom = next(i for i, c in enumerate(self.cliente.chamadas) if c[0] == "set_clip_properties")
+ indice_da_remocao = next(i for i, c in enumerate(self.cliente.chamadas) if c[0] == "remove_from_timeline")
+ self.assertLess(indice_do_marcador, indice_da_remocao)
+ self.assertLess(indice_do_zoom, indice_da_remocao)
+
+ def test_dois_cortes_sao_aplicados_do_fim_para_o_comeco(self):
+ plano = PlanoDeEdicao("0E6A8290.mp4", (
+ AcaoDeEdicao(TipoDeAcao.CORTE, inicio=0.0, fim=40.6, motivo="primeiro trecho"),
+ AcaoDeEdicao(TipoDeAcao.CORTE, inicio=86.2, fim=94.8, motivo="segundo trecho"),
+ ))
+ resultado = self.aplicador.aplicar(plano)
+ self.assertTrue(resultado.todas_bem_sucedidas, msg=[r.detalhe for r in resultado.falhas])
+ remocoes = [c["node_id"] for c in self.cliente.chamadas_de_remocao]
+ # o corte com início maior (86.2s) precisa ser removido antes do de início 0.0s
+ self.assertLess(remocoes.index("clipe_b_video"), remocoes.index("clipe_a_video"))
+
+ def test_corte_sem_clipe_sobreposto_falha_sem_interromper_o_resto(self):
+ plano = PlanoDeEdicao("0E6A8290.mp4", (
+ AcaoDeEdicao(TipoDeAcao.CORTE, inicio=1000.0, fim=1010.0, motivo="fora do vídeo"),
+ AcaoDeEdicao(TipoDeAcao.MARCADOR, inicio=90.0, fim=90.5, motivo="ok", parametros={"content": "X"}),
+ ))
+ resultado = self.aplicador.aplicar(plano)
+ self.assertFalse(resultado.todas_bem_sucedidas)
+ self.assertEqual(len(resultado.falhas), 1)
+ self.assertIn("Nenhum clipe", resultado.falhas[0].detalhe)
+ # a ação válida (marcador) ainda foi aplicada
+ self.assertTrue(any(c[0] == "add_marker" for c in self.cliente.chamadas))
+
+ def test_texto_e_convertido_em_marcador_por_falta_de_suporte_do_premiere(self):
+ plano = PlanoDeEdicao("0E6A8290.mp4", (
+ AcaoDeEdicao(TipoDeAcao.TEXTO, inicio=90.0, fim=91.0, motivo="reforçar termo", parametros={"content": "MASTOPEXIA"}),
+ ))
+ resultado = self.aplicador.aplicar(plano)
+ self.assertTrue(resultado.todas_bem_sucedidas)
+ chamadas_de_marcador = [c for c in self.cliente.chamadas if c[0] == "add_marker"]
+ self.assertEqual(len(chamadas_de_marcador), 1)
+ self.assertIn("MASTOPEXIA", chamadas_de_marcador[0][1]["comments"])
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/code/engine/testes/test_gerar_relatorio_de_analises.py b/code/engine/testes/test_gerar_relatorio_de_analises.py
new file mode 100644
index 0000000..c83ea0a
--- /dev/null
+++ b/code/engine/testes/test_gerar_relatorio_de_analises.py
@@ -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()
diff --git a/code/engine/testes/test_leitor_de_plano_de_edicao.py b/code/engine/testes/test_leitor_de_plano_de_edicao.py
new file mode 100644
index 0000000..ec3e00b
--- /dev/null
+++ b/code/engine/testes/test_leitor_de_plano_de_edicao.py
@@ -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()
diff --git a/code/engine/testes/test_mapeador_de_tempo.py b/code/engine/testes/test_mapeador_de_tempo.py
new file mode 100644
index 0000000..aadb533
--- /dev/null
+++ b/code/engine/testes/test_mapeador_de_tempo.py
@@ -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()