app.py

import os
import json
import glob
import sys
import time
import uuid
import threading
from pathlib import Path
from collections import deque, defaultdict
import telebot
import gradio as gr
import spaces
from openai import OpenAI
from huggingface_hub import CommitScheduler
from langchain_nvidia_ai_endpoints import NVIDIAEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document

@spaces.GPU
def apreton_de_manos_gpu():
    return "Hugging Face Zero-GPU Satisfecho"

print(f"🤖 {apreton_de_manos_gpu()}")

TELEGRAM_TOKEN = os.environ.get("TELEGRAM_TOKEN")
NVIDIA_API_KEY = os.environ.get("NVIDIA_API_KEY")
SYSTEM_PROMPT = os.environ.get("System_Prompt")

# Validación estricta de variables de entorno, incluyendo el System_Prompt
if not TELEGRAM_TOKEN or not NVIDIA_API_KEY or not SYSTEM_PROMPT:
    print("❌ ERROR: Faltan las variables de entorno TELEGRAM_TOKEN, NVIDIA_API_KEY o System_Prompt.")
    sys.exit(1)

# Configuración del bot con un pool de 20 hilos para procesar múltiples usuarios en paralelo
bot = telebot.TeleBot(TELEGRAM_TOKEN, num_threads=20)

nvidia_client = OpenAI(
    base_url="https://integrate.api.nvidia.com/v1",
    api_key=NVIDIA_API_KEY
)

class ControlDeTraficoIA:
    def __init__(self):
        self.lock = threading.Lock()
        
        # Estructuras por usuario individual
        self.peticiones_por_usuario = defaultdict(deque)
        self.tokens_por_usuario = defaultdict(deque)
        
        # Estructuras globales (para proteger la API de NVIDIA)
        self.peticiones_globales = deque()
        self.tokens_globales = deque()
        
        # --- CONFIGURACIÓN DE LÍMITES ---
        self.MAX_USER_RPM = 3       # Máximo 3 peticiones por minuto por usuario
        self.MAX_USER_TPM = 15000   # Máximo 15,000 tokens por minuto por usuario
        
        self.MAX_GLOBAL_RPM = 30    # Límite global seguro
        self.MAX_GLOBAL_TPM = 120000 # Límite global de tokens por minuto entre todos los usuarios

    def verificar_y_esperar(self, user_id, estimado_tokens):
        with self.lock:
            now = time.time()
            
            peticiones_usr = self.peticiones_por_usuario[user_id]
            tokens_usr = self.tokens_por_usuario[user_id]
            
            # Limpieza de registros con antigüedad mayor a 60 segundos
            while peticiones_usr and now - peticiones_usr[0] > 60:
                peticiones_usr.popleft()
            while tokens_usr and now - tokens_usr[0][0] > 60:
                tokens_usr.popleft()
                
            while self.peticiones_globales and now - self.peticiones_globales[0] > 60:
                self.peticiones_globales.popleft()
            while self.tokens_globales and now - self.tokens_globales[0][0] > 60:
                self.tokens_globales.popleft()
            
            # Evaluar límites individuales
            limite_usr_peticiones = len(peticiones_usr) >= self.MAX_USER_RPM
            limite_usr_tokens = (sum(t[1] for t in tokens_usr) + estimado_tokens) > self.MAX_USER_TPM
            
            # Evaluar límites globales de la API
            limite_glob_peticiones = len(self.peticiones_globales) >= self.MAX_GLOBAL_RPM
            limite_glob_tokens = (sum(t[1] for t in self.tokens_globales) + estimado_tokens) > self.MAX_GLOBAL_TPM
            
            # Si ningún límite se sobrepasa, registrar consumo y permitir la solicitud
            if not (limite_usr_peticiones or limite_usr_tokens or limite_glob_peticiones or limite_glob_tokens):
                peticiones_usr.append(now)
                tokens_usr.append((now, estimado_tokens))
                self.peticiones_globales.append(now)
                self.tokens_globales.append((now, estimado_tokens))
                return 0
            
            # Validación preventiva de longitud para evitar errores de tipo 'index out of range'
            espera_usr_pet = int(60 - (now - peticiones_usr[0])) + 1 if (limite_usr_peticiones and len(peticiones_usr) > 0) else 0
            espera_usr_tok = int(60 - (now - tokens_usr[0][0])) + 1 if (limite_usr_tokens and len(tokens_usr) > 0) else 0
            espera_glob_pet = int(60 - (now - self.peticiones_globales[0])) + 1 if (limite_glob_peticiones and len(self.peticiones_globales) > 0) else 0
            espera_glob_tok = int(60 - (now - self.tokens_globales[0][0])) + 1 if (limite_glob_tokens and len(self.tokens_globales) > 0) else 0
            
            return max(espera_usr_pet, espera_usr_tok, espera_glob_pet, espera_glob_tok)

    def registrar_consumo_forzado(self, user_id, estimado_tokens):
        with self.lock:
            now = time.time()
            self.peticiones_por_usuario[user_id].append(now)
            self.tokens_por_usuario[user_id].append((now, estimado_tokens))
            self.peticiones_globales.append(now)
            self.tokens_globales.append((now, estimado_tokens))

