jhonny-editor/code/engine/scanner/retakes/carregador_de_artefatos.py

"""Carrega as transcrições produzidas pelo scanner para análise de retakes.

O painel não retranscreve vídeos: os artefatos ``<caso>-audio-faixa-N.json``
gerados pelo scanner já contêm os segmentos de fala por clipe. Esta classe
converte esses artefatos em falas prontas para o ``CoordenadorDeRetakes``,
sem duplicar a lógica de transcrição.
"""

from __future__ import annotations

import json
from pathlib import Path
from types import SimpleNamespace

from .adaptador_de_falas import AdaptadorDeFalas
from .modelos_de_retakes import Fala


class ArtefatoDeTranscricaoInvalido(ValueError):
    """Erro levantado quando um artefato do scanner não tem transcrição útil."""


class CarregadorDeArtefatosDeAudio:
    """Lê os artefatos de transcrição do scanner e devolve falas ordenadas."""

    def __init__(self, adaptador: AdaptadorDeFalas | None = None) -> None:
        self.adaptador = adaptador or AdaptadorDeFalas()

    def carregar(self, caminhos_dos_artefatos: list[str | Path],
                 video_id: str) -> list[Fala]:
        """Converte os artefatos informados em falas prontas para análise.

        Artefatos sem transcrição concluída são ignorados. Quando nenhum
        artefato contiver segmentos, levanta ``ArtefatoDeTranscricaoInvalido``
        para que a tela possa avisar o usuário sem executar a análise.
        """
        falas: list[Fala] = []
        for caminho in caminhos_dos_artefatos:
            falas.extend(self._falas_do_artefato(Path(caminho), video_id))
        if not falas:
            raise ArtefatoDeTranscricaoInvalido(
                "Nenhum artefato de transcrição contém segmentos de fala. "
                "Execute o Scanner com transcrição antes de analisar retakes.")
        return falas

    def _falas_do_artefato(self, caminho: Path,
                           video_id: str) -> list[Fala]:
        """Extrai as falas de um único artefato ``-audio-faixa-N.json``."""
        if not caminho.is_file():
            return []
        dados = json.loads(caminho.read_text(encoding="utf-8"))
        if dados.get("status") != "concluida":
            return []
        pares: list[tuple[str, list[object]]] = []
        for item in dados.get("segmentos_por_clipe", []):
            segmentos = [SimpleNamespace(**segmento)
                         for segmento in item.get("segmentos", [])]
            if segmentos:
                pares.append((str(item.get("clipe", "clipe")), segmentos))
        return self.adaptador.converter_de_segmentos(pares, video_id=video_id)