app.py

import gradio as gr
import asyncio
import json
from typing import List, Dict, Any, Optional, Tuple, Generator
from datetime import datetime
from enum import Enum
import hashlib
import logging
from dataclasses import dataclass
from typing import Literal

# ==================== CONFIGURACIÓN DE LOGGING ESTRUCTURADO ====================
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('chat_interface.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# ==================== TIPADO Y ESTRUCTURAS DE DATOS ====================
@dataclass
class ChatMessage:
    """Estructura type-safe para mensajes de chat"""
    role: Literal["user", "assistant", "system"]
    content: str
    timestamp: str
    message_id: str
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "role": self.role,
            "content": self.content,
            "timestamp": self.timestamp,
            "id": self.message_id
        }

class IntentType(Enum):
    """Tipos de intención detectados"""
    QUERY = "consulta"
    CODE_GENERATION = "generacion_codigo"
    SYSTEM_DIAGNOSTIC = "diagnostico_sistema"
    OPTIMIZATION = "optimizacion"
    UNKNOWN = "desconocido"

@dataclass
class IntentAnalysis:
    """Resultado del análisis de intención"""
    intent: IntentType
    confidence: float
    entities: Dict[str, Any]
    requires_streaming: bool

# ==================== SERVICIO DE ANÁLISIS DE INTENCIÓN ====================
class IntentAnalyzer:
    """Servicio para análisis de intención con validación robusta"""
    
    @staticmethod
    def analyze_intent_detailed(
        message: str, 
        history: List[Dict[str, Any]]
    ) -> Tuple[IntentType, Dict[str, Any]]:
        """
        Analiza la intención del mensaje con validación de inputs.
        
        Args:
            message: Mensaje del usuario (validado y sanitizado)
            history: Historial de conversación en formato Gradio 4.x
            
        Returns:
            Tuple[IntentType, Dict]: Tipo de intención y metadatos
            
        Raises:
            ValueError: Si el mensaje está vacío o mal formado
        """
        # Validación de inputs
        if not isinstance(message, str):
            raise TypeError(f"Mensaje debe ser string, recibido: {type(message)}")
        
        if not message.strip():
            raise ValueError("Mensaje no puede estar vacío")
        
        # Sanitización básica
        sanitized_message = message.strip()[:2000]  # Limitación de longitud
        
        # Patrones de detección de intención
        message_lower = sanitized_message.lower()
        
        # Diccionario de palabras clave para cada intención
        intent_patterns = {
            IntentType.CODE_GENERATION: [
                "código", "programa", "función", "clase", "implementa",
                "desarrolla", "escribe un", "crea un", "genera código",
                "python", "javascript", "java", "typescript", "html", "css"
            ],
            IntentType.SYSTEM_DIAGNOSTIC: [
                "error", "falla", "problema", "debug", "depurar",
                "solucionar", "arreglar", "no funciona", "bug",
                "excepción", "traceback", "stack trace"
            ],
            IntentType.OPTIMIZATION: [
                "optimizar", "mejorar", "rendimiento", "eficiencia",
                "velocidad", "lento", "bottleneck", "cuello de botella",
                "escalar", "scaling", "performance", "memoria", "cpu"
            ],
            IntentType.QUERY: [
                "qué", "cómo", "cuándo", "dónde", "por qué",
                "explica", "describe", "información", "ayuda",
                "tutorial", "guía", "ejemplo"
            ]
        }
        
        # Análisis de frecuencia y confianza
        intent_scores = {intent: 0.0 for intent in IntentType}
        
        for intent, patterns in intent_patterns.items():
            for pattern in patterns:
                if pattern in message_lower:
                    intent_scores[intent] += 1.0
        
        # Ponderación por longitud de palabra clave
        word_count = len(sanitized_message.split())
        for intent in intent_scores:
            if intent_scores[intent] > 0:
                intent_scores[intent] = min(1.0, intent_scores[intent] / (word_count * 0.5))
        
        # Determinación de intención principal
        if max(intent_scores.values()) > 0:
            main_intent = max(intent_scores, key=intent_scores.get)
            confidence = intent_scores[main_intent]
        else:
            main_intent = IntentType.UNKNOWN
            confidence = 0.0
        
        # Metadatos adicionales
        metadata = {
            "confidence": round(confidence, 2),
            "message_length": len(sanitized_message),
            "word_count": word_count,
            "has_code_snippet": "```" in sanitized_message,
            "timestamp": datetime.now().isoformat(),
            "intent_scores": {k.name: v for k, v in intent_scores.items()}
        }
        
        logger.info(f"Intención detectada: {main_intent.value} (confianza: {confidence:.2f})")
        
        return main_intent, metadata

