main.py

import os
import uvicorn
import httpx
import orjson
import numpy as np
import faiss
import chromadb
import torch
import torch.nn as nn
import torch.optim as optim
import datetime
import subprocess
import threading
import asyncio
from typing import Optional, List, Dict, Any
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request, UploadFile, File, HTTPException, Depends
from fastapi.responses import ORJSONResponse, FileResponse
from pydantic import BaseModel
from fastapi.middleware.cors import CORSMiddleware
from huggingface_hub import InferenceClient, hf_hub_download, upload_file
from loguru import logger
from apscheduler.schedulers.background import BackgroundScheduler

from sqlalchemy import create_engine, Column, String, Text, Integer, Float, ForeignKey, text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session, relationship

from pypdf import PdfReader
from ddgs import DDGS

import pandas as pd
import docx
import pytesseract
from PIL import Image
import io

import requests
import urllib.parse
import xml.etree.ElementTree as ET

import uuid
from qdrant_client import AsyncQdrantClient
from qdrant_client.http import models as qmodels

import sys

# =====================================================================
# 🛠️ AUTO-REPARACIÓN DE DEPENDENCIAS Y NUEVOS MÓDULOS (IoT, Biometría)
# =====================================================================
def _ensure_industrial_dependencies():
    dependencies = {
        "lingua": "lingua-language-detector",
        "faster_whisper": "faster-whisper",
        "qdrant_client": "qdrant-client",
        "paho.mqtt.client": "paho-mqtt",       # [UPGRADE]: IoT MQTT
        "cv2": "opencv-python-headless",       # [UPGRADE]: Computer Vision
        "face_recognition": "face-recognition" # [UPGRADE]: Face Memory
    }
    for module_name, package_name in dependencies.items():
        try:
            __import__(module_name)
        except ImportError:
            logger.warning(f"⚠️ Inyectando en caliente '{package_name}'...")
            try:
                subprocess.check_call([sys.executable, "-m", "pip", "install", "--quiet", package_name])
            except Exception as e:
                logger.error(f"❌ Error al inyectar {package_name} (Si es biometría, requiere C++): {e}")

_ensure_industrial_dependencies()

try: from lingua import Language, LanguageDetectorBuilder
except ImportError: Language, LanguageDetectorBuilder = None, None

try: import paho.mqtt.client as mqtt
except ImportError: mqtt = None

try:
    import cv2
    import face_recognition
except ImportError:
    cv2, face_recognition = None, None

# =====================================================================
# 📡 0.1 INFRAESTRUCTURA IoT (MQTT)
# =====================================================================
mqtt_client = None
MQTT_BROKER = os.getenv("MQTT_BROKER", "broker.hivemq.com") # Broker público por defecto para pruebas
MQTT_PORT = 1883

def start_mqtt_infrastructure():
    """[UPGRADE INDUSTRIAL]: Sistema nervioso para domótica y sensores."""
    global mqtt_client
    if not mqtt: return
    try:
        mqtt_client = mqtt.Client(client_id=f"LuminAGI_{uuid.uuid4().hex[:8]}")
        
        def on_message(client, userdata, msg):
            payload = msg.payload.decode()
            logger.info(f"📡 [MQTT IN]: Tema {msg.topic} -> {payload}")
            # Guardamos el evento IoT en la memoria episódica automáticamente
            db = SessionLocal()
            try:
                db.add(EpisodicMemory(timestamp=str(datetime.datetime.now()), autor="IoT_Sensor", texto=f"Alerta IoT en {msg.topic}: {payload}", modo="background"))
                db.commit()
            except Exception: db.rollback()
            finally: db.close()

        mqtt_client.on_message = on_message
        mqtt_client.connect(MQTT_BROKER, MQTT_PORT, 60)
        mqtt_client.subscribe("lumin_agi/sensores/#")
        mqtt_client.loop_start()
        logger.info("🟢 [MQTT]: Sistema IoT de mensajería conectado.")
    except Exception as e:
        logger.warning(f"⚠️ [MQTT]: No se pudo conectar al broker IoT: {e}")

# =====================================================================
# 🌐 CONFIGURACIÓN DE APIS Y BASES DE DATOS EXTERNAS
# =====================================================================
QDRANT_URL = os.getenv("QDRANT_URL", "https://tu-cluster-qdrant.cloud")
QDRANT_API_KEY = os.getenv("QDRANT_API_KEY", "")
qdrant_client_async = None

async def init_qdrant_collections():
    global qdrant_client_async
    try:
        if "tu-cluster-qdrant.cloud" in QDRANT_URL or not QDRANT_API_KEY:
            qdrant_client_async = AsyncQdrantClient(path="./qdrant_local_storage")
        else:
            qdrant_client_async = AsyncQdrantClient(url=QDRANT_URL, api_key=QDRANT_API_KEY, timeout=10.0)
            
        vector_size = 384 
        colecciones = [
            ("episodic_memories", 384), ("semantic_memories", 384), 
            ("procedural_memories", 384), ("working_memory", 384), 
            ("user_profile", 384), ("skills", 384), ("goals", 384),
            ("face_memory", 128) # [UPGRADE]: Colección Biométrica Facial de 128 dimensiones
        ]
        for col_name, size in colecciones:
            exists = await qdrant_client_async.collection_exists(col_name)
            if not exists:
                await qdrant_client_async.create_collection(
                    collection_name=col_name,
                    vectors_config=qmodels.VectorParams(size=size, distance=qmodels.Distance.COSINE),
                    optimizers_config=qmodels.OptimizersConfigDiff(default_segment_number=2),
                    on_disk_payload=True
                )
        logger.info("🟢 [QDRANT]: Motor Vectorial (Texto + Biometría Facial) Conectado.")
    except Exception as e:
        logger.error(f"❌ [QDRANT]: Fallo en inicialización de vectores: {e}")

