"""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}"