591 lines
18 KiB
Python
591 lines
18 KiB
Python
"""
|
|
Notifications models for the Tenhal Multidisciplinary Healthcare Platform.
|
|
|
|
This module handles message templates, outbound messages (SMS/WhatsApp/Email),
|
|
and patient notification preferences.
|
|
"""
|
|
|
|
from django.db import models
|
|
from django.utils.translation import gettext_lazy as _
|
|
|
|
from core.models import (
|
|
UUIDPrimaryKeyMixin,
|
|
TimeStampedMixin,
|
|
TenantOwnedMixin,
|
|
)
|
|
from django.conf import settings
|
|
|
|
|
|
class MessageTemplate(UUIDPrimaryKeyMixin, TimeStampedMixin, TenantOwnedMixin):
|
|
"""
|
|
Reusable message templates for notifications.
|
|
Supports variable substitution for personalization.
|
|
"""
|
|
|
|
class Channel(models.TextChoices):
|
|
SMS = 'SMS', _('SMS')
|
|
WHATSAPP = 'WHATSAPP', _('WhatsApp')
|
|
EMAIL = 'EMAIL', _('Email')
|
|
|
|
code = models.CharField(
|
|
max_length=100,
|
|
unique=True,
|
|
help_text=_("Unique identifier for this template (e.g., 'appointment_reminder')"),
|
|
verbose_name=_("Template Code")
|
|
)
|
|
name = models.CharField(
|
|
max_length=200,
|
|
verbose_name=_("Template Name")
|
|
)
|
|
channel = models.CharField(
|
|
max_length=20,
|
|
choices=Channel.choices,
|
|
verbose_name=_("Channel")
|
|
)
|
|
subject = models.CharField(
|
|
max_length=200,
|
|
blank=True,
|
|
help_text=_("For email only"),
|
|
verbose_name=_("Subject")
|
|
)
|
|
body_en = models.TextField(
|
|
verbose_name=_("Body (English)")
|
|
)
|
|
body_ar = models.TextField(
|
|
blank=True,
|
|
verbose_name=_("Body (Arabic)")
|
|
)
|
|
variables = models.JSONField(
|
|
default=list,
|
|
help_text=_("List of variable names that can be used in the template (e.g., ['patient_name', 'appointment_date'])"),
|
|
verbose_name=_("Variables")
|
|
)
|
|
is_active = models.BooleanField(
|
|
default=True,
|
|
verbose_name=_("Is Active")
|
|
)
|
|
|
|
class Meta:
|
|
verbose_name = _("Message Template")
|
|
verbose_name_plural = _("Message Templates")
|
|
ordering = ['channel', 'name']
|
|
|
|
def __str__(self):
|
|
return f"{self.code} ({self.get_channel_display()})"
|
|
|
|
def render(self, language='en', **context):
|
|
"""
|
|
Render the template with the provided context variables.
|
|
|
|
Args:
|
|
language: 'en' or 'ar'
|
|
**context: Variable values to substitute in the template
|
|
|
|
Returns:
|
|
Rendered message body
|
|
"""
|
|
body = self.body_ar if language == 'ar' and self.body_ar else self.body_en
|
|
|
|
# Simple variable substitution
|
|
for key, value in context.items():
|
|
placeholder = f"{{{key}}}"
|
|
body = body.replace(placeholder, str(value))
|
|
|
|
return body
|
|
|
|
|
|
class Message(UUIDPrimaryKeyMixin, TimeStampedMixin, TenantOwnedMixin):
|
|
"""
|
|
Outbound messages sent to patients.
|
|
Tracks delivery status and provider responses.
|
|
"""
|
|
|
|
class Channel(models.TextChoices):
|
|
SMS = 'SMS', _('SMS')
|
|
WHATSAPP = 'WHATSAPP', _('WhatsApp')
|
|
EMAIL = 'EMAIL', _('Email')
|
|
|
|
class Status(models.TextChoices):
|
|
QUEUED = 'QUEUED', _('Queued')
|
|
SENT = 'SENT', _('Sent')
|
|
DELIVERED = 'DELIVERED', _('Delivered')
|
|
FAILED = 'FAILED', _('Failed')
|
|
BOUNCED = 'BOUNCED', _('Bounced')
|
|
READ = 'READ', _('Read')
|
|
|
|
template = models.ForeignKey(
|
|
MessageTemplate,
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
blank=True,
|
|
related_name='messages',
|
|
verbose_name=_("Template")
|
|
)
|
|
channel = models.CharField(
|
|
max_length=20,
|
|
choices=Channel.choices,
|
|
verbose_name=_("Channel")
|
|
)
|
|
recipient = models.CharField(
|
|
max_length=200,
|
|
help_text=_("Phone number or email address"),
|
|
verbose_name=_("Recipient")
|
|
)
|
|
subject = models.CharField(
|
|
max_length=200,
|
|
blank=True,
|
|
verbose_name=_("Subject")
|
|
)
|
|
body = models.TextField(
|
|
verbose_name=_("Body")
|
|
)
|
|
variables_used = models.JSONField(
|
|
default=dict,
|
|
help_text=_("Variables and their values used to render this message"),
|
|
verbose_name=_("Variables Used")
|
|
)
|
|
status = models.CharField(
|
|
max_length=20,
|
|
choices=Status.choices,
|
|
default=Status.QUEUED,
|
|
verbose_name=_("Status")
|
|
)
|
|
sent_at = models.DateTimeField(
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Sent At")
|
|
)
|
|
delivered_at = models.DateTimeField(
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Delivered At")
|
|
)
|
|
provider_message_id = models.CharField(
|
|
max_length=200,
|
|
blank=True,
|
|
help_text=_("Message ID from the provider (Twilio, etc.)"),
|
|
verbose_name=_("Provider Message ID")
|
|
)
|
|
provider_response = models.JSONField(
|
|
default=dict,
|
|
blank=True,
|
|
help_text=_("Full response from the messaging provider"),
|
|
verbose_name=_("Provider Response")
|
|
)
|
|
error_message = models.TextField(
|
|
blank=True,
|
|
verbose_name=_("Error Message")
|
|
)
|
|
retry_count = models.PositiveIntegerField(
|
|
default=0,
|
|
verbose_name=_("Retry Count")
|
|
)
|
|
|
|
class Meta:
|
|
verbose_name = _("Message")
|
|
verbose_name_plural = _("Messages")
|
|
ordering = ['-created_at']
|
|
indexes = [
|
|
models.Index(fields=['status', 'created_at']),
|
|
models.Index(fields=['channel', 'status']),
|
|
models.Index(fields=['recipient', 'created_at']),
|
|
models.Index(fields=['tenant', 'created_at']),
|
|
models.Index(fields=['provider_message_id']),
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"{self.get_channel_display()} to {self.recipient} - {self.get_status_display()}"
|
|
|
|
@property
|
|
def is_successful(self):
|
|
"""Check if message was successfully delivered."""
|
|
return self.status in [self.Status.DELIVERED, self.Status.READ]
|
|
|
|
@property
|
|
def can_retry(self):
|
|
"""Check if message can be retried."""
|
|
return self.status == self.Status.FAILED and self.retry_count < 3
|
|
|
|
|
|
class NotificationPreference(UUIDPrimaryKeyMixin, TimeStampedMixin):
|
|
"""
|
|
Patient notification preferences.
|
|
Controls which channels and types of notifications they receive.
|
|
"""
|
|
|
|
patient = models.OneToOneField(
|
|
'core.Patient',
|
|
on_delete=models.CASCADE,
|
|
related_name='notification_preferences',
|
|
verbose_name=_("Patient")
|
|
)
|
|
|
|
# Channel Preferences
|
|
sms_enabled = models.BooleanField(
|
|
default=True,
|
|
verbose_name=_("SMS Enabled")
|
|
)
|
|
whatsapp_enabled = models.BooleanField(
|
|
default=True,
|
|
verbose_name=_("WhatsApp Enabled")
|
|
)
|
|
email_enabled = models.BooleanField(
|
|
default=True,
|
|
verbose_name=_("Email Enabled")
|
|
)
|
|
|
|
# Notification Type Preferences
|
|
appointment_reminders = models.BooleanField(
|
|
default=True,
|
|
verbose_name=_("Appointment Reminders")
|
|
)
|
|
appointment_confirmations = models.BooleanField(
|
|
default=True,
|
|
verbose_name=_("Appointment Confirmations")
|
|
)
|
|
results_notifications = models.BooleanField(
|
|
default=True,
|
|
verbose_name=_("Results Notifications")
|
|
)
|
|
billing_notifications = models.BooleanField(
|
|
default=True,
|
|
verbose_name=_("Billing Notifications")
|
|
)
|
|
marketing_communications = models.BooleanField(
|
|
default=False,
|
|
verbose_name=_("Marketing Communications")
|
|
)
|
|
|
|
# Preferred Language
|
|
preferred_language = models.CharField(
|
|
max_length=5,
|
|
choices=[('en', _('English')), ('ar', _('Arabic'))],
|
|
default='en',
|
|
verbose_name=_("Preferred Language")
|
|
)
|
|
|
|
# Preferred Channel (priority order)
|
|
preferred_channel = models.CharField(
|
|
max_length=20,
|
|
choices=Message.Channel.choices,
|
|
default=Message.Channel.SMS,
|
|
verbose_name=_("Preferred Channel")
|
|
)
|
|
|
|
class Meta:
|
|
verbose_name = _("Notification Preference")
|
|
verbose_name_plural = _("Notification Preferences")
|
|
|
|
def __str__(self):
|
|
return f"Preferences for {self.patient}"
|
|
|
|
def can_send(self, channel, notification_type):
|
|
"""
|
|
Check if a notification can be sent via the specified channel and type.
|
|
|
|
Args:
|
|
channel: One of Message.Channel choices
|
|
notification_type: Type of notification (e.g., 'appointment_reminders')
|
|
|
|
Returns:
|
|
Boolean indicating if notification can be sent
|
|
"""
|
|
# Check channel preference
|
|
channel_enabled = {
|
|
Message.Channel.SMS: self.sms_enabled,
|
|
Message.Channel.WHATSAPP: self.whatsapp_enabled,
|
|
Message.Channel.EMAIL: self.email_enabled,
|
|
}.get(channel, False)
|
|
|
|
if not channel_enabled:
|
|
return False
|
|
|
|
# Check notification type preference
|
|
type_enabled = getattr(self, notification_type, True)
|
|
|
|
return type_enabled
|
|
|
|
|
|
class MessageLog(UUIDPrimaryKeyMixin, TimeStampedMixin):
|
|
"""
|
|
Detailed log of message lifecycle events.
|
|
Useful for debugging and auditing.
|
|
"""
|
|
|
|
class EventType(models.TextChoices):
|
|
CREATED = 'CREATED', _('Created')
|
|
QUEUED = 'QUEUED', _('Queued')
|
|
SENDING = 'SENDING', _('Sending')
|
|
SENT = 'SENT', _('Sent')
|
|
DELIVERED = 'DELIVERED', _('Delivered')
|
|
FAILED = 'FAILED', _('Failed')
|
|
BOUNCED = 'BOUNCED', _('Bounced')
|
|
READ = 'READ', _('Read')
|
|
RETRY = 'RETRY', _('Retry')
|
|
|
|
message = models.ForeignKey(
|
|
Message,
|
|
on_delete=models.CASCADE,
|
|
related_name='logs',
|
|
verbose_name=_("Message")
|
|
)
|
|
event_type = models.CharField(
|
|
max_length=20,
|
|
choices=EventType.choices,
|
|
verbose_name=_("Event Type")
|
|
)
|
|
details = models.JSONField(
|
|
default=dict,
|
|
blank=True,
|
|
verbose_name=_("Details")
|
|
)
|
|
|
|
class Meta:
|
|
verbose_name = _("Message Log")
|
|
verbose_name_plural = _("Message Logs")
|
|
ordering = ['created_at']
|
|
indexes = [
|
|
models.Index(fields=['message', 'created_at']),
|
|
models.Index(fields=['event_type', 'created_at']),
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"{self.get_event_type_display()} - {self.message}"
|
|
|
|
|
|
class Notification(UUIDPrimaryKeyMixin, TimeStampedMixin):
|
|
"""
|
|
In-app notifications for staff members.
|
|
Used for internal system alerts, appointment notifications, and status updates.
|
|
|
|
Supports three types of notifications:
|
|
1. Personal: Targeted to a specific user (user field is set)
|
|
2. General: System-wide announcements visible to all users (is_general=True)
|
|
3. Role-based: Visible to all users with specific roles (target_roles is set)
|
|
"""
|
|
|
|
class NotificationType(models.TextChoices):
|
|
INFO = 'INFO', _('Info')
|
|
SUCCESS = 'SUCCESS', _('Success')
|
|
WARNING = 'WARNING', _('Warning')
|
|
ERROR = 'ERROR', _('Error')
|
|
|
|
user = models.ForeignKey(
|
|
settings.AUTH_USER_MODEL,
|
|
on_delete=models.CASCADE,
|
|
related_name='notifications',
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("User"),
|
|
help_text=_("Specific user for personal notifications. Leave empty for general/role-based notifications.")
|
|
)
|
|
is_general = models.BooleanField(
|
|
default=False,
|
|
verbose_name=_("Is General"),
|
|
help_text=_("If True, this notification is visible to all users (system-wide announcement)")
|
|
)
|
|
target_roles = models.JSONField(
|
|
default=list,
|
|
blank=True,
|
|
verbose_name=_("Target Roles"),
|
|
help_text=_("List of user roles that should see this notification (e.g., ['ADMIN', 'FRONT_DESK'])")
|
|
)
|
|
title = models.CharField(
|
|
max_length=200,
|
|
verbose_name=_("Title")
|
|
)
|
|
message = models.TextField(
|
|
verbose_name=_("Message")
|
|
)
|
|
notification_type = models.CharField(
|
|
max_length=20,
|
|
choices=NotificationType.choices,
|
|
default=NotificationType.INFO,
|
|
verbose_name=_("Notification Type")
|
|
)
|
|
is_read = models.BooleanField(
|
|
default=False,
|
|
verbose_name=_("Is Read")
|
|
)
|
|
read_at = models.DateTimeField(
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Read At")
|
|
)
|
|
related_object_type = models.CharField(
|
|
max_length=50,
|
|
blank=True,
|
|
help_text=_("Type of related object (e.g., 'appointment', 'invoice')"),
|
|
verbose_name=_("Related Object Type")
|
|
)
|
|
related_object_id = models.UUIDField(
|
|
null=True,
|
|
blank=True,
|
|
help_text=_("UUID of related object"),
|
|
verbose_name=_("Related Object ID")
|
|
)
|
|
action_url = models.CharField(
|
|
max_length=500,
|
|
blank=True,
|
|
help_text=_("URL to navigate to when notification is clicked"),
|
|
verbose_name=_("Action URL")
|
|
)
|
|
|
|
class Meta:
|
|
verbose_name = _("Notification")
|
|
verbose_name_plural = _("Notifications")
|
|
ordering = ['-created_at']
|
|
indexes = [
|
|
models.Index(fields=['user', 'is_read', 'created_at']),
|
|
models.Index(fields=['user', 'created_at']),
|
|
models.Index(fields=['notification_type', 'created_at']),
|
|
models.Index(fields=['related_object_type', 'related_object_id']),
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"{self.title} - {self.user.username}"
|
|
|
|
def mark_as_read(self):
|
|
"""Mark notification as read."""
|
|
if not self.is_read:
|
|
from django.utils import timezone
|
|
self.is_read = True
|
|
self.read_at = timezone.now()
|
|
self.save(update_fields=['is_read', 'read_at'])
|
|
|
|
@classmethod
|
|
def get_unread_count(cls, user):
|
|
"""Get count of unread notifications for a user."""
|
|
return cls.get_for_user(user).filter(is_read=False).count()
|
|
|
|
@classmethod
|
|
def mark_all_as_read(cls, user):
|
|
"""Mark all notifications as read for a user."""
|
|
from django.utils import timezone
|
|
cls.get_for_user(user).filter(is_read=False).update(
|
|
is_read=True,
|
|
read_at=timezone.now()
|
|
)
|
|
|
|
@classmethod
|
|
def get_for_user(cls, user):
|
|
"""
|
|
Get all notifications relevant to a user.
|
|
|
|
Returns notifications that are:
|
|
- Personal (user field matches)
|
|
- General (is_general=True)
|
|
- Role-based (user's role is in target_roles)
|
|
|
|
Args:
|
|
user: User instance
|
|
|
|
Returns:
|
|
QuerySet of Notification objects
|
|
"""
|
|
from django.db.models import Q
|
|
|
|
# First, get personal and general notifications (these are straightforward)
|
|
base_query = Q(user=user) | Q(is_general=True)
|
|
|
|
# Get all notifications that could potentially match
|
|
all_notifications = cls.objects.all()
|
|
|
|
# Filter in Python for role-based notifications
|
|
# This is necessary because SQLite doesn't support JSONField contains lookup
|
|
matching_ids = []
|
|
|
|
for notif in all_notifications:
|
|
# Include if it's a personal notification for this user
|
|
if notif.user == user:
|
|
matching_ids.append(notif.id)
|
|
# Include if it's a general notification
|
|
elif notif.is_general:
|
|
matching_ids.append(notif.id)
|
|
# Include if it's role-based and user's role matches
|
|
elif notif.target_roles and user.role and user.role in notif.target_roles:
|
|
matching_ids.append(notif.id)
|
|
|
|
return cls.objects.filter(id__in=matching_ids) if matching_ids else cls.objects.none()
|
|
|
|
@classmethod
|
|
def create_personal(cls, user, title, message, notification_type='INFO',
|
|
related_object_type='', related_object_id=None, action_url=''):
|
|
"""
|
|
Create a personal notification for a specific user.
|
|
|
|
Args:
|
|
user: User instance
|
|
title: Notification title
|
|
message: Notification message
|
|
notification_type: Type of notification (INFO, SUCCESS, WARNING, ERROR)
|
|
related_object_type: Type of related object (optional)
|
|
related_object_id: UUID of related object (optional)
|
|
action_url: URL to navigate to when clicked (optional)
|
|
|
|
Returns:
|
|
Created Notification instance
|
|
"""
|
|
return cls.objects.create(
|
|
user=user,
|
|
title=title,
|
|
message=message,
|
|
notification_type=notification_type,
|
|
related_object_type=related_object_type,
|
|
related_object_id=related_object_id,
|
|
action_url=action_url
|
|
)
|
|
|
|
@classmethod
|
|
def create_general(cls, title, message, notification_type='INFO', action_url=''):
|
|
"""
|
|
Create a general system-wide notification visible to all users.
|
|
|
|
Args:
|
|
title: Notification title
|
|
message: Notification message
|
|
notification_type: Type of notification (INFO, SUCCESS, WARNING, ERROR)
|
|
action_url: URL to navigate to when clicked (optional)
|
|
|
|
Returns:
|
|
Created Notification instance
|
|
"""
|
|
return cls.objects.create(
|
|
is_general=True,
|
|
title=title,
|
|
message=message,
|
|
notification_type=notification_type,
|
|
action_url=action_url
|
|
)
|
|
|
|
@classmethod
|
|
def create_role_based(cls, roles, title, message, notification_type='INFO',
|
|
related_object_type='', related_object_id=None, action_url=''):
|
|
"""
|
|
Create a role-based notification visible to users with specific roles.
|
|
|
|
Args:
|
|
roles: List of role codes (e.g., ['ADMIN', 'FRONT_DESK'])
|
|
title: Notification title
|
|
message: Notification message
|
|
notification_type: Type of notification (INFO, SUCCESS, WARNING, ERROR)
|
|
related_object_type: Type of related object (optional)
|
|
related_object_id: UUID of related object (optional)
|
|
action_url: URL to navigate to when clicked (optional)
|
|
|
|
Returns:
|
|
Created Notification instance
|
|
"""
|
|
if not isinstance(roles, list):
|
|
roles = [roles]
|
|
|
|
return cls.objects.create(
|
|
target_roles=roles,
|
|
title=title,
|
|
message=message,
|
|
notification_type=notification_type,
|
|
related_object_type=related_object_type,
|
|
related_object_id=related_object_id,
|
|
action_url=action_url
|
|
)
|