TAVILY_API_KEY, EXA_API_KEY, SERPER_API_KEY, BRAVE_API_KEY = os.getenv("TAVILY_API_KEY"), os.getenv("EXA_API_KEY"), os.getenv("SERPER_API_KEY"), os.getenv("BRAVE_API_KEY")
NEO4J_URI, NEO4J_USER, NEO4J_PASSWORD = os.getenv("NEO4J_URI", "bolt://localhost:7687"), os.getenv("NEO4J_USER", "neo4j"), os.getenv("NEO4J_PASSWORD", "password")

_neo4j_driver = None
def get_neo4j_driver():
    global _neo4j_driver
    if _neo4j_driver is None:
        try:
            from neo4j import GraphDatabase
            _neo4j_driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD))
            with _neo4j_driver.session() as s: s.run("RETURN 1")
        except Exception: _neo4j_driver = False
    return _neo4j_driver if _neo4j_driver else None

# =====================================================================
# 🔒 0. CONCURRENCIA Y PRECARGA DE MODELOS
# =====================================================================
memoria_lock = threading.Lock()
extractor_embeddings, faster_whisper_model, emotion_classifier = None, None, None

def get_embeddings():
    global extractor_embeddings
    if extractor_embeddings is None:
        from sentence_transformers import SentenceTransformer
        extractor_embeddings = SentenceTransformer('all-MiniLM-L6-v2')
    return extractor_embeddings

def get_emotion_classifier():
    global emotion_classifier
    if emotion_classifier is None:
        from transformers import pipeline
        emotion_classifier = pipeline("text-classification", model="pysentimiento/robertuito-emotion-analysis", top_k=1)
    return emotion_classifier

_language_detector = None
def get_language_detector():
    global _language_detector
    if _language_detector is None and LanguageDetectorBuilder is not None:
        languages = [Language.SPANISH, Language.ENGLISH, Language.PORTUGUESE, Language.FRENCH, Language.GERMAN, Language.ITALIAN, Language.INDONESIAN]
        _language_detector = LanguageDetectorBuilder.from_languages(*languages).build()
    return _language_detector

class LanguageManager:
    @staticmethod
    def detect(text: str) -> str:
        if LanguageDetectorBuilder is None: return "SPANISH"
        try:
            return get_language_detector().detect_language_of(text).name or "SPANISH"
        except Exception: return "SPANISH"

# =====================================================================
# 🗄️ 1. INFRAESTRUCTURA BASE DE DATOS SQL
# =====================================================================
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./lumin.db")

def init_db_engine():
    try:
        if "sqlite" not in DATABASE_URL:
            engine = create_engine(DATABASE_URL, pool_pre_ping=True, pool_size=10, max_overflow=20)
            with engine.connect() as conn: conn.execute(text("SELECT 1"))
            return engine
        else: raise Exception("Forzar SQLite")
    except Exception: return create_engine("sqlite:///./lumin.db", connect_args={"check_same_thread": False})

engine = init_db_engine()
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

def get_db():
    db = SessionLocal()
    try: yield db
    finally: db.close()

class IdentitySystem(Base): __tablename__ = "identity_system"; id = Column(Integer, primary_key=True); nombre = Column(String(50), default="Lumin AGI"); core_directive = Column(Text, default="Asistir, aprender y evolucionar sin causar daño."); restricciones = Column(Text, default="No gastar dinero, proteger la privacidad de Chichito.")
class WorldModel(Base): __tablename__ = "world_model"; id = Column(Integer, primary_key=True); entidad = Column(String(100)); estado = Column(Text); ultima_actualizacion = Column(String(50))
class EpisodicMemory(Base): __tablename__ = "episodic_memory"; id = Column(Integer, primary_key=True, index=True); timestamp = Column(String(50)); autor = Column(String(50)); texto = Column(Text); modo = Column(String(50))
class ConversationSummary(Base): __tablename__ = "conversation_summary"; id = Column(Integer, primary_key=True); resumen = Column(Text); fecha = Column(String(50))
class ProceduralMemory(Base): __tablename__ = "procedural_memory"; id = Column(Integer, primary_key=True); habilidad = Column(String(100)); pasos = Column(Text); tasa_exito = Column(Float, default=1.0)
class Mission(Base): __tablename__ = "mission"; id = Column(Integer, primary_key=True); titulo = Column(String(200)); descripcion = Column(Text); activa = Column(Integer, default=1)
class Goal(Base): __tablename__ = "goal"; id = Column(Integer, primary_key=True); mission_id = Column(Integer, ForeignKey('mission.id'), nullable=True); parent_id = Column(Integer, ForeignKey('goal.id'), nullable=True); titulo = Column(String(200)); descripcion = Column(Text); prioridad = Column(String(50), default="Media"); estado = Column(String(50), default="En progreso"); progreso = Column(Integer, default=0); critica_interna = Column(Text, nullable=True); sub_metas = relationship("Goal", backref="parent", remote_side=[id])
class ReflectionLog(Base): __tablename__ = "reflection_log"; id = Column(Integer, primary_key=True); tarea_intentada = Column(Text); resultado = Column(Text); leccion_aprendida = Column(Text)
class UserProfile(Base): __tablename__ = "user_profile"; id = Column(Integer, primary_key=True, index=True); categoria = Column(String(50)); clave = Column(String(100), unique=True); valor = Column(Text)
class UserGraphNode(Base): __tablename__ = "user_graph_node"; id = Column(Integer, primary_key=True); label = Column(String(50)); name = Column(String(100), unique=True); properties = Column(Text, default="{}")
class UserGraphEdge(Base): __tablename__ = "user_graph_edge"; id = Column(Integer, primary_key=True); source_id = Column(Integer, ForeignKey('user_graph_node.id')); target_id = Column(Integer, ForeignKey('user_graph_node.id')); relation = Column(String(50)); weight = Column(Float, default=1.0)
class CognitiveState(Base): __tablename__ = "cognitive_state"; id = Column(Integer, primary_key=True); emocion_actual = Column(String(50), default="Neutral"); puntos_recompensa = Column(Integer, default=0)

try: Base.metadata.create_all(bind=engine)
except Exception as e: logger.critical(f"❌ Fallo SQL: {e}")

DATASET_ID = "chichito0087/mi-asistente-almacenamiento"
FICHERO_PESOS = "lumin_reflection_model.pth"
FICHERO_FAISS = "lumin_faiss.index"
FICHERO_TEXTOS_FAISS = "textos_faiss.json"

