# -*- coding: utf-8 -*- """ Adtlas Accounts Signals This module contains Django signals for the accounts app to handle automatic actions when user-related events occur. Features: - Auto-create user profiles - Log user activities - Manage user sessions - Handle role assignments - Password change tracking Author: Adtlas Development Team Version: 2.0.0 Last Updated: 2025-07-08 """ from django.utils import timezone from django.dispatch import receiver from django.contrib.sessions.models import Session from django.db.models.signals import post_save, pre_save, post_delete from django.contrib.auth.signals import user_logged_in, user_logged_out, user_login_failed from apps.accounts.models import User, Profile, UserRole, UserSession from apps.activities.models import Activity @receiver(post_save, sender=User) def create_user_profile(sender, instance, created, **kwargs): """ Create a Profile when a new user is created. """ if created: Profile.objects.get_or_create(user=instance) @receiver(post_save, sender=User) def save_user_profile(sender, instance, **kwargs): """ Save the Profile when the user is saved. """ if hasattr(instance, "profile"): instance.profile.save() @receiver(pre_save, sender=User) def track_password_change(sender, instance, **kwargs): """ Track when a user's password is changed. """ if instance.pk: try: old_instance = User.objects.get(pk=instance.pk) if old_instance.password != instance.password: instance.password_changed_at = timezone.now() # Reset failed login attempts on password change instance.failed_login_attempts = 0 instance.account_locked_until = None except User.DoesNotExist: pass @receiver(user_logged_in) def log_user_login(sender, request, user, **kwargs): """ Log user login activity and create/update session tracking. """ # Get client IP address x_forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR") if x_forwarded_for: ip_address = x_forwarded_for.split(",")[0] else: ip_address = request.META.get("REMOTE_ADDR") # Get user agent user_agent = request.META.get("HTTP_USER_AGENT", "") # Update user's last activity user.last_activity = timezone.now() user.failed_login_attempts = 0 # Reset on successful login user.save(update_fields=["last_activity", "failed_login_attempts"]) # Create user activity log Activity.log_activity( user=user, action="login", description=f"User {user.email if user else 'System'} logged in", metadata={ "ip_address": ip_address, "user_agent": user_agent } ) # Create or update session tracking if hasattr(request, "session"): session_key = request.session.session_key if session_key: session = Session.objects.filter(session_key=session_key).first() expires_at = session.expire_date if session else timezone.now() + timezone.timedelta(days=1) UserSession.objects.update_or_create( session_key=session_key, defaults={ "user": user, "ip_address": ip_address, "user_agent": user_agent, "is_active": True, "last_activity": timezone.now(), "expires_at": expires_at } ) @receiver(user_logged_out) def log_user_logout(sender, request, user, **kwargs): """ Log user logout activity and update session tracking. """ if user: # Get client IP address x_forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR") if x_forwarded_for: ip_address = x_forwarded_for.split(",")[0] else: ip_address = request.META.get("REMOTE_ADDR") # Get user agent user_agent = request.META.get("HTTP_USER_AGENT", "") # Create user activity log Activity.log_activity( user=user, action="logout", description=f"User {user.email if user else 'System'} logged out", metadata={ "ip_address": ip_address, "user_agent": user_agent } ) # Mark session as inactive if hasattr(request, "session") and request.session.session_key: UserSession.objects.filter( session_key=request.session.session_key ).update(is_active=False) @receiver(user_login_failed) def log_failed_login(sender, credentials, request, **kwargs): """ Log failed login attempts and implement account lockout. """ # Get client IP address x_forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR") if x_forwarded_for: ip_address = x_forwarded_for.split(",")[0] else: ip_address = request.META.get("REMOTE_ADDR") # Get user agent user_agent = request.META.get("HTTP_USER_AGENT", "") # Try to find the user by email/username email = credentials.get("email") or credentials.get("username") if email: try: user = User.objects.get(email=email) # Increment failed login attempts user.failed_login_attempts += 1 # Lock account after 5 failed attempts for 30 minutes if user.failed_login_attempts >= 5: user.account_locked_until = timezone.now() + timezone.timedelta(minutes=30) user.save(update_fields=["failed_login_attempts", "account_locked_until"]) # Create user activity log Activity.log_activity( user=user, action="login_failed", description=f"Failed login attempt for user {user.email}", metadata={ "ip_address": ip_address, "user_agent": user_agent, "attempt_count": user.failed_login_attempts } ) except User.DoesNotExist: # Skip logging for unknown emails since UserActivity requires a valid user # This prevents IntegrityError when user=None is passed # Consider implementing a separate SecurityLog model for tracking # failed attempts with non-existent emails if needed pass @receiver(post_save, sender=UserRole) def log_role_assignment(sender, instance, created, **kwargs): """ Log when roles are assigned to users. """ if created: Activity.log_activity( user=instance.user, action="role_assigned", description=f"Role {instance.role.name} assigned to user {instance.user.email}", metadata={ "role": instance.role.name, "role_code": instance.role.code, "assigned_by": instance.assigned_by.email if instance.assigned_by else None } ) @receiver(post_delete, sender=UserRole) def log_role_removal(sender, instance, **kwargs): """ Log when roles are removed from users. """ Activity.log_activity( user=instance.user, action="role_removed", description=f"Role {instance.role.name} removed from user {instance.user.email}", metadata={ "role": instance.role.name, "role_code": instance.role.code } ) def cleanup_expired_sessions(): """ Utility function to clean up expired sessions. This can be called by a management command or celery task. """ expired_sessions = UserSession.objects.filter( expires_at__lt=timezone.now(), is_active=True ) count = expired_sessions.count() expired_sessions.update(is_active=False) return count def cleanup_old_activities(days=90): """ Utility function to clean up old activity logs. Keep only activities from the last X days. """ cutoff_date = timezone.now() - timezone.timedelta(days=days) deleted_count, _ = Activity.objects.filter( created_at__lt=cutoff_date ).delete() return deleted_count # create default admin def create_default_admin(sender, **kwargs): """ Create a default admin user if none exists. This method checks if there are any users in the database. If not, it creates a default admin user with the specified email and password. Note: - This method should be called after migrations have been applied. - It's a one-time setup operation. """ try: from django.db.models import Q from apps.accounts.models import User # check if admin user already exists with the email admin@adtlas.tv and is_superuser=True and is_staff=True and is_active=True if not User.objects.filter( Q(email="admin@adtlas.tv") & Q(is_superuser=True) & Q(is_staff=True) & Q(is_active=True) ).exists(): User.objects.create_superuser( email="admin@adtlas.tv", password="admin123", ) print("Default admin user created.") else: print("Admin user already exists. Skipping creation.") except Exception as e: # Silently fail during migrations or if models aren"t ready pass # Create default roles def create_default_roles(sender, **kwargs): """ Create default system roles if they don"t exist. This method creates the basic role structure needed for the application to function properly. """ try: from django.contrib.auth.models import Permission from django.contrib.contenttypes.models import ContentType from apps.accounts.models import Role # Define default roles default_roles = [ { "name": "Super Administrator", "code": "super_admin", "description": "Full system access with all permissions", "role_type": "system", "level": 1, "is_active": True, "is_default": False }, { "name": "Administrator", "code": "admin", "description": "Administrative access to most features", "role_type": "system", "level": 2, "is_active": True, "is_default": False }, { "name": "Manager", "code": "manager", "description": "Management access to campaigns and analytics", "role_type": "system", "level": 3, "is_active": True, "is_default": False }, { "name": "Operator", "code": "operator", "description": "Operational access to create and manage content", "role_type": "system", "level": 4, "is_active": True, "is_default": False }, { "name": "Viewer", "code": "viewer", "description": "Read-only access to view data and reports", "role_type": "system", "level": 5, "is_active": True, "is_default": True } ] # Create roles if they don"t exist for role_data in default_roles: role, created = Role.objects.get_or_create( code=role_data["code"], defaults=role_data ) if created: print(f"Created default role: {role.name}") except Exception as e: # Silently fail during migrations or if models aren"t ready pass