""" Communications app models. """ import uuid from django.db import models from django.contrib.auth import get_user_model from django.utils import timezone from django.core.validators import MinValueValidator, MaxValueValidator from decimal import Decimal User = get_user_model() class Message(models.Model): """ Model for internal messaging system. """ MESSAGE_TYPES = [ ('INTERNAL', 'Internal Message'), ('EMAIL', 'Email'), ('SMS', 'SMS'), ('PUSH', 'Push Notification'), ('SLACK', 'Slack Message'), ('TEAMS', 'Microsoft Teams'), ('WEBHOOK', 'Webhook'), ('SYSTEM', 'System Message'), ('ALERT', 'Alert Message'), ] PRIORITY_LEVELS = [ ('LOW', 'Low'), ('NORMAL', 'Normal'), ('HIGH', 'High'), ('URGENT', 'Urgent'), ('CRITICAL', 'Critical'), ] STATUS_CHOICES = [ ('DRAFT', 'Draft'), ('PENDING', 'Pending'), ('SENDING', 'Sending'), ('SENT', 'Sent'), ('DELIVERED', 'Delivered'), ('READ', 'Read'), ('FAILED', 'Failed'), ('CANCELLED', 'Cancelled'), ] # Primary identification message_id = models.UUIDField( primary_key=True, default=uuid.uuid4, editable=False, help_text="Unique identifier for the message" ) tenant = models.ForeignKey( 'core.Tenant', on_delete=models.CASCADE, help_text="Tenant organization" ) # Message details subject = models.CharField( max_length=255, help_text="Message subject line" ) content = models.TextField( help_text="Message content/body" ) message_type = models.CharField( max_length=20, choices=MESSAGE_TYPES, default='INTERNAL', help_text="Type of message" ) priority = models.CharField( max_length=20, choices=PRIORITY_LEVELS, default='NORMAL', help_text="Message priority level" ) # Sender and recipients sender = models.ForeignKey( User, on_delete=models.CASCADE, related_name='sent_messages', help_text="Message sender" ) # Message status status = models.CharField( max_length=20, choices=STATUS_CHOICES, default='DRAFT', help_text="Message status" ) # Timing created_at = models.DateTimeField( auto_now_add=True, help_text="Message creation timestamp" ) scheduled_at = models.DateTimeField( null=True, blank=True, help_text="Scheduled send time" ) sent_at = models.DateTimeField( null=True, blank=True, help_text="Actual send timestamp" ) expires_at = models.DateTimeField( null=True, blank=True, help_text="Message expiration time" ) # Message configuration is_urgent = models.BooleanField( default=False, help_text="Urgent message flag" ) requires_acknowledgment = models.BooleanField( default=False, help_text="Requires recipient acknowledgment" ) is_confidential = models.BooleanField( default=False, help_text="Confidential message flag" ) # Delivery tracking delivery_attempts = models.PositiveIntegerField( default=0, help_text="Number of delivery attempts" ) max_delivery_attempts = models.PositiveIntegerField( default=3, help_text="Maximum delivery attempts" ) # Message metadata message_thread_id = models.UUIDField( null=True, blank=True, help_text="Thread ID for message grouping" ) reply_to_message = models.ForeignKey( 'self', on_delete=models.SET_NULL, null=True, blank=True, help_text="Original message if this is a reply" ) # External references external_message_id = models.CharField( max_length=255, null=True, blank=True, help_text="External system message ID" ) # Attachments and formatting has_attachments = models.BooleanField( default=False, help_text="Message has attachments" ) content_type = models.CharField( max_length=50, default='text/plain', help_text="Content MIME type" ) class Meta: db_table = 'communications_message' indexes = [ models.Index(fields=['tenant', 'status']), models.Index(fields=['sender', 'created_at']), models.Index(fields=['message_type', 'priority']), models.Index(fields=['scheduled_at']), models.Index(fields=['message_thread_id']), ] ordering = ['-created_at'] def __str__(self): return f"{self.subject} ({self.message_type})" @property def is_overdue(self): """Check if message is overdue for delivery.""" if self.scheduled_at and self.status in ['PENDING', 'SENDING']: return timezone.now() > self.scheduled_at return False @property def is_expired(self): """Check if message has expired.""" if self.expires_at: return timezone.now() > self.expires_at return False class MessageRecipient(models.Model): """ Model for message recipients. """ RECIPIENT_TYPES = [ ('USER', 'User'), ('EMAIL', 'Email Address'), ('PHONE', 'Phone Number'), ('ROLE', 'User Role'), ('DEPARTMENT', 'Department'), ('GROUP', 'User Group'), ] STATUS_CHOICES = [ ('PENDING', 'Pending'), ('SENT', 'Sent'), ('DELIVERED', 'Delivered'), ('READ', 'Read'), ('ACKNOWLEDGED', 'Acknowledged'), ('FAILED', 'Failed'), ('BOUNCED', 'Bounced'), ('UNSUBSCRIBED', 'Unsubscribed'), ] # Primary identification recipient_id = models.UUIDField( primary_key=True, default=uuid.uuid4, editable=False, help_text="Unique identifier for the recipient" ) message = models.ForeignKey( Message, on_delete=models.CASCADE, related_name='recipients', help_text="Associated message" ) # Recipient details recipient_type = models.CharField( max_length=20, choices=RECIPIENT_TYPES, help_text="Type of recipient" ) user = models.ForeignKey( User, on_delete=models.CASCADE, null=True, blank=True, help_text="User recipient" ) email_address = models.EmailField( null=True, blank=True, help_text="Email address recipient" ) phone_number = models.CharField( max_length=20, null=True, blank=True, help_text="Phone number recipient" ) role_name = models.CharField( max_length=100, null=True, blank=True, help_text="Role name for role-based recipients" ) # Delivery status status = models.CharField( max_length=20, choices=STATUS_CHOICES, default='PENDING', help_text="Delivery status" ) # Timing sent_at = models.DateTimeField( null=True, blank=True, help_text="Sent timestamp" ) delivered_at = models.DateTimeField( null=True, blank=True, help_text="Delivered timestamp" ) read_at = models.DateTimeField( null=True, blank=True, help_text="Read timestamp" ) acknowledged_at = models.DateTimeField( null=True, blank=True, help_text="Acknowledged timestamp" ) # Delivery tracking delivery_attempts = models.PositiveIntegerField( default=0, help_text="Number of delivery attempts" ) last_attempt_at = models.DateTimeField( null=True, blank=True, help_text="Last delivery attempt timestamp" ) error_message = models.TextField( null=True, blank=True, help_text="Last delivery error message" ) # External tracking external_delivery_id = models.CharField( max_length=255, null=True, blank=True, help_text="External delivery tracking ID" ) class Meta: db_table = 'communications_message_recipient' indexes = [ models.Index(fields=['message', 'status']), models.Index(fields=['user', 'status']), models.Index(fields=['recipient_type']), models.Index(fields=['sent_at']), ] unique_together = [ ['message', 'user'], ['message', 'email_address'], ['message', 'phone_number'], ] def __str__(self): if self.user: return f"{self.user.get_full_name()} - {self.message.subject}" elif self.email_address: return f"{self.email_address} - {self.message.subject}" elif self.phone_number: return f"{self.phone_number} - {self.message.subject}" return f"{self.recipient_type} - {self.message.subject}" class NotificationTemplate(models.Model): """ Model for notification templates. """ TEMPLATE_TYPES = [ ('EMAIL', 'Email Template'), ('SMS', 'SMS Template'), ('PUSH', 'Push Notification Template'), ('SLACK', 'Slack Template'), ('TEAMS', 'Teams Template'), ('WEBHOOK', 'Webhook Template'), ('SYSTEM', 'System Notification Template'), ] TEMPLATE_CATEGORIES = [ ('APPOINTMENT', 'Appointment Notifications'), ('MEDICATION', 'Medication Reminders'), ('LAB_RESULTS', 'Lab Results'), ('BILLING', 'Billing Notifications'), ('EMERGENCY', 'Emergency Alerts'), ('SYSTEM', 'System Notifications'), ('MARKETING', 'Marketing Communications'), ('CLINICAL', 'Clinical Notifications'), ('ADMINISTRATIVE', 'Administrative Messages'), ('QUALITY', 'Quality Alerts'), ] # Primary identification template_id = models.UUIDField( primary_key=True, default=uuid.uuid4, editable=False, help_text="Unique identifier for the template" ) tenant = models.ForeignKey( 'core.Tenant', on_delete=models.CASCADE, help_text="Tenant organization" ) # Template details name = models.CharField( max_length=255, help_text="Template name" ) description = models.TextField( null=True, blank=True, help_text="Template description" ) template_type = models.CharField( max_length=20, choices=TEMPLATE_TYPES, help_text="Type of template" ) category = models.CharField( max_length=30, choices=TEMPLATE_CATEGORIES, help_text="Template category" ) # Template content subject_template = models.CharField( max_length=255, null=True, blank=True, help_text="Subject line template" ) content_template = models.TextField( help_text="Message content template" ) # Template configuration variables = models.JSONField( default=dict, help_text="Available template variables" ) default_values = models.JSONField( default=dict, help_text="Default variable values" ) formatting_rules = models.JSONField( default=dict, help_text="Content formatting rules" ) # Template settings is_active = models.BooleanField( default=True, help_text="Template is active" ) is_system_template = models.BooleanField( default=False, help_text="System-defined template" ) requires_approval = models.BooleanField( default=False, help_text="Requires approval before use" ) # Usage tracking usage_count = models.PositiveIntegerField( default=0, help_text="Number of times template has been used" ) last_used_at = models.DateTimeField( null=True, blank=True, help_text="Last usage timestamp" ) # Metadata created_at = models.DateTimeField( auto_now_add=True, help_text="Template creation timestamp" ) updated_at = models.DateTimeField( auto_now=True, help_text="Last update timestamp" ) created_by = models.ForeignKey( User, on_delete=models.SET_NULL, null=True, blank=True, help_text="Template creator" ) class Meta: db_table = 'communications_notification_template' indexes = [ models.Index(fields=['tenant', 'template_type']), models.Index(fields=['category', 'is_active']), models.Index(fields=['is_system_template']), models.Index(fields=['usage_count']), ] unique_together = [['tenant', 'name', 'template_type']] def __str__(self): return f"{self.name} ({self.template_type})" class AlertRule(models.Model): """ Model for automated alert rules. """ TRIGGER_TYPES = [ ('THRESHOLD', 'Threshold Alert'), ('PATTERN', 'Pattern Alert'), ('SCHEDULE', 'Scheduled Alert'), ('EVENT', 'Event-based Alert'), ('ANOMALY', 'Anomaly Detection'), ('SYSTEM', 'System Alert'), ('CLINICAL', 'Clinical Alert'), ('OPERATIONAL', 'Operational Alert'), ] SEVERITY_LEVELS = [ ('INFO', 'Information'), ('WARNING', 'Warning'), ('ERROR', 'Error'), ('CRITICAL', 'Critical'), ('EMERGENCY', 'Emergency'), ] # Primary identification rule_id = models.UUIDField( primary_key=True, default=uuid.uuid4, editable=False, help_text="Unique identifier for the alert rule" ) tenant = models.ForeignKey( 'core.Tenant', on_delete=models.CASCADE, help_text="Tenant organization" ) # Rule details name = models.CharField( max_length=255, help_text="Alert rule name" ) description = models.TextField( null=True, blank=True, help_text="Alert rule description" ) trigger_type = models.CharField( max_length=20, choices=TRIGGER_TYPES, help_text="Type of alert trigger" ) severity = models.CharField( max_length=20, choices=SEVERITY_LEVELS, default='WARNING', help_text="Alert severity level" ) # Rule configuration trigger_conditions = models.JSONField( default=dict, help_text="Conditions that trigger the alert" ) evaluation_frequency = models.PositiveIntegerField( default=300, # 5 minutes help_text="Evaluation frequency in seconds" ) cooldown_period = models.PositiveIntegerField( default=3600, # 1 hour help_text="Cooldown period between alerts in seconds" ) # Notification configuration notification_template = models.ForeignKey( NotificationTemplate, on_delete=models.SET_NULL, null=True, blank=True, help_text="Notification template to use" ) notification_channels = models.JSONField( default=list, help_text="Notification channels to use" ) escalation_rules = models.JSONField( default=dict, help_text="Escalation configuration" ) # Recipients default_recipients = models.ManyToManyField( User, blank=True, help_text="Default alert recipients" ) recipient_roles = models.JSONField( default=list, help_text="Recipient roles" ) # Rule status is_active = models.BooleanField( default=True, help_text="Alert rule is active" ) is_system_rule = models.BooleanField( default=False, help_text="System-defined rule" ) # Tracking trigger_count = models.PositiveIntegerField( default=0, help_text="Number of times rule has triggered" ) last_triggered_at = models.DateTimeField( null=True, blank=True, help_text="Last trigger timestamp" ) last_evaluated_at = models.DateTimeField( null=True, blank=True, help_text="Last evaluation timestamp" ) # Metadata created_at = models.DateTimeField( auto_now_add=True, help_text="Rule creation timestamp" ) updated_at = models.DateTimeField( auto_now=True, help_text="Last update timestamp" ) created_by = models.ForeignKey( User, on_delete=models.SET_NULL, null=True, blank=True, related_name='created_alert_rules', help_text="Rule creator" ) class Meta: db_table = 'communications_alert_rule' indexes = [ models.Index(fields=['tenant', 'is_active']), models.Index(fields=['trigger_type', 'severity']), models.Index(fields=['last_evaluated_at']), models.Index(fields=['is_system_rule']), ] unique_together = [['tenant', 'name']] def __str__(self): return f"{self.name} ({self.severity})" @property def is_in_cooldown(self): """Check if rule is in cooldown period.""" if self.last_triggered_at: cooldown_end = self.last_triggered_at + timezone.timedelta( seconds=self.cooldown_period ) return timezone.now() < cooldown_end return False class AlertInstance(models.Model): """ Model for alert instances. """ STATUS_CHOICES = [ ('ACTIVE', 'Active'), ('ACKNOWLEDGED', 'Acknowledged'), ('RESOLVED', 'Resolved'), ('SUPPRESSED', 'Suppressed'), ('ESCALATED', 'Escalated'), ('EXPIRED', 'Expired'), ] # Primary identification alert_id = models.UUIDField( primary_key=True, default=uuid.uuid4, editable=False, help_text="Unique identifier for the alert instance" ) alert_rule = models.ForeignKey( AlertRule, on_delete=models.CASCADE, related_name='instances', help_text="Associated alert rule" ) # Alert details title = models.CharField( max_length=255, help_text="Alert title" ) description = models.TextField( help_text="Alert description" ) severity = models.CharField( max_length=20, choices=AlertRule.SEVERITY_LEVELS, help_text="Alert severity level" ) # Alert data trigger_data = models.JSONField( default=dict, help_text="Data that triggered the alert" ) context_data = models.JSONField( default=dict, help_text="Additional context data" ) # Alert status status = models.CharField( max_length=20, choices=STATUS_CHOICES, default='ACTIVE', help_text="Alert status" ) # Timing triggered_at = models.DateTimeField( auto_now_add=True, help_text="Alert trigger timestamp" ) acknowledged_at = models.DateTimeField( null=True, blank=True, help_text="Acknowledgment timestamp" ) resolved_at = models.DateTimeField( null=True, blank=True, help_text="Resolution timestamp" ) expires_at = models.DateTimeField( null=True, blank=True, help_text="Alert expiration time" ) # Response tracking acknowledged_by = models.ForeignKey( User, on_delete=models.SET_NULL, null=True, blank=True, related_name='acknowledged_alerts', help_text="User who acknowledged the alert" ) resolved_by = models.ForeignKey( User, on_delete=models.SET_NULL, null=True, blank=True, related_name='resolved_alerts', help_text="User who resolved the alert" ) resolution_notes = models.TextField( null=True, blank=True, help_text="Resolution notes" ) # Escalation tracking escalation_level = models.PositiveIntegerField( default=0, help_text="Current escalation level" ) escalated_at = models.DateTimeField( null=True, blank=True, help_text="Last escalation timestamp" ) # Notification tracking notifications_sent = models.PositiveIntegerField( default=0, help_text="Number of notifications sent" ) last_notification_at = models.DateTimeField( null=True, blank=True, help_text="Last notification timestamp" ) class Meta: db_table = 'communications_alert_instance' indexes = [ models.Index(fields=['alert_rule', 'status']), models.Index(fields=['severity', 'triggered_at']), models.Index(fields=['status', 'triggered_at']), models.Index(fields=['expires_at']), ] ordering = ['-triggered_at'] def __str__(self): return f"{self.title} ({self.severity})" @property def is_expired(self): """Check if alert has expired.""" if self.expires_at: return timezone.now() > self.expires_at return False @property def duration(self): """Get alert duration.""" end_time = self.resolved_at or timezone.now() return end_time - self.triggered_at class CommunicationChannel(models.Model): """ Model for communication channels. """ CHANNEL_TYPES = [ ('EMAIL', 'Email'), ('SMS', 'SMS'), ('PUSH', 'Push Notification'), ('SLACK', 'Slack'), ('TEAMS', 'Microsoft Teams'), ('WEBHOOK', 'Webhook'), ('PHONE', 'Phone Call'), ('FAX', 'Fax'), ('PAGER', 'Pager'), ] PROVIDER_TYPES = [ ('SMTP', 'SMTP Email'), ('SENDGRID', 'SendGrid'), ('MAILGUN', 'Mailgun'), ('TWILIO', 'Twilio SMS'), ('AWS_SNS', 'AWS SNS'), ('FIREBASE', 'Firebase'), ('SLACK_API', 'Slack API'), ('TEAMS_API', 'Teams API'), ('WEBHOOK', 'Webhook'), ('CUSTOM', 'Custom Provider'), ] # Primary identification channel_id = models.UUIDField( primary_key=True, default=uuid.uuid4, editable=False, help_text="Unique identifier for the channel" ) tenant = models.ForeignKey( 'core.Tenant', on_delete=models.CASCADE, help_text="Tenant organization" ) # Channel details name = models.CharField( max_length=255, help_text="Channel name" ) description = models.TextField( null=True, blank=True, help_text="Channel description" ) channel_type = models.CharField( max_length=20, choices=CHANNEL_TYPES, help_text="Type of communication channel" ) provider_type = models.CharField( max_length=20, choices=PROVIDER_TYPES, help_text="Provider type" ) # Configuration configuration = models.JSONField( default=dict, help_text="Channel configuration settings" ) authentication_config = models.JSONField( default=dict, help_text="Authentication configuration" ) rate_limits = models.JSONField( default=dict, help_text="Rate limiting configuration" ) # Channel status is_active = models.BooleanField( default=True, help_text="Channel is active" ) is_healthy = models.BooleanField( default=True, help_text="Channel health status" ) last_health_check = models.DateTimeField( null=True, blank=True, help_text="Last health check timestamp" ) health_check_interval = models.PositiveIntegerField( default=300, # 5 minutes help_text="Health check interval in seconds" ) # Usage tracking message_count = models.PositiveIntegerField( default=0, help_text="Total messages sent through channel" ) success_count = models.PositiveIntegerField( default=0, help_text="Successful message deliveries" ) failure_count = models.PositiveIntegerField( default=0, help_text="Failed message deliveries" ) last_used_at = models.DateTimeField( null=True, blank=True, help_text="Last usage timestamp" ) # Metadata created_at = models.DateTimeField( auto_now_add=True, help_text="Channel creation timestamp" ) updated_at = models.DateTimeField( auto_now=True, help_text="Last update timestamp" ) created_by = models.ForeignKey( User, on_delete=models.SET_NULL, null=True, blank=True, help_text="Channel creator" ) class Meta: db_table = 'communications_communication_channel' indexes = [ models.Index(fields=['tenant', 'channel_type']), models.Index(fields=['is_active', 'is_healthy']), models.Index(fields=['last_health_check']), models.Index(fields=['provider_type']), ] unique_together = [['tenant', 'name']] def __str__(self): return f"{self.name} ({self.channel_type})" @property def success_rate(self): """Calculate success rate.""" if self.message_count > 0: return (self.success_count / self.message_count) * 100 return 0 @property def needs_health_check(self): """Check if health check is needed.""" if not self.last_health_check: return True next_check = self.last_health_check + timezone.timedelta( seconds=self.health_check_interval ) return timezone.now() >= next_check class DeliveryLog(models.Model): """ Model for delivery logging. """ STATUS_CHOICES = [ ('PENDING', 'Pending'), ('PROCESSING', 'Processing'), ('SENT', 'Sent'), ('DELIVERED', 'Delivered'), ('FAILED', 'Failed'), ('BOUNCED', 'Bounced'), ('REJECTED', 'Rejected'), ('TIMEOUT', 'Timeout'), ] # Primary identification log_id = models.UUIDField( primary_key=True, default=uuid.uuid4, editable=False, help_text="Unique identifier for the delivery log" ) message = models.ForeignKey( Message, on_delete=models.CASCADE, related_name='delivery_logs', help_text="Associated message" ) recipient = models.ForeignKey( MessageRecipient, on_delete=models.CASCADE, related_name='delivery_logs', help_text="Associated recipient" ) channel = models.ForeignKey( CommunicationChannel, on_delete=models.CASCADE, related_name='delivery_logs', help_text="Communication channel used" ) # Delivery details status = models.CharField( max_length=20, choices=STATUS_CHOICES, default='PENDING', help_text="Delivery status" ) attempt_number = models.PositiveIntegerField( default=1, help_text="Delivery attempt number" ) # Timing started_at = models.DateTimeField( auto_now_add=True, help_text="Delivery start timestamp" ) completed_at = models.DateTimeField( null=True, blank=True, help_text="Delivery completion timestamp" ) # Response tracking external_id = models.CharField( max_length=255, null=True, blank=True, help_text="External delivery ID" ) response_code = models.CharField( max_length=50, null=True, blank=True, help_text="Response code from provider" ) response_message = models.TextField( null=True, blank=True, help_text="Response message from provider" ) error_details = models.JSONField( default=dict, help_text="Detailed error information" ) # Performance metrics processing_time_ms = models.PositiveIntegerField( null=True, blank=True, help_text="Processing time in milliseconds" ) payload_size_bytes = models.PositiveIntegerField( null=True, blank=True, help_text="Payload size in bytes" ) # Metadata metadata = models.JSONField( default=dict, help_text="Additional delivery metadata" ) class Meta: db_table = 'communications_delivery_log' indexes = [ models.Index(fields=['message', 'status']), models.Index(fields=['recipient', 'status']), models.Index(fields=['channel', 'started_at']), models.Index(fields=['status', 'started_at']), models.Index(fields=['external_id']), ] ordering = ['-started_at'] def __str__(self): return f"{self.message.subject} - {self.recipient} ({self.status})" @property def duration(self): """Get delivery duration.""" if self.completed_at: return self.completed_at - self.started_at return timezone.now() - self.started_at