# -*- coding: utf-8 -*- """ Notifications Services This module contains service classes for the notifications app, handling the core business logic for notification processing, delivery, template rendering, and user preferences. Services: - NotificationService: Core notification logic - EmailNotificationService: Email delivery - TemplateService: Template rendering - PreferenceService: User preferences - QueueService: Notification queuing - AnalyticsService: Notification analytics Author: AdTlas Development Team Version: 1.0.0 Last Updated: 2024 """ import json import logging from datetime import datetime, timedelta from typing import Dict, List, Optional, Union, Any from django.conf import settings from django.contrib.auth import get_user_model from django.core.mail import send_mail, EmailMultiAlternatives from django.template import Template, Context from django.template.loader import render_to_string from django.utils import timezone from django.core.cache import cache from django.db import transaction from django.db.models import Q, Count, Avg from django.utils.html import strip_tags from django.core.exceptions import ValidationError from apps.core.utils import get_client_ip from .models import ( NotificationType, Notification, NotificationPreference, NotificationTemplate, NotificationQueue, NotificationHistory, ) User = get_user_model() logger = logging.getLogger(__name__) class NotificationService: """ Core notification service for creating and managing notifications. This service handles the main notification logic including creation, validation, delivery coordination, and status management. """ def __init__(self): """ Initialize the notification service. """ self.email_service = EmailNotificationService() self.template_service = TemplateService() self.preference_service = PreferenceService() self.queue_service = QueueService() def create_notification( self, recipient: User, notification_type: Union[str, NotificationType], title: str, message: str, action_url: Optional[str] = None, expires_at: Optional[datetime] = None, metadata: Optional[Dict] = None, send_immediately: bool = True, template_context: Optional[Dict] = None ) -> Optional[Notification]: """ Create a new notification. Args: recipient: User to receive the notification notification_type: Type of notification (slug or instance) title: Notification title message: Notification message action_url: Optional action URL expires_at: Optional expiration date metadata: Optional metadata dictionary send_immediately: Whether to send immediately template_context: Context for template rendering Returns: Notification: Created notification or None if failed """ try: # Get notification type if isinstance(notification_type, str): notification_type = NotificationType.objects.get(slug=notification_type) # Check if user should receive this notification if not self.preference_service.should_receive_notification(recipient, notification_type): logger.info(f"Notification blocked by user preferences: {recipient.email} - {notification_type.slug}") return None # Render content using template if available if template_context: rendered_content = self.template_service.render_notification( notification_type, template_context ) if rendered_content: title = rendered_content.get('title', title) message = rendered_content.get('message', message) # Create notification notification = Notification.objects.create( recipient=recipient, notification_type=notification_type, title=title, message=message, action_url=action_url, expires_at=expires_at, metadata=metadata or {} ) # Send notification if requested if send_immediately: self.send_notification(notification) logger.info(f"Notification created: {notification.id} for {recipient.email}") return notification except NotificationType.DoesNotExist: logger.error(f"Notification type not found: {notification_type}") return None except Exception as e: logger.error(f"Error creating notification: {e}") return None def send_notification(self, notification: Notification) -> bool: """ Send a notification through appropriate channels. Args: notification: Notification to send Returns: bool: True if sent successfully """ try: user_preferences = self.preference_service.get_user_preferences( notification.recipient, notification.notification_type ) success = True # Send in-app notification (always sent) self._record_delivery_attempt( notification, 'in_app', 'delivered', 'In-app notification created' ) # Send email notification if enabled if user_preferences.email_enabled: if self._is_in_quiet_hours(notification.recipient, user_preferences): # Queue for later delivery self.queue_service.queue_notification( notification, scheduled_at=self._get_next_delivery_time(user_preferences) ) logger.info(f"Notification queued due to quiet hours: {notification.id}") else: email_success = self.email_service.send_email_notification(notification) if not email_success: success = False # Update notification status if success: self._update_notification_cache(notification) return success except Exception as e: logger.error(f"Error sending notification {notification.id}: {e}") self._record_delivery_attempt( notification, 'system', 'failed', str(e) ) return False def send_bulk_notifications( self, recipients: List[User], notification_type: Union[str, NotificationType], title: str, message: str, **kwargs ) -> Dict[str, Any]: """ Send notifications to multiple recipients. Args: recipients: List of users to receive notifications notification_type: Type of notification title: Notification title message: Notification message **kwargs: Additional notification parameters Returns: dict: Results of bulk operation """ try: results = { 'total': len(recipients), 'created': 0, 'failed': 0, 'skipped': 0, 'errors': [] } # Get notification type if isinstance(notification_type, str): notification_type = NotificationType.objects.get(slug=notification_type) # Create notifications in batches batch_size = getattr(settings, 'NOTIFICATION_BATCH_SIZE', 100) for i in range(0, len(recipients), batch_size): batch = recipients[i:i + batch_size] with transaction.atomic(): for recipient in batch: try: notification = self.create_notification( recipient=recipient, notification_type=notification_type, title=title, message=message, send_immediately=False, # Send in batch later **kwargs ) if notification: results['created'] += 1 else: results['skipped'] += 1 except Exception as e: results['failed'] += 1 results['errors'].append({ 'recipient': recipient.email, 'error': str(e) }) logger.error(f"Error creating notification for {recipient.email}: {e}") # Queue bulk sending if results['created'] > 0: self.queue_service.queue_bulk_notifications( notification_type=notification_type, title=title, message=message, recipients=[r.id for r in recipients if r], **kwargs ) logger.info(f"Bulk notification completed: {results}") return results except Exception as e: logger.error(f"Error in bulk notification: {e}") return { 'total': len(recipients), 'created': 0, 'failed': len(recipients), 'skipped': 0, 'errors': [{'error': str(e)}] } def mark_as_read(self, notification: Notification, user: User = None) -> bool: """ Mark notification as read. Args: notification: Notification to mark as read user: User marking as read (for validation) Returns: bool: True if marked successfully """ try: if user and notification.recipient != user: logger.warning(f"User {user.email} tried to mark notification {notification.id} as read") return False if not notification.is_read: notification.is_read = True notification.read_at = timezone.now() notification.save(update_fields=['is_read', 'read_at']) # Update cache self._update_notification_cache(notification) # Send signal from .signals import notification_read notification_read.send( sender=self.__class__, notification=notification ) logger.info(f"Notification marked as read: {notification.id}") return True except Exception as e: logger.error(f"Error marking notification as read: {e}") return False def mark_as_archived(self, notification: Notification, user: User = None) -> bool: """ Mark notification as archived. Args: notification: Notification to archive user: User archiving (for validation) Returns: bool: True if archived successfully """ try: if user and notification.recipient != user: logger.warning(f"User {user.email} tried to archive notification {notification.id}") return False if not notification.is_archived: notification.is_archived = True notification.archived_at = timezone.now() notification.save(update_fields=['is_archived', 'archived_at']) # Update cache self._update_notification_cache(notification) # Send signal from .signals import notification_archived notification_archived.send( sender=self.__class__, notification=notification ) logger.info(f"Notification archived: {notification.id}") return True except Exception as e: logger.error(f"Error archiving notification: {e}") return False def get_user_notifications( self, user: User, unread_only: bool = False, limit: int = 50, notification_type: Optional[str] = None ) -> List[Notification]: """ Get notifications for a user. Args: user: User to get notifications for unread_only: Whether to get only unread notifications limit: Maximum number of notifications notification_type: Filter by notification type slug Returns: list: List of notifications """ try: queryset = Notification.objects.filter( recipient=user, is_archived=False ).select_related('notification_type').order_by('-created_at') if unread_only: queryset = queryset.filter(is_read=False) if notification_type: queryset = queryset.filter(notification_type__slug=notification_type) return list(queryset[:limit]) except Exception as e: logger.error(f"Error getting user notifications: {e}") return [] def get_unread_count(self, user: User) -> int: """ Get count of unread notifications for user. Args: user: User to count notifications for Returns: int: Number of unread notifications """ try: cache_key = f"unread_notifications_{user.id}" count = cache.get(cache_key) if count is None: count = Notification.objects.filter( recipient=user, is_read=False, is_archived=False ).count() cache.set(cache_key, count, 300) # Cache for 5 minutes return count except Exception as e: logger.error(f"Error getting unread count: {e}") return 0 def cleanup_expired_notifications(self) -> int: """ Clean up expired notifications. Returns: int: Number of notifications cleaned up """ try: expired_notifications = Notification.objects.filter( expires_at__lt=timezone.now(), is_archived=False ) count = expired_notifications.count() expired_notifications.update( is_archived=True, archived_at=timezone.now() ) logger.info(f"Cleaned up {count} expired notifications") return count except Exception as e: logger.error(f"Error cleaning up expired notifications: {e}") return 0 def _record_delivery_attempt( self, notification: Notification, delivery_method: str, status: str, error_message: str = None ): """ Record a delivery attempt. Args: notification: Notification being delivered delivery_method: Method of delivery status: Delivery status error_message: Error message if failed """ try: NotificationHistory.objects.create( notification=notification, delivery_method=delivery_method, status=status, attempted_at=timezone.now(), delivered_at=timezone.now() if status == 'delivered' else None, error_message=error_message, metadata={ 'timestamp': timezone.now().isoformat(), } ) except Exception as e: logger.error(f"Error recording delivery attempt: {e}") def _update_notification_cache(self, notification: Notification): """ Update notification-related cache entries. Args: notification: Notification to update cache for """ try: # Clear unread count cache cache_key = f"unread_notifications_{notification.recipient.id}" cache.delete(cache_key) # Update other relevant caches # Add more cache invalidation as needed except Exception as e: logger.error(f"Error updating notification cache: {e}") def _is_in_quiet_hours(self, user: User, preferences: NotificationPreference) -> bool: """ Check if current time is in user's quiet hours. Args: user: User to check preferences: User's notification preferences Returns: bool: True if in quiet hours """ try: if not preferences.quiet_hours_start or not preferences.quiet_hours_end: return False # Get user's current time user_tz = preferences.get_timezone() current_time = timezone.now().astimezone(user_tz).time() start_time = preferences.quiet_hours_start end_time = preferences.quiet_hours_end # Handle overnight quiet hours if start_time <= end_time: return start_time <= current_time <= end_time else: return current_time >= start_time or current_time <= end_time except Exception as e: logger.error(f"Error checking quiet hours: {e}") return False def _get_next_delivery_time(self, preferences: NotificationPreference) -> datetime: """ Get next delivery time after quiet hours. Args: preferences: User's notification preferences Returns: datetime: Next delivery time """ try: if not preferences.quiet_hours_end: return timezone.now() + timedelta(hours=1) user_tz = preferences.get_timezone() current_time = timezone.now().astimezone(user_tz) # Calculate next delivery time (after quiet hours end) next_delivery = current_time.replace( hour=preferences.quiet_hours_end.hour, minute=preferences.quiet_hours_end.minute, second=0, microsecond=0 ) # If quiet hours end is tomorrow if next_delivery <= current_time: next_delivery += timedelta(days=1) return next_delivery.astimezone(timezone.utc) except Exception as e: logger.error(f"Error calculating next delivery time: {e}") return timezone.now() + timedelta(hours=1) class EmailNotificationService: """ Service for sending email notifications. This service handles email delivery including template rendering, HTML/text content generation, and delivery tracking. """ def __init__(self): """ Initialize the email notification service. """ self.template_service = TemplateService() def send_email_notification(self, notification: Notification) -> bool: """ Send email notification. Args: notification: Notification to send Returns: bool: True if sent successfully """ try: # Get email template template = self.template_service.get_email_template( notification.notification_type ) if not template: logger.warning(f"No email template found for {notification.notification_type.slug}") return self._send_simple_email(notification) # Render email content context = self._build_email_context(notification) rendered_content = self.template_service.render_email_template( template, context ) if not rendered_content: return self._send_simple_email(notification) # Send email success = self._send_rendered_email( notification, rendered_content ) # Record delivery attempt self._record_email_delivery( notification, 'delivered' if success else 'failed', None if success else 'Email sending failed' ) return success except Exception as e: logger.error(f"Error sending email notification: {e}") self._record_email_delivery( notification, 'failed', str(e) ) return False def _send_simple_email(self, notification: Notification) -> bool: """ Send simple email without template. Args: notification: Notification to send Returns: bool: True if sent successfully """ try: subject = f"[{settings.SITE_NAME}] {notification.title}" message = notification.message if notification.action_url: message += f"\n\nAction: {notification.action_url}" send_mail( subject=subject, message=message, from_email=settings.DEFAULT_FROM_EMAIL, recipient_list=[notification.recipient.email], fail_silently=False ) return True except Exception as e: logger.error(f"Error sending simple email: {e}") return False def _send_rendered_email( self, notification: Notification, rendered_content: Dict[str, str] ) -> bool: """ Send rendered email with HTML and text content. Args: notification: Notification to send rendered_content: Rendered email content Returns: bool: True if sent successfully """ try: subject = rendered_content.get('subject', notification.title) html_content = rendered_content.get('html_content') text_content = rendered_content.get('text_content') # Create email message email = EmailMultiAlternatives( subject=subject, body=text_content or strip_tags(html_content), from_email=settings.DEFAULT_FROM_EMAIL, to=[notification.recipient.email] ) # Add HTML content if available if html_content: email.attach_alternative(html_content, "text/html") # Send email email.send(fail_silently=False) return True except Exception as e: logger.error(f"Error sending rendered email: {e}") return False def _build_email_context(self, notification: Notification) -> Dict[str, Any]: """ Build context for email template rendering. Args: notification: Notification instance Returns: dict: Template context """ return { 'notification': notification, 'user': notification.recipient, 'site_name': getattr(settings, 'SITE_NAME', 'AdTlas'), 'site_url': getattr(settings, 'SITE_URL', 'http://localhost:8000'), 'title': notification.title, 'message': notification.message, 'action_url': notification.action_url, 'unsubscribe_url': f"/notifications/preferences/", 'current_year': timezone.now().year, 'metadata': notification.metadata, } def _record_email_delivery( self, notification: Notification, status: str, error_message: str = None ): """ Record email delivery attempt. Args: notification: Notification being delivered status: Delivery status error_message: Error message if failed """ try: from .models import NotificationHistory NotificationHistory.objects.create( notification=notification, delivery_method='email', status=status, attempted_at=timezone.now(), delivered_at=timezone.now() if status == 'delivered' else None, error_message=error_message, metadata={ 'recipient_email': notification.recipient.email, 'timestamp': timezone.now().isoformat(), } ) except Exception as e: logger.error(f"Error recording email delivery: {e}") class TemplateService: """ Service for managing notification templates. This service handles template rendering, variable substitution, and template management for notifications. """ def get_email_template(self, notification_type: NotificationType) -> Optional[NotificationTemplate]: """ Get email template for notification type. Args: notification_type: Notification type Returns: NotificationTemplate: Email template or None """ try: return NotificationTemplate.objects.filter( name__icontains=notification_type.slug, template_type='email', is_active=True ).first() except Exception as e: logger.error(f"Error getting email template: {e}") return None def get_in_app_template(self, notification_type: NotificationType) -> Optional[NotificationTemplate]: """ Get in-app template for notification type. Args: notification_type: Notification type Returns: NotificationTemplate: In-app template or None """ try: return NotificationTemplate.objects.filter( name__icontains=notification_type.slug, template_type='in_app', is_active=True ).first() except Exception as e: logger.error(f"Error getting in-app template: {e}") return None def render_email_template( self, template: NotificationTemplate, context: Dict[str, Any] ) -> Optional[Dict[str, str]]: """ Render email template with context. Args: template: Template to render context: Template context Returns: dict: Rendered content with subject, html_content, text_content """ try: rendered = {} # Render subject if template.subject: subject_template = Template(template.subject) rendered['subject'] = subject_template.render(Context(context)) # Render HTML content if template.html_content: html_template = Template(template.html_content) rendered['html_content'] = html_template.render(Context(context)) # Render text content if template.text_content: text_template = Template(template.text_content) rendered['text_content'] = text_template.render(Context(context)) elif template.html_content: # Generate text from HTML if no text content rendered['text_content'] = strip_tags(rendered['html_content']) return rendered except Exception as e: logger.error(f"Error rendering email template: {e}") return None def render_notification( self, notification_type: NotificationType, context: Dict[str, Any] ) -> Optional[Dict[str, str]]: """ Render notification content using templates. Args: notification_type: Notification type context: Template context Returns: dict: Rendered content with title and message """ try: template = self.get_in_app_template(notification_type) if not template: return None rendered = {} # Render title from HTML or text content if template.html_content: # Extract title from HTML (assuming first h1, h2, etc.) import re title_match = re.search(r']*>([^<]+)', template.html_content) if title_match: title_template = Template(title_match.group(1)) rendered['title'] = title_template.render(Context(context)) # Render message if template.text_content: message_template = Template(template.text_content) rendered['message'] = message_template.render(Context(context)) elif template.html_content: message_template = Template(strip_tags(template.html_content)) rendered['message'] = message_template.render(Context(context)) return rendered except Exception as e: logger.error(f"Error rendering notification template: {e}") return None class PreferenceService: """ Service for managing user notification preferences. This service handles user preference retrieval, updates, and validation for notification delivery. """ def get_user_preferences( self, user: User, notification_type: NotificationType ) -> NotificationPreference: """ Get user preferences for notification type. Args: user: User instance notification_type: Notification type Returns: NotificationPreference: User preferences """ try: preference, created = NotificationPreference.objects.get_or_create( user=user, notification_type=notification_type, defaults={ 'is_enabled': True, 'email_enabled': notification_type.default_email_enabled, 'in_app_enabled': notification_type.default_in_app_enabled, } ) return preference except Exception as e: logger.error(f"Error getting user preferences: {e}") # Return default preferences return NotificationPreference( user=user, notification_type=notification_type, is_enabled=True, email_enabled=notification_type.default_email_enabled, in_app_enabled=notification_type.default_in_app_enabled, ) def should_receive_notification( self, user: User, notification_type: NotificationType ) -> bool: """ Check if user should receive notification. Args: user: User instance notification_type: Notification type Returns: bool: True if user should receive notification """ try: if not user.is_active: return False if not notification_type.is_active: return False # Check if notification type can be disabled if notification_type.can_be_disabled: preferences = self.get_user_preferences(user, notification_type) return preferences.is_enabled return True except Exception as e: logger.error(f"Error checking notification permission: {e}") return False def update_user_preferences( self, user: User, preferences_data: Dict[str, Any] ) -> bool: """ Update user notification preferences. Args: user: User instance preferences_data: Preferences data Returns: bool: True if updated successfully """ try: with transaction.atomic(): for type_id, settings in preferences_data.items(): try: notification_type = NotificationType.objects.get(id=type_id) preference = self.get_user_preferences(user, notification_type) # Update preference fields for field, value in settings.items(): if hasattr(preference, field): setattr(preference, field, value) preference.save() except NotificationType.DoesNotExist: logger.warning(f"Notification type {type_id} not found") continue return True except Exception as e: logger.error(f"Error updating user preferences: {e}") return False class QueueService: """ Service for managing notification queues. This service handles queuing notifications for delayed delivery, batch processing, and retry logic. """ def queue_notification( self, notification: Notification, scheduled_at: Optional[datetime] = None ) -> Optional[NotificationQueue]: """ Queue notification for later delivery. Args: notification: Notification to queue scheduled_at: When to deliver notification Returns: NotificationQueue: Queued notification or None """ try: queue_item = NotificationQueue.objects.create( notification_type=notification.notification_type, title=notification.title, message=notification.message, recipients=[notification.recipient.id], scheduled_at=scheduled_at or timezone.now(), metadata={ 'original_notification_id': notification.id, 'action_url': notification.action_url, 'expires_at': notification.expires_at.isoformat() if notification.expires_at else None, } ) logger.info(f"Notification queued: {queue_item.id}") return queue_item except Exception as e: logger.error(f"Error queuing notification: {e}") return None def queue_bulk_notifications( self, notification_type: NotificationType, title: str, message: str, recipients: List[int], scheduled_at: Optional[datetime] = None, **kwargs ) -> Optional[NotificationQueue]: """ Queue bulk notifications for delivery. Args: notification_type: Type of notification title: Notification title message: Notification message recipients: List of recipient IDs scheduled_at: When to deliver notifications **kwargs: Additional notification data Returns: NotificationQueue: Queued notifications or None """ try: queue_item = NotificationQueue.objects.create( notification_type=notification_type, title=title, message=message, recipients=recipients, scheduled_at=scheduled_at or timezone.now(), metadata=kwargs ) logger.info(f"Bulk notifications queued: {queue_item.id} for {len(recipients)} recipients") return queue_item except Exception as e: logger.error(f"Error queuing bulk notifications: {e}") return None def process_queue(self) -> Dict[str, int]: """ Process queued notifications. Returns: dict: Processing results """ try: results = { 'processed': 0, 'failed': 0, 'skipped': 0 } # Get pending queue items pending_items = NotificationQueue.objects.filter( status='pending', scheduled_at__lte=timezone.now() ).order_by('scheduled_at') for item in pending_items: try: success = self._process_queue_item(item) if success: results['processed'] += 1 else: results['failed'] += 1 except Exception as e: logger.error(f"Error processing queue item {item.id}: {e}") results['failed'] += 1 return results except Exception as e: logger.error(f"Error processing notification queue: {e}") return {'processed': 0, 'failed': 0, 'skipped': 0} def _process_queue_item(self, item: NotificationQueue) -> bool: """ Process individual queue item. Args: item: Queue item to process Returns: bool: True if processed successfully """ try: # Mark as processing item.status = 'processing' item.save(update_fields=['status']) # Get recipients recipients = User.objects.filter( id__in=item.recipients, is_active=True ) if not recipients.exists(): item.status = 'completed' item.save(update_fields=['status']) return True # Create and send notifications notification_service = NotificationService() for recipient in recipients: notification_service.create_notification( recipient=recipient, notification_type=item.notification_type, title=item.title, message=item.message, action_url=item.metadata.get('action_url'), expires_at=item.metadata.get('expires_at'), metadata=item.metadata, send_immediately=True ) # Mark as completed item.status = 'completed' item.save(update_fields=['status']) return True except Exception as e: # Handle retry logic item.retry_count += 1 if item.retry_count >= item.max_retries: item.status = 'failed' item.error_message = str(e) else: item.status = 'pending' item.scheduled_at = timezone.now() + timedelta(minutes=5 * item.retry_count) item.save(update_fields=['status', 'retry_count', 'error_message', 'scheduled_at']) logger.error(f"Error processing queue item {item.id}: {e}") return False class AnalyticsService: """ Service for notification analytics and reporting. This service provides analytics data for notifications including delivery rates, user engagement, and trends. """ def get_notification_stats(self, user: User = None) -> Dict[str, Any]: """ Get notification statistics. Args: user: User to get stats for (None for global stats) Returns: dict: Notification statistics """ try: base_queryset = Notification.objects.all() if user: base_queryset = base_queryset.filter(recipient=user) now = timezone.now() today = now.date() week_ago = now - timedelta(days=7) month_ago = now - timedelta(days=30) stats = { 'total_notifications': base_queryset.count(), 'unread_notifications': base_queryset.filter(is_read=False).count(), 'read_notifications': base_queryset.filter(is_read=True).count(), 'archived_notifications': base_queryset.filter(is_archived=True).count(), 'expired_notifications': base_queryset.filter( expires_at__lt=now ).count(), 'notifications_today': base_queryset.filter( created_at__date=today ).count(), 'notifications_this_week': base_queryset.filter( created_at__gte=week_ago ).count(), 'notifications_this_month': base_queryset.filter( created_at__gte=month_ago ).count(), 'by_type': dict( base_queryset.values('notification_type__name') .annotate(count=Count('id')) .values_list('notification_type__name', 'count') ), 'by_status': { 'read': base_queryset.filter(is_read=True).count(), 'unread': base_queryset.filter(is_read=False).count(), 'archived': base_queryset.filter(is_archived=True).count(), }, 'recent_activity': list( base_queryset.order_by('-created_at')[:10] .values('id', 'title', 'created_at', 'is_read') ), } return stats except Exception as e: logger.error(f"Error getting notification stats: {e}") return {} def get_delivery_stats(self) -> Dict[str, Any]: """ Get notification delivery statistics. Returns: dict: Delivery statistics """ try: history_queryset = NotificationHistory.objects.all() stats = { 'total_attempts': history_queryset.count(), 'successful_deliveries': history_queryset.filter(status='delivered').count(), 'failed_deliveries': history_queryset.filter(status='failed').count(), 'by_method': dict( history_queryset.values('delivery_method') .annotate(count=Count('id')) .values_list('delivery_method', 'count') ), 'by_status': dict( history_queryset.values('status') .annotate(count=Count('id')) .values_list('status', 'count') ), } # Calculate delivery rate total = stats['total_attempts'] if total > 0: stats['delivery_rate'] = (stats['successful_deliveries'] / total) * 100 else: stats['delivery_rate'] = 0 return stats except Exception as e: logger.error(f"Error getting delivery stats: {e}") return {}