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