jhonny-editor/code/engine/persistencia/repositorio_de_planos.py

"""
Persistência dos planos de edição devolvidos pela IA e de suas aplicações.

Antes deste módulo o plano era escrito num arquivo temporário, aplicado na
timeline e descartado. Com isso o sistema não guardava nenhum registro de como
uma edição foi decidida — nem o que a IA propôs, nem se a aplicação funcionou.

Guardar plano e aplicação separadamente é deliberado: o mesmo plano pode ser
aplicado mais de uma vez (em outra sequência, ou depois de um backup), e a
proposta continua sendo a mesma. Comparar o que foi proposto com o que
sobreviveu na timeline é o que dá material para melhorar as decisões futuras.
"""

from __future__ import annotations

import json
import sqlite3
from dataclasses import dataclass, field
from typing import Sequence


class ErroDePlano(ValueError):
    """Plano malformado ou parâmetros inválidos para gravá-lo."""


@dataclass(frozen=True)
class AcaoDoPlano:
    """
    Uma operação proposta pela IA sobre a timeline.

    Atributos:
        tipo: Natureza da operação — corte, zoom, texto, marcador.
        inicio: Início do trecho afetado, em segundos.
        fim: Fim do trecho afetado, em segundos.
        motivo: Justificativa dada pela IA, quando houver.
        parametros: Campos específicos do tipo de ação.
    """

    tipo: str
    inicio: float | None = None
    fim: float | None = None
    motivo: str | None = None
    parametros: dict[str, object] = field(default_factory=dict)


@dataclass(frozen=True)
class PlanoDeEdicao:
    """
    Um plano completo, como a IA o devolveu.

    Atributos:
        origem: Mídia ou sequência a que o plano se refere.
        acoes: Operações propostas, na ordem em que devem ser aplicadas.
        video_id: Vídeo do banco correspondente, quando conhecido.
        tipo_de_video: Perfil editorial escolhido no painel.
        modelo_da_ia: Modelo que produziu o plano.
        intencao: Pedido do usuário que originou o plano.
    """

    origem: str
    acoes: tuple[AcaoDoPlano, ...]
    video_id: str | None = None
    tipo_de_video: str | None = None
    modelo_da_ia: str | None = None
    intencao: str | None = None


