250 lines
8.0 KiB
Python
250 lines
8.0 KiB
Python
"""
|
|
Notification processing service.
|
|
Matches events against rules and generates notifications for recipients.
|
|
"""
|
|
from typing import List, Dict, Any
|
|
|
|
from django.contrib.contenttypes.models import ContentType
|
|
|
|
from core.models.enums import NotificationStatusChoices, DeliveryStatusChoices
|
|
from core.models.events import Event, NotificationRule, Notification, NotificationDelivery
|
|
from core.models.profile import TeamProfile
|
|
|
|
|
|
class NotificationProcessor:
|
|
"""
|
|
Processes events and generates notifications based on rules.
|
|
"""
|
|
|
|
@staticmethod
|
|
def process_event(event: Event) -> List[Notification]:
|
|
"""
|
|
Process an event and generate notifications based on active rules.
|
|
|
|
Args:
|
|
event: The Event instance to process
|
|
|
|
Returns:
|
|
List of created Notification instances
|
|
"""
|
|
# Find active rules matching this event type
|
|
matching_rules = NotificationRule.objects.filter(
|
|
is_active=True,
|
|
event_types__contains=[event.event_type]
|
|
)
|
|
|
|
notifications = []
|
|
|
|
for rule in matching_rules:
|
|
# Check conditions if any
|
|
if rule.conditions and not NotificationProcessor._check_conditions(event, rule.conditions):
|
|
continue
|
|
|
|
# Get recipients for this rule
|
|
recipients = NotificationProcessor._get_recipients(rule, event)
|
|
|
|
# Generate notification subject and body
|
|
subject = NotificationProcessor._render_template(rule.template_subject, event)
|
|
body = NotificationProcessor._render_template(rule.template_body, event)
|
|
|
|
# Create action URL if applicable
|
|
action_url = NotificationProcessor._generate_action_url(event)
|
|
|
|
# Create notifications for each recipient
|
|
for recipient in recipients:
|
|
notification = NotificationProcessor._create_notification(
|
|
event=event,
|
|
rule=rule,
|
|
recipient=recipient,
|
|
subject=subject,
|
|
body=body,
|
|
action_url=action_url
|
|
)
|
|
notifications.append(notification)
|
|
|
|
# Queue delivery tasks for each channel
|
|
for channel in rule.channels:
|
|
NotificationProcessor._create_delivery(notification, channel)
|
|
|
|
return notifications
|
|
|
|
@staticmethod
|
|
def _check_conditions(event: Event, conditions: Dict[str, Any]) -> bool:
|
|
"""
|
|
Check if event metadata matches rule conditions.
|
|
|
|
Args:
|
|
event: Event to check
|
|
conditions: Conditions from NotificationRule
|
|
|
|
Returns:
|
|
True if conditions are met, False otherwise
|
|
"""
|
|
for key, value in conditions.items():
|
|
if event.metadata.get(key) != value:
|
|
return False
|
|
return True
|
|
|
|
@staticmethod
|
|
def _get_recipients(rule: NotificationRule, event: Event) -> List[Any]:
|
|
"""
|
|
Get list of recipients based on rule configuration.
|
|
|
|
Args:
|
|
rule: NotificationRule to process
|
|
event: Event being processed
|
|
|
|
Returns:
|
|
List of profile instances (TeamProfile or CustomerProfile)
|
|
"""
|
|
recipients = []
|
|
|
|
# If specific profiles are targeted, use them
|
|
team_profiles = list(rule.target_team_profiles.all())
|
|
customer_profiles = list(rule.target_customer_profiles.all())
|
|
|
|
if team_profiles or customer_profiles:
|
|
recipients.extend(team_profiles)
|
|
recipients.extend(customer_profiles)
|
|
# Otherwise, use role-based targeting
|
|
elif rule.target_roles:
|
|
recipients.extend(
|
|
TeamProfile.objects.filter(role__in=rule.target_roles)
|
|
)
|
|
# If no specific targeting, notify all team admins by default
|
|
else:
|
|
from core.models.enums import RoleChoices
|
|
recipients.extend(
|
|
TeamProfile.objects.filter(role=RoleChoices.ADMIN)
|
|
)
|
|
|
|
return recipients
|
|
|
|
@staticmethod
|
|
def _render_template(template: str, event: Event) -> str:
|
|
"""
|
|
Render a template string with event data.
|
|
|
|
Args:
|
|
template: Template string (supports simple variable substitution)
|
|
event: Event instance
|
|
|
|
Returns:
|
|
Rendered string
|
|
"""
|
|
if not template:
|
|
# Generate default message
|
|
return NotificationProcessor._generate_default_message(event)
|
|
|
|
# Simple template variable substitution
|
|
# Supports: {event_type}, {entity_type}, {entity_id}, and metadata fields
|
|
context = {
|
|
'event_type': event.event_type, # Use raw value instead of get_event_type_display()
|
|
'entity_type': event.entity_type,
|
|
'entity_id': str(event.entity_id),
|
|
**event.metadata
|
|
}
|
|
|
|
try:
|
|
return template.format(**context)
|
|
except KeyError as e:
|
|
# If template has unknown variables, try to provide defaults
|
|
# Add empty string for missing keys
|
|
import re
|
|
missing_keys = re.findall(r'\{(\w+)\}', template)
|
|
for key in missing_keys:
|
|
if key not in context:
|
|
context[key] = f'[{key} not available]'
|
|
|
|
try:
|
|
return template.format(**context)
|
|
except:
|
|
# If still fails, return template as-is
|
|
return template
|
|
|
|
@staticmethod
|
|
def _generate_default_message(event: Event) -> str:
|
|
"""Generate default notification message for an event"""
|
|
return f"{event.event_type}: {event.entity_type} {event.entity_id}"
|
|
|
|
@staticmethod
|
|
def _generate_action_url(event: Event) -> str:
|
|
"""
|
|
Generate action URL for the event entity.
|
|
|
|
Args:
|
|
event: Event instance
|
|
|
|
Returns:
|
|
URL string (can be empty)
|
|
"""
|
|
# This would ideally be configured based on your frontend routes
|
|
entity_type_map = {
|
|
'Project': f'/projects/{event.entity_id}',
|
|
'Report': f'/reports/{event.entity_id}',
|
|
'Invoice': f'/invoices/{event.entity_id}',
|
|
}
|
|
return entity_type_map.get(event.entity_type, '')
|
|
|
|
@staticmethod
|
|
def _create_notification(
|
|
event: Event,
|
|
rule: NotificationRule,
|
|
recipient: Any,
|
|
subject: str,
|
|
body: str,
|
|
action_url: str
|
|
) -> Notification:
|
|
"""
|
|
Create a Notification instance.
|
|
|
|
Args:
|
|
event: Event that triggered the notification
|
|
rule: Rule that matched
|
|
recipient: Profile receiving the notification
|
|
subject: Notification subject
|
|
body: Notification body
|
|
action_url: Action URL
|
|
|
|
Returns:
|
|
Created Notification instance
|
|
"""
|
|
content_type = ContentType.objects.get_for_model(recipient)
|
|
|
|
notification = Notification.objects.create(
|
|
event=event,
|
|
rule=rule,
|
|
recipient_content_type=content_type,
|
|
recipient_object_id=recipient.id,
|
|
status=NotificationStatusChoices.PENDING,
|
|
subject=subject,
|
|
body=body,
|
|
action_url=action_url
|
|
)
|
|
|
|
return notification
|
|
|
|
@staticmethod
|
|
def _create_delivery(notification: Notification, channel: str) -> NotificationDelivery:
|
|
"""
|
|
Create a NotificationDelivery instance and queue delivery task.
|
|
|
|
Args:
|
|
notification: Notification to deliver
|
|
channel: Delivery channel
|
|
|
|
Returns:
|
|
Created NotificationDelivery instance
|
|
"""
|
|
delivery = NotificationDelivery.objects.create(
|
|
notification=notification,
|
|
channel=channel,
|
|
status=DeliveryStatusChoices.PENDING
|
|
)
|
|
|
|
# Queue the appropriate delivery task
|
|
from core.tasks.notifications import deliver_notification
|
|
deliver_notification.delay(str(delivery.id))
|
|
|
|
return delivery
|