""" alert a user to activity """ from django.db import models from django.dispatch import receiver from .base_model import BookWyrmModel from . import Boost, Favorite, ImportJob, Report, Status, User # pylint: disable=line-too-long NotificationType = models.TextChoices( "NotificationType", "FAVORITE REPLY MENTION TAG FOLLOW FOLLOW_REQUEST BOOST IMPORT ADD REPORT INVITE ACCEPT JOIN LEAVE REMOVE GROUP_PRIVACY GROUP_NAME GROUP_DESCRIPTION", ) class Notification(BookWyrmModel): """you've been tagged, liked, followed, etc""" user = models.ForeignKey("User", on_delete=models.CASCADE) related_book = models.ForeignKey("Edition", on_delete=models.CASCADE, null=True) related_user = models.ForeignKey( "User", on_delete=models.CASCADE, null=True, related_name="related_user" ) related_group = models.ForeignKey( "Group", on_delete=models.CASCADE, null=True, related_name="notifications" ) related_status = models.ForeignKey("Status", on_delete=models.CASCADE, null=True) related_import = models.ForeignKey("ImportJob", on_delete=models.CASCADE, null=True) related_list_item = models.ForeignKey( "ListItem", on_delete=models.CASCADE, null=True ) related_report = models.ForeignKey("Report", on_delete=models.CASCADE, null=True) read = models.BooleanField(default=False) notification_type = models.CharField( max_length=255, choices=NotificationType.choices ) def save(self, *args, **kwargs): """save, but don't make dupes""" # there's probably a better way to do this if self.__class__.objects.filter( user=self.user, related_book=self.related_book, related_user=self.related_user, related_group=self.related_group, related_status=self.related_status, related_import=self.related_import, related_list_item=self.related_list_item, related_report=self.related_report, notification_type=self.notification_type, ).exists(): return super().save(*args, **kwargs) class Meta: """checks if notifcation is in enum list for valid types""" constraints = [ models.CheckConstraint( check=models.Q(notification_type__in=NotificationType.values), name="notification_type_valid", ) ] @receiver(models.signals.post_save, sender=Favorite) # pylint: disable=unused-argument def notify_on_fav(sender, instance, *args, **kwargs): """someone liked your content, you ARE loved""" if not instance.status.user.local or instance.status.user == instance.user: return Notification.objects.create( user=instance.status.user, notification_type="FAVORITE", related_user=instance.user, related_status=instance.status, ) @receiver(models.signals.post_delete, sender=Favorite) # pylint: disable=unused-argument def notify_on_unfav(sender, instance, *args, **kwargs): """oops, didn't like that after all""" if not instance.status.user.local: return Notification.objects.filter( user=instance.status.user, related_user=instance.user, related_status=instance.status, notification_type="FAVORITE", ).delete() @receiver(models.signals.post_save) # pylint: disable=unused-argument def notify_user_on_mention(sender, instance, *args, **kwargs): """creating and deleting statuses with @ mentions and replies""" if not issubclass(sender, Status): return if instance.deleted: Notification.objects.filter(related_status=instance).delete() return if ( instance.reply_parent and instance.reply_parent.user != instance.user and instance.reply_parent.user.local ): Notification.objects.create( user=instance.reply_parent.user, notification_type="REPLY", related_user=instance.user, related_status=instance, ) for mention_user in instance.mention_users.all(): # avoid double-notifying about this status if not mention_user.local or ( instance.reply_parent and mention_user == instance.reply_parent.user ): continue Notification.objects.create( user=mention_user, notification_type="MENTION", related_user=instance.user, related_status=instance, ) @receiver(models.signals.post_save, sender=Boost) # pylint: disable=unused-argument def notify_user_on_boost(sender, instance, *args, **kwargs): """boosting a status""" if ( not instance.boosted_status.user.local or instance.boosted_status.user == instance.user ): return Notification.objects.create( user=instance.boosted_status.user, related_status=instance.boosted_status, related_user=instance.user, notification_type="BOOST", ) @receiver(models.signals.post_delete, sender=Boost) # pylint: disable=unused-argument def notify_user_on_unboost(sender, instance, *args, **kwargs): """unboosting a status""" Notification.objects.filter( user=instance.boosted_status.user, related_status=instance.boosted_status, related_user=instance.user, notification_type="BOOST", ).delete() @receiver(models.signals.post_save, sender=ImportJob) # pylint: disable=unused-argument def notify_user_on_import_complete( sender, instance, *args, update_fields=None, **kwargs ): """we imported your books! aren't you proud of us""" update_fields = update_fields or [] if not instance.complete or "complete" not in update_fields: return Notification.objects.create( user=instance.user, notification_type="IMPORT", related_import=instance, ) @receiver(models.signals.post_save, sender=Report) # pylint: disable=unused-argument def notify_admins_on_report(sender, instance, *args, **kwargs): """something is up, make sure the admins know""" # moderators and superusers should be notified admins = User.objects.filter( models.Q(user_permissions__name__in=["moderate_user", "moderate_post"]) | models.Q(is_superuser=True) ).all() for admin in admins: Notification.objects.create( user=admin, related_report=instance, notification_type="REPORT", )