458 lines
15 KiB
Python
458 lines
15 KiB
Python
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()
|