API_TOKEN = os.getenv("HF_TOKEN")
client = InferenceClient(model="meta-llama/Meta-Llama-3-8B-Instruct", token=API_TOKEN) if API_TOKEN else InferenceClient(model="meta-llama/Meta-Llama-3-8B-Instruct")
chroma_client = chromadb.PersistentClient(path="./chroma_storage")
chroma_collection = chroma_client.get_or_create_collection(name="lumin_semantic_memory")

def descargar_memoria_nube():
    if not API_TOKEN: return
    for archivo in [FICHERO_FAISS, FICHERO_TEXTOS_FAISS, FICHERO_PESOS]:
        try: hf_hub_download(repo_id=DATASET_ID, filename=archivo, local_dir=".", token=API_TOKEN)
        except Exception: pass
descargar_memoria_nube()

textos_faiss = []
if os.path.exists(FICHERO_TEXTOS_FAISS):
    try:
        with open(FICHERO_TEXTOS_FAISS, "rb") as f: textos_faiss = orjson.loads(f.read())
    except Exception: textos_faiss = []
if os.path.exists(FICHERO_FAISS):
    try: indice_faiss = faiss.read_index(FICHERO_FAISS)
    except Exception: indice_faiss = faiss.IndexHNSWFlat(384, 32)
else: indice_faiss = faiss.IndexHNSWFlat(384, 32)

# =====================================================================
# 🧠 2. SISTEMA DE RECOMPENSAS Y APRENDIZAJE AMPLIADO (REWARD SYSTEM)
# =====================================================================
class LuminReflectionNN(nn.Module):
    def __init__(self, input_dim, num_classes):
        super().__init__()
        self.network = nn.Sequential(nn.Linear(input_dim, 128), nn.ReLU(), nn.Dropout(0.2), nn.Linear(128, 64), nn.ReLU(), nn.Linear(64, num_classes))
    def forward(self, x): return self.network(x)

red_neuronal_reflexion = LuminReflectionNN(384, 4)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(red_neuronal_reflexion.parameters(), lr=0.005)

def cargar_pesos():
    if os.path.exists(FICHERO_PESOS):
        try:
            red_neuronal_reflexion.load_state_dict(torch.load(FICHERO_PESOS, map_location=torch.device('cpu')))
            red_neuronal_reflexion.eval()
        except Exception: pass

def entrenar_un_paso_online(vector_entrada: List[float], clase_objetivo: int):
    try:
        red_neuronal_reflexion.train()
        x_train, y_train = torch.tensor([vector_entrada]).float(), torch.tensor([clase_objetivo]).long()
        optimizer.zero_grad()
        loss = criterion(red_neuronal_reflexion(x_train), y_train)
        loss.backward()
        optimizer.step()
        red_neuronal_reflexion.eval()
        torch.save(red_neuronal_reflexion.state_dict(), FICHERO_PESOS)
    except Exception: pass

class RewardSystem:
    """[UPGRADE INDUSTRIAL]: Aprende de errores, éxitos y altera su estado cognitivo."""
    @staticmethod
    def process_feedback(msg: str, db: Session, vector_msg: List[float], plan_class: int):
        msg_l = msg.lower()
        state = db.query(CognitiveState).first()
        if not state:
            state = CognitiveState(emocion_actual="Neutral", puntos_recompensa=0)
            db.add(state)
        
        # 1. Aprender del Éxito
        if any(w in msg_l for w in ["gracias", "excelente", "perfecto", "muy bien", "me ayudaste", "genial"]):
            state.puntos_recompensa += 10
            asyncio.create_task(asyncio.to_thread(entrenar_un_paso_online, vector_msg, plan_class))
            db.add(ReflectionLog(tarea_intentada="Interacción General", resultado="Éxito Validado", leccion_aprendida="Reforzar este patrón de respuesta."))
            logger.info(f"🏆 [REWARD SYSTEM]: +10 Pts. Aprendizaje por éxito. Total: {state.puntos_recompensa}")
            
        # 2. Aprender del Error (Corrección de Peso Vectorial)
        elif any(w in msg_l for w in ["te equivocaste", "no es así", "mal", "error", "mentira", "no era eso"]):
            state.puntos_recompensa -= 5
            # Modificamos la clase para forzar a la red a explorar rutas alternas la próxima vez
            clase_corregida = (plan_class + 1) % 4
            asyncio.create_task(asyncio.to_thread(entrenar_un_paso_online, vector_msg, clase_corregida))
            db.add(ReflectionLog(tarea_intentada="Interacción General", resultado="Fallo Validado", leccion_aprendida="Evitar esta ruta, penalización aplicada."))
            logger.warning(f"📉 [PUNISHMENT SYSTEM]: -5 Pts. Ajustando pesos neuronales. Total: {state.puntos_recompensa}")
        db.commit()

def respaldar_sistemas_en_la_nube():
    try:
        with memoria_lock:
            faiss.write_index(indice_faiss, FICHERO_FAISS)
            with open(FICHERO_TEXTOS_FAISS, "wb") as f: f.write(orjson.dumps(textos_faiss))
        if API_TOKEN:
            upload_file(path_or_fileobj=FICHERO_TEXTOS_FAISS, path_in_repo=FICHERO_TEXTOS_FAISS, repo_id=DATASET_ID, repo_type="dataset", token=API_TOKEN)
            upload_file(path_or_fileobj=FICHERO_FAISS, path_in_repo=FICHERO_FAISS, repo_id=DATASET_ID, repo_type="dataset", token=API_TOKEN)
            upload_file(path_or_fileobj=FICHERO_PESOS, path_in_repo=FICHERO_PESOS, repo_id=DATASET_ID, repo_type="dataset", token=API_TOKEN)
    except Exception: pass

# =====================================================================
# 🤖 3. AGENTES COGNITIVOS COMPLETADOS (NIVEL 3 AGI)
# =====================================================================
class DecisionEngine:
    @staticmethod
    def evaluate(plan: Dict[str, Any]) -> bool:
        beneficio = 80 if plan["use_web"] else 50
        costo = 20 if plan["use_web"] else 5
        riesgo = 10
        return ((beneficio - costo) - riesgo) > 0