control_trafico = ControlDeTraficoIA()

# Control de concurrencia para evitar múltiples solicitudes simultáneas de un mismo usuario
usuarios_procesando = set()
lock_usuarios = threading.Lock()

# Historial de conversación individual en memoria
historial_usuarios = defaultdict(lambda: deque(maxlen=5))
lock_historial = threading.Lock()

# --- ALMACENAMIENTO DE CHATS EN FORMATO JSON TRADICIONAL ---
LOG_DIR = Path("logs")
LOG_DIR.mkdir(exist_ok=True)

LOG_FILE_NAME = f"chat_logs_{uuid.uuid4()}.json"
LOG_FILE_PATH = LOG_DIR / LOG_FILE_NAME

lock_logs = threading.Lock()

# Inicializador asíncrono del scheduler para subir cambios al Dataset
try:
    scheduler = CommitScheduler(
        repo_id="danielgx300/ttdatset",  
        repo_type="dataset",
        folder_path=LOG_DIR,
        path_in_repo="data",
        every=5,
        private=True,
        token=os.environ.get("HF_TOKEN")
    )
    print("✅ CommitScheduler cargado para el dataset 'danielgx300/ttdatset'.")
except Exception as e:
    print(f"⚠️ Advertencia al iniciar CommitScheduler: {e}")
    scheduler = None

def registrar_conversacion_en_dataset(user_id, pregunta, respuesta):
    with lock_logs:
        try:
            log_entry = {
                "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
                "user_id": user_id,
                "pregunta": pregunta,
                "respuesta": respuesta
            }
            
            datos_existentes = []
            if LOG_FILE_PATH.exists() and LOG_FILE_PATH.stat().st_size > 0:
                try:
                    with open(LOG_FILE_PATH, "r", encoding="utf-8") as f:
                        datos_existentes = json.load(f)
                except Exception:
                    datos_existentes = []
            
            datos_existentes.append(log_entry)
            
            with open(LOG_FILE_PATH, "w", encoding="utf-8") as f:
                json.dump(datos_existentes, f, ensure_ascii=False, indent=4)
                
        except Exception as e:
            print(f"❌ Error al escribir log de chat local: {e}")


def parse_telegram_text(text_obj):
    if isinstance(text_obj, str):
        return text_obj
    elif isinstance(text_obj, list):
        out = ""
        for item in text_obj:
            if isinstance(item, str):
                out += item
            elif isinstance(item, dict):
                out += item.get('text', '')
        return out
    return ""

# Carga e indexación de archivos JSON (Arranque inicial)
json_files = glob.glob("result*.json")
print(f"📥 Archivos de historial encontrados: {json_files}")

valid_messages = []
seen_messages = set()

for file_path in sorted(json_files):
    print(f"📖 Leyendo {file_path}...")
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            chat_data = json.load(f)
        
        messages = chat_data.get('messages', [])
        file_messages_count = 0
        
        for msg in messages:
            if msg.get('type') != 'message': 
                continue
            text = parse_telegram_text(msg.get('text', ''))
            if not text.strip(): 
                continue
            
            user = msg.get('from', 'Desconocido')
            date = msg.get('date', '').replace('T', ' ')
            msg_str = f"[{date}] {user}: {text}"
            
            if msg_str not in seen_messages:
                seen_messages.add(msg_str)
                valid_messages.append(msg_str)
                file_messages_count += 1
                
        print(f"✅ Se cargaron {file_messages_count} mensajes únicos desde {file_path}")
    except Exception as e:
        print(f"❌ Error al procesar el archivo {file_path}: {e}")

del seen_messages

CHUNK_SIZE = 30
OVERLAP = 10
documents = []
for i in range(0, len(valid_messages), CHUNK_SIZE - OVERLAP):
    chunk_msgs = valid_messages[i : i + CHUNK_SIZE]
    chunk_text = "\n".join(chunk_msgs)
    documents.append(Document(page_content=chunk_text))

print(f"📊 Total consolidado: {len(documents)} bloques de conversación únicos.")

# --- CARGA DE EMBEDDINGS DESDE NVIDIA NIM ---
print("🚀 Cargando embeddings desde NVIDIA NIM...")
embeddings = NVIDIAEmbeddings(
    model="nvidia/nemotron-3-embed-1b",
    nvidia_api_key=NVIDIA_API_KEY
)

if documents:
    print("🧠 Creando base de datos unificada...")
    vectorstore = FAISS.from_documents(documents, embeddings)
    retriever = vectorstore.as_retriever(search_kwargs={"k": 15})
    print("✅ Base de datos lista.")
else:
    print("⚠️ No hay documentos para indexar.")
    retriever = None

