# -*- coding: utf-8 -*- """ Accounts Views This module contains all views for the accounts app, including both web views and API views for user management, authentication, and profile operations. Views: Web Views: - UserListView: Display paginated list of users - UserDetailView: Show user details - UserCreateView: Create new user - UserUpdateView: Update user information - UserDeleteView: Delete user - ProfileView: User profile display - ProfileUpdateView: Update user profile - AvatarUpdateView: Update user avatar - PasswordChangeView: Change user password - UserAdminView: Admin dashboard - UserStatisticsView: User statistics - UserExportView: Export users - UserImportView: Import users - BulkUserActionView: Bulk operations - AJAX views for dynamic interactions API Views: - UserViewSet: RESTful user operations - Authentication API views - Profile API views - Statistics API views - Bulk operations API views - Search and filter API views Features: - Permission-based access control - Pagination and filtering - AJAX support - File upload handling - Export/import functionality - Bulk operations - Activity logging - Error handling Security: - Authentication required - Permission checks - CSRF protection - Input validation - Rate limiting """ import json import csv import io from datetime import datetime, timedelta from typing import Any, Dict, List, Optional from django.contrib import messages from django.contrib.auth import get_user_model, login, logout from django.contrib.auth.decorators import login_required, permission_required from django.contrib.auth.mixins import LoginRequiredMixin, PermissionRequiredMixin from django.core.exceptions import ValidationError, PermissionDenied from django.core.paginator import Paginator, EmptyPage, PageNotAnInteger from django.db import transaction from django.db.models import Q, Count, Avg, Sum from django.http import ( HttpResponse, HttpResponseRedirect, JsonResponse, HttpResponseBadRequest, Http404 ) from django.shortcuts import render, get_object_or_404, redirect from django.urls import reverse_lazy, reverse from django.utils import timezone from django.utils.decorators import method_decorator from django.views import View from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_http_methods from django.views.generic import ( ListView, DetailView, CreateView, UpdateView, DeleteView, TemplateView, FormView ) from rest_framework import status, viewsets, permissions from rest_framework.decorators import action, api_view, permission_classes from rest_framework.permissions import IsAuthenticated, IsAdminUser from rest_framework.response import Response from rest_framework.views import APIView from rest_framework.generics import ( ListCreateAPIView, RetrieveUpdateDestroyAPIView, ListAPIView, CreateAPIView, UpdateAPIView ) from rest_framework.parsers import MultiPartParser, FormParser, JSONParser from rest_framework.filters import SearchFilter, OrderingFilter from django_filters.rest_framework import DjangoFilterBackend from apps.core.utils import generate_unique_id, send_email_async from apps.activities.models import Activity from .models import User from .forms import ( UserCreationForm, UserUpdateForm, ProfileForm, BulkUserActionForm ) from .serializers import ( UserSerializer, UserDetailSerializer, UserCreateSerializer, UserUpdateSerializer, UserListSerializer, PasswordChangeSerializer, BulkUserActionSerializer ) from .services import UserService, AuthenticationService, ProfileService, ExportService from .tasks import ( send_welcome_email_task, send_password_reset_email_task, cleanup_inactive_users_task, generate_user_report_task, bulk_user_operation_task ) User = get_user_model() # ============================================================================ # Web Views # ============================================================================ class UserListView(LoginRequiredMixin, PermissionRequiredMixin, ListView): """ Display paginated list of users with search and filter capabilities. Features: - Pagination - Search by name, email - Filter by status, role, date joined - Bulk selection - Export options """ model = User template_name = 'accounts/user_list.html' context_object_name = 'users' paginate_by = 25 permission_required = 'accounts.view_user' def get_queryset(self): queryset = User.objects.select_related().prefetch_related('groups') # Search functionality search_query = self.request.GET.get('search', '') if search_query: queryset = queryset.filter( Q(first_name__icontains=search_query) | Q(last_name__icontains=search_query) | Q(email__icontains=search_query) ) # Filter by status status_filter = self.request.GET.get('status', '') if status_filter == 'active': queryset = queryset.filter(is_active=True) elif status_filter == 'inactive': queryset = queryset.filter(is_active=False) # Filter by role role_filter = self.request.GET.get('role', '') if role_filter: queryset = queryset.filter(groups__name=role_filter) # Filter by date joined date_filter = self.request.GET.get('date_joined', '') if date_filter: if date_filter == 'today': queryset = queryset.filter(date_joined__date=timezone.now().date()) elif date_filter == 'week': week_ago = timezone.now() - timedelta(days=7) queryset = queryset.filter(date_joined__gte=week_ago) elif date_filter == 'month': month_ago = timezone.now() - timedelta(days=30) queryset = queryset.filter(date_joined__gte=month_ago) # Ordering order_by = self.request.GET.get('order_by', '-date_joined') if order_by in ['first_name', '-first_name', 'last_name', '-last_name', 'email', '-email', 'date_joined', '-date_joined', 'last_login', '-last_login']: queryset = queryset.order_by(order_by) return queryset.distinct() def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) context.update({ 'search_query': self.request.GET.get('search', ''), 'status_filter': self.request.GET.get('status', ''), 'role_filter': self.request.GET.get('role', ''), 'date_filter': self.request.GET.get('date_joined', ''), 'order_by': self.request.GET.get('order_by', '-date_joined'), 'total_users': User.objects.count(), 'active_users': User.objects.filter(is_active=True).count(), 'inactive_users': User.objects.filter(is_active=False).count(), 'bulk_form': BulkUserActionForm(), }) return context class UserDetailView(LoginRequiredMixin, PermissionRequiredMixin, DetailView): """ Display detailed information about a specific user. Features: - User profile information - Activity history - Statistics - Action buttons (edit, delete, etc.) """ model = User template_name = 'accounts/user_detail.html' context_object_name = 'user_obj' permission_required = 'accounts.view_user' def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) user_obj = self.get_object() # Get user activities activities = Activity.objects.filter( user=user_obj ).order_by('-created_at')[:10] # Get user statistics stats = UserService.get_user_statistics(user_obj) context.update({ 'activities': activities, 'stats': stats, 'can_edit': self.request.user.has_perm('accounts.change_user'), 'can_delete': self.request.user.has_perm('accounts.delete_user'), }) return context class UserCreateView(LoginRequiredMixin, PermissionRequiredMixin, CreateView): """ Create a new user account. Features: - User registration form - Email validation - Password strength validation - Welcome email sending - Activity logging """ model = User form_class = UserCreationForm template_name = 'accounts/user_create.html' permission_required = 'accounts.add_user' success_url = reverse_lazy('accounts:user_list') def form_valid(self, form): response = super().form_valid(form) # Send welcome email send_welcome_email_task.delay(self.object.id) # Log activity Activity.objects.create( user=self.request.user, action='user_created', description=f'Created user: {self.object.email}', metadata={'user_id': self.object.id} ) messages.success( self.request, f'User {self.object.email} has been created successfully.' ) return response class UserUpdateView(LoginRequiredMixin, PermissionRequiredMixin, UpdateView): """ Update user information. Features: - User profile editing - Permission management - Status changes - Activity logging """ model = User form_class = UserUpdateForm template_name = 'accounts/user_update.html' permission_required = 'accounts.change_user' def get_success_url(self): return reverse('accounts:user_detail', kwargs={'pk': self.object.pk}) def form_valid(self, form): response = super().form_valid(form) # Log activity Activity.objects.create( user=self.request.user, action='user_updated', description=f'Updated user: {self.object.email}', metadata={'user_id': self.object.id} ) messages.success( self.request, f'User {self.object.email} has been updated successfully.' ) return response class UserDeleteView(LoginRequiredMixin, PermissionRequiredMixin, DeleteView): """ Delete a user account. Features: - Confirmation dialog - Soft delete option - Data cleanup - Activity logging """ model = User template_name = 'accounts/user_delete.html' permission_required = 'accounts.delete_user' success_url = reverse_lazy('accounts:user_list') def delete(self, request, *args, **kwargs): user_obj = self.get_object() email = user_obj.email # Log activity before deletion Activity.objects.create( user=request.user, action='user_deleted', description=f'Deleted user: {email}', metadata={'user_id': user_obj.id} ) response = super().delete(request, *args, **kwargs) messages.success( request, f'User {email} has been deleted successfully.' ) return response class ProfileView(LoginRequiredMixin, TemplateView): """ Display user's own profile. Features: - Profile information display - Activity history - Statistics - Quick actions """ template_name = 'accounts/profile.html' def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) user = self.request.user # Get user activities activities = Activity.objects.filter( user=user ).order_by('-created_at')[:10] # Get user statistics stats = UserService.get_user_statistics(user) context.update({ 'activities': activities, 'stats': stats, }) return context class ProfileUpdateView(LoginRequiredMixin, UpdateView): """ Update user's own profile. Features: - Profile editing - Avatar upload - Preferences management - Activity logging """ model = User form_class = ProfileForm template_name = 'accounts/profile_update.html' success_url = reverse_lazy('accounts:profile') def get_object(self): return self.request.user def form_valid(self, form): response = super().form_valid(form) # Log activity Activity.objects.create( user=self.request.user, action='profile_updated', description='Updated profile information' ) messages.success( self.request, 'Your profile has been updated successfully.' ) return response class AvatarUpdateView(LoginRequiredMixin, View): """ Handle avatar upload and update. Features: - Image upload - Image validation - Image processing - AJAX support """ def post(self, request, *args, **kwargs): if 'avatar' not in request.FILES: return JsonResponse({ 'success': False, 'error': 'No avatar file provided' }) avatar_file = request.FILES['avatar'] try: # Update avatar using service ProfileService.update_avatar(request.user, avatar_file) # Log activity Activity.objects.create( user=request.user, action='avatar_updated', description='Updated profile avatar' ) return JsonResponse({ 'success': True, 'avatar_url': request.user.avatar.url if request.user.avatar else None }) except ValidationError as e: return JsonResponse({ 'success': False, 'error': str(e) }) class PasswordChangeView(LoginRequiredMixin, FormView): """ Change user password. Features: - Current password validation - New password strength validation - Password confirmation - Activity logging """ template_name = 'accounts/password_change.html' success_url = reverse_lazy('accounts:profile') def get_form_class(self): from django.contrib.auth.forms import PasswordChangeForm return PasswordChangeForm def get_form_kwargs(self): kwargs = super().get_form_kwargs() kwargs['user'] = self.request.user return kwargs def form_valid(self, form): form.save() # Log activity Activity.objects.create( user=self.request.user, action='password_changed', description='Changed account password' ) messages.success( self.request, 'Your password has been changed successfully.' ) return super().form_valid(form) class UserAdminView(LoginRequiredMixin, PermissionRequiredMixin, TemplateView): """ Admin dashboard for user management. Features: - User statistics - Recent activities - Quick actions - System health """ template_name = 'accounts/user_admin.html' permission_required = 'accounts.view_user' def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) # Get comprehensive statistics stats = UserService.get_comprehensive_statistics() # Get recent activities recent_activities = Activity.objects.filter( action__in=['user_created', 'user_updated', 'user_deleted'] ).order_by('-created_at')[:10] context.update({ 'stats': stats, 'recent_activities': recent_activities, }) return context class UserStatisticsView(LoginRequiredMixin, PermissionRequiredMixin, TemplateView): """ Detailed user statistics and analytics. Features: - User growth charts - Activity analytics - Geographic distribution - Export options """ template_name = 'accounts/user_statistics.html' permission_required = 'accounts.view_user' def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) # Get detailed statistics stats = UserService.get_detailed_statistics() context.update({ 'stats': stats, }) return context class UserExportView(LoginRequiredMixin, PermissionRequiredMixin, View): """ Export user data in various formats. Features: - CSV export - Excel export - JSON export - Filtered exports """ permission_required = 'accounts.view_user' def get(self, request, *args, **kwargs): export_format = request.GET.get('format', 'csv') # Get filtered queryset queryset = self.get_filtered_queryset() if export_format == 'csv': return ExportService.export_users_csv(queryset) elif export_format == 'excel': return ExportService.export_users_excel(queryset) elif export_format == 'json': return ExportService.export_users_json(queryset) else: return HttpResponseBadRequest('Invalid export format') def get_filtered_queryset(self): queryset = User.objects.all() # Apply filters similar to UserListView search_query = self.request.GET.get('search', '') if search_query: queryset = queryset.filter( Q(first_name__icontains=search_query) | Q(last_name__icontains=search_query) | Q(email__icontains=search_query) ) status_filter = self.request.GET.get('status', '') if status_filter == 'active': queryset = queryset.filter(is_active=True) elif status_filter == 'inactive': queryset = queryset.filter(is_active=False) return queryset class UserImportView(LoginRequiredMixin, PermissionRequiredMixin, FormView): """ Import users from file. Features: - CSV import - Excel import - Data validation - Error reporting """ template_name = 'accounts/user_import.html' permission_required = 'accounts.add_user' success_url = reverse_lazy('accounts:user_list') def post(self, request, *args, **kwargs): if 'import_file' not in request.FILES: messages.error(request, 'No import file provided') return redirect('accounts:user_import') import_file = request.FILES['import_file'] try: # Process import using service result = ExportService.import_users_from_file(import_file) # Log activity Activity.objects.create( user=request.user, action='users_imported', description=f'Imported {result["created"]} users, updated {result["updated"]} users', metadata=result ) messages.success( request, f'Successfully imported {result["created"]} users and updated {result["updated"]} users.' ) if result['errors']: messages.warning( request, f'{len(result["errors"])} errors occurred during import. Check the logs for details.' ) except Exception as e: messages.error(request, f'Import failed: {str(e)}') return redirect('accounts:user_list') class BulkUserActionView(LoginRequiredMixin, PermissionRequiredMixin, View): """ Handle bulk operations on users. Features: - Bulk activate/deactivate - Bulk delete - Bulk export - Bulk role assignment """ permission_required = 'accounts.change_user' def post(self, request, *args, **kwargs): form = BulkUserActionForm(request.POST) if not form.is_valid(): messages.error(request, 'Invalid form data') return redirect('accounts:user_list') action = form.cleaned_data['action'] user_ids = form.cleaned_data['user_ids'] try: # Execute bulk operation asynchronously task = bulk_user_operation_task.delay( action=action, user_ids=user_ids, performed_by=request.user.id ) # Log activity Activity.objects.create( user=request.user, action='bulk_user_action', description=f'Initiated bulk {action} for {len(user_ids)} users', metadata={ 'action': action, 'user_count': len(user_ids), 'task_id': task.id } ) messages.success( request, f'Bulk {action} operation has been initiated for {len(user_ids)} users.' ) except Exception as e: messages.error(request, f'Bulk operation failed: {str(e)}') return redirect('accounts:user_list') # ============================================================================ # AJAX Views # ============================================================================ class UserSearchAjaxView(LoginRequiredMixin, View): """ AJAX endpoint for user search. """ def get(self, request, *args, **kwargs): query = request.GET.get('q', '') if len(query) < 2: return JsonResponse({'users': []}) users = User.objects.filter( Q(first_name__icontains=query) | Q(last_name__icontains=query) | Q(email__icontains=query) )[:10] user_data = [{ 'id': user.id, 'name': user.get_full_name(), 'email': user.email, 'avatar': user.avatar.url if user.avatar else None } for user in users] return JsonResponse({'users': user_data}) class ToggleUserStatusAjaxView(LoginRequiredMixin, PermissionRequiredMixin, View): """ AJAX endpoint to toggle user active status. """ permission_required = 'accounts.change_user' def post(self, request, pk, *args, **kwargs): try: user = get_object_or_404(User, pk=pk) user.is_active = not user.is_active user.save() # Log activity Activity.objects.create( user=request.user, action='user_status_toggled', description=f'Toggled status for user: {user.email}', metadata={'user_id': user.id, 'new_status': user.is_active} ) return JsonResponse({ 'success': True, 'is_active': user.is_active, 'status_text': 'Active' if user.is_active else 'Inactive' }) except Exception as e: return JsonResponse({ 'success': False, 'error': str(e) }) class UserInfoAjaxView(LoginRequiredMixin, View): """ AJAX endpoint to get user information. """ def get(self, request, pk, *args, **kwargs): try: user = get_object_or_404(User, pk=pk) return JsonResponse({ 'success': True, 'user': { 'id': user.id, 'name': user.get_full_name(), 'email': user.email, 'is_active': user.is_active, 'date_joined': user.date_joined.isoformat(), 'last_login': user.last_login.isoformat() if user.last_login else None, 'avatar': user.avatar.url if user.avatar else None } }) except Exception as e: return JsonResponse({ 'success': False, 'error': str(e) }) class CheckEmailAjaxView(View): """ AJAX endpoint to check email availability. """ def get(self, request, *args, **kwargs): 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, 'email': email }) # ============================================================================ # API Views # ============================================================================ class UserViewSet(viewsets.ModelViewSet): """ ViewSet for User model providing full CRUD operations. Features: - List, create, retrieve, update, delete users - Search and filtering - Pagination - Permission-based access """ queryset = User.objects.all() permission_classes = [IsAuthenticated] filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter] search_fields = ['first_name', 'last_name', 'email'] ordering_fields = ['first_name', 'last_name', 'email', 'date_joined'] ordering = ['-date_joined'] def get_serializer_class(self): if self.action == 'list': return UserListSerializer elif self.action == 'create': return UserCreateSerializer elif self.action in ['update', 'partial_update']: return UserUpdateSerializer else: return UserDetailSerializer def get_permissions(self): if self.action == 'create': permission_classes = [permissions.AllowAny] elif self.action in ['update', 'partial_update', 'destroy']: permission_classes = [IsAuthenticated, IsAdminUser] else: permission_classes = [IsAuthenticated] return [permission() for permission in permission_classes] def perform_create(self, serializer): user = serializer.save() # Send welcome email send_welcome_email_task.delay(user.id) # Log activity Activity.objects.create( user=self.request.user if self.request.user.is_authenticated else None, action='user_created_api', description=f'Created user via API: {user.email}', metadata={'user_id': user.id} ) def perform_update(self, serializer): user = serializer.save() # Log activity Activity.objects.create( user=self.request.user, action='user_updated_api', description=f'Updated user via API: {user.email}', metadata={'user_id': user.id} ) def perform_destroy(self, instance): email = instance.email # Log activity before deletion Activity.objects.create( user=self.request.user, action='user_deleted_api', description=f'Deleted user via API: {email}', metadata={'user_id': instance.id} ) instance.delete() @action(detail=True, methods=['post']) def activate(self, request, pk=None): """Activate a user.""" user = self.get_object() user.is_active = True user.save() Activity.objects.create( user=request.user, action='user_activated_api', description=f'Activated user via API: {user.email}', metadata={'user_id': user.id} ) return Response({'status': 'User activated'}) @action(detail=True, methods=['post']) def deactivate(self, request, pk=None): """Deactivate a user.""" user = self.get_object() user.is_active = False user.save() Activity.objects.create( user=request.user, action='user_deactivated_api', description=f'Deactivated user via API: {user.email}', metadata={'user_id': user.id} ) return Response({'status': 'User deactivated'}) @action(detail=True, methods=['get']) def activities(self, request, pk=None): """Get user activities.""" user = self.get_object() activities = Activity.objects.filter(user=user).order_by('-created_at')[:20] activity_data = [{ 'id': activity.id, 'action': activity.action, 'description': activity.description, 'created_at': activity.created_at.isoformat(), 'metadata': activity.metadata } for activity in activities] return Response({'activities': activity_data}) class LoginAPIView(APIView): """ API endpoint for user login. """ permission_classes = [permissions.AllowAny] def post(self, request): email = request.data.get('email') password = request.data.get('password') if not email or not password: return Response({ 'error': 'Email and password are required' }, status=status.HTTP_400_BAD_REQUEST) try: user = AuthenticationService.authenticate_user(email, password) if user: # Log activity Activity.objects.create( user=user, action='login_api', description='Logged in via API' ) serializer = UserDetailSerializer(user) return Response({ 'user': serializer.data, 'message': 'Login successful' }) else: return Response({ 'error': 'Invalid credentials' }, status=status.HTTP_401_UNAUTHORIZED) except Exception as e: return Response({ 'error': str(e) }, status=status.HTTP_500_INTERNAL_SERVER_ERROR) class LogoutAPIView(APIView): """ API endpoint for user logout. """ permission_classes = [IsAuthenticated] def post(self, request): try: # Log activity Activity.objects.create( user=request.user, action='logout_api', description='Logged out via API' ) # Perform logout AuthenticationService.logout_user(request) return Response({ 'message': 'Logout successful' }) except Exception as e: return Response({ 'error': str(e) }, status=status.HTTP_500_INTERNAL_SERVER_ERROR) class RegisterAPIView(CreateAPIView): """ API endpoint for user registration. """ queryset = User.objects.all() serializer_class = UserCreateSerializer permission_classes = [permissions.AllowAny] def perform_create(self, serializer): user = serializer.save() # Send welcome email send_welcome_email_task.delay(user.id) # Log activity Activity.objects.create( user=user, action='register_api', description='Registered via API' ) class PasswordChangeAPIView(APIView): """ API endpoint for password change. """ permission_classes = [IsAuthenticated] def post(self, request): serializer = PasswordChangeSerializer( data=request.data, context={'request': request} ) if serializer.is_valid(): serializer.save() # Log activity Activity.objects.create( user=request.user, action='password_changed_api', description='Changed password via API' ) return Response({ 'message': 'Password changed successfully' }) return Response( serializer.errors, status=status.HTTP_400_BAD_REQUEST ) class PasswordResetAPIView(APIView): """ API endpoint for password reset. """ permission_classes = [permissions.AllowAny] def post(self, request): email = request.data.get('email') if not email: return Response({ 'error': 'Email is required' }, status=status.HTTP_400_BAD_REQUEST) try: user = User.objects.get(email=email) # Send password reset email send_password_reset_email_task.delay(user.id) # Log activity Activity.objects.create( user=user, action='password_reset_requested_api', description='Requested password reset via API' ) return Response({ 'message': 'Password reset email sent' }) except User.DoesNotExist: # Don't reveal if email exists return Response({ 'message': 'Password reset email sent' }) class ProfileAPIView(RetrieveUpdateDestroyAPIView): """ API endpoint for user profile management. """ serializer_class = UserDetailSerializer permission_classes = [IsAuthenticated] def get_object(self): return self.request.user def perform_update(self, serializer): serializer.save() # Log activity Activity.objects.create( user=self.request.user, action='profile_updated_api', description='Updated profile via API' ) class AvatarUpdateAPIView(APIView): """ API endpoint for avatar update. """ permission_classes = [IsAuthenticated] parser_classes = [MultiPartParser, FormParser] def post(self, request): if 'avatar' not in request.FILES: return Response({ 'error': 'No avatar file provided' }, status=status.HTTP_400_BAD_REQUEST) avatar_file = request.FILES['avatar'] try: # Update avatar using service ProfileService.update_avatar(request.user, avatar_file) # Log activity Activity.objects.create( user=request.user, action='avatar_updated_api', description='Updated avatar via API' ) return Response({ 'message': 'Avatar updated successfully', 'avatar_url': request.user.avatar.url if request.user.avatar else None }) except ValidationError as e: return Response({ 'error': str(e) }, status=status.HTTP_400_BAD_REQUEST) class UserStatisticsAPIView(APIView): """ API endpoint for user statistics. """ permission_classes = [IsAuthenticated] def get(self, request): stats = UserService.get_comprehensive_statistics() return Response(stats) class DashboardStatsAPIView(APIView): """ API endpoint for dashboard statistics. """ permission_classes = [IsAuthenticated] def get(self, request): stats = UserService.get_dashboard_statistics() return Response(stats) class BulkUserActionAPIView(APIView): """ API endpoint for bulk user operations. """ permission_classes = [IsAuthenticated, IsAdminUser] def post(self, request): serializer = BulkUserActionSerializer(data=request.data) if serializer.is_valid(): action = serializer.validated_data['action'] user_ids = serializer.validated_data['user_ids'] try: # Execute bulk operation asynchronously task = bulk_user_operation_task.delay( action=action, user_ids=user_ids, performed_by=request.user.id ) # Log activity Activity.objects.create( user=request.user, action='bulk_user_action_api', description=f'Initiated bulk {action} for {len(user_ids)} users via API', metadata={ 'action': action, 'user_count': len(user_ids), 'task_id': task.id } ) return Response({ 'message': f'Bulk {action} operation initiated', 'task_id': task.id, 'user_count': len(user_ids) }) except Exception as e: return Response({ 'error': str(e) }, status=status.HTTP_500_INTERNAL_SERVER_ERROR) return Response( serializer.errors, status=status.HTTP_400_BAD_REQUEST ) class BulkExportAPIView(APIView): """ API endpoint for bulk user export. """ permission_classes = [IsAuthenticated] def get(self, request): export_format = request.GET.get('format', 'csv') # Get filtered queryset queryset = User.objects.all() # Apply filters search = request.GET.get('search') if search: queryset = queryset.filter( Q(first_name__icontains=search) | Q(last_name__icontains=search) | Q(email__icontains=search) ) is_active = request.GET.get('is_active') if is_active is not None: queryset = queryset.filter(is_active=is_active.lower() == 'true') try: if export_format == 'csv': return ExportService.export_users_csv(queryset) elif export_format == 'excel': return ExportService.export_users_excel(queryset) elif export_format == 'json': return ExportService.export_users_json(queryset) else: return Response({ 'error': 'Invalid export format' }, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({ 'error': str(e) }, status=status.HTTP_500_INTERNAL_SERVER_ERROR) class UserSearchAPIView(ListAPIView): """ API endpoint for user search. """ serializer_class = UserListSerializer permission_classes = [IsAuthenticated] filter_backends = [SearchFilter, DjangoFilterBackend] search_fields = ['first_name', 'last_name', 'email'] def get_queryset(self): queryset = User.objects.all() # Apply search search = self.request.GET.get('search') if search: queryset = queryset.filter( Q(first_name__icontains=search) | Q(last_name__icontains=search) | Q(email__icontains=search) ) return queryset[:20] # Limit results class UserFilterAPIView(ListAPIView): """ API endpoint for user filtering. """ serializer_class = UserListSerializer permission_classes = [IsAuthenticated] filter_backends = [DjangoFilterBackend, OrderingFilter] filterset_fields = ['is_active', 'is_staff', 'date_joined'] ordering_fields = ['first_name', 'last_name', 'email', 'date_joined'] def get_queryset(self): return User.objects.all() class UserRolesAPIView(APIView): """ API endpoint for user roles management. """ permission_classes = [IsAuthenticated, IsAdminUser] def get(self, request): from django.contrib.auth.models import Group roles = Group.objects.all().values('id', 'name') return Response({'roles': list(roles)}) def post(self, request): user_id = request.data.get('user_id') role_ids = request.data.get('role_ids', []) try: user = User.objects.get(id=user_id) user.groups.set(role_ids) # Log activity Activity.objects.create( user=request.user, action='user_roles_updated_api', description=f'Updated roles for user: {user.email}', metadata={'user_id': user.id, 'role_ids': role_ids} ) return Response({ 'message': 'User roles updated successfully' }) except User.DoesNotExist: return Response({ 'error': 'User not found' }, status=status.HTTP_404_NOT_FOUND) class UserPermissionsAPIView(APIView): """ API endpoint for user permissions management. """ permission_classes = [IsAuthenticated, IsAdminUser] def get(self, request): user_id = request.GET.get('user_id') if user_id: try: user = User.objects.get(id=user_id) permissions = user.get_all_permissions() return Response({'permissions': list(permissions)}) except User.DoesNotExist: return Response({ 'error': 'User not found' }, status=status.HTTP_404_NOT_FOUND) from django.contrib.auth.models import Permission all_permissions = Permission.objects.all().values( 'id', 'name', 'codename', 'content_type__app_label' ) return Response({'permissions': list(all_permissions)}) class UserActivityAPIView(APIView): """ API endpoint for user activity history. """ permission_classes = [IsAuthenticated] def get(self, request, pk): try: user = User.objects.get(id=pk) # Check permission if not (request.user == user or request.user.has_perm('accounts.view_user')): raise PermissionDenied activities = Activity.objects.filter( user=user ).order_by('-created_at')[:50] activity_data = [{ 'id': activity.id, 'action': activity.action, 'description': activity.description, 'created_at': activity.created_at.isoformat(), 'metadata': activity.metadata } for activity in activities] return Response({'activities': activity_data}) except User.DoesNotExist: return Response({ 'error': 'User not found' }, status=status.HTTP_404_NOT_FOUND)