51d78bacf4
Sistema completo para monitoreo y gestion de flotas de vehiculos con: - Backend FastAPI con PostgreSQL/TimescaleDB - Frontend React con TypeScript y TailwindCSS - App movil React Native con Expo - Soporte para dispositivos GPS, Meshtastic y celulares - Video streaming en vivo con MediaMTX - Geocercas, alertas, viajes y reportes - Autenticacion JWT y WebSockets en tiempo real Documentacion completa y guias de usuario incluidas.
349 lines
12 KiB
Python
349 lines
12 KiB
Python
"""
|
|
Servicio para envío de notificaciones.
|
|
|
|
Maneja el envío de notificaciones por email, push y SMS.
|
|
"""
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from email.mime.multipart import MIMEMultipart
|
|
from email.mime.text import MIMEText
|
|
from typing import List, Optional
|
|
|
|
import aiosmtplib
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.config import settings
|
|
from app.models.alerta import Alerta
|
|
|
|
|
|
class NotificacionService:
|
|
"""Servicio para envío de notificaciones."""
|
|
|
|
def __init__(self, db: AsyncSession = None):
|
|
"""
|
|
Inicializa el servicio.
|
|
|
|
Args:
|
|
db: Sesión de base de datos async (opcional).
|
|
"""
|
|
self.db = db
|
|
|
|
async def enviar_notificacion_alerta(
|
|
self,
|
|
alerta: Alerta,
|
|
destinatarios_email: List[str] = None,
|
|
) -> dict:
|
|
"""
|
|
Envía notificaciones para una alerta.
|
|
|
|
Args:
|
|
alerta: Alerta a notificar.
|
|
destinatarios_email: Lista de emails (opcional, usa config si no se especifica).
|
|
|
|
Returns:
|
|
Resultado del envío.
|
|
"""
|
|
resultado = {
|
|
"email_enviado": False,
|
|
"push_enviado": False,
|
|
"sms_enviado": False,
|
|
}
|
|
|
|
# Determinar si enviar cada tipo de notificación
|
|
tipo_alerta = alerta.tipo_alerta
|
|
|
|
if tipo_alerta.notificar_email:
|
|
resultado["email_enviado"] = await self.enviar_email_alerta(
|
|
alerta,
|
|
destinatarios_email,
|
|
)
|
|
|
|
if tipo_alerta.notificar_push:
|
|
resultado["push_enviado"] = await self.enviar_push_alerta(alerta)
|
|
|
|
if tipo_alerta.notificar_sms:
|
|
resultado["sms_enviado"] = await self.enviar_sms_alerta(alerta)
|
|
|
|
# Actualizar estado de notificaciones en la alerta
|
|
if self.db:
|
|
alerta.notificacion_email_enviada = resultado["email_enviado"]
|
|
alerta.notificacion_push_enviada = resultado["push_enviado"]
|
|
alerta.notificacion_sms_enviada = resultado["sms_enviado"]
|
|
await self.db.commit()
|
|
|
|
return resultado
|
|
|
|
async def enviar_email_alerta(
|
|
self,
|
|
alerta: Alerta,
|
|
destinatarios: List[str] = None,
|
|
) -> bool:
|
|
"""
|
|
Envía notificación de alerta por email.
|
|
|
|
Args:
|
|
alerta: Alerta a notificar.
|
|
destinatarios: Lista de emails.
|
|
|
|
Returns:
|
|
True si se envió correctamente.
|
|
"""
|
|
if not settings.SMTP_HOST or not settings.SMTP_USER:
|
|
return False
|
|
|
|
destinatarios = destinatarios or [settings.SMTP_FROM_EMAIL]
|
|
|
|
# Crear mensaje
|
|
mensaje = MIMEMultipart("alternative")
|
|
mensaje["Subject"] = f"[{alerta.severidad.upper()}] {alerta.mensaje[:50]}"
|
|
mensaje["From"] = f"{settings.SMTP_FROM_NAME} <{settings.SMTP_FROM_EMAIL}>"
|
|
mensaje["To"] = ", ".join(destinatarios)
|
|
|
|
# Contenido HTML
|
|
html_content = self._crear_html_alerta(alerta)
|
|
mensaje.attach(MIMEText(html_content, "html"))
|
|
|
|
# Contenido texto plano
|
|
text_content = self._crear_texto_alerta(alerta)
|
|
mensaje.attach(MIMEText(text_content, "plain"))
|
|
|
|
try:
|
|
async with aiosmtplib.SMTP(
|
|
hostname=settings.SMTP_HOST,
|
|
port=settings.SMTP_PORT,
|
|
use_tls=settings.SMTP_TLS,
|
|
) as smtp:
|
|
await smtp.login(settings.SMTP_USER, settings.SMTP_PASSWORD)
|
|
await smtp.send_message(mensaje)
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"Error enviando email: {e}")
|
|
return False
|
|
|
|
async def enviar_push_alerta(
|
|
self,
|
|
alerta: Alerta,
|
|
) -> bool:
|
|
"""
|
|
Envía notificación push de alerta.
|
|
|
|
Args:
|
|
alerta: Alerta a notificar.
|
|
|
|
Returns:
|
|
True si se envió correctamente.
|
|
"""
|
|
if not settings.FIREBASE_ENABLED:
|
|
return False
|
|
|
|
# TODO: Implementar con Firebase Cloud Messaging
|
|
# from firebase_admin import messaging
|
|
#
|
|
# message = messaging.Message(
|
|
# notification=messaging.Notification(
|
|
# title=f"Alerta: {alerta.tipo_alerta.nombre}",
|
|
# body=alerta.mensaje,
|
|
# ),
|
|
# topic="alertas",
|
|
# )
|
|
# messaging.send(message)
|
|
|
|
return False
|
|
|
|
async def enviar_sms_alerta(
|
|
self,
|
|
alerta: Alerta,
|
|
) -> bool:
|
|
"""
|
|
Envía notificación SMS de alerta.
|
|
|
|
Args:
|
|
alerta: Alerta a notificar.
|
|
|
|
Returns:
|
|
True si se envió correctamente.
|
|
"""
|
|
# TODO: Implementar con Twilio u otro proveedor SMS
|
|
return False
|
|
|
|
async def enviar_email(
|
|
self,
|
|
destinatarios: List[str],
|
|
asunto: str,
|
|
contenido_html: str,
|
|
contenido_texto: str = None,
|
|
) -> bool:
|
|
"""
|
|
Envía un email genérico.
|
|
|
|
Args:
|
|
destinatarios: Lista de emails.
|
|
asunto: Asunto del email.
|
|
contenido_html: Contenido HTML.
|
|
contenido_texto: Contenido texto plano (opcional).
|
|
|
|
Returns:
|
|
True si se envió correctamente.
|
|
"""
|
|
if not settings.SMTP_HOST or not settings.SMTP_USER:
|
|
return False
|
|
|
|
mensaje = MIMEMultipart("alternative")
|
|
mensaje["Subject"] = asunto
|
|
mensaje["From"] = f"{settings.SMTP_FROM_NAME} <{settings.SMTP_FROM_EMAIL}>"
|
|
mensaje["To"] = ", ".join(destinatarios)
|
|
|
|
mensaje.attach(MIMEText(contenido_html, "html"))
|
|
if contenido_texto:
|
|
mensaje.attach(MIMEText(contenido_texto, "plain"))
|
|
|
|
try:
|
|
async with aiosmtplib.SMTP(
|
|
hostname=settings.SMTP_HOST,
|
|
port=settings.SMTP_PORT,
|
|
use_tls=settings.SMTP_TLS,
|
|
) as smtp:
|
|
await smtp.login(settings.SMTP_USER, settings.SMTP_PASSWORD)
|
|
await smtp.send_message(mensaje)
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"Error enviando email: {e}")
|
|
return False
|
|
|
|
def _crear_html_alerta(self, alerta: Alerta) -> str:
|
|
"""Crea el contenido HTML para el email de alerta."""
|
|
color_severidad = {
|
|
"baja": "#10B981",
|
|
"media": "#F59E0B",
|
|
"alta": "#EF4444",
|
|
"critica": "#DC2626",
|
|
}
|
|
|
|
color = color_severidad.get(alerta.severidad, "#6B7280")
|
|
|
|
html = f"""
|
|
<!DOCTYPE html>
|
|
<html>
|
|
<head>
|
|
<style>
|
|
body {{ font-family: Arial, sans-serif; line-height: 1.6; }}
|
|
.container {{ max-width: 600px; margin: 0 auto; padding: 20px; }}
|
|
.header {{ background-color: {color}; color: white; padding: 20px; border-radius: 8px 8px 0 0; }}
|
|
.content {{ background-color: #f9fafb; padding: 20px; border: 1px solid #e5e7eb; }}
|
|
.footer {{ padding: 10px; text-align: center; color: #6b7280; font-size: 12px; }}
|
|
.badge {{ display: inline-block; padding: 4px 8px; border-radius: 4px; font-size: 12px; }}
|
|
.info-row {{ margin: 10px 0; }}
|
|
.label {{ color: #6b7280; }}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<div class="container">
|
|
<div class="header">
|
|
<h2 style="margin: 0;">Alerta: {alerta.tipo_alerta.nombre if alerta.tipo_alerta else 'Sistema'}</h2>
|
|
<span class="badge" style="background-color: rgba(255,255,255,0.2);">
|
|
{alerta.severidad.upper()}
|
|
</span>
|
|
</div>
|
|
<div class="content">
|
|
<p><strong>{alerta.mensaje}</strong></p>
|
|
|
|
<div class="info-row">
|
|
<span class="label">Fecha/Hora:</span>
|
|
{alerta.creado_en.strftime('%Y-%m-%d %H:%M:%S')}
|
|
</div>
|
|
|
|
{'<div class="info-row"><span class="label">Vehiculo ID:</span> ' + str(alerta.vehiculo_id) + '</div>' if alerta.vehiculo_id else ''}
|
|
|
|
{'<div class="info-row"><span class="label">Ubicacion:</span> ' + str(alerta.lat) + ', ' + str(alerta.lng) + '</div>' if alerta.lat else ''}
|
|
|
|
{'<div class="info-row"><span class="label">Velocidad:</span> ' + str(alerta.velocidad) + ' km/h</div>' if alerta.velocidad else ''}
|
|
|
|
{f'<div class="info-row"><span class="label">Descripcion:</span> {alerta.descripcion}</div>' if alerta.descripcion else ''}
|
|
</div>
|
|
<div class="footer">
|
|
<p>Este es un mensaje automatico de {settings.APP_NAME}</p>
|
|
</div>
|
|
</div>
|
|
</body>
|
|
</html>
|
|
"""
|
|
return html
|
|
|
|
def _crear_texto_alerta(self, alerta: Alerta) -> str:
|
|
"""Crea el contenido de texto plano para el email de alerta."""
|
|
texto = f"""
|
|
ALERTA: {alerta.tipo_alerta.nombre if alerta.tipo_alerta else 'Sistema'}
|
|
Severidad: {alerta.severidad.upper()}
|
|
|
|
{alerta.mensaje}
|
|
|
|
Fecha/Hora: {alerta.creado_en.strftime('%Y-%m-%d %H:%M:%S')}
|
|
"""
|
|
if alerta.vehiculo_id:
|
|
texto += f"Vehiculo ID: {alerta.vehiculo_id}\n"
|
|
if alerta.lat and alerta.lng:
|
|
texto += f"Ubicacion: {alerta.lat}, {alerta.lng}\n"
|
|
if alerta.velocidad:
|
|
texto += f"Velocidad: {alerta.velocidad} km/h\n"
|
|
|
|
texto += f"\n--\nEste es un mensaje automatico de {settings.APP_NAME}"
|
|
return texto
|
|
|
|
async def enviar_recordatorio_mantenimiento(
|
|
self,
|
|
vehiculo_nombre: str,
|
|
vehiculo_placa: str,
|
|
tipo_mantenimiento: str,
|
|
fecha_programada: str,
|
|
destinatarios: List[str],
|
|
) -> bool:
|
|
"""
|
|
Envía recordatorio de mantenimiento por email.
|
|
|
|
Args:
|
|
vehiculo_nombre: Nombre del vehículo.
|
|
vehiculo_placa: Placa del vehículo.
|
|
tipo_mantenimiento: Tipo de mantenimiento.
|
|
fecha_programada: Fecha programada.
|
|
destinatarios: Lista de emails.
|
|
|
|
Returns:
|
|
True si se envió correctamente.
|
|
"""
|
|
asunto = f"Recordatorio: Mantenimiento próximo - {vehiculo_placa}"
|
|
|
|
html = f"""
|
|
<!DOCTYPE html>
|
|
<html>
|
|
<head>
|
|
<style>
|
|
body {{ font-family: Arial, sans-serif; line-height: 1.6; }}
|
|
.container {{ max-width: 600px; margin: 0 auto; padding: 20px; }}
|
|
.header {{ background-color: #3B82F6; color: white; padding: 20px; border-radius: 8px 8px 0 0; }}
|
|
.content {{ background-color: #f9fafb; padding: 20px; border: 1px solid #e5e7eb; }}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<div class="container">
|
|
<div class="header">
|
|
<h2>Recordatorio de Mantenimiento</h2>
|
|
</div>
|
|
<div class="content">
|
|
<p>Se aproxima la fecha de mantenimiento programado:</p>
|
|
<ul>
|
|
<li><strong>Vehiculo:</strong> {vehiculo_nombre} ({vehiculo_placa})</li>
|
|
<li><strong>Tipo:</strong> {tipo_mantenimiento}</li>
|
|
<li><strong>Fecha programada:</strong> {fecha_programada}</li>
|
|
</ul>
|
|
<p>Por favor, programe el mantenimiento con anticipacion.</p>
|
|
</div>
|
|
</div>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
return await self.enviar_email(destinatarios, asunto, html)
|