from enum import Enum from typing import Optional import structlog from notifications.signals import notify from sendgrid import Mail, SendGridAPIClient from storages.utils import setting from vbv_lernwelt.core.models import User from vbv_lernwelt.notify.models import Notification, NotificationType logger = structlog.get_logger(__name__) class EmailTemplate(Enum): """Enum for the different Sendgrid email templates.""" # VBV - Erinnerung Präsenzkurse ATTENDANCE_COURSE_REMINDER = { "de": "d-9af079f98f524d85ac6e4166de3480da", "it": "d-ab78ddca8a7a46b8afe50aaba3efee81", "fr": "d-f88d9912e5484e55a879571463e4a166", } # VBV - Geleitete Fallarbeit abgegeben CASEWORK_SUBMITTED = {"de": "d-599f0b35ddcd4fac99314cdf8f5446a2"} # VBV - Geleitete Fallarbeit bewertet CASEWORK_EVALUATED = {"de": "d-8c57fa13116b47be8eec95dfaf2aa030"} # VBV - Neues Feedback für Circle NEW_FEEDBACK = {"de": "d-40fb94d5149949e7b8e7ddfcf0fcfdde"} class EmailService: """Email service class implemented using the Sendgrid API""" _sendgrid_client = SendGridAPIClient(setting("SENDGRID_API_KEY")) @classmethod def send_email( cls, recipient: User, template: EmailTemplate, template_data: dict, ) -> None: message = Mail( from_email="noreply@my.vbv-afa.ch", to_emails=recipient.email, ) message.template_id = template.value.get( recipient.language, template.value["de"] ) message.dynamic_template_data = template_data cls._sendgrid_client.send(message) class NotificationService: @classmethod def send_user_interaction_notification( cls, recipient: User, verb: str, sender: User, course: str, target_url: str, email_template: EmailTemplate, template_data: dict = {}, ) -> None: cls._send_notification( recipient=recipient, verb=verb, sender=sender, course=course, target_url=target_url, notification_type=NotificationType.USER_INTERACTION, email_template=email_template, template_data=template_data, ) @classmethod def send_progress_notification( cls, recipient: User, verb: str, course: str, target_url: str, email_template: EmailTemplate, template_data: dict = {}, ) -> None: cls._send_notification( recipient=recipient, verb=verb, course=course, target_url=target_url, notification_type=NotificationType.PROGRESS, email_template=email_template, template_data=template_data, ) @classmethod def send_information_notification( cls, recipient: User, verb: str, target_url: str, email_template: EmailTemplate, template_data: dict = {}, ) -> None: cls._send_notification( recipient=recipient, verb=verb, target_url=target_url, notification_type=NotificationType.INFORMATION, email_template=email_template, template_data=template_data, ) @classmethod def _send_notification( cls, recipient: User, verb: str, notification_type: NotificationType, email_template: EmailTemplate, template_data: dict, sender: User | None = None, course: str | None = None, target_url: str | None = None, ) -> None: actor_avatar_url: Optional[str] = None if not sender: sender = User.objects.get(email="admin") else: actor_avatar_url = sender.avatar_url log = logger.bind( recipient=recipient.get_full_name(), sender=sender.get_full_name(), verb=verb, notification_type=notification_type, course=course, target_url=target_url, template_data=template_data, ) if NotificationService._is_duplicate_notification( recipient=recipient, verb=verb, notification_type=notification_type, target_url=target_url, template_name=email_template.name, template_data=template_data, ): log.warn("A duplicate notification was omitted from being sent") return emailed = False if cls._should_send_email(notification_type, recipient): emailed = cls._send_email( recipient=recipient, template=email_template, template_data={ "target_url": f"https://my.vbv-afa.ch{target_url}", **template_data, }, ) try: response = notify.send( sender=sender, recipient=recipient, verb=verb, emailed=emailed, # The metadata is saved in the 'data' member of the AbstractNotification model email_template=email_template.name, template_data=template_data, ) sent_notification: Notification = response[0][1][0] # 🫨 sent_notification.target_url = target_url sent_notification.notification_type = notification_type sent_notification.course = course sent_notification.actor_avatar_url = actor_avatar_url sent_notification.save() except Exception as e: log.error("Failed to send notification", exception=str(e)) else: log.info("Notification sent successfully") @staticmethod def _should_send_email( notification_type: NotificationType, recipient: User ) -> bool: return str(notification_type) in recipient.additional_json_data.get( "email_notification_types", [] ) @staticmethod def _send_email( recipient: User, template: EmailTemplate, template_data: dict | None, ) -> bool: log = logger.bind( recipient=recipient.username, template=template.name, template_data=template_data, ) try: EmailService.send_email( recipient=recipient, template=template, template_data=template_data, ) log.info("Email sent successfully") return True except Exception as e: log.error( "Failed to send Email", exception=str(e), exc_info=True, stack_info=True ) return False @staticmethod def _is_duplicate_notification( recipient: User, verb: str, notification_type: NotificationType, template_name: str, template_data: dict, target_url: str | None, ) -> bool: """Check if a notification with the same parameters has already been sent to the recipient. This is to prevent duplicate notifications from being sent and to protect against potential programming errors. """ return Notification.objects.filter( recipient=recipient, verb=verb, notification_type=notification_type, target_url=target_url, data={ "email_template": template_name, "template_data": template_data, }, ).exists()