jhonny-editor/code/engine/gerar_relatorio_de_analises.py

"""Relatório dos J-SOM de análise (imagem e som) lidos do banco de análises.

Este módulo lê o banco ``analises.db`` e monta um único documento JSON com as
duas análises já persistidas pelo Scanner:

- **Som**: transcrição por clipe/cena, com falante, emoção e confiança.
- **Imagem**: evidências visuais por clipe (Apple Vision), com resumo por tipo.

É a base da aba **Análises** do painel CEP, que permite abrir e copiar esses
resultados sem precisar de SQL nem de reprocessar o vídeo.
"""

from __future__ import annotations

import argparse
import json
import sys
import traceback
from pathlib import Path

# O painel chama este arquivo com cwd ``code/engine``; injetar ``code`` (pai do
# pacote ``engine``) para que os importes funcionem, como em ``executar_retakes.py``.
CAMINHO_DO_CODIGO = Path(__file__).resolve().parent.parent
if str(CAMINHO_DO_CODIGO) not in sys.path:
    sys.path.insert(0, str(CAMINHO_DO_CODIGO))

from engine.persistencia.consultas import ConsultasDeAnalises  # noqa: E402


class ErroDeBancoSemVideos(RuntimeError):
    """Representa um banco de análises que ainda não tem nenhum vídeo."""


class MontadorDeRelatorioDeAnalises:
    """
    Monta o relatório JSON dos J-SOM de análise a partir do banco de análises.

    Reaproveita ``ConsultasDeAnalises`` (somente leitura) e organiza os dados
    por clipe/cena, unindo a transcrição (som) e as evidências visuais (imagem)
    de cada trecho da timeline.

    Atributos:
        consultas: Consultas de leitura sobre o banco de análises.
    """

    def __init__(self, banco: str | Path) -> None:
        """
        Inicializa o montador sobre um banco de análises existente.

        Parâmetros:
            banco: Caminho do arquivo ``analises.db``.
        """
        self.consultas = ConsultasDeAnalises(banco)

    def montar(self, video_id: str | None = None) -> dict:
        """
        Monta o relatório completo de um vídeo (ou do primeiro vídeo do banco).

        Parâmetros:
            video_id: Identificador do vídeo. Quando None, usa o primeiro
                vídeo encontrado no banco.

        Retorna:
            Dicionário com o resumo do vídeo e a lista de cenas, cada uma com
            as falas (som) e as evidências visuais (imagem) daquele clipe.

        Pode gerar:
            ErroDeBancoSemVideos: quando o banco não tem nenhum vídeo.
        """
        video_id = video_id or self._primeiro_video()
        resumo = self.consultas.consultar_resumo_do_video(video_id)
        if resumo is None:
            raise ErroDeBancoSemVideos(f"Vídeo '{video_id}' não existe no banco de análises.")
        falas = self.consultas.listar_falas_no_intervalo(video_id, 0.0, None)
        return {
            "video": resumo,
            "cenas": self._cenas_do_relatorio(video_id, falas),
        }

    def _primeiro_video(self) -> str:
        """Devolve o identificador do primeiro vídeo do banco."""
        linha = self.consultas.conexao.execute(
            "SELECT id FROM videos ORDER BY criado_em LIMIT 1").fetchone()
        if linha is None:
            raise ErroDeBancoSemVideos(
                "O banco de análises não tem nenhum vídeo. Execute o Scanner primeiro.")
        return linha["id"]

    def _cenas_do_relatorio(self, video_id: str, falas: list[dict]) -> list[dict]:
        """
        Reúne, por clipe, as falas e as evidências visuais do vídeo.

        A base é a tabela ``clipes`` (timeline completa), porque a tabela
        ``cenas`` pode cobrir só parte dos clipes; o campo
        ``cena_registrada`` indica se aquele clipe tem corte de cena gravado.

        Parâmetros:
            video_id: Identificador do vídeo.
            falas: Falas já consultadas (saída de ``listar_falas_no_intervalo``).

        Retorna:
            Lista de entradas por clipe, ordenadas pela posição na timeline.
        """
        falas_por_clipe: dict[str, list[dict]] = {}
        for fala in falas:
            falas_por_clipe.setdefault(fala["clipe_id"], []).append(fala)

        clipes_com_cena = {
            linha["clipe_id"] for linha in self.consultas.conexao.execute(
                "SELECT DISTINCT clipe_id FROM cenas WHERE video_id = ? AND clipe_id IS NOT NULL",
                (video_id,),
            ).fetchall()
        }

        cenas: list[dict] = []
        for clipe in self.consultas.conexao.execute(
            """SELECT id AS clipe_id, nome, inicio_na_timeline AS inicio,
                      fim_na_timeline AS fim FROM clipes
               WHERE video_id = ? ORDER BY inicio_na_timeline""",
            (video_id,),
        ).fetchall():
            clipe_id = clipe["clipe_id"]
            evidencias = self.consultas.listar_evidencias_visuais_do_clipe(video_id, clipe_id)
            cenas.append({
                "clipe_id": clipe_id,
                "nome": clipe["nome"],
                "inicio": clipe["inicio"],
                "fim": clipe["fim"],
                "cena_registrada": clipe_id in clipes_com_cena,
                "falas": falas_por_clipe.get(clipe_id, []),
                "resumo_de_evidencias": self._resumo_de_evidencias(evidencias),
                "evidencias_visuais": evidencias,
            })
        return cenas

    @staticmethod
    def _resumo_de_evidencias(evidencias: list[dict]) -> dict:
        """Conta as evidências visuais por tipo (ex.: ``rosto: 2``)."""
        resumo: dict[str, int] = {}
        for evidencia in evidencias:
            resumo[evidencia["tipo"]] = resumo.get(evidencia["tipo"], 0) + 1
        return resumo


def executar(banco: Path, saida: Path, video_id: str | None = None) -> dict:
    """
    Gera o arquivo JSON do relatório e devolve o documento montado.

    Parâmetros:
        banco: Caminho do arquivo ``analises.db``.
        saida: Caminho do arquivo JSON a gravar.
        video_id: Vídeo desejado; quando None, usa o primeiro do banco.

    Retorna:
        O dicionário do relatório (mesmo conteúdo gravado em ``saida``).
    """
    relatorio = MontadorDeRelatorioDeAnalises(banco).montar(video_id)
    saida.parent.mkdir(parents=True, exist_ok=True)
    saida.write_text(json.dumps(relatorio, ensure_ascii=False, indent=2), encoding="utf-8")
    return relatorio


def main() -> None:
    """Ponto de entrada da linha de comando usada pelo painel."""
    parser = argparse.ArgumentParser(
        description="Gera o relatório JSON dos J-SOM de análise (som e imagem) do banco.")
    parser.add_argument("banco", type=Path, help="Caminho do analises.db.")
    parser.add_argument("saida", type=Path, help="Arquivo JSON de saída.")
    parser.add_argument("--video-id", default=None, help="Vídeo desejado (opcional).")
    args = parser.parse_args()
    try:
        relatorio = executar(args.banco, args.saida, args.video_id)
    except Exception as erro:  # fronteira do processo: repassa a causa para a tela decidir como exibir
        traceback.print_exc()  # traceback completo no stderr, para não perder um bug real por trás do "erro"
        print(json.dumps({"evento": "erro", "mensagem": str(erro)}, ensure_ascii=False), flush=True)
        sys.exit(1)
    print(json.dumps({"evento": "concluido", "arquivo": str(args.saida),
                      "video": relatorio["video"]["nome"]}, ensure_ascii=False), flush=True)


if __name__ == "__main__":
    main()