hf_spaces_config.py

"""
Configuración para espacios de Hugging Face duplicados
Este archivo maneja la configuración y comunicación con los espacios de HF
"""

import requests
import json
import time
from typing import Dict, Any, Optional
import os

class HFSpacesManager:
    """Gestor de espacios de Hugging Face"""
    
    def __init__(self):
        self.spaces_config = {
            "NTIA-VEO3-Free": {
                "url": "https://huggingface.co/spaces/Ntdeseb/NTIA-VEO3-Free",
                "api_url": "https://ntdeseb-ntia-veo3-free.hf.space",
                "description": "Generación de video con VEO3-Free",
                "type": "text-to-video",
                "enabled": True,
                "timeout": 120,
                "retries": 3
            },
            "NTIA-animated": {
                "url": "https://huggingface.co/spaces/Ntdeseb/NTIA-animated", 
                "api_url": "https://ntdeseb-ntia-animated.hf.space",
                "description": "Animación con wan2-1-fast",
                "type": "animation",
                "enabled": True,
                "timeout": 90,
                "retries": 3
            },
            "NTIA-LivePortrait": {
                "url": "https://huggingface.co/spaces/Ntdeseb/NTIA-LivePortrait",
                "api_url": "https://ntdeseb-ntia-liveportrait.hf.space", 
                "description": "Retratos animados con LivePortrait",
                "type": "live-portrait",
                "enabled": False,  # Deshabilitado por error de runtime
                "timeout": 60,
                "retries": 2
            },
            "NTIA-stable-video-diffusion": {
                "url": "https://huggingface.co/spaces/Ntdeseb/NTIA-stable-video-diffusion",
                "api_url": "https://ntdeseb-ntia-stable-video-diffusion.hf.space",
                "description": "Video diffusion estable",
                "type": "stable-video-diffusion", 
                "enabled": True,
                "timeout": 150,
                "retries": 3
            }
        }
        
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'NTIA-Assistant/1.0',
            'Accept': 'application/json'
        })
    
    def check_space_status(self, space_name: str) -> Dict[str, Any]:
        """Verificar el estado de un espacio específico"""
        if space_name not in self.spaces_config:
            return {"status": "error", "message": f"Espacio {space_name} no encontrado"}
        
        config = self.spaces_config[space_name]
        
        try:
            # Verificar si el espacio está habilitado
            if not config["enabled"]:
                return {
                    "status": "disabled",
                    "message": f"Espacio {space_name} está deshabilitado",
                    "reason": "Runtime error o mantenimiento"
                }
            
            # Verificar conectividad
            response = self.session.get(
                config["api_url"],
                timeout=10
            )
            
            if response.status_code == 200:
                return {
                    "status": "online",
                    "message": f"Espacio {space_name} está en línea",
                    "url": config["url"],
                    "api_url": config["api_url"]
                }
            else:
                return {
                    "status": "error",
                    "message": f"Error HTTP {response.status_code}",
                    "url": config["url"]
                }
                
        except requests.exceptions.Timeout:
            return {
                "status": "timeout",
                "message": f"Timeout al conectar con {space_name}",
                "url": config["url"]
            }
        except requests.exceptions.ConnectionError:
            return {
                "status": "offline",
                "message": f"Espacio {space_name} no está disponible",
                "url": config["url"]
            }
        except Exception as e:
            return {
                "status": "error",
                "message": f"Error desconocido: {str(e)}",
                "url": config["url"]
            }
    
    def call_space_api(self, space_name: str, prompt: str, **kwargs) -> Dict[str, Any]:
        """Llamar a la API de un espacio específico"""
        if space_name not in self.spaces_config:
            raise ValueError(f"Espacio {space_name} no encontrado")
        
        config = self.spaces_config[space_name]
        
        if not config["enabled"]:
            raise ValueError(f"Espacio {space_name} está deshabilitado")
        
        # Preparar payload según el tipo de espacio
        payload = self._prepare_payload(space_name, prompt, **kwargs)
        
        # Intentar llamada con reintentos
        for attempt in range(config["retries"]):
            try:
                print(f"🌐 Llamando a {space_name} (intento {attempt + 1}/{config['retries']})")
                print(f"📤 Payload: {payload}")
                
                response = self.session.post(
                    f"{config['api_url']}/run/predict",
                    json={"data": [payload]},
                    timeout=config["timeout"]
                )
                
                if response.status_code == 200:
                    result = response.json()
                    print(f"✅ Respuesta exitosa de {space_name}")
                    return self._process_response(space_name, result)
                else:
                    print(f"❌ Error HTTP {response.status_code} en {space_name}")
                    print(f"📄 Respuesta: {response.text}")
                    
                    if attempt < config["retries"] - 1:
                        wait_time = (attempt + 1) * 5
                        print(f"⏳ Reintentando en {wait_time} segundos...")
                        time.sleep(wait_time)
                    else:
                        raise Exception(f"Error HTTP {response.status_code} después de {config['retries']} intentos")
                        
            except requests.exceptions.Timeout:
                print(f"⏰ Timeout en {space_name} (intento {attempt + 1})")
                if attempt < config["retries"] - 1:
                    wait_time = (attempt + 1) * 10
                    print(f"⏳ Reintentando en {wait_time} segundos...")
                    time.sleep(wait_time)
                else:
                    raise Exception(f"Timeout después de {config['retries']} intentos")
                    
            except Exception as e:
                print(f"❌ Error en {space_name}: {e}")
                if attempt < config["retries"] - 1:
                    wait_time = (attempt + 1) * 5
                    print(f"⏳ Reintentando en {wait_time} segundos...")
                    time.sleep(wait_time)
                else:
                    raise
    
    def _prepare_payload(self, space_name: str, prompt: str, **kwargs) -> Dict[str, Any]:
        """Preparar el payload específico para cada tipo de espacio"""
        config = self.spaces_config[space_name]
        
        if config["type"] == "text-to-video":
            return {
                "prompt": prompt,
                "num_frames": kwargs.get("num_frames", 16),
                "num_inference_steps": kwargs.get("num_inference_steps", 20),
                "height": kwargs.get("height", 256),
                "width": kwargs.get("width", 256)
            }
        elif config["type"] == "animation":
            return {
                "text": prompt,
                "steps": kwargs.get("num_inference_steps", 20),
                "guidance_scale": kwargs.get("guidance_scale", 7.5)
            }
        elif config["type"] == "stable-video-diffusion":
            return {
                "prompt": prompt,
                "video_length": kwargs.get("num_frames", 16),
                "num_inference_steps": kwargs.get("num_inference_steps", 20),
                "guidance_scale": kwargs.get("guidance_scale", 7.5)
            }
        elif config["type"] == "live-portrait":
            return {
                "prompt": prompt,
                "source_image": kwargs.get("source_image", None),
                "driving_video": kwargs.get("driving_video", None)
            }
        else:
            return {"prompt": prompt}
    
    def _process_response(self, space_name: str, response: Dict[str, Any]) -> Dict[str, Any]:
        """Procesar la respuesta del espacio"""
        config = self.spaces_config[space_name]
        
        try:
            if "data" in response and len(response["data"]) > 0:
                data = response["data"][0]
                
                # Procesar según el tipo de espacio
                if config["type"] in ["text-to-video", "animation", "stable-video-diffusion"]:
                    return {
                        "success": True,
                        "space_name": space_name,
                        "type": config["type"],
                        "video_data": data,
                        "format": self._detect_video_format(data)
                    }
                elif config["type"] == "live-portrait":
                    return {
                        "success": True,
                        "space_name": space_name,
                        "type": config["type"],
                        "portrait_data": data
                    }
                else:
                    return {
                        "success": True,
                        "space_name": space_name,
                        "type": config["type"],
                        "data": data
                    }
            else:
                return {
                    "success": False,
                    "space_name": space_name,
                    "error": "Respuesta inválida del espacio",
                    "response": response
                }
                
        except Exception as e:
            return {
                "success": False,
                "space_name": space_name,
                "error": f"Error procesando respuesta: {str(e)}",
                "response": response
            }
    
    def _detect_video_format(self, data: Any) -> str:
        """Detectar el formato del video en la respuesta"""
        if isinstance(data, str):
            if data.startswith("data:video"):
                return "base64_video"
            elif data.endswith((".mp4", ".avi", ".mov")):
                return "file_path"
        elif isinstance(data, list):
            return "frames_list"
        elif isinstance(data, dict) and "video" in data:
            return "video_object"
        else:
            return "unknown"
    
    def get_all_spaces_status(self) -> Dict[str, Dict[str, Any]]:
        """Obtener el estado de todos los espacios"""
        status = {}
        for space_name in self.spaces_config.keys():
            status[space_name] = self.check_space_status(space_name)
        return status
    
    def get_enabled_spaces(self) -> Dict[str, Dict[str, Any]]:
        """Obtener solo los espacios habilitados"""
        return {
            name: config for name, config in self.spaces_config.items() 
            if config["enabled"]
        }
    
    def enable_space(self, space_name: str) -> bool:
        """Habilitar un espacio"""
        if space_name in self.spaces_config:
            self.spaces_config[space_name]["enabled"] = True
            return True
        return False
    
    def disable_space(self, space_name: str) -> bool:
        """Deshabilitar un espacio"""
        if space_name in self.spaces_config:
            self.spaces_config[space_name]["enabled"] = False
            return True
        return False