def construir_prompt(context_text, chat_history_list, user_query):
    historial_str = ""
    if chat_history_list:
        historial_str = "\nHISTORIAL DE CONVERSACIÓN RECIENTE CON ESTE USUARIO:\n"
        for user_msg, assistant_msg in chat_history_list:
            historial_str += f"- Usuario: {user_msg}\n- Asistente (tú): {assistant_msg}\n"
        historial_str += "\nNota: Utiliza este historial de conversación únicamente para interpretar pronombres o dar continuidad. Prioriza siempre la información del CONTEXTO DEL CHAT general.\n"

    return f"""{SYSTEM_PROMPT}

CONTEXTO DEL CHAT:
{context_text}
{historial_str}
PREGUNTA DEL USUARIO:
{user_query}

Responde de forma estructurada en texto plano siguiendo estrictamente todas las reglas anteriores:
"""

def obtener_respuesta_rag(user_query, user_id):
    if not retriever:
        return "❌ El sistema de base de datos no está disponible temporalmente."
        
    try:
        docs = retriever.invoke(user_query)
        context_text = "\n\n---\n\n".join([doc.page_content for doc in docs])
        
        # Obtener historial individual del usuario
        with lock_historial:
            chat_history_list = list(historial_usuarios[user_id])
        
        prompt = construir_prompt(context_text, chat_history_list, user_query)
        estimacion_tokens = len(prompt) // 3
        
        # Verificar límites de tráfico
        espera_necesaria = control_trafico.verificar_y_esperar(user_id, estimacion_tokens)
        if espera_necesaria > 0:
            print(f"⚠️ [COLA DE ESPERA - USUARIO {user_id}] Esperando {espera_necesaria} segundos...")
            time.sleep(espera_necesaria)
            control_trafico.registrar_consumo_forzado(user_id, estimacion_tokens)

        print(f"🚀 [BOT] Enviando prompt a NVIDIA NIM para el usuario {user_id}...")
        completion = nvidia_client.chat.completions.create(
            model="thinkingmachines/inkling",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.6,
            top_p=0.95,
            max_tokens=16384,
            extra_body={"chat_template_kwargs": {"thinking": True, "reasoning_effort": "high"}},
            stream=False
        )
        
        reasoning = getattr(completion.choices[0].message, "reasoning", None) or getattr(completion.choices[0].message, "reasoning_content", None)
        if reasoning:
            print(f"\n🧠 === PROCESO DE RAZONAMIENTO - USUARIO {user_id} ===")
            print(reasoning)
            print("==================================================\n")
            
        respuesta_ia = completion.choices[0].message.content
        
        # Guardar interacción en la memoria en caché del usuario
        with lock_historial:
            historial_usuarios[user_id].append((user_query, respuesta_ia))
            
        registrar_conversacion_en_dataset(user_id, user_query, respuesta_ia)
            
        return respuesta_ia
        
    except Exception as e:
        return f"❌ Error en el proceso de consulta: {e}"

@bot.message_handler(commands=['start', 'help'])
def send_welcome(message):
    bienvenida = (
        "¡Hola! Soy tu asistente inteligente del grupo HDL. 🌶️\n\n"
        "Pregúntame lo que quieras sobre el historial de conversaciones consolidado y buscaré la información para ti de inmediato."
    )
    bot.reply_to(message, bienvenida)

@bot.message_handler(func=lambda message: True)
def handle_query(message):
    user_id = message.from_user.id
    
    with lock_usuarios:
        if user_id in usuarios_procesando:
            bot.reply_to(message, "⚠️ Ya estoy procesando una consulta para ti. Por favor, espera a que termine de responderte.")
            return
        usuarios_procesando.add(user_id)
        
    try:
        bot.send_chat_action(message.chat.id, 'typing')
        msg_espera = bot.reply_to(message, "🔍 Buscando en el historial y procesando...")
        
        respuesta_ia = obtener_respuesta_rag(message.text, user_id)
        
        if len(respuesta_ia) > 4000:
            respuesta_ia = respuesta_ia[:4000] + "\n\n⚠️ Respuesta truncada por longitud máxima..."
            
        bot.edit_message_text(chat_id=message.chat.id, message_id=msg_espera.message_id, text=respuesta_ia)
    except Exception as e:
        print(f"❌ Error al gestionar mensaje de Telegram para el usuario {user_id}: {e}")
    finally:
        with lock_usuarios:
            usuarios_procesando.discard(user_id)

def run_telegram_bot():
    print("🤖 Polling del Bot de Telegram activo...")
    try:
        bot.delete_webhook(drop_pending_updates=True)
        bot.infinity_polling()
    except Exception as e:
        print(f"❌ Error en el loop del bot: {e}")

threading.Thread(target=run_telegram_bot, daemon=True).start()

with gr.Blocks() as demo:
    gr.Markdown("# 🌶️ Servidor del Asistente HDL activo")
    gr.Markdown("El bot de Telegram está escuchando preguntas en segundo plano utilizando el modelo de IA.")

demo.launch()