121 lines
3.9 KiB
Python
121 lines
3.9 KiB
Python
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}: <a href='{target_url}'>Link</a>",
|
|
)
|
|
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,
|
|
)
|