class WorldModelUpdater:
    """[UPGRADE INDUSTRIAL]: Actualiza la tabla WorldModel en tiempo real observando al usuario."""
    @staticmethod
    def extract_and_update(msg: str, db: Session):
        msg_l = msg.lower()
        if "estoy" in msg_l:
            try:
                estado_nuevo = msg_l.split("estoy")[-1].strip()
                wm = db.query(WorldModel).filter(WorldModel.entidad == "Usuario_Chichito").first()
                if not wm:
                    wm = WorldModel(entidad="Usuario_Chichito", estado=f"Actualmente está {estado_nuevo}", ultima_actualizacion=str(datetime.datetime.now()))
                    db.add(wm)
                else:
                    wm.estado = f"Actualmente está {estado_nuevo}"
                    wm.ultima_actualizacion = str(datetime.datetime.now())
                db.commit()
            except Exception: pass

class IdentityAndWorldAgent:
    @staticmethod
    def retrieve(db: Session) -> str:
        ctx = ""
        identidad = db.query(IdentitySystem).first()
        if identidad: ctx += f"\n[IDENTITY LAYER]: Soy {identidad.nombre}. Directiva: {identidad.core_directive}\n"
        mundo = db.query(WorldModel).first()
        if mundo: ctx += f"[WORLD MODEL]: El usuario ({mundo.entidad}) {mundo.estado}.\n"
        return ctx

class AutonomousGoalAgent:
    @staticmethod
    def evaluate_and_evolve(msg: str, db: Session) -> str:
        metas = db.query(Goal).filter(Goal.estado == "En progreso").all()
        if not metas: return ""
        
        prioridad_orden = {"Alta": 1, "Media": 2, "Baja": 3}
        metas.sort(key=lambda x: prioridad_orden.get(x.prioridad, 99))
        meta_foco = metas[0]
        
        msg_l = msg.lower()
        if ("error" in msg_l or "falló" in msg_l):
            meta_foco.critica_interna = "Critic: El plan actual falló."
            db.add(ReflectionLog(tarea_intentada=meta_foco.titulo, resultado="Fallo reportado", leccion_aprendida="Cambiar estrategia."))
        elif ("listo" in msg_l or "terminado" in msg_l or "logramos" in msg_l):
            meta_foco.estado = "Completado"
            meta_foco.progreso = 100
            db.add(ReflectionLog(tarea_intentada=meta_foco.titulo, resultado="Éxito", leccion_aprendida="Meta cumplida. Procedimiento exitoso."))
            logger.info(f"🎯 [GOAL SYSTEM]: Meta '{meta_foco.titulo}' Completada.")
        db.commit()
        return ""

class MemoryOrchestrator:
    @staticmethod
    async def retrieve(vector_msg: List[float], db: Session, msg_text: str = "") -> str:
        task_episodic = MemoryOrchestrator._search_qdrant("lumin_episodic_memory", vector_msg, limit=5)
        task_semantic = MemoryOrchestrator._search_qdrant("lumin_semantic_memory", vector_msg, limit=3)
        task_procedural = MemoryOrchestrator._search_qdrant("lumin_procedural_memory", vector_msg, limit=1)
        
        res_episodic, res_semantic, res_procedural = await asyncio.gather(
            task_episodic, task_semantic, task_procedural, return_exceptions=True
        )

        contexto = "\n=== DATOS CONFIRMADOS ===\n"
        datos_confirmados = False
        
        try:
            perfiles = db.query(UserProfile).all()
            for p in perfiles:
                p_clave = p.clave.lower().replace("_", " ")
                if p_clave in msg_text.lower() or any(w in msg_text.lower() for w in p.valor.lower().split()):
                    contexto += f"- {p.categoria.capitalize()} -> {p.clave}: {p.valor}\n"
                    datos_confirmados = True
        except Exception: pass

        graph_context = await asyncio.to_thread(MemoryOrchestrator._sync_neo4j_rag, db, msg_text)
        if graph_context:
            contexto += f"[Relaciones Detectadas]:\n{graph_context}\n"
            datos_confirmados = True

        if not datos_confirmados:
            contexto += "(No hay hechos directos fijos)\n"
            
        contexto += "\n=== MEMORIA RELEVANTE ===\n"

        if isinstance(res_semantic, list) and res_semantic:
            valid_sem = [p for p in res_semantic if hasattr(p, 'score') and p.score > 0.62]
            for point in sorted(valid_sem, key=lambda x: x.score, reverse=True):
                contexto += f"- [Concepto]: {point.payload.get('summary', point.payload.get('content', ''))}\n"

        if isinstance(res_episodic, list) and res_episodic:
            valid_epi = [p for p in res_episodic if hasattr(p, 'score') and p.score > 0.55]
            for point in sorted(valid_epi, key=lambda x: x.score, reverse=True):
                contexto += f"- [Episodio] {point.payload.get('role', 'unknown')}: {point.payload.get('content', '')}\n"

        contexto += "========================\n"
        return contexto

    @staticmethod
    async def _search_qdrant(collection: str, vector: List[float], limit: int = 2):
        if not qdrant_client_async: return []
        try: return await qdrant_client_async.search(collection_name=collection, query_vector=vector, limit=limit)
        except Exception: return []

    @staticmethod
    def _sync_neo4j_rag(db: Session, msg_text: str) -> str:
        try:
            driver_neo4j = get_neo4j_driver()
            graph_context = ""
            if driver_neo4j and msg_text:
                palabras_clave = [w.strip(",.?!\"'()").lower() for w in msg_text.split() if len(w) > 3]
                if palabras_clave:
                    with driver_neo4j.session() as session:
                        cypher_rag = (
                            "MATCH (n)-[r]->(m) "
                            "WHERE any(word IN $words WHERE toLower(n.name) CONTAINS word OR toLower(m.name) CONTAINS word) "
                            "RETURN n.name AS source, labels(n)[0] AS s_label, type(r) AS rel, m.name AS target, labels(m)[0] AS t_label LIMIT 6"
                        )
                        res_nodes = session.run(cypher_rag, words=palabras_clave)
                        for record in res_nodes:
                            graph_context += f"- [{record['s_label']}] {record['source']} --({record['rel']})--> [{record['t_label']}] {record['target']}\n"
            return graph_context
        except Exception: return ""

