408 lines
15 KiB
Python
408 lines
15 KiB
Python
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()
|