from __future__ import annotations from typing import TYPE_CHECKING import structlog from django.db.models import Model from notifications.signals import notify from vbv_lernwelt.assignment.models import AssignmentType from vbv_lernwelt.core.models import User from vbv_lernwelt.course.models import CourseSession from vbv_lernwelt.course_session.models import CourseSessionAssignment from vbv_lernwelt.notify.email.email_services import ( create_template_data_from_course_session_attendance_course, EmailTemplate, format_swiss_datetime, send_email, ) from vbv_lernwelt.notify.models import ( Notification, NotificationCategory, NotificationTrigger, ) if TYPE_CHECKING: from vbv_lernwelt.assignment.models import AssignmentCompletion from vbv_lernwelt.course_session.models import ( CourseSessionAttendanceCourse, CourseSessionEdoniqTest, ) from vbv_lernwelt.feedback.models import FeedbackResponse logger = structlog.get_logger(__name__) class NotificationService: @classmethod def send_assignment_submitted_notification( cls, recipient: User, sender: User, assignment_completion: AssignmentCompletion ): texts = { "de": "%(sender)s hat die geleitete Fallarbeit «%(assignment_title)s» abgegeben.", "fr": "%(sender)s a soumis l'étude de cas dirigée «%(assignment_title)s».", "it": "%(sender)s ha consegnato il caso di studio guidato «%(assignment_title)s».", } verb = texts.get(recipient.language, "de") % { "sender": sender.get_full_name(), "assignment_title": assignment_completion.assignment.title, } return cls._send_notification( recipient=recipient, verb=verb, notification_category=NotificationCategory.USER_INTERACTION, notification_trigger=NotificationTrigger.CASEWORK_SUBMITTED, sender=sender, target_url=assignment_completion.get_assignment_evaluation_frontend_url(), course_session=assignment_completion.course_session, action_object=assignment_completion, email_template=EmailTemplate.CASEWORK_SUBMITTED, ) @classmethod def send_assignment_evaluated_notification( cls, recipient: User, sender: User, assignment_completion: AssignmentCompletion, target_url: str, ): texts = { "de": "%(sender)s hat die geleitete Fallarbeit «%(assignment_title)s» bewertet.", "fr": "%(sender)s a évalué l'étude de cas dirigée «%(assignment_title)s».", "it": "%(sender)s ha valutato il caso di studio guidato «%(assignment_title)s».", } verb = texts.get(recipient.language, "de") % { "sender": sender.get_full_name(), "assignment_title": assignment_completion.assignment.title, } return cls._send_notification( recipient=recipient, verb=verb, notification_category=NotificationCategory.USER_INTERACTION, notification_trigger=NotificationTrigger.CASEWORK_EVALUATED, sender=sender, target_url=target_url, course_session=assignment_completion.course_session, action_object=assignment_completion, email_template=EmailTemplate.CASEWORK_EVALUATED, ) @classmethod def send_new_feedback_notification( cls, recipient: User, feedback_response: FeedbackResponse, ): texts = { "de": "Feedback abgeschickt für Circle «%(circle_title)s»", "fr": "Feedback envoyé pour le cercle «%(circle_title)s»", "it": "Feedback inviato per il cerchio «%(circle_title)s»", } verb = texts.get(recipient.language, "de") % { "circle_title": feedback_response.circle.title, } return cls._send_notification( recipient=recipient, verb=verb, notification_category=NotificationCategory.INFORMATION, notification_trigger=NotificationTrigger.NEW_FEEDBACK, target_url=f"/course/{feedback_response.course_session.course.slug}/cockpit/feedback/{feedback_response.circle_id}/", course_session=feedback_response.course_session, action_object=feedback_response, email_template=EmailTemplate.NEW_FEEDBACK, ) @classmethod def send_attendance_course_reminder_notification( cls, recipient: User, attendance_course: CourseSessionAttendanceCourse, ): texts = { "de": "Erinnerung: Bald findet ein Präsenzkurs statt", "fr": "Rappel: Un cours de présence aura lieu bientôt.", "it": "Promemoria: Un corso di presenza avrà luogo presto.", } verb = texts.get(recipient.language, "de") return cls._send_notification( recipient=recipient, verb=verb, notification_category=NotificationCategory.INFORMATION, notification_trigger=NotificationTrigger.ATTENDANCE_COURSE_REMINDER, target_url=attendance_course.learning_content.get_frontend_url(), action_object=attendance_course, course_session=attendance_course.course_session, email_template=EmailTemplate.ATTENDANCE_COURSE_REMINDER, template_data=create_template_data_from_course_session_attendance_course( attendance_course=attendance_course ), ) @classmethod def send_assignment_reminder_notification_member( cls, recipient: User, assignment: CourseSessionAssignment, ): texts = { "de": "Erinnerung: Bald ist ein Abgabetermin", "fr": "Rappel: Une date limite approche", "it": "Promemoria: Una scadenza si avvicina", } templates = { AssignmentType.CASEWORK: EmailTemplate.ASSIGNMENT_REMINDER_CASEWORK_MEMBER, AssignmentType.PREP_ASSIGNMENT: EmailTemplate.ASSIGNMENT_REMINDER_PREP_ASSIGNMENT_MEMBER, } verb = texts.get(recipient.language, "de") circle = assignment.learning_content.get_parent_circle().title due_at = assignment.submission_deadline.start return cls._send_notification( recipient=recipient, verb=verb, notification_category=NotificationCategory.INFORMATION, notification_trigger=NotificationTrigger.ASSIGNMENT_REMINDER, target_url=assignment.learning_content.get_frontend_url(), action_object=assignment, course_session=assignment.course_session, email_template=templates[ AssignmentType(assignment.learning_content.assignment_type) ], template_data={ "circle": circle, "due_date": format_swiss_datetime(due_at), }, ) @classmethod def send_edoniq_test_reminder_notification_member( cls, recipient: User, edoniq_test: CourseSessionEdoniqTest, ): texts = { "de": "Erinnerung: Bald ist ein Abgabetermin", "fr": "Rappel: Une date limite approche", "it": "Promemoria: Una scadenza si avvicina", } verb = texts.get(recipient.language, "de") circle = edoniq_test.learning_content.get_parent_circle().title due_at = edoniq_test.deadline.start return cls._send_notification( recipient=recipient, verb=verb, notification_category=NotificationCategory.INFORMATION, notification_trigger=NotificationTrigger.ASSIGNMENT_REMINDER, target_url=edoniq_test.learning_content.get_frontend_url(), action_object=edoniq_test, course_session=edoniq_test.course_session, email_template=EmailTemplate.ASSIGNMENT_REMINDER_EDONIQ_MEMBER, template_data={ "circle": circle, "due_date": format_swiss_datetime(due_at), }, ) @classmethod def send_casework_expert_evaluation_reminder( cls, recipient: User, assignment: CourseSessionAssignment, ): texts = { "de": "Erinnerung: Bald ist ein Bewertungstermin", "fr": "Rappel: Une date limite approche", "it": "Promemoria: Una scadenza si avvicina", } verb = texts.get(recipient.language, "de") circle = assignment.learning_content.get_parent_circle().title due_at = assignment.evaluation_deadline.start return cls._send_notification( recipient=recipient, verb=verb, notification_category=NotificationCategory.INFORMATION, notification_trigger=NotificationTrigger.CASEWORK_EXPERT_EVALUATION_REMINDER, target_url=assignment.evaluation_deadline.url_expert, action_object=assignment, course_session=assignment.course_session, email_template=EmailTemplate.EVALUATION_REMINDER_CASEWORK_EXPERT, template_data={ "circle": circle, "due_date": format_swiss_datetime(due_at), }, ) @classmethod def _send_notification( cls, recipient: User, verb: str, notification_category: NotificationCategory, notification_trigger: NotificationTrigger, sender: User | None = None, action_object: Model | None = None, target_url: str | None = None, course_session: CourseSession | None = None, email_template: EmailTemplate | None = None, template_data: dict | None = None, fail_silently: bool = True, ) -> str: if template_data is None: template_data = {} notification_identifier = ( f"{notification_category.name}_{notification_trigger.name}" ) actor_avatar_url = "" 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_category=notification_category, notification_trigger=notification_trigger, course_session=course_session.title if course_session else "", target_url=target_url, template_data=template_data, ) emailed = False try: notification = NotificationService._find_duplicate_notification( recipient=recipient, verb=verb, notification_category=notification_category, notification_trigger=notification_trigger, target_url=target_url, course_session=course_session, ) emailed = False if notification and notification.emailed: emailed = True if ( email_template and cls._should_send_email(notification_category, recipient) and not emailed ): log.debug("Try to send email") try: emailed = cls._send_email( recipient=recipient, template=email_template, template_data={ "target_url": f"https://my.vbv-afa.ch{target_url}", **template_data, }, fail_silently=False, ) except Exception as e: notification_identifier += "_email_error" if not fail_silently: raise e return notification_identifier if emailed: notification_identifier += "_emailed" if notification: notification.emailed = True notification.save() else: log.debug("Should not send email") if notification: log.info("Duplicate notification was omitted from being sent") notification_identifier += "_duplicate" return notification_identifier else: response = notify.send( sender=sender, recipient=recipient, verb=verb, action_object=action_object, emailed=emailed, # The extra arguments are saved in the 'data' member email_template=email_template.name if email_template else "", template_data=template_data, ) sent_notification: Notification = response[0][1][0] # 🫨 sent_notification.target_url = target_url sent_notification.notification_category = notification_category sent_notification.notification_trigger = notification_trigger sent_notification.course_session = course_session sent_notification.actor_avatar_url = actor_avatar_url sent_notification.save() log.info("Notification sent successfully", emailed=emailed) return f"{notification_identifier}_success" except Exception as e: log.error( "Failed to send notification", exception=str(e), exc_info=True, stack_info=True, emailed=emailed, ) if not fail_silently: raise e return f"{notification_identifier}_error" @staticmethod def _should_send_email( notification_category: NotificationCategory, recipient: User ) -> bool: return str(notification_category) in recipient.additional_json_data.get( "email_notification_categories", [] ) @staticmethod def _send_email( recipient: User, template: EmailTemplate, template_data: dict, fail_silently: bool = True, ) -> bool: return send_email( recipient_email=recipient.email, template=template, template_data=template_data, template_language=recipient.language, fail_silently=fail_silently, ) @staticmethod def _find_duplicate_notification( recipient: User, verb: str, notification_category: NotificationCategory, notification_trigger: NotificationTrigger, target_url: str | None, course_session: CourseSession | None, ) -> Notification | None: """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_category=notification_category, notification_trigger=notification_trigger, target_url=target_url, course_session=course_session, ).first()