mirror of
https://github.com/pacnpal/thrillwiki_django_no_react.git
synced 2026-02-05 08:45:19 -05:00
299 lines
9.4 KiB
Python
299 lines
9.4 KiB
Python
"""
|
|
Notifications models.
|
|
|
|
Provides models for:
|
|
- Subscriber: User notification profile (legacy, kept for compatibility)
|
|
- NotificationPreference: User notification preferences
|
|
- NotificationLog: Audit trail of sent notifications
|
|
- SystemAnnouncement: System-wide announcements
|
|
|
|
Note: Now using django-notifications-hq for the core notification system.
|
|
Subscriber model is kept for backward compatibility but is optional.
|
|
"""
|
|
|
|
from django.conf import settings
|
|
from django.db import models
|
|
|
|
|
|
class Subscriber(models.Model):
|
|
"""
|
|
User notification profile.
|
|
|
|
Note: This model is kept for backward compatibility. The new
|
|
django-notifications-hq system uses User directly for notifications.
|
|
This can be used for storing additional notification-related user data.
|
|
"""
|
|
|
|
user = models.OneToOneField(
|
|
settings.AUTH_USER_MODEL,
|
|
on_delete=models.CASCADE,
|
|
related_name="notification_subscriber",
|
|
)
|
|
# Legacy field - kept for migration compatibility
|
|
novu_subscriber_id = models.CharField(
|
|
max_length=255,
|
|
unique=True,
|
|
db_index=True,
|
|
help_text="Legacy Novu subscriber ID (deprecated)"
|
|
)
|
|
first_name = models.CharField(max_length=100, blank=True)
|
|
last_name = models.CharField(max_length=100, blank=True)
|
|
email = models.EmailField(blank=True)
|
|
phone = models.CharField(max_length=20, blank=True)
|
|
avatar = models.URLField(blank=True)
|
|
locale = models.CharField(max_length=10, default="en")
|
|
data = models.JSONField(default=dict, blank=True, help_text="Custom subscriber data")
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
|
|
class Meta:
|
|
verbose_name = "Notification Subscriber"
|
|
verbose_name_plural = "Notification Subscribers"
|
|
|
|
def __str__(self):
|
|
return f"Subscriber({self.user.username})"
|
|
|
|
|
|
class NotificationPreference(models.Model):
|
|
"""
|
|
User notification preferences across channels and workflows.
|
|
"""
|
|
|
|
user = models.OneToOneField(
|
|
settings.AUTH_USER_MODEL,
|
|
on_delete=models.CASCADE,
|
|
related_name="novu_notification_prefs", # Renamed to avoid conflict with User.notification_preferences JSONField
|
|
)
|
|
# Channel preferences
|
|
channel_preferences = models.JSONField(
|
|
default=dict,
|
|
blank=True,
|
|
help_text="Preferences per channel (email, push, in_app, sms)",
|
|
)
|
|
# Workflow-specific preferences
|
|
workflow_preferences = models.JSONField(
|
|
default=dict,
|
|
blank=True,
|
|
help_text="Preferences per notification workflow",
|
|
)
|
|
# Frequency settings
|
|
frequency_settings = models.JSONField(
|
|
default=dict,
|
|
blank=True,
|
|
help_text="Digest and frequency settings",
|
|
)
|
|
# Global opt-out
|
|
is_opted_out = models.BooleanField(default=False)
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
|
|
class Meta:
|
|
verbose_name = "Notification Preference"
|
|
verbose_name_plural = "Notification Preferences"
|
|
|
|
def __str__(self):
|
|
return f"Preferences({self.user.username})"
|
|
|
|
|
|
class NotificationLog(models.Model):
|
|
"""
|
|
Audit log of sent notifications.
|
|
"""
|
|
|
|
class Status(models.TextChoices):
|
|
PENDING = "pending", "Pending"
|
|
SENT = "sent", "Sent"
|
|
DELIVERED = "delivered", "Delivered"
|
|
FAILED = "failed", "Failed"
|
|
|
|
user = models.ForeignKey(
|
|
settings.AUTH_USER_MODEL,
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
related_name="notification_logs",
|
|
)
|
|
workflow_id = models.CharField(max_length=100, db_index=True)
|
|
notification_type = models.CharField(max_length=50)
|
|
channel = models.CharField(max_length=20) # email, push, in_app, sms
|
|
status = models.CharField(
|
|
max_length=20,
|
|
choices=Status.choices,
|
|
default=Status.PENDING,
|
|
)
|
|
payload = models.JSONField(default=dict, blank=True)
|
|
error_message = models.TextField(blank=True)
|
|
novu_transaction_id = models.CharField(max_length=255, blank=True, db_index=True)
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
|
|
class Meta:
|
|
verbose_name = "Notification Log"
|
|
verbose_name_plural = "Notification Logs"
|
|
ordering = ["-created_at"]
|
|
indexes = [
|
|
models.Index(fields=["user", "-created_at"]),
|
|
models.Index(fields=["workflow_id", "-created_at"]),
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"Log({self.workflow_id}, {self.status})"
|
|
|
|
|
|
class SystemAnnouncement(models.Model):
|
|
"""
|
|
System-wide announcements.
|
|
"""
|
|
|
|
class Severity(models.TextChoices):
|
|
INFO = "info", "Information"
|
|
WARNING = "warning", "Warning"
|
|
CRITICAL = "critical", "Critical"
|
|
|
|
title = models.CharField(max_length=255)
|
|
message = models.TextField()
|
|
severity = models.CharField(
|
|
max_length=20,
|
|
choices=Severity.choices,
|
|
default=Severity.INFO,
|
|
)
|
|
action_url = models.URLField(blank=True)
|
|
is_active = models.BooleanField(default=True)
|
|
created_by = models.ForeignKey(
|
|
settings.AUTH_USER_MODEL,
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
related_name="announcements_created",
|
|
)
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
expires_at = models.DateTimeField(null=True, blank=True)
|
|
|
|
class Meta:
|
|
verbose_name = "System Announcement"
|
|
verbose_name_plural = "System Announcements"
|
|
ordering = ["-created_at"]
|
|
|
|
def __str__(self):
|
|
return f"{self.title} ({self.severity})"
|
|
|
|
|
|
class Notification(models.Model):
|
|
"""
|
|
In-app notification model.
|
|
|
|
This is a Django-native implementation for storing user notifications,
|
|
supporting both in-app and email notification channels.
|
|
"""
|
|
|
|
class Level(models.TextChoices):
|
|
INFO = "info", "Info"
|
|
SUCCESS = "success", "Success"
|
|
WARNING = "warning", "Warning"
|
|
ERROR = "error", "Error"
|
|
|
|
# Who receives the notification
|
|
recipient = models.ForeignKey(
|
|
settings.AUTH_USER_MODEL,
|
|
on_delete=models.CASCADE,
|
|
related_name="in_app_notifications", # Renamed to avoid clash with accounts.UserNotification
|
|
)
|
|
# Who triggered the notification (can be null for system notifications)
|
|
actor = models.ForeignKey(
|
|
settings.AUTH_USER_MODEL,
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
blank=True,
|
|
related_name="notifications_sent",
|
|
)
|
|
# What happened
|
|
verb = models.CharField(max_length=255)
|
|
description = models.TextField(blank=True)
|
|
level = models.CharField(
|
|
max_length=20,
|
|
choices=Level.choices,
|
|
default=Level.INFO,
|
|
)
|
|
# The object that was acted upon (generic foreign key)
|
|
action_object_content_type = models.ForeignKey(
|
|
"contenttypes.ContentType",
|
|
on_delete=models.CASCADE,
|
|
blank=True,
|
|
null=True,
|
|
related_name="notification_action_objects",
|
|
)
|
|
action_object_id = models.PositiveIntegerField(blank=True, null=True)
|
|
# The target of the action (generic foreign key)
|
|
target_content_type = models.ForeignKey(
|
|
"contenttypes.ContentType",
|
|
on_delete=models.CASCADE,
|
|
blank=True,
|
|
null=True,
|
|
related_name="notification_targets",
|
|
)
|
|
target_id = models.PositiveIntegerField(blank=True, null=True)
|
|
# Additional data
|
|
data = models.JSONField(default=dict, blank=True)
|
|
# Status
|
|
unread = models.BooleanField(default=True, db_index=True)
|
|
# Timestamps
|
|
timestamp = models.DateTimeField(auto_now_add=True)
|
|
read_at = models.DateTimeField(null=True, blank=True)
|
|
|
|
class Meta:
|
|
verbose_name = "Notification"
|
|
verbose_name_plural = "Notifications"
|
|
ordering = ["-timestamp"]
|
|
indexes = [
|
|
models.Index(fields=["recipient", "-timestamp"]),
|
|
models.Index(fields=["recipient", "unread"]),
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"{self.verb} -> {self.recipient}"
|
|
|
|
def mark_as_read(self):
|
|
"""Mark this notification as read."""
|
|
if self.unread:
|
|
from django.utils import timezone
|
|
self.unread = False
|
|
self.read_at = timezone.now()
|
|
self.save(update_fields=["unread", "read_at"])
|
|
|
|
@property
|
|
def action_object(self):
|
|
"""Get the action object instance."""
|
|
if self.action_object_content_type and self.action_object_id:
|
|
return self.action_object_content_type.get_object_for_this_type(
|
|
pk=self.action_object_id
|
|
)
|
|
return None
|
|
|
|
@property
|
|
def target(self):
|
|
"""Get the target instance."""
|
|
if self.target_content_type and self.target_id:
|
|
return self.target_content_type.get_object_for_this_type(pk=self.target_id)
|
|
return None
|
|
|
|
|
|
class NotificationManager(models.Manager):
|
|
"""Custom manager for Notification model."""
|
|
|
|
def unread(self):
|
|
"""Return only unread notifications."""
|
|
return self.filter(unread=True)
|
|
|
|
def read(self):
|
|
"""Return only read notifications."""
|
|
return self.filter(unread=False)
|
|
|
|
def mark_all_as_read(self):
|
|
"""Mark all notifications as read."""
|
|
from django.utils import timezone
|
|
return self.filter(unread=True).update(unread=False, read_at=timezone.now())
|
|
|
|
|
|
# Add custom manager to Notification model
|
|
Notification.objects = NotificationManager()
|
|
Notification.objects.model = Notification
|
|
|