import logging from typing import Optional 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 = logging.getLogger(__name__) class EmailService: _sendgrid_client = SendGridAPIClient(setting("SENDGRID_API_KEY")) @classmethod def send_email(cls, recipient: User, verb: str, target_url) -> bool: message = Mail( from_email="info@iterativ.ch", to_emails=recipient.email, subject=f"myVBV - {verb}", ## TODO: Add HTML content. html_content=f"{verb}: Link", ) try: cls._sendgrid_client.send(message) logger.info(f"Successfully sent email to {recipient}") return True except Exception as e: logger.error(f"Failed to send email to {recipient}: {e}") return False class NotificationService: @classmethod def send_user_interaction_notification( cls, recipient: User, verb: str, sender: User, course: str, target_url: str ) -> None: cls._send_notification( recipient=recipient, verb=verb, sender=sender, course=course, target_url=target_url, notification_type=NotificationType.USER_INTERACTION, ) @classmethod def send_progress_notification( cls, recipient: User, verb: str, course: str, target_url: str ) -> None: cls._send_notification( recipient=recipient, verb=verb, course=course, target_url=target_url, notification_type=NotificationType.PROGRESS, ) @classmethod def send_information_notification( cls, recipient: User, verb: str, target_url: str ) -> None: cls._send_notification( recipient=recipient, verb=verb, target_url=target_url, notification_type=NotificationType.INFORMATION, ) @classmethod def _send_notification( cls, recipient: User, verb: str, notification_type: NotificationType, sender: Optional[User] = None, course: Optional[str] = None, target_url: Optional[str] = None, ) -> None: actor_avatar_url: Optional[str] = None if not sender: sender = User.objects.get(email="admin") else: actor_avatar_url = sender.avatar_url emailed = False if cls._should_send_email(notification_type, recipient): emailed = cls._send_email(recipient, verb, target_url) response = notify.send( sender=sender, recipient=recipient, verb=verb, ) # Custom Notification model fields cannot be set using the notify.send() method. # https://github.com/django-notifications/django-notifications/issues/301 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.target_url = target_url sent_notification.actor_avatar_url = actor_avatar_url sent_notification.emailed = emailed sent_notification.save() @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, verb: str, target_url: Optional[str]) -> bool: return EmailService.send_email( recipient=recipient, verb=verb, target_url=target_url, )