class RepositorioDePlanos:
    """
    Grava e lê planos de edição e o resultado de suas aplicações.

    Atributos:
        conexao: Conexão SQLite já aberta e com o esquema aplicado.
    """

    def __init__(self, conexao: sqlite3.Connection) -> None:
        """
        Inicializa o repositório sobre uma conexão existente.

        Parâmetros:
            conexao: Conexão SQLite já aberta e com o esquema aplicado.
        """
        self.conexao = conexao

    def registrar_plano(self, plano: PlanoDeEdicao) -> int:
        """
        Grava um plano e suas ações, devolvendo o identificador criado.

        Planos nunca são substituídos: cada devolução da IA é um registro
        novo, porque o histórico de tentativas é justamente o que se quer
        preservar.

        Parâmetros:
            plano: Plano a gravar.

        Retorna:
            O identificador do plano gravado.

        Pode gerar:
            ErroDePlano: quando o plano não tem origem ou não tem ação alguma.
        """
        if not plano.origem or not plano.origem.strip():
            raise ErroDePlano("O plano precisa indicar a origem (mídia ou sequência).")
        if not plano.acoes:
            raise ErroDePlano("Um plano sem ações não é gravável.")

        with self.conexao:
            if plano.video_id:
                self.conexao.execute(
                    "INSERT OR IGNORE INTO videos (id) VALUES (?)", (plano.video_id,))
            cursor = self.conexao.execute(
                """INSERT INTO planos_de_edicao
                   (video_id, origem, tipo_de_video, modelo_da_ia, intencao)
                   VALUES (?, ?, ?, ?, ?)""",
                (plano.video_id, plano.origem, plano.tipo_de_video,
                 plano.modelo_da_ia, plano.intencao),
            )
            plano_id = int(cursor.lastrowid)
            self.conexao.executemany(
                """INSERT INTO acoes_do_plano
                   (plano_id, ordem, tipo, inicio, fim, motivo, parametros)
                   VALUES (?, ?, ?, ?, ?, ?, ?)""",
                [(plano_id, ordem, acao.tipo, acao.inicio, acao.fim, acao.motivo,
                  json.dumps(acao.parametros, ensure_ascii=False))
                 for ordem, acao in enumerate(plano.acoes)],
            )
        return plano_id

    def registrar_aplicacao(
        self,
        plano_id: int,
        sucesso: bool,
        acoes_aplicadas: int = 0,
        sequencia: str | None = None,
        mensagem: str | None = None,
    ) -> int:
        """
        Grava o resultado de aplicar um plano na timeline.

        A falha é gravada tanto quanto o sucesso: saber que um plano não pôde
        ser aplicado, e por quê, é informação de diagnóstico que hoje se perde.

        Parâmetros:
            plano_id: Plano que foi aplicado.
            sucesso: Se a aplicação terminou sem erro.
            acoes_aplicadas: Quantas ações efetivamente entraram na timeline.
            sequencia: Nome da sequência onde foi aplicado.
            mensagem: Erro ou observação da aplicação.

        Retorna:
            O identificador da aplicação registrada.

        Pode gerar:
            ErroDePlano: quando o plano informado não existe.
        """
        existe = self.conexao.execute(
            "SELECT 1 FROM planos_de_edicao WHERE id = ?", (plano_id,)).fetchone()
        if existe is None:
            raise ErroDePlano(f"Plano {plano_id} não existe.")
        with self.conexao:
            cursor = self.conexao.execute(
                """INSERT INTO aplicacoes_do_plano
                   (plano_id, sequencia, sucesso, acoes_aplicadas, mensagem)
                   VALUES (?, ?, ?, ?, ?)""",
                (plano_id, sequencia, 1 if sucesso else 0, acoes_aplicadas, mensagem),
            )
        return int(cursor.lastrowid)

    def carregar_plano(self, plano_id: int) -> PlanoDeEdicao | None:
        """
        Lê um plano gravado, com suas ações em ordem.

        Parâmetros:
            plano_id: Plano a carregar.

        Retorna:
            O plano, ou None quando o identificador não existe.
        """
        cabecalho = self.conexao.execute(
            "SELECT * FROM planos_de_edicao WHERE id = ?", (plano_id,)).fetchone()
        if cabecalho is None:
            return None
        acoes = tuple(
            AcaoDoPlano(
                tipo=linha["tipo"], inicio=linha["inicio"], fim=linha["fim"],
                motivo=linha["motivo"],
                parametros=json.loads(linha["parametros"] or "{}"),
            )
            for linha in self.conexao.execute(
                "SELECT * FROM acoes_do_plano WHERE plano_id = ? ORDER BY ordem",
                (plano_id,),
            )
        )
        return PlanoDeEdicao(
            origem=cabecalho["origem"], acoes=acoes, video_id=cabecalho["video_id"],
            tipo_de_video=cabecalho["tipo_de_video"],
            modelo_da_ia=cabecalho["modelo_da_ia"], intencao=cabecalho["intencao"],
        )

    def listar_planos_do_video(self, video_id: str) -> list[dict[str, object]]:
        """
        Lista o histórico de planos de um vídeo, do mais recente ao mais antigo.

        Parâmetros:
            video_id: Vídeo cujos planos serão listados.

        Retorna:
            Um resumo por plano, com contagem de ações e de aplicações.
        """
        return [
            dict(linha) for linha in self.conexao.execute(
                """SELECT p.id, p.criado_em, p.tipo_de_video, p.modelo_da_ia,
                          (SELECT count(*) FROM acoes_do_plano a WHERE a.plano_id = p.id)
                              AS total_de_acoes,
                          (SELECT count(*) FROM aplicacoes_do_plano ap
                            WHERE ap.plano_id = p.id AND ap.sucesso = 1)
                              AS aplicacoes_com_sucesso
                   FROM planos_de_edicao p
                   WHERE p.video_id = ?
                   ORDER BY p.criado_em DESC""",
                (video_id,),
            )
        ]

    def listar_todos_os_planos(self, limite: int = 50) -> list[dict[str, object]]:
        """
        Lista os planos mais recentes de qualquer vídeo, para escolha manual por id.

        Existe para o caso em que o plano foi registrado sem ``video_id`` (a
        mídia não tinha edição correspondente em ``edicoes_de_video`` no
        momento do registro) — ``listar_planos_do_video`` não o encontraria,
        já que filtra por vídeo conhecido.

        Parâmetros:
            limite: Quantos planos devolver, do mais recente ao mais antigo.

        Retorna:
            Um resumo por plano — id, data de criação, origem, intenção,
            tipo de vídeo e contagem de ações — para montar uma lista de
            seleção sem carregar o plano inteiro.
        """
        return [
            dict(linha) for linha in self.conexao.execute(
                """SELECT p.id, p.criado_em, p.origem, p.intencao, p.tipo_de_video,
                          p.modelo_da_ia, p.video_id,
                          (SELECT count(*) FROM acoes_do_plano a WHERE a.plano_id = p.id)
                              AS total_de_acoes
                   FROM planos_de_edicao p
                   ORDER BY p.criado_em DESC
                   LIMIT ?""",
                (limite,),
            )
        ]


