rag_api.py

import os
import re
import httpx
import asyncio
from typing import List, Optional, Tuple
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, Request, Response
from pydantic import BaseModel, Field

# --- IMPORTS DE LANGCHAIN ---
from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_core.runnables import RunnablePassthrough
from langchain_core.prompts import PromptTemplate
from langchain_groq import ChatGroq

# ------------------------------------------------------------------------------
# 1. CONFIGURACIÓN DE ENTORNO, TOKENS Y CACHÉ
# ------------------------------------------------------------------------------
TEMP_CACHE_DIR = '/tmp/huggingface_cache'
os.environ['TRANSFORMERS_CACHE'] = TEMP_CACHE_DIR
os.environ['HF_HOME'] = TEMP_CACHE_DIR
os.environ['SENTENCE_TRANSFORMERS_HOME'] = TEMP_CACHE_DIR
os.makedirs(TEMP_CACHE_DIR, exist_ok=True)

# Variables de entorno obtenidas desde Secrets
VERIFY_TOKEN = os.getenv("VERIFY_TOKEN")
WHATSAPP_TOKEN = os.getenv("WHATSAPP_TOKEN")
WHATSAPP_PHONE_ID = os.getenv("WHATSAPP_PHONE_ID")

# ------------------------------------------------------------------------------
# 2. MODELOS DE DATOS (DTOs)
# ------------------------------------------------------------------------------
class QueryRequest(BaseModel):
    query: str = Field(..., min_length=2, max_length=500, description="Consulta del usuario sobre Godot Engine")
    phone_number: Optional[str] = Field(None, max_length=20, pattern=r"^\+?[0-9\s\-]+$")

class QueryResponse(BaseModel):
    query: str
    response: str
    intent: str
    sources: List[str]

# ------------------------------------------------------------------------------
# 3. NOTIFICADOR EXTERNO (Meta / WhatsApp Cloud API)
# ------------------------------------------------------------------------------
class WhatsAppNotifier:
    """Envía mensajes directamente usando la API Oficial de Meta WhatsApp Cloud."""
    def __init__(self):
        # trust_env=False evita interferencias del entorno del contenedor HF
        self.client = httpx.AsyncClient(
            timeout=httpx.Timeout(30.0, connect=15.0),
            trust_env=False
        )

    async def send_message(self, phone_number: str, message_text: str) -> bool:
        token = os.getenv("WHATSAPP_TOKEN")
        phone_id = os.getenv("WHATSAPP_PHONE_ID")
        api_url = f"https://graph.facebook.com/v19.0/{phone_id}/messages" if phone_id else None

        if not phone_number or not token or not api_url:
            print(f"[WhatsAppNotifier Error] Faltan credenciales: phone_number={bool(phone_number)}, token={bool(token)}, api_url={bool(api_url)}")
            return False

        headers = {
            "Authorization": f"Bearer {token.strip()}",
            "Content-Type": "application/json"
        }
        payload = {
            "messaging_product": "whatsapp",
            "recipient_type": "individual",
            "to": phone_number,
            "type": "text",
            "text": {"preview_url": False, "body": message_text}
        }
        
        try:
            response = await self.client.post(api_url, json=payload, headers=headers)
            if response.status_code != 200:
                print(f"[WhatsAppNotifier Error] Meta API respondió Status {response.status_code}: {response.text}")
                return False
            print(f"[WhatsAppNotifier Success] Mensaje enviado correctamente a {phone_number}")
            return True
        except Exception as err:
            print(f"[WhatsAppNotifier Error] Excepción al realizar POST a Meta: {type(err).__name__} - {err}")
            return False

    async def close(self):
        await self.client.aclose()

