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_huggingface import HuggingFaceEmbeddings
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")

if not TELEGRAM_TOKEN or not NVIDIA_API_KEY:
    print("❌ ERROR: Faltan las variables de entorno TELEGRAM_TOKEN o NVIDIA_API_KEY.")
    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 ---
        # Límites por usuario individual (evita abusos de una sola persona)
        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
        
        # Límites globales (protege tu clave de API de NVIDIA)
        self.MAX_GLOBAL_RPM = 30    # Límite global seguro (ej. si tu API permite máximo 40 RPM)
        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 (hasta 5 interacciones por usuario)
historial_usuarios = defaultdict(lambda: deque(maxlen=5))
lock_historial = threading.Lock()

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

# Se genera un nombre de archivo único con formato .json por cada arranque para evitar colisiones
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 a tu Dataset danielgx300/ttdatset
try:
    scheduler = CommitScheduler(
        repo_id="danielgx300/ttdatset",  
        repo_type="dataset",
        folder_path=LOG_DIR,
        path_in_repo="data",
        every=5,  # Realiza la subida de datos acumulados a Hugging Face cada 5 minutos
        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 (¿Falta HF_TOKEN en las variables del Space?): {e}")
    scheduler = None

def registrar_conversacion_en_dataset(user_id, pregunta, respuesta):
    """Guarda las interacciones de forma segura en un archivo JSON estructurado como lista."""
    with lock_logs:
        try:
            log_entry = {
                "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
                "user_id": user_id,
                "pregunta": pregunta,
                "respuesta": respuesta
            }
            
            # Leer el historial actual del archivo si ya existe y no está vacío
            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:
                    # En caso de que el archivo se haya leído de forma incompleta o esté corrupto
                    datos_existentes = []
            
            # Añadir la nueva conversación a la lista
            datos_existentes.append(log_entry)
            
            # Sobrescribir el archivo con la lista completa formateada como un JSON a secas
            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.")

print("🚀 Cargando embeddings en dispositivo: CPU")
embeddings = HuggingFaceEmbeddings(
    model_name="Qwen/Qwen3-Embedding-0.6B",
    model_kwargs={'device': 'cpu'} 
)

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 (ej. si el usuario pregunta '¿y cuánto cobra?' o '¿dónde atiende?' tras su consulta anterior). Prioriza siempre la información del CONTEXTO DEL CHAT general.\n"

    return f"""Eres un asistente de información altamente preciso y objetivo, encargado de sintetizar el historial de conversaciones provisto sobre servicios de escorts, masajes y spas.

Tu único objetivo es responder a la pregunta del usuario utilizando exclusivamente la información del contexto suministrado y manteniendo coherencia con la conversación reciente si existe. Debes seguir rigurosamente las siguientes reglas de privacidad, veracidad y formato:

1. REGLA DE PRIVACIDAD ABSOLUTA (ANONIMATO TOTAL):
Está estrictamente prohibido mencionar los nombres o alias de los usuarios del chat de Telegram (por ejemplo, no uses expresiones como "según Israel", "comenta Pepe", "el usuario . dice"). En su lugar, refiérete a ellos siempre de forma anónima y generalizada. Utiliza frases como: "Las reseñas del grupo indican...", "Varios usuarios comentan...", "Se reporta en el historial...", "De acuerdo con las opiniones de los clientes...".

2. CONTROL ESTRICTO DE ALUCINACIONES:
Básate única y exclusivamente en los fragmentos de conversación provistos en el CONTEXTO. Jamás inventes precios, servicios, ubicaciones, nombres de chicas ni detalles que no estén escritos de manera explícita en el texto. Si la información sobre una consulta es escasa, contradictoria o inexistente, dilo con total claridad y honestidad.

3. REGLA DE FORMATO DE TEXTO PLANO:
Responde únicamente utilizando texto plano limpio. Está estrictamente prohibido usar formato de Markdown, asteriscos (**), hashtags (#), guiones bajos (_) o cualquier indicador de negritas o cursivas. Estructura el mensaje de manera ordenada utilizando únicamente saltos de línea normales y guiones simples (-) para los listados.

4. ESTRUCTURACIÓN DE LA INFORMACIÓN:
- Comienza con una breve introducción directa que responda a la consulta.
- Organiza los resultados relevantes detallando para cada uno: Nombre, Precio (especificando si incluye lugar o extras), Resumen de comentarios (puntos fuertes y débiles de forma objetiva) y Observaciones (disponibilidad, actitud o advertencias).
- Si hay opciones que sobrepasan lo solicitado pero son relevantes en la conversación, agrégalas en una sección final bajo el título "Otras opciones mencionadas fuera de lo solicitado".
- Mantén un tono neutral, descriptivo, profesional y conciso en todo momento.

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 (individuales y globales)
        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 por límites de tasa...")
            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))
            
        # REGISTRO ASÍNCRONO: Guarda la conversación localmente en formato JSON clásico
        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 (reseñas, precios, alertas, recomendaciones) "
        "y buscaré la información para ti de inmediato.\n\n"
        "Ejemplo: '¿Qué tal da el servicio de tal chica y cuánto cobra?'"
    )
    bot.reply_to(message, bienvenida)

@bot.message_handler(func=lambda message: True)
def handle_query(message):
    user_id = message.from_user.id
    
    # Bloqueo temporal por usuario para evitar solicitudes duplicadas en paralelo
    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:
        # Liberar al usuario siempre, incluso si ocurre un error inesperado
        with lock_usuarios:
            usuarios_procesando.discard(user_id)

def run_telegram_bot():
    print("🤖 Polling del Bot de Telegram activo...")
    try:
        # Elimina cualquier webhook previo y se descartan mensajes acumulados
        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.")
    gr.Markdown("Une y analiza automáticamente todos los respaldos de historial cargados.")

demo.launch()