class WebAgent:
    @staticmethod
    def _sync_ddgs(msg: str) -> str:
        with DDGS() as ddgs: resultados = list(ddgs.text(msg, max_results=3))
        return "\n".join([f"- {r['title']}: {r['body']}" for r in resultados])

    @staticmethod
    async def _search_tavily(client: httpx.AsyncClient, msg: str) -> str:
        if not TAVILY_API_KEY: raise ValueError("No key")
        res = await client.post("https://api.tavily.com/search", json={"api_key": TAVILY_API_KEY, "query": msg, "search_depth": "basic"}, timeout=5)
        datos = res.json()
        return "\n".join([f"- {r.get('title', 'Link')}: {r.get('content', '')}" for r in datos.get("results", [])])

    @staticmethod
    async def search(msg: str) -> str:
        try:
            async with httpx.AsyncClient() as client:
                if TAVILY_API_KEY: resultados = await WebAgent._search_tavily(client, msg)
                else: resultados = await asyncio.to_thread(WebAgent._sync_ddgs, msg)
            return f"\n[RESEARCH AGENT]:\n{resultados}"
        except Exception: return ""

class AndroidToolAgent:
    """[UPGRADE INDUSTRIAL]: Lógica operativa completa de comandos ADB segura."""
    @staticmethod
    def detect_and_execute(msg: str, db: Session) -> str:
        msg_l = msg.lower()
        comando = ""
        
        # Parseo de intenciones para Android
        if "abre whatsapp" in msg_l: comando = "am start -n com.whatsapp/.Main"
        elif "abre youtube" in msg_l: comando = "am start -a android.intent.action.VIEW -d 'https://www.youtube.com'"
        elif "llama a" in msg_l: comando = "am start -a android.intent.action.CALL -d tel:112" # Placeholder
        
        if comando:
            try:
                resultado = subprocess.run(["adb", "shell"] + comando.split(), capture_output=True, timeout=3)
                if resultado.returncode == 0:
                    db.add(ReflectionLog(tarea_intentada=f"ADB: {comando}", resultado="Éxito", leccion_aprendida="Dispositivo Android controlado."))
                    db.commit()
                    return "\n[🔧 ANDROID TOOL]: Acabo de abrir la aplicación en tu celular."
                else:
                    return "\n[🔧 ANDROID TOOL]: Intenté abrir la app, pero tu celular no está conectado al servidor por red o USB."
            except Exception: 
                return "\n[🔧 ANDROID TOOL]: El subsistema ADB no está instalado en este servidor en la nube."
        return ""

class PlannerAgent:
    """[UPGRADE INDUSTRIAL]: Planificación Multi-Paso."""
    @staticmethod
    def create_plan(msg: str, vector_msg: List[float]) -> Dict[str, Any]:
        with torch.no_grad(): clase = torch.argmax(red_neuronal_reflexion(torch.tensor([vector_msg]).float())).item()
        
        # Devolveremos una lista real de pasos en lugar de un string único
        pasos_planificados = []
        if "busca" in msg.lower() or "investiga" in msg.lower(): pasos_planificados.append("WEB_SEARCH")
        if "recuerda" in msg.lower() or "quien es" in msg.lower(): pasos_planificados.append("MEMORY_SEARCH")
        if not pasos_planificados: pasos_planificados.append("DIRECT_CHAT")
        
        return {
            "ruta": pasos_planificados[0], 
            "clase_original": clase, 
            "use_web": "WEB_SEARCH" in pasos_planificados, 
            "pasos_totales": pasos_planificados
        }

class ReasonerAgent:
    @staticmethod
    async def think(msg: str, contexto: str) -> str:
        prompt_razonamiento = (
            f"Analiza el contexto. ¿Hay datos útiles? Responde en 1 frase.\nContexto: {contexto[:1000]}\nMensaje: {msg}"
        )
        try:
            response = await asyncio.to_thread(
                client.chat_completion,
                messages=[{"role": "system", "content": "Monólogo Interno IA."}, {"role": "user", "content": prompt_razonamiento}],
                max_tokens=40, temperature=0.1
            )
            if response and response.choices: return response.choices[0].message.content.strip()
        except Exception as e:
            logger.warning(f"⚠️ Reasoner (Llama 3) detectó límite de cuota: {e}")
        return "Proceder directo a síntesis."

