450 lines
13 KiB
Python
450 lines
13 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.
|
|
"""
|
|
|
|
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',
|
|
verbose_name=_("User")
|
|
)
|
|
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.objects.filter(user=user, 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.objects.filter(user=user, is_read=False).update(
|
|
is_read=True,
|
|
read_at=timezone.now()
|
|
)
|