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

"""Persistência do catálogo independente de trilhas musicais."""

from __future__ import annotations

import hashlib
import json
import sqlite3
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable

from ..integracoes.audio.modelos_de_analise_musical import ResultadoDaAnaliseMusical
from .conexao import abrir_banco


@dataclass(frozen=True)
class TrilhaMusical:
    """Representa um arquivo de música catalogado, sem vínculo com vídeo."""

    identificador: int
    caminho: Path
    nome_do_arquivo: str
    status: str
    erro: str | None = None


class RepositorioDeTrilhasSQLite:
    """Salva pastas, trilhas e versões das análises musicais no SQLite."""

    def __init__(self, banco: str | Path | sqlite3.Connection = ".jhonny/analises.db") -> None:
        """Abre o banco informado e garante que o esquema esteja disponível."""
        self.conexao = banco if isinstance(banco, sqlite3.Connection) else abrir_banco(banco)

    def registrar_pasta_e_arquivos(self, pasta: str | Path, arquivos: Iterable[Path]) -> list[TrilhaMusical]:
        """Registra a pasta e torna sua lista de arquivos pronta para análise."""
        caminho_da_pasta = Path(pasta).expanduser().resolve(strict=True)
        if not caminho_da_pasta.is_dir():
            raise NotADirectoryError(f"A pasta de trilhas não existe: {caminho_da_pasta}")
        arquivos_materializados = tuple(Path(arquivo).resolve() for arquivo in arquivos)
        with self.conexao:
            self.conexao.execute(
                """INSERT INTO pastas_de_trilhas(caminho) VALUES (?)
                   ON CONFLICT(caminho) DO UPDATE SET atualizada_em = strftime('%Y-%m-%dT%H:%M:%fZ','now')""",
                (str(caminho_da_pasta),),
            )
            pasta_id = self.conexao.execute(
                "SELECT id FROM pastas_de_trilhas WHERE caminho = ?", (str(caminho_da_pasta),)
            ).fetchone()[0]
            resultado: list[TrilhaMusical] = []
            for arquivo in arquivos_materializados:
                linha = self.conexao.execute(
                    """INSERT INTO trilhas_musicais(pasta_id, caminho, nome_do_arquivo, tamanho_em_bytes)
                       VALUES (?, ?, ?, ?)
                       ON CONFLICT(caminho) DO UPDATE SET nome_do_arquivo = excluded.nome_do_arquivo,
                       tamanho_em_bytes = excluded.tamanho_em_bytes, atualizada_em = strftime('%Y-%m-%dT%H:%M:%fZ','now')""",
                    (pasta_id, str(arquivo), arquivo.name, arquivo.stat().st_size),
                )
                identificador = linha.lastrowid or self.conexao.execute(
                    "SELECT id FROM trilhas_musicais WHERE caminho = ?", (str(arquivo),)
                ).fetchone()[0]
                registro = self.conexao.execute(
                    "SELECT id, caminho, nome_do_arquivo, status, erro FROM trilhas_musicais WHERE id = ?",
                    (identificador,),
                ).fetchone()
                resultado.append(TrilhaMusical(registro[0], Path(registro[1]), registro[2], registro[3], registro[4]))
            return resultado

    def iniciar_analise(self, trilha_id: int) -> None:
        """Marca uma trilha como em análise e remove erro anterior."""
        with self.conexao:
            self.conexao.execute(
                "UPDATE trilhas_musicais SET status = 'analisando', erro = NULL, atualizada_em = strftime('%Y-%m-%dT%H:%M:%fZ','now') WHERE id = ?",
                (trilha_id,),
            )

    def registrar_analise(self, trilha: TrilhaMusical, resultado: ResultadoDaAnaliseMusical) -> None:
        """Grava descritores e etiquetas produzidos pelo áudio da trilha."""
        bruto = json.dumps({"generos": resultado.generos, "humores": resultado.humores}, ensure_ascii=False)
        with self.conexao:
            self.conexao.execute(
                "UPDATE trilhas_musicais SET hash_do_conteudo = ?, status = 'concluida', erro = NULL, atualizada_em = strftime('%Y-%m-%dT%H:%M:%fZ','now') WHERE id = ?",
                (self._calcular_hash(trilha.caminho), trilha.identificador),
            )
            cursor = self.conexao.execute(
                """INSERT INTO analises_musicais
                   (trilha_id, provedor, modelo, duracao_em_segundos, batidas_por_minuto, tonalidade, modo, intensidade, dancabilidade, resultado_bruto_json)
                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                (trilha.identificador, resultado.provedor, resultado.modelo, resultado.duracao_em_segundos,
                 resultado.batidas_por_minuto, resultado.tonalidade, resultado.modo, resultado.intensidade,
                 resultado.dancabilidade, bruto),
            )
            analise_id = cursor.lastrowid
            for tipo, etiquetas in (("genero", resultado.pontuacoes_de_genero), ("humor", resultado.pontuacoes_de_humor)):
                self.conexao.executemany(
                    "INSERT INTO etiquetas_musicais(analise_id, tipo, etiqueta, confianca, ordem) VALUES (?, ?, ?, ?, ?)",
                    ((analise_id, tipo, item.etiqueta, item.confianca, ordem) for ordem, item in enumerate(etiquetas)),
                )

    def registrar_erro(self, trilha_id: int, erro: str) -> None:
        """Marca uma trilha como erro sem interromper o lote inteiro."""
        with self.conexao:
            self.conexao.execute(
                "UPDATE trilhas_musicais SET status = 'erro', erro = ?, atualizada_em = strftime('%Y-%m-%dT%H:%M:%fZ','now') WHERE id = ?",
                (erro[:2000], trilha_id),
            )

    @staticmethod
    def _calcular_hash(caminho: Path) -> str:
        """Calcula SHA-256 em blocos para não carregar o áudio inteiro na memória."""
        resumo = hashlib.sha256()
        with caminho.open("rb") as arquivo:
            for bloco in iter(lambda: arquivo.read(1024 * 1024), b""):
                resumo.update(bloco)
        return resumo.hexdigest()