392 lines
15 KiB
Python
392 lines
15 KiB
Python
"""
|
|
Communications app forms for CRUD operations.
|
|
"""
|
|
|
|
from django import forms
|
|
from django.contrib.auth import get_user_model
|
|
from django.core.exceptions import ValidationError
|
|
import json
|
|
|
|
from .models import (
|
|
Message, MessageRecipient, NotificationTemplate, AlertRule,
|
|
AlertInstance, CommunicationChannel, DeliveryLog
|
|
)
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
class MessageForm(forms.ModelForm):
|
|
"""
|
|
Form for creating and updating messages.
|
|
"""
|
|
recipients = forms.ModelMultipleChoiceField(
|
|
queryset=User.objects.none(),
|
|
widget=forms.CheckboxSelectMultiple,
|
|
required=True,
|
|
help_text="Select message recipients"
|
|
)
|
|
|
|
class Meta:
|
|
model = Message
|
|
fields = [
|
|
'subject', 'content', 'message_type', 'priority',
|
|
'scheduled_at', 'recipients'
|
|
]
|
|
widgets = {
|
|
'subject': forms.TextInput(attrs={
|
|
'class': 'form-control',
|
|
'placeholder': 'Enter message subject'
|
|
}),
|
|
'content': forms.Textarea(attrs={
|
|
'class': 'form-control',
|
|
'rows': 8,
|
|
'placeholder': 'Enter message content'
|
|
}),
|
|
'message_type': forms.Select(attrs={'class': 'form-control'}),
|
|
'priority': forms.Select(attrs={'class': 'form-control'}),
|
|
'scheduled_at': forms.DateTimeInput(attrs={
|
|
'class': 'form-control',
|
|
'type': 'datetime-local'
|
|
}),
|
|
}
|
|
help_texts = {
|
|
'subject': 'Brief description of the message',
|
|
'body': 'Full message content',
|
|
'message_type': 'Type of message being sent',
|
|
'priority': 'Message priority level',
|
|
'template': 'Optional: Use a predefined template',
|
|
'scheduled_time': 'Optional: Schedule message for later delivery',
|
|
}
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
self.user = kwargs.pop('user', None)
|
|
super().__init__(*args, **kwargs)
|
|
|
|
if self.user:
|
|
# Filter recipients by tenant
|
|
self.fields['recipients'].queryset = User.objects.filter(
|
|
tenant=self.user.tenant,
|
|
is_active=True
|
|
).order_by('first_name', 'last_name')
|
|
|
|
# Filter templates by tenant
|
|
# self.fields['template_type'].queryset = NotificationTemplate.objects.filter(
|
|
# tenant=self.user.tenant,
|
|
# is_active=True
|
|
# ).order_by('name')
|
|
|
|
def clean_scheduled_time(self):
|
|
scheduled_time = self.cleaned_data.get('scheduled_time')
|
|
if scheduled_time:
|
|
from django.utils import timezone
|
|
if scheduled_time <= timezone.now():
|
|
raise ValidationError("Scheduled time must be in the future.")
|
|
return scheduled_time
|
|
|
|
def save(self, commit=True):
|
|
message = super().save(commit=commit)
|
|
|
|
if commit:
|
|
# Create MessageRecipient entries
|
|
recipients = self.cleaned_data.get('recipients', [])
|
|
MessageRecipient.objects.filter(message=message).delete()
|
|
|
|
for recipient in recipients:
|
|
MessageRecipient.objects.create(
|
|
tenant=message.tenant,
|
|
message=message,
|
|
recipient=recipient
|
|
)
|
|
|
|
return message
|
|
|
|
|
|
class NotificationTemplateForm(forms.ModelForm):
|
|
"""
|
|
Form for creating and updating notification templates.
|
|
"""
|
|
|
|
class Meta:
|
|
model = NotificationTemplate
|
|
fields = [
|
|
'name', 'description', 'template_type', 'category',
|
|
'subject_template', 'content_template', 'variables', 'is_active'
|
|
]
|
|
widgets = {
|
|
'name': forms.TextInput(attrs={
|
|
'class': 'form-control',
|
|
'placeholder': 'Enter template name'
|
|
}),
|
|
'description': forms.Textarea(attrs={
|
|
'class': 'form-control',
|
|
'rows': 3,
|
|
'placeholder': 'Describe the template purpose'
|
|
}),
|
|
'template_type': forms.Select(attrs={'class': 'form-control'}),
|
|
'category': forms.Select(attrs={'class': 'form-control'}),
|
|
'subject_template': forms.TextInput(attrs={
|
|
'class': 'form-control',
|
|
'placeholder': 'Subject template with variables like {{patient_name}}'
|
|
}),
|
|
'content_template': forms.Textarea(attrs={
|
|
'class': 'form-control',
|
|
'rows': 8,
|
|
'placeholder': 'Body template with variables like {{appointment_date}}'
|
|
}),
|
|
'variables': forms.Textarea(attrs={
|
|
'class': 'form-control',
|
|
'rows': 4,
|
|
'placeholder': 'JSON format: {"patient_name": "Patient Name", "appointment_date": "Appointment Date"}'
|
|
}),
|
|
'is_active': forms.CheckboxInput(attrs={'class': 'form-check-input'}),
|
|
}
|
|
help_texts = {
|
|
'name': 'Unique name for the template',
|
|
'description': 'Brief description of when to use this template',
|
|
'template_type': 'Type of notification this template is for',
|
|
'category': 'Category for organizing templates',
|
|
'subject_template': 'Subject line with variable placeholders',
|
|
'content_template': 'Message body with variable placeholders',
|
|
'variables': 'JSON object defining available variables and their descriptions',
|
|
'is_active': 'Whether this template is available for use',
|
|
}
|
|
|
|
def clean_variables(self):
|
|
variables = self.cleaned_data.get('variables')
|
|
if variables:
|
|
try:
|
|
json.loads(variables)
|
|
except json.JSONDecodeError:
|
|
raise ValidationError("Variables must be valid JSON format.")
|
|
return variables
|
|
|
|
def clean_template_name(self):
|
|
template_name = self.cleaned_data.get('template_name')
|
|
if template_name:
|
|
# Check for duplicate names within tenant (excluding current instance)
|
|
queryset = NotificationTemplate.objects.filter(
|
|
template_name=template_name
|
|
)
|
|
if self.instance.pk:
|
|
queryset = queryset.exclude(pk=self.instance.pk)
|
|
|
|
if queryset.exists():
|
|
raise ValidationError("A template with this name already exists.")
|
|
|
|
return template_name
|
|
|
|
|
|
class AlertRuleForm(forms.ModelForm):
|
|
"""
|
|
Form for creating and updating alert rules.
|
|
"""
|
|
|
|
class Meta:
|
|
model = AlertRule
|
|
fields = [
|
|
'name', 'description', 'trigger_type', 'severity',
|
|
'trigger_conditions', 'evaluation_frequency', 'cooldown_period',
|
|
'notification_template', 'notification_channels', 'is_active'
|
|
]
|
|
widgets = {
|
|
'name': forms.TextInput(attrs={
|
|
'class': 'form-control',
|
|
'placeholder': 'Enter alert rule name'
|
|
}),
|
|
'description': forms.Textarea(attrs={
|
|
'class': 'form-control',
|
|
'rows': 3,
|
|
'placeholder': 'Describe what this alert monitors'
|
|
}),
|
|
'trigger_type': forms.Select(attrs={'class': 'form-control'}),
|
|
'severity': forms.Select(attrs={'class': 'form-control'}),
|
|
'trigger_conditions': forms.Textarea(attrs={
|
|
'class': 'form-control',
|
|
'rows': 4,
|
|
'placeholder': 'JSON conditions that trigger the alert'
|
|
}),
|
|
'evaluation_frequency': forms.NumberInput(attrs={
|
|
'class': 'form-control',
|
|
'min': '60'
|
|
}),
|
|
'cooldown_period': forms.NumberInput(attrs={
|
|
'class': 'form-control',
|
|
'min': '300'
|
|
}),
|
|
'notification_template': forms.Select(attrs={'class': 'form-control'}),
|
|
'notification_channels': forms.Textarea(attrs={
|
|
'class': 'form-control',
|
|
'rows': 2,
|
|
'placeholder': 'JSON array of notification channels'
|
|
}),
|
|
'is_active': forms.CheckboxInput(attrs={'class': 'form-check-input'}),
|
|
}
|
|
help_texts = {
|
|
'name': 'Unique name for the alert rule',
|
|
'description': 'What condition this alert monitors',
|
|
'trigger_type': 'Type of alert trigger',
|
|
'severity': 'Severity level when alert is triggered',
|
|
'condition_query': 'Query or condition to evaluate for triggering',
|
|
'threshold_value': 'Numeric threshold for comparison',
|
|
'threshold_operator': 'Comparison operator for threshold',
|
|
'notification_template': 'Template to use for alert notifications',
|
|
'notification_channels': 'Channels to send notifications through',
|
|
'is_active': 'Whether this rule is actively monitoring',
|
|
}
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
|
|
# Filter templates and channels by tenant if user is available
|
|
if hasattr(self, 'user') and self.user:
|
|
self.fields['notification_template'].queryset = NotificationTemplate.objects.filter(
|
|
tenant=self.user.tenant,
|
|
is_active=True
|
|
).order_by('template_name')
|
|
|
|
self.fields['notification_channels'].queryset = CommunicationChannel.objects.filter(
|
|
tenant=self.user.tenant,
|
|
is_active=True
|
|
).order_by('channel_name')
|
|
|
|
def clean_rule_name(self):
|
|
rule_name = self.cleaned_data.get('rule_name')
|
|
if rule_name:
|
|
# Check for duplicate names within tenant (excluding current instance)
|
|
queryset = AlertRule.objects.filter(rule_name=rule_name)
|
|
if self.instance.pk:
|
|
queryset = queryset.exclude(pk=self.instance.pk)
|
|
|
|
if queryset.exists():
|
|
raise ValidationError("An alert rule with this name already exists.")
|
|
|
|
return rule_name
|
|
|
|
def clean_condition_query(self):
|
|
condition_query = self.cleaned_data.get('condition_query')
|
|
if condition_query:
|
|
# Basic validation for SQL injection prevention
|
|
dangerous_keywords = ['DROP', 'DELETE', 'UPDATE', 'INSERT', 'ALTER', 'CREATE']
|
|
query_upper = condition_query.upper()
|
|
for keyword in dangerous_keywords:
|
|
if keyword in query_upper:
|
|
raise ValidationError(f"Query cannot contain '{keyword}' statements.")
|
|
|
|
return condition_query
|
|
|
|
|
|
class CommunicationChannelForm(forms.ModelForm):
|
|
"""
|
|
Form for creating and updating communication channels.
|
|
"""
|
|
|
|
class Meta:
|
|
model = CommunicationChannel
|
|
fields = [
|
|
'name', 'description', 'channel_type',
|
|
'configuration', 'is_active'
|
|
]
|
|
widgets = {
|
|
'name': forms.TextInput(attrs={
|
|
'class': 'form-control',
|
|
'placeholder': 'Enter channel name'
|
|
}),
|
|
'description': forms.Textarea(attrs={
|
|
'class': 'form-control',
|
|
'rows': 3,
|
|
'placeholder': 'Describe this communication channel'
|
|
}),
|
|
'channel_type': forms.Select(attrs={'class': 'form-control'}),
|
|
'configuration': forms.Textarea(attrs={
|
|
'class': 'form-control',
|
|
'rows': 8,
|
|
'placeholder': 'JSON configuration for the channel'
|
|
}),
|
|
'is_active': forms.CheckboxInput(attrs={'class': 'form-check-input'}),
|
|
}
|
|
help_texts = {
|
|
'name': 'Unique name for the communication channel',
|
|
'description': 'Brief description of this channel',
|
|
'channel_type': 'Type of communication channel',
|
|
'configuration': 'JSON configuration specific to channel type',
|
|
'is_active': 'Whether this channel is available for use',
|
|
}
|
|
|
|
def clean_configuration(self):
|
|
configuration = self.cleaned_data.get('configuration')
|
|
if configuration:
|
|
try:
|
|
config_data = json.loads(configuration)
|
|
|
|
# Validate required fields based on channel type
|
|
channel_type = self.cleaned_data.get('channel_type')
|
|
if channel_type == 'EMAIL':
|
|
required_fields = ['smtp_server', 'smtp_port', 'username', 'password']
|
|
elif channel_type == 'SMS':
|
|
required_fields = ['api_key', 'sender_id']
|
|
elif channel_type == 'PUSH':
|
|
required_fields = ['api_key', 'app_id']
|
|
else:
|
|
required_fields = []
|
|
|
|
for field in required_fields:
|
|
if field not in config_data:
|
|
raise ValidationError(f"Configuration must include '{field}' for {channel_type} channels.")
|
|
|
|
except json.JSONDecodeError:
|
|
raise ValidationError("Configuration must be valid JSON format.")
|
|
|
|
return configuration
|
|
|
|
def clean_channel_name(self):
|
|
channel_name = self.cleaned_data.get('channel_name')
|
|
if channel_name:
|
|
# Check for duplicate names within tenant (excluding current instance)
|
|
queryset = CommunicationChannel.objects.filter(channel_name=channel_name)
|
|
if self.instance.pk:
|
|
queryset = queryset.exclude(pk=self.instance.pk)
|
|
|
|
if queryset.exists():
|
|
raise ValidationError("A communication channel with this name already exists.")
|
|
|
|
return channel_name
|
|
|
|
|
|
class MessageRecipientForm(forms.ModelForm):
|
|
"""
|
|
Form for managing message recipients (used in inline formsets).
|
|
"""
|
|
|
|
class Meta:
|
|
model = MessageRecipient
|
|
fields = ['recipient_type', 'user', 'email_address', 'phone_number']
|
|
widgets = {
|
|
'recipient_type': forms.Select(attrs={'class': 'form-control'}),
|
|
'user': forms.Select(attrs={'class': 'form-control'}),
|
|
'email_address': forms.EmailInput(attrs={'class': 'form-control'}),
|
|
'phone_number': forms.TextInput(attrs={'class': 'form-control'}),
|
|
}
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
self.user = kwargs.pop('user', None)
|
|
super().__init__(*args, **kwargs)
|
|
|
|
if self.user:
|
|
# Filter recipients by tenant
|
|
self.fields['user'].queryset = User.objects.filter(
|
|
tenant=self.user.tenant,
|
|
is_active=True
|
|
).order_by('first_name', 'last_name')
|
|
|
|
|
|
# Formset for managing message recipients
|
|
MessageRecipientFormSet = forms.inlineformset_factory(
|
|
Message,
|
|
MessageRecipient,
|
|
form=MessageRecipientForm,
|
|
extra=1,
|
|
can_delete=True,
|
|
fields=['recipient_type', 'user', 'email_address', 'phone_number']
|
|
)
|
|
|