# ==================== SERVICIO DE GENERACIÓN DE RESPUESTAS ====================
class ResponseGenerator:
    """Servicio para generación de respuestas con streaming"""
    
    def __init__(self):
        self.intent_analyzer = IntentAnalyzer()
        self.response_templates = self._load_templates()
    
    def _load_templates(self) -> Dict[IntentType, str]:
        """Carga plantillas de respuesta según intención"""
        return {
            IntentType.CODE_GENERATION: """
            **🔧 Generando solución de código...**
            
            He analizado tu solicitud y procedo a generar una implementación robusta y lista para producción.
            
            **Consideraciones técnicas aplicadas:**
            • Type hints completos para mypy
            • Logging estructurado con niveles apropiados
            • Manejo de errores con excepciones específicas
            • Principios SOLID y Clean Code
            • Validación de inputs y sanitización
            
            **Código generado:**
            ```python
            # Implementación específica basada en tu solicitud
            """
            ,
            IntentType.SYSTEM_DIAGNOSTIC: """
            **🔍 Analizando diagnóstico del sistema...**
            
            Procedo a analizar el problema reportado con metodología sistemática:
            
            1. **Identificación del root cause**
            2. **Análisis de logs y métricas**
            3. **Propuesta de solución escalable**
            4. **Prevención de recurrencia**
            
            """
            ,
            IntentType.OPTIMIZATION: """
            **⚡ Iniciando proceso de optimización...**
            
            Optimización aplicada con enfoque en:
            • Complejidad algorítmica (Big O)
            • Uso eficiente de memoria
            • Paralelización y concurrencia
            • Caching estratégico
            • Minimización de I/O
            
            """
            ,
            IntentType.QUERY: """
            **📚 Proporcionando respuesta detallada...**
            
            Basado en tu consulta, te proporciono información completa y verificada:
            
            """
            ,
            IntentType.UNKNOWN: """
            **🤖 Procesando tu solicitud...**
            
            He recibido tu mensaje. Para brindarte la mejor asistencia:
            
            • ¿Podrías especificar si necesitas generación de código?
            • ¿O prefieres una explicación técnica detallada?
            • ¿Estás enfrentando un problema específico del sistema?
            
            Estoy listo para ayudarte en cualquiera de estas áreas.
            """
        }
    
    async def generate_streaming_response(
        self,
        message: str,
        history: List[Dict[str, Any]]
    ) -> Generator[Tuple[List[Dict[str, Any]], str], None, None]:
        """
        Genera respuesta con streaming en tiempo real.
        
        Args:
            message: Mensaje del usuario
            history: Historial de conversación
            
        Yields:
            Tupla[List[Dict], str]: Historial actualizado y texto de modo
        """
        try:
            # 1. Análisis de intención
            intent, metadata = self.intent_analyzer.analyze_intent_detailed(message, history)
            mode_text = f"**Modo activo**: {intent.value.replace('_', ' ').title()}"
            
            # 2. Crear historial temporal para streaming
            temp_history = history.copy()
            
            # 3. Añadir mensaje del usuario al historial
            user_message = {
                "role": "user",
                "content": message,
                "timestamp": datetime.now().isoformat()
            }
            temp_history.append(user_message)
            
            # 4. Añadir placeholder de asistente
            assistant_placeholder = {
                "role": "assistant",
                "content": "",
                "timestamp": datetime.now().isoformat()
            }
            temp_history.append(assistant_placeholder)
            
            yield temp_history, mode_text
            
            # 5. Generar respuesta con streaming simulado
            template = self.response_templates.get(intent, self.response_templates[IntentType.UNKNOWN])
            
            # Streaming chunk por chunk
            chunks = self._split_into_chunks(template)
            accumulated_response = ""
            
            for chunk in chunks:
                await asyncio.sleep(0.05)  # Simular procesamiento
                accumulated_response += chunk
                
                # Actualizar contenido del asistente en el historial
                temp_history[-1]["content"] = accumulated_response
                yield temp_history, mode_text
            
            # 6. Añadir sección específica basada en el mensaje
            specific_response = f"\n\n**📝 Detalles específicos para tu solicitud:**\n\n"
            specific_response += f"• **Mensaje analizado:** `{message[:100]}{'...' if len(message) > 100 else ''}`\n"
            specific_response += f"• **Confianza de intención:** {metadata['confidence'] * 100:.1f}%\n"
            specific_response += f"• **Longitud del mensaje:** {metadata['message_length']} caracteres\n"
            
            chunks_specific = self._split_into_chunks(specific_response)
            for chunk in chunks_specific:
                await asyncio.sleep(0.03)
                accumulated_response += chunk
                temp_history[-1]["content"] = accumulated_response
                yield temp_history, mode_text
            
            # 7. Añadir marca de finalización
            completion_marker = "\n\n---\n*Respuesta generada con SISTEMA_PROTOCOLO_POTENCIACION_ACTIVO v2.0*"
            temp_history[-1]["content"] = accumulated_response + completion_marker
            
            yield temp_history, mode_text
            
        except Exception as e:
            logger.error(f"Error en generación de respuesta: {str(e)}", exc_info=True)
            
            # Respuesta de error controlada
            error_history = history.copy()
            error_history.append({
                "role": "user",
                "content": message,
                "timestamp": datetime.now().isoformat()
            })
            error_history.append({
                "role": "assistant",
                "content": f"**⚠️ Error en el procesamiento:**\n\n{str(e)}\n\nPor favor, intenta nuevamente o reformula tu solicitud.",
                "timestamp": datetime.now().isoformat()
            })
            
            yield error_history, "**Modo activo**: Manejo de errores"
    
    def _split_into_chunks(self, text: str, chunk_size: int = 20) -> List[str]:
        """Divide texto en chunks para streaming"""
        words = text.split()
        chunks = []
        current_chunk = []
        current_length = 0
        
        for word in words:
            if current_length + len(word) + 1 > chunk_size and current_chunk:
                chunks.append(" ".join(current_chunk) + " ")
                current_chunk = [word]
                current_length = len(word)
            else:
                current_chunk.append(word)
                current_length += len(word) + 1
        
        if current_chunk:
            chunks.append(" ".join(current_chunk))
        
        return chunks

