# -*- coding: utf-8 -*- """ Adtlas Accounts Views - Class-Based View Implementation This module contains class-based views for user authentication, profile management, and account-related functionality in the Adtlas DAI Management System. Features: - Email-based authentication - User profile management - Role-based access control - Password reset functionality - Session management - Activity tracking Author: Adtlas Development Team Version: 2.0.0 Last Updated: 2025-07-08 """ from django.shortcuts import render, redirect, get_object_or_404 from django.contrib.auth import authenticate, login as auth_login, logout as auth_logout from django.contrib.auth.mixins import LoginRequiredMixin from django.contrib import messages from django.http import JsonResponse from django.utils import timezone from django.core.exceptions import ValidationError from django.db import transaction from django.views.decorators.http import require_http_methods from django.views.decorators.csrf import csrf_protect from django.utils.decorators import method_decorator from django.contrib.auth.password_validation import validate_password from django.core.paginator import Paginator from django.db.models import Q from django.views.generic import TemplateView, View from django.views.generic.base import RedirectView from .models import User, Profile, UserSession, UserActivity, Role, UserRole from apps.core.models import ActivityLog class LoginView(View): """ User login view with enhanced security features. Handles email-based authentication with account lockout protection, activity logging, and session management. """ template_name = 'accounts/login.html' @method_decorator(csrf_protect) def dispatch(self, request, *args, **kwargs): return super().dispatch(request, *args, **kwargs) def get(self, request): if request.user.is_authenticated: return redirect('core:dashboard') return render(request, self.template_name) def post(self, request): if request.user.is_authenticated: return redirect('core:dashboard') email = request.POST.get('email') or request.POST.get('username') password = request.POST.get('password') if not email or not password: messages.error(request, 'Please provide both email and password.') return render(request, self.template_name) # Check for account lockout try: user = User.objects.get(email=email) if user.is_account_locked(): messages.error( request, f'Account is locked due to too many failed attempts. ' f'Try again after {user.account_locked_until.strftime("%H:%M")}.' ) return render(request, self.template_name) except User.DoesNotExist: # Still show generic error to prevent email enumeration pass # Authenticate user user = authenticate(request, email=email, password=password) if user is not None: if user.is_active: auth_login(request, user) # Update last login and activity user.last_login = timezone.now() user.last_activity = timezone.now() user.save(update_fields=['last_login', 'last_activity']) # Redirect to next or dashboard next_url = request.GET.get('next', '/dashboard/') return redirect(next_url) else: messages.error(request, 'Your account has been deactivated. Please contact support.') else: messages.error(request, 'Invalid email or password.') return render(request, self.template_name) class LogoutView(LoginRequiredMixin, RedirectView): """ User logout view with session cleanup. """ url = 'accounts:login' def get(self, request, *args, **kwargs): auth_logout(request) messages.success(request, 'You have been successfully logged out.') return super().get(request, *args, **kwargs) class ProfileView(LoginRequiredMixin, TemplateView): """ User profile view and edit functionality. """ template_name = 'accounts/profile.html' def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) profile, created = Profile.objects.get_or_create(user=self.request.user) # Get user's roles user_roles = UserRole.objects.filter( user=self.request.user, is_active=True ).select_related('role') # Get user's recent activities recent_activities = UserActivity.objects.filter( user=self.request.user ).order_by('-created_at')[:10] context.update({ 'profile': profile, 'user_roles': user_roles, 'recent_activities': recent_activities, 'timezone_choices': [ ('UTC', 'UTC'), ('America/New_York', 'Eastern Time'), ('America/Chicago', 'Central Time'), ('America/Denver', 'Mountain Time'), ('America/Los_Angeles', 'Pacific Time'), ('Europe/London', 'London'), ('Europe/Paris', 'Paris'), ('Asia/Tokyo', 'Tokyo'), ], 'language_choices': [ ('en', 'English'), ('es', 'Spanish'), ('fr', 'French'), ('de', 'German'), ], 'theme_choices': [ ('light', 'Light'), ('dark', 'Dark'), ('auto', 'Auto'), ] }) return context def post(self, request, *args, **kwargs): profile, created = Profile.objects.get_or_create(user=request.user) # Update profile information profile.phone = request.POST.get('phone', '') profile.bio = request.POST.get('bio', '') profile.job_title = request.POST.get('job_title', '') profile.timezone = request.POST.get('timezone', 'UTC') profile.language = request.POST.get('language', 'en') profile.theme = request.POST.get('theme', 'light') profile.notifications_enabled = request.POST.get('notifications_enabled') == 'on' profile.email_notifications = request.POST.get('email_notifications') == 'on' # Handle avatar upload if 'avatar' in request.FILES: profile.avatar = request.FILES['avatar'] try: profile.save() # Update user basic info request.user.first_name = request.POST.get('first_name', '') request.user.last_name = request.POST.get('last_name', '') request.user.save(update_fields=['first_name', 'last_name']) messages.success(request, 'Profile updated successfully.') # Log activity UserActivity.objects.create( user=request.user, action='update', object_type='Profile', object_id=str(profile.id), object_repr=str(profile), details={'updated_fields': list(request.POST.keys())} ) except Exception as e: messages.error(request, f'Error updating profile: {str(e)}') return redirect('accounts:profile') class ChangePasswordView(LoginRequiredMixin, View): """ Change user password with validation. """ http_method_names = ['post'] def post(self, request): current_password = request.POST.get('current_password') new_password = request.POST.get('new_password') confirm_password = request.POST.get('confirm_password') if not all([current_password, new_password, confirm_password]): messages.error(request, 'All password fields are required.') return redirect('accounts:profile') # Verify current password if not request.user.check_password(current_password): messages.error(request, 'Current password is incorrect.') return redirect('accounts:profile') # Check new password confirmation if new_password != confirm_password: messages.error(request, 'New passwords do not match.') return redirect('accounts:profile') # Validate new password try: validate_password(new_password, request.user) except ValidationError as e: messages.error(request, ' '.join(e.messages)) return redirect('accounts:profile') # Update password request.user.set_password(new_password) request.user.password_changed_at = timezone.now() request.user.save() # Log activity UserActivity.objects.create( user=request.user, action='update', object_type='User', object_id=str(request.user.id), object_repr=str(request.user), details={'action': 'password_changed'} ) messages.success(request, 'Password changed successfully. Please log in again.') auth_logout(request) return redirect('accounts:login') class UserSessionsAPIView(LoginRequiredMixin, View): """ API endpoint to get user's active sessions. """ def get(self, request): sessions = UserSession.objects.filter( user=request.user, is_active=True ).order_by('-last_activity') session_data = [] for session in sessions: session_data.append({ 'id': session.id, 'session_key': session.session_key[:8] + '...', 'ip_address': session.ip_address, 'location': session.location, 'last_activity': session.last_activity.isoformat(), 'expires_at': session.expires_at.isoformat(), 'is_current': session.session_key == request.session.session_key }) return JsonResponse({'sessions': session_data}) class TerminateSessionView(LoginRequiredMixin, View): """ Terminate a specific user session. """ http_method_names = ['post'] def post(self, request, session_id): try: session = UserSession.objects.get( id=session_id, user=request.user, is_active=True ) # Don't allow terminating current session if session.session_key == request.session.session_key: return JsonResponse({ 'success': False, 'error': 'Cannot terminate current session' }) session.is_active = False session.save() # Log activity UserActivity.objects.create( user=request.user, action='admin', object_type='UserSession', object_id=str(session.id), object_repr=str(session), details={'action': 'session_terminated'} ) return JsonResponse({'success': True}) except UserSession.DoesNotExist: return JsonResponse({ 'success': False, 'error': 'Session not found' }) class UserActivitiesView(LoginRequiredMixin, TemplateView): """ View user's activity history with filtering and pagination. """ template_name = 'accounts/user_activities.html' def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) activities = UserActivity.objects.filter( user=self.request.user ).order_by('-created_at') # Filter by action type action_filter = self.request.GET.get('action') if action_filter: activities = activities.filter(action=action_filter) # Filter by date range date_from = self.request.GET.get('date_from') date_to = self.request.GET.get('date_to') if date_from: activities = activities.filter(created_at__date__gte=date_from) if date_to: activities = activities.filter(created_at__date__lte=date_to) # Search in object representation search = self.request.GET.get('search') if search: activities = activities.filter( Q(object_repr__icontains=search) | Q(object_type__icontains=search) ) # Pagination paginator = Paginator(activities, 25) page_number = self.request.GET.get('page') page_obj = paginator.get_page(page_number) context.update({ 'page_obj': page_obj, 'action_choices': UserActivity.ACTION_TYPES, 'current_filters': { 'action': action_filter, 'date_from': date_from, 'date_to': date_to, 'search': search, } }) return context class PasswordResetRequestView(TemplateView): """ Request password reset (placeholder for future implementation). """ template_name = 'accounts/password_reset_request.html' def post(self, request): email = request.POST.get('email') try: user = User.objects.get(email=email, is_active=True) # TODO: Implement password reset email sending messages.success( request, 'If the email exists in our system, you will receive password reset instructions.' ) except User.DoesNotExist: # Don't reveal whether the email exists messages.success( request, 'If the email exists in our system, you will receive password reset instructions.' ) return redirect('accounts:login') # API Views for AJAX requests class CheckEmailAvailabilityView(LoginRequiredMixin, View): """ Check if an email address is available for registration. """ def get(self, request): email = request.GET.get('email') if not email: return JsonResponse({'available': False, 'error': 'Email is required'}) exists = User.objects.filter(email=email).exists() return JsonResponse({'available': not exists}) class UserSearchAPIView(LoginRequiredMixin, View): """ Search users for mentions, assignments, etc. Requires appropriate permissions. """ def get(self, request): query = request.GET.get('q', '') if len(query) < 2: return JsonResponse({'users': []}) # Check if user has permission to search users if not request.user.has_perm('accounts.view_User'): return JsonResponse({'error': 'Permission denied'}, status=403) users = User.objects.filter( Q(email__icontains=query) | Q(first_name__icontains=query) | Q(last_name__icontains=query), is_active=True )[:10] user_data = [] for user in users: user_data.append({ 'id': str(user.id), 'email': user.email, 'name': user.get_full_name(), 'avatar': user.profile.avatar.url if hasattr(user, 'profile') and user.profile.avatar else None }) return JsonResponse({'users': user_data}) # Convenience function-based view aliases for backward compatibility login_view = LoginView.as_view() logout_view = LogoutView.as_view() profile_view = ProfileView.as_view() change_password = ChangePasswordView.as_view() user_sessions_api = UserSessionsAPIView.as_view() terminate_session = TerminateSessionView.as_view() user_activities_view = UserActivitiesView.as_view() password_reset_request = PasswordResetRequestView.as_view() check_email_availability = CheckEmailAvailabilityView.as_view() user_search_api = UserSearchAPIView.as_view()