att base de dados
This commit is contained in:
JotaChina committed
1 parent
ceb7908da4
commit
73693e5da9
14 files changed
+3010
-1
No files matched your search
@@ -0,0 +1,457 @@
|
||||
import yaml
|
||||
import json
|
||||
import os
|
||||
import logging
|
||||
import csv
|
||||
import unicodedata
|
||||
import time
|
||||
import psutil
|
||||
import glob # Import necessário para ler múltiplos arquivos
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
# --- CONFIGURAÇÕES GERAIS ---
|
||||
BASE_DIR = "/var/www/html/trens/src_mmpSearch"
|
||||
LOG_DIR = os.path.join(BASE_DIR, "logs/classificacao_pedagogica")
|
||||
os.makedirs(LOG_DIR, exist_ok=True)
|
||||
TIMESTAMP = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
|
||||
|
||||
# Configuração de Log
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s - CLASSIFICADOR - %(levelname)s - %(message)s",
|
||||
handlers=[
|
||||
logging.FileHandler(os.path.join(LOG_DIR, f"log_pedagogico_{TIMESTAMP}.log")),
|
||||
logging.StreamHandler(),
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
# --- ARQUIVOS DE ENTRADA/SAÍDA ---
|
||||
def encontrar_caminho(nome_arquivo):
|
||||
"""Procura o arquivo em locais comuns do projeto."""
|
||||
candidatos = [
|
||||
os.path.join(BASE_DIR, "saida_analises", nome_arquivo),
|
||||
os.path.join(BASE_DIR, "data", nome_arquivo),
|
||||
os.path.join(BASE_DIR, nome_arquivo),
|
||||
nome_arquivo,
|
||||
]
|
||||
for c in candidatos:
|
||||
if os.path.exists(c):
|
||||
return c
|
||||
return nome_arquivo
|
||||
|
||||
|
||||
# --- MUDANÇA AQUI: Aponta para a pasta dos lotes ---
|
||||
PASTA_LOTES_YAML = os.path.join(BASE_DIR, "lotes_yaml")
|
||||
# ARQUIVO_YAML = "all.yml" <-- Removido/Comentado
|
||||
|
||||
ARQUIVO_AUDIO_INPUT = encontrar_caminho("db_final_completo.json")
|
||||
ARQUIVO_FINAL_OUTPUT = os.path.join(
|
||||
BASE_DIR, "saida_analises", "db_projetos_classificados.json"
|
||||
)
|
||||
ARQUIVO_AUDITORIA = os.path.join(
|
||||
BASE_DIR, "saida_analises", "audit_classificacao_pedagogica.csv"
|
||||
)
|
||||
|
||||
# --- LISTAS DE REFERÊNCIA ---
|
||||
PLUGINS_NATIVOS = [
|
||||
"tripleoscillator",
|
||||
"zynaddsubfx",
|
||||
"sfxr",
|
||||
"organic",
|
||||
"audiofileprocessor",
|
||||
"lb302",
|
||||
"kicker",
|
||||
"watsyn",
|
||||
"bitinvader",
|
||||
"freeboy",
|
||||
"mallets",
|
||||
"vibed",
|
||||
]
|
||||
|
||||
|
||||
# --- CLASSE DE AUDITORIA DE PERFORMANCE ---
|
||||
class AuditoriaPerformance:
|
||||
def __init__(self):
|
||||
self.inicio_global = time.time()
|
||||
self.process = psutil.Process(os.getpid())
|
||||
self.marcos = [1, 10, 100, 500, 1000, 2000, 5000]
|
||||
|
||||
def get_metricas(self):
|
||||
"""Retorna uso atual de RAM (MB) e CPU (%)"""
|
||||
try:
|
||||
ram_mb = self.process.memory_info().rss / (1024 * 1024)
|
||||
cpu_pct = self.process.cpu_percent(interval=None)
|
||||
return round(ram_mb, 2), cpu_pct
|
||||
except:
|
||||
return 0.0, 0.0
|
||||
|
||||
def verificar_marco(self, contagem):
|
||||
if contagem in self.marcos:
|
||||
tempo_decorrido = time.time() - self.inicio_global
|
||||
logging.info(
|
||||
f"--- MARCO: {contagem} projetos processados em {str(timedelta(seconds=int(tempo_decorrido)))} ---"
|
||||
)
|
||||
|
||||
|
||||
# --- FUNÇÕES AUXILIARES ---
|
||||
|
||||
|
||||
def normalizar_chave(texto):
|
||||
if not texto:
|
||||
return ""
|
||||
s = str(texto)
|
||||
if "." in s:
|
||||
s = os.path.splitext(s)[0]
|
||||
s = unicodedata.normalize("NFKD", s).encode("ASCII", "ignore").decode("ASCII")
|
||||
return s.lower().replace(" ", "").replace("-", "").replace("_", "").strip()
|
||||
|
||||
|
||||
def calcular_nivel_ponderado(
|
||||
num_tracks, tem_automacao, usa_zyn, qtd_secoes_audio, fx_criativos
|
||||
):
|
||||
pontos = 0
|
||||
|
||||
# 1. Volume
|
||||
if num_tracks >= 8:
|
||||
pontos += 1
|
||||
if num_tracks >= 16:
|
||||
pontos += 2
|
||||
if num_tracks >= 32:
|
||||
pontos += 3
|
||||
if num_tracks >= 64:
|
||||
pontos += 4
|
||||
elif num_tracks >= 128:
|
||||
pontos += 5
|
||||
|
||||
# 2. Técnica
|
||||
if tem_automacao:
|
||||
pontos += 2
|
||||
if usa_zyn:
|
||||
pontos += 1.5
|
||||
if fx_criativos:
|
||||
pontos += 1
|
||||
|
||||
# 3. Estrutura (Áudio)
|
||||
if qtd_secoes_audio >= 3:
|
||||
pontos += 1.5
|
||||
elif qtd_secoes_audio >= 5:
|
||||
pontos += 3
|
||||
elif qtd_secoes_audio >= 10:
|
||||
pontos += 5
|
||||
|
||||
if pontos < 3:
|
||||
return "Iniciante"
|
||||
if pontos < 6:
|
||||
return "Intermediário"
|
||||
return "Avançado"
|
||||
|
||||
|
||||
def analisar_tags_pedagogicas(tracks, bpm, qtd_secoes):
|
||||
tags = []
|
||||
tem_kicker = False
|
||||
tem_zyn = False
|
||||
tem_automacao = False
|
||||
tem_fx_chain = False
|
||||
rolagem_trap = False
|
||||
|
||||
for t in tracks:
|
||||
inst = t.get("instrumenttrack")
|
||||
if not inst:
|
||||
continue
|
||||
|
||||
nome = inst.get("name", "").lower()
|
||||
plugin = inst.get("plugin_name", "").lower()
|
||||
|
||||
if plugin == "kicker":
|
||||
tem_kicker = True
|
||||
if plugin == "zynaddsubfx":
|
||||
tem_zyn = True
|
||||
if "automation" in str(inst).lower():
|
||||
tem_automacao = True
|
||||
|
||||
fx = inst.get("fxchain", {})
|
||||
if fx and int(fx.get("numofeffects", 0)) > 0:
|
||||
tem_fx_chain = True
|
||||
|
||||
patterns = inst.get("patterns", [])
|
||||
if patterns:
|
||||
for pat in patterns:
|
||||
steps = pat.get("steps", [])
|
||||
consecutivos = 0
|
||||
for step in steps:
|
||||
val = 1 if step else 0
|
||||
consecutivos = (consecutivos + 1) if val else 0
|
||||
if consecutivos >= 3 and ("hat" in nome):
|
||||
rolagem_trap = True
|
||||
|
||||
if tem_kicker:
|
||||
tags.append("Síntese de Bateria")
|
||||
if tem_zyn:
|
||||
tags.append("Design de Som Avançado")
|
||||
if tem_automacao:
|
||||
tags.append("Automação")
|
||||
if tem_fx_chain:
|
||||
tags.append("Mixagem com FX")
|
||||
|
||||
if rolagem_trap and bpm > 115:
|
||||
tags.append("Trap Rhythm")
|
||||
if bpm < 100 and not rolagem_trap:
|
||||
tags.append("Boombap Groove")
|
||||
|
||||
if qtd_secoes >= 4:
|
||||
tags.append("Arranjo Completo")
|
||||
elif qtd_secoes <= 1:
|
||||
tags.append("Loop Estático")
|
||||
else:
|
||||
tags.append("Estrutura Simples")
|
||||
|
||||
return tags, tem_automacao, tem_zyn, tem_fx_chain
|
||||
|
||||
|
||||
# --- FUNÇÃO PARA CARREGAR LOTES YAML ---
|
||||
def carregar_todos_yamls(pasta_lotes):
|
||||
"""Lê todos os arquivos YAML da pasta especificada e retorna uma lista única."""
|
||||
logging.info(f"Buscando arquivos YAML em: {pasta_lotes}")
|
||||
|
||||
padrao = os.path.join(pasta_lotes, "*.yml")
|
||||
arquivos = sorted(glob.glob(padrao))
|
||||
|
||||
todos_projetos = []
|
||||
|
||||
if not arquivos:
|
||||
logging.warning("Nenhum arquivo YAML encontrado na pasta de lotes.")
|
||||
# Tenta fallback para o arquivo único antigo se existir
|
||||
caminho_unico = encontrar_caminho("all.yml")
|
||||
if os.path.exists(caminho_unico):
|
||||
logging.info(f"Fallback: Carregando arquivo único {caminho_unico}")
|
||||
with open(caminho_unico, "r", encoding="utf-8") as f:
|
||||
return yaml.safe_load(f)
|
||||
return []
|
||||
|
||||
for arq in arquivos:
|
||||
try:
|
||||
logging.info(f"Carregando lote: {os.path.basename(arq)}")
|
||||
with open(arq, "r", encoding="utf-8") as f:
|
||||
lote = yaml.safe_load(f)
|
||||
if lote:
|
||||
# Garante que seja uma lista
|
||||
if isinstance(lote, list):
|
||||
todos_projetos.extend(lote)
|
||||
elif isinstance(lote, dict):
|
||||
todos_projetos.append(lote)
|
||||
except Exception as e:
|
||||
logging.error(f"Erro ao ler arquivo {arq}: {e}")
|
||||
|
||||
logging.info(f"Total de projetos carregados dos lotes: {len(todos_projetos)}")
|
||||
return todos_projetos
|
||||
|
||||
|
||||
# --- MAIN ---
|
||||
|
||||
|
||||
def main():
|
||||
logging.info("--- INICIANDO CLASSIFICADOR PEDAGÓGICO (COM AUDITORIA E LOTES) ---")
|
||||
|
||||
auditor = AuditoriaPerformance()
|
||||
|
||||
# 1. Carregar YAML (Adaptado para Lotes)
|
||||
if not os.path.exists(PASTA_LOTES_YAML):
|
||||
logging.warning(f"Pasta de lotes não encontrada: {PASTA_LOTES_YAML}")
|
||||
# Tenta criar se não existir (embora deva existir com arquivos)
|
||||
# os.makedirs(PASTA_LOTES_YAML, exist_ok=True)
|
||||
|
||||
dados_projetos = carregar_todos_yamls(PASTA_LOTES_YAML)
|
||||
|
||||
if not dados_projetos:
|
||||
logging.critical("Nenhum dado de projeto carregado. Encerrando.")
|
||||
return
|
||||
|
||||
# 2. Carregar JSON de Áudio
|
||||
lookup_audio = {}
|
||||
total_audio_carregado = 0
|
||||
|
||||
if os.path.exists(ARQUIVO_AUDIO_INPUT):
|
||||
logging.info(f"Lendo Análise de Áudio: {ARQUIVO_AUDIO_INPUT}")
|
||||
with open(ARQUIVO_AUDIO_INPUT, "r", encoding="utf-8") as f:
|
||||
lista_audio = json.load(f)
|
||||
for item in lista_audio:
|
||||
chave = normalizar_chave(item.get("arquivo", ""))
|
||||
if chave:
|
||||
lookup_audio[chave] = item
|
||||
total_audio_carregado = len(lista_audio)
|
||||
else:
|
||||
logging.warning(f"Arquivo de áudio {ARQUIVO_AUDIO_INPUT} não encontrado.")
|
||||
|
||||
db_final = []
|
||||
estatisticas = {"Iniciante": 0, "Intermediário": 0, "Avançado": 0, "Total": 0}
|
||||
|
||||
# --- PREPARAR CSV DE AUDITORIA ---
|
||||
with open(ARQUIVO_AUDITORIA, "w", newline="", encoding="utf-8") as csvfile:
|
||||
writer = csv.writer(csvfile)
|
||||
# Cabeçalho expandido com métricas de performance
|
||||
writer.writerow(
|
||||
[
|
||||
"Timestamp",
|
||||
"ID_Projeto",
|
||||
"Audio_Encontrado",
|
||||
"Qtd_Secoes",
|
||||
"Nivel",
|
||||
"Genero_Final",
|
||||
"Tempo_Proc_ms",
|
||||
"RAM_Uso_MB",
|
||||
"Tags",
|
||||
]
|
||||
)
|
||||
|
||||
logging.info(f"Processando {len(dados_projetos)} projetos...")
|
||||
|
||||
contagem_processados = 0
|
||||
|
||||
for proj in dados_projetos:
|
||||
if not proj or not isinstance(proj, dict):
|
||||
continue
|
||||
|
||||
# --- INÍCIO TIMER INDIVIDUAL ---
|
||||
inicio_item = time.perf_counter()
|
||||
|
||||
nome_arquivo = proj.get("file", "")
|
||||
if not nome_arquivo:
|
||||
continue
|
||||
|
||||
try:
|
||||
# Lógica de Classificação
|
||||
chave_proj = normalizar_chave(nome_arquivo)
|
||||
dados_audio = lookup_audio.get(chave_proj, {})
|
||||
|
||||
analise_tec = dados_audio.get("analise_tecnica", {})
|
||||
analise_ia = dados_audio.get("analise_ia", {})
|
||||
|
||||
raw_bpm = proj.get("bpm", 120)
|
||||
try:
|
||||
bpm_projeto = float(raw_bpm)
|
||||
except:
|
||||
bpm_projeto = 120.0
|
||||
|
||||
est_seg = analise_tec.get("estrutura_segundos", [])
|
||||
qtd_secoes = len(est_seg) if isinstance(est_seg, list) else 1
|
||||
|
||||
tracks = proj.get("tracks", [])
|
||||
tags, tem_auto, tem_zyn, tem_fx = analisar_tags_pedagogicas(
|
||||
tracks, bpm_projeto, qtd_secoes
|
||||
)
|
||||
|
||||
plugins_usados = set()
|
||||
usa_vst_externo = False
|
||||
for t in tracks:
|
||||
inst_track = t.get("instrumenttrack", {})
|
||||
if not inst_track:
|
||||
continue
|
||||
p = inst_track.get("plugin_name", "")
|
||||
if p:
|
||||
p_norm = p.lower()
|
||||
plugins_usados.add(p_norm)
|
||||
if p_norm not in PLUGINS_NATIVOS:
|
||||
usa_vst_externo = True
|
||||
|
||||
nivel = calcular_nivel_ponderado(
|
||||
len(tracks), tem_auto, tem_zyn, qtd_secoes, tem_fx
|
||||
)
|
||||
estatisticas[nivel] += 1
|
||||
estatisticas["Total"] += 1
|
||||
|
||||
genero = "Desconhecido"
|
||||
if analise_ia and "estilo_principal" in analise_ia:
|
||||
genero = analise_ia["estilo_principal"]
|
||||
elif (
|
||||
"genero_macro" in analise_ia
|
||||
and analise_ia["genero_macro"] != "Unknown"
|
||||
):
|
||||
genero = analise_ia["genero_macro"]
|
||||
|
||||
if genero == "Desconhecido" or genero == "Unknown":
|
||||
if "Trap Rhythm" in tags:
|
||||
genero = "Trap (Rule-based)"
|
||||
elif bpm_projeto > 120:
|
||||
genero = "Electronic (Rule-based)"
|
||||
else:
|
||||
genero = "HipHop/Downtempo (Rule-based)"
|
||||
|
||||
# Objeto Final
|
||||
item = {
|
||||
"id": nome_arquivo,
|
||||
"titulo": proj.get("original_title", nome_arquivo),
|
||||
"match_audio": bool(dados_audio),
|
||||
"pedagogia": {
|
||||
"nivel": nivel,
|
||||
"tags_aprendizado": tags,
|
||||
"plugins_principais": list(plugins_usados),
|
||||
"requer_vst_externo": usa_vst_externo,
|
||||
},
|
||||
"tecnica": {
|
||||
"bpm": bpm_projeto,
|
||||
"tom": analise_tec.get("tom", "N/A"),
|
||||
"escala": analise_tec.get("escala", "N/A"),
|
||||
"estrutura_qtd_secoes": qtd_secoes,
|
||||
"intensidade_media": analise_tec.get("intensidade_db", "N/A"),
|
||||
},
|
||||
"classificacao": {
|
||||
"genero": genero,
|
||||
"mood": "Energético"
|
||||
if analise_tec.get("intensidade_db", -20) > -10
|
||||
else "Suave",
|
||||
},
|
||||
}
|
||||
|
||||
db_final.append(item)
|
||||
|
||||
# --- FIM TIMER INDIVIDUAL ---
|
||||
tempo_ms = (time.perf_counter() - inicio_item) * 1000
|
||||
ram_atual, _ = auditor.get_metricas()
|
||||
ts_agora = datetime.now().strftime("%H:%M:%S")
|
||||
|
||||
# Registra Auditoria com Performance
|
||||
writer.writerow(
|
||||
[
|
||||
ts_agora,
|
||||
nome_arquivo,
|
||||
"SIM" if dados_audio else "NAO",
|
||||
qtd_secoes,
|
||||
nivel,
|
||||
genero,
|
||||
f"{tempo_ms:.3f}", # Tempo em milissegundos com 3 casas
|
||||
f"{ram_atual:.2f}", # RAM em MB
|
||||
"|".join(tags),
|
||||
]
|
||||
)
|
||||
|
||||
contagem_processados += 1
|
||||
auditor.verificar_marco(contagem_processados)
|
||||
|
||||
except Exception as e:
|
||||
logging.error(
|
||||
f"Erro ao processar projeto '{proj.get('file', 'unknown')}': {e}"
|
||||
)
|
||||
continue
|
||||
|
||||
# Salvar Resultado JSON
|
||||
logging.info(f"Salvando JSON Final: {ARQUIVO_FINAL_OUTPUT}")
|
||||
with open(ARQUIVO_FINAL_OUTPUT, "w", encoding="utf-8") as f:
|
||||
json.dump(db_final, f, indent=4, ensure_ascii=False)
|
||||
|
||||
tempo_total = time.time() - auditor.inicio_global
|
||||
|
||||
logging.info("--- RELATÓRIO FINAL ---")
|
||||
logging.info(f"Tempo Total de Execução: {str(timedelta(seconds=int(tempo_total)))}")
|
||||
logging.info(f"Total Processado: {estatisticas['Total']}")
|
||||
logging.info(f"Iniciantes: {estatisticas['Iniciante']}")
|
||||
logging.info(f"Intermediários: {estatisticas['Intermediário']}")
|
||||
logging.info(f"Avançados: {estatisticas['Avançado']}")
|
||||
logging.info(
|
||||
f"Taxa de Casamento com Áudio: {len(lookup_audio)}/{total_audio_carregado}"
|
||||
)
|
||||
logging.info(f"Auditoria salva em: {ARQUIVO_AUDITORIA}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in new issue
Block a user