# -*- coding: utf-8 -*- """ Adtlas Activities Signals This module contains Django signal handlers for the activities app, providing automatic activity logging, model change tracking, and event-driven functionality. Features: - Automatic activity logging for model changes - User authentication activity tracking - System event monitoring - Custom signal handlers - Activity cleanup and maintenance - Real-time notifications Author: Adtlas Development Team Version: 1.0.0 Last Updated: 2025-01-27 """ import json import logging from datetime import datetime, timedelta from django.db.models.signals import ( post_save, post_delete, pre_save, pre_delete, m2m_changed ) from django.contrib.auth.signals import ( user_logged_in, user_logged_out, user_login_failed ) from django.core.signals import request_started, request_finished from django.dispatch import receiver, Signal from django.contrib.auth import get_user_model from django.contrib.contenttypes.models import ContentType from django.utils import timezone from django.conf import settings from django.core.cache import cache from django.db import transaction from .models import Activity, ActivityCategory, ActivitySummary from .utils import get_client_ip, get_user_agent, should_log_activity # Get the user model User = get_user_model() # Set up logging logger = logging.getLogger(__name__) # Custom signals activity_logged = Signal() bulk_activity_logged = Signal() activity_cleanup_completed = Signal() # Authentication signals @receiver(user_logged_in) def log_user_login(sender, request, user, **kwargs): """ Log user login activity. This signal handler automatically creates an activity record when a user successfully logs in to the system. Args: sender: The sender of the signal request: Django request object user: User instance that logged in **kwargs: Additional keyword arguments """ try: # Get or create authentication category auth_category, _ = ActivityCategory.objects.get_or_create( code='auth', defaults={ 'name': 'Authentication', 'description': 'User authentication activities', 'color': '#007bff', 'icon': 'fas fa-sign-in-alt' } ) # Create login activity Activity.log_activity( user=user, action='login', description=f'User {user.email} logged in successfully', category=auth_category, request=request, is_successful=True, metadata={ 'login_method': getattr(request, 'login_method', 'standard'), 'remember_me': getattr(request, 'remember_me', False), 'login_timestamp': timezone.now().isoformat() } ) # Update user's last login activity cache cache.set(f'user_last_login_{user.id}', timezone.now(), 3600) logger.info(f"Login activity logged for user {user.email}") except Exception as e: logger.error(f"Failed to log login activity for user {user.email}: {str(e)}") @receiver(user_logged_out) def log_user_logout(sender, request, user, **kwargs): """ Log user logout activity. This signal handler automatically creates an activity record when a user logs out of the system. Args: sender: The sender of the signal request: Django request object user: User instance that logged out (may be None) **kwargs: Additional keyword arguments """ try: # Skip if user is None (anonymous logout) if not user: return # Get authentication category auth_category = ActivityCategory.objects.filter( code='auth' ).first() if not auth_category: return # Calculate session duration last_login = cache.get(f'user_last_login_{user.id}') session_duration = None if last_login: session_duration = (timezone.now() - last_login).total_seconds() * 1000 # Create logout activity Activity.log_activity( user=user, action='logout', description=f'User {user.email} logged out', category=auth_category, request=request, is_successful=True, duration_ms=session_duration, metadata={ 'logout_method': getattr(request, 'logout_method', 'standard'), 'session_duration_ms': session_duration, 'logout_timestamp': timezone.now().isoformat() } ) # Clear login cache cache.delete(f'user_last_login_{user.id}') logger.info(f"Logout activity logged for user {user.email}") except Exception as e: logger.error(f"Failed to log logout activity: {str(e)}") @receiver(user_login_failed) def log_failed_login(sender, credentials, request, **kwargs): """ Log failed login attempts. This signal handler automatically creates an activity record when a login attempt fails. Args: sender: The sender of the signal credentials: Login credentials used request: Django request object **kwargs: Additional keyword arguments """ try: # Get authentication category auth_category, _ = ActivityCategory.objects.get_or_create( code='auth', defaults={ 'name': 'Authentication', 'description': 'User authentication activities', 'color': '#007bff', 'icon': 'fas fa-sign-in-alt' } ) # Get username from credentials username = credentials.get('username', 'Unknown') # Create failed login activity Activity.log_activity( user=None, # No user for failed login action='login_failed', description=f'Failed login attempt for username: {username}', category=auth_category, request=request, is_successful=False, error_message='Invalid credentials provided', metadata={ 'attempted_username': username, 'failure_reason': 'invalid_credentials', 'attempt_timestamp': timezone.now().isoformat() } ) # Track failed attempts for security monitoring ip_address = get_client_ip(request) cache_key = f'failed_login_attempts_{ip_address}' attempts = cache.get(cache_key, 0) + 1 cache.set(cache_key, attempts, 3600) # 1 hour logger.warning(f"Failed login attempt for username {username} from IP {ip_address}") except Exception as e: logger.error(f"Failed to log failed login activity: {str(e)}") # Model change signals @receiver(post_save) def log_model_creation(sender, instance, created, **kwargs): """ Log model creation activities. This signal handler automatically creates activity records when new model instances are created. Args: sender: Model class that sent the signal instance: Model instance that was saved created: Boolean indicating if instance was created **kwargs: Additional keyword arguments """ # Skip if not created or if it's an Activity model (avoid recursion) if not created or sender == Activity: return # Check if we should log this model if not should_log_activity(sender): return try: # Get current user from thread local storage or request user = getattr(instance, '_current_user', None) # Skip if no user context if not user: return # Get or create appropriate category app_label = sender._meta.app_label category = get_or_create_category_for_app(app_label) # Create activity Activity.log_activity( user=user, action='create', description=f'Created {sender._meta.verbose_name}: {str(instance)}', category=category, content_type=ContentType.objects.get_for_model(sender), object_id=instance.pk, is_successful=True, metadata={ 'model': sender._meta.label, 'object_repr': str(instance), 'created_fields': get_model_fields(instance) } ) logger.debug(f"Creation activity logged for {sender._meta.label}: {instance.pk}") except Exception as e: logger.error(f"Failed to log creation activity for {sender._meta.label}: {str(e)}") @receiver(post_save) def log_model_update(sender, instance, created, **kwargs): """ Log model update activities. This signal handler automatically creates activity records when model instances are updated. Args: sender: Model class that sent the signal instance: Model instance that was saved created: Boolean indicating if instance was created **kwargs: Additional keyword arguments """ # Skip if created or if it's an Activity model (avoid recursion) if created or sender == Activity: return # Check if we should log this model if not should_log_activity(sender): return try: # Get current user from thread local storage or request user = getattr(instance, '_current_user', None) # Skip if no user context if not user: return # Get changed fields changed_fields = getattr(instance, '_changed_fields', {}) # Skip if no fields changed if not changed_fields: return # Get or create appropriate category app_label = sender._meta.app_label category = get_or_create_category_for_app(app_label) # Create activity Activity.log_activity( user=user, action='update', description=f'Updated {sender._meta.verbose_name}: {str(instance)}', category=category, content_type=ContentType.objects.get_for_model(sender), object_id=instance.pk, is_successful=True, metadata={ 'model': sender._meta.label, 'object_repr': str(instance), 'changed_fields': changed_fields, 'field_count': len(changed_fields) } ) logger.debug(f"Update activity logged for {sender._meta.label}: {instance.pk}") except Exception as e: logger.error(f"Failed to log update activity for {sender._meta.label}: {str(e)}") @receiver(post_delete) def log_model_deletion(sender, instance, **kwargs): """ Log model deletion activities. This signal handler automatically creates activity records when model instances are deleted. Args: sender: Model class that sent the signal instance: Model instance that was deleted **kwargs: Additional keyword arguments """ # Skip if it's an Activity model (avoid recursion) if sender == Activity: return # Check if we should log this model if not should_log_activity(sender): return try: # Get current user from thread local storage or request user = getattr(instance, '_current_user', None) # Skip if no user context if not user: return # Get or create appropriate category app_label = sender._meta.app_label category = get_or_create_category_for_app(app_label) # Create activity Activity.log_activity( user=user, action='delete', description=f'Deleted {sender._meta.verbose_name}: {str(instance)}', category=category, content_type=ContentType.objects.get_for_model(sender), object_id=instance.pk, is_successful=True, metadata={ 'model': sender._meta.label, 'object_repr': str(instance), 'deleted_fields': get_model_fields(instance) } ) logger.debug(f"Deletion activity logged for {sender._meta.label}: {instance.pk}") except Exception as e: logger.error(f"Failed to log deletion activity for {sender._meta.label}: {str(e)}") # Custom signal handlers @receiver(activity_logged) def handle_activity_logged(sender, activity, **kwargs): """ Handle activity logged signal. This signal handler performs additional processing when an activity is logged, such as updating summaries and triggering notifications. Args: sender: The sender of the signal activity: Activity instance that was logged **kwargs: Additional keyword arguments """ try: # Update activity summary asynchronously if hasattr(settings, 'CELERY_TASK_ALWAYS_EAGER') and not settings.CELERY_TASK_ALWAYS_EAGER: from .tasks import update_activity_summary update_activity_summary.delay( activity.created_at.date(), activity.user_id if activity.user else None, activity.category_id if activity.category else None ) else: # Update synchronously if Celery is not available update_activity_summary_sync( activity.created_at.date(), activity.user_id if activity.user else None, activity.category_id if activity.category else None ) # Trigger real-time notifications if enabled if getattr(settings, 'ACTIVITIES_REALTIME_ENABLED', False): send_realtime_notification(activity) logger.debug(f"Activity logged signal handled for activity {activity.id}") except Exception as e: logger.error(f"Failed to handle activity logged signal: {str(e)}") @receiver(bulk_activity_logged) def handle_bulk_activity_logged(sender, activities, **kwargs): """ Handle bulk activity logged signal. This signal handler performs batch processing when multiple activities are logged at once. Args: sender: The sender of the signal activities: List of Activity instances that were logged **kwargs: Additional keyword arguments """ try: # Group activities by date, user, and category for efficient summary updates summary_keys = set() for activity in activities: summary_key = ( activity.created_at.date(), activity.user_id if activity.user else None, activity.category_id if activity.category else None ) summary_keys.add(summary_key) # Update summaries if hasattr(settings, 'CELERY_TASK_ALWAYS_EAGER') and not settings.CELERY_TASK_ALWAYS_EAGER: from .tasks import update_activity_summaries_bulk update_activity_summaries_bulk.delay(list(summary_keys)) else: # Update synchronously if Celery is not available for date, user_id, category_id in summary_keys: update_activity_summary_sync(date, user_id, category_id) logger.info(f"Bulk activity logged signal handled for {len(activities)} activities") except Exception as e: logger.error(f"Failed to handle bulk activity logged signal: {str(e)}") # Utility functions def get_or_create_category_for_app(app_label): """ Get or create an activity category for a Django app. Args: app_label: Django app label Returns: ActivityCategory: Category instance for the app """ category_mapping = { 'accounts': { 'name': 'User Management', 'code': 'user_management', 'color': '#007bff', 'icon': 'fas fa-users' }, 'campaigns': { 'name': 'Campaign Management', 'code': 'campaign_management', 'color': '#28a745', 'icon': 'fas fa-bullhorn' }, 'channels': { 'name': 'Channel Management', 'code': 'channel_management', 'color': '#ffc107', 'icon': 'fas fa-tv' }, 'analytics': { 'name': 'Analytics & Reporting', 'code': 'analytics', 'color': '#17a2b8', 'icon': 'fas fa-chart-bar' }, 'vast': { 'name': 'VAST Ad Serving', 'code': 'vast_serving', 'color': '#6f42c1', 'icon': 'fas fa-play-circle' } } category_data = category_mapping.get(app_label, { 'name': f'{app_label.title()} Management', 'code': f'{app_label}_management', 'color': '#6c757d', 'icon': 'fas fa-cog' }) category, created = ActivityCategory.objects.get_or_create( code=category_data['code'], defaults=category_data ) return category def get_model_fields(instance): """ Get model fields and their values for logging. Args: instance: Model instance Returns: dict: Dictionary of field names and values """ fields = {} for field in instance._meta.fields: try: value = getattr(instance, field.name) # Convert non-serializable values if hasattr(value, 'isoformat'): # datetime objects value = value.isoformat() elif hasattr(value, '__str__') and not isinstance(value, (str, int, float, bool, type(None))): value = str(value) fields[field.name] = value except Exception: fields[field.name] = '' return fields def update_activity_summary_sync(date, user_id, category_id): """ Update activity summary synchronously. Args: date: Date for the summary user_id: User ID (optional) category_id: Category ID (optional) """ try: with transaction.atomic(): # Get or create summary summary, created = ActivitySummary.objects.get_or_create( date=date, user_id=user_id, category_id=category_id ) # Calculate statistics activities = Activity.objects.filter( created_at__date=date ) if user_id: activities = activities.filter(user_id=user_id) if category_id: activities = activities.filter(category_id=category_id) # Update summary fields summary.total_activities = activities.count() summary.successful_activities = activities.filter(is_successful=True).count() summary.failed_activities = activities.filter(is_successful=False).count() # Calculate average duration durations = activities.filter( duration_ms__isnull=False ).values_list('duration_ms', flat=True) if durations: summary.avg_duration_ms = sum(durations) / len(durations) # Count unique IPs summary.unique_ips = activities.values('ip_address').distinct().count() summary.save() except Exception as e: logger.error(f"Failed to update activity summary: {str(e)}") def send_realtime_notification(activity): """ Send real-time notification for activity. Args: activity: Activity instance """ try: # This would integrate with WebSocket or Server-Sent Events # For now, we'll just cache the latest activities cache_key = 'latest_activities' latest = cache.get(cache_key, []) # Add new activity to the beginning latest.insert(0, { 'id': str(activity.id), 'user': activity.user.email if activity.user else 'System', 'action': activity.get_action_display(), 'description': activity.description, 'timestamp': activity.created_at.isoformat(), 'is_successful': activity.is_successful }) # Keep only last 50 activities latest = latest[:50] # Cache for 1 hour cache.set(cache_key, latest, 3600) except Exception as e: logger.error(f"Failed to send real-time notification: {str(e)}") # Cleanup signals @receiver(activity_cleanup_completed) def handle_cleanup_completed(sender, deleted_count, **kwargs): """ Handle activity cleanup completion. Args: sender: The sender of the signal deleted_count: Number of activities deleted **kwargs: Additional keyword arguments """ logger.info(f"Activity cleanup completed: {deleted_count} activities deleted") # Log cleanup activity try: Activity.log_activity( user=None, action='system_cleanup', description=f'Automatic cleanup completed: {deleted_count} old activities removed', category=get_or_create_category_for_app('activities'), is_successful=True, metadata={ 'deleted_count': deleted_count, 'cleanup_timestamp': timezone.now().isoformat() } ) except Exception as e: logger.error(f"Failed to log cleanup activity: {str(e)}") def create_default_activity_categories(sender, **kwargs): """ Create default activity categories if they don't exist. This method creates the basic activity categories needed for proper activity classification and filtering. Categories created: - Authentication (login, logout, password changes) - Content Management (create, update, delete content) - Campaign Management (campaign operations) - System Administration (user management, settings) - Analytics (report generation, data export) - Security (failed logins, permission changes) """ try: from .models import ActivityCategory # Define default activity categories default_categories = [ { "name": "Authentication", "code": "auth", "description": "User authentication activities (login, logout, password changes)", "color": "#007bff", "icon": "fas fa-sign-in-alt", "is_system": True, "retention_days": 90 }, { "name": "Content Management", "code": "content", "description": "Content creation, modification, and deletion activities", "color": "#28a745", "icon": "fas fa-file-alt", "is_system": True, "retention_days": 365 }, { "name": "Campaign Management", "code": "campaign", "description": "Campaign creation, modification, and management activities", "color": "#ffc107", "icon": "fas fa-bullhorn", "is_system": True, "retention_days": 730 }, { "name": "System Administration", "code": "admin", "description": "Administrative activities (user management, system settings)", "color": "#dc3545", "icon": "fas fa-cogs", "is_system": True, "retention_days": 1095 }, { "name": "Analytics & Reporting", "code": "analytics", "description": "Analytics viewing, report generation, and data export activities", "color": "#6f42c1", "icon": "fas fa-chart-bar", "is_system": True, "retention_days": 180 }, { "name": "Security Events", "code": "security", "description": "Security-related events (failed logins, permission changes)", "color": "#fd7e14", "icon": "fas fa-shield-alt", "is_system": True, "retention_days": 1095 } ] # Create categories if they don't exist for category_data in default_categories: category, created = ActivityCategory.objects.get_or_create( code=category_data["code"], defaults=category_data ) if created: print(f"Created default activity category: {category.name}") except Exception as e: # Silently fail during migrations or if models aren't ready print(f"Warning: Could not create default activity categories: {e}") def setup_periodic_cleanup(sender, **kwargs): """ Set up periodic cleanup tasks for old activity records. This method configures Celery periodic tasks to automatically clean up old activity records based on retention policies. Tasks configured: - Daily cleanup of expired activity records - Weekly aggregation of activity statistics - Monthly archival of old activity data """ try: from django_celery_beat.models import PeriodicTask, CrontabSchedule import json # Create daily cleanup task schedule, created = CrontabSchedule.objects.get_or_create( minute=0, hour=2, # Run at 2 AM daily day_of_week='*', day_of_month='*', month_of_year='*', ) PeriodicTask.objects.get_or_create( crontab=schedule, name='activities_daily_cleanup', task='apps.activities.tasks.cleanup_expired_activities', defaults={ 'args': json.dumps([]), 'kwargs': json.dumps({}), 'enabled': True, } ) print("Activities periodic cleanup tasks configured") except Exception as e: print(f"Warning: Could not setup periodic cleanup: {e}")