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

"""
Ingestão no banco de análises do JSON produzido pelo pipeline de voz.

O pipeline de voz (transcrição, diarização e métricas de fala) roda hoje fora
deste repositório e entrega um ``dados-para-ia.json`` cujo ``segments`` já é
uma tabela plana: cada item traz o texto e os tempos da frase, o falante e os
seis escalares de métrica lado a lado. Este módulo trata esse arquivo como
formato de entrada — nunca como armazenamento — e o converte em linhas do
banco de análises.

A leitura e a gravação ficam em classes separadas: ``LeitorDeDadosParaIA`` não
conhece banco algum e ``IngestorDeVozNoBanco`` não conhece o formato do
arquivo. Assim uma mudança no JSON de origem não alcança a persistência, e
outra origem de transcrição pode reaproveitar a gravação.
"""

from __future__ import annotations

import json
import sqlite3
from dataclasses import dataclass, field
from pathlib import Path
from typing import Sequence

# Tradução entre as chaves de métrica do pipeline de voz e as colunas do banco.
# São seis escalares de conjunto fechado: por isso viram coluna, e não um JSON
# opaco que não se consegue filtrar em SQL.
COLUNAS_DE_METRICA: dict[str, str] = {
    "energy_rms": "energia_rms",
    "pitch_hz_median": "pitch_mediano_hz",
    "pitch_hz_std": "pitch_desvio_hz",
    "speaking_rate_wps": "velocidade_de_fala_pps",
    "longest_internal_pause_s": "maior_pausa_interna_s",
    "gap_before_s": "intervalo_anterior_s",
}


class ErroDeIngestaoDeVoz(ValueError):
    """Arquivo de voz ausente, ilegível ou fora do formato esperado."""


@dataclass(frozen=True)
class PalavraDoPipelineDeVoz:
    """Uma palavra transcrita, com os tempos que permitem cortar sem picotá-la."""

    texto: str
    inicio: float
    fim: float
    confianca: float | None = None


@dataclass(frozen=True)
class FalaDoPipelineDeVoz:
    """
    Uma frase transcrita com as três análises de áudio reunidas.

    Reúne numa só unidade o que o pipeline produz em etapas diferentes sobre o
    mesmo intervalo de tempo: o texto (transcrição), quem falou (diarização) e
    como falou (métricas). São descrições do mesmo trecho, e por isso ocupam a
    mesma linha do banco.

    Atributos:
        inicio: Início da frase em segundos, relativo à mídia de origem.
        fim: Fim da frase em segundos, relativo à mídia de origem.
        texto: Texto transcrito da frase.
        falante: Identificador do falante atribuído pela diarização.
        metricas: Escalares de métrica de fala, já nas chaves do pipeline.
        palavras: Palavras da frase, em ordem.
    """

    inicio: float
    fim: float
    texto: str
    falante: str | None = None
    metricas: dict[str, float | None] = field(default_factory=dict)
    palavras: tuple[PalavraDoPipelineDeVoz, ...] = ()

    @property
    def duracao(self) -> float:
        """Duração da frase em segundos."""
        return self.fim - self.inicio


@dataclass(frozen=True)
class ResultadoDaIngestao:
    """Contagem do que foi gravado, para o chamador relatar sem reconsultar."""

    falas: int
    palavras: int
    falantes: int


class LeitorDeDadosParaIA:
    """
    Converte um ``dados-para-ia.json`` em falas do pipeline de voz.

    Não acessa banco de dados nem filesystem além da leitura do arquivo
    indicado, e não decide nada de editorial: apenas normaliza o formato.
    """

    def ler(self, caminho: str | Path) -> tuple[FalaDoPipelineDeVoz, ...]:
        """
        Lê o arquivo e devolve as falas nele contidas, em ordem de tempo.

        Parâmetros:
            caminho: Caminho do ``dados-para-ia.json`` a carregar.

        Retorna:
            As falas do arquivo, ordenadas pelo início.

        Pode gerar:
            ErroDeIngestaoDeVoz: quando o arquivo não existe, não é JSON
                válido, não traz ``segments`` ou traz um segmento sem os
                tempos obrigatórios.
        """
        caminho = Path(caminho)
        if not caminho.is_file():
            raise ErroDeIngestaoDeVoz(f"Arquivo de voz não encontrado: {caminho}")
        try:
            dados = json.loads(caminho.read_text(encoding="utf-8"))
        except json.JSONDecodeError as erro:
            raise ErroDeIngestaoDeVoz(f"JSON inválido em {caminho}: {erro}") from erro
        except OSError as erro:
            raise ErroDeIngestaoDeVoz(f"Não foi possível ler {caminho}: {erro}") from erro

        if not isinstance(dados, dict) or not isinstance(dados.get("segments"), list):
            raise ErroDeIngestaoDeVoz(
                f"{caminho} não tem a lista 'segments' esperada do pipeline de voz."
            )

        falas = [self._converter_fala(item, indice, caminho)
                 for indice, item in enumerate(dados["segments"])]
        return tuple(sorted(falas, key=lambda fala: fala.inicio))

    def _converter_fala(
        self, item: object, indice: int, caminho: Path,
    ) -> FalaDoPipelineDeVoz:
        """Converte um item de ``segments`` numa fala validada."""
        if not isinstance(item, dict):
            raise ErroDeIngestaoDeVoz(f"Segmento {indice} de {caminho} não é um objeto.")
        inicio = self._numero_obrigatorio(item, "start", indice, caminho)
        fim = self._numero_obrigatorio(item, "end", indice, caminho)
        if fim < inicio:
            raise ErroDeIngestaoDeVoz(
                f"Segmento {indice} de {caminho} termina ({fim}) antes de começar ({inicio})."
            )
        metricas = {chave: self._numero_opcional(item.get(chave))
                    for chave in COLUNAS_DE_METRICA}
        return FalaDoPipelineDeVoz(
            inicio=inicio,
            fim=fim,
            texto=str(item.get("text", "")).strip(),
            falante=item.get("speaker") or None,
            metricas=metricas,
            palavras=self._converter_palavras(item.get("words")),
        )

    def _converter_palavras(self, bruto: object) -> tuple[PalavraDoPipelineDeVoz, ...]:
        """Converte a lista ``words`` de um segmento, ignorando itens malformados."""
        if not isinstance(bruto, list):
            return ()
        palavras = []
        for item in bruto:
            if not isinstance(item, dict):
                continue
            inicio = self._numero_opcional(item.get("start"))
            fim = self._numero_opcional(item.get("end"))
            if inicio is None or fim is None:
                continue
            palavras.append(PalavraDoPipelineDeVoz(
                texto=str(item.get("text", "")),
                inicio=inicio,
                fim=fim,
                confianca=self._numero_opcional(item.get("confidence")),
            ))
        return tuple(palavras)

    @staticmethod
    def _numero_obrigatorio(item: dict, chave: str, indice: int, caminho: Path) -> float:
        """Lê um número que precisa existir, com erro que diz qual campo faltou."""
        valor = item.get(chave)
        if not isinstance(valor, (int, float)) or isinstance(valor, bool):
            raise ErroDeIngestaoDeVoz(
                f"Segmento {indice} de {caminho} não tem o campo numérico '{chave}'."
            )
        return float(valor)

    @staticmethod
    def _numero_opcional(valor: object) -> float | None:
        """Lê um número que pode faltar, devolvendo None quando ausente."""
        if isinstance(valor, bool) or not isinstance(valor, (int, float)):
            return None
        return float(valor)