# ------------------------------------------------------------------------------
# 4. RAG PIPELINE ASÍNCRONO
# ------------------------------------------------------------------------------
class GodotRAGPipeline:
    def __init__(self, doc_url: str):
        self.doc_url = doc_url
        self.embeddings = None
        self.vectorstore = None
        self.retriever = None
        self.llm = None
        self.intent_chain = None
        self.saludo_chain = None
        self.rag_chain = None
        self._init_prompts()

    def _init_prompts(self):
        self.intent_prompt = PromptTemplate(
            template="""Eres un sistema clasificador estricto de intenciones para un asistente de Godot Engine.
Tu ÚNICA tarea es devolver una de estas palabras clave: SALUDO, GODOT o OTRO.

Categorías:
- SALUDO: saludos, despedidas, conversación casual.
- GODOT: preguntas técnicas sobre código, nodos, funciones, errores o GDScript.
- OTRO: cualquier otra pregunta ajena a Godot.

Entrada del usuario a clasificar:
\"\"\"{query}\"\"\"

Responde ÚNICAMENTE con la palabra de la categoría (SALUDO, GODOT o OTRO):""",
            input_variables=["query"]
        )
        self.saludo_prompt = PromptTemplate(
            template="""Hola, soy tu asistente experto en Godot Engine. ¿En qué script o problema técnico te puedo ayudar hoy? 🤖
Mensaje del usuario: \"\"\"{query}\"\"\"
Respuesta:""",
            input_variables=["query"]
        )
        self.rag_prompt = PromptTemplate(
            template="""Eres un desarrollador experto en Godot Engine. Responde la pregunta del usuario utilizando la información del contexto extraído de la documentación oficial.
Contexto oficial:
{context}

Pregunta del usuario: \"\"\"{question}\"\"\"
Respuesta técnica:""",
            input_variables=["context", "question"]
        )

    def _sanitize_input(self, text: str) -> str:
        text = text.replace('"""', '""')
        text = re.sub(r'[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]', '', text)
        return text.strip()

    def initialize(self):
        try:
            print(f"[GodotRAGPipeline] Cargando contenido desde: {self.doc_url}")
            loader = WebBaseLoader(self.doc_url)
            docs = loader.load()
            text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
            splits = text_splitter.split_documents(docs)

            print("[GodotRAGPipeline] Cargando modelo de Embeddings...")
            self.embeddings = HuggingFaceEmbeddings(
                model_name="sentence-transformers/all-MiniLM-L6-v2",
                model_kwargs={'device': 'cpu'},
                cache_folder=TEMP_CACHE_DIR
            )
            print("[GodotRAGPipeline] Construyendo índice vectorial FAISS...")
            self.vectorstore = FAISS.from_documents(splits, self.embeddings)
            self.retriever = self.vectorstore.as_retriever(search_kwargs={"k": 3})

            print("[GodotRAGPipeline] Inicializando modelo LLM (Groq)...")
            self.llm = ChatGroq(temperature=0.2, model_name="qwen/qwen3.6-27b", max_tokens=1024)
            self.intent_chain = self.intent_prompt | self.llm
            self.saludo_chain = self.saludo_prompt | self.llm
            print("[GodotRAGPipeline] Pipeline inicializado de forma segura.")
        except Exception as e:
            print(f"[GodotRAGPipeline CRITICAL] Error al inicializar: {e}")
            raise RuntimeError(f"Falla crítica en RAG Pipeline: {e}")

    async def process_query(self, query_text: str) -> Tuple[str, str, List[str]]:
        clean_query = self._sanitize_input(query_text)
        intent_res = await self.intent_chain.ainvoke({"query": clean_query})
        intent = intent_res.content.strip().upper()
        sources = []

        if "SALUDO" in intent:
            response = await self.saludo_chain.ainvoke({"query": clean_query})
            return response.content, "SALUDO", sources
        elif "OTRO" in intent:
            response = "Disculpa, solo puedo ayudarte con temas de programación y desarrollo en Godot Engine. 🕹️"
            return response, "OTRO", sources
        else:
            docs = await self.retriever.ainvoke(clean_query)
            context_text = "\n\n".join([doc.page_content for doc in docs])
            response = await (self.rag_prompt | self.llm).ainvoke({
                "context": context_text, 
                "question": clean_query
            })
            sources = list(set([doc.metadata.get("source", self.doc_url) for doc in docs]))
            return response.content, "GODOT_INFO", sources

