""" Notification scheduling and batching service for efficient delivery """ from django.contrib.auth import get_user_model from django.utils import timezone from django.db.models import Q from django.core.mail import send_mail from django.conf import settings from .models import Notification from .notification_models import ( NotificationBatch, NotificationDelivery, NotificationTemplate, NotificationPreference ) from .notification_routing_service import NotificationRoutingService from datetime import datetime, timedelta from typing import List, Dict, Any, Optional import logging import json from celery import shared_task User = get_user_model() logger = logging.getLogger(__name__) class NotificationSchedulingService: """Service for scheduling and batching notifications""" def __init__(self): self.routing_service = NotificationRoutingService() self.batch_size = 100 # Process notifications in batches of 100 self.max_retries = 3 def schedule_notification(self, notification_type: str, context: Dict[str, Any], scheduled_at: datetime, priority: str = 'medium', channels: List[str] = None) -> str: """ Schedule a notification for future delivery Args: notification_type: Type of notification context: Context data for the notification scheduled_at: When to send the notification priority: Notification priority channels: Delivery channels Returns: Task ID for the scheduled notification """ try: # Schedule the task using Celery (or store for later processing) task_data = { 'notification_type': notification_type, 'context': context, 'priority': priority, 'channels': channels or ['in_app'] } # In a real implementation, this would use Celery's apply_async # For now, we'll store it and process later return self._store_scheduled_notification(task_data, scheduled_at) except Exception as e: logger.error(f"Error scheduling notification: {str(e)}") return None def _store_scheduled_notification(self, task_data: Dict[str, Any], scheduled_at: datetime) -> str: """Store scheduled notification for later processing""" # This would typically be stored in a task queue or database # For now, we'll create a simple record import uuid task_id = str(uuid.uuid4()) # Store in a simple format (in production, use proper task queue) scheduled_notification = { 'task_id': task_id, 'scheduled_at': scheduled_at.isoformat(), 'task_data': task_data, 'status': 'scheduled' } # In production, store this in Redis or database logger.info(f"Scheduled notification {task_id} for {scheduled_at}") return task_id def create_notification_batch(self, name: str, notification_type: str, template_id: str, target_criteria: Dict[str, Any], scheduled_at: Optional[datetime] = None, created_by: User = None) -> NotificationBatch: """ Create a batch notification job Args: name: Batch name notification_type: Type of notifications to send template_id: Template to use target_criteria: Criteria for selecting recipients scheduled_at: When to process the batch created_by: User who created the batch Returns: Created NotificationBatch instance """ try: template = NotificationTemplate.objects.get(id=template_id) # Calculate recipient count recipients = self._get_batch_recipients(target_criteria) batch = NotificationBatch.objects.create( name=name, notification_type=notification_type, template=template, target_criteria=target_criteria, total_recipients=len(recipients), scheduled_at=scheduled_at, created_by=created_by ) # Schedule processing if needed if scheduled_at: self._schedule_batch_processing(batch) return batch except Exception as e: logger.error(f"Error creating notification batch: {str(e)}") raise def _get_batch_recipients(self, criteria: Dict[str, Any]) -> List[User]: """Get recipients based on batch criteria""" users = User.objects.filter(is_active=True) if 'roles' in criteria and criteria['roles']: users = users.filter(role__in=criteria['roles']) if 'branches' in criteria and criteria['branches']: users = users.filter(branch_id__in=criteria['branches']) if 'portfolio_managers' in criteria and criteria['portfolio_managers']: users = users.filter(portfolio_manager_id__in=criteria['portfolio_managers']) if 'user_ids' in criteria and criteria['user_ids']: users = users.filter(id__in=criteria['user_ids']) # Additional filters if 'exclude_roles' in criteria and criteria['exclude_roles']: users = users.exclude(role__in=criteria['exclude_roles']) if 'min_last_login_days' in criteria: min_date = timezone.now() - timedelta(days=criteria['min_last_login_days']) users = users.filter(last_login__gte=min_date) return list(users) def _schedule_batch_processing(self, batch: NotificationBatch): """Schedule batch processing""" if batch.scheduled_at: # In production, use Celery's apply_async with eta logger.info(f"Scheduled batch {batch.id} for processing at {batch.scheduled_at}") else: # Process immediately self.process_notification_batch(batch.id) def process_notification_batch(self, batch_id: str) -> Dict[str, Any]: """ Process a notification batch Args: batch_id: ID of the batch to process Returns: Processing results """ try: batch = NotificationBatch.objects.get(id=batch_id) if batch.status != 'pending': return {'error': f'Batch is not in pending status: {batch.status}'} batch.start_processing() # Get recipients recipients = self._get_batch_recipients(batch.target_criteria) success_count = 0 failed_count = 0 errors = [] # Process in smaller batches for i in range(0, len(recipients), self.batch_size): batch_recipients = recipients[i:i + self.batch_size] for recipient in batch_recipients: try: # Create context for this recipient context = { 'user': recipient, 'batch': batch, 'template': batch.template, } # Add any additional context from batch if hasattr(batch, 'context_data'): context.update(batch.context_data) # Route notification notifications = self.routing_service.route_notification( batch.notification_type, context, priority='medium' ) if notifications: success_count += 1 else: failed_count += 1 errors.append(f"No notifications created for user {recipient.id}") except Exception as e: failed_count += 1 error_msg = f"Failed for user {recipient.id}: {str(e)}" errors.append(error_msg) logger.error(error_msg) # Small delay between batches to avoid overwhelming the system import time time.sleep(0.1) # Update batch status batch.sent_count = success_count batch.failed_count = failed_count batch.error_log = '\n'.join(errors[:100]) # Limit error log size batch.complete_processing() return { 'success': True, 'batch_id': str(batch.id), 'total_recipients': len(recipients), 'success_count': success_count, 'failed_count': failed_count, 'errors': errors[:10] # Return first 10 errors } except NotificationBatch.DoesNotExist: return {'error': 'Batch not found'} except Exception as e: logger.error(f"Error processing batch {batch_id}: {str(e)}") try: batch = NotificationBatch.objects.get(id=batch_id) batch.fail_processing(str(e)) except: pass return {'error': str(e)} def process_digest_notifications(self, frequency: str = 'daily') -> Dict[str, Any]: """ Process digest notifications for users who prefer batched delivery Args: frequency: Digest frequency ('hourly', 'daily', 'weekly') Returns: Processing results """ try: # Get users who prefer digest notifications digest_preferences = NotificationPreference.objects.filter( frequency=frequency, is_enabled=True ).select_related('user') results = { 'frequency': frequency, 'processed_users': 0, 'total_notifications': 0, 'errors': [] } # Determine time range based on frequency if frequency == 'hourly': start_time = timezone.now() - timedelta(hours=1) elif frequency == 'daily': start_time = timezone.now() - timedelta(days=1) elif frequency == 'weekly': start_time = timezone.now() - timedelta(weeks=1) else: start_time = timezone.now() - timedelta(days=1) # Group preferences by user user_preferences = {} for pref in digest_preferences: if pref.user_id not in user_preferences: user_preferences[pref.user_id] = [] user_preferences[pref.user_id].append(pref) # Process each user's digest for user_id, preferences in user_preferences.items(): try: user = User.objects.get(id=user_id) # Get unread notifications for this user in the time range notification_types = [pref.notification_type for pref in preferences] unread_notifications = Notification.objects.filter( user=user, notification_type__in=notification_types, read_at__isnull=True, created_at__gte=start_time ).order_by('-created_at') if unread_notifications.exists(): # Create digest notification digest_content = self._create_digest_content( user, unread_notifications, frequency ) # Send digest self._send_digest_notification(user, digest_content, frequency) results['processed_users'] += 1 results['total_notifications'] += unread_notifications.count() # Mark original notifications as processed (optional) # unread_notifications.update(read_at=timezone.now()) except Exception as e: error_msg = f"Error processing digest for user {user_id}: {str(e)}" results['errors'].append(error_msg) logger.error(error_msg) return results except Exception as e: logger.error(f"Error processing digest notifications: {str(e)}") return {'error': str(e)} def _create_digest_content(self, user: User, notifications: List[Notification], frequency: str) -> Dict[str, str]: """Create digest notification content""" notification_count = notifications.count() # Group notifications by type by_type = {} for notification in notifications: if notification.notification_type not in by_type: by_type[notification.notification_type] = [] by_type[notification.notification_type].append(notification) # Create summary summary_lines = [] for notification_type, type_notifications in by_type.items(): count = len(type_notifications) type_name = notification_type.replace('_', ' ').title() summary_lines.append(f"• {count} {type_name} notification{'s' if count > 1 else ''}") title = f"{frequency.title()} Notification Digest" message = f"""Hello {user.get_full_name()}, You have {notification_count} unread notification{'s' if notification_count > 1 else ''} from the past {frequency}: {chr(10).join(summary_lines)} Please log in to view your notifications. Best regards, Haven Grazuri Investment Limited Team""" return { 'title': title, 'message': message, 'summary': summary_lines } def _send_digest_notification(self, user: User, content: Dict[str, str], frequency: str): """Send digest notification to user""" # Create in-app notification Notification.objects.create( user=user, notification_type=f'{frequency}_digest', title=content['title'], message=content['message'], priority='low', icon='fa-list' ) # Send email if user prefers email digests email_pref = NotificationPreference.objects.filter( user=user, notification_type=f'{frequency}_digest', channel='email', is_enabled=True ).first() if email_pref and user.email: try: send_mail( subject=content['title'], message=content['message'], from_email=settings.DEFAULT_FROM_EMAIL, recipient_list=[user.email], fail_silently=False ) except Exception as e: logger.error(f"Error sending digest email to {user.email}: {str(e)}") def cleanup_old_notifications(self, days: int = 90) -> Dict[str, int]: """ Clean up old notifications and delivery records Args: days: Number of days to keep notifications Returns: Cleanup statistics """ try: cutoff_date = timezone.now() - timedelta(days=days) # Count records to be deleted old_notifications = Notification.objects.filter(created_at__lt=cutoff_date) old_deliveries = NotificationDelivery.objects.filter(created_at__lt=cutoff_date) old_batches = NotificationBatch.objects.filter( created_at__lt=cutoff_date, status__in=['completed', 'failed', 'cancelled'] ) notification_count = old_notifications.count() delivery_count = old_deliveries.count() batch_count = old_batches.count() # Delete records old_notifications.delete() old_deliveries.delete() old_batches.delete() logger.info(f"Cleaned up {notification_count} notifications, " f"{delivery_count} deliveries, {batch_count} batches") return { 'notifications_deleted': notification_count, 'deliveries_deleted': delivery_count, 'batches_deleted': batch_count, 'cutoff_date': cutoff_date.isoformat() } except Exception as e: logger.error(f"Error cleaning up old notifications: {str(e)}") return {'error': str(e)} def get_processing_status(self) -> Dict[str, Any]: """Get current processing status and statistics""" try: # Batch statistics pending_batches = NotificationBatch.objects.filter(status='pending').count() processing_batches = NotificationBatch.objects.filter(status='processing').count() # Delivery statistics pending_deliveries = NotificationDelivery.objects.filter(status='pending').count() failed_deliveries = NotificationDelivery.objects.filter(status='failed').count() # Recent activity recent_batches = NotificationBatch.objects.order_by('-created_at')[:5].values( 'id', 'name', 'status', 'total_recipients', 'sent_count', 'failed_count' ) return { 'pending_batches': pending_batches, 'processing_batches': processing_batches, 'pending_deliveries': pending_deliveries, 'failed_deliveries': failed_deliveries, 'recent_batches': list(recent_batches), 'status': 'healthy' if processing_batches == 0 else 'processing' } except Exception as e: logger.error(f"Error getting processing status: {str(e)}") return {'error': str(e)} # Celery tasks (if using Celery) @shared_task def process_notification_batch_task(batch_id: str): """Celery task for processing notification batches""" service = NotificationSchedulingService() return service.process_notification_batch(batch_id) @shared_task def process_digest_notifications_task(frequency: str = 'daily'): """Celery task for processing digest notifications""" service = NotificationSchedulingService() return service.process_digest_notifications(frequency) @shared_task def cleanup_old_notifications_task(days: int = 90): """Celery task for cleaning up old notifications""" service = NotificationSchedulingService() return service.cleanup_old_notifications(days)