class IngestorDeVozNoBanco:
    """
    Grava falas do pipeline de voz no banco de análises.

    Substitui integralmente as falas do clipe recebido, para que reprocessar a
    mesma mídia seja idempotente sem apagar análises de outros clipes.

    Atributos:
        conexao: Conexão SQLite já aberta e com o esquema aplicado.
    """

    def __init__(self, conexao: sqlite3.Connection) -> None:
        """
        Inicializa o ingestor com a conexão onde as falas serão gravadas.

        Parâmetros:
            conexao: Conexão SQLite já aberta e com o esquema aplicado.
        """
        self.conexao = conexao

    def ingerir(
        self, video_id: str, clipe_id: str, falas: Sequence[FalaDoPipelineDeVoz],
    ) -> ResultadoDaIngestao:
        """
        Grava as falas do clipe, substituindo o que houver dele no banco.

        Parâmetros:
            video_id: Identificador do vídeo/timeline dono das falas.
            clipe_id: Identificador do clipe a que as falas pertencem.
            falas: Falas a gravar, já normalizadas pelo leitor.

        Retorna:
            As contagens de falas, palavras e falantes distintos gravados.

        Pode gerar:
            ErroDeIngestaoDeVoz: quando ``video_id`` ou ``clipe_id`` for vazio.
        """
        self._validar_identificador("video_id", video_id)
        self._validar_identificador("clipe_id", clipe_id)

        colunas_de_metrica = tuple(COLUNAS_DE_METRICA.values())
        insercao = (
            "INSERT INTO segmentos_de_transcricao "
            "(video_id, clipe_id, inicio, fim, texto, falante, "
            + ", ".join(colunas_de_metrica)
            + ") VALUES (?, ?, ?, ?, ?, ?, "
            + ", ".join("?" * len(colunas_de_metrica))
            + ")"
        )
        total_de_palavras = 0
        with self.conexao:
            self.conexao.execute("INSERT OR IGNORE INTO videos (id) VALUES (?)", (video_id,))
            self.conexao.execute(
                "DELETE FROM segmentos_de_transcricao WHERE video_id = ? AND clipe_id = ?",
                (video_id, clipe_id),
            )
            for fala in falas:
                cursor = self.conexao.execute(insercao, (
                    video_id, clipe_id, fala.inicio, fala.fim, fala.texto, fala.falante,
                    *(fala.metricas.get(chave) for chave in COLUNAS_DE_METRICA),
                ))
                total_de_palavras += self._inserir_palavras(
                    int(cursor.lastrowid), fala,
                )
        return ResultadoDaIngestao(
            falas=len(falas),
            palavras=total_de_palavras,
            falantes=len({fala.falante for fala in falas if fala.falante}),
        )

    def _inserir_palavras(self, segmento_id: int, fala: FalaDoPipelineDeVoz) -> int:
        """Grava as palavras de uma fala e devolve quantas foram gravadas."""
        self.conexao.executemany(
            """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, fala.falante)
             for ordem, palavra in enumerate(fala.palavras)],
        )
        return len(fala.palavras)

    @staticmethod
    def _validar_identificador(nome: str, valor: str) -> None:
        """Recusa identificador vazio antes de escrever qualquer linha."""
        if not isinstance(valor, str) or not valor.strip():
            raise ErroDeIngestaoDeVoz(f"O parâmetro '{nome}' é obrigatório.")