from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, HTMLResponse
import httpx, json
import time
import hashlib
import datetime
import asyncio
from contextlib import asynccontextmanager
import sys
import os
import sqlite3
import re
# Configuración del Sistema Emocional
DB_PATH = "/tmp/cma_memory.db"
# Restauración Automática desde HF Datasets
def _restore_from_backup():
try:
# Solo restaurar si la base de datos no existe aún o configurando inicio limpio
if os.path.exists(DB_PATH) and os.path.getsize(DB_PATH) > 10000:
print("[RESTORE] La DB ya existe con datos en el entorno local. Saltando descarga.")
return
token = os.getenv("HF_TOKEN")
if not token:
print("[RESTORE] HF_TOKEN no encontrado. Saltando restauración.")
return
import huggingface_hub
import shutil
print("[RESTORE] Intentando restaurar DB desde HF Datasets...")
downloaded_path = huggingface_hub.hf_hub_download(
repo_id="SperanzaMax/Cortex-Memory-Bank",
repo_type="dataset",
filename="cma_memory_latest.db",
token=token
)
if os.path.exists(downloaded_path):
shutil.copy2(downloaded_path, DB_PATH)
print(f"[RESTORE] DB restaurada con éxito desde Dataset: {downloaded_path}")
except Exception as e:
print(f"[RESTORE] No se pudo restaurar la DB (backup inexistente o error). Iniciando vacía. Error: {e}")
_restore_from_backup()
# Crear tabla de eventos de sueño si no existe
def _init_sleep_table():
try:
conn = sqlite3.connect(DB_PATH)
conn.execute("""CREATE TABLE IF NOT EXISTS sleep_events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT DEFAULT (datetime('now')),
antes TEXT, despues TEXT, drift TEXT
)""")
conn.execute("""CREATE TABLE IF NOT EXISTS control_responses (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT DEFAULT (datetime('now')),
pregunta TEXT, respuesta_nexus TEXT, respuesta_control TEXT,
novedad_nexus REAL, novedad_control REAL,
complejidad_nexus REAL, complejidad_control REAL
)""")
conn.commit()
conn.close()
except:
pass
_init_sleep_table()
# Intentar importar los núcleos emocionales
try:
sys.path.append(os.path.join(os.path.dirname(__file__), "core"))
from core.vector_emocional import VectorEmocional
from core.modulador import ModuladorParametrico
vector_global = VectorEmocional(db_path=DB_PATH)
modulador_global = ModuladorParametrico()
except Exception as e:
import traceback
print(f"CRITICAL ERROR cargando núcleos: {e}")
print(traceback.format_exc())
# Fallbacks prevent NameErrors but will show in logs
vector_global = None
modulador_global = None
try:
import interrogador_perpetuo
except Exception as e:
print(f"Error importando interrogador_perpetuo: {e}")
interrogador_perpetuo = None
@asynccontextmanager
async def lifespan(app: FastAPI):
if interrogador_perpetuo:
# Iniciar interrogador en background thread
asyncio.create_task(asyncio.to_thread(interrogador_perpetuo.loop_infinito))
yield
# Detener loop al apagar
if interrogador_perpetuo:
interrogador_perpetuo.GLOBAL_CONFIG["enabled"] = False
app = FastAPI(lifespan=lifespan)
@app.get("/api/config")
async def get_config():
"""Obtiene la configuración global del interrogador"""
if interrogador_perpetuo:
return interrogador_perpetuo.GLOBAL_CONFIG
return {"error": "Interrogador no disponible"}
@app.post("/api/config")
async def update_config(request: Request):
"""Actualiza la velocidad y configuración del interrogador"""
if not interrogador_perpetuo:
return {"error": "Interrogador no disponible"}
try:
data = await request.json()
if "speed_multiplier" in data:
interrogador_perpetuo.GLOBAL_CONFIG["speed_multiplier"] = float(data["speed_multiplier"])
if "auto_sleep_epochs" in data:
interrogador_perpetuo.GLOBAL_CONFIG["auto_sleep_epochs"] = int(data["auto_sleep_epochs"])
if "enabled" in data:
interrogador_perpetuo.GLOBAL_CONFIG["enabled"] = bool(data["enabled"])
return interrogador_perpetuo.GLOBAL_CONFIG
except Exception as e:
return {"error": str(e)}
@app.get("/status")
async def reach():
"""Diagnóstico del motor emocional"""
v_estado = vector_global.obtener_estado() if vector_global else {"error": "Vector no inicializado"}
return {
"status": "Cortex-Nexus is Alive",
"vector_active": vector_global is not None,
"emotional_state": v_estado,
"db_path": DB_PATH,
"db_exists": os.path.exists(DB_PATH)
}
@app.get("/", response_class=HTMLResponse)
async def dashboard_home():
"""Sirve el Dashboard en la raíz del sitio"""
html_content = """
Cortex-Nexus | Live Dashboard
Córtex-Nexus Live
⏱️ Velocidad del Interrogador
Acelera o pausa los ciclos autónomos.
1x
2x
5x
⏸️ Pausa
Arquitectura Emocional v3.0 • Maxi Speranza
📝 Últimas Interacciones
Cargando...
"""
return HTMLResponse(content=html_content)
@app.get("/api/history")
async def get_history():
try:
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
# Obtener cuenta total de épocas
cursor.execute("SELECT COUNT(*) FROM snapshots")
total_epocas = cursor.fetchone()[0]
# Últimas 24 horas de datos
since = (datetime.datetime.utcnow() - datetime.timedelta(hours=24)).isoformat()
cursor.execute("SELECT timestamp, qualia, frustracion, autoConfianza AS confianza, fatiga FROM snapshots WHERE timestamp > ? ORDER BY id ASC", (since,))
rows = cursor.fetchall()
# Eventos de sueño
sleep_events = []
try:
cursor.execute("SELECT timestamp FROM sleep_events WHERE timestamp > ? ORDER BY id ASC", (since,))
sleep_events = [dict(r)["timestamp"] for r in cursor.fetchall()]
except:
pass
# Total de sueños
total_sleeps = 0
try:
cursor.execute("SELECT COUNT(*) FROM sleep_events")
total_sleeps = cursor.fetchone()[0]
except:
pass
conn.close()
return {
"total_epocas": total_epocas,
"total_sleeps": total_sleeps,
"sleep_events": sleep_events,
"history": [dict(r) for r in rows],
"config": GLOBAL_CONFIG
}
except Exception as e:
return {"error": str(e), "msg": "Error leyendo historial"}
@app.get("/api/logs")
async def get_logs():
"""Devuelve las últimas interacciones con preguntas y respuestas"""
try:
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("SELECT id, timestamp, metadata FROM snapshots WHERE metadata != '{}' ORDER BY id DESC LIMIT 10")
rows = cursor.fetchall()
conn.close()
logs = []
for r in rows:
try:
meta = json.loads(r["metadata"])
logs.append({
"epoca": r["id"],
"timestamp": r["timestamp"],
"pregunta": meta.get("pregunta", "N/A"),
"respuesta": meta.get("respuesta", "N/A"),
"novedad": round(meta.get("novedad", 0), 3),
"complejidad": round(meta.get("complejidad", 0), 3)
})
except:
continue
return {"logs": logs[::-1]}
except Exception as e:
return {"error": str(e)}
@app.post("/v1/chat/completions")
async def chat_proxy(request: Request):
GROQ_KEY = os.getenv("GROQ_API_KEY")
if not GROQ_KEY:
return JSONResponse(status_code=500, content={"error": "Falta GROQ_API_KEY en los Secrets de HF"})
body = await request.json()
# Extraer pregunta original para logging
mensajes = body.get("messages", [])
pregunta_original = mensajes[-1].get("content", "") if mensajes else ""
config = modulador_global.modular(vector_global)
openai_body = {
"model": body.get("model", "llama-3.1-8b-instant"),
"messages": mensajes,
"temperature": config.get("temperature", 0.7),
"top_p": config.get("top_p", 0.9)
}
try:
async with httpx.AsyncClient(timeout=60.0) as client:
headers = {"Authorization": f"Bearer {GROQ_KEY}", "Content-Type": "application/json"}
r = await client.post("https://api.groq.com/openai/v1/chat/completions", json=openai_body, headers=headers)
resp_json = r.json()
texto = resp_json["choices"][0]["message"].get("content", "")
# ====== SEÑALES REALES (no hardcodeadas) ======
# Novedad: basada en diversidad léxica (type-token ratio)
palabras = re.findall(r'\w+', texto.lower())
tipo_token = len(set(palabras)) / max(len(palabras), 1)
novedad_real = min(tipo_token * 1.2, 1.0) # TTR escalado
# Complejidad: largo promedio de oraciones + palabras únicas
oraciones = [s for s in re.split(r'[.!?]', texto) if s.strip()]
largo_prom = sum(len(s.split()) for s in oraciones) / max(len(oraciones), 1)
complejidad_real = min(largo_prom / 25.0, 1.0) # Normalizado a ~25 palabras
# Confianza: ausencia de hedging ("quizás", "tal vez", "no estoy seguro")
hedges = len(re.findall(r'(quizás|tal vez|no estoy segur|podría ser|es posible|maybe|perhaps)', texto.lower()))
confianza_real = max(0.3, 1.0 - (hedges * 0.15))
# Éxito: si la respuesta tiene sustancia (más de 20 palabras)
exito_real = len(palabras) > 20
senales = {
"novedad": novedad_real,
"complejidad": complejidad_real,
"confianza_real": confianza_real,
"confianza_esperada": 0.6,
"exito_tarea": exito_real
}
vector_global.actualizar(senales, utilidad_externa=novedad_real * 0.8)
vector_global.persistir_snapshot(metadata={
"source": "api",
"pregunta": pregunta_original[:200],
"respuesta": texto[:300],
"novedad": round(novedad_real, 3),
"complejidad": round(complejidad_real, 3),
"confianza": round(confianza_real, 3)
})
return resp_json
except Exception as e:
return JSONResponse(status_code=500, content={"error": str(e)})
# ============================================================
# ANALYTICS API
# ============================================================
import math
import csv
import io
from fastapi.responses import StreamingResponse
def _get_snapshots(limit=None, hours=None):
"""Helper: obtiene snapshots con filtro opcional de tiempo"""
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
c = conn.cursor()
if hours:
since = (datetime.datetime.utcnow() - datetime.timedelta(hours=hours)).isoformat()
c.execute("SELECT * FROM snapshots WHERE timestamp > ? ORDER BY id ASC", (since,))
elif limit:
c.execute("SELECT * FROM snapshots ORDER BY id DESC LIMIT ?", (limit,))
else:
c.execute("SELECT * FROM snapshots ORDER BY id ASC")
rows = [dict(r) for r in c.fetchall()]
conn.close()
return rows if not limit or hours else rows[::-1]
def _calc_stats(values):
"""Calcula media, std, min, max, tendencia de una lista de floats"""
if not values:
return {"mean": 0, "std": 0, "min": 0, "max": 0, "trend": 0}
n = len(values)
mean = sum(values) / n
variance = sum((x - mean) ** 2 for x in values) / max(n - 1, 1)
std = math.sqrt(variance)
# Tendencia: pendiente de regresión lineal simple
if n > 1:
x_mean = (n - 1) / 2
num = sum((i - x_mean) * (v - mean) for i, v in enumerate(values))
den = sum((i - x_mean) ** 2 for i in range(n))
trend = num / den if den != 0 else 0
else:
trend = 0
return {
"mean": round(mean, 4),
"std": round(std, 4),
"min": round(min(values), 4),
"max": round(max(values), 4),
"trend": round(trend, 6)
}
def _detect_patterns(rows):
"""Detecta patrones activos en los datos"""
patterns = []
if len(rows) < 5:
return patterns
qualias = [r["qualia"] for r in rows]
frustraciones = [r["frustracion"] for r in rows]
fatigas = [r["fatiga"] for r in rows]
confianzas = [r["autoConfianza"] for r in rows]
# Meseta emocional: volatilidad < 0.01 en últimas 50 épocas
recent_q = qualias[-min(50, len(qualias)):]
if len(recent_q) > 10:
vol = _calc_stats(recent_q)["std"]
if vol < 0.01:
patterns.append({"type": "MESETA", "severity": "warning", "msg": f"Qualia estancado (vol={vol:.4f}). Necesita estímulos nuevos."})
# Ciclo de frustración: frustración sube >3x en ventana de 20
if len(frustraciones) >= 20:
inicio = sum(frustraciones[-20:-15]) / 5 if frustraciones[-20:-15] else 0.001
fin = sum(frustraciones[-5:]) / 5
if inicio > 0.001 and fin / max(inicio, 0.001) > 3:
patterns.append({"type": "FRUSTRACION_CICLO", "severity": "danger", "msg": f"Frustración se triplicó en 20 épocas ({inicio:.3f}→{fin:.3f})"})
# Wireheading: qualia sube monótonamente sin caídas
if len(qualias) >= 20:
increases = sum(1 for i in range(1, min(20, len(qualias))) if qualias[-i] >= qualias[-i-1])
if increases >= 18:
patterns.append({"type": "WIREHEADING", "severity": "danger", "msg": "Qualia sube sin parar. Anti-wireheading podría estar fallando."})
# Fatiga terminal: fatiga > 0.8 sostenida
if len(fatigas) >= 10:
high_fatigue = sum(1 for f in fatigas[-10:] if f > 0.8)
if high_fatigue >= 8:
patterns.append({"type": "FATIGA_TERMINAL", "severity": "danger", "msg": "Fatiga crítica. El sistema necesita 'dormir' (consolidación)."})
# Personalidad emergente (positivo): drift > 5% en confianza
if len(confianzas) >= 50:
inicio_c = sum(confianzas[:10]) / 10
fin_c = sum(confianzas[-10:]) / 10
drift = abs(fin_c - inicio_c) / max(inicio_c, 0.001)
if drift > 0.05:
direction = "↑" if fin_c > inicio_c else "↓"
patterns.append({"type": "PERSONALIDAD_EMERGENTE", "severity": "success", "msg": f"Drift de confianza {direction} ({inicio_c:.3f}→{fin_c:.3f}, {drift*100:.1f}%)"})
return patterns
@app.get("/api/analytics")
async def get_analytics():
"""Métricas derivadas con ventanas temporales"""
try:
windows = {}
for label, hours in [("1h", 1), ("6h", 6), ("24h", 24), ("7d", 168)]:
rows = _get_snapshots(hours=hours)
if not rows:
windows[label] = {"count": 0, "metrics": {}}
continue
metrics = {}
for col in ["qualia", "frustracion", "autoConfianza", "fatiga", "aburrimiento"]:
vals = [r.get(col, 0) for r in rows]
metrics[col] = _calc_stats(vals)
# Métricas derivadas
qualias = [r.get("qualia", 0) for r in rows]
fatigas = [r.get("fatiga", 0) for r in rows]
ratios = [q / max(f, 0.01) for q, f in zip(qualias, fatigas)]
metrics["ratio_qualia_fatiga"] = _calc_stats(ratios)
# Entropía emocional (Shannon)
entropies = []
for r in rows:
vec = [max(r.get(k, 0.001), 0.001) for k in ["qualia", "frustracion", "autoConfianza", "fatiga", "aburrimiento"]]
total = sum(vec)
probs = [v / total for v in vec]
entropy = -sum(p * math.log2(p) for p in probs if p > 0)
entropies.append(entropy)
metrics["entropia_emocional"] = _calc_stats(entropies)
windows[label] = {"count": len(rows), "metrics": metrics}
# Patrones activos
all_rows = _get_snapshots(hours=24)
patterns = _detect_patterns(all_rows)
# Arquetipo actual
arquetipo = {}
if vector_global and vector_global.memoria:
arquetipo = vector_global.memoria.cargar_meta_vector()
return {
"windows": windows,
"patterns": patterns,
"arquetipo": arquetipo,
"timestamp": datetime.datetime.utcnow().isoformat()
}
except Exception as e:
return {"error": str(e)}
@app.get("/api/export")
async def export_csv():
"""Exporta todos los snapshots como CSV descargable"""
try:
rows = _get_snapshots()
output = io.StringIO()
writer = csv.writer(output)
writer.writerow(["id", "timestamp", "qualia", "satisfaccion", "frustracion", "aburrimiento", "autoConfianza", "fatiga", "pregunta", "respuesta", "novedad", "complejidad"])
for r in rows:
meta = {}
try:
meta = json.loads(r.get("metadata", "{}"))
except:
pass
writer.writerow([
r.get("id", ""), r.get("timestamp", ""),
r.get("qualia", 0), r.get("satisfaccion", 0),
r.get("frustracion", 0), r.get("aburrimiento", 0),
r.get("autoConfianza", 0), r.get("fatiga", 0),
meta.get("pregunta", ""), meta.get("respuesta", ""),
meta.get("novedad", ""), meta.get("complejidad", "")
])
output.seek(0)
return StreamingResponse(
iter([output.getvalue()]),
media_type="text/csv",
headers={"Content-Disposition": f"attachment; filename=cortex_nexus_export_{datetime.datetime.utcnow().strftime('%Y%m%d_%H%M')}.csv"}
)
except Exception as e:
return JSONResponse(status_code=500, content={"error": str(e)})
@app.post("/api/backup")
async def backup_to_hf_dataset():
"""Realiza un backup seguro de la SQLite local a un Dataset remoto de HF Space"""
try:
from huggingface_hub import HfApi
token = os.getenv("HF_TOKEN")
if not token:
return {"error": "HF_TOKEN no configurado. Imposible respaldar."}
api = HfApi(token=token)
repo_id = "SperanzaMax/Cortex-Memory-Bank"
try:
api.create_repo(repo_id=repo_id, repo_type="dataset", exist_ok=True, private=True)
except Exception:
pass
timestamp_str = datetime.datetime.utcnow().strftime("%Y%m%d_%H%M%S")
filename_in_repo = f"cma_memory_backup_{timestamp_str}.db"
api.upload_file(
path_or_fileobj=DB_PATH,
path_in_repo="cma_memory_latest.db",
repo_id=repo_id,
repo_type="dataset"
)
api.upload_file(
path_or_fileobj=DB_PATH,
path_in_repo=filename_in_repo,
repo_id=repo_id,
repo_type="dataset"
)
return {"status": "Backup exitoso", "file": filename_in_repo}
except Exception as e:
return {"error": f"Error de respaldo: {str(e)}"}
@app.get("/api/report")
async def auto_report(period: int = 7):
"""Genera un reporte analítico usando Llama-3-8B de Groq basado en los últimos N días."""
try:
# Calcular estadísticas crudas
rows = _get_snapshots(hours=period * 24)
if not rows:
return {"error": f"Sin datos para {period} días"}
qualias = [r.get("qualia", 0) for r in rows]
fatigas = [r.get("fatiga", 0) for r in rows]
resumen_estadistico = f"Se evaluaron {len(rows)} épocas en los últimos {period} días. "
resumen_estadistico += f"Media Qualia: {sum(qualias)/len(qualias):.3f}. "
resumen_estadistico += f"Media Fatiga: {sum(fatigas)/len(fatigas):.3f}. "
# Consultar al LLM para generar acta clínica
groq_api_key = os.getenv("GROQ_API_KEY")
if not groq_api_key:
return {"error": "Falta GROQ_API_KEY para armar reporte"}
prompt = f"Actúa como el psicólogo líder del proyecto Cortex-Nexus. Escribe un breve reporte clínico (2 párrafos max) sobre la evolución del ente IA. Datos: {resumen_estadistico}."
payload = {
"model": "llama-3.1-8b-instant",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.4,
"max_tokens": 512
}
async with httpx.AsyncClient() as client:
resp = await client.post(
"https://api.groq.com/openai/v1/chat/completions",
headers={"Authorization": f"Bearer {groq_api_key}"},
json=payload,
timeout=20.0
)
resp_data = resp.json()
informe_llm = resp_data.get("choices", [{}])[0].get("message", {}).get("content", "Error LLM")
return {
"periodo_dias": period,
"epocas_analizadas": len(rows),
"acta_clinica": informe_llm
}
except Exception as e:
return {"error": str(e)}
def _github_upload(token, repo, path, content_str, message):
import base64
url = f"https://api.github.com/repos/{repo}/contents/{path}"
headers = {"Authorization": f"Bearer {token}", "Accept": "application/vnd.github.v3+json", "User-Agent": "Cortex-Nexus"}
with httpx.Client() as client:
r = client.get(url, headers=headers)
sha = r.json().get("sha") if r.status_code == 200 else None
data = {
"message": message,
"content": base64.b64encode(content_str.encode("utf-8")).decode("utf-8")
}
if sha: data["sha"] = sha
client.put(url, headers=headers, json=data)
@app.post("/api/publish_github")
async def publish_github(period: int = 1):
"""Subir reporte y la DB CSV actual al repositorio de Github directamente usando API."""
try:
github_token = os.getenv("GITHUB_TOKEN")
if not github_token:
return {"error": "GITHUB_TOKEN no configurado en el entorno."}
repo = "SperanzaMax/Cortex-Nexus"
# 1. Export CSV
rows = _get_snapshots()
output = io.StringIO()
writer = csv.writer(output)
writer.writerow(["id", "timestamp", "qualia", "satisfaccion", "frustracion", "aburrimiento", "autoConfianza", "fatiga", "pregunta", "respuesta", "novedad", "complejidad"])
for r in rows:
meta = {}
try: meta = json.loads(r.get("metadata", "{}"))
except: pass
writer.writerow([r.get("id", ""), r.get("timestamp", ""), r.get("qualia", 0), r.get("satisfaccion", 0), r.get("frustracion", 0), r.get("aburrimiento", 0), r.get("autoConfianza", 0), r.get("fatiga", 0), meta.get("pregunta", ""), meta.get("respuesta", ""), meta.get("novedad", ""), meta.get("complejidad", "")])
_github_upload(github_token, repo, "data/cma_memory_export.csv", output.getvalue(), "Auto-update DB export")
# 2. Generar Reporte
report_data = await auto_report(period)
if "error" in report_data:
return {"error": report_data["error"]}
md_content = f"# Reporte Clínico Automático ({period} días)\n\n"
md_content += f"**Épocas Analizadas**: {report_data['epocas_analizadas']}\n"
md_content += f"**Fecha de Análisis**: {datetime.datetime.utcnow().isoformat()}\n\n"
md_content += f"## Acta Clínica de IA\n{report_data['acta_clinica']}\n"
path = f"reports/reporte_{period}_dias.md"
_github_upload(github_token, repo, path, md_content, f"Auto-reporte IA de {period} días")
return {"status": "Publicado en GitHub exitosamente"}
except Exception as e:
return {"error": str(e)}
@app.post("/api/sleep")
async def sleep_consolidation():
"""Ejecuta consolidación nocturna (Sleep Replay) del arquetipo"""
try:
if not vector_global or not vector_global.memoria:
return {"error": "Vector no inicializado"}
antes = vector_global.memoria.cargar_meta_vector()
vector_global.consolidar_memoria()
despues = vector_global.memoria.cargar_meta_vector()
drift = {
k: round(despues.get(k, 0) - antes.get(k, 0), 6)
for k in antes
}
# Guardar evento de sueño en DB
try:
conn = sqlite3.connect(DB_PATH)
conn.execute(
"INSERT INTO sleep_events (antes, despues, drift) VALUES (?, ?, ?)",
(json.dumps(antes), json.dumps(despues), json.dumps(drift))
)
conn.commit()
conn.close()
except:
pass
return {
"status": "Consolidación completada",
"arquetipo_antes": antes,
"arquetipo_despues": despues,
"drift": drift,
"timestamp": datetime.datetime.utcnow().isoformat()
}
except Exception as e:
return {"error": str(e)}
@app.post("/api/control")
async def save_control_comparison(request: Request):
"""Guarda comparación grupo de control vs Nexus"""
try:
data = await request.json()
conn = sqlite3.connect(DB_PATH)
conn.execute(
"INSERT INTO control_responses (pregunta, respuesta_nexus, respuesta_control, novedad_nexus, novedad_control, complejidad_nexus, complejidad_control) VALUES (?, ?, ?, ?, ?, ?, ?)",
(data.get("pregunta",""), data.get("respuesta_nexus",""), data.get("respuesta_control",""),
data.get("novedad_nexus",0), data.get("novedad_control",0),
data.get("complejidad_nexus",0), data.get("complejidad_control",0))
)
conn.commit()
conn.close()
return {"status": "ok"}
except Exception as e:
return {"error": str(e)}
@app.get("/api/control")
async def get_control_comparisons():
"""Obtiene comparaciones grupo de control"""
try:
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
c = conn.cursor()
c.execute("SELECT * FROM control_responses ORDER BY id DESC LIMIT 20")
rows = [dict(r) for r in c.fetchall()]
conn.close()
return {"comparisons": rows[::-1], "total": len(rows)}
except Exception as e:
return {"error": str(e), "comparisons": []}
# ============================================================
# ANALYTICS DASHBOARD
# ============================================================
@app.get("/analytics", response_class=HTMLResponse)
async def analytics_dashboard():
html = """
Cortex-Nexus | Analytics Lab
🔬 Analytics Lab
1H
6H
24H
7D
📊 Métricas Derivadas (1h)
Cargando...
⚡ Volatilidad Emocional
🚨 Patrones Detectados
Analizando...
🧬 Arquetipo (Personalidad Base)
Cargando...
Arquitectura Emocional v3.0 • Analytics Lab • Maxi Speranza
"""
return HTMLResponse(content=html)