""" 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"""

Alerta: {alerta.tipo_alerta.nombre if alerta.tipo_alerta else 'Sistema'}

{alerta.severidad.upper()}

{alerta.mensaje}

Fecha/Hora: {alerta.creado_en.strftime('%Y-%m-%d %H:%M:%S')}
{'
Vehiculo ID: ' + str(alerta.vehiculo_id) + '
' if alerta.vehiculo_id else ''} {'
Ubicacion: ' + str(alerta.lat) + ', ' + str(alerta.lng) + '
' if alerta.lat else ''} {'
Velocidad: ' + str(alerta.velocidad) + ' km/h
' if alerta.velocidad else ''} {f'
Descripcion: {alerta.descripcion}
' if alerta.descripcion else ''}
""" 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"""

Recordatorio de Mantenimiento

Se aproxima la fecha de mantenimiento programado:

Por favor, programe el mantenimiento con anticipacion.

""" return await self.enviar_email(destinatarios, asunto, html)