2025-12-24 12:42:31 +03:00

202 lines
5.9 KiB
Python

"""
Notifications models - Multi-channel notification delivery
This module implements the notification system that:
- Sends SMS, WhatsApp, and Email notifications
- Logs all notification attempts
- Tracks delivery status
- Supports retry logic
"""
from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType
from django.db import models
from apps.core.models import BaseChoices, TimeStampedModel, UUIDModel
class NotificationChannel(BaseChoices):
"""Notification channel choices"""
SMS = 'sms', 'SMS'
WHATSAPP = 'whatsapp', 'WhatsApp'
EMAIL = 'email', 'Email'
PUSH = 'push', 'Push Notification'
class NotificationStatus(BaseChoices):
"""Notification delivery status"""
PENDING = 'pending', 'Pending'
SENDING = 'sending', 'Sending'
SENT = 'sent', 'Sent'
DELIVERED = 'delivered', 'Delivered'
FAILED = 'failed', 'Failed'
BOUNCED = 'bounced', 'Bounced'
class NotificationLog(UUIDModel, TimeStampedModel):
"""
Notification log - tracks all notification attempts.
Logs every SMS, WhatsApp, and Email sent by the system.
Used for:
- Delivery tracking
- Debugging
- Compliance
- Analytics
"""
# Channel and recipient
channel = models.CharField(
max_length=20,
choices=NotificationChannel.choices,
db_index=True
)
recipient = models.CharField(
max_length=200,
help_text="Phone number or email address"
)
# Message content
subject = models.CharField(max_length=500, blank=True)
message = models.TextField()
# Related object (generic foreign key)
content_type = models.ForeignKey(
ContentType,
on_delete=models.SET_NULL,
null=True,
blank=True
)
object_id = models.UUIDField(null=True, blank=True)
content_object = GenericForeignKey('content_type', 'object_id')
# Delivery status
status = models.CharField(
max_length=20,
choices=NotificationStatus.choices,
default=NotificationStatus.PENDING,
db_index=True
)
# Timestamps
sent_at = models.DateTimeField(null=True, blank=True)
delivered_at = models.DateTimeField(null=True, blank=True)
# Provider response
provider = models.CharField(
max_length=50,
blank=True,
help_text="SMS/Email provider used"
)
provider_message_id = models.CharField(
max_length=200,
blank=True,
help_text="Message ID from provider"
)
provider_response = models.JSONField(
default=dict,
blank=True,
help_text="Full response from provider"
)
# Error tracking
error = models.TextField(blank=True)
retry_count = models.IntegerField(default=0)
# Metadata
metadata = models.JSONField(
default=dict,
blank=True,
help_text="Additional metadata (campaign, template, etc.)"
)
class Meta:
ordering = ['-created_at']
indexes = [
models.Index(fields=['channel', 'status', '-created_at']),
models.Index(fields=['recipient', '-created_at']),
models.Index(fields=['content_type', 'object_id']),
]
def __str__(self):
return f"{self.channel} to {self.recipient} ({self.status})"
def mark_sent(self, provider_message_id=None):
"""Mark notification as sent"""
from django.utils import timezone
self.status = NotificationStatus.SENT
self.sent_at = timezone.now()
if provider_message_id:
self.provider_message_id = provider_message_id
self.save(update_fields=['status', 'sent_at', 'provider_message_id'])
def mark_delivered(self):
"""Mark notification as delivered"""
from django.utils import timezone
self.status = NotificationStatus.DELIVERED
self.delivered_at = timezone.now()
self.save(update_fields=['status', 'delivered_at'])
def mark_failed(self, error_message):
"""Mark notification as failed"""
self.status = NotificationStatus.FAILED
self.error = error_message
self.retry_count += 1
self.save(update_fields=['status', 'error', 'retry_count'])
class NotificationTemplate(UUIDModel, TimeStampedModel):
"""
Notification template for consistent messaging.
Supports:
- Bilingual templates (AR/EN)
- Variable substitution
- Multiple channels
"""
name = models.CharField(max_length=200, unique=True)
description = models.TextField(blank=True)
# Template type
template_type = models.CharField(
max_length=50,
choices=[
('survey_invitation', 'Survey Invitation'),
('survey_reminder', 'Survey Reminder'),
('complaint_acknowledgment', 'Complaint Acknowledgment'),
('complaint_update', 'Complaint Update'),
('action_assignment', 'Action Assignment'),
('sla_reminder', 'SLA Reminder'),
('sla_breach', 'SLA Breach'),
],
db_index=True
)
# Channel-specific templates
sms_template = models.TextField(
blank=True,
help_text="SMS template with {{variables}}"
)
sms_template_ar = models.TextField(blank=True)
whatsapp_template = models.TextField(
blank=True,
help_text="WhatsApp template with {{variables}}"
)
whatsapp_template_ar = models.TextField(blank=True)
email_subject = models.CharField(max_length=500, blank=True)
email_subject_ar = models.CharField(max_length=500, blank=True)
email_template = models.TextField(
blank=True,
help_text="Email HTML template with {{variables}}"
)
email_template_ar = models.TextField(blank=True)
# Configuration
is_active = models.BooleanField(default=True)
class Meta:
ordering = ['name']
def __str__(self):
return self.name