"""Persistência dos grupos de retakes. Na primeira versão, grava os grupos em um arquivo JSON idempotente, no mesmo espírito do cache de transcrição da engine. O `RepositorioDeRetakes` define o contrato (``registrar``/``carregar``/``ja_analisado``) para que a camada de banco de dados (SQLite/Postgres, prevista na Etapa 8 da arquitetura) possa ser plugada depois sem alterar o coordenador. """ from __future__ import annotations import hashlib import json from dataclasses import asdict from pathlib import Path from typing import Iterable from .modelos_de_retakes import ( STATUS_DE_REVISAO_VALIDOS, GrupoDeRetake, ) # Versões da transição e das regras para a idempotência. VERSAO_REGRA_FALAS = 1 VERSAO_REGRA_TEXTO = 1 VERSAO_MODELO_PADRAO = "lexico" class GrupoDeRetakeInexistente(LookupError): """Erro levantado quando um grupo de retakes não existe no repositório.""" def __init__(self, video_id: str, grupo_id: str) -> None: super().__init__( f"Grupo {grupo_id} não encontrado para o vídeo {video_id}.") self.video_id = video_id self.grupo_id = grupo_id class RepositorioDeRetakes: """Persiste e recupera grupos de retakes em um arquivo JSON.""" def __init__( self, diretoria_de_trabalho: str | Path = ".retakes", versao_transcricao: str = "1", versao_regras: str = f"regras:{VERSAO_REGRA_FALAS}.{VERSAO_REGRA_TEXTO}", versao_modelo: str = VERSAO_MODELO_PADRAO, ) -> None: self.diretoria_de_trabalho = Path(diretoria_de_trabalho) self.versao_transcricao = versao_transcricao self.versao_regras = versao_regras self.versao_modelo = versao_modelo def _caminho(self, video_id: str) -> Path: return self.diretoria_de_trabalho / f"retakes-{self._seguro(video_id)}.json" @staticmethod def _seguro(valor: str) -> str: return "".join(c if c.isalnum() or c in "._-" else "_" for c in valor) or "video" 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. A ``versao_visual`` separa análises com e sem cenas visuais, de modo que uma mudança de configuração visual força uma nova análise (§10). """ caminho = self._caminho(video_id) if not caminho.is_file(): return False try: dados = json.loads(caminho.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return False return dados.get("_hash") == self._hash_de_identificacao(video_id, versao_visual) def registrar( self, video_id: str, grupos: Iterable[GrupoDeRetake], versao_visual: str = "sem_visual", ) -> Path: """Grava os grupos de forma idempotente e retorna o arquivo gerado.""" self.diretoria_de_trabalho.mkdir(parents=True, exist_ok=True) dados = { "video_id": video_id, "versao_transcricao": self.versao_transcricao, "versao_regras": self.versao_regras, "versao_modelo": self.versao_modelo, "versao_visual": versao_visual, "_hash": self._hash_de_identificacao(video_id, versao_visual), "grupos_de_retakes": [asdict(grupo) for grupo in grupos], } caminho = self._caminho(video_id) caminho.write_text(json.dumps(dados, ensure_ascii=False, indent=2), encoding="utf-8") return caminho def carregar(self, video_id: str) -> list[GrupoDeRetake]: """Carrega os grupos já persistidos para o vídeo.""" caminho = self._caminho(video_id) if not caminho.is_file(): return [] dados = json.loads(caminho.read_text(encoding="utf-8")) grupos: list[GrupoDeRetake] = [] for item in dados.get("grupos_de_retakes", []): tomadas = item.get("tomadas", []) evidencias = item.get("evidencias", []) grupos.append(GrupoDeRetake( id=item["id"], video_id=item.get("video_id", video_id), faixa_id=item.get("faixa_id", ""), tipo=item["tipo"], confianca=item["confianca"], tomadas=[self._tomada_do_dict(t) for t in tomadas], evidencias=[self._evidencia_do_dict(e) for e in evidencias], criado_em=item.get("criado_em", ""), status_de_revisao=item.get("status_de_revisao", "pendente_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: apenas grava o estado escolhido (confirmado, não é retake, ignorado ou pendente) no arquivo já existente do vídeo, preservando os demais grupos. Levanta ``ValueError`` se o status não for válido e ``GrupoDeRetakeInexistente`` se o grupo não existir. """ 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}. " f"Válidos: {validos}.") caminho = self._caminho(video_id) if not caminho.is_file(): raise GrupoDeRetakeInexistente(video_id, grupo_id) dados = json.loads(caminho.read_text(encoding="utf-8")) grupos = dados.get("grupos_de_retakes", []) alvo = next((g for g in grupos if g.get("id") == grupo_id), None) if alvo is None: raise GrupoDeRetakeInexistente(video_id, grupo_id) alvo["status_de_revisao"] = novo_status caminho.write_text(json.dumps(dados, ensure_ascii=False, indent=2), encoding="utf-8") return self.carregar(video_id)[grupos.index(alvo)] @staticmethod def _tomada_do_dict(item: dict) -> object: from .modelos_de_retakes import TomadaDeRetake return TomadaDeRetake( ordem=item["ordem"], fala_id=item["fala_id"], segmento_id=item.get("segmento_id", ""), inicio=item["inicio"], fim=item["fim"], texto=item.get("texto", ""), similaridade_com_anterior=item.get("similaridade_com_anterior", 0.0), confianca=item.get("confianca", 0.0), ) @staticmethod def _evidencia_do_dict(item: dict) -> object: from .modelos_de_retakes import EvidenciaDeRetake return EvidenciaDeRetake( tipo=item["tipo"], descricao=item.get("descricao", ""), valor=item.get("valor"), )