# -*- coding: utf-8 -*- """ Notifications Celery Tasks This module contains Celery tasks for the notifications app, handling asynchronous notification processing, email delivery, bulk operations, and maintenance tasks. Tasks: - send_notification_task: Send individual notifications - send_bulk_notifications_task: Send bulk notifications - send_email_notification_task: Send email notifications - process_notification_queue_task: Process queued notifications - cleanup_expired_notifications_task: Clean up expired notifications - generate_notification_reports_task: Generate analytics reports - retry_failed_notifications_task: Retry failed notifications - send_digest_notifications_task: Send digest notifications Author: AdTlas Development Team Version: 1.0.0 Last Updated: 2024 """ import logging from datetime import datetime, timedelta from typing import Dict, List, Optional, Any, Union from celery import shared_task, group, chord from celery.exceptions import Retry from django.conf import settings from django.contrib.auth import get_user_model from django.core.mail import send_mail, EmailMultiAlternatives from django.template.loader import render_to_string from django.utils import timezone from django.db import transaction from django.core.cache import cache from django.db.models import Q, Count from .models import ( NotificationType, Notification, NotificationPreference, NotificationTemplate, NotificationQueue, NotificationHistory, ) from .services import ( NotificationService, EmailNotificationService, QueueService, AnalyticsService, ) User = get_user_model() logger = logging.getLogger(__name__) @shared_task(bind=True, max_retries=3, default_retry_delay=60) def send_notification_task( self, recipient_id: int, notification_type_slug: str, title: str, message: str, action_url: Optional[str] = None, expires_at: Optional[str] = None, metadata: Optional[Dict] = None, template_context: Optional[Dict] = None ) -> Dict[str, Any]: """ Send individual notification asynchronously. Args: recipient_id: ID of the recipient user notification_type_slug: Slug of notification type title: Notification title message: Notification message action_url: Optional action URL expires_at: Optional expiration datetime (ISO format) metadata: Optional metadata dictionary template_context: Optional template context Returns: dict: Task result with success status and details """ try: # Get recipient user try: recipient = User.objects.get(id=recipient_id, is_active=True) except User.DoesNotExist: logger.error(f"Recipient user not found: {recipient_id}") return { 'success': False, 'error': 'Recipient not found', 'recipient_id': recipient_id } # Get notification type try: notification_type = NotificationType.objects.get( slug=notification_type_slug, is_active=True ) except NotificationType.DoesNotExist: logger.error(f"Notification type not found: {notification_type_slug}") return { 'success': False, 'error': 'Notification type not found', 'notification_type': notification_type_slug } # Parse expires_at if provided expires_at_dt = None if expires_at: try: expires_at_dt = datetime.fromisoformat(expires_at.replace('Z', '+00:00')) except ValueError: logger.warning(f"Invalid expires_at format: {expires_at}") # Create and send notification notification_service = NotificationService() notification = notification_service.create_notification( recipient=recipient, notification_type=notification_type, title=title, message=message, action_url=action_url, expires_at=expires_at_dt, metadata=metadata or {}, send_immediately=True, template_context=template_context or {} ) if notification: logger.info(f"Notification sent successfully: {notification.id}") return { 'success': True, 'notification_id': notification.id, 'recipient_id': recipient_id, 'task_id': self.request.id } else: logger.warning(f"Notification not created (possibly blocked by preferences): {recipient_id}") return { 'success': False, 'error': 'Notification blocked by user preferences', 'recipient_id': recipient_id } except Exception as exc: logger.error(f"Error in send_notification_task: {exc}") # Retry logic if self.request.retries < self.max_retries: logger.info(f"Retrying send_notification_task (attempt {self.request.retries + 1})") raise self.retry(exc=exc, countdown=60 * (self.request.retries + 1)) return { 'success': False, 'error': str(exc), 'recipient_id': recipient_id, 'max_retries_exceeded': True } @shared_task(bind=True, max_retries=2) def send_bulk_notifications_task( self, recipient_ids: List[int], notification_type_slug: str, title: str, message: str, action_url: Optional[str] = None, expires_at: Optional[str] = None, metadata: Optional[Dict] = None, template_context: Optional[Dict] = None, batch_size: int = 50 ) -> Dict[str, Any]: """ Send bulk notifications asynchronously. Args: recipient_ids: List of recipient user IDs notification_type_slug: Slug of notification type title: Notification title message: Notification message action_url: Optional action URL expires_at: Optional expiration datetime (ISO format) metadata: Optional metadata dictionary template_context: Optional template context batch_size: Number of notifications per batch Returns: dict: Task result with success status and statistics """ try: logger.info(f"Starting bulk notification task for {len(recipient_ids)} recipients") # Validate notification type try: notification_type = NotificationType.objects.get( slug=notification_type_slug, is_active=True ) except NotificationType.DoesNotExist: logger.error(f"Notification type not found: {notification_type_slug}") return { 'success': False, 'error': 'Notification type not found', 'notification_type': notification_type_slug } # Get active recipients active_recipients = User.objects.filter( id__in=recipient_ids, is_active=True ).values_list('id', flat=True) if not active_recipients: logger.warning("No active recipients found for bulk notification") return { 'success': True, 'total_recipients': 0, 'sent': 0, 'failed': 0, 'skipped': len(recipient_ids) } # Create batches and send notifications results = { 'success': True, 'total_recipients': len(active_recipients), 'sent': 0, 'failed': 0, 'skipped': 0, 'batch_results': [] } # Process in batches for i in range(0, len(active_recipients), batch_size): batch_recipients = list(active_recipients[i:i + batch_size]) # Create group of individual notification tasks job = group( send_notification_task.s( recipient_id=recipient_id, notification_type_slug=notification_type_slug, title=title, message=message, action_url=action_url, expires_at=expires_at, metadata=metadata, template_context=template_context ) for recipient_id in batch_recipients ) # Execute batch batch_result = job.apply_async() batch_results = batch_result.get() # Aggregate results batch_stats = { 'batch_number': (i // batch_size) + 1, 'batch_size': len(batch_recipients), 'sent': 0, 'failed': 0, 'skipped': 0 } for result in batch_results: if result.get('success'): if result.get('notification_id'): batch_stats['sent'] += 1 results['sent'] += 1 else: batch_stats['skipped'] += 1 results['skipped'] += 1 else: batch_stats['failed'] += 1 results['failed'] += 1 results['batch_results'].append(batch_stats) logger.info(f"Batch {batch_stats['batch_number']} completed: {batch_stats}") logger.info(f"Bulk notification task completed: {results}") return results except Exception as exc: logger.error(f"Error in send_bulk_notifications_task: {exc}") if self.request.retries < self.max_retries: logger.info(f"Retrying send_bulk_notifications_task (attempt {self.request.retries + 1})") raise self.retry(exc=exc, countdown=300) # 5 minute delay for bulk retries return { 'success': False, 'error': str(exc), 'total_recipients': len(recipient_ids), 'sent': 0, 'failed': len(recipient_ids), 'skipped': 0, 'max_retries_exceeded': True } @shared_task(bind=True, max_retries=3, default_retry_delay=120) def send_email_notification_task( self, notification_id: int, force_send: bool = False ) -> Dict[str, Any]: """ Send email notification asynchronously. Args: notification_id: ID of the notification to send force_send: Whether to force send even if preferences say no Returns: dict: Task result with success status """ try: # Get notification try: notification = Notification.objects.select_related( 'recipient', 'notification_type' ).get(id=notification_id) except Notification.DoesNotExist: logger.error(f"Notification not found: {notification_id}") return { 'success': False, 'error': 'Notification not found', 'notification_id': notification_id } # Check if recipient is active if not notification.recipient.is_active: logger.warning(f"Recipient is inactive: {notification.recipient.email}") return { 'success': False, 'error': 'Recipient is inactive', 'notification_id': notification_id } # Check user preferences unless forced if not force_send: from .services import PreferenceService preference_service = PreferenceService() user_preferences = preference_service.get_user_preferences( notification.recipient, notification.notification_type ) if not user_preferences.email_enabled: logger.info(f"Email disabled for user: {notification.recipient.email}") return { 'success': False, 'error': 'Email notifications disabled by user', 'notification_id': notification_id } # Send email email_service = EmailNotificationService() success = email_service.send_email_notification(notification) if success: logger.info(f"Email notification sent successfully: {notification_id}") return { 'success': True, 'notification_id': notification_id, 'recipient_email': notification.recipient.email, 'task_id': self.request.id } else: logger.error(f"Failed to send email notification: {notification_id}") return { 'success': False, 'error': 'Email sending failed', 'notification_id': notification_id } except Exception as exc: logger.error(f"Error in send_email_notification_task: {exc}") # Retry logic if self.request.retries < self.max_retries: logger.info(f"Retrying send_email_notification_task (attempt {self.request.retries + 1})") raise self.retry(exc=exc, countdown=120 * (self.request.retries + 1)) return { 'success': False, 'error': str(exc), 'notification_id': notification_id, 'max_retries_exceeded': True } @shared_task(bind=True) def process_notification_queue_task(self) -> Dict[str, Any]: """ Process queued notifications. Returns: dict: Processing results """ try: logger.info("Starting notification queue processing") queue_service = QueueService() results = queue_service.process_queue() logger.info(f"Queue processing completed: {results}") return { 'success': True, 'task_id': self.request.id, **results } except Exception as exc: logger.error(f"Error in process_notification_queue_task: {exc}") return { 'success': False, 'error': str(exc), 'task_id': self.request.id } @shared_task(bind=True) def cleanup_expired_notifications_task( self, days_to_keep: int = 30, batch_size: int = 1000 ) -> Dict[str, Any]: """ Clean up expired and old notifications. Args: days_to_keep: Number of days to keep notifications batch_size: Number of notifications to process per batch Returns: dict: Cleanup results """ try: logger.info(f"Starting notification cleanup (keeping {days_to_keep} days)") cutoff_date = timezone.now() - timedelta(days=days_to_keep) results = { 'success': True, 'expired_archived': 0, 'old_deleted': 0, 'history_cleaned': 0, 'task_id': self.request.id } # Archive expired notifications expired_notifications = Notification.objects.filter( expires_at__lt=timezone.now(), is_archived=False ) expired_count = expired_notifications.count() if expired_count > 0: # Process in batches for i in range(0, expired_count, batch_size): batch = expired_notifications[i:i + batch_size] batch.update( is_archived=True, archived_at=timezone.now() ) results['expired_archived'] += len(batch) # Delete very old notifications (beyond retention period) old_notifications = Notification.objects.filter( created_at__lt=cutoff_date, is_archived=True ) old_count = old_notifications.count() if old_count > 0: # Delete in batches for i in range(0, old_count, batch_size): batch_ids = list(old_notifications[i:i + batch_size].values_list('id', flat=True)) deleted_count = Notification.objects.filter(id__in=batch_ids).delete()[0] results['old_deleted'] += deleted_count # Clean up old notification history old_history = NotificationHistory.objects.filter( attempted_at__lt=cutoff_date ) history_count = old_history.count() if history_count > 0: # Delete in batches for i in range(0, history_count, batch_size): batch_ids = list(old_history[i:i + batch_size].values_list('id', flat=True)) deleted_count = NotificationHistory.objects.filter(id__in=batch_ids).delete()[0] results['history_cleaned'] += deleted_count # Clean up completed queue items old_queue_items = NotificationQueue.objects.filter( created_at__lt=cutoff_date, status__in=['completed', 'failed'] ) queue_deleted = old_queue_items.delete()[0] results['queue_cleaned'] = queue_deleted logger.info(f"Notification cleanup completed: {results}") return results except Exception as exc: logger.error(f"Error in cleanup_expired_notifications_task: {exc}") return { 'success': False, 'error': str(exc), 'task_id': self.request.id } @shared_task(bind=True) def generate_notification_reports_task( self, report_type: str = 'daily', date_str: Optional[str] = None ) -> Dict[str, Any]: """ Generate notification analytics reports. Args: report_type: Type of report ('daily', 'weekly', 'monthly') date_str: Date for the report (ISO format, defaults to today) Returns: dict: Report generation results """ try: logger.info(f"Generating {report_type} notification report") # Parse date if date_str: report_date = datetime.fromisoformat(date_str.replace('Z', '+00:00')).date() else: report_date = timezone.now().date() # Calculate date range based on report type if report_type == 'daily': start_date = report_date end_date = report_date elif report_type == 'weekly': start_date = report_date - timedelta(days=report_date.weekday()) end_date = start_date + timedelta(days=6) elif report_type == 'monthly': start_date = report_date.replace(day=1) if start_date.month == 12: end_date = start_date.replace(year=start_date.year + 1, month=1) - timedelta(days=1) else: end_date = start_date.replace(month=start_date.month + 1) - timedelta(days=1) else: raise ValueError(f"Invalid report type: {report_type}") # Generate analytics analytics_service = AnalyticsService() # Get notifications in date range notifications = Notification.objects.filter( created_at__date__gte=start_date, created_at__date__lte=end_date ) # Get delivery history in date range delivery_history = NotificationHistory.objects.filter( attempted_at__date__gte=start_date, attempted_at__date__lte=end_date ) # Generate report data report_data = { 'report_type': report_type, 'start_date': start_date.isoformat(), 'end_date': end_date.isoformat(), 'generated_at': timezone.now().isoformat(), 'notification_stats': { 'total_created': notifications.count(), 'total_read': notifications.filter(is_read=True).count(), 'total_unread': notifications.filter(is_read=False).count(), 'total_archived': notifications.filter(is_archived=True).count(), 'by_type': dict( notifications.values('notification_type__name') .annotate(count=Count('id')) .values_list('notification_type__name', 'count') ), 'by_day': dict( notifications.extra(select={'day': 'date(created_at)'}) .values('day') .annotate(count=Count('id')) .values_list('day', 'count') ), }, 'delivery_stats': { 'total_attempts': delivery_history.count(), 'successful_deliveries': delivery_history.filter(status='delivered').count(), 'failed_deliveries': delivery_history.filter(status='failed').count(), 'by_method': dict( delivery_history.values('delivery_method') .annotate(count=Count('id')) .values_list('delivery_method', 'count') ), 'by_status': dict( delivery_history.values('status') .annotate(count=Count('id')) .values_list('status', 'count') ), }, } # Calculate delivery rate total_attempts = report_data['delivery_stats']['total_attempts'] if total_attempts > 0: success_rate = (report_data['delivery_stats']['successful_deliveries'] / total_attempts) * 100 report_data['delivery_stats']['success_rate'] = round(success_rate, 2) else: report_data['delivery_stats']['success_rate'] = 0 # Cache report data cache_key = f"notification_report_{report_type}_{start_date}_{end_date}" cache.set(cache_key, report_data, 86400) # Cache for 24 hours logger.info(f"Report generated successfully: {report_type} for {start_date} to {end_date}") return { 'success': True, 'report_type': report_type, 'start_date': start_date.isoformat(), 'end_date': end_date.isoformat(), 'cache_key': cache_key, 'task_id': self.request.id, 'report_data': report_data } except Exception as exc: logger.error(f"Error in generate_notification_reports_task: {exc}") return { 'success': False, 'error': str(exc), 'report_type': report_type, 'task_id': self.request.id } @shared_task(bind=True) def retry_failed_notifications_task( self, max_age_hours: int = 24, max_retry_count: int = 3 ) -> Dict[str, Any]: """ Retry failed notifications. Args: max_age_hours: Maximum age of notifications to retry (in hours) max_retry_count: Maximum number of retry attempts Returns: dict: Retry results """ try: logger.info(f"Starting retry of failed notifications (max age: {max_age_hours}h)") cutoff_time = timezone.now() - timedelta(hours=max_age_hours) # Get failed queue items to retry failed_queue_items = NotificationQueue.objects.filter( status='failed', retry_count__lt=max_retry_count, created_at__gte=cutoff_time ) results = { 'success': True, 'total_failed': failed_queue_items.count(), 'retried': 0, 'skipped': 0, 'task_id': self.request.id } for queue_item in failed_queue_items: try: # Reset status and increment retry count queue_item.status = 'pending' queue_item.retry_count += 1 queue_item.scheduled_at = timezone.now() + timedelta(minutes=5) queue_item.error_message = None queue_item.save() results['retried'] += 1 logger.info(f"Queued retry for notification queue item: {queue_item.id}") except Exception as e: logger.error(f"Error retrying queue item {queue_item.id}: {e}") results['skipped'] += 1 # Also retry failed notification history items failed_history = NotificationHistory.objects.filter( status='failed', attempted_at__gte=cutoff_time ).select_related('notification') for history_item in failed_history: try: if history_item.notification and history_item.delivery_method == 'email': # Retry email notification send_email_notification_task.delay( notification_id=history_item.notification.id, force_send=False ) results['retried'] += 1 except Exception as e: logger.error(f"Error retrying notification {history_item.notification.id}: {e}") results['skipped'] += 1 logger.info(f"Failed notification retry completed: {results}") return results except Exception as exc: logger.error(f"Error in retry_failed_notifications_task: {exc}") return { 'success': False, 'error': str(exc), 'task_id': self.request.id } @shared_task(bind=True) def send_digest_notifications_task( self, digest_type: str = 'daily', user_ids: Optional[List[int]] = None ) -> Dict[str, Any]: """ Send digest notifications to users. Args: digest_type: Type of digest ('daily', 'weekly') user_ids: Optional list of specific user IDs to send to Returns: dict: Digest sending results """ try: logger.info(f"Starting {digest_type} digest notification task") # Determine date range for digest now = timezone.now() if digest_type == 'daily': start_date = now - timedelta(days=1) elif digest_type == 'weekly': start_date = now - timedelta(days=7) else: raise ValueError(f"Invalid digest type: {digest_type}") # Get users to send digest to if user_ids: users = User.objects.filter(id__in=user_ids, is_active=True) else: # Get users who have unread notifications and digest enabled users = User.objects.filter( is_active=True, notifications__is_read=False, notifications__created_at__gte=start_date ).distinct() results = { 'success': True, 'digest_type': digest_type, 'total_users': users.count(), 'sent': 0, 'failed': 0, 'skipped': 0, 'task_id': self.request.id } # Get or create digest notification type digest_notification_type, created = NotificationType.objects.get_or_create( slug=f'{digest_type}_digest', defaults={ 'name': f'{digest_type.title()} Digest', 'description': f'{digest_type.title()} notification digest', 'default_email_enabled': True, 'default_in_app_enabled': False, 'can_be_disabled': True, } ) for user in users: try: # Get user's unread notifications for the period unread_notifications = Notification.objects.filter( recipient=user, is_read=False, created_at__gte=start_date, is_archived=False ).order_by('-created_at') if not unread_notifications.exists(): results['skipped'] += 1 continue # Check if user wants digest notifications from .services import PreferenceService preference_service = PreferenceService() if not preference_service.should_receive_notification(user, digest_notification_type): results['skipped'] += 1 continue # Create digest content notification_count = unread_notifications.count() title = f"You have {notification_count} unread notifications" # Group notifications by type notifications_by_type = {} for notification in unread_notifications[:10]: # Limit to 10 for digest type_name = notification.notification_type.name if type_name not in notifications_by_type: notifications_by_type[type_name] = [] notifications_by_type[type_name].append(notification) # Build digest message message_parts = [f"Here's your {digest_type} notification digest:"] for type_name, notifications in notifications_by_type.items(): message_parts.append(f"\n{type_name}: {len(notifications)} notifications") for notification in notifications[:3]: # Show first 3 message_parts.append(f" • {notification.title}") if len(notifications) > 3: message_parts.append(f" • ... and {len(notifications) - 3} more") if notification_count > 10: message_parts.append(f"\n... and {notification_count - 10} more notifications") message = "\n".join(message_parts) # Send digest notification digest_task = send_notification_task.delay( recipient_id=user.id, notification_type_slug=digest_notification_type.slug, title=title, message=message, action_url="/notifications/", metadata={ 'digest_type': digest_type, 'notification_count': notification_count, 'period_start': start_date.isoformat(), 'period_end': now.isoformat(), } ) results['sent'] += 1 logger.info(f"Digest sent to user {user.id}: {notification_count} notifications") except Exception as e: logger.error(f"Error sending digest to user {user.id}: {e}") results['failed'] += 1 logger.info(f"Digest notification task completed: {results}") return results except Exception as exc: logger.error(f"Error in send_digest_notifications_task: {exc}") return { 'success': False, 'error': str(exc), 'digest_type': digest_type, 'task_id': self.request.id } # Periodic task helpers @shared_task def daily_notification_maintenance(): """ Daily maintenance task for notifications. This task runs daily to: - Process notification queue - Clean up expired notifications - Generate daily reports - Retry failed notifications """ logger.info("Starting daily notification maintenance") # Chain tasks together maintenance_tasks = [ process_notification_queue_task.s(), cleanup_expired_notifications_task.s(), generate_notification_reports_task.s('daily'), retry_failed_notifications_task.s(), ] # Execute tasks in sequence from celery import chain job = chain(*maintenance_tasks) result = job.apply_async() logger.info(f"Daily maintenance tasks queued: {result.id}") return {'maintenance_job_id': result.id} @shared_task def weekly_notification_digest(): """ Weekly digest notification task. Sends weekly digest notifications to users. """ logger.info("Starting weekly notification digest") result = send_digest_notifications_task.delay('weekly') logger.info(f"Weekly digest task queued: {result.id}") return {'digest_job_id': result.id}