jhonny-editor/code/engine/testes/duplos_de_premiere.py

"""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,
    ) -> 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._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,
            "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_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}"