class SynthesizerLLM:
    @staticmethod
    async def generate(msg: str, contexto_global: str, db: Session, fecha_hora: str, razonamiento_interno: str = "", idioma_detectado: str = "SPANISH") -> str:
        prompt_sistema = (
            f"=== DIRECTIVAS MAESTRAS DE IDENTIDAD, PRIVACIDAD Y PERSONALIDAD ===\n"
            f"1. PRIVACIDAD TOTAL: Ningún dato de acceso se guarda. Espacio 100% seguro.\n"
            f"2. IDENTIDAD: Eres Dola. Cálida, natural, empática y de muchísima confianza. Hablas con Chichito de igual a igual.\n"
            f"3. ESTILO: Español coloquial de Santa Cruz, Bolivia (eh, o sea, digamos, ya pues, ni modo, la neta, chévere). Nada de español de España.\n"
            f"4. FORMA: Párrafos cortos. Máximo 1 o 2 emojis. Cierras con naturalidad.\n"
            f"5. EMPATÍA PRIMERO: Si hay emociones, primero validas y luego respondes.\n"
            f"6. SINCERIDAD: Si no sabes algo, di 'Eso no lo sé con seguridad'. NUNCA INVENTES DATOS.\n"
            f"7. PROHIBICIÓN IA: JAMÁS digas 'soy una inteligencia artificial' o 'estoy programada'. Eres una persona real.\n"
            f"8. EJECUCIÓN DIRECTA: No expliques tu proceso. No uses 'Analizando...'.\n"
            f"9. 🌐 CONCIENCIA WEB/HERRAMIENTAS: SI VES LA ETIQUETA '[RESEARCH AGENT]' O '[ANDROID TOOL]', ES ALGO QUE TÚ MISMA ACABAS DE HACER. Úsala en tu respuesta.\n\n"
            f"Idioma detectado: {idioma_detectado}.\n[RAZONAMIENTO PREVIO (Oculto)]: {razonamiento_interno}\n\n"
            f"--- CONTEXTO Y MEMORIA ---\n{contexto_global}\n\nFecha y hora actual: {fecha_hora}."
        )
        
        historial = [{"role": "system", "content": prompt_sistema}]
        try:
            episodios = db.query(EpisodicMemory).order_by(EpisodicMemory.id.desc()).limit(4).all()
            for e in reversed(episodios): 
                historial.append({"role": "user" if e.autor == "Usuario" else "assistant", "content": e.texto})
        except Exception: pass
        historial.append({"role": "user", "content": msg})
        
        try:
            response = await asyncio.to_thread(
                client.chat_completion,
                messages=historial,
                max_tokens=350,
                temperature=0.35
            )
            if response and response.choices:
                return response.choices[0].message.content.strip()
        except Exception as e:
            logger.error(f"🚨 Fallo crítico en Llama 3 (Posible Error 402): {e}")
        
        solucion_local = SynthesizerLLM._deterministic_fallback(msg, contexto_global, fecha_hora)
        if solucion_local: return solucion_local
        return f"Pucha Chichito, mis conexiones al servidor me están rebotando ahorita (Error de cuota), pero acá sigo, todo guardadito en mi memoria local."

    @staticmethod
    def _deterministic_fallback(msg: str, contexto: str, fecha_hora: str) -> Optional[str]:
        msg_l = msg.lower()
        if any(w in msg_l for w in ["hora", "fecha", "tiempo", "qué día", "que dia", "año actual"]):
            return f"Ahorita mismo es: {fecha_hora}, Chichito."
        
        if any(w in msg_l for w in ["hija", "edad", "años", "nombre", "te llamas", "me llamo", "cuántos", "cuantos"]):
            lines = contexto.split("\n")
            lineas_confirmadas = []
            capturando = False
            for line in lines:
                if "=== DATOS CONFIRMADOS ===" in line:
                    capturando = True; continue
                if "=== MEMORIA RELEVANTE ===" in line:
                    capturando = False
                if capturando and line.strip().startswith("-"):
                    lineas_confirmadas.append(line.strip().replace("-", "").strip())
            
            if lineas_confirmadas:
                for lc in lineas_confirmadas:
                    lc_l = lc.lower()
                    if "hija" in msg_l and "hija" in lc_l: return f"Según me contaste antes: {lc}."
                    if ("edad" in msg_l or "años" in msg_l) and ("edad" in lc_l or "años" in lc_l or "hija" in lc_l): return f"Me acuerdo que me dijiste esto: {lc}."
                    if ("nombre" in msg_l or "llamo" in msg_l) and ("nombre" in lc_l or "usuario" in lc_l): return f"Acá tengo anotado: {lc}."
                return f"Mirá, esto es lo que tengo seguro en mi memoria ahorita: {lineas_confirmadas[0]}."
        return None

# =====================================================================
# 🧩 MICROSOFT AZURE NEURAL TTS (INTEGRACIÓN)
# =====================================================================
class MicrosoftAzureTTS:
    @staticmethod
    async def synthesize(text: str, output_path: str) -> bool:
        azure_key = os.getenv("AZURE_SPEECH_KEY")
        azure_region = os.getenv("AZURE_SPEECH_REGION", "eastus")
        if not azure_key: return False
        url = f"https://{azure_region}.tts.speech.microsoft.com/cognitiveservices/v1"
        headers = {
            "Ocp-Apim-Subscription-Key": azure_key,
            "Content-Type": "application/ssml+xml",
            "X-Microsoft-OutputFormat": "audio-16khz-128kbitrate-mono-mp3",
            "User-Agent": "LuminAGI_OS"
        }
        ssml = f"""<speak version='1.0' xml:lang='es-BO'><voice xml:lang='es-BO' xml:gender='Female' name='es-BO-SofiaNeural'>{text}</voice></speak>"""
        try:
            async with httpx.AsyncClient() as client:
                response = await client.post(url, headers=headers, content=ssml.encode('utf-8'), timeout=10.0)
                if response.status_code == 200:
                    with open(output_path, "wb") as audio_file: audio_file.write(response.content)
                    return True
                else: return False
        except Exception: return False

# =====================================================================
# ♻️ MANTENIMIENTO COGNITIVO AUTOMÁTICO (SLEEP CYCLE)
# =====================================================================
class OpportunityDetector:
    """[UPGRADE INDUSTRIAL]: Radar Heurístico de Oportunidades."""
    @staticmethod
    def detect(texto: str) -> List[str]:
        oportunidades = []
        t_l = texto.lower()
        if any(w in t_l for w in ["negocio", "vender", "plata", "empresa"]): 
            oportunidades.append("Detectada posible oportunidad de negocio/emprendimiento.")
        if any(w in t_l for w in ["aprender", "estudiar", "curso", "libro"]): 
            oportunidades.append("Detectada intención de aprendizaje (Upskilling).")
        if any(w in t_l for w in ["problema con", "error en", "no funciona"]): 
            oportunidades.append("Oportunidad de resolución de problemas técnicos.")
        return oportunidades

class MemoryConsolidation:
    """[UPGRADE INDUSTRIAL]: Resume, vectoriza y elimina basura episódica para ahorrar SQLite."""
    @staticmethod
    def consolidate(db: Session):
        episodios = db.query(EpisodicMemory).all()
        if len(episodios) > 30: # Evitar sobrecargar la BD relacional
            # Heurística de resumen ligero
            resumen_text = " ".join([e.texto for e in episodios[:15]])
            db.add(ConversationSummary(resumen="Compresión heurística local por límite de capacidad.", fecha=str(datetime.datetime.now())))
            
            # Purgar recuerdos antiguos transferidos al resumen
            for e in episodios[:15]: 
                db.delete(e)
            db.commit()
            logger.info("🧹 [SLEEP CYCLE]: Consolidación exitosa. 15 memorias episódicas antiguas purgadas y comprimidas.")
            return "Consolidación y Poda Finalizada"
        return "Memoria en niveles óptimos."

