""" Intelligent notification routing service with role-based filtering and escalation """ from django.contrib.auth import get_user_model from django.utils import timezone from django.db.models import Q from django.core.exceptions import ObjectDoesNotExist from .models import Notification from .notification_models import ( NotificationPreference, NotificationRule, NotificationEscalation, NotificationTemplate, NotificationDelivery, NotificationBatch ) import logging from typing import List, Dict, Any, Optional from datetime import datetime, timedelta User = get_user_model() logger = logging.getLogger(__name__) class NotificationRoutingService: """Service for intelligent notification routing based on roles and rules""" def __init__(self): self.default_channels = ['in_app'] self.role_hierarchy = { 'admin': ['admin'], 'team_leader': ['team_leader', 'admin'], 'loan_officer': ['loan_officer', 'team_leader', 'admin'], 'secretary': ['secretary', 'team_leader', 'admin'], 'auditor': ['auditor', 'admin'], 'borrower': ['borrower'] } def route_notification(self, notification_type: str, context: Dict[str, Any], priority: str = 'medium', channels: List[str] = None) -> List[Notification]: """ Route notification based on type, context, and rules Args: notification_type: Type of notification context: Context data for routing decisions priority: Notification priority channels: Preferred delivery channels Returns: List of created notifications """ try: # Get applicable routing rules rules = self._get_applicable_rules(notification_type, context) # Determine recipients based on rules recipients = self._determine_recipients(rules, context) # Get notification template template = self._get_template(notification_type, 'in_app') # Create notifications notifications = [] for recipient in recipients: # Check user preferences if self._should_notify_user(recipient, notification_type, channels or self.default_channels): notification = self._create_notification( recipient, notification_type, template, context, priority ) notifications.append(notification) # Schedule delivery across channels self._schedule_delivery(notification, recipient, channels or self.default_channels) # Handle escalation rules self._setup_escalations(notifications, rules) logger.info(f"Routed {len(notifications)} notifications for type {notification_type}") return notifications except Exception as e: logger.error(f"Error routing notification {notification_type}: {str(e)}") return [] def _get_applicable_rules(self, notification_type: str, context: Dict[str, Any]) -> List[NotificationRule]: """Get rules that apply to this notification type and context""" rules = NotificationRule.objects.filter( notification_type=notification_type, is_active=True ).order_by('priority') applicable_rules = [] for rule in rules: if rule.evaluate_condition(context): applicable_rules.append(rule) return applicable_rules def _determine_recipients(self, rules: List[NotificationRule], context: Dict[str, Any]) -> List[User]: """Determine who should receive the notification based on rules""" recipients = set() if not rules: # Default routing based on context recipients.update(self._get_default_recipients(context)) else: # Rule-based routing for rule in rules: rule_recipients = self._get_recipients_for_rule(rule, context) recipients.update(rule_recipients) return list(recipients) def _get_default_recipients(self, context: Dict[str, Any]) -> List[User]: """Get default recipients when no specific rules apply""" recipients = [] # If there's a specific user in context if 'user' in context and context['user']: recipients.append(context['user']) # If there's a loan/application, notify portfolio manager if 'loan' in context and hasattr(context['loan'], 'portfolio_manager'): if context['loan'].portfolio_manager: recipients.append(context['loan'].portfolio_manager) # If there's a branch context, notify branch managers if 'branch' in context and context['branch']: branch_managers = User.objects.filter( branch=context['branch'], role__in=['team_leader', 'admin'] ) recipients.extend(branch_managers) return recipients def _get_recipients_for_rule(self, rule: NotificationRule, context: Dict[str, Any]) -> List[User]: """Get recipients based on a specific rule""" recipients = [] # Get users matching target roles if rule.target_roles: users = User.objects.filter(role__in=rule.target_roles, is_active=True) # Apply additional filtering based on context if 'branch' in context and context['branch']: users = users.filter(Q(branch=context['branch']) | Q(is_superuser=True)) recipients.extend(users) return recipients def _should_notify_user(self, user: User, notification_type: str, channels: List[str]) -> bool: """Check if user should receive notification based on preferences""" try: # Check if user has preferences for this notification type preferences = NotificationPreference.objects.filter( user=user, notification_type=notification_type, channel__in=channels, is_enabled=True ) if not preferences.exists(): # No specific preferences, use defaults return True # Check quiet hours now = timezone.now().time() for pref in preferences: if pref.quiet_hours_start and pref.quiet_hours_end: if pref.quiet_hours_start <= now <= pref.quiet_hours_end: continue # Skip this channel during quiet hours if pref.frequency != 'never': return True return False except Exception as e: logger.error(f"Error checking user preferences for {user.id}: {str(e)}") return True # Default to sending notification def _get_template(self, notification_type: str, channel: str) -> Optional[NotificationTemplate]: """Get template for notification type and channel""" try: return NotificationTemplate.objects.filter( notification_type=notification_type, channel=channel, is_active=True ).first() except Exception: return None def _create_notification(self, user: User, notification_type: str, template: Optional[NotificationTemplate], context: Dict[str, Any], priority: str) -> Notification: """Create notification instance""" if template: rendered = template.render(context) title = rendered['title'] message = rendered['message'] else: # Fallback to basic notification title = context.get('title', f'New {notification_type.replace("_", " ").title()}') message = context.get('message', 'You have a new notification') notification = Notification.objects.create( user=user, notification_type=notification_type, title=title, message=message, priority=priority, action_url=context.get('action_url'), icon=context.get('icon') or Notification.get_default_icon(notification_type), loan_app=context.get('loan_app'), related_loan=context.get('loan'), alert_data=context.get('alert_data', {}), action_required=context.get('action_required', False) ) return notification def _schedule_delivery(self, notification: Notification, user: User, channels: List[str]): """Schedule notification delivery across channels""" for channel in channels: # Check user preferences for this channel preference = NotificationPreference.objects.filter( user=user, notification_type=notification.notification_type, channel=channel ).first() if preference and not preference.is_enabled: continue # Create delivery record delivery = NotificationDelivery.objects.create( notification=notification, channel=channel, recipient=user, status='pending' ) # Schedule immediate delivery for in-app notifications if channel == 'in_app': delivery.mark_sent() delivery.mark_delivered() def _setup_escalations(self, notifications: List[Notification], rules: List[NotificationRule]): """Setup escalation rules for notifications""" for notification in notifications: for rule in rules: if rule.escalation_delay > 0 and rule.escalation_roles: # Schedule escalation (this would typically be handled by a background task) self._schedule_escalation(notification, rule) def _schedule_escalation(self, notification: Notification, rule: NotificationRule): """Schedule escalation for a notification""" # This would typically be handled by a task queue like Celery # For now, we'll create the escalation record escalation_time = timezone.now() + timedelta(minutes=rule.escalation_delay) # Get escalation recipients escalation_users = User.objects.filter( role__in=rule.escalation_roles, is_active=True ) for user in escalation_users: NotificationEscalation.objects.create( original_notification=notification, rule=rule, escalated_to=user, escalation_level=1 ) def process_escalations(self): """Process pending escalations (called by background task)""" pending_escalations = NotificationEscalation.objects.filter( acknowledged_at__isnull=True, resolved_at__isnull=True, escalated_at__lte=timezone.now() ) for escalation in pending_escalations: # Check if original notification was acknowledged if escalation.original_notification.read_at: escalation.resolve("Original notification was read") continue # Create escalation notification self.route_notification( 'escalation_alert', { 'user': escalation.escalated_to, 'original_notification': escalation.original_notification, 'escalation_level': escalation.escalation_level, 'title': f'Escalation: {escalation.original_notification.title}', 'message': f'This notification requires attention: {escalation.original_notification.message}', 'action_url': escalation.original_notification.action_url, 'action_required': True }, priority='high' ) escalation.acknowledge("Escalation notification sent") def get_user_notification_stats(self, user: User, days: int = 30) -> Dict[str, Any]: """Get notification statistics for a user""" start_date = timezone.now() - timedelta(days=days) notifications = Notification.objects.filter( user=user, created_at__gte=start_date ) total_count = notifications.count() read_count = notifications.filter(read_at__isnull=False).count() unread_count = total_count - read_count # Get delivery stats deliveries = NotificationDelivery.objects.filter( recipient=user, created_at__gte=start_date ) delivery_stats = {} for channel in ['in_app', 'email', 'sms', 'push']: channel_deliveries = deliveries.filter(channel=channel) delivery_stats[channel] = { 'total': channel_deliveries.count(), 'delivered': channel_deliveries.filter(status='delivered').count(), 'read': channel_deliveries.filter(status='read').count(), 'failed': channel_deliveries.filter(status='failed').count(), } return { 'total_notifications': total_count, 'read_notifications': read_count, 'unread_notifications': unread_count, 'read_rate': (read_count / total_count * 100) if total_count > 0 else 0, 'delivery_stats': delivery_stats, 'most_common_types': list( notifications.values('notification_type') .annotate(count=models.Count('id')) .order_by('-count')[:5] ) } class NotificationPreferenceService: """Service for managing user notification preferences""" def __init__(self): self.default_preferences = { 'loan_approved': {'in_app': True, 'email': True, 'sms': False}, 'loan_rejected': {'in_app': True, 'email': True, 'sms': False}, 'payment_due': {'in_app': True, 'email': True, 'sms': True}, 'payment_received': {'in_app': True, 'email': False, 'sms': False}, 'loan_overdue': {'in_app': True, 'email': True, 'sms': True}, 'system_maintenance': {'in_app': True, 'email': False, 'sms': False}, 'security_alert': {'in_app': True, 'email': True, 'sms': True}, } def get_user_preferences(self, user: User) -> Dict[str, Dict[str, Any]]: """Get user's notification preferences""" preferences = NotificationPreference.objects.filter(user=user) result = {} for pref in preferences: if pref.notification_type not in result: result[pref.notification_type] = {} result[pref.notification_type][pref.channel] = { 'enabled': pref.is_enabled, 'frequency': pref.frequency, 'quiet_hours_start': pref.quiet_hours_start, 'quiet_hours_end': pref.quiet_hours_end, } return result def update_user_preferences(self, user: User, preferences: Dict[str, Dict[str, Any]]): """Update user's notification preferences""" for notification_type, channels in preferences.items(): for channel, settings in channels.items(): pref, created = NotificationPreference.objects.get_or_create( user=user, notification_type=notification_type, channel=channel, defaults={ 'is_enabled': settings.get('enabled', True), 'frequency': settings.get('frequency', 'immediate'), 'quiet_hours_start': settings.get('quiet_hours_start'), 'quiet_hours_end': settings.get('quiet_hours_end'), } ) if not created: pref.is_enabled = settings.get('enabled', pref.is_enabled) pref.frequency = settings.get('frequency', pref.frequency) pref.quiet_hours_start = settings.get('quiet_hours_start', pref.quiet_hours_start) pref.quiet_hours_end = settings.get('quiet_hours_end', pref.quiet_hours_end) pref.save() def setup_default_preferences(self, user: User): """Setup default preferences for a new user""" for notification_type, channels in self.default_preferences.items(): for channel, enabled in channels.items(): NotificationPreference.objects.get_or_create( user=user, notification_type=notification_type, channel=channel, defaults={ 'is_enabled': enabled, 'frequency': 'immediate', } )