# Instancia global del gestor
hf_spaces_manager = HFSpacesManager()

def get_hf_spaces_manager() -> HFSpacesManager:
    """Obtener la instancia global del gestor de espacios"""
    return hf_spaces_manager

# Funciones de utilidad
def test_space_connection(space_name: str) -> Dict[str, Any]:
    """Probar la conexión con un espacio específico"""
    manager = get_hf_spaces_manager()
    return manager.check_space_status(space_name)

def list_available_spaces() -> Dict[str, Dict[str, Any]]:
    """Listar todos los espacios disponibles"""
    manager = get_hf_spaces_manager()
    return manager.get_all_spaces_status()

def generate_video_with_space(space_name: str, prompt: str, **kwargs) -> Dict[str, Any]:
    """Generar video usando un espacio específico"""
    manager = get_hf_spaces_manager()
    return manager.call_space_api(space_name, prompt, **kwargs)

if __name__ == "__main__":
    # Pruebas básicas
    print("🧪 Probando configuración de espacios de HF...")
    
    manager = get_hf_spaces_manager()
    
    # Verificar estado de todos los espacios
    print("\n📊 Estado de todos los espacios:")
    for space_name, status in manager.get_all_spaces_status().items():
        print(f"  {space_name}: {status['status']} - {status['message']}")
    
    # Listar espacios habilitados
    print("\n✅ Espacios habilitados:")
    enabled_spaces = manager.get_enabled_spaces()
    for space_name in enabled_spaces.keys():
        print(f"  • {space_name}")
    
    print(f"\n🎯 Total de espacios: {len(manager.spaces_config)}")
    print(f"✅ Habilitados: {len(enabled_spaces)}")
    print(f"❌ Deshabilitados: {len(manager.spaces_config) - len(enabled_spaces)}")