from typing import Optional import structlog from notifications.signals import notify from vbv_lernwelt.core.models import User from vbv_lernwelt.notify.email.email_services import EmailTemplate, send_email from vbv_lernwelt.notify.models import Notification, NotificationType logger = structlog.get_logger(__name__) 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, ) -> 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, ) -> 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, ) -> 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 | None = None, sender: User | None = None, course: str | None = None, target_url: str | None = None, ) -> None: if template_data is None: template_data = {} 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.email, sender=sender.email, 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.info("A duplicate notification was omitted from being sent") return emailed = False if cls._should_send_email(notification_type, recipient): log.debug("Try to send email") emailed = cls._send_email( recipient=recipient, template=email_template, template_data={ "target_url": f"https://my.vbv-afa.ch{target_url}", **template_data, }, ) else: log.debug("Should not send email") 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", emailed=emailed) @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, ) -> bool: return send_email( recipient_email=recipient.email, template=template, template_data=template_data, template_language=recipient.language, ) @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()