#!/usr/bin/env python3 """ Servidor de transcripción WhisperX en proceso separado Evita completamente el problema de deadlock en Gradio ULTRA LIGERA: Optimizado para CPU con int8 quantization - Usa CTranslate2 (faster-whisper) backend con int8 para máximo rendimiento en CPU - Reduce significativamente el uso de memoria RAM - Acelera la inferencia en hardware limitado - Mantiene calidad aceptable para casos de uso generales """ import os import sys import json import time import pickle import tempfile import gc from pathlib import Path # --- Integrar FFmpeg local (si existe) antes de cualquier carga de audio --- def _inject_local_ffmpeg(): """Add bundled ffmpeg bin folder to PATH so whisperx.load_audio finds ffmpeg/ffprobe. Safe no-op if already present.""" try: base_dir = Path(__file__).parent # Estructura esperada: ffmpeg/ffmpeg-8.0-essentials_build/bin/ffmpeg.exe candidates = [ base_dir / 'ffmpeg' / 'ffmpeg-8.0-essentials_build' / 'bin', base_dir / 'ffmpeg' / 'bin', ] for c in candidates: if c.is_dir(): bin_path = str(c) if bin_path not in os.environ.get('PATH',''): os.environ['PATH'] = bin_path + os.pathsep + os.environ.get('PATH','') print(f"🔧 FFmpeg agregado al PATH: {bin_path}") # Comprobar ejecutables for exe in ('ffmpeg.exe','ffprobe.exe'): if (c / exe).is_file(): print(f"✅ Detectado {exe} en {c}") return print("⚠️ No se encontró carpeta local de FFmpeg; se usará el PATH del sistema") except Exception as e: print(f"⚠️ Error inyectando FFmpeg local: {e}") _inject_local_ffmpeg() def transcribe_audio_subprocess(audio_path, model_name="large-v3", force_cpu: bool = True): """Transcribir audio usando WhisperX optimizado para CPU únicamente. ULTRA LIGERA: Solo CPU con int8 quantization para máximo rendimiento. force_cpu: Ignorado - siempre usa CPU (versión ultra ligera optimizada). """ try: import torch try: import whisperx except Exception as e_imp: # Diagnóstico extendido para errores tipo "_ufuncs" (SciPy faltante) import platform py = sys.version.replace("\n", " ") arch = platform.architecture() np_ok = True np_ver = None sp_ok = True sp_err = None try: import numpy as _np np_ver = getattr(_np, '__version__', 'unknown') except Exception: np_ok = False try: # SciPy opcional pero muchos deps (librosa) lo requieren import scipy _ = getattr(scipy, '__version__', 'unknown') # Forzar import de special._ufuncs para detectar DLLs from scipy import special as _spc # noqa: F401 except Exception as _se: sp_ok = False sp_err = str(_se) msg = { "success": False, "error": f"Fallo importando whisperx: {e_imp}", "python": py, "arch": arch, "numpy_present": np_ok, "numpy_version": np_ver, "scipy_present": sp_ok, "scipy_error": sp_err, "hint": "Instala numpy y scipy compatibles en python_embebido. Ej: numpy==1.26.4, scipy==1.11.4" } try: with open('transcription_result.json','w', encoding='utf-8') as f: json.dump(msg, f, ensure_ascii=False, indent=2) except Exception: pass print("❌ Error importando whisperx - probable dependencia SciPy faltante") print(f"NumPy ok={np_ok} ver={np_ver}; SciPy ok={sp_ok} err={sp_err}") return False # ULTRA LIGERA: Solo CPU - versión optimizada print("💻 ULTRA LIGERA: Ejecutando siempre en CPU con int8 quantization") # Configuración fija para CPU únicamente device = "cpu" compute_type = "int8" # Optimización máxima para CPU print(f"⚙️ Device: {device} compute_type={compute_type} (CPU optimizado)") # Cargar modelo con configuración fija CPU + int8 print(f"🚀 Cargando modelo {model_name} en CPU con int8...") print("💡 ULTRA LIGERA: int8 quantization puede ser más lento pero usa menos memoria") model = whisperx.load_model(model_name, device=device, compute_type=compute_type, download_root=None) if not os.path.isfile(audio_path): raise FileNotFoundError(f"Audio no encontrado en ruta recibida por subprocess: {audio_path}") print("📁 Cargando archivo de audio...") audio = whisperx.load_audio(audio_path) audio_duration = len(audio)/16000 print(f"⏱️ Duración audio: {audio_duration:.1f}s") # ULTRA LIGERA: Batch size adaptativo para CPU + int8 if audio_duration > 3600: # Más de 1 hora batch_size = 2 # Muy conservador elif audio_duration > 1800: # Más de 30 minutos batch_size = 3 else: batch_size = 4 # Normal print(f"🎯 Batch size CPU optimizado: {batch_size} (duración: {audio_duration:.1f}s)") print("📝 Iniciando transcribe...") print(f"⏱️ Tiempo estimado CPU: ~{audio_duration/60:.1f} min para {audio_duration/60:.1f} min de audio") lang_hint = os.environ.get('WHISPERX_LANG_HINT', '').strip() kwargs = {"batch_size": batch_size} if lang_hint: print(f"🌐 Pista de idioma recibida: {lang_hint}") # whisperx.transcribe acepta language como hint kwargs["language"] = lang_hint # Iniciar medición de tiempo transcribe_start = time.time() result = model.transcribe(audio, **kwargs) transcribe_time = time.time() - transcribe_start print(f"✅ Transcripción completada en {transcribe_time:.1f}s (ratio: {transcribe_time/audio_duration:.2f}x)") # Limpieza CPU optimizada del model; del audio gc.collect() # Solo garbage collection en CPU result_file = "transcription_result.json" out = { "success": True, "result": result, "device": device, "compute_type": compute_type, "audio_path": str(audio_path) } with open(result_file, 'w', encoding='utf-8') as f: json.dump(out, f, ensure_ascii=False, indent=2) print("✅ Transcripción completada OK") return True except Exception as e: try: # ULTRA LIGERA: Solo CPU - sin diagnósticos GPU import torch gc.collect() # Solo garbage collection CPU except Exception: pass err_out = {"success": False, "error": str(e), "audio_path": str(audio_path)} try: with open('transcription_result.json','w', encoding='utf-8') as f: json.dump(err_out, f, ensure_ascii=False, indent=2) except Exception: pass print(f"❌ Error en transcripción: {e}") return False if __name__ == "__main__": try: if len(sys.argv) < 2: print("❌ Error: No se proporcionó ruta de audio") print("Uso: python transcriber_subprocess.py [model_name] [--force-cpu]") sys.exit(1) audio_path = None model_name = "large-v2" # ULTRA LIGERA: modelo fijo force_cpu = True # ULTRA LIGERA: siempre CPU for arg in sys.argv[1:]: if arg == "--force-cpu": pass # Ignorado - siempre CPU elif audio_path is None: audio_path = arg elif model_name == "large-v2": model_name = arg if audio_path is None: raise ValueError("Ruta de audio no detectada en argumentos") # ULTRA LIGERA: Solo CPU - sin configuraciones GPU print(f"🚀 Iniciando transcripción CPU optimizada: {audio_path}") print(f"🧠 Modelo: {model_name} (int8 quantization)") print("💻 Modo: CPU únicamente (ultra ligera)") success = transcribe_audio_subprocess(audio_path, model_name, force_cpu=force_cpu) if success: print("✅ Transcripción subprocess completada exitosamente") sys.exit(0) else: print("❌ Transcripción subprocess falló") sys.exit(1) except Exception as e: print(f"❌ Error crítico en transcriber subprocess: {e}") import traceback print(f"❌ Traceback: {traceback.format_exc()}") error_data = {"success": False, "error": f"Error crítico: {str(e)}", "traceback": traceback.format_exc()} try: with open("transcription_result.json", "w", encoding='utf-8') as f: json.dump(error_data, f, ensure_ascii=False, indent=2) except Exception as save_error: print(f"❌ No se pudo guardar error: {save_error}") sys.exit(1)