feat: criado repositório jhonny-editor no Gitea
criado repositório jhonny-editor no Gitea adicionado script admin/deploy.command com commit automático atualizado admin/DEV-NOTES.md com template limpo Resumo: - 4 arquivos alterados - 2 novos - 2 modificados - 0 removidos 2 files changed, 48 insertions(+) Arquivos: - .jhonny/analises.db - code/engine/persistencia/esquema.py - code/engine/analisar_trilhas.py - code/engine/persistencia/repositorio_de_trilhas.py
This commit is contained in:
@@ -0,0 +1,114 @@
|
||||
"""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:
|
||||
cursor = 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 = cursor.lastrowid or 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()
|
||||
Reference in New Issue
Block a user