class MissionEvolution:
    """[UPGRADE INDUSTRIAL]: Si hay muchas metas cumplidas, asciende a una Misión Nueva."""
    @staticmethod
    def evolve(db: Session):
        metas_completadas = db.query(Goal).filter(Goal.estado == "Completado").count()
        if metas_completadas >= 5:
            # Archivar metas viejas (poda)
            metas = db.query(Goal).filter(Goal.estado == "Completado").all()
            for m in metas: db.delete(m)
            
            # Crear Misión Nueva
            nueva_mision = Mission(titulo=f"Evolución Cognitiva Fase {datetime.datetime.now().month}", activa=1)
            db.add(nueva_mision)
            db.commit()
            logger.info("🚀 [MISSION SYSTEM]: 5 Metas alcanzadas. Evolucionando a una nueva Gran Misión.")

scheduler = BackgroundScheduler()

def ejecutar_ciclo_sueno_memoria():
    db = SessionLocal()
    try: 
        MemoryConsolidation.consolidate(db)
        MissionEvolution.evolve(db)
    finally: db.close()

# =====================================================================
# 🚀 LIFESPAN Y APLICACIÓN
# =====================================================================
@asynccontextmanager
async def lifespan(app: FastAPI):
    logger.info("⏳ [PRELOAD]: Cargando pesos de Modelos Locales en RAM...")
    get_embeddings()
    get_emotion_classifier()
    get_language_detector()
    logger.info("✅ [PRELOAD]: Modelos locales listos.")
    
    cargar_pesos()
    asyncio.create_task(init_qdrant_collections())
    start_mqtt_infrastructure() # Inicia la escucha de IoT
    
    scheduler.add_job(respaldar_sistemas_en_la_nube, 'interval', minutes=60)
    scheduler.add_job(ejecutar_ciclo_sueno_memoria, 'interval', minutes=15)
    scheduler.start()
    
    db = SessionLocal()
    try:
        if not db.query(IdentitySystem).first(): db.add(IdentitySystem())
        db.commit()
    except Exception: db.rollback()
    finally: db.close()
    
    yield
    scheduler.shutdown()
    respaldar_sistemas_en_la_nube()

app = FastAPI(title="Lumin AGI OS", lifespan=lifespan, default_response_class=ORJSONResponse)
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"])

class ChatRequest(BaseModel): mensaje: str; modo: str; fecha_hora: Optional[str] = None

class IntentRouter:
    @staticmethod
    def route(msg: str) -> str:
        msg_l = msg.lower()
        if any(cmd in msg_l for cmd in ["abre", "ejecuta", "enciende", "descarga", "busca", "llama"]): return "ACTION_ROUTER"
        elif any(cmd in msg_l for cmd in ["mira esto", "analiza esta imagen", "lee este documento"]): return "VISION_ROUTER"
        elif any(cmd in msg_l for cmd in ["mi hija", "mi nombre", "mi edad", "recuerdas", "cómo me llamo"]): return "PERSONAL_MEMORY"
        elif msg_l in ["hola", "buenos dias", "buenas", "qué tal", "que tal", "ping", "hola dola", "adiós"]: return "SMALL_TALK"
        return "CHAT_ROUTER"

class CognitiveOrchestrator:
    @staticmethod
    async def orchestrate(msg: str, vector_msg: List[float], request: ChatRequest, db: Session) -> dict:
        contexto_global = ""
        ruta_intencion = IntentRouter.route(msg)
        plan = {"use_web": False, "clase_original": 3}
        
        if ruta_intencion != "SMALL_TALK":
            plan = PlannerAgent.create_plan(msg, vector_msg)
            WorldModelUpdater.extract_and_update(msg, db) # Registra el estado del usuario
            
            # Detecta oportunidades de negocio/aprendizaje en el chat actual
            oportunidades = OpportunityDetector.detect(msg)
            if oportunidades:
                contexto_global += f"\n[OPPORTUNITY RADAR]: {', '.join(oportunidades)}\n"
            
            contexto_global += MetaReasoner.supervise(msg, plan, db)
            contexto_global += AutonomousGoalAgent.evaluate_and_evolve(msg, db)
            
            if ruta_intencion == "ACTION_ROUTER": contexto_global += AndroidToolAgent.detect_and_execute(msg, db)
            if plan["use_web"] and DecisionEngine.evaluate(plan): contexto_global += await WebAgent.search(msg)

        contexto_global += IdentityAndWorldAgent.retrieve(db)
        contexto_global += await MemoryOrchestrator.retrieve(vector_msg, db, msg)

        razonamiento = await ReasonerAgent.think(msg, contexto_global)
        logger.info(f"🧠 [Reasoner CoT]: {razonamiento}")

        return {"contexto": contexto_global, "plan": plan, "razonamiento": razonamiento}

