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

"""Persistência SQLite de metadados de arquivo, transcrição e evidências visuais.

Complementa ``RepositorioDeRetakesSQLite`` gravando o restante do que o
``ContextoDeAnalise`` do pipeline produz e que hoje só vive em memória
(metadados técnicos) ou em cache JSON por arquivo (transcrição).
"""

from __future__ import annotations

import json
import sqlite3
from pathlib import Path
from typing import Iterable

from ..integracoes.midia.extracao_de_metadados import MetadadosDoArquivo
from ..scanner.modelos import Cena, EvidenciaVisual, PalavraDeTranscricao, SegmentoDeTranscricao
from ..scanner.transcricao_da_timeline import TranscricaoDoClipe
from .conexao import abrir_banco


def _numero_ou_nulo(valor: object) -> float | None:
    """Converte um valor de métrica em float, ou None quando ausente/ilegível."""
    if isinstance(valor, bool) or not isinstance(valor, (int, float)):
        return None
    return float(valor)


class RepositorioDeAnalisesSQLite:
    """Grava metadados de arquivo, transcrição e evidências/cenas visuais."""

    def __init__(self, banco: str | Path | sqlite3.Connection = ".jhonny/analises.db") -> None:
        self.conexao = banco if isinstance(banco, sqlite3.Connection) else abrir_banco(banco)

    def _garantir_video(self, video_id: str) -> None:
        self.conexao.execute("INSERT OR IGNORE INTO videos (id) VALUES (?)", (video_id,))

    def registrar_metadados_de_arquivo(
        self, video_id: str, metadados: MetadadosDoArquivo, hash_do_conteudo: str | None = None,
    ) -> None:
        """
        Grava os metadados técnicos de um arquivo de mídia do vídeo.

        Regravar o mesmo caminho atualiza a linha existente em vez de criar
        outra, então o método é seguro para reprocessamento.

        Parâmetros:
            video_id: Identificador do vídeo dono do arquivo.
            metadados: Metadados técnicos extraídos da mídia.
            hash_do_conteudo: Hash do conteúdo do arquivo, quando calculado.
        """
        with self.conexao:
            self._garantir_video(video_id)
            self.conexao.execute(
                """INSERT INTO arquivos
                   (video_id, caminho, formato, duracao, codec_de_video, codec_de_audio,
                    largura, altura, taxa_de_quadros, canais_de_audio, taxa_de_amostragem,
                    tamanho_em_bytes, orientacao, timecode, hash_do_conteudo)
                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                   ON CONFLICT (video_id, caminho) DO UPDATE SET
                       formato = excluded.formato, duracao = excluded.duracao,
                       codec_de_video = excluded.codec_de_video,
                       codec_de_audio = excluded.codec_de_audio,
                       largura = excluded.largura, altura = excluded.altura,
                       taxa_de_quadros = excluded.taxa_de_quadros,
                       canais_de_audio = excluded.canais_de_audio,
                       taxa_de_amostragem = excluded.taxa_de_amostragem,
                       tamanho_em_bytes = excluded.tamanho_em_bytes,
                       orientacao = excluded.orientacao, timecode = excluded.timecode,
                       hash_do_conteudo = excluded.hash_do_conteudo""",
                (video_id, str(metadados.caminho), metadados.formato, metadados.duracao,
                 metadados.codec_de_video, metadados.codec_de_audio, metadados.largura,
                 metadados.altura, metadados.taxa_de_quadros, metadados.canais_de_audio,
                 metadados.taxa_de_amostragem, metadados.tamanho_em_bytes,
                 metadados.orientacao, metadados.timecode, hash_do_conteudo),
            )

    def registrar_transcricoes(
        self, video_id: str, transcricoes: Iterable[TranscricaoDoClipe],
    ) -> None:
        """
        Acrescenta transcrições sem remover as já gravadas.

        Para reprocessamento use ``substituir_transcricoes``: este método
        acumula, e chamá-lo duas vezes para o mesmo clipe duplica as falas.

        Parâmetros:
            video_id: Identificador do vídeo dono das transcrições.
            transcricoes: Transcrições por clipe a gravar.
        """
        with self.conexao:
            self._garantir_video(video_id)
            for transcricao in transcricoes:
                for segmento in transcricao.segmentos:
                    self._inserir_segmento(video_id, transcricao.identificador_do_clipe, segmento)

    def substituir_transcricoes(
        self, video_id: str, transcricoes: Iterable[TranscricaoDoClipe],
    ) -> None:
        """Substitui a transcrição somente dos clipes recebidos.

        O método torna o reprocessamento idempotente sem apagar análises de
        outros clipes ou faixas que não participaram da execução atual.
        """
        with self.conexao:
            self._garantir_video(video_id)
            transcricoes_materializadas = tuple(transcricoes)
            for transcricao in transcricoes_materializadas:
                self.conexao.execute(
                    """DELETE FROM segmentos_de_transcricao
                       WHERE video_id = ? AND clipe_id = ?""",
                    (video_id, transcricao.identificador_do_clipe),
                )
                for segmento in transcricao.segmentos:
                    self._inserir_segmento(video_id, transcricao.identificador_do_clipe, segmento)

    def atualizar_enriquecimento_de_transcricao(
        self,
        video_id: str,
        clipe_id: str,
        inicio: float,
        fim: float,
        *,
        falante: str | None = None,
        caracteristicas_acusticas: dict[str, object] | None = None,
    ) -> None:
        """Atualiza diarização e métricas de um segmento já persistido."""
        campos = []
        parametros: list[object] = []
        if falante is not None:
            campos.append("falante = ?")
            parametros.append(falante)
        if caracteristicas_acusticas is not None:
            campos.append("caracteristicas_acusticas = ?")
            parametros.append(json.dumps(caracteristicas_acusticas, ensure_ascii=False))
        if not campos:
            return
        parametros.extend((video_id, clipe_id, inicio, fim))
        with self.conexao:
            self.conexao.execute(
                f"""UPDATE segmentos_de_transcricao
                    SET {', '.join(campos)}
                    WHERE video_id = ? AND clipe_id = ? AND inicio = ? AND fim = ?""",
                parametros,
            )

    def _inserir_segmento(
        self, video_id: str, clipe_id: str, segmento: SegmentoDeTranscricao,
    ) -> int:
        """Grava uma fala e devolve o id gerado, para as palavras se ligarem a ela.

        As métricas acústicas são gravadas duas vezes de propósito: em
        ``caracteristicas_acusticas`` (JSON completo, para reconstrução fiel)
        e em colunas dedicadas (para filtrar e ordenar por elas em SQL sem
        precisar abrir o JSON).
        """
        metricas = segmento.caracteristicas_acusticas or {}
        cursor = self.conexao.execute(
            """INSERT INTO segmentos_de_transcricao
               (video_id, clipe_id, inicio, fim, texto, confianca, falante, voz_aparente,
                confianca_voz, emocao, confianca_emocao, caracteristicas_acusticas,
                energia_rms, pitch_mediano_hz, pitch_desvio_hz, velocidade_de_fala_pps,
                maior_pausa_interna_s, intervalo_anterior_s)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
            (video_id, clipe_id, segmento.inicio, segmento.fim, segmento.texto,
             segmento.confianca, segmento.falante, segmento.voz_aparente,
             segmento.confianca_voz, segmento.emocao, segmento.confianca_emocao,
             json.dumps(segmento.caracteristicas_acusticas, ensure_ascii=False),
             _numero_ou_nulo(metricas.get("energy_rms")),
             _numero_ou_nulo(metricas.get("pitch_hz_median")),
             _numero_ou_nulo(metricas.get("pitch_hz_std")),
             _numero_ou_nulo(metricas.get("speaking_rate_wps")),
             _numero_ou_nulo(metricas.get("longest_internal_pause_s")),
             _numero_ou_nulo(metricas.get("gap_before_s"))),
        )
        segmento_id = cursor.lastrowid
        for ordem, palavra in enumerate(segmento.palavras):
            self.conexao.execute(
                """INSERT INTO palavras_de_transcricao
                   (segmento_id, ordem, texto, inicio, fim, confianca, falante)
                   VALUES (?, ?, ?, ?, ?, ?, ?)""",
                (segmento_id, ordem, palavra.texto, palavra.inicio, palavra.fim,
                 palavra.confianca, palavra.falante),
            )
        return segmento_id

    def carregar_transcricoes(self, video_id: str, clipe_id: str) -> list[SegmentoDeTranscricao]:
        """
        Carrega as falas de um clipe, já com as palavras de cada uma.

        Parâmetros:
            video_id: Identificador do vídeo dono das falas.
            clipe_id: Clipe cujas falas serão lidas.

        Retorna:
            As falas do clipe ordenadas por início, com as palavras em ordem.
        """
        segmentos: list[SegmentoDeTranscricao] = []
        for linha in self.conexao.execute(
            """SELECT * FROM segmentos_de_transcricao
               WHERE video_id = ? AND clipe_id = ? ORDER BY inicio""",
            (video_id, clipe_id),
        ).fetchall():
            palavras = tuple(
                PalavraDeTranscricao(
                    texto=item["texto"], inicio=item["inicio"], fim=item["fim"],
                    confianca=item["confianca"], falante=item["falante"],
                )
                for item in self.conexao.execute(
                    "SELECT * FROM palavras_de_transcricao WHERE segmento_id = ? ORDER BY ordem",
                    (linha["id"],),
                ).fetchall()
            )
            segmentos.append(SegmentoDeTranscricao(
                inicio=linha["inicio"], fim=linha["fim"], texto=linha["texto"],
                confianca=linha["confianca"], palavras=palavras,
                emocao=linha["emocao"], confianca_emocao=linha["confianca_emocao"],
                caracteristicas_acusticas=json.loads(linha["caracteristicas_acusticas"] or "{}"),
                falante=linha["falante"], voz_aparente=linha["voz_aparente"],
                confianca_voz=linha["confianca_voz"],
            ))
        return segmentos

    def substituir_evidencias_visuais(
        self, video_id: str, clipe_id: str, evidencias: Iterable[EvidenciaVisual],
    ) -> int:
        """
        Regrava as evidências visuais de um clipe, apagando as anteriores.

        Torna o reprocessamento idempotente. ``registrar_evidencias_visuais``
        só acrescenta, então reanalisar o mesmo clipe com ele acumula cópias
        da mesma evidência e infla qualquer contagem feita sobre a tabela.

        Parâmetros:
            video_id: Identificador do vídeo dono das evidências.
            clipe_id: Clipe cujas evidências serão substituídas.
            evidencias: Evidências a gravar.

        Retorna:
            A quantidade de evidências gravadas.
        """
        evidencias_materializadas = tuple(evidencias)
        with self.conexao:
            self._garantir_video(video_id)
            self.conexao.execute(
                "DELETE FROM evidencias_visuais WHERE video_id = ? AND clipe_id = ?",
                (video_id, clipe_id),
            )
        self.registrar_evidencias_visuais(video_id, clipe_id, evidencias_materializadas)
        return len(evidencias_materializadas)

    def substituir_cenas(
        self, video_id: str, cenas: Iterable[Cena], clipe_id: str | None = None,
    ) -> int:
        """
        Regrava as cenas de um clipe, apagando as anteriores.

        Parâmetros:
            video_id: Identificador do vídeo dono das cenas.
            cenas: Cenas a gravar.
            clipe_id: Clipe cujas cenas serão substituídas. Quando ``None``,
                substitui as cenas do vídeo que não pertencem a clipe algum.

        Retorna:
            A quantidade de cenas gravadas.
        """
        cenas_materializadas = tuple(cenas)
        with self.conexao:
            self._garantir_video(video_id)
            if clipe_id is None:
                self.conexao.execute(
                    "DELETE FROM cenas WHERE video_id = ? AND clipe_id IS NULL", (video_id,))
            else:
                self.conexao.execute(
                    "DELETE FROM cenas WHERE video_id = ? AND clipe_id = ?", (video_id, clipe_id))
        self.registrar_cenas(video_id, cenas_materializadas, clipe_id)
        return len(cenas_materializadas)

    def registrar_evidencias_visuais(
        self, video_id: str, clipe_id: str, evidencias: Iterable[EvidenciaVisual],
    ) -> None:
        """Acrescenta evidências visuais sem remover as já gravadas.

        Para reprocessamento use ``substituir_evidencias_visuais``: este método
        acumula, e chamá-lo duas vezes para o mesmo clipe duplica as linhas.
        """
        with self.conexao:
            self._garantir_video(video_id)
            for evidencia in evidencias:
                self.conexao.execute(
                    """INSERT INTO evidencias_visuais
                       (video_id, clipe_id, tipo, inicio, fim, valor, confianca, provider, modelo)
                       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                    (video_id, clipe_id, evidencia.tipo, evidencia.inicio, evidencia.fim,
                     json.dumps(evidencia.valor, ensure_ascii=False), evidencia.confianca,
                     evidencia.provider, evidencia.modelo),
                )

    def registrar_cenas(
        self, video_id: str, cenas: Iterable[Cena], clipe_id: str | None = None,
    ) -> None:
        """Acrescenta cenas sem remover as já gravadas.

        Para reprocessamento use ``substituir_cenas``: este método acumula, e
        chamá-lo duas vezes para o mesmo clipe duplica as cenas.

        Parâmetros:
            video_id: Identificador do vídeo dono das cenas.
            cenas: Cenas detectadas a gravar.
            clipe_id: Clipe a que as cenas pertencem, quando houver.
        """
        with self.conexao:
            self._garantir_video(video_id)
            for cena in cenas:
                self.conexao.execute(
                    """INSERT INTO cenas (video_id, clipe_id, inicio, fim, confianca, referencias)
                       VALUES (?, ?, ?, ?, ?, ?)""",
                    (video_id, clipe_id, cena.inicio, cena.fim, cena.confianca,
                     json.dumps(list(cena.referencias))),
                )