515 lines
20 KiB
Python
515 lines
20 KiB
Python
"""
|
|
===============================================================
|
|
SISTEMA DE NOTIFICACIÓN PRIVADA DE RECOLECCIÓN DE RESIDUOS
|
|
Backend MVP - Hackathon 24h
|
|
===============================================================
|
|
Stack: FastAPI + SQLite (SQLAlchemy) + Firebase Admin SDK
|
|
|
|
CÓMO CORRER:
|
|
pip install fastapi uvicorn sqlalchemy firebase-admin
|
|
uvicorn main:app --reload --port 8000
|
|
|
|
ATAJO DE HACKATHON:
|
|
Usamos SQLite para cero configuración. En producción
|
|
cambiaría a PostgreSQL solo cambiando DATABASE_URL.
|
|
===============================================================
|
|
"""
|
|
|
|
from fastapi import FastAPI, HTTPException, Depends
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
|
|
from sqlalchemy.orm import declarative_base, sessionmaker, Session, relationship
|
|
from pydantic import BaseModel
|
|
from typing import Optional
|
|
import logging
|
|
|
|
# ---------------------------------------------------------------
|
|
# CONFIGURACIÓN DE LOGGING
|
|
# Útil para ver en consola qué está pasando sin un debugger real
|
|
# ---------------------------------------------------------------
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ---------------------------------------------------------------
|
|
# CONFIGURACIÓN DE BASE DE DATOS (SQLite - Hackathon mode)
|
|
#
|
|
# DATABASE_URL apunta a un archivo local "hackathon.db".
|
|
# check_same_thread=False es NECESARIO para FastAPI porque
|
|
# maneja requests en múltiples threads con el mismo engine.
|
|
# ---------------------------------------------------------------
|
|
DATABASE_URL = "sqlite:///./hackathon.db"
|
|
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
|
|
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)
|
|
Base = declarative_base()
|
|
|
|
|
|
# ===============================================================
|
|
# MODELOS DE BASE DE DATOS (SQLAlchemy ORM)
|
|
# ===============================================================
|
|
|
|
class Usuario(Base):
|
|
"""
|
|
Tabla de usuarios del sistema.
|
|
|
|
fcm_token: Token de Firebase Cloud Messaging. Cada dispositivo
|
|
móvil genera uno único. Es lo que necesitamos para enviar
|
|
notificaciones push. Se guarda aquí al hacer login en la app.
|
|
|
|
ATAJO: No hay autenticación real (JWT, OAuth). Para el hackathon
|
|
el usuario_id es suficiente. En producción: agrega auth.
|
|
"""
|
|
__tablename__ = "usuarios"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
nombre = Column(String, nullable=False)
|
|
# Token FCM que Flutter registrará al iniciar la app
|
|
fcm_token = Column(String, nullable=True)
|
|
|
|
# Relación 1-a-1 con Domicilio (un usuario, un domicilio registrado)
|
|
domicilio = relationship("Domicilio", back_populates="usuario", uselist=False)
|
|
|
|
|
|
class Domicilio(Base):
|
|
"""
|
|
Tabla de domicilios asociados a cada usuario.
|
|
|
|
colonia: Nombre de la colonia, es la clave para buscar
|
|
en nuestros datos mockeados y obtener el route_id correspondiente.
|
|
|
|
PRIVACIDAD POR DISEÑO: El route_id se guarda internamente
|
|
pero NUNCA se expone en los endpoints públicos. El usuario
|
|
solo ve su ETA, no la ruta completa del camión.
|
|
"""
|
|
__tablename__ = "domicilios"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
usuario_id = Column(Integer, ForeignKey("usuarios.id"), unique=True)
|
|
colonia = Column(String, nullable=False)
|
|
# route_id es dato interno - no lo devolvemos al cliente
|
|
route_id = Column(String, nullable=False)
|
|
|
|
usuario = relationship("Usuario", back_populates="domicilio")
|
|
|
|
|
|
# Crea las tablas en hackathon.db si no existen
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
|
|
# ===============================================================
|
|
# DATOS MOCKEADOS EN MEMORIA
|
|
#
|
|
# ATAJO DE HACKATHON: Evitamos una tabla extra de "Rutas" y
|
|
# "Horarios" usando simples diccionarios. Esto nos ahorra 2h de
|
|
# desarrollo. En producción, estos datos vendrían de la DB.
|
|
# ===============================================================
|
|
|
|
# Mapeo: Nombre de Colonia -> ID de Ruta interna
|
|
# El Flutter usa estas colonias para el Dropdown del login
|
|
COLONIAS_A_RUTAS: dict[str, str] = {
|
|
"Zona Centro": "RUTA-01",
|
|
"Col. Hidalgo": "RUTA-01",
|
|
"Col. Independencia":"RUTA-02",
|
|
"Col. Obrera": "RUTA-02",
|
|
"Col. San Juan": "RUTA-03",
|
|
"Fracc. Los Pinos": "RUTA-03",
|
|
"Col. Reforma": "RUTA-04",
|
|
}
|
|
|
|
# Horarios estimados por ruta (ETA en texto amigable para el usuario)
|
|
# Formato: { route_id: { "eta_texto": str, "eta_minutos": int } }
|
|
HORARIOS_POR_RUTA: dict[str, dict] = {
|
|
"RUTA-01": {"eta_texto": "Llega en aproximadamente 15 minutos", "eta_minutos": 15},
|
|
"RUTA-02": {"eta_texto": "Llega en aproximadamente 30 minutos", "eta_minutos": 30},
|
|
"RUTA-03": {"eta_texto": "Llega en aproximadamente 45 minutos", "eta_minutos": 45},
|
|
"RUTA-04": {"eta_texto": "Llega en aproximadamente 60 minutos", "eta_minutos": 60},
|
|
}
|
|
|
|
# Tipos de evento válidos para el simulador
|
|
TIPOS_EVENTO_VALIDOS = ["en_camino", "llegando", "completado", "retrasado"]
|
|
|
|
|
|
# ===============================================================
|
|
# CONFIGURACIÓN FIREBASE ADMIN SDK
|
|
#
|
|
# CÓMO ACTIVARLO:
|
|
# 1. Ve a Firebase Console -> Configuración del Proyecto ->
|
|
# Cuentas de Servicio -> Generar nueva clave privada
|
|
# 2. Guarda el JSON como "firebase-credentials.json" junto a main.py
|
|
# 3. Descomenta el bloque de inicialización de abajo
|
|
# ===============================================================
|
|
import firebase_admin
|
|
from firebase_admin import credentials, messaging
|
|
|
|
# --- DESCOMENTA ESTO CUANDO TENGAS EL ARCHIVO DE CREDENCIALES ---
|
|
# try:
|
|
# cred = credentials.Certificate("firebase-credentials.json")
|
|
# firebase_admin.initialize_app(cred)
|
|
# logger.info("✅ Firebase Admin SDK inicializado correctamente")
|
|
# except Exception as e:
|
|
# logger.error(f"❌ Error inicializando Firebase: {e}")
|
|
# ---------------------------------------------------------------
|
|
|
|
FIREBASE_ACTIVO = False # Cambia a True al desbloquear Firebase
|
|
|
|
|
|
# ===============================================================
|
|
# INICIALIZACIÓN DE FASTAPI
|
|
# ===============================================================
|
|
app = FastAPI(
|
|
title="Sistema de Notificación de Residuos - MVP Hackathon",
|
|
description="API privada para notificaciones de recolección de basura",
|
|
version="0.1.0-hackathon"
|
|
)
|
|
|
|
# CORS abierto para desarrollo. En producción: restringe origins.
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------
|
|
# DEPENDENCIA: Sesión de Base de Datos
|
|
#
|
|
# FastAPI usa Dependency Injection. Esta función provee una sesión
|
|
# de DB a cada endpoint y garantiza que se cierre al terminar,
|
|
# sin importar si hubo error. Es el patrón estándar de FastAPI+SQLAlchemy.
|
|
# ---------------------------------------------------------------
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# ===============================================================
|
|
# SCHEMAS PYDANTIC (Validación de request/response)
|
|
# Pydantic valida automáticamente los tipos. FastAPI los usa para
|
|
# generar la documentación en /docs sin esfuerzo adicional.
|
|
# ===============================================================
|
|
|
|
class ETAResponse(BaseModel):
|
|
"""Respuesta del endpoint de ETA. Sin route_id por privacidad."""
|
|
usuario_id: int
|
|
colonia: str
|
|
eta_texto: str
|
|
eta_minutos: int
|
|
mensaje_preventivo: str # Ej: "No saques tu basura aún"
|
|
|
|
|
|
class SimularEventoRequest(BaseModel):
|
|
"""
|
|
Payload para simular un evento de ruta.
|
|
route_id: ID interno de la ruta (RUTA-01, RUTA-02...)
|
|
tipo_evento: Tipo de evento a simular
|
|
"""
|
|
route_id: str
|
|
tipo_evento: str # "en_camino" | "llegando" | "completado" | "retrasado"
|
|
|
|
|
|
class SimularEventoResponse(BaseModel):
|
|
"""Respuesta del simulador de eventos."""
|
|
usuarios_notificados: int
|
|
route_id: str
|
|
tipo_evento: str
|
|
detalle: list[str] # Log de qué pasó con cada usuario
|
|
|
|
|
|
# ===============================================================
|
|
# UTILIDADES INTERNAS
|
|
# ===============================================================
|
|
|
|
def generar_mensaje_preventivo(eta_minutos: int) -> str:
|
|
"""
|
|
Genera un mensaje contextual basado en el tiempo de llegada.
|
|
|
|
La MENSAJERÍA PREVENTIVA es el core del sistema:
|
|
evitamos que el usuario saque la basura demasiado pronto
|
|
o demasiado tarde, mejorando la experiencia y la higiene.
|
|
"""
|
|
if eta_minutos <= 5:
|
|
return "🚛 ¡El camión está muy cerca! Saca tu basura AHORA."
|
|
elif eta_minutos <= 15:
|
|
return "⏰ Prepárate, el camión llega pronto. No saques tu basura aún."
|
|
elif eta_minutos <= 30:
|
|
return "🕐 Tienes tiempo. No saques tu basura todavía."
|
|
else:
|
|
return "😌 Aún falta bastante. Mantén tu basura adentro por ahora."
|
|
|
|
|
|
def enviar_notificacion_firebase(fcm_token: str, titulo: str, cuerpo: str) -> bool:
|
|
"""
|
|
Envía una notificación push via Firebase Cloud Messaging.
|
|
|
|
ATAJO: La función existe y está lista, pero si FIREBASE_ACTIVO=False
|
|
solo simula el envío en los logs. Esto nos permite desarrollar
|
|
el flujo completo sin credenciales reales.
|
|
|
|
Retorna True si el envío fue exitoso (o simulado), False si falló.
|
|
"""
|
|
if not FIREBASE_ACTIVO:
|
|
# Modo simulación: log del intento sin llamar a Firebase
|
|
logger.info(f"[SIMULADO] Push -> Token: {fcm_token[:20]}... | {titulo}: {cuerpo}")
|
|
return True
|
|
|
|
# --- CÓDIGO REAL DE FIREBASE (desbloquear cuando FIREBASE_ACTIVO=True) ---
|
|
try:
|
|
message = messaging.Message(
|
|
notification=messaging.Notification(title=titulo, body=cuerpo),
|
|
token=fcm_token,
|
|
)
|
|
response = messaging.send(message)
|
|
logger.info(f"✅ Notificación enviada: {response}")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"❌ Error enviando push a {fcm_token[:20]}...: {e}")
|
|
return False
|
|
|
|
|
|
# ===============================================================
|
|
# ENDPOINT: SEED DE DATOS DE PRUEBA
|
|
#
|
|
# Crea usuarios de prueba para poder testear sin un frontend.
|
|
# Llama: POST /api/seed
|
|
# ATAJO: En producción, eliminar este endpoint.
|
|
# ===============================================================
|
|
@app.post("/api/seed", tags=["Utilidades"])
|
|
def seed_datos(db: Session = Depends(get_db)):
|
|
"""
|
|
Crea usuarios de prueba en la DB para demos rápidas.
|
|
Idempotente: si los usuarios ya existen, no hace nada.
|
|
"""
|
|
# Verifica si ya hay datos para no duplicar
|
|
if db.query(Usuario).count() > 0:
|
|
return {"mensaje": "Ya hay datos en la DB. No se hizo nada."}
|
|
|
|
usuarios_seed = [
|
|
{"nombre": "Ana García", "colonia": "Zona Centro", "fcm_token": "token-ana-fake-001"},
|
|
{"nombre": "Carlos López", "colonia": "Col. Hidalgo", "fcm_token": "token-carlos-fake-002"},
|
|
{"nombre": "María Torres", "colonia": "Col. Independencia", "fcm_token": "token-maria-fake-003"},
|
|
{"nombre": "Pedro Ruiz", "colonia": "Col. San Juan", "fcm_token": "token-pedro-fake-004"},
|
|
]
|
|
|
|
for u in usuarios_seed:
|
|
colonia = u["colonia"]
|
|
route_id = COLONIAS_A_RUTAS.get(colonia, "RUTA-01")
|
|
|
|
usuario = Usuario(nombre=u["nombre"], fcm_token=u["fcm_token"])
|
|
db.add(usuario)
|
|
db.flush() # flush para obtener el id antes del commit
|
|
|
|
domicilio = Domicilio(usuario_id=usuario.id, colonia=colonia, route_id=route_id)
|
|
db.add(domicilio)
|
|
|
|
db.commit()
|
|
logger.info("✅ Seed completado: 4 usuarios creados")
|
|
return {"mensaje": "Seed exitoso. Usuarios IDs: 1, 2, 3, 4"}
|
|
|
|
|
|
# ===============================================================
|
|
# ENDPOINT 1: GET /api/eta/{usuario_id}
|
|
#
|
|
# Consulta el ETA de la ruta asignada al domicilio del usuario.
|
|
#
|
|
# PRIVACIDAD POR DISEÑO:
|
|
# - El usuario_id es la única info que el cliente manda
|
|
# - El route_id se resuelve internamente, NUNCA se devuelve
|
|
# - El cliente ve el ETA y el mensaje, no la infraestructura
|
|
# ===============================================================
|
|
@app.get("/api/eta/{usuario_id}", response_model=ETAResponse, tags=["Core"])
|
|
def obtener_eta(usuario_id: int, db: Session = Depends(get_db)):
|
|
"""
|
|
Devuelve el tiempo estimado de llegada del camión para el usuario.
|
|
|
|
Flujo:
|
|
1. Busca el usuario en la DB
|
|
2. Obtiene su domicilio (colonia + route_id interno)
|
|
3. Consulta el horario de esa ruta en los datos mockeados
|
|
4. Retorna ETA + mensaje preventivo SIN exponer el route_id
|
|
"""
|
|
# Paso 1: Verificar que el usuario existe
|
|
usuario = db.query(Usuario).filter(Usuario.id == usuario_id).first()
|
|
if not usuario:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"Usuario {usuario_id} no encontrado. ¿Corriste /api/seed?"
|
|
)
|
|
|
|
# Paso 2: Verificar que tiene domicilio registrado
|
|
if not usuario.domicilio:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"El usuario {usuario_id} no tiene domicilio registrado."
|
|
)
|
|
|
|
colonia = usuario.domicilio.colonia
|
|
route_id = usuario.domicilio.route_id # Uso INTERNO, no se devuelve
|
|
|
|
# Paso 3: Buscar el horario de la ruta en los datos mockeados
|
|
horario = HORARIOS_POR_RUTA.get(route_id)
|
|
if not horario:
|
|
# Fallback gracioso: si la ruta no tiene horario, decimos que no hay info
|
|
raise HTTPException(
|
|
status_code=503,
|
|
detail="No hay información de horario disponible para esta zona."
|
|
)
|
|
|
|
# Paso 4: Construir respuesta con mensaje preventivo
|
|
eta_minutos = horario["eta_minutos"]
|
|
return ETAResponse(
|
|
usuario_id=usuario_id,
|
|
colonia=colonia,
|
|
eta_texto=horario["eta_texto"],
|
|
eta_minutos=eta_minutos,
|
|
mensaje_preventivo=generar_mensaje_preventivo(eta_minutos),
|
|
# NOTA: route_id NO está en ETAResponse -> privacidad garantizada
|
|
)
|
|
|
|
|
|
# ===============================================================
|
|
# ENDPOINT 2: POST /api/simular-evento
|
|
#
|
|
# Simula que un camión generó un evento (ej. "llegando") y
|
|
# dispara notificaciones push a todos los usuarios de esa ruta.
|
|
#
|
|
# En un sistema real, este endpoint sería llamado por el GPS
|
|
# del camión o un sistema de despacho, no por el usuario.
|
|
# ===============================================================
|
|
@app.post("/api/simular-evento", response_model=SimularEventoResponse, tags=["Core"])
|
|
def simular_evento(payload: SimularEventoRequest, db: Session = Depends(get_db)):
|
|
"""
|
|
Recibe un evento de ruta y notifica a todos sus usuarios.
|
|
|
|
Flujo:
|
|
1. Valida el tipo de evento y que la ruta exista
|
|
2. Busca todos los Domicilios asignados a esa ruta
|
|
3. Para cada domicilio -> obtiene el usuario -> envía push
|
|
4. Retorna un log de lo que pasó con cada usuario
|
|
"""
|
|
# Validación del tipo de evento
|
|
if payload.tipo_evento not in TIPOS_EVENTO_VALIDOS:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"tipo_evento inválido. Opciones: {TIPOS_EVENTO_VALIDOS}"
|
|
)
|
|
|
|
# Validar que la ruta existe en nuestros datos
|
|
if payload.route_id not in HORARIOS_POR_RUTA:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"Ruta {payload.route_id} no encontrada. Rutas válidas: {list(HORARIOS_POR_RUTA.keys())}"
|
|
)
|
|
|
|
# Buscar todos los domicilios asignados a esta ruta
|
|
domicilios = db.query(Domicilio).filter(Domicilio.route_id == payload.route_id).all()
|
|
|
|
if not domicilios:
|
|
return SimularEventoResponse(
|
|
usuarios_notificados=0,
|
|
route_id=payload.route_id,
|
|
tipo_evento=payload.tipo_evento,
|
|
detalle=["No hay usuarios registrados en esta ruta."]
|
|
)
|
|
|
|
# Construir el mensaje según el tipo de evento
|
|
mensajes_por_evento = {
|
|
"en_camino": ("🚛 Camión en camino", "El camión de recolección está en ruta hacia tu zona."),
|
|
"llegando": ("⚠️ ¡El camión está cerca!", "Saca tu basura ahora, el camión llega en minutos."),
|
|
"completado": ("✅ Recolección completada", "El camión ya pasó por tu zona. Nos vemos mañana."),
|
|
"retrasado": ("🕐 Retraso en ruta", "El camión se ha retrasado. Te avisaremos cuando esté cerca."),
|
|
}
|
|
titulo, cuerpo = mensajes_por_evento[payload.tipo_evento]
|
|
|
|
# Enviar notificación a cada usuario de la ruta
|
|
detalle_log = []
|
|
usuarios_notificados = 0
|
|
|
|
for domicilio in domicilios:
|
|
usuario = domicilio.usuario
|
|
|
|
if not usuario:
|
|
detalle_log.append(f"Domicilio ID {domicilio.id}: Sin usuario asociado (dato corrupto).")
|
|
continue
|
|
|
|
if not usuario.fcm_token:
|
|
# Sin token no hay push. En producción: guardar en cola para reintentar
|
|
detalle_log.append(f"Usuario '{usuario.nombre}' (ID {usuario.id}): Sin FCM token. Push omitido.")
|
|
continue
|
|
|
|
# Intentar enviar la notificación (real o simulada)
|
|
exito = enviar_notificacion_firebase(usuario.fcm_token, titulo, cuerpo)
|
|
|
|
if exito:
|
|
usuarios_notificados += 1
|
|
detalle_log.append(f"✅ Push enviado a '{usuario.nombre}' (ID {usuario.id}) en {domicilio.colonia}.")
|
|
else:
|
|
detalle_log.append(f"❌ Fallo push para '{usuario.nombre}' (ID {usuario.id}).")
|
|
|
|
logger.info(f"Evento '{payload.tipo_evento}' en {payload.route_id}: {usuarios_notificados}/{len(domicilios)} notificados.")
|
|
|
|
return SimularEventoResponse(
|
|
usuarios_notificados=usuarios_notificados,
|
|
route_id=payload.route_id,
|
|
tipo_evento=payload.tipo_evento,
|
|
detalle=detalle_log
|
|
)
|
|
|
|
|
|
# ===============================================================
|
|
# ENDPOINT: GET /api/colonias
|
|
#
|
|
# Devuelve la lista de colonias disponibles para el Dropdown
|
|
# del login en Flutter. Simple y directo.
|
|
# ===============================================================
|
|
@app.get("/api/colonias", tags=["Utilidades"])
|
|
def listar_colonias():
|
|
"""
|
|
Lista todas las colonias disponibles.
|
|
Flutter las usa para poblar el Dropdown del login screen.
|
|
"""
|
|
return {"colonias": list(COLONIAS_A_RUTAS.keys())}
|
|
|
|
|
|
# ===============================================================
|
|
# ENDPOINT: PUT /api/usuarios/{usuario_id}/fcm-token
|
|
#
|
|
# Flutter llama este endpoint al iniciar la app para registrar
|
|
# o actualizar el FCM token del dispositivo.
|
|
# ===============================================================
|
|
class ActualizarTokenRequest(BaseModel):
|
|
fcm_token: str
|
|
|
|
@app.put("/api/usuarios/{usuario_id}/fcm-token", tags=["Utilidades"])
|
|
def actualizar_fcm_token(
|
|
usuario_id: int,
|
|
payload: ActualizarTokenRequest,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
Actualiza el FCM token de un usuario.
|
|
|
|
Flutter llama esto cuando:
|
|
- El usuario inicia sesión por primera vez
|
|
- Firebase renueva el token del dispositivo (pasa periódicamente)
|
|
"""
|
|
usuario = db.query(Usuario).filter(Usuario.id == usuario_id).first()
|
|
if not usuario:
|
|
raise HTTPException(status_code=404, detail="Usuario no encontrado.")
|
|
|
|
usuario.fcm_token = payload.fcm_token
|
|
db.commit()
|
|
logger.info(f"FCM token actualizado para usuario {usuario_id}")
|
|
return {"mensaje": f"Token actualizado para usuario {usuario_id}"}
|
|
|
|
|
|
# ---------------------------------------------------------------
|
|
# PUNTO DE ENTRADA PARA DESARROLLO DIRECTO
|
|
# Corre con: python main.py (o preferiblemente: uvicorn main:app --reload)
|
|
# ---------------------------------------------------------------
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)
|