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

"""Persistência SQLite dos grupos de retakes.

Implementação alternativa a ``scanner/retakes/repositorio_de_retakes.RepositorioDeRetakes``,
com a mesma interface pública (``registrar``/``carregar``/``ja_analisado``), gravando nas
tabelas ``grupos_de_retake``, ``tomadas_de_retake`` e ``evidencias_de_retake`` e usando
``analises_versao`` (etapa ``"retakes"``) para a idempotência por hash de versão.
"""

from __future__ import annotations

import hashlib
import sqlite3
from pathlib import Path
from typing import Iterable

from ..scanner.retakes.modelos_de_retakes import (
    STATUS_DE_REVISAO_VALIDOS,
    EvidenciaDeRetake,
    GrupoDeRetake,
    TomadaDeRetake,
)
from ..scanner.retakes.repositorio_de_retakes import GrupoDeRetakeInexistente
from .conexao import abrir_banco

VERSAO_REGRA_FALAS = 1
VERSAO_REGRA_TEXTO = 1
VERSAO_MODELO_PADRAO = "lexico"
ETAPA = "retakes"


class RepositorioDeRetakesSQLite:
    """Persiste e recupera grupos de retakes em um banco SQLite."""

    def __init__(
        self,
        banco: str | Path | sqlite3.Connection = ".jhonny/analises.db",
        versao_transcricao: str = "1",
        versao_regras: str = f"regras:{VERSAO_REGRA_FALAS}.{VERSAO_REGRA_TEXTO}",
        versao_modelo: str = VERSAO_MODELO_PADRAO,
    ) -> None:
        self.conexao = banco if isinstance(banco, sqlite3.Connection) else abrir_banco(banco)
        self.versao_transcricao = versao_transcricao
        self.versao_regras = versao_regras
        self.versao_modelo = versao_modelo

    def _hash_de_identificacao(self, video_id: str, versao_visual: str = "sem_visual") -> str:
        origem = "|".join([video_id, self.versao_transcricao, self.versao_regras,
                           self.versao_modelo, versao_visual])
        return hashlib.sha256(origem.encode("utf-8")).hexdigest()[:12]

    def ja_analisado(self, video_id: str, versao_visual: str = "sem_visual") -> bool:
        """Verdadeiro se o vídeo já foi analisado com exatamente estas versões."""
        linha = self.conexao.execute(
            "SELECT hash_versao FROM analises_versao WHERE video_id = ? AND etapa = ?",
            (video_id, ETAPA),
        ).fetchone()
        if linha is None:
            return False
        return linha["hash_versao"] == self._hash_de_identificacao(video_id, versao_visual)

    def registrar(
        self,
        video_id: str,
        grupos: Iterable[GrupoDeRetake],
        versao_visual: str = "sem_visual",
    ) -> None:
        """Grava os grupos de forma idempotente (substitui os grupos anteriores do vídeo)."""
        grupos = list(grupos)
        with self.conexao:
            self.conexao.execute(
                "INSERT OR IGNORE INTO videos (id) VALUES (?)", (video_id,)
            )
            self.conexao.execute(
                "DELETE FROM grupos_de_retake WHERE video_id = ?", (video_id,)
            )
            for grupo in grupos:
                self.conexao.execute(
                    """INSERT INTO grupos_de_retake
                       (id, video_id, faixa_id, tipo, confianca, criado_em, status_de_revisao)
                       VALUES (?, ?, ?, ?, ?, ?, ?)""",
                    (grupo.id, grupo.video_id, grupo.faixa_id, grupo.tipo,
                     grupo.confianca, grupo.criado_em, grupo.status_de_revisao),
                )
                for tomada in grupo.tomadas:
                    self.conexao.execute(
                        """INSERT INTO tomadas_de_retake
                           (grupo_id, ordem, fala_id, segmento_id, inicio, fim, texto,
                            similaridade_com_anterior, confianca)
                           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                        (grupo.id, tomada.ordem, tomada.fala_id, tomada.segmento_id,
                         tomada.inicio, tomada.fim, tomada.texto,
                         tomada.similaridade_com_anterior, tomada.confianca),
                    )
                for evidencia in grupo.evidencias:
                    self.conexao.execute(
                        """INSERT INTO evidencias_de_retake (grupo_id, tipo, descricao, valor)
                           VALUES (?, ?, ?, ?)""",
                        (grupo.id, evidencia.tipo, evidencia.descricao, evidencia.valor),
                    )
            self.conexao.execute(
                """INSERT INTO analises_versao (video_id, etapa, hash_versao)
                   VALUES (?, ?, ?)
                   ON CONFLICT (video_id, etapa)
                   DO UPDATE SET hash_versao = excluded.hash_versao,
                                 concluido_em = strftime('%Y-%m-%dT%H:%M:%fZ','now')""",
                (video_id, ETAPA, self._hash_de_identificacao(video_id, versao_visual)),
            )

    def carregar(self, video_id: str) -> list[GrupoDeRetake]:
        """Carrega os grupos já persistidos para o vídeo."""
        linhas_de_grupos = self.conexao.execute(
            "SELECT * FROM grupos_de_retake WHERE video_id = ? ORDER BY criado_em",
            (video_id,),
        ).fetchall()

        grupos: list[GrupoDeRetake] = []
        for linha in linhas_de_grupos:
            tomadas = [
                TomadaDeRetake(
                    ordem=item["ordem"],
                    fala_id=item["fala_id"],
                    segmento_id=item["segmento_id"],
                    inicio=item["inicio"],
                    fim=item["fim"],
                    texto=item["texto"],
                    similaridade_com_anterior=item["similaridade_com_anterior"],
                    confianca=item["confianca"],
                )
                for item in self.conexao.execute(
                    "SELECT * FROM tomadas_de_retake WHERE grupo_id = ? ORDER BY ordem",
                    (linha["id"],),
                ).fetchall()
            ]
            evidencias = [
                EvidenciaDeRetake(
                    tipo=item["tipo"], descricao=item["descricao"], valor=item["valor"],
                )
                for item in self.conexao.execute(
                    "SELECT * FROM evidencias_de_retake WHERE grupo_id = ? ORDER BY id",
                    (linha["id"],),
                ).fetchall()
            ]
            grupos.append(GrupoDeRetake(
                id=linha["id"],
                video_id=linha["video_id"],
                faixa_id=linha["faixa_id"],
                tipo=linha["tipo"],
                confianca=linha["confianca"],
                tomadas=tomadas,
                evidencias=evidencias,
                criado_em=linha["criado_em"],
                status_de_revisao=linha["status_de_revisao"],
            ))
        return grupos

    def atualizar_status_de_revisao(
        self, video_id: str, grupo_id: str, novo_status: str,
    ) -> GrupoDeRetake:
        """
        Registra a decisão manual do usuário sobre um grupo de retakes.

        A revisão não altera a detecção automática: grava apenas o estado
        escolhido, preservando os demais grupos do vídeo.

        Parâmetros:
            video_id: Vídeo dono do grupo revisado.
            grupo_id: Grupo que recebeu a decisão.
            novo_status: Estado escolhido pelo usuário.

        Retorna:
            O grupo já com o novo status.

        Pode gerar:
            ValueError: quando o status não é um dos válidos.
            GrupoDeRetakeInexistente: quando o grupo não existe no vídeo.
        """
        if novo_status not in STATUS_DE_REVISAO_VALIDOS:
            validos = ", ".join(sorted(STATUS_DE_REVISAO_VALIDOS))
            raise ValueError(
                f"Status de revisão inválido: {novo_status!r}. Válidos: {validos}."
            )
        with self.conexao:
            cursor = self.conexao.execute(
                """UPDATE grupos_de_retake SET status_de_revisao = ?
                   WHERE video_id = ? AND id = ?""",
                (novo_status, video_id, grupo_id),
            )
        if cursor.rowcount == 0:
            raise GrupoDeRetakeInexistente(video_id, grupo_id)
        grupo = next(
            (item for item in self.carregar(video_id) if item.id == grupo_id), None)
        if grupo is None:
            raise GrupoDeRetakeInexistente(video_id, grupo_id)
        return grupo