""" Enhanced notification models for intelligent routing and management """ from django.db import models from django.contrib.auth import get_user_model from django.utils import timezone from django.contrib.contenttypes.models import ContentType from django.contrib.contenttypes.fields import GenericForeignKey import uuid import json User = get_user_model() class NotificationPreference(models.Model): """User notification preferences for different types and channels""" CHANNEL_CHOICES = [ ('in_app', 'In-App Notification'), ('email', 'Email'), ('sms', 'SMS'), ('push', 'Push Notification'), ] FREQUENCY_CHOICES = [ ('immediate', 'Immediate'), ('hourly', 'Hourly Digest'), ('daily', 'Daily Digest'), ('weekly', 'Weekly Digest'), ('never', 'Never'), ] id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) user = models.ForeignKey(User, on_delete=models.CASCADE, related_name='notification_preferences') notification_type = models.CharField(max_length=50) channel = models.CharField(max_length=20, choices=CHANNEL_CHOICES) frequency = models.CharField(max_length=20, choices=FREQUENCY_CHOICES, default='immediate') is_enabled = models.BooleanField(default=True) quiet_hours_start = models.TimeField(null=True, blank=True, help_text="Start of quiet hours (no notifications)") quiet_hours_end = models.TimeField(null=True, blank=True, help_text="End of quiet hours") created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: unique_together = ['user', 'notification_type', 'channel'] db_table = 'notification_preferences' def __str__(self): return f"{self.user.get_full_name()} - {self.notification_type} via {self.channel}" class NotificationRule(models.Model): """Rules for intelligent notification routing based on conditions""" CONDITION_TYPES = [ ('role', 'User Role'), ('branch', 'User Branch'), ('portfolio', 'Portfolio Assignment'), ('loan_amount', 'Loan Amount'), ('priority', 'Notification Priority'), ('time', 'Time-based'), ('custom', 'Custom Condition'), ] OPERATOR_CHOICES = [ ('equals', 'Equals'), ('not_equals', 'Not Equals'), ('greater_than', 'Greater Than'), ('less_than', 'Less Than'), ('contains', 'Contains'), ('in', 'In List'), ('not_in', 'Not In List'), ] id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) name = models.CharField(max_length=200) description = models.TextField(blank=True) notification_type = models.CharField(max_length=50) condition_type = models.CharField(max_length=20, choices=CONDITION_TYPES) condition_field = models.CharField(max_length=100, help_text="Field to check (e.g., 'user.role', 'loan.amount')") operator = models.CharField(max_length=20, choices=OPERATOR_CHOICES) condition_value = models.TextField(help_text="Value to compare against (JSON for complex values)") target_roles = models.JSONField(default=list, help_text="Roles to notify when condition is met") escalation_delay = models.IntegerField(default=0, help_text="Minutes to wait before escalation") escalation_roles = models.JSONField(default=list, help_text="Roles to escalate to if not acknowledged") is_active = models.BooleanField(default=True) priority = models.IntegerField(default=1, help_text="Rule priority (lower number = higher priority)") created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ['priority', 'name'] db_table = 'notification_rules' def __str__(self): return f"{self.name} - {self.notification_type}" def evaluate_condition(self, context): """Evaluate if the rule condition is met given the context""" try: # Get the field value from context field_parts = self.condition_field.split('.') value = context for part in field_parts: if hasattr(value, part): value = getattr(value, part) elif isinstance(value, dict) and part in value: value = value[part] else: return False # Parse condition value try: condition_value = json.loads(self.condition_value) except (json.JSONDecodeError, TypeError): condition_value = self.condition_value # Apply operator if self.operator == 'equals': return value == condition_value elif self.operator == 'not_equals': return value != condition_value elif self.operator == 'greater_than': return float(value) > float(condition_value) elif self.operator == 'less_than': return float(value) < float(condition_value) elif self.operator == 'contains': return str(condition_value).lower() in str(value).lower() elif self.operator == 'in': return value in condition_value if isinstance(condition_value, list) else False elif self.operator == 'not_in': return value not in condition_value if isinstance(condition_value, list) else True return False except Exception: return False class NotificationEscalation(models.Model): """Track notification escalations""" id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) original_notification = models.ForeignKey('utils.Notification', on_delete=models.CASCADE, related_name='escalations') rule = models.ForeignKey(NotificationRule, on_delete=models.CASCADE) escalated_to = models.ForeignKey(User, on_delete=models.CASCADE, related_name='escalated_notifications') escalated_at = models.DateTimeField(auto_now_add=True) acknowledged_at = models.DateTimeField(null=True, blank=True) resolved_at = models.DateTimeField(null=True, blank=True) escalation_level = models.IntegerField(default=1) notes = models.TextField(blank=True) class Meta: db_table = 'notification_escalations' ordering = ['-escalated_at'] def __str__(self): return f"Escalation {self.escalation_level} for {self.original_notification.title}" def acknowledge(self, notes=""): """Mark escalation as acknowledged""" self.acknowledged_at = timezone.now() self.notes = notes self.save() def resolve(self, notes=""): """Mark escalation as resolved""" self.resolved_at = timezone.now() if notes: self.notes = notes self.save() class NotificationTemplate(models.Model): """Templates for different notification types and channels""" CHANNEL_CHOICES = [ ('in_app', 'In-App Notification'), ('email', 'Email'), ('sms', 'SMS'), ('push', 'Push Notification'), ] id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) name = models.CharField(max_length=200) notification_type = models.CharField(max_length=50) channel = models.CharField(max_length=20, choices=CHANNEL_CHOICES) subject_template = models.CharField(max_length=200, blank=True, help_text="Subject template (for email/push)") title_template = models.CharField(max_length=200, help_text="Title template") message_template = models.TextField(help_text="Message template with placeholders") html_template = models.TextField(blank=True, help_text="HTML template for rich content") variables = models.JSONField(default=dict, help_text="Available template variables") is_active = models.BooleanField(default=True) is_default = models.BooleanField(default=False) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: unique_together = ['notification_type', 'channel', 'is_default'] db_table = 'notification_templates' ordering = ['notification_type', 'channel', 'name'] def __str__(self): return f"{self.name} - {self.notification_type} ({self.channel})" def render(self, context): """Render template with given context""" import re def replace_placeholder(match): placeholder = match.group(1) parts = placeholder.split('.') value = context try: for part in parts: if hasattr(value, part): value = getattr(value, part) elif isinstance(value, dict) and part in value: value = value[part] else: return match.group(0) # Return original if not found return str(value) except: return match.group(0) # Replace placeholders in format {{variable.field}} pattern = r'\{\{([^}]+)\}\}' rendered_title = re.sub(pattern, replace_placeholder, self.title_template) rendered_message = re.sub(pattern, replace_placeholder, self.message_template) rendered_subject = re.sub(pattern, replace_placeholder, self.subject_template) if self.subject_template else "" rendered_html = re.sub(pattern, replace_placeholder, self.html_template) if self.html_template else "" return { 'title': rendered_title, 'message': rendered_message, 'subject': rendered_subject, 'html': rendered_html, } class NotificationDelivery(models.Model): """Track notification delivery across different channels""" STATUS_CHOICES = [ ('pending', 'Pending'), ('sent', 'Sent'), ('delivered', 'Delivered'), ('failed', 'Failed'), ('bounced', 'Bounced'), ('read', 'Read'), ('clicked', 'Clicked'), ] id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) notification = models.ForeignKey('utils.Notification', on_delete=models.CASCADE, related_name='deliveries') channel = models.CharField(max_length=20, choices=NotificationTemplate.CHANNEL_CHOICES) recipient = models.ForeignKey(User, on_delete=models.CASCADE) status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='pending') sent_at = models.DateTimeField(null=True, blank=True) delivered_at = models.DateTimeField(null=True, blank=True) read_at = models.DateTimeField(null=True, blank=True) clicked_at = models.DateTimeField(null=True, blank=True) failed_at = models.DateTimeField(null=True, blank=True) error_message = models.TextField(blank=True) external_id = models.CharField(max_length=200, blank=True, help_text="External service message ID") metadata = models.JSONField(default=dict, help_text="Additional delivery metadata") created_at = models.DateTimeField(auto_now_add=True) class Meta: db_table = 'notification_deliveries' ordering = ['-created_at'] indexes = [ models.Index(fields=['notification', 'channel']), models.Index(fields=['recipient', 'status']), models.Index(fields=['status', 'created_at']), ] def __str__(self): return f"{self.notification.title} to {self.recipient.get_full_name()} via {self.channel}" def mark_sent(self, external_id=""): """Mark delivery as sent""" self.status = 'sent' self.sent_at = timezone.now() self.external_id = external_id self.save() def mark_delivered(self): """Mark delivery as delivered""" self.status = 'delivered' self.delivered_at = timezone.now() self.save() def mark_read(self): """Mark delivery as read""" self.status = 'read' self.read_at = timezone.now() self.save() def mark_clicked(self): """Mark delivery as clicked""" self.status = 'clicked' self.clicked_at = timezone.now() self.save() def mark_failed(self, error_message=""): """Mark delivery as failed""" self.status = 'failed' self.failed_at = timezone.now() self.error_message = error_message self.save() class NotificationBatch(models.Model): """Batch processing for bulk notifications""" STATUS_CHOICES = [ ('pending', 'Pending'), ('processing', 'Processing'), ('completed', 'Completed'), ('failed', 'Failed'), ('cancelled', 'Cancelled'), ] id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) name = models.CharField(max_length=200) notification_type = models.CharField(max_length=50) template = models.ForeignKey(NotificationTemplate, on_delete=models.CASCADE) target_criteria = models.JSONField(help_text="Criteria for selecting recipients") status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='pending') total_recipients = models.IntegerField(default=0) sent_count = models.IntegerField(default=0) failed_count = models.IntegerField(default=0) scheduled_at = models.DateTimeField(null=True, blank=True) started_at = models.DateTimeField(null=True, blank=True) completed_at = models.DateTimeField(null=True, blank=True) created_by = models.ForeignKey(User, on_delete=models.CASCADE, related_name='created_notification_batches') error_log = models.TextField(blank=True) created_at = models.DateTimeField(auto_now_add=True) class Meta: db_table = 'notification_batches' ordering = ['-created_at'] def __str__(self): return f"{self.name} - {self.status}" def start_processing(self): """Start batch processing""" self.status = 'processing' self.started_at = timezone.now() self.save() def complete_processing(self): """Complete batch processing""" self.status = 'completed' self.completed_at = timezone.now() self.save() def fail_processing(self, error_message=""): """Mark batch as failed""" self.status = 'failed' self.completed_at = timezone.now() self.error_log = error_message self.save()