Limpa dependências versionadas e reconstrói catálogo com 20 projetos

scripts/classificacao/analise_audio.py

@@ -48,14 +48,19 @@ essentia.log.info.active = False
 essentia.log.warning.active = False
 
 # ================= CONFIGURAÇÕES =================
-BASE_DIR = "/var/www/html/trens/src_mmpSearch"
+import sys
+from pathlib import Path
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from etl_settings import DATA_ROOT, MODELS_ROOT, worker_count
+
+BASE_DIR = str(DATA_ROOT)
 PASTA_WAVS = os.path.join(BASE_DIR, "wav")
 PASTA_LOTES_YAML = os.path.join(BASE_DIR, "lotes_yaml")
 
 # --- SISTEMA DE DOIS MODELOS ---
-MODELO_EMBEDDING = "discogs-effnet-bs64-1.pb"
-MODELO_CLASSIFIER = "genre_discogs400.pb"
-MODELO_CLASSES = "genre_discogs400.json"
+MODELO_EMBEDDING = str(MODELS_ROOT / "discogs-effnet-bs64-1.pb")
+MODELO_CLASSIFIER = str(MODELS_ROOT / "genre_discogs400.pb")
+MODELO_CLASSES = str(MODELS_ROOT / "genre_discogs400.json")
 
 # Saídas
 DIR_SAIDA = os.path.join(BASE_DIR, "saida_analises")
