jhonny-editor/code/engine/analisar_trilhas.py

"""Executa a análise em lote de uma pasta de trilhas e emite progresso JSON."""

from __future__ import annotations

import argparse
import json
from pathlib import Path
import sys

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.integracoes.audio import AnalisadorDeMusicaInstrumentalEssentia, ClassificadorDeMusicaEssentia
from engine.persistencia.conexao import abrir_banco
from engine.persistencia.repositorio_de_trilhas import RepositorioDeTrilhasSQLite

EXTENSOES_DE_AUDIO = {".wav", ".mp3", ".flac", ".m4a", ".aiff", ".ogg"}


def emitir(evento: str, **dados: object) -> None:
    """Emite uma linha de progresso consumível pelo painel CEP."""
    print("@@PROGRESS " + json.dumps({"evento": evento, **dados}, ensure_ascii=False), flush=True)


def executar(pasta: Path, banco: Path, diretorio_dos_modelos: Path | None, apenas_listar: bool = False) -> int:
    """Atualiza o catálogo e, opcionalmente, analisa as trilhas pendentes."""
    arquivos = tuple(sorted(
        arquivo for arquivo in pasta.rglob("*")
        if arquivo.is_file() and arquivo.suffix.lower() in EXTENSOES_DE_AUDIO
    ))
    conexao = abrir_banco(banco)
    try:
        repositorio = RepositorioDeTrilhasSQLite(conexao)
        trilhas = repositorio.registrar_pasta_e_arquivos(pasta, arquivos)
        if apenas_listar:
            emitir("catalogo_atualizado", trilhas=[
                {"nome": trilha.nome_do_arquivo, "caminho": str(trilha.caminho), "status": trilha.status}
                for trilha in trilhas
            ])
            return 0
        classificador = (
            ClassificadorDeMusicaEssentia(diretorio_dos_modelos)
            if diretorio_dos_modelos and diretorio_dos_modelos.is_dir()
            else None
        )
        analisador = AnalisadorDeMusicaInstrumentalEssentia(classificador=classificador)
        pendentes = tuple(trilha for trilha in trilhas if trilha.status != "concluida")
        emitir(
            "lote_iniciado",
            total=len(pendentes),
            pasta=str(pasta),
            nomes=[trilha.nome_do_arquivo for trilha in pendentes],
            trilhas=[{"nome": trilha.nome_do_arquivo, "caminho": str(trilha.caminho), "status": trilha.status} for trilha in trilhas],
        )
        for ordem, trilha in enumerate(pendentes, start=1):
            repositorio.iniciar_analise(trilha.identificador)
            emitir("trilha_iniciada", nome=trilha.nome_do_arquivo, indice=ordem, total=len(trilhas), percentual=(ordem - 1) * 100 / max(1, len(trilhas)))
            try:
                resultado = analisador.analisar(trilha.caminho)
                repositorio.registrar_analise(trilha, resultado)
                emitir("trilha_concluida", nome=trilha.nome_do_arquivo, indice=ordem, total=len(trilhas), percentual=ordem * 100 / max(1, len(trilhas)))
            except (OSError, RuntimeError, ValueError) as erro:
                repositorio.registrar_erro(trilha.identificador, str(erro))
                emitir("trilha_com_erro", nome=trilha.nome_do_arquivo, indice=ordem, total=len(trilhas), erro=str(erro), percentual=ordem * 100 / max(1, len(trilhas)))
        emitir("lote_concluido", total=len(trilhas), percentual=100)
        return 0
    finally:
        conexao.close()


def principal() -> None:
    """Lê argumentos e executa o lote de análise."""
    argumentos = argparse.ArgumentParser()
    argumentos.add_argument("pasta", type=Path)
    argumentos.add_argument("--banco", type=Path, default=Path(".jhonny/analises.db"))
    argumentos.add_argument("--modelos", type=Path, default=None)
    argumentos.add_argument("--listar", action="store_true")
    opcoes = argumentos.parse_args()
    try:
        raise SystemExit(executar(opcoes.pasta, opcoes.banco, opcoes.modelos, opcoes.listar))
    except (OSError, RuntimeError, ValueError) as erro:
        emitir("lote_com_erro", erro=str(erro))
        raise SystemExit(1) from erro


if __name__ == "__main__":
    principal()