# ==================== INTERFAZ GRADIO 4.x COMPATIBLE ====================
def create_advanced_interface():
    """
    Crea interfaz avanzada de chat compatible con Gradio 4.x.
    Implementa formato correcto de mensajes: {"role": "rol", "content": "texto"}
    """
    
    # Instanciar servicios
    intent_analyzer = IntentAnalyzer()
    response_generator = ResponseGenerator()
    
    # Cache para historiales de sesión
    session_histories = {}
    
    def get_session_history(session_id: str) -> List[Dict[str, Any]]:
        """Obtiene o crea historial para sesión"""
        if session_id not in session_histories:
            # Mensaje inicial del sistema
            session_histories[session_id] = [{
                "role": "system",
                "content": "Eres un asistente especializado en ingeniería de software, DevOps y arquitectura de sistemas distribuidos. Proporcionas soluciones técnicas precisas y listas para producción.",
                "timestamp": datetime.now().isoformat()
            }]
        return session_histories[session_id]
    
    def generate_session_id() -> str:
        """Genera ID único de sesión"""
        timestamp = datetime.now().isoformat()
        hash_obj = hashlib.md5(timestamp.encode())
        return hash_obj.hexdigest()[:12]
    
    # ==================== FUNCIONES DE INTERACCIÓN ====================
    def process_message(
        message: str, 
        history: List[Dict[str, Any]], 
        session_id: str
    ) -> Tuple[str, List[Dict[str, Any]], str]:
        """
        Procesa mensaje del usuario - FORMATO GRADIO 4.x CORRECTO
        
        Args:
            message: Mensaje del usuario
            history: Historial en formato {"role": "rol", "content": "texto"}
            session_id: ID de sesión para persistencia
            
        Returns:
            Tuple[str, List[Dict], str]: Mensaje vacío, historial actualizado, texto de modo
        """
        # Validación robusta
        if not message or not isinstance(message, str):
            return "", history, "**Modo**: Error - Mensaje inválido"
        
        sanitized_message = message.strip()
        if not sanitized_message:
            return "", history, "**Modo**: Esperando entrada válida..."
        
        # Obtener historial de sesión
        session_history = get_session_history(session_id)
        
        # Crear mensaje de usuario en formato Gradio 4.x
        user_message = {
            "role": "user",
            "content": sanitized_message,
            "timestamp": datetime.now().isoformat()
        }
        
        # Actualizar historiales
        new_history = history + [user_message] if history else [user_message]
        session_history.append(user_message)
        
        try:
            # Análisis de intención
            intent, metadata = intent_analyzer.analyze_intent_detailed(
                sanitized_message, 
                new_history
            )
            
            mode_text = (
                f"**Modo activo**: {intent.value.replace('_', ' ').title()}\n"
                f"**Confianza**: {metadata['confidence'] * 100:.1f}%\n"
                f"**Tokens**: {metadata['word_count']}"
            )
            
        except Exception as e:
            logger.warning(f"Error en análisis de intención: {e}")
            mode_text = "**Modo activo**: Procesamiento directo"
        
        # Limpiar input
        return "", new_history, mode_text
    
    async def generate_response_stream(
        message: str,
        history: List[Dict[str, Any]],
        session_id: str
    ):
        """
        Genera respuesta con streaming - FORMATO GRADIO 4.x CORRECTO
        
        Args:
            message: Mensaje del usuario
            history: Historial en formato correcto
            session_id: ID de sesión
            
        Yields:
            Historial actualizado y texto de modo
        """
        async for updated_history, mode_text in response_generator.generate_streaming_response(
            message, 
            history
        ):
            # Actualizar historial de sesión
            session_histories[session_id] = updated_history.copy()
            yield updated_history, mode_text
    
    # ==================== INTERFAZ GRADIO ====================
    with gr.Blocks(
        title="🧠 SISTEMA_PROTOCOLO_POTENCIACION_ACTIVO v2.0",
        theme=gr.themes.Soft(),
        css="""
        .gradio-container {
            max-width: 1200px !important;
            margin: 0 auto !important;
        }
        .chatbot {
            height: 600px !important;
            overflow-y: auto !important;
            border: 1px solid #e0e0e0 !important;
            border-radius: 10px !important;
            padding: 20px !important;
        }
        .status-panel {
            background: linear-gradient(135deg, #667eea 0%, #764ba2 100%) !important;
            color: white !important;
            padding: 15px !important;
            border-radius: 10px !important;
            margin-bottom: 20px !important;
        }
        """
    ) as interface:
        
        # Estado de sesión
        session_state = gr.State(value=generate_session_id())
        
        gr.Markdown("""
        # 🧠 SISTEMA_PROTOCOLO_POTENCIACION_ACTIVO v2.0
        ### *Asistente de IA Senior en Ingeniería de Software y DevOps*
        
        ---
        
        **Especialidades:** 
        • Arquitectura de Sistemas Distribuidos • Microservicios • DevOps • Cloud Security
        • Optimización de Rendimiento • Code Review Profesional • Soluciones Production-Ready
        
        **Modo actual:** `OPTIMIZACION_ASCENDENTE_EXCLUSIVA`
        """)
        
        with gr.Row():
            with gr.Column(scale=3):
                # Chatbot con formato correcto para Gradio 4.x
                chatbot = gr.Chatbot(
                    label="Conversación",
                    height=600,
                    show_label=False,
                    bubble_full_width=False,
                    avatar_images=(
                        "https://cdn-icons-png.flaticon.com/512/3135/3135715.png",  # User
                        "https://cdn-icons-png.flaticon.com/512/4712/4712035.png"   # Assistant
                    ),
                    sanitize_html=False,
                    show_copy_button=True
                )
                
                # Panel de estado
                status_display = gr.Markdown(
                    "**Modo**: Inicializando sistema...",
                    elem_classes="status-panel"
                )
                
                # Input con validación
                msg_input = gr.Textbox(
                    label="Escribe tu mensaje técnico",
                    placeholder="Ejemplo: 'Optimiza este microservicio Python para alta concurrencia...'",
                    lines=3,
                    max_lines=6,
                    show_label=False
                )
                
                with gr.Row():
                    submit_btn = gr.Button(
                        "🚀 Enviar Consulta Técnica",
                        variant="primary",
                        scale=2
                    )
                    clear_btn = gr.Button(
                        "🧹 Limpiar Chat",
                        variant="secondary",
                        scale=1
                    )
            
            with gr.Column(scale=1):
                # Panel de información técnica
                gr.Markdown("### 📊 Métricas del Sistema")
                
                session_id_display = gr.Textbox(
                    label="ID de Sesión",
                    value=session_state.value,
                    interactive=False
                )
                
                message_count = gr.Number(
                    label="Mensajes en sesión",
                    value=0,
                    interactive=False
                )
                
                gr.Markdown("""
                ### ⚙️ Configuración
                
                - **Formato:** Gradio 4.x Compatible
                - **Streaming:** Activado
                - **Persistencia:** Por sesión
                - **Logging:** Estructurado
                - **Seguridad:** Input Sanitization
                """)
        
        # ==================== EVENT HANDLERS ====================
        
        def update_message_count(history):
            """Actualiza contador de mensajes"""
            user_messages = [m for m in history if m.get("role") == "user"]
            return len(user_messages)
        
        # Submit handler
        submit_event = msg_input.submit(
            fn=process_message,
            inputs=[msg_input, chatbot, session_state],
            outputs=[msg_input, chatbot, status_display],
            queue=True
        ).then(
            fn=update_message_count,
            inputs=[chatbot],
            outputs=[message_count]
        ).then(
            fn=generate_response_stream,
            inputs=[msg_input, chatbot, session_state],
            outputs=[chatbot, status_display]
        )
        
        # Button click handler
        submit_btn.click(
            fn=process_message,
            inputs=[msg_input, chatbot, session_state],
            outputs=[msg_input, chatbot, status_display],
            queue=True
        ).then(
            fn=update_message_count,
            inputs=[chatbot],
            outputs=[message_count]
        ).then(
            fn=generate_response_stream,
            inputs=[msg_input, chatbot, session_state],
            outputs=[chatbot, status_display]
        )
        
        # Clear handler
        def clear_chat():
            """Limpia el chat y genera nueva sesión"""
            new_session_id = generate_session_id()
            initial_message = [{
                "role": "system",
                "content": "Sesión reiniciada. Listo para asistencia técnica.",
                "timestamp": datetime.now().isoformat()
            }]
            return (
                initial_message,  # chatbot
                "**Modo**: Sistema reiniciado - Esperando consulta técnica",  # status
                new_session_id,  # session_state
                0,  # message_count
                new_session_id  # session_id_display
            )
        
        clear_btn.click(
            fn=clear_chat,
            outputs=[chatbot, status_display, session_state, message_count, session_id_display]
        )
        
        # ==================== CONFIGURACIÓN ADICIONAL ====================
        
        # Ejemplos de consultas
        gr.Examples(
            examples=[
                ["Optimiza este endpoint REST para manejar 10K RPS con Python async/await"],
                ["Diseña una arquitectura de microservicios con auto-scaling y circuit breaker"],
                ["Implementa un sistema de logging distribuido con OpenTelemetry y Grafana"],
                ["Genera un Dockerfile production-ready para aplicación Python con seguridad hardening"],
                ["Crea un sistema de CI/CD con GitHub Actions para despliegue blue-green en Kubernetes"]
            ],
            inputs=msg_input,
            label="📋 Ejemplos de consultas técnicas"
        )
        
        # Footer
        gr.Markdown("""
        ---
        
        **🔒 Seguridad activa:** Todas las consultas son procesadas con sanitización de inputs y validación de tipos.
        **📈 Logging:** Todas las interacciones son registradas para mejora continua del sistema.
        **⚡ Performance:** Respuestas optimizadas con streaming en tiempo real.
        
        *Sistema compatible con Gradio 4.x - Formato de mensajes: `{"role": "rol", "content": "texto"}`*
        """)
    
    return interface

