Tetando pós scripts

scripts/handler/main.py

@@ -1,7 +1,6 @@
 # main.py
 
 import os
-import json
 import shutil
 import subprocess
 import multiprocessing
@@ -11,7 +10,9 @@ import platform
 import sys
 import re
 import unicodedata
-from datetime import datetime
+import csv
+import psutil
+from datetime import datetime, timedelta
 
 # Importando seus módulos locais
 from generate_manifest import generate_manifests
@@ -20,7 +21,6 @@ from file_saver import save_to_json, save_to_yaml
 from dependency_checker import check_dependencies
 from utils import (
     create_folders_if_not_exist,
-    BASE_PATH,
     DATA_FOLDER,
     METADATA_FOLDER,
     WAV_FOLDER,
@@ -28,27 +28,45 @@ from utils import (
     MMP_FOLDER,
     SRC_MMPSEARCH,
     LOG_FOLDER,
+    SAIDA_ANALISES,
 )
 
+# === CONFIGURAÇÕES DE SEGURANÇA E LOTES ===
+TIMEOUT_RENDER_SECONDS = 300  # 5 minutos máximo por arquivo
+MIN_RAM_FREE_MB = 500  # Mínimo de RAM livre para iniciar worker
+BATCH_SIZE = 250  # Tamanho do lote para o YAML
+LOTES_FOLDER = os.path.join(SRC_MMPSEARCH, "lotes_yaml")  # Pasta dos lotes
+LIMIT_FILES = 1000  # <--- 0 = PROCESSA TUDO | >0 = LIMITA A QUANTIDADE (EX: 50)
+
+# === VARIÁVEL GLOBAL PARA O WORKER (CORREÇÃO MULTIPROCESSING) ===
+# Necessário para evitar erro de pickle ao passar Lock via argumentos
+global_counter = None
+
+
+def init_worker(counter):
+    """Inicializa a variável global dentro de cada processo worker"""
+    global global_counter
+    global_counter = counter
+
+
 # === Função de Sanitização (Slugify) ===
 def slugify(value):
     """
     Normaliza a string: remove acentos, converte para minúsculas,
     remove caracteres não alfanuméricos e substitui espaços por hifens.
-    Ex: 'Ação & Reação!' -> 'acao-reacao'
     """
     value = str(value)
-    # Normaliza unicode (ex: ã -> a)
-    value = unicodedata.normalize('NFKD', value).encode('ascii', 'ignore').decode('ascii')
+    value = (
+        unicodedata.normalize("NFKD", value).encode("ascii", "ignore").decode("ascii")
+    )
     value = value.lower()
-    # Remove tudo que não for letra, número ou espaço/hífen
-    value = re.sub(r'[^\w\s-]', '', value)
-    # Substitui espaços e underlines por hífen
-    value = re.sub(r'[-\s_]+', '-', value)
-    return value.strip('-_')
+    value = re.sub(r"[^\w\s-]", "", value)
+    value = re.sub(r"[-\s_]+", "-", value)
+    return value.strip("-_")
+
 
 def check_system_dependencies():
-    """Verifica se as ferramentas necessárias estão instaladas antes de iniciar."""
+    """Verifica se as ferramentas necessárias estão instaladas."""
     required_tools = ["lmms"]
     missing = []
     for tool in required_tools:
@@ -61,6 +79,16 @@ def check_system_dependencies():
         sys.exit(1)
 
 
+def get_cpu_safe_count():
+    """Retorna contagem de CPU menos 1 para não travar o SO."""
+    try:
+        count = multiprocessing.cpu_count()
+        return max(1, count - 1)
+    except:
+        return 1
+
+
+# --- Funções de Info de Sistema ---
 def get_linux_mem_info():
     try:
         with open("/proc/meminfo", "r") as f:
@@ -95,9 +123,12 @@ def log_system_info():
         logging.info(f"Node:    {uname.node}")
         cpu_model = get_cpu_model_name()
         cores_logical = multiprocessing.cpu_count()
+        safe_cores = get_cpu_safe_count()
         mem_total_gb = get_linux_mem_info()
         logging.info(f"CPU Modelo:   {cpu_model}")
-        logging.info(f"Núcleos:      {cores_logical}")
+        logging.info(
+            f"Núcleos (Total): {cores_logical} | Workers Seguros: {safe_cores}"
+        )
         logging.info(f"Memória Total: {mem_total_gb:.2f} GB")
         total, used, free = shutil.disk_usage(".")
         logging.info(f"Disco (Livre): {free // (2**30)} GB")
@@ -126,146 +157,194 @@ def setup_logger():
 
 def process_single_file(args):
     """
-    Worker robusto.
-    Garante geração de WAV tanto para .mmpz quanto para .mmp.
+    Worker robusto com CONTADOR COMPARTILHADO (THREAD-SAFE).
     """
-    # === TRATAMENTO DE ENTRADA (Tupla ou String) ===
-    if isinstance(args, str):
-        file_name = args
-        original_base = os.path.splitext(file_name)[0]
-        clean_slug = slugify(original_base)
-        if not clean_slug:
-            clean_slug = f"upload-incorreto-{int(time.time())}"
-    else:
-        file_name, clean_slug = args
-    # ===============================================
-    
+    # Desempacota os argumentos (O CONTADOR NÃO VEM MAIS AQUI)
+    file_name, clean_slug, total_files = args
+
     file_path = os.path.join(MMP_FOLDER, file_name)
     pid = os.getpid()
-    result = {"success": False, "file": file_name, "data": None, "error": None}
 
-    # Recupera o nome original base para exibição
-    original_base_name = os.path.splitext(file_name)[0]
-    
-    # Define os nomes de saída baseados no Slug
-    wav_name = clean_slug + ".wav"
-    json_name = clean_slug + ".json"
-    yml_name = clean_slug + ".yml"
-    
-    # Variável que vai armazenar o caminho final do arquivo .mmp a ser processado/renderizado
-    target_mmp_path = ""
+    # === Lógica do Contador (Acessando via Global) ===
+    # O global_counter foi injetado pelo init_worker
+    global global_counter
+    current_idx = 0
+
+    if global_counter:
+        with global_counter.get_lock():
+            global_counter.value += 1
+            current_idx = global_counter.value
+
+    remaining = total_files - current_idx
+
+    # Log com status de progresso
+    logging.info(
+        f"[PID {pid}] [Progresso: {current_idx}/{total_files} | Faltam: {remaining}] Iniciando: {file_name}"
+    )
+
+    # Métricas para Auditoria
+    start_time = time.time()
+    ram_usage_mb = 0.0
+
+    result = {
+        "success": False,
+        "file": file_name,
+        "slug": clean_slug,
+        "data": None,
+        "error": None,
+        "duration_s": 0.0,
+        "ram_mb": 0.0,
+        "file_size_mb": 0.0,
+    }
 
     try:
-        logging.info(f"[PID {pid}] Processando: {file_name} -> Slug: {clean_slug}")
+        # Checagem de Tamanho
+        if os.path.exists(file_path):
+            result["file_size_mb"] = os.path.getsize(file_path) / (1024 * 1024)
+
+        # Checagem de Segurança de RAM
+        mem = psutil.virtual_memory()
+        if (mem.available / (1024 * 1024)) < MIN_RAM_FREE_MB:
+            result["error"] = "RAM Insuficiente para iniciar worker"
+            logging.warning(f"[PID {pid}] Pulei {file_name}: RAM Crítica")
+            return result
+
+        original_base_name = os.path.splitext(file_name)[0]
+        wav_name = clean_slug + ".wav"
+        json_name = clean_slug + ".json"
+        yml_name = clean_slug + ".yml"
+
+        target_mmp_path = ""
 
         # 1. Tratamento e Extração
         if file_name.endswith(".mmpz"):
             destination_path = os.path.join(MMPZ_FOLDER, file_name)
 
-            # Move o mmpz original para a pasta de backup/mmpz
             if not os.path.exists(destination_path):
                 shutil.move(file_path, destination_path)
             elif os.path.exists(file_path):
                 os.remove(file_path)
 
-            # Define o nome do arquivo MMP extraído usando o SLUG
-            mmp_temp_name = clean_slug + ".mmp" 
+            mmp_temp_name = clean_slug + ".mmp"
             output_mmp_path = os.path.join(MMP_FOLDER, mmp_temp_name)
-
             abs_dest = os.path.abspath(destination_path)
             abs_mmp_out = os.path.abspath(output_mmp_path)
 
-            # Comando para descompactar (dump)
-            dump_cmd = ["lmms", "--dump", abs_dest]
-
-            try:
-                with open(abs_mmp_out, "w") as outfile:
-                    subprocess.run(
-                        dump_cmd,
-                        stdout=outfile,
-                        stderr=subprocess.PIPE,
-                        check=True,
-                        env={"QT_QPA_PLATFORM": "offscreen", **os.environ},
-                    )
-                # Define o arquivo extraído como o alvo para renderização e parsing
-                target_mmp_path = output_mmp_path
-
-            except subprocess.CalledProcessError as e:
-                result["error"] = f"Erro no dump MMP: {e.stderr.decode('utf-8')}"
-                logging.error(f"[PID {pid}] {result['error']}")
-                return result
+            with open(abs_mmp_out, "w") as outfile:
+                subprocess.run(
+                    ["lmms", "--dump", abs_dest],
+                    stdout=outfile,
+                    stderr=subprocess.PIPE,
+                    check=True,
+                    timeout=60,
+                    env={"QT_QPA_PLATFORM": "offscreen", **os.environ},
+                )
+            target_mmp_path = output_mmp_path
 
         elif file_name.endswith(".mmp"):
-            # Se já for MMP, ele é o próprio alvo
             target_mmp_path = file_path
 
-        # 2. GERAÇÃO DE ÁUDIO (AGORA FORA DO IF/ELSE)
-        # Isso garante que rode tanto para MMPZ (usando o extraído) quanto para MMP puro
+        # 2. GERAÇÃO DE ÁUDIO
         if os.path.exists(target_mmp_path):
             abs_target_mmp = os.path.abspath(target_mmp_path)
             abs_wav_out = os.path.abspath(os.path.join(WAV_FOLDER, wav_name))
 
-            wav_cmd = [
-                "lmms",
-                "-r",
-                abs_target_mmp, # Usa sempre o .mmp descompactado
-                "-o",
-                abs_wav_out,
-                "-f",
-                "wav",
-            ]
+            render_process = subprocess.run(
+                ["lmms", "-r", abs_target_mmp, "-o", abs_wav_out, "-f", "wav"],
+                check=False,
+                capture_output=True,
+                text=True,
+                timeout=TIMEOUT_RENDER_SECONDS,
+                env={"QT_QPA_PLATFORM": "offscreen", **os.environ},
+            )
 
-            try:
-                subprocess.run(wav_cmd, check=True, capture_output=True, text=True)
-                logging.info(f"[PID {pid}] Áudio WAV gerado com sucesso: {wav_name}")
-            except subprocess.CalledProcessError as e:
-                logging.warning(f"[PID {pid}] Falha na geração do WAV: {e.stderr}")
+            if render_process.returncode != 0:
+                err_msg = (
+                    render_process.stderr
+                    if render_process.stderr
+                    else "Erro desconhecido"
+                )
+                if render_process.returncode == -11:
+                    err_msg = f"LMMS CRASH (SIGSEGV): {err_msg}"
+
+                raise subprocess.CalledProcessError(
+                    render_process.returncode,
+                    "lmms render",
+                    output=render_process.stdout,
+                    stderr=err_msg,
+                )
+
+            logging.info(f"[PID {pid}] Áudio WAV gerado: {wav_name}")
         else:
-            result["error"] = "Arquivo alvo MMP não encontrado para renderização."
-            logging.error(f"[PID {pid}] {result['error']}")
-            return result
+            raise FileNotFoundError(
+                "Arquivo MMP alvo não encontrado para renderização."
+            )
 
-        # 3. Parsing e Salvamento de Dados
+        # 3. Parsing e Salvamento INDIVIDUAL
         if os.path.exists(target_mmp_path):
             mmp_data = parse_mmp_file(target_mmp_path)
 
             if mmp_data:
-                # Injeta metadados de padronização
-                mmp_data["file"] = clean_slug 
+                mmp_data["file"] = clean_slug
                 mmp_data["original_title"] = original_base_name
-                
-                # Salva metadados
-                save_to_json(
-                    mmp_data, os.path.join(METADATA_FOLDER, json_name)
-                )
+
+                # Salva metadados individuais
+                save_to_json(mmp_data, os.path.join(METADATA_FOLDER, json_name))
                 save_to_yaml(mmp_data, os.path.join(DATA_FOLDER, yml_name))
 
                 result["success"] = True
                 result["data"] = mmp_data
-                
-                # Mantemos o arquivo .mmp na pasta para referência futura
             else:
                 result["error"] = "Dados vazios após parsing."
         else:
             result["error"] = "Arquivo MMP não encontrado para parsing."
 
+        try:
+            process = psutil.Process(pid)
+            ram_usage_mb = process.memory_info().rss / (1024 * 1024)
+        except:
+            ram_usage_mb = 0.0
+
+    except subprocess.TimeoutExpired:
+        result["error"] = f"Timeout ({TIMEOUT_RENDER_SECONDS}s) excedido."
+        logging.error(f"[PID {pid}] TIMEOUT em {file_name}")
+
+    except subprocess.CalledProcessError as e:
+        if isinstance(e.stderr, bytes):
+            err_msg = e.stderr.decode("utf-8", errors="ignore")
+        else:
+            err_msg = str(e.stderr) if e.stderr else str(e)
+
+        result["error"] = f"Erro LMMS (Code {e.returncode}): {err_msg}"
+        logging.error(f"[PID {pid}] Falha LMMS: {result['error']}")
+
     except Exception as e:
         result["error"] = f"Erro geral: {str(e)}"
         logging.error(f"[PID {pid}] {file_name}: {result['error']}")
 
+    result["duration_s"] = time.time() - start_time
+    result["ram_mb"] = ram_usage_mb
+
     return result
 
 
 def main_parallel():
     logger, log_file_path = setup_logger()
-    start_time = time.time()
+    start_time_global = time.time()
+
+    audit_csv_path = os.path.join(SAIDA_ANALISES, "audit_pipeline.csv")
 
     check_system_dependencies()
-
     log_system_info()
-    logging.info("=== Iniciando Pipeline Otimizado (Com Sanitização e Persistência MMP) ===")
 
-    create_folders_if_not_exist([MMPZ_FOLDER, WAV_FOLDER, METADATA_FOLDER, DATA_FOLDER])
+    # Cria pastas (incluindo a de Lotes)
+    create_folders_if_not_exist(
+        [MMPZ_FOLDER, WAV_FOLDER, METADATA_FOLDER, DATA_FOLDER, LOTES_FOLDER]
+    )
+
+    logging.info(
+        "=== Iniciando Pipeline Otimizado (Multi-Batch + Paralelo + Contador) ==="
+    )
 
     if not os.path.exists(MMP_FOLDER):
         logging.critical(f"Pasta {MMP_FOLDER} não encontrada.")
@@ -277,48 +356,132 @@ def main_parallel():
         logging.warning("Nenhum arquivo encontrado.")
         return
 
+    # === APLICAÇÃO DO LIMITE DE ARQUIVOS (ALTERAÇÃO AQUI) ===
+    if LIMIT_FILES and LIMIT_FILES > 0:
+        logging.info(
+            f"MODO LIMITADO: Processando apenas os primeiros {LIMIT_FILES} arquivos."
+        )
+        all_files_raw = all_files_raw[:LIMIT_FILES]
+
+    total_files = len(all_files_raw)
+    logging.info(f"Total de arquivos a processar: {total_files}")
+
+    # === CONFIGURAÇÃO DO CONTADOR COMPARTILHADO ===
+    # Usamos multiprocessing.Value diretamente (Shared Memory)
+    shared_counter = multiprocessing.Value("i", 0)
+
     # === PRÉ-PROCESSAMENTO DOS SLUGS ===
     tasks = []
     invalid_count = 0
-    
-    logging.info("Calculando nomes sanitizados e preparando tarefas...")
-    
+
+    logging.info("Preparando tarefas...")
+
     for file_name in all_files_raw:
         original_base = os.path.splitext(file_name)[0]
         slug = slugify(original_base)
-        
-        # Se o nome sanitizado ficou vazio (ex: "!!!.mmp" -> ""), gera nome padrão
+
         if not slug:
             invalid_count += 1
             slug = f"titulo-incorreto-{invalid_count}"
-            logging.warning(f"Nome inválido detectado: '{file_name}'. Renomeando para '{slug}'")
-            
-        tasks.append((file_name, slug))
+            logging.warning(
+                f"Nome inválido detectado: '{file_name}'. Renomeando para '{slug}'"
+            )
 
-    num_cores = multiprocessing.cpu_count()
-    logging.info(f"Processando {len(tasks)} arquivos com {num_cores} workers.")
+        # Passamos APENAS os dados estáticos. O contador vai pelo initializer.
+        tasks.append((file_name, slug, total_files))
 
-    # O pool.map envia a tupla (arquivo, slug)
-    with multiprocessing.Pool(processes=num_cores) as pool:
+    # === EXECUÇÃO PARALELA ===
+    num_cores = get_cpu_safe_count()
+    logging.info(f"Processando com {num_cores} workers (Safe Mode).")
+
+    with open(audit_csv_path, "w", newline="", encoding="utf-8") as f:
+        writer = csv.writer(f)
+        writer.writerow(
+            [
+                "Timestamp",
+                "Arquivo_Original",
+                "Slug",
+                "Tamanho_MB",
+                "Duracao_Sec",
+                "RAM_Worker_MB",
+                "Status",
+                "Erro",
+            ]
+        )
+
+    successful_data = []
+    failed_files = []
+
+    # === O POOL AGORA USA INITIALIZER ===
+    # Isso injeta o shared_counter em cada worker de forma segura antes das tarefas começarem
+    with multiprocessing.Pool(
+        processes=num_cores, initializer=init_worker, initargs=(shared_counter,)
+    ) as pool:
         results = pool.map(process_single_file, tasks)
 
-    successful_data = [r["data"] for r in results if r["success"]]
-    failed_files = [r for r in results if not r["success"]]
+    # Coleta de resultados
+    with open(audit_csv_path, "a", newline="", encoding="utf-8") as f:
+        writer = csv.writer(f)
 
+        for res in results:
+            ts = datetime.now().strftime("%H:%M:%S")
+            status = "SUCESSO" if res["success"] else "FALHA"
+
+            writer.writerow(
+                [
+                    ts,
+                    res["file"],
+                    res["slug"],
+                    f"{res['file_size_mb']:.2f}",
+                    f"{res['duration_s']:.2f}",
+                    f"{res['ram_mb']:.2f}",
+                    status,
+                    res["error"] or "",
+                ]
+            )
+
+            if res["success"]:
+                successful_data.append(res["data"])
+            else:
+                failed_files.append(res)
+
+    # === SALVAMENTO EM LOTES (BATCHING) ===
     if successful_data:
-        save_to_json(successful_data, os.path.join(METADATA_FOLDER, "all.json"))
-        save_to_yaml(successful_data, os.path.join(DATA_FOLDER, "all.yml"))
+        logging.info(f"Total processado com sucesso: {len(successful_data)}")
+        logging.info(f"Iniciando gravação em lotes (Batch Size: {BATCH_SIZE})...")
 
+        # 1. Salvar JSON Completo (Geralmente seguro)
+        logging.info("Salvando all.json completo...")
+        save_to_json(successful_data, os.path.join(METADATA_FOLDER, "all.json"))
+
+        # 2. Salvar YAML em Fatias (Para evitar memory overflow)
+        total_items = len(successful_data)
+        for i in range(0, total_items, BATCH_SIZE):
+            batch = successful_data[i : i + BATCH_SIZE]
+            batch_num = (i // BATCH_SIZE) + 1
+            batch_filename = os.path.join(
+                LOTES_FOLDER, f"projetos_lote_{batch_num:03d}.yml"
+            )
+
+            logging.info(
+                f"Gravando Lote {batch_num} ({len(batch)} itens) em {batch_filename}..."
+            )
+            save_to_yaml(batch, batch_filename)
+
+        logging.info("Gravação de lotes concluída.")
+
+    # Geração dos manifestos
     try:
         manifest_report = generate_manifests(SRC_MMPSEARCH)
     except Exception as e:
         manifest_report = {"generated": [], "failed": [str(e)]}
 
-    duration = time.time() - start_time
+    duration = time.time() - start_time_global
     logging.info("=" * 60)
     logging.info(
-        f"FIM - Tempo: {duration:.2f}s | Sucessos: {len(successful_data)} | Falhas: {len(failed_files)}"
+        f"FIM - Tempo Total: {str(timedelta(seconds=int(duration)))} | Sucessos: {len(successful_data)} | Falhas: {len(failed_files)}"
     )
+    logging.info(f"Relatório de Auditoria: {audit_csv_path}")
 
     if failed_files:
         logging.info("--- Detalhe das Falhas ---")
@@ -335,36 +498,5 @@ def main_parallel():
         pass
 
 
-def rebuild_indexes():
-    logging.info("Regerando índices globais (all.json / all.yml)...")
-
-    all_data = []
-    ignored_files = [
-        "all.json",
-        "samples-manifest.json",
-        "mmp-manifest.json",
-        "dependency_report.json",
-    ]
-
-    if os.path.exists(METADATA_FOLDER):
-        for f in os.listdir(METADATA_FOLDER):
-            if f.endswith(".json") and f not in ignored_files:
-                try:
-                    with open(
-                        os.path.join(METADATA_FOLDER, f), "r", encoding="utf-8"
-                    ) as json_file:
-                        data = json.load(json_file)
-                        all_data.append(data)
-                except Exception as e:
-                    logging.error(f"Erro ao ler {f} para índice global: {e}")
-
-    if all_data:
-        save_to_json(all_data, os.path.join(METADATA_FOLDER, "all.json"))
-        save_to_yaml(all_data, os.path.join(DATA_FOLDER, "all.yml"))
-        logging.info("Índices globais atualizados com sucesso.")
-
-    return len(all_data)
-
-
 if __name__ == "__main__":
-    main_parallel()
+    main_parallel()

scripts/handler/utils.py

@@ -7,6 +7,7 @@ BASE_PATH_MMPSEARCH = "/var/www/html/trens/mmpSearch"
 BASE_DATA = "/nethome/jotachina/projetos/mmpSearch"
 SRC_MMPSEARCH = "/var/www/html/trens/src_mmpSearch"
 
+SAIDA_ANALISES = os.path.join(BASE_PATH, SRC_MMPSEARCH, "saida_analises")
 MMP_FOLDER = os.path.join(BASE_PATH, "src_mmpSearch", "mmp")
 MMPZ_FOLDER = os.path.join(BASE_PATH, "src_mmpSearch", "mmpz")
 WAV_FOLDER = os.path.join(BASE_PATH, "src_mmpSearch", "wav")