def plano_de_dict(
    dados: dict,
    video_id: str | None = None,
    tipo_de_video: str | None = None,
    modelo_da_ia: str | None = None,
    intencao: str | None = None,
) -> PlanoDeEdicao:
    """
    Converte o JSON devolvido pela IA no plano gravável.

    Aceita o mesmo formato que o painel já valida — ``source`` mais uma lista
    ``actions`` com ``kind``, ``start`` e ``end`` — para que gravar o plano não
    exija mudar o contrato com a IA.

    Parâmetros:
        dados: JSON do plano, já desserializado.
        video_id: Vídeo do banco correspondente, quando conhecido.
        tipo_de_video: Perfil editorial escolhido no painel.
        modelo_da_ia: Modelo que produziu o plano.
        intencao: Pedido do usuário que originou o plano.

    Retorna:
        O plano pronto para ``RepositorioDePlanos.registrar_plano``.

    Pode gerar:
        ErroDePlano: quando falta ``source`` ou ``actions`` não é uma lista
            não vazia.
    """
    origem = dados.get("source")
    acoes_brutas = dados.get("actions")
    if not isinstance(origem, str) or not origem.strip():
        raise ErroDePlano("O plano precisa ter 'source'.")
    if not isinstance(acoes_brutas, list) or not acoes_brutas:
        raise ErroDePlano("O plano precisa ter uma lista 'actions' não vazia.")

    conhecidos = {"kind", "start", "end", "reason", "motivo"}
    acoes: list[AcaoDoPlano] = []
    for indice, bruta in enumerate(acoes_brutas):
        if not isinstance(bruta, dict) or not bruta.get("kind"):
            raise ErroDePlano(f"A ação {indice} não tem 'kind'.")
        acoes.append(AcaoDoPlano(
            tipo=str(bruta["kind"]),
            inicio=_numero(bruta.get("start")),
            fim=_numero(bruta.get("end")),
            motivo=bruta.get("reason") or bruta.get("motivo"),
            parametros={chave: valor for chave, valor in bruta.items()
                        if chave not in conhecidos},
        ))
    return PlanoDeEdicao(
        origem=origem, acoes=tuple(acoes), video_id=video_id,
        tipo_de_video=tipo_de_video, modelo_da_ia=modelo_da_ia, intencao=intencao,
    )


def _numero(valor: object) -> float | None:
    """Converte um valor em float quando ele for numérico, senão devolve None."""
    if isinstance(valor, bool) or not isinstance(valor, (int, float)):
        return None
    return float(valor)


__all__ = [
    "AcaoDoPlano",
    "ErroDePlano",
    "PlanoDeEdicao",
    "RepositorioDePlanos",
    "plano_de_dict",
]