"""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``, ``remove_linked_from_timeline``, ``duplicate_linked_range_to_start``, ``set_clip_properties``, ``add_marker``, ``add_track``, ``set_playhead_position``, ``add_adjustment_layer``, ``get_clip_at_position``, ``trim_clip`` e ``move_clip``. """ 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, simular_falha_razor: bool = False, ) -> 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``. ``simular_falha_razor`` reproduz o bug QE do Premiere 26.x: o ``razor_all_tracks`` reporta clipes elegíveis mas não corta nenhum (``razored=0, eligibleTracks=N``).""" self.sem_origem_no_audio = sem_origem_no_audio self.desvio_do_split = desvio_do_split self.simular_falha_razor = simular_falha_razor self.chamadas: list[tuple[str, dict[str, Any]]] = [] self._proximo_id = 0 self._historico: list[dict[str, list[_ClipeFalso]]] = [] self._playhead = 0.0 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 remoção com ripple, na ordem em que ocorreram.""" return [argumentos for nome, argumentos in self.chamadas if nome in {"ripple_delete", "remove_from_timeline", "remove_linked_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_ripple_delete, "remove_linked_from_timeline": self._simular_remocao_vinculada, "duplicate_linked_range_to_start": self._simular_gancho_vinculado_no_inicio, "set_clip_properties": self._simular_set_clip_properties, "add_marker": self._simular_add_marker, "undo": self._simular_undo, "add_track": self._simular_add_track, "set_playhead_position": self._simular_set_playhead_position, "add_adjustment_layer": self._simular_add_adjustment_layer, "razor_all_tracks": self._simular_razor_all_tracks, "get_clip_at_position": self._simular_get_clip_at_position, "trim_clip": self._simular_trim_clip, "move_clip": self._simular_move_clip, } 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]], } faixas_de_video = sorted( (identificador for identificador in self._faixas if identificador.startswith("video_")), key=lambda identificador: int(identificador.split("_")[1]), ) faixas_de_audio = sorted( (identificador for identificador in self._faixas if identificador.startswith("audio_")), key=lambda identificador: int(identificador.split("_")[1]), ) dados = { "id": "sequencia_falsa", "name": "Sequência de teste", "duration": 200.0, "frameRate": 29.97, "frameSizeHorizontal": 3840, "frameSizeVertical": 2160, "videoTracks": [ _faixa_como_dict(identificador, "video", int(identificador.split("_")[1])) for identificador in faixas_de_video ], "audioTracks": [ _faixa_como_dict(identificador, "audio", int(identificador.split("_")[1])) for identificador in faixas_de_audio ], } 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_razor_all_tracks(self, argumentos: dict[str, Any]) -> dict[str, Any]: """Divide todos os clipes de todas as faixas no instante informado. Reproduz o ``razor_all_tracks`` do bridge real: corta todas as faixas de vídeo e/ou áudio no mesmo ponto de tempo, garantindo sincronização entre elas. Quando ``simular_falha_razor`` está ativo, reporta clipes elegíveis mas não corta nenhum, reproduzindo o bug QE do Premiere 26.x. """ instante_pedido = float(argumentos["time_seconds"]) track_type = argumentos.get("track_type", "both") elegiveis = 0 for identificador_da_faixa, faixa in self._faixas.items(): if track_type == "video" and not identificador_da_faixa.startswith("video_"): continue if track_type == "audio" and not identificador_da_faixa.startswith("audio_"): continue for clipe in faixa: if clipe.inicio_na_timeline < instante_pedido < clipe.fim_na_timeline: elegiveis += 1 break if self.simular_falha_razor: return {"razored": 0, "eligibleTracks": elegiveis, "verified": False} cortados = 0 self._salvar_estado() for identificador_da_faixa, faixa in self._faixas.items(): if track_type == "video" and not identificador_da_faixa.startswith("video_"): continue if track_type == "audio" and not identificador_da_faixa.startswith("audio_"): continue for posicao, clipe in enumerate(faixa): if clipe.inicio_na_timeline < instante_pedido < clipe.fim_na_timeline: instante = min( max(instante_pedido + self.desvio_do_split, clipe.inicio_na_timeline + 0.001), clipe.fim_na_timeline - 0.001, ) 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] cortados += 1 break return {"razored": cortados, "eligibleTracks": elegiveis, "verified": True} 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_remocao_vinculada(self, argumentos: dict[str, Any]) -> dict[str, Any]: """Remove o par de vídeo e áudio, fechando o espaço nos dois lados.""" identificadores = argumentos["node_ids"] if len(identificadores) != 2: raise ErroDeFerramentaMCP("O par vinculado precisa conter dois clipes.") for identificador in identificadores: self._simular_ripple_delete({"node_id": identificador, "ripple": True}) return {"removed": True, "verified": True} def _simular_gancho_vinculado_no_inicio(self, argumentos: dict[str, Any]) -> dict[str, Any]: """Duplica um par já isolado no início e preserva sua ocorrência original. Espelha o bridge real: não recebe limites de origem à parte, lê o in/out do próprio clipe. Por isso os identificadores precisam já corresponder a um par isolado por ``split_clip`` — o mesmo clipe inteiro, sem sobra de trecho vizinho. """ identificadores = argumentos["node_ids"] encontrados: list[tuple[str, _ClipeFalso]] = [] for identificador_da_faixa, clipes in self._faixas.items(): for clipe in clipes: if clipe.identificador in identificadores: encontrados.append((identificador_da_faixa, clipe)) if len(encontrados) != 2 or {faixa.split("_")[0] for faixa, _ in encontrados} != {"video", "audio"}: raise ErroDeFerramentaMCP("O gancho exige um par de vídeo e áudio existente.") inicios = {round(clipe.inicio_na_timeline, 6) for _, clipe in encontrados} fins = {round(clipe.fim_na_timeline, 6) for _, clipe in encontrados} if len(inicios) != 1 or len(fins) != 1: raise ErroDeFerramentaMCP("O par não ocupa o mesmo intervalo.") if next(iter(inicios)) <= 0.0: raise ErroDeFerramentaMCP("O trecho pedido já está no início da sequência.") self._salvar_estado() duracao = next(iter(fins)) - next(iter(inicios)) novos_ids: list[str] = [] for identificador_da_faixa, clipe in encontrados: faixa = self._faixas[identificador_da_faixa] for outro in faixa: outro.inicio_na_timeline += duracao outro.fim_na_timeline += duracao novo = self._novo_clipe( clipe.nome, 0.0, duracao, clipe.inicio_na_origem, clipe.fim_na_origem, ) faixa.append(novo) faixa.sort(key=lambda item: item.inicio_na_timeline) novos_ids.append(novo.identificador) return {"duplicated": True, "verified": True, "insertedNodeIds": novos_ids} 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_track(self, argumentos: dict[str, Any]) -> dict[str, Any]: """Cria ``count`` faixas de vídeo novas, vazias, no topo da pilha. Espelha ``add_track`` do bridge real (usado pelo `.mjs` para criar a track de camada de ajuste): devolve ``totalTracks`` já contando as novas, de forma que ``totalTracks - 1`` seja o índice da última faixa criada. """ if argumentos.get("track_type") != "video": raise ErroDeFerramentaMCP("Este dublê só simula add_track para vídeo.") contagem = int(argumentos.get("count", 1)) indices_existentes = [ int(identificador.split("_")[1]) for identificador in self._faixas if identificador.startswith("video_") ] proximo_indice = max(indices_existentes, default=-1) + 1 for deslocamento in range(contagem): self._faixas[f"video_{proximo_indice + deslocamento}"] = [] total_de_faixas_de_video = proximo_indice + contagem return {"totalTracks": total_de_faixas_de_video} def _simular_set_playhead_position(self, argumentos: dict[str, Any]) -> dict[str, Any]: """Guarda a posição do playhead, usada por :meth:`_simular_add_adjustment_layer`.""" self._playhead = float(argumentos["time_seconds"]) return {"positioned": True} def _simular_add_adjustment_layer(self, argumentos: dict[str, Any]) -> dict[str, Any]: """Cria uma camada de ajuste na track pedida, no instante atual do playhead. A camada nasce com uma duração fixa arbitrária (o `.mjs` real conta com isso: sempre chama ``trim_clip``/``move_clip`` em seguida para ajustar duração e posição definitivas). """ indice_da_faixa = int(argumentos["track_index"]) identificador_da_faixa = self._identificador_da_faixa(indice_da_faixa, "video") if identificador_da_faixa not in self._faixas: raise ErroDeFerramentaMCP(f"Faixa de vídeo inexistente: {identificador_da_faixa!r}.") duracao_inicial = 5.0 clipe = self._novo_clipe( "Adjustment Layer", self._playhead, self._playhead + duracao_inicial, self._playhead, self._playhead + duracao_inicial, ) self._faixas[identificador_da_faixa].append(clipe) self._faixas[identificador_da_faixa].sort(key=lambda item: item.inicio_na_timeline) return {"added": True, "nodeId": clipe.identificador} def _simular_get_clip_at_position(self, argumentos: dict[str, Any]) -> dict[str, Any]: """Devolve o clipe da faixa pedida que cobre ``time_seconds``.""" identificador_da_faixa = self._identificador_da_faixa(argumentos["track_index"], argumentos["track_type"]) instante = float(argumentos["time_seconds"]) faixa = self._faixas.get(identificador_da_faixa, []) clipe = next( (c for c in faixa if c.inicio_na_timeline <= instante < c.fim_na_timeline), None ) if clipe is None: raise ErroDeFerramentaMCP(f"Nenhum clipe em {identificador_da_faixa!r} no instante {instante}s.") return {"nodeId": clipe.identificador} def _simular_trim_clip(self, argumentos: dict[str, Any]) -> dict[str, Any]: """Ajusta a duração do clipe, medida a partir do próprio início atual.""" clipe = self._localizar_clipe(argumentos["node_id"]) nova_duracao = float(argumentos["new_out_seconds"]) clipe.fim_na_timeline = clipe.inicio_na_timeline + nova_duracao clipe.fim_na_origem = clipe.inicio_na_origem + nova_duracao return {"trimmed": True} def _simular_move_clip(self, argumentos: dict[str, Any]) -> dict[str, Any]: """Move o clipe para ``new_start_seconds``, preservando sua duração.""" clipe = self._localizar_clipe(argumentos["node_id"]) duracao = clipe.fim_na_timeline - clipe.inicio_na_timeline novo_inicio = float(argumentos["new_start_seconds"]) clipe.inicio_na_timeline = novo_inicio clipe.fim_na_timeline = novo_inicio + duracao for faixa in self._faixas.values(): faixa.sort(key=lambda item: item.inicio_na_timeline) return {"moved": True} def _localizar_clipe(self, identificador_do_clipe: str) -> _ClipeFalso: """Encontra um clipe pelo identificador em qualquer faixa, ou levanta erro.""" for faixa in self._faixas.values(): for clipe in faixa: if clipe.identificador == identificador_do_clipe: return clipe raise ErroDeFerramentaMCP(f"Clipe não encontrado: {identificador_do_clipe!r}.") 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}"