100 lines
3.9 KiB
Python
100 lines
3.9 KiB
Python
import logging
|
|
|
|
from actstream.models import Action
|
|
from django.contrib.auth import get_user_model
|
|
from django.db import models
|
|
from django.template import loader, TemplateDoesNotExist
|
|
|
|
from constants.activity import Verb
|
|
from utils import absolutify
|
|
|
|
User = get_user_model()
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Notification(models.Model):
|
|
"""Notification records are created in Action's post_save signal.
|
|
When a user marks a notification as read, read_at is set.
|
|
send_notification_emails management command runs periodically in background and sends all
|
|
notifications that haven't been processed yet, read_at is not checked when sending emails.
|
|
email_sent flag is used only to record the fact that we attempted to send an email.
|
|
A user can unsubscribe from notification emails in their profile settings.
|
|
"""
|
|
|
|
recipient = models.ForeignKey(User, null=False, on_delete=models.CASCADE)
|
|
action = models.ForeignKey(Action, null=False, on_delete=models.CASCADE)
|
|
email_sent = models.BooleanField(default=False, null=False)
|
|
processed_by_mailer_at = models.DateTimeField(default=None, null=True)
|
|
read_at = models.DateTimeField(default=None, null=True)
|
|
|
|
class Meta:
|
|
indexes = [
|
|
models.Index(fields=['processed_by_mailer_at']),
|
|
models.Index(fields=['recipient', 'read_at']),
|
|
]
|
|
unique_together = ['recipient', 'action']
|
|
|
|
@property
|
|
def original_template_name(self) -> str:
|
|
"""Template name constructed from action type, target and so on.
|
|
|
|
If we want to override email template for this specific notification,
|
|
this is the name of the template that should be created.
|
|
"""
|
|
action = self.action
|
|
action_object = action.action_object
|
|
target_name = action.target.__class__.__name__.lower()
|
|
action_object_name = action_object.__class__.__name__.lower() if action_object else ''
|
|
return f"new_activity_{action_object_name}_{target_name}_{action.verb.replace(' ', '_')}"
|
|
|
|
@property
|
|
def template_name(self) -> str:
|
|
"""Template name to be used for constructing notification email."""
|
|
default_name = 'new_activity'
|
|
name = self.original_template_name
|
|
args = {'name': name, 'default_name': default_name}
|
|
|
|
try:
|
|
loader.get_template(f'emails/{name}.html')
|
|
except TemplateDoesNotExist:
|
|
logger.warning('Template %(name)s does not exist, using %(default_name)s', args)
|
|
return default_name
|
|
return name
|
|
|
|
def get_template_context(self):
|
|
"""Return template context to be used in the notification email."""
|
|
action = self.action
|
|
action_object = action.action_object
|
|
quoted_message = getattr(action_object, 'message', getattr(action_object, 'text', ''))
|
|
context = {
|
|
'Verb': Verb,
|
|
'action': action,
|
|
'action_object': action_object,
|
|
'notification': self,
|
|
'quoted_message': quoted_message,
|
|
'target': action.target,
|
|
'url': self.get_absolute_url(),
|
|
'user': self.recipient,
|
|
}
|
|
return context
|
|
|
|
def get_absolute_url(self):
|
|
if self.action.verb == Verb.RATED_EXTENSION:
|
|
fragment = f'#review-{self.action.action_object.pk}'
|
|
url = self.action.target.get_ratings_url() + fragment
|
|
elif self.action.verb in [
|
|
Verb.APPROVED,
|
|
Verb.COMMENTED,
|
|
Verb.REQUESTED_CHANGES,
|
|
Verb.REQUESTED_REVIEW,
|
|
]:
|
|
fragment = f'#activity-{self.action.action_object.pk}'
|
|
url = self.action.target.get_review_url() + fragment
|
|
elif self.action.action_object is not None:
|
|
url = self.action.action_object.get_absolute_url()
|
|
else:
|
|
url = self.action.target.get_absolute_url()
|
|
|
|
# TODO? url cloacking to mark visited notifications as read automatically
|
|
return absolutify(url)
|