# -*- coding: utf-8 -*- """ Focus Accounts Views Module This module contains all class-based views for the accounts application, following the same pattern as the channels app for consistency. Author: Focus Development Team Version: 2.0.0 Created: 2025-01-01 Last Updated: 2025-01-01 """ # Standard library imports import logging from datetime import datetime, timedelta from typing import Any, Dict, Optional # Django core imports from django.contrib import messages from django.contrib.auth.mixins import LoginRequiredMixin, PermissionRequiredMixin from django.contrib.auth.models import Permission from django.core.exceptions import PermissionDenied from django.core.paginator import Paginator, PageNotAnInteger, EmptyPage from django.db import transaction from django.db.models import Q, Count, Avg from django.http import HttpRequest, HttpResponse, JsonResponse, Http404 from django.shortcuts import get_object_or_404, redirect, render from django.urls import reverse_lazy, reverse from django.utils import timezone from django.core.exceptions import ValidationError from django.utils.decorators import method_decorator from django.utils.translation import gettext_lazy as _ from django.views.decorators.cache import cache_page from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_http_methods, require_POST from django.views.generic import ( ListView, DetailView, CreateView, UpdateView, DeleteView, TemplateView, View ) from django.contrib.auth.decorators import login_required from rest_framework import viewsets from rest_framework.permissions import IsAuthenticated # Local application imports from .models import User, Profile, Role, UserRole, UserSession from .forms import ( UserUpdateForm, ProfileUpdateForm, RoleForm, UserCreateForm ) from apps.common.utils import get_client_ip from apps.activities.models import Activity # Configure logging for this module logger = logging.getLogger(__name__) # ============================================================================ # User Management Views # ============================================================================ class UserListView(LoginRequiredMixin, PermissionRequiredMixin, ListView): """ Display a paginated list of users with filtering and search capabilities. This view provides comprehensive user management functionality including: - Search by name, email, username - Filter by status, role, date joined - Bulk operations - Export functionality - AJAX support for dynamic loading """ model = User template_name = 'accounts/users/list.html' context_object_name = 'users' permission_required = 'accounts.view_user' paginate_by = 20 def get_context_data(self, **kwargs): """ Add additional context data for the template. Returns: dict: Updated context data """ context = super().get_context_data(**kwargs) context.update({ "title": _("Users"), }) return context def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse: """ Handles the HTTP GET request and returns the rendered users list page template. Args: request (HttpRequest): The HTTP request object. *args: Variable length argument list. **kwargs: Arbitrary keyword arguments. Returns: HttpResponse: The rendered users list page template. Query Parameters: - search: Search term for user name/email/username - status: Filter by user status - role: Filter by user role - date_from: Start date for filtering - date_to: End date for filtering - date_range: Predefined date range - sort_by: Sorting field and direction - page: Page number for pagination - per_page: Number of items per page (default: 20) """ # Check for AJAX requests if request.headers.get('X-Requested-With') == 'XMLHttpRequest': return self._handle_ajax_request(request) # Get filtered queryset queryset = self._get_filtered_queryset(request) # Apply pagination paginator = Paginator(queryset, self._get_items_per_page(request)) page_number = request.GET.get('page', 1) try: page_obj = paginator.get_page(page_number) except PageNotAnInteger: page_obj = paginator.get_page(1) except EmptyPage: page_obj = paginator.get_page(paginator.num_pages) # Log user activity for auditing self._log_user_activity(request) # Build context context = self._build_context(request, page_obj, paginator) return render(request, self.template_name, context) def _get_filtered_queryset(self, request: HttpRequest): """ Build filtered queryset based on request parameters. Args: request: Django HTTP request object Returns: QuerySet: Filtered user queryset """ # Base queryset with optimizations queryset = User.objects.all().select_related().prefetch_related('roles') # Apply filters queryset = self._apply_search_filter(queryset, request) queryset = self._apply_status_filter(queryset, request) queryset = self._apply_role_filter(queryset, request) queryset = self._apply_date_filters(queryset, request) # Apply sorting queryset = self._apply_sorting(queryset, request) return queryset def _apply_search_filter(self, queryset, request): """ Apply search filter to queryset. Args: queryset: Base queryset request: Django HTTP request object Returns: QuerySet: Filtered queryset """ search_query = request.GET.get('search', '').strip() if search_query: queryset = queryset.filter( Q(first_name__icontains=search_query) | Q(last_name__icontains=search_query) | Q(email__icontains=search_query) | Q(username__icontains=search_query) ) return queryset def _apply_status_filter(self, queryset, request): """ Apply status filter to queryset. Args: queryset: Base queryset request: Django HTTP request object Returns: QuerySet: Filtered queryset """ status = request.GET.get('status') if status == 'active': queryset = queryset.filter(is_active=True) elif status == 'inactive': queryset = queryset.filter(is_active=False) elif status == 'verified': queryset = queryset.filter(is_verified=True) elif status == 'unverified': queryset = queryset.filter(is_verified=False) return queryset def _apply_role_filter(self, queryset, request): """ Apply role filter to queryset. Args: queryset: Base queryset request: Django HTTP request object Returns: QuerySet: Filtered queryset """ role_id = request.GET.get('role') if role_id: try: queryset = queryset.filter(roles__id=role_id) except (ValueError, TypeError): pass return queryset def _apply_date_filters(self, queryset, request): """ Apply date range filters to queryset. Args: queryset: Base queryset request: Django HTTP request object Returns: QuerySet: Filtered queryset """ date_from = request.GET.get('date_from') date_to = request.GET.get('date_to') date_range = request.GET.get('date_range') # Handle predefined date ranges if date_range: now = timezone.now() if date_range == 'today': date_from = now.date() date_to = now.date() elif date_range == 'week': date_from = (now - timedelta(days=7)).date() date_to = now.date() elif date_range == 'month': date_from = (now - timedelta(days=30)).date() date_to = now.date() elif date_range == 'year': date_from = (now - timedelta(days=365)).date() date_to = now.date() # Apply date filters if date_from: try: queryset = queryset.filter(date_joined__date__gte=date_from) except (ValueError, TypeError): pass if date_to: try: queryset = queryset.filter(date_joined__date__lte=date_to) except (ValueError, TypeError): pass return queryset def _apply_sorting(self, queryset, request): """ Apply sorting to queryset. Args: queryset: Base queryset request: Django HTTP request object Returns: QuerySet: Sorted queryset """ sort_by = request.GET.get('sort_by', '-date_joined') # Validate sort field valid_sort_fields = [ 'first_name', '-first_name', 'last_name', '-last_name', 'email', '-email', 'username', '-username', 'date_joined', '-date_joined', 'last_login', '-last_login', 'is_active', '-is_active', ] if sort_by in valid_sort_fields: queryset = queryset.order_by(sort_by) else: queryset = queryset.order_by('-date_joined') return queryset def _get_items_per_page(self, request): """ Get number of items per page from request. Args: request: Django HTTP request object Returns: int: Number of items per page """ try: per_page = int(request.GET.get('per_page', self.paginate_by)) return min(max(per_page, 10), 100) # Limit between 10 and 100 except (ValueError, TypeError): return self.paginate_by def _handle_ajax_request(self, request): """ Handle AJAX requests for dynamic content loading. Args: request: Django HTTP request object Returns: JsonResponse: JSON response with user data """ queryset = self._get_filtered_queryset(request) # Apply pagination paginator = Paginator(queryset, self._get_items_per_page(request)) page_number = request.GET.get('page', 1) page_obj = paginator.get_page(page_number) # Serialize user data users_data = [] for user in page_obj: users_data.append({ 'id': str(user.id), 'username': user.username, 'email': user.email, 'first_name': user.first_name, 'last_name': user.last_name, 'is_active': user.is_active, 'is_verified': user.is_verified, 'date_joined': user.date_joined.isoformat() if user.date_joined else None, 'last_login': user.last_login.isoformat() if user.last_login else None, }) return JsonResponse({ 'users': users_data, 'pagination': { 'current_page': page_obj.number, 'total_pages': paginator.num_pages, 'total_items': paginator.count, 'has_previous': page_obj.has_previous(), 'has_next': page_obj.has_next(), } }) def _log_user_activity(self, request): """ Log user activity for auditing purposes. Args: request: Django HTTP request object """ try: logger.info( f"User {request.user.username} accessed user list view", extra={ 'user_id': request.user.id, 'ip_address': get_client_ip(request), 'user_agent': request.META.get('HTTP_USER_AGENT', ''), } ) except Exception as e: logger.error(f"Failed to log user activity: {e}") def _build_context(self, request, page_obj, paginator): """ Build template context data. Args: request: Django HTTP request object page_obj: Paginated page object paginator: Paginator instance Returns: dict: Template context data """ # Get filter options roles = Role.objects.all().order_by('name') # Get statistics total_users = User.objects.count() active_users = User.objects.filter(is_active=True).count() verified_users = User.objects.filter(is_verified=True).count() context = { 'users': page_obj, 'page_obj': page_obj, 'paginator': paginator, 'title': _("Users"), 'roles': roles, 'stats': { 'total': total_users, 'active': active_users, 'verified': verified_users, 'inactive': total_users - active_users, }, 'current_filters': { 'search': request.GET.get('search', ''), 'status': request.GET.get('status', ''), 'role': request.GET.get('role', ''), 'date_from': request.GET.get('date_from', ''), 'date_to': request.GET.get('date_to', ''), 'date_range': request.GET.get('date_range', ''), 'sort_by': request.GET.get('sort_by', '-date_joined'), 'per_page': request.GET.get('per_page', self.paginate_by), } } return context class UserDetailView(LoginRequiredMixin, PermissionRequiredMixin, DetailView): """ Display detailed user information. """ model = User template_name = 'accounts/users/detail.html' context_object_name = 'user_obj' permission_required = 'accounts.view_user' pk_url_kwarg = 'user_id' class UserCreateView(LoginRequiredMixin, PermissionRequiredMixin, CreateView): """ Create a new user. """ model = User form_class = UserCreateForm template_name = 'accounts/users/form.html' permission_required = 'accounts.add_user' success_url = reverse_lazy('accounts:user_list') class UserUpdateView(LoginRequiredMixin, PermissionRequiredMixin, UpdateView): """ Update an existing user. """ model = User form_class = UserUpdateForm template_name = 'accounts/users/form.html' permission_required = 'accounts.change_user' pk_url_kwarg = 'user_id' success_url = reverse_lazy('accounts:user_list') class UserDeleteView(LoginRequiredMixin, PermissionRequiredMixin, DeleteView): """ Delete a user. """ model = User template_name = 'accounts/users/confirm_delete.html' permission_required = 'accounts.delete_user' pk_url_kwarg = 'user_id' success_url = reverse_lazy('accounts:user_list') # ============================================================================ # Role Management Views # ============================================================================ class EmptyRole: """Empty role object for template consistency.""" def __init__(self): self.name = '' self.description = '' self.code = '' self.is_active = True self.priority = 50 self.is_system_role = False self.can_be_assigned = True self.max_users = None self.expires_at = None self.parent_role = None def __bool__(self): return False class RoleListView(LoginRequiredMixin, PermissionRequiredMixin, View): """ Role List View This view is responsible for rendering the roles list page of the application. It extends the Django View class and defines the HTTP GET method to handle the request and return the rendered roles list page template with advanced filtering, searching, pagination, and AJAX support. Methods: get(request): Handles the HTTP GET request and returns the rendered roles list page template. Template: - 'accounts/roles/list.html': The HTML template used to render the roles list page. Features: - Advanced filtering by permi ssion type, status, and date ranges - Full-text search across role name and description - Pagination with customizable items per page - AJAX support for dynamic updates - Statistics and analytics - Real-time updates """ http_method_names = ["get"] template_name = "accounts/roles/list.html" permission_required = 'accounts.view_role' extra_context = { "title": "Roles - Adtlas", "author": "Adtlas Development Team", "description": "Role management for Adtlas advertising solutions. This page provides comprehensive role management and insights.", "keywords": "Adtlas, roles, permissions, role management, user management, data visualization", "active_menu": "accounts", } def get_context_data(self, **kwargs): """ Add roles page context data. Args: **kwargs: Additional keyword arguments Returns: dict: Updated context data """ context = super().get_context_data(**kwargs) context.update({ "title": _("Roles"), }) return context def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse: """ Handles the HTTP GET request and returns the rendered roles list page template. Args: request (HttpRequest): The HTTP request object. *args: Variable length argument list. **kwargs: Arbitrary keyword arguments. Returns: HttpResponse: The rendered roles list page template. Query Parameters: - search: Search term for role name/description - permission_type: Filter by permission type - status: Filter by role status - sort_by: Sorting field and direction - page: Page number for pagination - per_page: Number of items per page (default: 20) """ # Check for AJAX requests if request.headers.get('X-Requested-With') == 'XMLHttpRequest': return self._handle_ajax_request(request) # Get filtered queryset queryset = self._get_filtered_queryset(request) # Apply pagination paginator = Paginator(queryset, self._get_items_per_page(request)) page_number = request.GET.get('page', 1) try: page_obj = paginator.get_page(page_number) except (PageNotAnInteger, EmptyPage): page_obj = paginator.get_page(1) # Log user activity for auditing self._log_user_activity(request) # Build context context = self._build_context(request, page_obj, paginator) return render(request, self.template_name, context) def _get_filtered_queryset(self, request: HttpRequest): """ Build filtered queryset based on request parameters. Args: request: Django HTTP request object Returns: QuerySet: Filtered role queryset """ # Base queryset with optimizations and user count annotation queryset = Role.objects.all().prefetch_related('permissions').annotate( user_count=Count('userrole') ) # Apply filters queryset = self._apply_search_filter(queryset, request) queryset = self._apply_permission_type_filter(queryset, request) queryset = self._apply_status_filter(queryset, request) # Apply sorting queryset = self._apply_sorting(queryset, request) return queryset def _apply_search_filter(self, queryset, request): """ Apply search filter to queryset. Args: queryset: Base queryset request: Django HTTP request object Returns: QuerySet: Filtered queryset """ search_query = request.GET.get('search', '').strip() if search_query: queryset = queryset.filter( Q(name__icontains=search_query) | Q(description__icontains=search_query) ) return queryset def _apply_permission_type_filter(self, queryset, request): """ Apply permission type filter to queryset. Args: queryset: Base queryset request: Django HTTP request object Returns: QuerySet: Filtered queryset """ permission_type = request.GET.get('permission_type', '').strip() if permission_type: queryset = queryset.filter(permissions__codename__icontains=permission_type).distinct() return queryset def _apply_status_filter(self, queryset, request): """ Apply status filter to queryset. Args: queryset: Base queryset request: Django HTTP request object Returns: QuerySet: Filtered queryset """ status = request.GET.get('status', '').strip() if status == 'active': queryset = queryset.filter(is_active=True) elif status == 'inactive': queryset = queryset.filter(is_active=False) return queryset def _apply_sorting(self, queryset, request): """ Apply sorting to queryset. Args: queryset: Base queryset request: Django HTTP request object Returns: QuerySet: Sorted queryset """ sort_by = request.GET.get('sort_by', 'name') if sort_by == '-created_at': queryset = queryset.order_by('-created_at') elif sort_by == 'created_at': queryset = queryset.order_by('created_at') elif sort_by == 'user_count': queryset = queryset.order_by('-user_count') else: queryset = queryset.order_by('name') return queryset def _get_items_per_page(self, request: HttpRequest) -> int: """ Get number of items per page from request. Args: request: Django HTTP request object Returns: int: Items per page (default: 20) """ try: per_page = int(request.GET.get('per_page', 20)) return max(1, min(per_page, 100)) # Limit between 1 and 100 except (ValueError, TypeError): return 20 def _build_context(self, request: HttpRequest, page_obj, paginator): """ Build template context. Args: request: Django HTTP request object page_obj: Page object from paginator paginator: Paginator instance Returns: dict: Template context """ user = request.user context = { # Pagination data 'roles': page_obj.object_list, 'page_obj': page_obj, 'paginator': paginator, 'is_paginated': paginator.num_pages > 1, # Filter data 'permission_types': self._get_permission_types(), 'status_choices': self._get_status_choices(), 'sort_options': self._get_sort_options(), 'filter_params': self._get_current_filters(request), # Statistics 'stats': self._get_stats(user), # Additional data 'total_results': paginator.count, 'current_page': page_obj.number, 'total_pages': paginator.num_pages, # Filter states for template 'search_query': request.GET.get('search', ''), 'selected_permission_type': request.GET.get('permission_type', ''), 'selected_status': request.GET.get('status', ''), } # Add extra context context.update(self.extra_context) return context def _get_permission_types(self): """ Get available permission types for dropdown filter. Returns: list: Permission type choices """ return [ ('admin', 'Admin'), ('user', 'User'), ('content', 'Content'), ('system', 'System'), ] def _get_status_choices(self): """ Get available status choices for dropdown filter. Returns: list: Status choice tuples """ return [ ('all', _('All Status')), ('active', _('Active')), ('inactive', _('Inactive')), ] def _get_sort_options(self): """ Get available sort options for dropdown filter. Returns: list: Sort option tuples """ return [ ('name', _('Name')), ('-created_at', _('Newest')), ('created_at', _('Oldest')), ('user_count', _('User Count')), ] def _get_current_filters(self, request: HttpRequest) -> Dict[str, str]: """ Get current filter parameters for template. Args: request: Django HTTP request object Returns: dict: Current filter values """ return { 'search': request.GET.get('search', ''), 'permission_type': request.GET.get('permission_type', ''), 'status': request.GET.get('status', ''), 'sort_by': request.GET.get('sort_by', 'name'), } def _get_stats(self, user): """ Get role statistics for the user. Args: user: User object Returns: dict: Statistics data """ try: total_roles = Role.objects.count() active_roles = Role.objects.filter(is_active=True).count() total_permissions = Permission.objects.count() # Count assigned users (users with roles) assigned_users = UserRole.objects.filter(is_active=True).values('user').distinct().count() return { 'total_roles': total_roles, 'active_roles': active_roles, 'total_permissions': total_permissions, 'assigned_users': assigned_users, } except Exception: return { 'total_roles': 0, 'active_roles': 0, 'total_permissions': 0, 'assigned_users': 0, } def _log_user_activity(self, request: HttpRequest): """ Log user activity for auditing. Args: request: Django HTTP request object """ try: user = request.user Activity.log_activity( user=user, action="view_roles", description=f"User {user.email if user else 'System'} viewed roles list", request=request ) except Exception: # Silently fail if logging fails pass def _handle_ajax_request(self, request: HttpRequest) -> JsonResponse: """ Handle AJAX requests for dynamic updates. Args: request: Django HTTP request object Returns: JsonResponse: JSON response """ try: # Get filtered queryset queryset = self._get_filtered_queryset(request) # Apply pagination paginator = Paginator(queryset, self._get_items_per_page(request)) page_number = request.GET.get('page', 1) try: page_obj = paginator.get_page(page_number) except (PageNotAnInteger, EmptyPage): page_obj = paginator.get_page(1) # Build context for AJAX context = self._build_context(request, page_obj, paginator) # Return JSON response return JsonResponse({ 'success': True, 'data': { 'roles': [self._serialize_role(role) for role in context['roles']], 'pagination': { 'current_page': context['current_page'], 'total_pages': context['total_pages'], 'total_results': context['total_results'], 'has_previous': page_obj.has_previous(), 'has_next': page_obj.has_next(), }, 'stats': context['stats'], } }) except Exception as e: return JsonResponse({ 'success': False, 'error': str(e) }, status=500) def _serialize_role(self, role): """ Serialize role object for AJAX response. Args: role: Role object Returns: dict: Serialized role data """ return { 'id': role.id, 'name': role.name, 'description': role.description, 'is_active': role.is_active, 'created_at': role.created_at.isoformat() if role.created_at else None, 'user_count': getattr(role, 'user_count', 0), 'permission_count': role.permissions.count(), } class RoleCreateView(LoginRequiredMixin, PermissionRequiredMixin, View): """ Create a new Role with comprehensive form handling. """ http_method_names = ["get", "post"] template_name = "accounts/roles/form.html" permission_required = 'accounts.add_role' model = Role def get(self, request: HttpRequest) -> HttpResponse: """Render the role creation form.""" # Get available permissions for assignment available_permissions = self._get_available_permissions() permission_groups = self._get_permission_groups() existing_roles = self._get_existing_roles() priority_choices = self._get_priority_choices() role_type_choices = self._get_role_type_choices() context = { 'form_title': 'Create New Role', 'submit_text': 'Create Role', 'is_create': True, 'available_permissions': available_permissions, 'permission_groups': permission_groups, 'existing_roles': existing_roles, 'priority_choices': priority_choices, 'role_type_choices': role_type_choices, 'title': 'Create Role - Focus', 'object': None, # Changed from EmptyRole() to None 'active_menu': 'roles', } return render(request, self.template_name, context) def post(self, request: HttpRequest) -> HttpResponse: """Handle role creation.""" try: form_data = self._extract_form_data(request) validation_errors = self._validate_form_data(form_data, request) if validation_errors: return self._render_form_with_errors(request, validation_errors, form_data) role = self._create_role(form_data, request) if form_data['permissions']: self._assign_permissions(role, form_data['permissions']) if form_data['parent_role_id']: self._set_parent_role(role, form_data['parent_role_id']) self._log_activity(request, role, 'create') messages.success(request, f'Role "{role.name}" has been created successfully!') return redirect('accounts:role_detail', pk=role.pk) except ValidationError as e: messages.error(request, f'Validation error: {str(e)}') return self.get(request) except Exception as e: messages.error(request, f'An unexpected error occurred: {str(e)}') return self.get(request) def _extract_form_data(self, request: HttpRequest) -> dict: """Extract form data matching your model fields.""" return { # Basic information 'name': request.POST.get('name', '').strip(), 'description': request.POST.get('description', '').strip(), 'code': request.POST.get('code', '').strip(), # Role configuration - UPDATED TO MATCH YOUR MODEL 'is_active': request.POST.get('is_active') == 'on', 'level': request.POST.get('priority', '').strip(), # Your model uses 'level' not 'priority' 'role_type': request.POST.get('role_type', 'custom'), 'is_default': request.POST.get('is_default') == 'on', # Permissions and hierarchy 'permissions': request.POST.getlist('permissions'), 'parent_role_id': request.POST.get('parent_role', '').strip(), } def _validate_form_data(self, form_data: dict, request: HttpRequest) -> list: """Validate form data against your actual model.""" errors = [] # Required field validation if not form_data['name']: errors.append('Role name is required.') elif len(form_data['name']) > 100: errors.append('Role name cannot exceed 100 characters.') # Validate role code if not form_data['code']: errors.append('Role code is required.') elif len(form_data['code']) > 50: errors.append('Role code cannot exceed 50 characters.') elif not self._is_valid_role_code(form_data['code']): errors.append('Role code must be lowercase with underscores only.') # Description validation if form_data['description'] and len(form_data['description']) > 500: errors.append('Description cannot exceed 500 characters.') # Validate level (priority) - YOUR MODEL USES 'level' if form_data['level']: try: level = int(form_data['level']) if level < 1 or level > 5: # Based on your help text: 1=highest, 5=lowest errors.append('Level must be between 1 and 5.') except ValueError: errors.append('Level must be a valid number.') # Validate role type valid_role_types = [choice[0] for choice in Role.ROLE_TYPES] if form_data['role_type'] not in valid_role_types: errors.append('Invalid role type selected.') # Validate parent role if form_data['parent_role_id']: try: parent_role = Role.objects.get(id=form_data['parent_role_id']) if not parent_role.is_active: errors.append('Parent role must be active.') except Role.DoesNotExist: errors.append('Selected parent role does not exist.') # Validate permissions if form_data['permissions']: invalid_permissions = self._validate_permissions(form_data['permissions']) if invalid_permissions: errors.append(f'Invalid permissions: {", ".join(invalid_permissions)}') # Check for duplicate role name if form_data['name']: existing = Role.objects.filter( name__iexact=form_data['name'] ).exclude(is_deleted=True).exists() if existing: errors.append('A role with this name already exists.') # Check for duplicate role code if form_data['code']: existing = Role.objects.filter( code__iexact=form_data['code'] ).exclude(is_deleted=True).exists() if existing: errors.append('A role with this code already exists.') return errors def _create_role(self, form_data: dict, request: HttpRequest) -> Role: """Create role with your actual model fields.""" # Prepare role data matching YOUR model role_data = { 'name': form_data['name'], 'code': form_data['code'], 'is_active': form_data['is_active'], 'role_type': form_data['role_type'], 'is_default': form_data['is_default'], 'created_by': request.user, # From BaseModel } # Add optional description if form_data['description']: role_data['description'] = form_data['description'] # Handle level (priority in your form, but 'level' in model) if form_data['level']: role_data['level'] = int(form_data['level']) else: role_data['level'] = 1 # Default to highest level # Create the role role = Role.objects.create(**role_data) return role def _set_parent_role(self, role: Role, parent_role_id: str): """Set parent role - this field exists in your model.""" try: parent_role = Role.objects.get(id=parent_role_id) # Check for circular reference if self._would_create_circular_reference(role, parent_role): raise ValidationError('Setting this parent role would create a circular reference.') role.parent_role = parent_role role.save() except Role.DoesNotExist: raise ValidationError('Parent role does not exist.') def _assign_permissions(self, role: Role, permission_ids: list): """Assign permissions - this relationship exists in your model.""" from django.contrib.auth.models import Permission try: permissions = Permission.objects.filter(id__in=permission_ids) role.permissions.set(permissions) except Exception as e: raise ValidationError(f'Error assigning permissions: {str(e)}') def _validate_permissions(self, permission_ids: list) -> list: """Validate permission IDs.""" from django.contrib.auth.models import Permission try: valid_ids = set( Permission.objects.filter(id__in=permission_ids).values_list('id', flat=True) ) provided_ids = set(int(pid) for pid in permission_ids) invalid_ids = provided_ids - valid_ids return [str(pid) for pid in invalid_ids] except (ValueError, TypeError): return permission_ids def _would_create_circular_reference(self, role: Role, parent_role: Role) -> bool: """Check for circular reference in hierarchy.""" current = parent_role while current.parent_role: if current.parent_role.id == role.id: return True current = current.parent_role return False def _generate_role_code(self, name: str) -> str: """Generate role code matching your model's validator.""" import re # Convert to lowercase and replace spaces with underscores # Remove any characters that aren't lowercase letters or underscores code = re.sub(r'[^a-z_]', '', name.lower().replace(' ', '_').replace('-', '_')) # Ensure uniqueness base_code = code counter = 1 while Role.objects.filter(code=code).exclude(is_deleted=True).exists(): code = f"{base_code}_{counter}" counter += 1 return code[:50] def _render_form_with_errors(self, request: HttpRequest, errors: list, form_data: dict) -> HttpResponse: """Render form with errors.""" for error in errors: messages.error(request, error) context = { 'form_title': 'Create New Role', 'submit_text': 'Create Role', 'is_create': True, 'available_permissions': self._get_available_permissions(), 'permission_groups': self._get_permission_groups(), 'existing_roles': self._get_existing_roles(), 'priority_choices': self._get_priority_choices(), 'role_type_choices': self._get_role_type_choices(), 'form_data': form_data, 'object': None, 'title': 'Create Role - Focus', 'active_menu': 'roles', } return render(request, self.template_name, context) def _log_activity(self, request: HttpRequest, role: Role, action: str): """Log activity.""" try: Activity.log_activity( user=request.user, action=f"{action}_role_cbv", description=f"User {request.user.email} {action}d role: {role.name}", request=request, ip_address=get_client_ip(request), user_agent=request.META.get("HTTP_USER_AGENT", ""), metadata={ 'role_id': str(role.id), 'role_name': role.name, 'role_code': role.code, 'action': action } ) except Exception as e: print(f"Failed to log activity: {e}") # Helper methods def _get_available_permissions(self): """Get available permissions.""" from django.contrib.auth.models import Permission try: return Permission.objects.select_related('content_type').order_by( 'content_type__app_label', 'content_type__model', 'codename' ) except Exception: return [] def _get_permission_groups(self): """Get permissions grouped by app.""" from django.contrib.auth.models import Permission from collections import defaultdict try: permissions = Permission.objects.select_related('content_type').order_by( 'content_type__app_label', 'content_type__model', 'codename' ) groups = defaultdict(list) for perm in permissions: app_label = perm.content_type.app_label groups[app_label].append(perm) return dict(groups) except Exception: return {} def _get_existing_roles(self): """Get existing roles for parent selection.""" try: return Role.objects.filter( is_active=True ).exclude(is_deleted=True).order_by('name') except Exception: return [] def _get_priority_choices(self): """Get priority choices - Updated for your model's 'level' field.""" return [ ('', 'Select level...'), ('1', 'Level 1 (Highest)'), ('2', 'Level 2 (High)'), ('3', 'Level 3 (Medium)'), ('4', 'Level 4 (Low)'), ('5', 'Level 5 (Lowest)'), ] def _get_role_type_choices(self): """Get role type choices from your model.""" return [('', 'Select type...')] + list(Role.ROLE_TYPES) def _is_valid_role_code(self, code: str) -> bool: """Validate role code format matching your model's validator.""" import re pattern = r'^[a-z_]+$' return re.match(pattern, code) is not None class RoleDetailView(LoginRequiredMixin, PermissionRequiredMixin, View): """ Display detailed role information with comprehensive data handling. """ http_method_names = ["get"] template_name = "accounts/roles/detail.html" permission_required = 'accounts.view_role' model = Role def get(self, request: HttpRequest, pk: int) -> HttpResponse: """Display role details with comprehensive information.""" try: role = self._get_role_object(pk) if not role: messages.error(request, 'Role not found.') return redirect('accounts:role_list') # Get comprehensive role data role_stats = self._get_role_statistics(role) user_data = self._get_user_data(role) permission_data = self._get_permission_data(role) hierarchy_data = self._get_hierarchy_data(role) activity_data = self._get_activity_data(role, request) related_data = self._get_related_data(role) context = { 'role': role, 'object': role, # For consistency with DetailView 'title': f'Role: {role.name} - Focus', 'active_menu': 'roles', # Role statistics **role_stats, # User data **user_data, # Permission data **permission_data, # Hierarchy data **hierarchy_data, # Activity data **activity_data, # Related data **related_data, # Additional context 'can_edit': self._user_can_edit_role(request.user, role), 'can_delete': self._user_can_delete_role(request.user, role), 'breadcrumbs': self._get_breadcrumbs(role), } # Log view activity self._log_activity(request, role, 'view') return render(request, self.template_name, context) except Exception as e: messages.error(request, f'An error occurred while loading the role: {str(e)}') return redirect('accounts:role_list') def _get_role_object(self, pk: int) -> Role: """Get role object with error handling.""" try: return Role.objects.select_related( 'created_by', 'updated_by', 'parent_role' ).prefetch_related( 'permissions__content_type', 'child_roles', 'userrole_set__user__profile' ).get(pk=pk, is_deleted=False) except Role.DoesNotExist: return None except Exception as e: print(f"Error fetching role: {e}") return None def _get_role_statistics(self, role: Role) -> dict: """Get comprehensive role statistics.""" try: # User counts via UserRole relationship total_user_count = ( UserRole.objects.filter(role=role) .values('user') .distinct() .count() ) active_user_count = ( UserRole.objects.filter(role=role, is_active=True) .values('user') .distinct() .count() ) inactive_user_count = total_user_count - active_user_count # Permission counts permission_count = role.permissions.count() # Hierarchy counts child_role_count = role.child_roles.filter(is_deleted=False).count() parent_role_count = 1 if role.parent_role else 0 return { 'user_count': total_user_count, 'active_user_count': active_user_count, 'inactive_user_count': inactive_user_count, 'permission_count': permission_count, 'child_role_count': child_role_count, 'parent_role_count': parent_role_count, 'hierarchy_depth': self._calculate_hierarchy_depth(role), } except Exception as e: print(f"Error calculating role statistics: {e}") return { 'user_count': 0, 'active_user_count': 0, 'inactive_user_count': 0, 'permission_count': 0, 'child_role_count': 0, 'parent_role_count': 0, 'hierarchy_depth': 0, } def _get_user_data(self, role: Role) -> dict: """Get users assigned to this role.""" try: # Get assigned user IDs (active assignments) assigned_user_ids = ( UserRole.objects.filter(role=role, is_active=True) .values_list('user_id', flat=True) .distinct() ) # Get user objects with profiles assigned_users = User.objects.filter( id__in=assigned_user_ids ).select_related('profile').order_by('first_name', 'last_name') # Get recent assignments (last 30 days) from datetime import datetime, timedelta recent_date = datetime.now() - timedelta(days=30) recent_assignments = ( UserRole.objects.filter( role=role, created_at__gte=recent_date ).select_related('user__profile', 'assigned_by') .order_by('-created_at')[:10] ) return { 'assigned_users': assigned_users, 'recent_assignments': recent_assignments, } except Exception as e: print(f"Error getting user data: {e}") return { 'assigned_users': [], 'recent_assignments': [], } def _get_permission_data(self, role: Role) -> dict: """Get permission data grouped by content type.""" try: from collections import defaultdict permissions = role.permissions.select_related('content_type').order_by( 'content_type__app_label', 'content_type__model', 'codename' ) # Group permissions by app and model permission_groups = defaultdict(lambda: defaultdict(list)) for perm in permissions: app_label = perm.content_type.app_label model_name = perm.content_type.model permission_groups[app_label][model_name].append(perm) # Convert to regular dict for template permission_groups = { app: dict(models) for app, models in permission_groups.items() } # Get permission categories permission_categories = self._categorize_permissions(permissions) return { 'permissions': permissions, 'permission_groups': permission_groups, 'permission_categories': permission_categories, } except Exception as e: print(f"Error getting permission data: {e}") return { 'permissions': [], 'permission_groups': {}, 'permission_categories': {}, } def _get_hierarchy_data(self, role: Role) -> dict: """Get role hierarchy information.""" try: # Get parent chain parent_chain = [] current = role.parent_role while current and len(parent_chain) < 10: # Prevent infinite loops parent_chain.append(current) current = current.parent_role # Get child roles child_roles = role.child_roles.filter(is_deleted=False).order_by('name') # Get sibling roles (same parent) sibling_roles = [] if role.parent_role: sibling_roles = role.parent_role.child_roles.filter( is_deleted=False ).exclude(id=role.id).order_by('name') return { 'parent_chain': parent_chain, 'child_roles': child_roles, 'sibling_roles': sibling_roles, 'hierarchy_level': len(parent_chain), } except Exception as e: print(f"Error getting hierarchy data: {e}") return { 'parent_chain': [], 'child_roles': [], 'sibling_roles': [], 'hierarchy_level': 0, } def _get_activity_data(self, role: Role, request: HttpRequest) -> dict: """Get recent activity related to this role.""" try: # Get recent activities if Activity model is available recent_activities = [] try: from django.contrib.contenttypes.models import ContentType role_ct = ContentType.objects.get_for_model(Role) recent_activities = Activity.objects.filter( content_type=role_ct, object_id=role.id ).select_related('user').order_by('-created_at')[:10] except: pass # Activity model might not be available return { 'recent_activities': recent_activities, } except Exception as e: print(f"Error getting activity data: {e}") return { 'recent_activities': [], } def _get_related_data(self, role: Role) -> dict: """Get related data for the role.""" try: # Get roles with similar permissions similar_roles = [] if role.permissions.exists(): role_permissions = set(role.permissions.values_list('id', flat=True)) similar_roles = Role.objects.filter( is_deleted=False, is_active=True ).exclude(id=role.id).prefetch_related('permissions') # Calculate similarity score role_similarities = [] for other_role in similar_roles: other_permissions = set(other_role.permissions.values_list('id', flat=True)) if other_permissions: intersection = len(role_permissions & other_permissions) union = len(role_permissions | other_permissions) similarity = intersection / union if union > 0 else 0 if similarity > 0.3: # Only include roles with >30% similarity role_similarities.append({ 'role': other_role, 'similarity': similarity, 'common_permissions': intersection }) # Sort by similarity similar_roles = sorted(role_similarities, key=lambda x: x['similarity'], reverse=True)[:5] return { 'similar_roles': similar_roles, } except Exception as e: print(f"Error getting related data: {e}") return { 'similar_roles': [], } def _calculate_hierarchy_depth(self, role: Role) -> int: """Calculate the depth of the role in hierarchy.""" depth = 0 current = role while current.parent_role and depth < 10: # Prevent infinite loops depth += 1 current = current.parent_role return depth def _categorize_permissions(self, permissions) -> dict: """Categorize permissions by type.""" categories = { 'admin': [], 'content': [], 'user': [], 'system': [], 'other': [] } for perm in permissions: perm_name = perm.name.lower() codename = perm.codename.lower() if any(word in perm_name or word in codename for word in ['admin', 'superuser', 'staff']): categories['admin'].append(perm) elif any(word in perm_name or word in codename for word in ['content', 'post', 'article', 'page']): categories['content'].append(perm) elif any(word in perm_name or word in codename for word in ['user', 'profile', 'account']): categories['user'].append(perm) elif any(word in perm_name or word in codename for word in ['system', 'setting', 'config']): categories['system'].append(perm) else: categories['other'].append(perm) return categories def _user_can_edit_role(self, user, role: Role) -> bool: """Check if user can edit this role.""" try: # Check basic permission if not user.has_perm('accounts.change_role'): return False # Additional business logic # Users can't edit roles higher than their own level if hasattr(user, 'get_highest_role_level'): user_level = user.get_highest_role_level() if role.level and user_level and role.level < user_level: return False return True except Exception: return False def _user_can_delete_role(self, user, role: Role) -> bool: """Check if user can delete this role.""" try: # Check basic permission if not user.has_perm('accounts.delete_role'): return False # Can't delete system roles if role.role_type == 'system': return False # Can't delete default roles if role.is_default: return False # Can't delete roles with assigned users if UserRole.objects.filter(role=role, is_active=True).exists(): return False return True except Exception: return False def _get_breadcrumbs(self, role: Role) -> list: """Generate breadcrumbs for the role detail page.""" breadcrumbs = [ {'title': 'Dashboard', 'url': 'core:dashboard'}, {'title': 'Roles', 'url': 'accounts:role_list'}, {'title': role.name, 'url': None, 'active': True}, ] return breadcrumbs def _log_activity(self, request: HttpRequest, role: Role, action: str): """Log activity - same pattern as RoleCreateView.""" try: Activity.log_activity( user=request.user, action=f"{action}_role_cbv", description=f"User {request.user.email} viewed role: {role.name}", request=request, ip_address=get_client_ip(request), user_agent=request.META.get("HTTP_USER_AGENT", ""), metadata={ 'role_id': str(role.id), 'role_name': role.name, 'role_code': role.code, 'action': action } ) except Exception as e: print(f"Failed to log activity: {e}") class RoleUpdateView(LoginRequiredMixin, PermissionRequiredMixin, View): """ Update an existing Role with comprehensive form handling. """ http_method_names = ["get", "post"] template_name = "accounts/roles/form.html" permission_required = 'accounts.change_role' model = Role def get(self, request: HttpRequest, pk: int) -> HttpResponse: """Render the role update form.""" try: role = self._get_role_object(pk) if not role: messages.error(request, 'Role not found.') return redirect('accounts:role_list') # Check if user can edit this role if not self._user_can_edit_role(request.user, role): messages.error(request, 'You do not have permission to edit this role.') return redirect('accounts:role_detail', pk=role.pk) # Get form data available_permissions = self._get_available_permissions() permission_groups = self._get_permission_groups() existing_roles = self._get_existing_roles(exclude_id=role.id) priority_choices = self._get_priority_choices() role_type_choices = self._get_role_type_choices() # Prepare form data from existing role form_data = self._prepare_form_data_from_role(role) context = { 'form_title': f'Edit Role: {role.name}', 'submit_text': 'Update Role', 'is_create': False, 'available_permissions': available_permissions, 'permission_groups': permission_groups, 'existing_roles': existing_roles, 'priority_choices': priority_choices, 'role_type_choices': role_type_choices, 'title': f'Edit Role: {role.name} - Focus', 'object': role, 'role': role, 'form_data': form_data, 'active_menu': 'roles', 'cancel_url': reverse('accounts:role_detail', kwargs={'pk': role.pk}), } return render(request, self.template_name, context) except Exception as e: messages.error(request, f'An error occurred while loading the role: {str(e)}') return redirect('accounts:role_list') def post(self, request: HttpRequest, pk: int) -> HttpResponse: """Handle role update.""" try: role = self._get_role_object(pk) if not role: messages.error(request, 'Role not found.') return redirect('accounts:role_list') # Check permissions if not self._user_can_edit_role(request.user, role): messages.error(request, 'You do not have permission to edit this role.') return redirect('accounts:role_detail', pk=role.pk) # Store original data for comparison original_data = self._get_role_original_data(role) form_data = self._extract_form_data(request) validation_errors = self._validate_form_data(form_data, request, role_id=role.id) if validation_errors: return self._render_form_with_errors(request, validation_errors, form_data, role) # Update the role updated_role = self._update_role(role, form_data, request) # Update permissions if changed if form_data['permissions'] != list(role.permissions.values_list('id', flat=True)): self._assign_permissions(updated_role, form_data['permissions']) # Update parent role if changed current_parent_id = role.parent_role.id if role.parent_role else None new_parent_id = int(form_data['parent_role_id']) if form_data['parent_role_id'] else None if current_parent_id != new_parent_id: self._set_parent_role(updated_role, form_data['parent_role_id']) # Log the changes changes = self._get_role_changes(original_data, form_data) self._log_activity(request, updated_role, 'update', changes) messages.success(request, f'Role "{updated_role.name}" has been updated successfully!') return redirect('accounts:role_detail', pk=updated_role.pk) except ValidationError as e: messages.error(request, f'Validation error: {str(e)}') return self.get(request, pk) except Exception as e: messages.error(request, f'An unexpected error occurred: {str(e)}') return self.get(request, pk) def _get_role_object(self, pk: int) -> Role: """Get role object with error handling.""" try: return Role.objects.select_related( 'created_by', 'updated_by', 'parent_role' ).prefetch_related( 'permissions', 'child_roles' ).get(pk=pk, is_deleted=False) except Role.DoesNotExist: return None except Exception as e: print(f"Error fetching role: {e}") return None def _prepare_form_data_from_role(self, role: Role) -> dict: """Prepare form data from existing role.""" return { 'name': role.name, 'description': role.description or '', 'code': role.code, 'is_active': role.is_active, 'level': str(role.level) if role.level else '', 'role_type': role.role_type, 'is_default': role.is_default, 'permissions': list(role.permissions.values_list('id', flat=True)), 'parent_role_id': str(role.parent_role.id) if role.parent_role else '', } def _get_role_original_data(self, role: Role) -> dict: """Get original role data for change tracking.""" return { 'name': role.name, 'description': role.description, 'code': role.code, 'is_active': role.is_active, 'level': role.level, 'role_type': role.role_type, 'is_default': role.is_default, 'permissions': set(role.permissions.values_list('id', flat=True)), 'parent_role_id': role.parent_role.id if role.parent_role else None, } def _extract_form_data(self, request: HttpRequest) -> dict: """Extract form data - same as create view.""" return { 'name': request.POST.get('name', '').strip(), 'description': request.POST.get('description', '').strip(), 'code': request.POST.get('code', '').strip(), 'is_active': request.POST.get('is_active') == 'on', 'level': request.POST.get('priority', '').strip(), 'role_type': request.POST.get('role_type', 'custom'), 'is_default': request.POST.get('is_default') == 'on', 'permissions': request.POST.getlist('permissions'), 'parent_role_id': request.POST.get('parent_role', '').strip(), } def _validate_form_data(self, form_data: dict, request: HttpRequest, role_id: int) -> list: """Validate form data - similar to create but excludes current role.""" errors = [] # Required field validation if not form_data['name']: errors.append('Role name is required.') elif len(form_data['name']) > 100: errors.append('Role name cannot exceed 100 characters.') # Validate role code if not form_data['code']: errors.append('Role code is required.') elif len(form_data['code']) > 50: errors.append('Role code cannot exceed 50 characters.') elif not self._is_valid_role_code(form_data['code']): errors.append('Role code must be lowercase with underscores only.') # Description validation if form_data['description'] and len(form_data['description']) > 500: errors.append('Description cannot exceed 500 characters.') # Validate level if form_data['level']: try: level = int(form_data['level']) if level < 1 or level > 5: errors.append('Level must be between 1 and 5.') except ValueError: errors.append('Level must be a valid number.') # Validate role type valid_role_types = [choice[0] for choice in Role.ROLE_TYPES] if form_data['role_type'] not in valid_role_types: errors.append('Invalid role type selected.') # Validate parent role if form_data['parent_role_id']: try: parent_role = Role.objects.get(id=form_data['parent_role_id']) if not parent_role.is_active: errors.append('Parent role must be active.') # Check for circular reference if self._would_create_circular_reference_update(role_id, parent_role): errors.append('Setting this parent role would create a circular reference.') except Role.DoesNotExist: errors.append('Selected parent role does not exist.') # Validate permissions if form_data['permissions']: invalid_permissions = self._validate_permissions(form_data['permissions']) if invalid_permissions: errors.append(f'Invalid permissions: {", ".join(invalid_permissions)}') # Check for duplicate role name (exclude current role) if form_data['name']: existing = Role.objects.filter( name__iexact=form_data['name'] ).exclude(id=role_id).exclude(is_deleted=True).exists() if existing: errors.append('A role with this name already exists.') # Check for duplicate role code (exclude current role) if form_data['code']: existing = Role.objects.filter( code__iexact=form_data['code'] ).exclude(id=role_id).exclude(is_deleted=True).exists() if existing: errors.append('A role with this code already exists.') return errors def _update_role(self, role: Role, form_data: dict, request: HttpRequest) -> Role: """Update role with form data.""" # Update basic fields role.name = form_data['name'] role.code = form_data['code'] role.is_active = form_data['is_active'] role.role_type = form_data['role_type'] role.is_default = form_data['is_default'] role.updated_by = request.user # Update optional fields role.description = form_data['description'] if form_data['description'] else None # Handle level if form_data['level']: role.level = int(form_data['level']) else: role.level = 1 role.save() return role def _get_role_changes(self, original_data: dict, form_data: dict) -> dict: """Track changes made to the role.""" changes = {} # Compare basic fields if original_data['name'] != form_data['name']: changes['name'] = {'old': original_data['name'], 'new': form_data['name']} if original_data['code'] != form_data['code']: changes['code'] = {'old': original_data['code'], 'new': form_data['code']} if original_data['description'] != form_data['description']: changes['description'] = {'old': original_data['description'], 'new': form_data['description']} if original_data['is_active'] != form_data['is_active']: changes['is_active'] = {'old': original_data['is_active'], 'new': form_data['is_active']} level_new = int(form_data['level']) if form_data['level'] else 1 if original_data['level'] != level_new: changes['level'] = {'old': original_data['level'], 'new': level_new} if original_data['role_type'] != form_data['role_type']: changes['role_type'] = {'old': original_data['role_type'], 'new': form_data['role_type']} if original_data['is_default'] != form_data['is_default']: changes['is_default'] = {'old': original_data['is_default'], 'new': form_data['is_default']} # Compare permissions new_permissions = set(int(p) for p in form_data['permissions']) if original_data['permissions'] != new_permissions: changes['permissions'] = { 'old': list(original_data['permissions']), 'new': list(new_permissions) } # Compare parent role parent_new = int(form_data['parent_role_id']) if form_data['parent_role_id'] else None if original_data['parent_role_id'] != parent_new: changes['parent_role_id'] = {'old': original_data['parent_role_id'], 'new': parent_new} return changes def _would_create_circular_reference_update(self, role_id: int, parent_role: Role) -> bool: """Check for circular reference when updating.""" current = parent_role while current.parent_role: if current.parent_role.id == role_id: return True current = current.parent_role return False # Reuse helper methods from create view def _assign_permissions(self, role: Role, permission_ids: list): """Assign permissions - same as create view.""" from django.contrib.auth.models import Permission try: permissions = Permission.objects.filter(id__in=permission_ids) role.permissions.set(permissions) except Exception as e: raise ValidationError(f'Error assigning permissions: {str(e)}') def _set_parent_role(self, role: Role, parent_role_id: str): """Set parent role - same as create view.""" try: if parent_role_id: parent_role = Role.objects.get(id=parent_role_id) role.parent_role = parent_role else: role.parent_role = None role.save() except Role.DoesNotExist: raise ValidationError('Parent role does not exist.') def _validate_permissions(self, permission_ids: list) -> list: """Validate permission IDs - same as create view.""" from django.contrib.auth.models import Permission try: valid_ids = set( Permission.objects.filter(id__in=permission_ids).values_list('id', flat=True) ) provided_ids = set(int(pid) for pid in permission_ids) invalid_ids = provided_ids - valid_ids return [str(pid) for pid in invalid_ids] except (ValueError, TypeError): return permission_ids def _user_can_edit_role(self, user, role: Role) -> bool: """Check if user can edit this role - same logic as detail view.""" try: if not user.has_perm('accounts.change_role'): return False if hasattr(user, 'get_highest_role_level'): user_level = user.get_highest_role_level() if role.level and user_level and role.level < user_level: return False return True except Exception: return False def _render_form_with_errors(self, request: HttpRequest, errors: list, form_data: dict, role: Role) -> HttpResponse: """Render form with errors.""" for error in errors: messages.error(request, error) context = { 'form_title': f'Edit Role: {role.name}', 'submit_text': 'Update Role', 'is_create': False, 'available_permissions': self._get_available_permissions(), 'permission_groups': self._get_permission_groups(), 'existing_roles': self._get_existing_roles(exclude_id=role.id), 'priority_choices': self._get_priority_choices(), 'role_type_choices': self._get_role_type_choices(), 'form_data': form_data, 'object': role, 'role': role, 'title': f'Edit Role: {role.name} - Focus', 'active_menu': 'roles', 'cancel_url': reverse('accounts:role_detail', kwargs={'pk': role.pk}), } return render(request, self.template_name, context) def _log_activity(self, request: HttpRequest, role: Role, action: str, changes: dict = None): """Log activity with changes.""" try: metadata = { 'role_id': str(role.id), 'role_name': role.name, 'role_code': role.code, 'action': action } if changes: metadata['changes'] = changes Activity.log_activity( user=request.user, action=f"{action}_role_cbv", description=f"User {request.user.email} {action}d role: {role.name}", request=request, ip_address=get_client_ip(request), user_agent=request.META.get("HTTP_USER_AGENT", ""), metadata=metadata ) except Exception as e: print(f"Failed to log activity: {e}") # Helper methods - reuse from create view def _get_available_permissions(self): """Get available permissions.""" from django.contrib.auth.models import Permission try: return Permission.objects.select_related('content_type').order_by( 'content_type__app_label', 'content_type__model', 'codename' ) except Exception: return [] def _get_permission_groups(self): """Get permissions grouped by app.""" from django.contrib.auth.models import Permission from collections import defaultdict try: permissions = Permission.objects.select_related('content_type').order_by( 'content_type__app_label', 'content_type__model', 'codename' ) groups = defaultdict(list) for perm in permissions: app_label = perm.content_type.app_label groups[app_label].append(perm) return dict(groups) except Exception: return {} def _get_existing_roles(self, exclude_id=None): """Get existing roles for parent selection.""" try: queryset = Role.objects.filter( is_active=True ).exclude(is_deleted=True) if exclude_id: queryset = queryset.exclude(id=exclude_id) return queryset.order_by('name') except Exception: return [] def _get_priority_choices(self): """Get priority choices.""" return [ ('', 'Select level...'), ('1', 'Level 1 (Highest)'), ('2', 'Level 2 (High)'), ('3', 'Level 3 (Medium)'), ('4', 'Level 4 (Low)'), ('5', 'Level 5 (Lowest)'), ] def _get_role_type_choices(self): """Get role type choices from model.""" return [('', 'Select type...')] + list(Role.ROLE_TYPES) def _is_valid_role_code(self, code: str) -> bool: """Validate role code format.""" import re pattern = r'^[a-z_]+$' return re.match(pattern, code) is not None class RoleDeleteView(LoginRequiredMixin, PermissionRequiredMixin, View): """ Delete a role with comprehensive validation and confirmation. """ http_method_names = ["get", "post"] template_name = "accounts/roles/confirm_delete.html" permission_required = 'accounts.delete_role' model = Role def get(self, request: HttpRequest, pk: int) -> HttpResponse: """Show delete confirmation page.""" try: role = self._get_role_object(pk) if not role: messages.error(request, 'Role not found.') return redirect('accounts:role_list') # Check if user can delete this role can_delete, error_message = self._user_can_delete_role(request.user, role) if not can_delete: messages.error(request, error_message) return redirect('accounts:role_detail', pk=role.pk) # Get deletion impact data deletion_impact = self._get_deletion_impact(role) context = { 'role': role, 'object': role, 'title': f'Delete Role: {role.name} - Focus', 'active_menu': 'roles', 'deletion_impact': deletion_impact, 'cancel_url': reverse('accounts:role_detail', kwargs={'pk': role.pk}), } return render(request, self.template_name, context) except Exception as e: messages.error(request, f'An error occurred: {str(e)}') return redirect('accounts:role_list') def post(self, request: HttpRequest, pk: int) -> HttpResponse: """Handle role deletion.""" try: role = self._get_role_object(pk) if not role: messages.error(request, 'Role not found.') return redirect('accounts:role_list') # Check permissions again can_delete, error_message = self._user_can_delete_role(request.user, role) if not can_delete: messages.error(request, error_message) return redirect('accounts:role_detail', pk=role.pk) # Get role data for logging before deletion role_data = { 'id': role.id, 'name': role.name, 'code': role.code, 'user_count': self._get_assigned_user_count(role), 'permission_count': role.permissions.count(), 'child_role_count': role.child_roles.filter(is_deleted=False).count() } # Handle deletion strategy deletion_strategy = request.POST.get('deletion_strategy', 'soft_delete') if deletion_strategy == 'hard_delete': self._hard_delete_role(role, request) else: self._soft_delete_role(role, request) # Log the deletion self._log_activity(request, role_data, 'delete', deletion_strategy) messages.success( request, f'Role "{role_data["name"]}" has been deleted successfully.' ) return redirect('accounts:role_list') except Exception as e: messages.error(request, f'An error occurred while deleting the role: {str(e)}') return redirect('accounts:role_detail', pk=pk) def _get_role_object(self, pk: int) -> Role: """Get role object with related data.""" try: return Role.objects.select_related( 'created_by', 'updated_by', 'parent_role' ).prefetch_related( 'permissions', 'child_roles', 'userrole_set__user' ).get(pk=pk, is_deleted=False) except Role.DoesNotExist: return None except Exception as e: print(f"Error fetching role: {e}") return None def _user_can_delete_role(self, user, role: Role) -> tuple: """Check if user can delete this role with detailed error message.""" try: # Check basic permission if not user.has_perm('accounts.delete_role'): return False, 'You do not have permission to delete roles.' # Can't delete system roles if role.role_type == 'system': return False, 'System roles cannot be deleted.' # Can't delete default roles if role.is_default: return False, 'Default roles cannot be deleted.' # Check for assigned users assigned_count = self._get_assigned_user_count(role) if assigned_count > 0: return False, f'Cannot delete role with {assigned_count} assigned user(s). Please reassign users first.' # Check for child roles child_count = role.child_roles.filter(is_deleted=False).count() if child_count > 0: return False, f'Cannot delete role with {child_count} child role(s). Please reassign or delete child roles first.' # Check user level permissions if hasattr(user, 'get_highest_role_level'): user_level = user.get_highest_role_level() if role.level and user_level and role.level < user_level: return False, 'You cannot delete roles with higher level than yours.' return True, '' except Exception as e: print(f"Error checking delete permissions: {e}") return False, 'An error occurred while checking permissions.' def _get_deletion_impact(self, role: Role) -> dict: """Get information about what will be affected by deletion.""" try: assigned_user_count = self._get_assigned_user_count(role) child_roles = role.child_roles.filter(is_deleted=False) permission_count = role.permissions.count() # Get recent activities recent_activities = [] try: from django.contrib.contenttypes.models import ContentType role_ct = ContentType.objects.get_for_model(Role) recent_activities = Activity.objects.filter( content_type=role_ct, object_id=role.id ).select_related('user').order_by('-created_at')[:5] except: pass return { 'assigned_user_count': assigned_user_count, 'child_roles': child_roles, 'child_role_count': child_roles.count(), 'permission_count': permission_count, 'recent_activities': recent_activities, 'has_dependencies': assigned_user_count > 0 or child_roles.count() > 0 } except Exception as e: print(f"Error getting deletion impact: {e}") return { 'assigned_user_count': 0, 'child_roles': [], 'child_role_count': 0, 'permission_count': 0, 'recent_activities': [], 'has_dependencies': False } def _get_assigned_user_count(self, role: Role) -> int: """Get count of users assigned to this role.""" try: return UserRole.objects.filter( role=role, is_active=True ).values('user').distinct().count() except Exception: return 0 def _soft_delete_role(self, role: Role, request: HttpRequest): """Perform soft delete (mark as deleted).""" role.is_deleted = True role.is_active = False role.updated_by = request.user role.save() # Deactivate all user assignments UserRole.objects.filter(role=role, is_active=True).update( is_active=False, updated_by=request.user ) def _hard_delete_role(self, role: Role, request: HttpRequest): """Perform hard delete (actually delete from database).""" # Remove all user assignments first UserRole.objects.filter(role=role).delete() # Remove permission assignments role.permissions.clear() # Handle child roles - set their parent to None or to this role's parent if role.parent_role: role.child_roles.update(parent_role=role.parent_role) else: role.child_roles.update(parent_role=None) # Finally delete the role role.delete() def _log_activity(self, request: HttpRequest, role_data: dict, action: str, deletion_strategy: str): """Log deletion activity.""" try: Activity.log_activity( user=request.user, action=f"{action}_role_cbv", description=f"User {request.user.email} deleted role: {role_data['name']} (Strategy: {deletion_strategy})", request=request, ip_address=get_client_ip(request), user_agent=request.META.get("HTTP_USER_AGENT", ""), metadata={ 'role_id': str(role_data['id']), 'role_name': role_data['name'], 'role_code': role_data['code'], 'deletion_strategy': deletion_strategy, 'user_count': role_data['user_count'], 'permission_count': role_data['permission_count'], 'child_role_count': role_data['child_role_count'], 'action': action } ) except Exception as e: print(f"Failed to log activity: {e}") # ============================================================================ # Profile Management Views # ============================================================================ class ProfileListView(LoginRequiredMixin, PermissionRequiredMixin, ListView): """Display a list of user profiles.""" model = Profile template_name = 'accounts/profiles/list.html' context_object_name = 'profiles' paginate_by = 25 permission_required = 'accounts.view_profile' class ProfileDetailView(LoginRequiredMixin, DetailView): """Display the details of a user profile.""" model = Profile template_name = 'accounts/profiles/detail.html' context_object_name = 'profile' def get_object(self, queryset=None): """Ensure user can only view allowed profiles.""" obj = super().get_object(queryset) # Users can only view their own profile unless they have permission if not self.request.user.has_perm('accounts.view_profile'): if obj.user != self.request.user: raise PermissionDenied("You don't have permission to view this profile.") return obj class ProfileCreateView(LoginRequiredMixin, PermissionRequiredMixin, CreateView): """Create a new profile.""" model = Profile form_class = ProfileUpdateForm template_name = 'accounts/profiles/form.html' permission_required = 'accounts.add_profile' success_url = reverse_lazy('accounts:profile_list') class ProfileUpdateView(LoginRequiredMixin, UpdateView): """Update an existing profile.""" model = Profile form_class = ProfileUpdateForm template_name = 'accounts/profiles/form.html' def get_success_url(self): """Return to profile detail page after successful update.""" return reverse('accounts:profile_detail', kwargs={'pk': self.object.pk}) def get_object(self, queryset=None): """Ensure user can only edit allowed profiles.""" obj = super().get_object(queryset) # Users can only edit their own profile unless they have permission if not self.request.user.has_perm('accounts.change_profile'): if obj.user != self.request.user: raise PermissionDenied("You don't have permission to edit this profile.") return obj class ProfileDeleteView(LoginRequiredMixin, PermissionRequiredMixin, DeleteView): """Delete a profile.""" model = Profile template_name = 'accounts/profiles/confirm_delete.html' permission_required = 'accounts.delete_profile' success_url = reverse_lazy('accounts:profile_list') class CurrentUserProfileView(LoginRequiredMixin, TemplateView): """Display current user's profile.""" template_name = 'accounts/profiles/current_profile.html' def get_context_data(self, **kwargs): """Add current user's profile to context.""" context = super().get_context_data(**kwargs) # Get or create profile for current user profile, created = Profile.objects.get_or_create(user=self.request.user) context['profile'] = profile context['user_obj'] = self.request.user # Add statistics context['total_sessions'] = UserSession.objects.filter( user=self.request.user ).count() context['active_sessions'] = UserSession.objects.filter( user=self.request.user, expires_at__gt=timezone.now() ).count() return context class CurrentUserProfileUpdateView(LoginRequiredMixin, UpdateView): """Update current user's profile.""" model = Profile form_class = ProfileUpdateForm template_name = 'accounts/profiles/current_profile_edit.html' success_url = reverse_lazy('accounts:current_profile') def get_object(self, queryset=None): """Get or create profile for current user.""" profile, created = Profile.objects.get_or_create(user=self.request.user) return profile def form_valid(self, form): """Handle successful form submission.""" response = super().form_valid(form) messages.success( self.request, 'Your profile has been updated successfully.' ) logger.info(f"Profile updated by {self.request.user.username}") return response # ============================================================================ # Security and Authentication Views # ============================================================================ class UserSecurityView(LoginRequiredMixin, TemplateView): """Display user security dashboard.""" template_name = 'accounts/security/dashboard.html' def get_context_data(self, **kwargs): """Add security-related context.""" context = super().get_context_data(**kwargs) # Add recent security logs context['recent_logs'] = SecurityLog.objects.filter( user=self.request.user ).order_by('-timestamp')[:10] # Add active sessions context['active_sessions'] = UserSession.objects.filter( user=self.request.user, expires_at__gt=timezone.now() ).order_by('-created_at') # Add API keys context['api_keys'] = APIKey.objects.filter( user=self.request.user, is_active=True ).order_by('-created_at') return context class SecurityDashboardView(LoginRequiredMixin, PermissionRequiredMixin, TemplateView): """Security dashboard for administrators.""" template_name = 'accounts/security/admin_dashboard.html' permission_required = 'accounts.view_securitylog' def get_context_data(self, **kwargs): """Add security statistics and recent logs.""" context = super().get_context_data(**kwargs) # Add security statistics context['total_users'] = User.objects.count() context['active_users'] = User.objects.filter(is_active=True).count() context['recent_logins'] = User.objects.filter( last_login__gte=timezone.now() - timedelta(days=7) ).count() # Add recent security logs context['recent_logs'] = SecurityLog.objects.select_related('user').order_by('-timestamp')[:20] # Add failed login attempts context['failed_logins'] = SecurityLog.objects.filter( action='login_failed', timestamp__gte=timezone.now() - timedelta(hours=24) ).count() return context # ============================================================================ # Authentication Views # ============================================================================ @login_required def change_password(request): """Handle password change for logged-in users.""" if request.method == 'POST': form = PasswordChangeForm(request.user, request.POST) if form.is_valid(): user = form.save() update_session_auth_hash(request, user) # Important! # Log password change SecurityLog.objects.create( user=user, action='password_changed', ip_address=request.META.get('REMOTE_ADDR'), user_agent=request.META.get('HTTP_USER_AGENT', '')[:255], details="Password changed successfully" ) messages.success(request, 'Your password was successfully updated!') return redirect('accounts:current_profile') else: messages.error(request, 'Please correct the error below.') else: form = PasswordChangeForm(request.user) return render(request, 'accounts/auth/change_password.html', { 'form': form }) # ============================================================================ # AJAX and Utility Views # ============================================================================ @login_required @require_http_methods(["GET"]) def user_search_ajax(request): """AJAX endpoint for user search.""" query = request.GET.get('q', '').strip() if len(query) < 2: return JsonResponse({'users': []}) users = User.objects.filter( Q(username__icontains=query) | Q(first_name__icontains=query) | Q(last_name__icontains=query) | Q(email__icontains=query) ).filter(is_active=True)[:10] users_data = [] for user in users: users_data.append({ 'id': user.id, 'username': user.username, 'full_name': user.get_full_name(), 'email': user.email, }) return JsonResponse({'users': users_data}) @login_required @require_http_methods(["GET"]) def role_permissions_ajax(request, role_id): """AJAX endpoint to get role permissions.""" try: role = get_object_or_404(Group, id=role_id) permissions = role.permissions.all().values('id', 'name', 'codename') return JsonResponse({'permissions': list(permissions)}) except Exception as e: logger.error(f"Error fetching role permissions: {str(e)}") return JsonResponse({'error': 'Failed to fetch permissions'}, status=500) @login_required @require_http_methods(["POST"]) def bulk_user_action(request): """Handle bulk actions on users.""" action = request.POST.get('action') user_ids = request.POST.getlist('user_ids[]') if not action or not user_ids: return JsonResponse({'error': 'Missing action or user IDs'}, status=400) try: users = User.objects.filter(id__in=user_ids) if action == 'activate': users.update(is_active=True) message = f'{users.count()} users activated successfully.' elif action == 'deactivate': users.update(is_active=False) message = f'{users.count()} users deactivated successfully.' elif action == 'delete': count = users.count() users.delete() message = f'{count} users deleted successfully.' else: return JsonResponse({'error': 'Invalid action'}, status=400) # Log bulk action SecurityLog.objects.create( user=request.user, action=f'bulk_{action}', ip_address=request.META.get('REMOTE_ADDR'), user_agent=request.META.get('HTTP_USER_AGENT', '')[:255], details=f"Bulk {action} on {len(user_ids)} users" ) return JsonResponse({'message': message}) except Exception as e: logger.error(f"Error in bulk user action: {str(e)}") return JsonResponse({'error': 'Failed to perform bulk action'}, status=500) # ============================================================================ # API Key Management Views # ============================================================================ @login_required def api_key_list(request): """Display list of user's API keys.""" api_keys = APIKey.objects.filter(user=request.user).order_by('-created_at') return render(request, 'accounts/api_keys/list.html', { 'api_keys': api_keys }) @login_required @require_POST def generate_api_key(request): """Generate a new API key for the user.""" name = request.POST.get('name', '').strip() if not name: messages.error(request, 'API key name is required.') return redirect('accounts:api-key-list') # Check if user already has an API key with this name if APIKey.objects.filter(user=request.user, name=name).exists(): messages.error(request, 'You already have an API key with this name.') return redirect('accounts:api-key-list') # Generate new API key api_key = APIKey.objects.create( user=request.user, name=name, key=generate_api_key() ) # Log API key generation SecurityLog.objects.create( user=request.user, action='api_key_generated', ip_address=request.META.get('REMOTE_ADDR'), user_agent=request.META.get('HTTP_USER_AGENT', '')[:255], details=f"API key '{name}' generated" ) messages.success(request, f'API key "{name}" generated successfully.') logger.info(f"API key {name} generated for user {request.user.username}") return redirect('accounts:api-key-list') @login_required @require_POST def revoke_api_key(request, key_id): """Revoke an API key.""" try: api_key = get_object_or_404(APIKey, id=key_id, user=request.user) key_name = api_key.name api_key.is_active = False api_key.save() # Log API key revocation SecurityLog.objects.create( user=request.user, action='api_key_revoked', ip_address=request.META.get('REMOTE_ADDR'), user_agent=request.META.get('HTTP_USER_AGENT', '')[:255], details=f"API key '{key_name}' revoked" ) messages.success(request, f'API key "{key_name}" has been revoked.') logger.info(f"API key {key_name} revoked by user {request.user.username}") except Exception as e: logger.error(f"Error revoking API key: {str(e)}") messages.error(request, 'Failed to revoke API key.') return redirect('accounts:api-key-list') # ============================================================================ # Session Management Views # ============================================================================ @login_required def session_list(request): """Display list of user's active sessions.""" sessions = UserSession.objects.filter( user=request.user, expires_at__gt=timezone.now() ).order_by('-created_at') return render(request, 'accounts/sessions/list.html', { 'sessions': sessions }) @login_required @require_POST def terminate_session(request): """Terminate a specific session.""" session_id = request.POST.get('session_id') try: session = get_object_or_404(UserSession, id=session_id, user=request.user) session.delete() # Log session termination SecurityLog.objects.create( user=request.user, action='session_terminated', ip_address=request.META.get('REMOTE_ADDR'), user_agent=request.META.get('HTTP_USER_AGENT', '')[:255], details=f"Session terminated manually" ) messages.success(request, 'Session terminated successfully.') except Exception as e: logger.error(f"Error terminating session: {str(e)}") messages.error(request, 'Failed to terminate session.') return redirect('accounts:session-list')