jhonny-editor/code/engine/aplicar_plano_de_edicao.py

"""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 argparse
import json
import os
import shutil
import sqlite3
import sys
from pathlib import Path

CAMINHO_DO_CODIGO = Path(__file__).resolve().parent.parent
if str(CAMINHO_DO_CODIGO) not in sys.path:
    sys.path.insert(0, str(CAMINHO_DO_CODIGO))

from engine.editor import (
    AplicadorDePlanoDeEdicao,
    BackupDaSequencia,
    EscritaNoEditor,
    ErroDePlanoInvalido,
    LeitorDePlanoDeEdicao,
    MapeadorDeTempoDeOrigemParaTimeline,
    PlanoDeEdicao,
    ResultadoDaVerificacao,
    TipoDeAcao,
    ValidadorSemanticoDePlano,
    VerificadorDaAplicacao,
)
from engine.integracoes.premiere.cliente_mcp import ClienteMCPPorStdio
from engine.integracoes.premiere.cliente_mcp_com_pausa import ClienteMCPComPausa
from engine.integracoes.premiere.conversores import ConversorDeTimeline
from engine.integracoes.premiere.leitura import AcessoAoEditor, AcessoATimeline
from engine.integracoes.premiere.sessao_mcp import SessaoMCP

CAMINHO_DO_SERVIDOR_MCP = CAMINHO_DO_CODIGO / "dist" / "index.js"
# Mesmo banco de análises do Scanner: o plano aplicado passa a ficar ao lado
# das falas e evidências que o originaram.
CAMINHO_DO_BANCO = CAMINHO_DO_CODIGO.parent / ".jhonny" / "analises.db"


def _garantir_backup_para_ganchos(plano: PlanoDeEdicao, backup_criado: bool) -> None:
    """Bloqueia ganchos quando não há uma cópia recuperável da sequência."""
    if plano.acoes_do_tipo(TipoDeAcao.GANCHO) and not backup_criado:
        raise ErroDePlanoInvalido(
            "A ação hook não foi aplicada porque o Premiere não confirmou o backup da sequência."
        )


def _validar_integridade_semantica(
    plano: PlanoDeEdicao,
    plano_id: int | None,
    caminho_do_banco: Path,
    exigir_validacao: bool = False,
) -> None:
    """Valida palavras, frases e respiros antes de iniciar a sessão do Premiere."""
    if not (
        plano.acoes_do_tipo(TipoDeAcao.CORTE)
        or plano.acoes_do_tipo(TipoDeAcao.GANCHO)
    ):
        return
    if plano_id is None:
        if exigir_validacao:
            raise ErroDePlanoInvalido(
                "Salve ou carregue o plano no banco antes de aplicar: sem o identificador "
                "não é possível conferir a transcrição palavra por palavra."
            )
        return
    from engine.persistencia.conexao import abrir_banco
    from engine.persistencia.leitura_semantica_do_plano import LeitorDeTranscricaoDoPlano

    conexao = abrir_banco(caminho_do_banco)
    try:
        falas = LeitorDeTranscricaoDoPlano(conexao).carregar(plano_id)
    finally:
        conexao.close()
    ValidadorSemanticoDePlano().garantir_valido(plano, falas)


def resolver_caminho_do_node() -> str:
    """Resolve o executável Node mesmo quando o painel CEP possui um PATH reduzido.

    A variável ``PREMIERE_NODE_BIN`` permite que o painel ou o sistema informe
    um caminho explícito. Em seguida, a resolução tenta o ``PATH`` e os
    caminhos comuns de instalação local. Levanta :class:`FileNotFoundError`
    com orientação acionável quando o Node não está disponível.
    """
    caminho_configurado = os.environ.get("PREMIERE_NODE_BIN", "").strip()
    if caminho_configurado:
        if os.path.isfile(caminho_configurado) and os.access(caminho_configurado, os.X_OK):
            return caminho_configurado
        raise FileNotFoundError(
            f"Node.js configurado em PREMIERE_NODE_BIN não é executável: {caminho_configurado}"
        )

    encontrado_no_path = shutil.which("node.exe" if os.name == "nt" else "node")
    if encontrado_no_path:
        return encontrado_no_path

    caminhos_conhecidos = (
        ("C:/Program Files/nodejs/node.exe", "C:/Program Files/nodejs/nodejs.exe")
        if os.name == "nt"
        else ("/opt/homebrew/bin/node", "/usr/local/bin/node", "/usr/bin/node")
    )
    for caminho in caminhos_conhecidos:
        if os.path.isfile(caminho) and os.access(caminho, os.X_OK):
            return caminho
    raise FileNotFoundError(
        "Node.js não encontrado. Instale o Node.js ou configure PREMIERE_NODE_BIN "
        "com o caminho completo do executável."
    )


def _registrar_no_banco(
    caminho_do_plano: Path,
    resultado: object,
    sequencia: str,
    plano_id: int | None = None,
    caminho_do_banco: Path = CAMINHO_DO_BANCO,
    verificacao: ResultadoDaVerificacao | None = None,
) -> None:
    """Guarda no banco de análises o plano aplicado e o resultado da aplicação.

    O registro é o único vestígio de como uma edição foi decidida: sem ele o
    plano sumiria junto com o arquivo temporário. Uma falha ao gravar não
    interrompe a edição já aplicada na timeline — apenas avisa em stderr,
    porque perder o histórico é bem menos grave do que abortar um corte que
    já foi feito.

    Parâmetros:
        caminho_do_plano: Arquivo JSON do plano que foi aplicado.
        resultado: Resultado devolvido pelo aplicador.
        sequencia: Nome da sequência onde o plano foi aplicado.
    """
    try:
        from engine.persistencia.conexao import abrir_banco
        from engine.persistencia.repositorio_de_planos import (
            RepositorioDePlanos,
            plano_de_dict,
        )

        dados = json.loads(caminho_do_plano.read_text(encoding="utf-8"))
        repositorio = RepositorioDePlanos(abrir_banco(caminho_do_banco))
        if plano_id is None:
            plano_id = repositorio.registrar_plano(plano_de_dict(dados))
        aplicadas = sum(1 for item in resultado.resultados if item.sucesso)
        verificacao_aprovada = verificacao is None or verificacao.aprovada
        mensagem = verificacao.mensagem if verificacao is not None else None
        repositorio.registrar_aplicacao(
            plano_id,
            sucesso=resultado.todas_bem_sucedidas and verificacao_aprovada,
            acoes_aplicadas=aplicadas,
            sequencia=sequencia,
            mensagem=mensagem,
        )
    except (OSError, ValueError, sqlite3.Error) as erro:
        print(f"Plano aplicado, mas não foi possível registrá-lo no banco: {erro}",
              file=sys.stderr)


def executar(
    caminho_do_plano: Path,
    plano_id: int | None = None,
    caminho_do_banco: Path = CAMINHO_DO_BANCO,
    exigir_validacao_semantica: bool = False,
) -> 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)
    _validar_integridade_semantica(
        plano,
        plano_id,
        caminho_do_banco,
        exigir_validacao_semantica,
    )

    # A pausa é aplicada no cliente, e não dentro do aplicador, para valer
    # igualmente nas leituras da timeline e nas escritas — são todas
    # round-trips síncronos no processo principal do Premiere.
    caminho_do_node = resolver_caminho_do_node()
    cliente = ClienteMCPComPausa(ClienteMCPPorStdio([caminho_do_node, str(CAMINHO_DO_SERVIDOR_MCP)]))
    with SessaoMCP(cliente):
        acesso_ao_editor = AcessoAoEditor(AcessoATimeline(cliente))
        conversor_de_timeline = ConversorDeTimeline()

        # A engine é a única responsável pelo backup. A validação semântica já
        # aconteceu antes da conexão, portanto um plano bloqueado não duplica
        # a sequência. Gancho é recusado sem backup porque combina mais de
        # uma mutação estrutural; as demais ações preservam o fluxo legado.
        sequencia_ativa = conversor_de_timeline.converter(acesso_ao_editor.obter_timeline_ativa())
        backup_ok = BackupDaSequencia(cliente).criar_backup(sequencia_ativa.identificador)
        _garantir_backup_para_ganchos(plano, backup_ok)
        print(
            "Backup da sequência criado." if backup_ok else "Não foi possível criar o backup automático; aplicando mesmo assim.",
            file=sys.stderr,
        )

        aplicador = AplicadorDePlanoDeEdicao(
            acesso_ao_editor=acesso_ao_editor,
            conversor_de_timeline=conversor_de_timeline,
            escrita=EscritaNoEditor(cliente),
            mapeador=MapeadorDeTempoDeOrigemParaTimeline(),
        )
        resultado = aplicador.aplicar(plano)
        timeline_depois = conversor_de_timeline.converter(
            acesso_ao_editor.obter_timeline_ativa()
        )
        verificacao = VerificadorDaAplicacao().verificar(
            sequencia_ativa, timeline_depois, plano
        )

    _registrar_no_banco(
        caminho_do_plano, resultado, sequencia_ativa.nome,
        plano_id=plano_id, caminho_do_banco=caminho_do_banco,
        verificacao=verificacao,
    )

    return {
        "arquivo_de_origem": plano.arquivo_de_origem,
        "todas_bem_sucedidas": resultado.todas_bem_sucedidas and verificacao.aprovada,
        "verificacao": {
            "aprovada": verificacao.aprovada,
            "mensagem": verificacao.mensagem,
            "divergencias": list(verificacao.divergencias),
            "avisos": list(verificacao.avisos),
        },
        "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."""
    argumentos = argparse.ArgumentParser()
    argumentos.add_argument("plano", type=Path)
    argumentos.add_argument("--plano-id", type=int)
    argumentos.add_argument("--banco", type=Path, default=CAMINHO_DO_BANCO)
    argumentos.add_argument("--exigir-validacao-semantica", action="store_true")
    opcoes = argumentos.parse_args()
    try:
        resultado = executar(
            opcoes.plano,
            opcoes.plano_id,
            opcoes.banco,
            opcoes.exigir_validacao_semantica,
        )
    except ErroDePlanoInvalido as erro:
        print(json.dumps({
            "todas_bem_sucedidas": False,
            "erro": str(erro),
            "acoes": [],
        }, ensure_ascii=False, indent=2))
        sys.exit(1)
    print(json.dumps(resultado, ensure_ascii=False, indent=2))
    if not resultado["todas_bem_sucedidas"]:
        sys.exit(1)


if __name__ == "__main__":
    main()