# ==================== PUNTO DE ENTRADA ====================
if __name__ == "__main__":
    # Validación de entorno
    try:
        import gradio as gr
        GRADIO_VERSION = gr.__version__
        logger.info(f"Gradio versión detectada: {GRADIO_VERSION}")
        
        if not GRADIO_VERSION.startswith("4."):
            logger.warning(
                f"Esta interfaz está optimizada para Gradio 4.x. "
                f"Versión actual: {GRADIO_VERSION}. "
                f"Puede haber incompatibilidades."
            )
        
        # Crear e iniciar interfaz
        interface = create_advanced_interface()
        
        # Configuración para Hugging Face Spaces
        interface.queue(
            max_size=20,
            default_concurrency_limit=10,
            api_open=True
        ).launch(
            server_name="0.0.0.0" if "SPACE_ID" in os.environ else None,
            server_port=7860,
            share=False,
            debug=False,
            show_error=True,
            favicon_path="https://cdn-icons-png.flaticon.com/512/4712/4712035.png"
        )
        
    except Exception as e:
        logger.critical(f"Error crítico al iniciar la aplicación: {str(e)}", exc_info=True)
        
        # Interfaz de fallback mínima
        with gr.Blocks() as fallback:
            gr.Markdown(f"""
            # ⚠️ Error en la inicialización del sistema
            
            **Detalles técnicos:**
            ```python
            {str(e)}
            ```
            
            Por favor, verifica:
            1. Dependencias de Gradio 4.x instaladas
            2. Formato de mensajes correcto
            3. Logs del sistema para diagnóstico
            """)
        
        fallback.launch()