Administrator/dashboard-leads
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
f68926f07815a334bb2370499f6cf223ebea0ea8
dashboard-leads/backend/main.py

722 lines
27 KiB
Python
Raw Blame History

# backend/main.py
"""API REST del Dashboard de LEADS (FastAPI)."""
from fastapi import FastAPI, Query, HTTPException, Body
from fastapi.middleware.cors import CORSMiddleware
from datetime import datetime
import uvicorn
import services
from cache_manager import start_background_refresh, cache_stats
app = FastAPI(title="Dashboard Leads API", version="1.0")
app.add_middleware(
CORSMiddleware, allow_origins=["*"], allow_credentials=True,
allow_methods=["*"], allow_headers=["*"],
)
@app.on_event("startup")
def _startup():
# Precarga en SEGUNDO PLANO (hilo aparte) para no bloquear el arranque del server.
import threading
def _precarga_bg():
print("[startup] Precargando datos de Leads (2do plano)...")
try:
services.precargar_todo()
print("[startup] Precarga completa.")
except Exception as e:
print(f"[startup] Precarga falló: {e}")
threading.Thread(target=_precarga_bg, daemon=True).start()
# Otros General en un hilo aparte (sus consultas son mas pesadas)
def _precarga_otros_bg():
try:
services.precargar_otros_general()
except Exception as e:
print(f"[startup] Precarga Otros falló: {e}")
threading.Thread(target=_precarga_otros_bg, daemon=True).start()
start_background_refresh(services.refrescar_todo, interval=1800) # 30 min (con horario)
@app.get("/api/health")
def health():
return {"status": "ok", "hora": datetime.now().isoformat()}
@app.get("/api/cache/stats")
def stats():
return cache_stats()
@app.post("/api/cache/refresh")
def refresh():
services.refrescar_todo()
return {"status": "refrescado"}
@app.get("/api/leads/filtros")
def get_filtros():
try:
return services.opciones_filtros()
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/leads")
def get_leads(
ano: str = Query("TODOS"),
mes: str = Query("TODOS"),
dia: str = Query("TODOS"),
programa: str = Query("TODOS"),
sede: str = Query("TODOS"),
):
try:
return services.leads_dashboard(ano, mes, dia, programa, sede)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/vendedores")
def get_vendedores(
ano: str = Query("TODOS"),
mes: str = Query("TODOS"),
dia: str = Query("TODOS"),
):
try:
return services.vendedores_dashboard(ano, mes, dia)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/tiempo-respuesta/detalle")
def get_tiempo_respuesta_detalle(
dia: int = Query(...),
modo: str = Query("asignacion"),
vendedor: str = Query("TODOS"),
ano: str = Query("TODOS"),
mes: str = Query("TODOS"),
):
try:
return services.tiempo_respuesta_detalle(dia, modo, vendedor, ano, mes)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/programa/opciones")
def get_programa_opciones(ano: str = Query("TODOS"), mes: str = Query("TODOS")):
try:
return {"programas": services.programa_opciones(ano, mes)}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/programa")
def get_programa(
ano: str = Query("TODOS"),
mes: str = Query("TODOS"),
programa: str = Query("TODOS"),
):
try:
return services.programa_dashboard(ano, mes, programa)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/alertas")
def get_alertas_dash():
"""Cursos proximos a iniciar (<=30d) clasificados critico/advertencia/buen rumbo."""
try:
return services.alertas_dashboard()
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/plantillas")
def get_plantillas():
"""Lista las plantillas de WhatsApp del inbox de Chatwoot (nombre + categoría)."""
import os, requests
CHATWOOT_URL = os.getenv("CHATWOOT_URL", "https://gestor.escueladerefrigeracion.edu.pe")
TOKEN = os.getenv("CHATWOOT_TOKEN", "4anazHvZnvKLtup8biu5Zuoh")
ACCOUNT_ID = os.getenv("CHATWOOT_ACCOUNT_ID", "1")
INBOX_ID = os.getenv("CHATWOOT_INBOX_ID", "8")
headers = {"api_access_token": TOKEN, "Content-Type": "application/json"}
try:
r = requests.get(f"{CHATWOOT_URL}/api/v1/accounts/{ACCOUNT_ID}/inboxes/{INBOX_ID}",
headers=headers, timeout=15)
r.raise_for_status()
inbox = r.json()
tpls = inbox.get("message_templates") or []
out = []
for t in tpls:
# El texto está en el componente tipo BODY (components[].text).
texto = ""
for comp in (t.get("components") or []):
if str(comp.get("type", "")).upper() == "BODY":
texto = comp.get("text", "") or ""
break
out.append({
"nombre": t.get("name", ""),
"categoria": t.get("category", ""),
"texto": texto,
})
out.sort(key=lambda x: x["nombre"])
return {"plantillas": out}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/plantillas/enviar")
def post_plantillas_enviar(payload: dict = Body(...)):
"""Envía una plantilla de WhatsApp (Chatwoot) a una lista de números.
payload: {nombre, categoria, idioma, texto, telefonos:[...]}.
Solo se envía a números peruanos con código 51 (formato 51XXXXXXXXX)."""
import os, requests, time
CHATWOOT_URL = os.getenv("CHATWOOT_URL", "https://gestor.escueladerefrigeracion.edu.pe")
TOKEN = os.getenv("CHATWOOT_TOKEN", "4anazHvZnvKLtup8biu5Zuoh")
ACCOUNT_ID = os.getenv("CHATWOOT_ACCOUNT_ID", "1")
INBOX_ID = int(os.getenv("CHATWOOT_INBOX_ID", "8"))
headers = {"api_access_token": TOKEN, "Content-Type": "application/json"}
nombre = payload.get("nombre", "")
categoria = payload.get("categoria", "MARKETING")
idioma = payload.get("idioma", "es_PE")
texto = payload.get("texto", "")
telefonos_raw = payload.get("telefonos", []) or []
def ok(s): return s in (200, 201)
def normaliza(v):
s = str(v or "").strip()
if s.endswith(".0"): s = s[:-2]
for x in ("+", " ", "-", "(", ")"): s = s.replace(x, "")
return s
# Se envía a CUALQUIER país. Solo se exige que sea un número (dígitos).
# NO se crean contactos nuevos: si no existe en Chatwoot, se salta y se reporta.
validos, invalidos = [], []
for t in telefonos_raw:
s = normaliza(t)
if s.isdigit() and len(s) >= 8:
validos.append(s)
else:
invalidos.append(str(t))
# Números que NO existen como contacto en Chatwoot (no se crean, solo se reportan)
no_encontrados = []
def chat_id(tel):
tel_p = "+" + tel
try:
r = requests.get(f"{CHATWOOT_URL}/api/v1/accounts/{ACCOUNT_ID}/contacts/search",
headers=headers, params={"q": tel_p}, timeout=20)
data = r.json()
except Exception:
return None
# Si NO existe el contacto → no crear, marcar como no encontrado
if not (ok(r.status_code) and data.get("payload")):
no_encontrados.append(tel)
return None
cid = data["payload"][0]["id"]
rconv = requests.get(f"{CHATWOOT_URL}/api/v1/accounts/{ACCOUNT_ID}/contacts/{cid}/conversations",
headers=headers, timeout=20)
for conv in rconv.json().get("payload", []):
if conv.get("inbox_id") == INBOX_ID:
return conv["id"]
rn = requests.post(f"{CHATWOOT_URL}/api/v1/accounts/{ACCOUNT_ID}/conversations",
headers=headers, json={"source_id": str(cid), "inbox_id": INBOX_ID, "contact_id": cid}, timeout=20)
return rn.json().get("id") if ok(rn.status_code) else None
enviados, fallidos = 0, []
for tel in validos:
cid = chat_id(tel)
if not cid:
fallidos.append(tel); continue
body = {"content": texto, "private": False,
"template_params": {"name": nombre, "category": categoria,
"language": idioma, "processed_params": {}}}
try:
r = requests.post(f"{CHATWOOT_URL}/api/v1/accounts/{ACCOUNT_ID}/conversations/{cid}/messages",
headers=headers, json=body, timeout=30)
if ok(r.status_code):
enviados += 1
else:
fallidos.append(tel)
except Exception:
fallidos.append(tel)
time.sleep(1)
# "fallidos" incluye los que no existen (no_encontrados) + los que sí existían
# pero falló el envío. Se separan para poder avisar cuáles no están en la base.
fallo_envio = [t for t in fallidos if t not in set(no_encontrados)]
return {"status": "ok", "enviados": enviados,
"fallidos": fallidos, "invalidos": invalidos,
"no_encontrados": no_encontrados, "fallo_envio": fallo_envio,
"total": len(telefonos_raw)}
@app.get("/api/plantillas/campanias")
def get_plantillas_campanias():
"""Lista las campañas (código + frase) para el checklist."""
try:
return services.plantillas_campanias()
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/plantillas/asesores")
def get_plantillas_asesores():
"""Lista los asesores (para el checklist de filtro)."""
try:
return services.plantillas_asesores()
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/plantillas/estados")
def get_plantillas_estados():
"""Lista los estados/objeción (para el checklist de filtro)."""
try:
return services.plantillas_estados()
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/plantillas/base")
def post_plantillas_base(payload: dict = Body(...)):
"""Devuelve la base de números (teléfono + 2 fechas + asesor + estado) de los
códigos elegidos, filtrando por asesores ('__VACIO__' = sin asesor) y estados."""
try:
return services.plantillas_base_por_codigos(
payload.get("codigos", []), payload.get("asesores"), payload.get("estados"))
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# ── SUBIR BASE: etiquetas, asesores y envío masivo con verificación ────────
def _cw_cfg():
import os
return {
"URL": os.getenv("CHATWOOT_URL", "https://gestor.escueladerefrigeracion.edu.pe"),
"TOKEN": os.getenv("CHATWOOT_TOKEN", "4anazHvZnvKLtup8biu5Zuoh"),
"ACCOUNT_ID": os.getenv("CHATWOOT_ACCOUNT_ID", "1"),
"INBOX_ID": int(os.getenv("CHATWOOT_INBOX_ID", "8")),
}
@app.get("/api/subir-base/etiquetas")
def get_subir_base_etiquetas():
"""Lista las etiquetas (labels) de Chatwoot, EN VIVO (sin caché)."""
import requests
c = _cw_cfg()
h = {"api_access_token": c["TOKEN"], "Content-Type": "application/json"}
try:
r = requests.get(f"{c['URL']}/api/v1/accounts/{c['ACCOUNT_ID']}/labels", headers=h, timeout=15)
r.raise_for_status()
data = r.json().get("payload", r.json())
etiquetas = [x.get("title") for x in (data or []) if x.get("title")]
return {"etiquetas": sorted(etiquetas)}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/subir-base/asesores")
def get_subir_base_asesores():
"""Lista los asesores (agents) de Chatwoot con su id y nombre, EN VIVO."""
import requests
c = _cw_cfg()
h = {"api_access_token": c["TOKEN"], "Content-Type": "application/json"}
try:
r = requests.get(f"{c['URL']}/api/v1/accounts/{c['ACCOUNT_ID']}/agents", headers=h, timeout=15)
r.raise_for_status()
ags = [{"id": a.get("id"), "nombre": a.get("name") or a.get("email")}
for a in (r.json() or []) if a.get("id")]
ags.sort(key=lambda x: str(x["nombre"]).upper())
return {"asesores": ags}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/subir-base/enviar")
def post_subir_base_enviar(payload: dict = Body(...)):
"""Sube una base de números: crea contacto (o usa el existente), le asigna
asesor y etiquetas, le envía la plantilla, y tras unos segundos verifica el
estado. Si WhatsApp rechazó el número (error 131009 → 'Error al enviar'), BORRA
la conversación y el contacto. Los que se enviaron bien se quedan con todo.
payload: {nombre, categoria, idioma, texto, telefonos:[], etiquetas:[],
asesor_id (int o null para SIN ASIGNACIÓN)}."""
import requests, time
c = _cw_cfg()
URL, ACC, INBOX = c["URL"], c["ACCOUNT_ID"], c["INBOX_ID"]
h = {"api_access_token": c["TOKEN"], "Content-Type": "application/json"}
def ok(s): return s in (200, 201)
nombre = payload.get("nombre", "")
categoria = payload.get("categoria", "MARKETING")
idioma = payload.get("idioma", "es_PE")
texto = payload.get("texto", "")
etiquetas = [str(x).strip() for x in (payload.get("etiquetas") or []) if str(x).strip()]
asesor_id = payload.get("asesor_id") # None/"" = SIN ASIGNACIÓN
tels_raw = payload.get("telefonos", []) or []
def norm(v):
s = str(v or "").strip()
if s.endswith(".0"): s = s[:-2]
for x in ("+", " ", "-", "(", ")"): s = s.replace(x, "")
return s
validos, invalidos = [], []
for t in tels_raw:
s = norm(t)
if s.isdigit() and len(s) >= 8:
validos.append(s)
else:
invalidos.append(str(t))
enviados, borrados, fallo = [], [], []
seguimiento = [] # [{tel, contact_id, conv_id, msg_id, creado_contacto}]
for tel in validos:
tel_p = "+" + tel
# 1) Buscar contacto; si no existe, crearlo
creado = False
contact_id = None
try:
rb = requests.get(f"{URL}/api/v1/accounts/{ACC}/contacts/search",
headers=h, params={"q": tel_p}, timeout=20)
pay = rb.json().get("payload", []) if ok(rb.status_code) else []
if pay:
contact_id = pay[0]["id"]
except Exception:
pass
if not contact_id:
try:
# 1) Crear el contacto (sin inbox todavía).
rc = requests.post(f"{URL}/api/v1/accounts/{ACC}/contacts", headers=h,
json={"name": "Contacto " + tel, "phone_number": tel_p}, timeout=20)
jc = rc.json()
contact_id = (jc.get("payload", {}).get("contact", {}) or {}).get("id") or jc.get("id")
creado = True
print(f"[subir-base] {tel}: POST contacto status={rc.status_code} contact_id={contact_id}")
except Exception as e:
print(f"[subir-base] {tel}: error creando contacto: {e}")
fallo.append(tel); continue
if not contact_id:
fallo.append(tel); continue
# Vincular el contacto al inbox con source_id = número (igual que el SQL que
# inserta en contact_inboxes). Así WhatsApp reconoce el destino. Idempotente.
try:
requests.post(f"{URL}/api/v1/accounts/{ACC}/contacts/{contact_id}/contact_inboxes",
headers=h, json={"inbox_id": INBOX, "source_id": tel}, timeout=20)
except Exception as e:
print(f"[subir-base] {tel}: error vinculando inbox: {e}")
# 2) Conversación (usar la del inbox o crear)
conv_id = None
try:
rconv = requests.get(f"{URL}/api/v1/accounts/{ACC}/contacts/{contact_id}/conversations",
headers=h, timeout=20)
for cv in rconv.json().get("payload", []):
if cv.get("inbox_id") == INBOX:
conv_id = cv["id"]; break
except Exception:
pass
if not conv_id:
try:
# source_id = número de teléfono (así WhatsApp reconoce el destino).
# Usar el contact_id como source_id rompe el envío (131009).
rn = requests.post(f"{URL}/api/v1/accounts/{ACC}/conversations", headers=h,
json={"source_id": tel, "inbox_id": INBOX,
"contact_id": contact_id}, timeout=20)
conv_id = rn.json().get("id") if ok(rn.status_code) else None
print(f"[subir-base] {tel}: POST conv status={rn.status_code} conv_id={conv_id}")
except Exception as e:
print(f"[subir-base] {tel}: error creando conversacion: {e}")
if not conv_id:
if creado:
try: requests.delete(f"{URL}/api/v1/accounts/{ACC}/contacts/{contact_id}", headers=h, timeout=15)
except Exception: pass
fallo.append(tel); continue
# 3) Asignar asesor (si no es SIN ASIGNACIÓN) — DIRECTO en Postgres, para NO
# generar el evento "Asignado a ..." (que dispararía el panel de alertas).
if asesor_id:
try:
import services as _svc
_conn = _svc.get_dm().pg_conn() if hasattr(_svc, "get_dm") else None
except Exception:
_conn = None
if _conn is None:
# fallback: usar psycopg2 directo con los mismos env que el data_manager
try:
import psycopg2, os as _os
_conn = psycopg2.connect(
host=_os.getenv("PG_HOST", "191.98.134.81"),
dbname=_os.getenv("PG_DB", _os.getenv("PG_DATABASE", "chatwoot_production")),
user=_os.getenv("PG_USER", "postgres"),
password=_os.getenv("PG_PASS", _os.getenv("PG_PASSWORD", "")),
port=_os.getenv("PG_PORT", "5432"))
except Exception as e:
_conn = None
print(f"[subir-base] {tel}: sin conexion Postgres para asignar: {e}")
if _conn is not None:
try:
_cur = _conn.cursor()
_cur.execute("UPDATE conversations SET assignee_id=%s, updated_at=NOW() WHERE id=%s",
(int(asesor_id), int(conv_id)))
_conn.commit()
_cur.close(); _conn.close()
except Exception as e:
print(f"[subir-base] {tel}: error asignando en Postgres: {e}")
try: _conn.close()
except Exception: pass
# 4) Etiquetas (todas las elegidas)
if etiquetas:
try:
requests.post(f"{URL}/api/v1/accounts/{ACC}/conversations/{conv_id}/labels",
headers=h, json={"labels": etiquetas}, timeout=15)
except Exception:
pass
# 5) Enviar plantilla
msg_id = None
try:
body = {"content": texto, "private": False,
"template_params": {"name": nombre, "category": categoria,
"language": idioma, "processed_params": {}}}
rm = requests.post(f"{URL}/api/v1/accounts/{ACC}/conversations/{conv_id}/messages",
headers=h, json=body, timeout=30)
if ok(rm.status_code):
msg_id = rm.json().get("id")
print(f"[subir-base] {tel}: POST msg status={rm.status_code} msg_id={msg_id}")
except Exception as e:
print(f"[subir-base] {tel}: error enviando plantilla: {e}")
seguimiento.append({"tel": tel, "contact_id": contact_id, "conv_id": conv_id,
"msg_id": msg_id, "creado": creado})
time.sleep(0.4)
# 6) Esperar a que WhatsApp marque los estados (131009 tarda unos segundos)
time.sleep(22) # dar tiempo a WhatsApp a marcar el estado del mensaje
# 7) Revisar cada mensaje: SOLO borrar si está explícitamente "failed".
# Si no se encuentra el mensaje o el estado no es failed → NO se borra (se queda).
for s in seguimiento:
fallo_wa = False
encontrado = False
try:
rmsgs = requests.get(f"{URL}/api/v1/accounts/{ACC}/conversations/{s['conv_id']}/messages",
headers=h, timeout=20)
for m in rmsgs.json().get("payload", []):
if s["msg_id"] and m.get("id") == s["msg_id"]:
encontrado = True
st = str(m.get("status", "")).lower()
print(f"[subir-base] {s['tel']}: estado mensaje = '{st}'")
if st == "failed":
fallo_wa = True
break
if not encontrado:
print(f"[subir-base] {s['tel']}: mensaje NO encontrado (msg_id={s['msg_id']})")
except Exception as e:
print(f"[subir-base] {s['tel']}: error leyendo estado: {e}")
# Solo se borra si WhatsApp lo marcó como failed de verdad
if fallo_wa:
# Borrar conversación y contacto (número no existe)
try: requests.delete(f"{URL}/api/v1/accounts/{ACC}/conversations/{s['conv_id']}", headers=h, timeout=15)
except Exception: pass
try: requests.delete(f"{URL}/api/v1/accounts/{ACC}/contacts/{s['contact_id']}", headers=h, timeout=15)
except Exception: pass
borrados.append(s["tel"])
else:
enviados.append(s["tel"])
return {"status": "ok", "enviados": enviados, "borrados": borrados,
"fallo": fallo, "invalidos": invalidos, "total": len(tels_raw)}
@app.get("/api/roas")
def get_roas(
ano: str = Query("TODOS"),
mes: str = Query("TODOS"),
dia: str = Query("TODOS"),
):
try:
return services.roas_dashboard(ano, mes, dia)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/conjuntos-sin-sede")
def get_conjuntos_sin_sede():
try:
return services.conjuntos_sin_sede()
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/conjuntos-sin-sede/guardar")
def post_conjunto_sede(body: dict = Body(...)):
try:
cambios = body.get("cambios") or []
return services.guardar_conjunto_sede(cambios)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/conjuntos-sin-sede/borrar")
def post_borrar_conjunto_sede(body: dict = Body(...)):
try:
return services.borrar_conjunto_sede(body.get("conjunto") or "")
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/otros-general")
def get_otros_general(
ano: str = Query("TODOS"),
mes: str = Query("TODOS"),
dia: str = Query("TODOS"),
):
try:
return services.otros_general(ano, mes, dia)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/conjuntos-sin-pauta")
def get_conjuntos_sin_pauta():
try:
return {"conjuntos": services.conjuntos_sin_pauta()}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/programas-ocultos")
def get_programas_ocultos():
try:
return services.programas_ocultos()
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/programas-ocultos/encender")
def post_encender_programas(body: dict = Body(...)):
try:
nis = body.get("num_indices") or []
return services.encender_programas(nis)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/programas-disponibles")
def get_programas_disponibles():
try:
return services.programas_disponibles()
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/leyenda-anuncios")
def get_leyenda():
try:
return services.leyenda_anuncios()
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/leyenda-anuncios/guardar")
def post_leyenda(body: dict = Body(...)):
try:
cambios = body.get("cambios") or []
return services.guardar_leyenda(cambios)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# ── LEYENDA (campanias de Supabase): listar / agregar / borrar ──
@app.get("/api/campanias")
def get_campanias():
try:
return {"filas": services.campanias_listar()}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/campanias/agregar")
def post_campania(body: dict = Body(...)):
try:
return services.campania_agregar(body or {})
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/campanias/editar")
def post_editar_campania(body: dict = Body(...)):
try:
cid = body.get("id")
return services.campania_editar(cid, body or {})
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/campanias/borrar")
def post_borrar_campania(body: dict = Body(...)):
try:
return services.campania_borrar(body.get("id"))
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/pauta/uso")
def get_uso_pauta(pauta: str = Query(...), excluir: str = Query("")):
try:
return services.uso_de_pauta(pauta, excluir)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/curso/guardar-pauta")
def post_guardar_pauta(body: dict = Body(...)):
try:
ni = str(body.get("num_indice", "")).strip()
pa = str(body.get("pauta", "")).strip()
cj = body.get("conjunto") or None
contar = body.get("contar") # "SI" / "NO" / None
if not ni:
raise HTTPException(status_code=400, detail="num_indice es obligatorio")
return services.guardar_edicion_curso(ni, pa, cj, contar)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/ultima-actualizacion")
def get_ultima_actualizacion():
try:
return services.ultima_actualizacion()
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/leads/alertas")
def get_alertas():
"""Valores de programa/sede en cartera_junta que NO están en el diccionario."""
try:
return services.alertas()
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/debug/cursos")
def debug_cursos(ano: str = "2026", mes: str = "1"):
"""Diagnóstico: lista cursos del periodo con inscritos, tipo y meta."""
try:
return services.debug_cursos(ano, mes)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=8001, reload=False)
Reference in New Issue View Git Blame Copy Permalink