""" 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", ]