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

"""Leitura da transcrição vinculada a um plano para validação prévia."""

from __future__ import annotations

import sqlite3

from engine.editor.erros import ErroDePlanoInvalido
from engine.scanner.modelos import PalavraDeTranscricao, SegmentoDeTranscricao


class LeitorDeTranscricaoDoPlano:
    """Reconstrói tempos de origem das falas associadas a um plano salvo."""

    def __init__(self, conexao: sqlite3.Connection) -> None:
        """Recebe a conexão de leitura do banco de análises."""
        self._conexao = conexao

    def carregar(self, plano_id: int) -> tuple[SegmentoDeTranscricao, ...]:
        """Carrega falas e palavras do plano, convertidas para a mídia original."""
        plano = self._conexao.execute(
            "SELECT video_id FROM planos_de_edicao WHERE id = ?",
            (plano_id,),
        ).fetchone()
        if plano is None:
            raise ErroDePlanoInvalido(f"O plano {plano_id} não existe no banco de análises.")
        video_id = plano["video_id"]
        if not video_id:
            raise ErroDePlanoInvalido(
                f"O plano {plano_id} não está associado a um vídeo analisado."
            )
        linhas = self._conexao.execute(
            """SELECT s.id, s.inicio, s.fim, s.texto, s.confianca, s.falante,
                      c.inicio_na_timeline, c.inicio_na_origem
                 FROM segmentos_de_transcricao AS s
                 LEFT JOIN clipes AS c
                   ON c.video_id = s.video_id AND c.id = s.clipe_id
                WHERE s.video_id = ?
                ORDER BY s.inicio, s.id""",
            (video_id,),
        ).fetchall()
        if not linhas:
            raise ErroDePlanoInvalido(
                "Não há transcrição para validar as frases deste plano. "
                "Rode o Scanner novamente."
            )
        falas = tuple(self._converter_fala(linha) for linha in linhas)
        if not any(fala.palavras for fala in falas):
            raise ErroDePlanoInvalido(
                "A transcrição não possui tempos por palavra. "
                "Rode o Scanner novamente antes de aplicar."
            )
        return falas

    def _converter_fala(self, linha: sqlite3.Row) -> SegmentoDeTranscricao:
        deslocamento = self._calcular_deslocamento(linha)
        palavras = tuple(
            PalavraDeTranscricao(
                texto=palavra["texto"],
                inicio=max(0.0, float(palavra["inicio"]) + deslocamento),
                fim=max(0.0, float(palavra["fim"]) + deslocamento),
                confianca=palavra["confianca"],
                falante=palavra["falante"],
            )
            for palavra in self._conexao.execute(
                """SELECT texto, inicio, fim, confianca, falante
                     FROM palavras_de_transcricao
                    WHERE segmento_id = ?
                    ORDER BY ordem""",
                (linha["id"],),
            ).fetchall()
        )
        return SegmentoDeTranscricao(
            inicio=max(0.0, float(linha["inicio"]) + deslocamento),
            fim=max(0.0, float(linha["fim"]) + deslocamento),
            texto=linha["texto"],
            confianca=linha["confianca"],
            palavras=palavras,
            falante=linha["falante"],
        )

    @staticmethod
    def _calcular_deslocamento(linha: sqlite3.Row) -> float:
        if linha["inicio_na_origem"] is None or linha["inicio_na_timeline"] is None:
            return 0.0
        return float(linha["inicio_na_origem"]) - float(linha["inicio_na_timeline"])