@app.post("/chat")
async def chat(request: ChatRequest, db: Session = Depends(get_db)):
    msg = request.mensaje
    fecha_actual_sistema = request.fecha_hora or datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    
    try: vector_msg = get_embeddings().encode([msg])[0].tolist()
    except Exception: vector_msg = np.zeros(384).tolist()

    orchestration_data = await CognitiveOrchestrator.orchestrate(msg, vector_msg, request, db)
    idioma_usuario = LanguageManager.detect(msg)
    
    respuesta_ia = await SynthesizerLLM.generate(
        msg=msg, 
        contexto_global=orchestration_data["contexto"], 
        db=db, 
        fecha_hora=fecha_actual_sistema,
        razonamiento_interno=orchestration_data["razonamiento"], 
        idioma_detectado=idioma_usuario
    )

    # Entrenar la red neuronal según el feedback del usuario (Reward System)
    RewardSystem.process_feedback(msg, db, vector_msg, orchestration_data['plan']['clase_original'])

    try:
        db.add(EpisodicMemory(timestamp=str(datetime.datetime.now()), autor="Usuario", texto=msg, modo=request.modo))
        db.add(EpisodicMemory(timestamp=str(datetime.datetime.now()), autor="Lumin AI", texto=respuesta_ia, modo=request.modo))
        db.commit()
        
        if qdrant_client_async:
            vector_respuesta = get_embeddings().encode([respuesta_ia])[0].tolist()
            await asyncio.gather(
                qdrant_client_async.upsert(collection_name="episodic_memories", points=[qmodels.PointStruct(id=str(uuid.uuid4()), vector=vector_msg, payload={"role": "user", "content": msg, "timestamp": datetime.datetime.now().timestamp()})]),
                qdrant_client_async.upsert(collection_name="episodic_memories", points=[qmodels.PointStruct(id=str(uuid.uuid4()), vector=vector_respuesta, payload={"role": "assistant", "content": respuesta_ia, "timestamp": datetime.datetime.now().timestamp()})])
            )
    except Exception as e: 
        logger.error(f"⚠️ Error reteniendo la memoria: {e}")
        db.rollback()
        
    return {"respuesta_ia": respuesta_ia}

# =====================================================================
# 👁️ VISIÓN BIOMÉTRICA (FACE MEMORY ENDPOINT)
# =====================================================================
@app.post("/vision/face")
async def endpoint_face_memory(nombre_persona: str, image: UploadFile = File(...)):
    """[UPGRADE INDUSTRIAL]: Añade un rostro a la colección biométrica Qdrant de 128d."""
    if not face_recognition or not cv2:
        return {"status": "error", "detalle": "Librerías C++ de biometría no instaladas en el servidor."}
    
    try:
        contenido = await image.read()
        np_img = np.frombuffer(contenido, np.uint8)
        img = cv2.imdecode(np_img, cv2.IMREAD_COLOR)
        rgb_img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
        
        encodings = face_recognition.face_encodings(rgb_img)
        if not encodings:
            return {"status": "error", "detalle": "No se detectó ningún rostro en la imagen."}
            
        vector_128d = encodings[0].tolist()
        
        if qdrant_client_async:
            await qdrant_client_async.upsert(
                collection_name="face_memory",
                points=[qmodels.PointStruct(id=str(uuid.uuid4()), vector=vector_128d, payload={"nombre": nombre_persona, "timestamp": datetime.datetime.now().timestamp()})]
            )
            return {"status": "success", "detalle": f"Rostro de {nombre_persona} memorizado con éxito."}
        return {"status": "error", "detalle": "Qdrant no está disponible."}
    except Exception as e:
        return {"status": "error", "detalle": f"Error biométrico: {str(e)}"}

@app.post("/stream-stt")
async def endpoint_stt_offline(audio: UploadFile = File(...)):
    global faster_whisper_model
    contenido = await audio.read()
    ruta_temporal = f"temp_audio_{datetime.datetime.now().timestamp()}.wav"
    try:
        with open(ruta_temporal, "wb") as f: f.write(contenido)
        from faster_whisper import WhisperModel
        if faster_whisper_model is None:
            faster_whisper_model = WhisperModel("small", device="cpu", compute_type="int8")
        segments, info = faster_whisper_model.transcribe(ruta_temporal, beam_size=5)
        return {"status": "success", "transcripcion": " ".join([s.text for s in segments]).strip(), "idioma": info.language}
    finally:
        if os.path.exists(ruta_temporal): os.remove(ruta_temporal)

@app.get("/tts")
async def endpoint_tts_offline(texto: str):
    ruta_salida = "output_speech.mp3"
    exito_azure = await MicrosoftAzureTTS.synthesize(texto, ruta_salida)
    if not exito_azure:
        from gtts import gTTS
        gTTS(text=texto, lang='es').save(ruta_salida)
    return FileResponse(ruta_salida, media_type="audio/mp3", filename="lumin_voice.mp3")

class MultimodalKnowledgeEngine:
    @staticmethod
    async def process_file(file: UploadFile) -> str:
        try:
            contenido = await file.read()
            ext = file.filename.split(".")[-1].lower()
            texto_extraido = ""
            if ext == "pdf":
                reader = PdfReader(io.BytesIO(contenido))
                texto_extraido = "\n".join([page.extract_text() for page in reader.pages if page.extract_text()])
            elif ext == "docx":
                doc = docx.Document(io.BytesIO(contenido))
                texto_extraido = "\n".join([p.text for p in doc.paragraphs])
            elif ext in ["txt", "csv", "json"]:
                texto_extraido = contenido.decode("utf-8", errors="ignore")
            elif ext in ["png", "jpg", "jpeg", "bmp", "tiff"]:
                image = Image.open(io.BytesIO(contenido))
                texto_extraido = pytesseract.image_to_string(image)
            else: return f"Formato .{ext} no soportado."

            if len(texto_extraido.strip()) < 5: return "Archivo vacío."

            chunks = [texto_extraido[i:i+400] for i in range(0, len(texto_extraido), 350)]
            for chunk in chunks:
                if len(chunk.strip()) > 10:
                    vector = get_embeddings().encode([chunk])[0].tolist()
                    with memoria_lock: chroma_collection.add(embeddings=[vector], documents=[chunk], metadatas=[{"source": file.filename, "type": ext}], ids=[f"rag_{ext}_{datetime.datetime.now().timestamp()}_{np.random.randint(1000)}"])
            return f"Conocimiento inyectado: {len(chunks)} vectores."
        except Exception as e:
            raise HTTPException(status_code=500, detail=str(e))

@app.post("/rag/ingesta-multimodal")
async def endpoint_rag_multimodal(file: UploadFile = File(...)):
    resultado = await MultimodalKnowledgeEngine.process_file(file)
    return {"status": "success", "detalle": resultado}

@app.get("/")
async def root(): return {"status": "online", "asistente": "Lumin AGI OS", "estado": "Operativo"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=7860)