@@ -516,9 +521,9 @@ def analisar_faixa(caminho_arquivo, embedding_model, classifier_model, classes_r
         except Exception as e:
             msg = f"Erro IA Pipeline: {str(e)}"
             logging.error(msg)
-            metadata["analise_ia"] = {"erro": msg}
+            raise RuntimeError(msg) from e
     else:
-        metadata["analise_ia"] = {"status": "Modelos nao carregados"}
+        raise RuntimeError("Modelos de IA nao inicializados no worker.")
 
     # Limpeza Limpa
     del audio_hq
@@ -530,35 +535,10 @@ def analisar_faixa(caminho_arquivo, embedding_model, classifier_model, classes_r
 
 
 # ================= MAIN (Multiprocessing) =================
-def worker_analise(caminho):
-    # Processa usando as instâncias dos modelos carregadas na thread mestre
-    global GLOBAL_EMBEDDING, GLOBAL_CLASSIFIER, GLOBAL_CLASSES
-    return analisar_faixa(caminho, GLOBAL_EMBEDDING, GLOBAL_CLASSIFIER, GLOBAL_CLASSES)
-
-
-def main():
-    print("\n--- INICIANDO PROCESSADOR + AUDITORIA ---")
-
-    # Referencia variávies globais do script para alimentar os childs process do pool
+def initialize_analysis_worker():
+    # Cada processo cria suas proprias sessoes TensorFlow.
     global DATABASE_YAML, GLOBAL_EMBEDDING, GLOBAL_CLASSIFIER, GLOBAL_CLASSES
-
     DATABASE_YAML = carregar_database_yaml()
-
-    ja_processados, lista_resultados = carregar_checkpoint()
-
-    auditor = Auditoria(ARQUIVO_AUDITORIA)
-
-    if not os.path.exists(PASTA_WAVS):
-        print("CRÍTICO: Pasta WAV não encontrada!")
-        return
-
-    todos = [f for f in os.listdir(PASTA_WAVS) if f.lower().endswith((".wav", ".ogg"))]
-    a_fazer = [os.path.join(PASTA_WAVS, f) for f in todos if f not in ja_processados]
-
-    total = len(a_fazer)
-    print(f"Total Arquivos: {len(todos)} | Novos a Fazer: {total}")
-    print(f"Auditoria será salva em: {ARQUIVO_AUDITORIA}")
-
     # === CARREGAMENTO MODELOS GLOBAIS ===
     path_embed = (
         MODELO_EMBEDDING
@@ -608,24 +588,61 @@ def main():
             print("[2/2] Classificador de Gênero carregado.")
 
         except Exception as e:
-            print(f"ERRO CRÍTICO AO CARREGAR IA: {e}")
-            return
+            raise RuntimeError("Falha ao carregar modelos de IA") from e
     else:
-        print("Arquivos de modelo incompletos.")
-        return
+        raise FileNotFoundError("Arquivos de modelo incompletos.")
+
+
+
+def worker_analise(caminho):
+    # Modelos inicializados neste processo, sem herdar sessoes via fork.
+    print(f"[PID {os.getpid()}] Iniciando analise: {os.path.basename(caminho)}", flush=True)
+    global GLOBAL_EMBEDDING, GLOBAL_CLASSIFIER, GLOBAL_CLASSES
+    result = analisar_faixa(caminho, GLOBAL_EMBEDDING, GLOBAL_CLASSIFIER, GLOBAL_CLASSES)
+    print(f"[PID {os.getpid()}] Finalizou analise: {os.path.basename(caminho)}", flush=True)
+    return result
+
+
+def main():
+    print("\n--- INICIANDO PROCESSADOR + AUDITORIA ---")
+
+    # Referencia variávies globais do script para alimentar os childs process do pool
+    global DATABASE_YAML, GLOBAL_EMBEDDING, GLOBAL_CLASSIFIER, GLOBAL_CLASSES
+
+    DATABASE_YAML = carregar_database_yaml()
+
+    ja_processados, lista_resultados = carregar_checkpoint()
+
+    auditor = Auditoria(ARQUIVO_AUDITORIA)
+
+    if not os.path.exists(PASTA_WAVS):
+        raise RuntimeError("Pasta de audio nao encontrada.")
+
+    todos = [f for f in os.listdir(PASTA_WAVS) if f.lower().endswith((".wav", ".ogg"))]
+    a_fazer = [os.path.join(PASTA_WAVS, f) for f in todos if f not in ja_processados]
+
+    total = len(a_fazer)
+    print(f"Total Arquivos: {len(todos)} | Novos a Fazer: {total}")
+    print(f"Auditoria será salva em: {ARQUIVO_AUDITORIA}")
 
     if total == 0:
-        print("Tudo atualizado.")
+        print("Nenhum audio novo; reconstruindo o consolidado do checkpoint.")
+        if not lista_resultados:
+            raise RuntimeError("Nao ha audios nem resultados no checkpoint.")
+        with open(ARQUIVO_FINAL_JSON, "w", encoding="utf-8") as f:
+            json.dump(lista_resultados, f, indent=4, ensure_ascii=False)
         return
 
     # ProcessPoolExecutor lidará com N filas ao mesmo tempo
-    max_workers = max(1, multiprocessing.cpu_count() - 1)
+    max_workers = worker_count()
     print(f"-> Iniciando processamento paralelo com {max_workers} workers.")
     contagem_sessao = 0
 
     try:
         with concurrent.futures.ProcessPoolExecutor(
-            max_workers=max_workers
+            max_workers=max_workers,
+            mp_context=multiprocessing.get_context("spawn"),
+            initializer=initialize_analysis_worker,
         ) as executor:
             resultados_futuros = {
                 executor.submit(worker_analise, caminho): caminho for caminho in a_fazer
@@ -657,7 +674,7 @@ def main():
                     logging.error(f"Erro ao processar um dos arquivos no worker: {e}")
 
     except KeyboardInterrupt:
-        print("\nParando paralelismo graciosamente...")
+        raise RuntimeError("Analise interrompida pelo operador.")
 
     finally:
         auditor.fechar()  # Fecha o CSV com segurança
@@ -670,6 +687,12 @@ def main():
     with open(ARQUIVO_FINAL_JSON, "w", encoding="utf-8") as f:
         json.dump(lista_resultados, f, indent=4, ensure_ascii=False)
 
+    expected = {os.path.basename(path) for path in a_fazer}
+    obtained = {item.get("arquivo") for item in lista_resultados}
+    missing = expected - obtained
+    if missing:
+        raise RuntimeError(f"Analises ausentes para {len(missing)} audios: {sorted(missing)}")
+
     print(f"Dados JSON: {ARQUIVO_FINAL_JSON}")
     print(f"Auditoria CSV: {ARQUIVO_AUDITORIA}")