# ------------------------------------------------------------------------------
# 5. MANEJO DE TAREAS Y CONTROLADOR API (FastAPI)
# ------------------------------------------------------------------------------
DOC_URL = "https://docs.godotengine.org/es/stable/tutorials/scripting/gdscript/gdscript_basics.html"
rag_pipeline = GodotRAGPipeline(doc_url=DOC_URL)
notifier = WhatsAppNotifier()

# Estructura para almacenar referencias fuertes de tareas asíncronas
background_tasks_set = set()

async def process_and_send_whatsapp(sender_phone: str, user_text: str):
    """Procesa la respuesta del RAG en el Event Loop en segundo plano."""
    try:
        print(f"[AsyncTask] Procesando consulta RAG para {sender_phone}: '{user_text}'")
        response_text, intent, _ = await rag_pipeline.process_query(user_text)
        print(f"[AsyncTask] Respuesta generada ({intent}). Enviando POST a Meta...")
        await notifier.send_message(sender_phone, response_text)
    except Exception as e:
        print(f"[AsyncTask Error] Error durante la ejecución asíncrona: {type(e).__name__} - {e}")

@asynccontextmanager
async def lifespan(app: FastAPI):
    try:
        rag_pipeline.initialize()
    except Exception as e:
        print(f"Error al iniciar el servicio: {e}")
    yield
    await notifier.close()

app = FastAPI(title="Godot Link RAG API", lifespan=lifespan)

@app.get("/")
def health_check():
    return {"status": "active", "message": "Servicio RAG de Godot operativo y protegido."}

# --- WEBHOOK PARA VALIDACIÓN Y RECEPCIÓN DE WHATSAPP ---
@app.get("/webhook")
async def verify_webhook(request: Request):
    params = request.query_params
    mode = params.get("hub.mode")
    token = params.get("hub.verify_token")
    challenge = params.get("hub.challenge")

    if mode == "subscribe" and token == VERIFY_TOKEN:
        print("[Webhook] Verificación exitosa de Meta.")
        return Response(content=challenge, status_code=200)
    
    return Response(content="Token de verificación inválido", status_code=403)

@app.post("/webhook")
async def receive_whatsapp_message(request: Request):
    data = await request.json()
    try:
        entries = data.get("entry", [])
        for entry in entries:
            for change in entry.get("changes", []):
                value = change.get("value", {})
                messages = value.get("messages", [])
                if messages:
                    message = messages[0]
                    sender_phone = message.get("from")

                    if message.get("type") == "text":
                        user_text = message["text"]["body"]
                        print(f"[Webhook] Mensaje de {sender_phone}: {user_text}")
                        
                        # Crear tarea y guardar referencia fuerte para evitar GC (Garbage Collection)
                        task = asyncio.create_task(process_and_send_whatsapp(sender_phone, user_text))
                        background_tasks_set.add(task)
                        task.add_done_callback(background_tasks_set.discard)

    except Exception as e:
        print(f"[Webhook Error] Fallo al procesar evento entrante: {e}")

    # Retorna HTTP 200 inmediatamente a Meta para responder antes de los 3 segundos
    return {"status": "success"}

@app.post("/query", response_model=QueryResponse)
async def handle_query(request: QueryRequest):
    try:
        response_text, intent, sources = await rag_pipeline.process_query(request.query)
        if request.phone_number:
            await notifier.send_message(request.phone_number, response_text)
        return QueryResponse(
            query=request.query,
            response=response_text,
            intent=intent,
            sources=sources
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error interno procesando la consulta: {str(e)}")