import os import subprocess import logging from typing import Optional from fastapi import FastAPI, HTTPException import gradio as gr # Configuración de logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # Directorios y rutas UPLOAD_DIR = os.path.join(os.getcwd(), "uploads") FFMPEG_PATH = "./ffmpeg" # Crear directorios os.makedirs(UPLOAD_DIR, exist_ok=True) # Definir formatos de video y audio soportados VIDEO_FORMATS = ['.mp4', '.avi', '.mov', '.mkv', '.webm', '.flv', '.wmv'] AUDIO_FORMATS = ['.mp3', '.wav', '.aac', '.flac', '.ogg', '.m4a', '.wma'] def detect_media_type(file_path: str) -> str: """Detectar si el archivo es de audio o video.""" ext = os.path.splitext(file_path)[1].lower() if ext in VIDEO_FORMATS: return 'video' elif ext in AUDIO_FORMATS: return 'audio' else: return 'unsupported' def sanitize_filename(filename: str) -> str: """Limpiar y validar nombre de archivo para prevenir riesgos de seguridad.""" return ''.join(c for c in filename if c.isalnum() or c in ('.', '_', '-')).rstrip() def ensure_unique_filename(directory: str, filename: str) -> str: """Generar un nombre de archivo único para evitar sobreescrituras.""" base, ext = os.path.splitext(filename) counter = 1 new_filename = filename while os.path.exists(os.path.join(directory, new_filename)): new_filename = f"{base}_{counter}{ext}" counter += 1 return new_filename def make_ffmpeg_executable(ffmpeg_path: str): """Asegurar permisos correctos para FFmpeg.""" try: subprocess.run(["chmod", "+x", ffmpeg_path], check=True) logger.info(f"Permisos de FFmpeg configurados para {ffmpeg_path}") except subprocess.CalledProcessError as e: logger.error(f"Error al configurar permisos de FFmpeg: {e}") raise def convert_media(input_file: str, output_dir: str) -> str: """Convertir archivos multimedia con configuraciones optimizadas.""" try: media_type = detect_media_type(input_file) base_name = os.path.basename(input_file) # Elegir extensión de salida según el tipo de medio output_extension = 'mp4' if media_type == 'video' else 'm4a' output_filename = ensure_unique_filename( output_dir, f"{os.path.splitext(base_name)[0]}_converted.{output_extension}" ) output_file = os.path.join(output_dir, output_filename) if media_type == 'video': # Configuraciones de conversión de video ffmpeg_command = [ FFMPEG_PATH, '-i', input_file, '-vf', 'fps=24', # Cambiar la tasa de fotogramas '-c:v', 'libx264', # Codificador de video '-c:a', 'libfdk_aac', # Codificador de audio '-profile:a', 'aac_he_v2', # Perfil de audio '-crf', '28', # Tasa de compresión '-b:a', '32k', # Tasa de bits de audio '-preset', 'slow', # Preajuste de codificación '-movflags', '+faststart', # Web optimization output_file ] elif media_type == 'audio': # Configuraciones de conversión de audio ffmpeg_command = [ FFMPEG_PATH, '-i', input_file, '-vn', # Ignorar video '-c:a', 'libfdk_aac', # Codificador AAC para M4A '-profile:a', 'aac_he_v2', '-b:a', '32k', # Calidad de audio alta '-ar', '44100', # Frecuencia de muestreo output_file ] else: raise ValueError("Formato no soportado") # Ejecutar conversión result = subprocess.run( ffmpeg_command, check=True, capture_output=True, text=True ) logger.info(f"Medio convertido exitosamente: {output_file}") return output_file except subprocess.CalledProcessError as e: logger.error(f"Error de conversión de FFmpeg: {e.stderr}") raise HTTPException(status_code=500, detail=f"Fallo en la conversión de medio: {e.stderr}") except Exception as e: logger.error(f"Error inesperado durante la conversión: {e}") raise HTTPException(status_code=500, detail="Error inesperado durante la conversión") def process_media(file_path: str) -> str: """Procesar medio completo.""" return convert_media(file_path, UPLOAD_DIR) def gradio_interface(media: Optional[str]) -> Optional[str]: """Interfaz de Gradio para conversión de medios.""" if not media: raise gr.Error("No se ha subido ningún medio. Por favor, sube un archivo.") try: converted_media = process_media(media) return converted_media except Exception as e: raise gr.Error(f"Conversión fallida: {str(e)}") # Asegurar permisos de FFmpeg make_ffmpeg_executable(FFMPEG_PATH) # Crear aplicación FastAPI app = FastAPI() # Configurar interfaz de Gradio iface = gr.Interface( fn=gradio_interface, inputs=gr.File( label="Subir Archivo", type="filepath" ), outputs=gr.File( label="Descargar Archivo Convertido", type="filepath" ), title="🎥🎵 Convertidor Universal", description="Convierte videos a MP4 y audios a M4A con configuraciones optimizadas", theme="huggingface" ) # Lanzar interfaz de Gradio if __name__ == "__main__": iface.launch(share=True)