from django.shortcuts import render, get_object_or_404, redirect from django.contrib.auth.decorators import login_required, user_passes_test from django.contrib import messages from django.db.models import Q, Sum, Count from django.db.models.functions import Coalesce from django.http import JsonResponse, HttpResponseRedirect, HttpResponse from django.core.paginator import Paginator from django.utils import timezone from django.contrib.auth import authenticate, login, logout from django.contrib.auth.forms import AuthenticationForm from django.views.decorators.csrf import csrf_protect from django.views.decorators.http import require_http_methods from django.urls import reverse from django.views.decorators.clickjacking import xframe_options_sameorigin from .models import CustomUser from .decorators import require_role, admin_required, staff_required from loans.models import Loan, LoanApplication from reports.models import LoanScoring from utils.models import AuditLog, Document import json from PIL import Image from io import BytesIO from django.core.files.base import ContentFile from utils.models import Notification import magic from loans.models import Repayment def invalidate_client_popup_cache(client_id): """Invalidate cached client popup data when client is updated""" from django.core.cache import cache cache_key = f'client_popup_{client_id}' cache.delete(cache_key) print(f"Invalidated cache for client popup: {cache_key}") @csrf_protect @require_http_methods(["GET", "POST"]) def custom_login(request): """ Custom login view with enhanced security and email-based authentication """ # Redirect if user is already authenticated if request.user.is_authenticated: return HttpResponseRedirect(reverse('main_dashboard')) if request.method == 'POST': # Get credentials from POST data username = request.POST.get('username') password = request.POST.get('password') # Try to authenticate user = authenticate(request, username=username, password=password) if user is not None: # Check if user is active and not suspended if user.is_active and user.status != 'suspended': login(request, user) # Update last login time user.last_login_at = timezone.now() user.save() # Create audit log AuditLog.objects.create( user=user, action='login', model_name='CustomUser', object_id=str(user.id), description=f'User logged in via email: {username}', ip_address=request.META.get('REMOTE_ADDR'), user_agent=request.META.get('HTTP_USER_AGENT', '') ) # Redirect to main dashboard return HttpResponseRedirect(reverse('main_dashboard')) else: messages.error(request, 'Your account has been suspended. Please contact support.') else: messages.error(request, 'Invalid email or password. Please try again.') # If we get here, authentication failed - just re-render the form # GET request or failed POST return render(request, 'users/login.html', { 'form': AuthenticationForm() }) @login_required def custom_logout(request): """Custom logout view""" # Create audit log AuditLog.objects.create( user=request.user, action='logout', model_name='CustomUser', object_id=str(request.user.id), description=f'User logged out: {request.user.get_full_name()}' ) # Perform logout logout(request) messages.success(request, 'You have been logged out successfully') return redirect('users:login') @login_required def client_list(request): """List all clients with filtering and search""" try: # Get filter parameters search = request.GET.get('search', '') status_filter = request.GET.get('status', '') business_type_filter = request.GET.get('business_type', '') sort_by = request.GET.get('sort_by', 'first_name') # Default sort by first name sort_order = request.GET.get('sort_order', 'asc') # Default ascending order # Base queryset clients = CustomUser.objects.filter(role='borrower').select_related() # Apply filters if search: clients = clients.filter( Q(first_name__icontains=search) | Q(last_name__icontains=search) | Q(business_name__icontains=search) | Q(id_number__icontains=search) | Q(phone_number__icontains=search) | Q(email__icontains=search) ) if status_filter: clients = clients.filter(status=status_filter) if business_type_filter: clients = clients.filter(business_type=business_type_filter) # Apply sorting if sort_by == 'first_name': if sort_order == 'desc': clients = clients.order_by('-first_name', '-last_name') else: clients = clients.order_by('first_name', 'last_name') elif sort_by == 'last_name': if sort_order == 'desc': clients = clients.order_by('-last_name', '-first_name') else: clients = clients.order_by('last_name', 'first_name') elif sort_by == 'business_name': if sort_order == 'desc': clients = clients.order_by('-business_name') else: clients = clients.order_by('business_name') elif sort_by == 'status': if sort_order == 'desc': clients = clients.order_by('-status') else: clients = clients.order_by('status') elif sort_by == 'created_at': if sort_order == 'desc': clients = clients.order_by('-created_at') else: clients = clients.order_by('created_at') else: # Default sorting by first name ascending clients = clients.order_by('first_name', 'last_name') # Get loan statistics for each client for client in clients: try: client.active_loans = Loan.objects.filter(borrower=client, status='active').count() client.total_borrowed = Loan.objects.filter(borrower=client).aggregate( total=Sum('principal_amount') )['total'] or 0 client.total_repaid = Loan.objects.filter(borrower=client).aggregate( total=Sum('amount_paid') )['total'] or 0 # Get credit score try: credit_score = LoanScoring.objects.get(user=client) client.credit_score = credit_score.total_score except LoanScoring.DoesNotExist: client.credit_score = 0 except Exception as e: # Handle individual client errors client.active_loans = 0 client.total_borrowed = 0 client.total_repaid = 0 client.credit_score = 0 print(f"Error processing client {client.id}: {str(e)}") # Pagination paginator = Paginator(clients, 10) page_number = request.GET.get('page') page_obj = paginator.get_page(page_number) context = { 'clients': page_obj, 'search': search, 'status_filter': status_filter, 'business_type_filter': business_type_filter, 'sort_by': sort_by, 'sort_order': sort_order, 'total_clients': clients.count(), } return render(request, 'users/client_list.html', context) except Exception as e: # Log the error and return a simple error page print(f"Error in client_list view: {str(e)}") return render(request, 'users/client_list.html', { 'clients': [], 'search': '', 'status_filter': 'all', 'business_type_filter': 'all', 'sort_by': 'first_name', 'sort_order': 'asc', 'total_clients': 0, 'error_message': f'An error occurred while loading clients: {str(e)}' }) @login_required def client_list_test(request): """Simple test view for client list debugging""" try: clients = CustomUser.objects.filter(role='borrower')[:5] # Just get first 5 clients # Simple data without complex queries for client in clients: client.active_loans = 0 client.total_borrowed = 0 client.total_repaid = 0 client.credit_score = 0 context = { 'clients': clients, 'search': '', 'status_filter': 'all', 'business_type_filter': 'all', 'total_clients': clients.count(), } return render(request, 'users/client_list.html', context) except Exception as e: print(f"Error in test view: {str(e)}") return render(request, 'users/client_list.html', { 'clients': [], 'search': '', 'status_filter': 'all', 'business_type_filter': 'all', 'total_clients': 0, 'error_message': f'Test error: {str(e)}' }) @login_required def client_detail(request, client_id): """View for client details""" client = get_object_or_404(CustomUser, id=client_id) # Check permissions if not request.user.is_staff and request.user != client: messages.error(request, 'You do not have permission to view this client') return redirect('users:client_list') # Get document counts documents = Document.objects.filter(uploaded_by=client) # Get loan history loans = Loan.objects.filter(borrower=client).order_by('-created_at') applications = LoanApplication.objects.filter(borrower=client).order_by('-submitted_at') # Get credit score try: credit_score = LoanScoring.objects.get(user=client) except LoanScoring.DoesNotExist: credit_score = None # Calculate statistics total_borrowed = loans.aggregate(total=Sum('principal_amount'))['total'] or 0 total_repaid = loans.aggregate(total=Sum('amount_paid'))['total'] or 0 active_loans = loans.filter(status='active').count() context = { 'client': client, 'id_documents': documents.filter(document_type='id_document'), 'selfie_documents': documents.filter(document_type='selfie'), 'utility_documents': documents.filter(document_type='utility_bill'), 'bank_documents': documents.filter(document_type='bank_statement'), 'loans': loans, 'applications': applications, 'credit_score': credit_score, 'total_borrowed': total_borrowed, 'total_repaid': total_repaid, 'active_loans': active_loans, } return render(request, 'users/client_detail.html', context) @login_required def client_create(request): """Create a new client with comprehensive registration form""" if request.method == 'POST': try: # Extract form data full_name = request.POST.get('first_name', '').strip() # Split full name into first and last name name_parts = full_name.split(' ', 1) first_name = name_parts[0] if name_parts else '' last_name = name_parts[1] if len(name_parts) > 1 else '' personal_data = { 'first_name': first_name, 'last_name': last_name, 'email': request.POST.get('email'), 'phone_number': request.POST.get('phone_number'), 'date_of_birth': request.POST.get('date_of_birth'), 'gender': request.POST.get('gender'), 'id_number': request.POST.get('id_number'), 'nationality': request.POST.get('nationality', 'Kenyan'), 'marital_status': request.POST.get('marital_status'), 'physical_address': request.POST.get('address'), 'city': request.POST.get('city'), 'postal_code': request.POST.get('postal_code'), } # Get business type and handle "Other" option business_type = request.POST.get('business_type') if business_type == 'other': business_type = request.POST.get('other_business_type') business_data = { 'business_name': request.POST.get('business_name'), 'business_type': business_type, 'business_address': request.POST.get('business_address'), } # Validate required fields required_fields = ['phone_number', 'date_of_birth', 'gender', 'id_number', 'nationality', 'physical_address', 'city', 'business_name', 'business_type'] missing_fields = [] for field in required_fields: if not request.POST.get(field): missing_fields.append(field.replace("_", " ").title()) # Validate required file fields required_file_fields = ['id_document', 'selfie'] missing_file_fields = [] for field in required_file_fields: if not request.FILES.get(field): missing_file_fields.append(field.replace('_', ' ').title()) if missing_fields or missing_file_fields: all_missing = missing_fields + missing_file_fields messages.error(request, f'Required fields missing: {", ".join(all_missing)}') return render(request, 'users/client_create.html', { 'business_types': [ ('retail', 'Retail'), ('agriculture', 'Agriculture'), ('transport', 'Transport'), ('manufacturing', 'Manufacturing'), ('services', 'Services'), ('technology', 'Technology'), ], 'genders': CustomUser._meta.get_field('gender').choices, 'form_data': request.POST }) # Check if user already exists if personal_data['email'] and CustomUser.objects.filter(email=personal_data['email']).exists(): messages.error(request, 'This email address is already registered. Please use a different email address.') return render(request, 'users/client_create.html', { 'business_types': [ ('retail', 'Retail'), ('agriculture', 'Agriculture'), ('transport', 'Transport'), ('manufacturing', 'Manufacturing'), ('services', 'Services'), ('technology', 'Technology'), ], 'genders': CustomUser._meta.get_field('gender').choices, 'form_data': request.POST }) if CustomUser.objects.filter(phone_number=personal_data['phone_number']).exists(): messages.error(request, 'This phone number is already registered. Please use a different phone number.') return render(request, 'users/client_create.html', { 'business_types': [ ('retail', 'Retail'), ('agriculture', 'Agriculture'), ('transport', 'Transport'), ('manufacturing', 'Manufacturing'), ('services', 'Services'), ('technology', 'Technology'), ], 'genders': CustomUser._meta.get_field('gender').choices, 'form_data': request.POST }) # Create user # Remove email and phone_number from personal_data to avoid duplicates user_data = personal_data.copy() email = user_data.pop('email') phone_number = user_data.pop('phone_number') user = CustomUser.objects.create_user( username=phone_number, email=email, phone_number=phone_number, **user_data, **business_data, role='borrower', status='active' ) # Handle file uploads if request.FILES.get('id_document'): user.id_document = request.FILES['id_document'] if request.FILES.get('selfie'): user.selfie = request.FILES['selfie'] if request.FILES.get('utility_bill'): user.utility_bill = request.FILES['utility_bill'] if request.FILES.get('bank_statement'): user.bank_statement = request.FILES['bank_statement'] user.save() # Create audit log AuditLog.objects.create( user=request.user, action='create', model_name='CustomUser', object_id=str(user.id), description=f'Created new client: {user.get_full_name()}' ) messages.success(request, f'Client {user.get_full_name()} registered successfully!') return redirect('users:client_list') except Exception as e: # Make error messages more user-friendly error_msg = str(e) if "Duplicate entry" in error_msg and "id_number" in error_msg: error_msg = "This ID number is already registered. Please check and try again." elif "Duplicate entry" in error_msg and "phone_number" in error_msg: error_msg = "This phone number is already registered. Please use a different phone number." elif "Duplicate entry" in error_msg and "email" in error_msg: error_msg = "This email address is already registered. Please use a different email address." messages.error(request, f'Unable to create client: {error_msg}') # Get system settings for file upload from utils.models import SystemSetting max_file_size_mb = SystemSetting.get_int('max_file_size_mb', 10) context = { 'business_types': [ ('retail', 'Retail'), ('agriculture', 'Agriculture'), ('transport', 'Transport'), ('manufacturing', 'Manufacturing'), ('services', 'Services'), ('technology', 'Technology'), ], 'genders': CustomUser._meta.get_field('gender').choices, 'max_file_size_mb': max_file_size_mb, } return render(request, 'users/client_create.html', context) @login_required def client_update(request, client_id): """Update client information""" client = get_object_or_404(CustomUser, pk=client_id, role='borrower') if request.method == 'POST': try: # Update fields for field in ['first_name', 'last_name', 'email', 'phone_number', 'date_of_birth', 'gender', 'id_number', 'marital_status', 'address', 'city', 'county', 'postal_code', 'business_name', 'business_type', 'employer']: if field in request.POST: setattr(client, field, request.POST[field]) # Handle monthly_income separately monthly_income = request.POST.get('monthly_income') if monthly_income: try: client.monthly_income = float(monthly_income) except ValueError: client.monthly_income = None else: client.monthly_income = None # Handle nationality field specifically if 'nationality' in request.POST: client.nationality = request.POST['nationality'] # Only update file fields if a new file is uploaded: for field in ['id_document', 'selfie', 'utility_bill', 'bank_statement']: if request.FILES.get(field): setattr(client, field, request.FILES[field]) client.save() # Invalidate cache for this client's popup data invalidate_client_popup_cache(client_id) # Create audit log AuditLog.objects.create( user=request.user, action='update', model_name='CustomUser', object_id=str(client.id), description=f'Updated client: {client.get_full_name()}' ) messages.success(request, 'Client information updated successfully!', extra_tags='client_update') return redirect('users:client_list') except Exception as e: messages.error(request, f'Error updating client: {str(e)}', extra_tags='client_update') # Get system settings for file upload from utils.models import SystemSetting max_file_size_mb = SystemSetting.get_int('max_file_size_mb', 10) context = { 'client': client, 'business_types': [ ('retail', 'Retail'), ('agriculture', 'Agriculture'), ('transport', 'Transport'), ('manufacturing', 'Manufacturing'), ('services', 'Services'), ('technology', 'Technology'), ], 'genders': CustomUser._meta.get_field('gender').choices, 'max_file_size_mb': max_file_size_mb, } return render(request, 'users/client_update.html', context) @login_required def client_delete(request, client_id): """Delete a client with option to blacklist or permanently delete""" client = get_object_or_404(CustomUser, pk=client_id, role='borrower') if request.method == 'POST': action = request.POST.get('action') if action == 'blacklist': # Blacklist the client (soft delete) client.status = 'blacklisted' client.save() # Invalidate cache for this client's popup data invalidate_client_popup_cache(client_id) # Create audit log AuditLog.objects.create( user=request.user, action='blacklist', model_name='CustomUser', object_id=str(client.id), description=f'Blacklisted client: {client.get_full_name()}' ) messages.success(request, f'Client {client.get_full_name()} has been blacklisted.', extra_tags='client_list') elif action == 'delete': # Permanently delete the client client_name = client.get_full_name() # Invalidate cache for this client's popup data before deletion invalidate_client_popup_cache(client_id) client.delete() # Create audit log AuditLog.objects.create( user=request.user, action='delete', model_name='CustomUser', object_id=str(client_id), description=f'Permanently deleted client: {client_name}' ) messages.success(request, f'Client {client_name} has been permanently deleted.', extra_tags='client_list') return redirect('users:client_list') return render(request, 'users/client_confirm_delete.html', {'client': client}) @login_required def client_export(request): """Export clients data to CSV/Excel""" # Implementation for data export pass @login_required def profile(request): """User profile view""" return render(request, 'users/profile.html') @login_required def edit_profile(request): """Edit user profile""" return render(request, 'users/edit_profile.html') @login_required def kyc_documents(request): """KYC documents management""" return render(request, 'users/kyc_documents.html') @login_required def upload_kyc(request): """Upload KYC documents""" return render(request, 'users/upload_kyc.html') # Admin Management Views @login_required @require_role('admin') def admin_list(request): """List all staff users - accessible by admins""" try: # Get filter parameters search = request.GET.get('search', '') status_filter = request.GET.get('status', 'all') # Base queryset - get all staff users (admin, team_leader, loan_officer, secretary) admins = CustomUser.objects.filter(is_staff=True).select_related() # Apply filters if search: admins = admins.filter( Q(first_name__icontains=search) | Q(last_name__icontains=search) | Q(email__icontains=search) | Q(phone_number__icontains=search) | Q(username__icontains=search) ) role_filter = request.GET.get('role', 'all') if role_filter != 'all': admins = admins.filter(role=role_filter) if status_filter != 'all': admins = admins.filter(status=status_filter) # Pagination paginator = Paginator(admins, 10) page_number = request.GET.get('page') page_obj = paginator.get_page(page_number) # Get statistics for all staff users all_staff = CustomUser.objects.filter(is_staff=True) context = { 'admins': page_obj, 'search': search, 'status_filter': status_filter, 'role_filter': role_filter, 'total_staff': all_staff.count(), 'admin_count': all_staff.filter(role='admin').count(), 'team_leader_count': all_staff.filter(role='team_leader').count(), 'loan_officer_count': all_staff.filter(role='loan_officer').count(), 'secretary_count': all_staff.filter(role='secretary').count(), } return render(request, 'users/admin_list.html', context) except Exception as e: print(f"Error in admin_list view: {str(e)}") return render(request, 'users/admin_list.html', { 'admins': [], 'search': '', 'status_filter': 'all', 'role_filter': 'all', 'total_staff': 0, 'admin_count': 0, 'team_leader_count': 0, 'loan_officer_count': 0, 'secretary_count': 0, 'error_message': str(e) }) @login_required @require_role('admin') def admin_create(request): """Create a new staff user with role-based permissions""" if request.method == 'POST': # Get form data username = request.POST.get('username') email = request.POST.get('email') phone_number = request.POST.get('phone_number') first_name = request.POST.get('first_name') last_name = request.POST.get('last_name') role = request.POST.get('role') password = request.POST.get('password') confirm_password = request.POST.get('confirm_password') status = request.POST.get('status', 'active') # Get custom permissions custom_permissions = {} for module in ['users', 'loans', 'reports', 'documents', 'settings', 'notifications', 'audit', 'kyc', 'payments', 'communications']: custom_permissions[module] = {} for action in ['view', 'create', 'edit', 'delete', 'approve', 'reject', 'export', 'import', 'manage']: custom_permissions[module][action] = request.POST.get(f'perm_{module}_{action}') == 'on' # Validate required fields if not all([username, email, phone_number, first_name, last_name, role, password]): messages.error(request, 'All fields are required.') return redirect('users:admin_create') # Validate password if password != confirm_password: messages.error(request, 'Passwords do not match.') return redirect('users:admin_create') # Check if username or email already exists if CustomUser.objects.filter(username=username).exists(): messages.error(request, 'Username already exists.') return redirect('users:admin_create') if CustomUser.objects.filter(email=email).exists(): messages.error(request, 'Email already exists.') return redirect('users:admin_create') if CustomUser.objects.filter(phone_number=phone_number).exists(): messages.error(request, 'Phone number already exists.') return redirect('users:admin_create') try: # Create user user = CustomUser.objects.create_user( username=username, email=email, phone_number=phone_number, first_name=first_name, last_name=last_name, role=role, password=password, status=status, is_staff=True, is_superuser=(role == 'admin'), is_phone_verified=True, is_email_verified=True, ) # Create custom permissions if they differ from defaults from users.models import RolePermission for module, actions in custom_permissions.items(): for action, is_allowed in actions.items(): # Check if this differs from default permissions try: default_perm = RolePermission.objects.get(role=role, module=module, action=action) if default_perm.is_allowed != is_allowed: # Update the permission default_perm.is_allowed = is_allowed default_perm.save() except RolePermission.DoesNotExist: # Create new permission RolePermission.objects.create( role=role, module=module, action=action, is_allowed=is_allowed ) # Log the creation request.user.log_access( action='create', module='users', object_type='CustomUser', object_id=str(user.id), description=f'Created new {role} user: {user.get_full_name()} with custom permissions', request=request ) messages.success(request, f'Successfully created {role} user: {user.get_full_name()}') return redirect('users:admin_list') except Exception as e: messages.error(request, f'Error creating user: {str(e)}') return redirect('users:admin_create') # Get available roles for staff members staff_roles = [role for role in CustomUser.ROLE_CHOICES if role[0] in ['admin', 'team_leader', 'loan_officer', 'secretary']] # Get default permissions for each role from users.models import RolePermission default_permissions = {} for role, _ in staff_roles: default_permissions[role] = {} for module in ['users', 'loans', 'reports', 'documents', 'settings', 'notifications', 'audit', 'kyc', 'payments', 'communications']: default_permissions[role][module] = {} for action in ['view', 'create', 'edit', 'delete', 'approve', 'reject', 'export', 'import', 'manage']: try: perm = RolePermission.objects.get(role=role, module=module, action=action) default_permissions[role][module][action] = perm.is_allowed except RolePermission.DoesNotExist: default_permissions[role][module][action] = False context = { 'staff_roles': staff_roles, 'default_permissions': default_permissions, 'status_choices': CustomUser.STATUS_CHOICES, } return render(request, 'users/admin_create.html', context) @login_required @require_role('admin') def admin_detail(request, admin_id): """Detailed view of a staff user""" admin = get_object_or_404(CustomUser, id=admin_id, is_staff=True) # Get access logs for this admin try: from .models import UserAccessLog audit_logs = UserAccessLog.objects.filter(user=admin).order_by('-accessed_at')[:10] except ImportError: audit_logs = [] context = { 'admin': admin, 'audit_logs': audit_logs, } return render(request, 'users/admin_detail.html', context) @login_required @require_role('admin') def admin_update(request, admin_id): """Update staff user information""" admin = get_object_or_404(CustomUser, id=admin_id, is_staff=True) if request.method == 'POST': try: # Update basic fields admin.first_name = request.POST.get('first_name') admin.last_name = request.POST.get('last_name') admin.email = request.POST.get('email') admin.phone_number = request.POST.get('phone_number') admin.status = request.POST.get('status') admin.is_staff = request.POST.get('is_staff') == 'on' admin.is_superuser = request.POST.get('is_superuser') == 'on' # Update password if provided new_password = request.POST.get('new_password') confirm_password = request.POST.get('confirm_password') if new_password: if new_password != confirm_password: messages.error(request, 'Passwords do not match.', extra_tags='admin_update') return redirect('users:admin_update', admin_id=admin_id) admin.set_password(new_password) admin.save() # Create access log request.user.log_access( action='update', module='users', object_type='CustomUser', object_id=str(admin.id), description=f'Updated staff user: {admin.get_full_name()}', request=request ) messages.success(request, 'Admin information updated successfully!', extra_tags='admin_list') return redirect('users:admin_list') except Exception as e: messages.error(request, f'Error updating admin: {str(e)}', extra_tags='admin_update') return redirect('users:admin_update', admin_id=admin_id) context = { 'admin': admin, 'status_choices': CustomUser.STATUS_CHOICES, } return render(request, 'users/admin_update.html', context) @login_required @require_role('admin') def admin_delete(request, pk): """Delete a staff user (soft delete)""" admin_user = get_object_or_404(CustomUser, pk=pk, is_staff=True) # Prevent deleting self if admin_user == request.user: messages.error(request, 'You cannot delete your own account.') return redirect('users:admin_list') if request.method == 'POST': admin_user.status = 'suspended' admin_user.is_active = False admin_user.save() # Create access log request.user.log_access( action='delete', module='users', object_type='CustomUser', object_id=str(admin_user.id), description=f'Suspended staff user: {admin_user.get_full_name()}', request=request ) messages.success(request, f'Admin {admin_user.get_full_name()} has been suspended.') return redirect('users:admin_list') return render(request, 'users/admin_confirm_delete.html', {'admin_user': admin_user}) @login_required @require_role('admin') def admin_reactivate(request, pk): """Reactivate a suspended staff user""" admin_user = get_object_or_404(CustomUser, pk=pk, is_staff=True) if request.method == 'POST': admin_user.status = 'active' admin_user.is_active = True admin_user.save() # Create access log request.user.log_access( action='update', module='users', object_type='CustomUser', object_id=str(admin_user.id), description=f'Reactivated staff user: {admin_user.get_full_name()}', request=request ) messages.success(request, f'Admin {admin_user.get_full_name()} has been reactivated.') return redirect('users:admin_list') return render(request, 'users/admin_confirm_reactivate.html', {'admin_user': admin_user}) @login_required def client_popup(request, client_id): """Get client data for popup modal - OPTIMIZED VERSION WITH CACHING""" try: # Check cache first for performance from django.core.cache import cache cache_key = f'client_popup_{client_id}' cached_data = cache.get(cache_key) if cached_data: return JsonResponse(cached_data) # Get client with select_related to optimize queries client = get_object_or_404(CustomUser, id=client_id, role='borrower') # OPTIMIZED: Use lightweight queries instead of comprehensive analytics from django.db.models import Sum, Count, Q from decimal import Decimal # Get basic loan statistics with optimized queries client_loans = Loan.objects.filter(borrower=client) # Single query for all loan statistics loan_stats = client_loans.aggregate( total_loans=Count('id'), active_loans=Count('id', filter=Q(status='active')), total_principal=Sum('principal_amount'), total_interest=Sum('interest_amount'), total_paid=Sum('amount_paid') ) # Calculate derived values total_borrowed = float(loan_stats['total_principal'] or 0) total_repaid = float(loan_stats['total_paid'] or 0) outstanding_amount = total_borrowed - total_repaid active_loans = loan_stats['active_loans'] or 0 # Simple credit score calculation (much faster than comprehensive analytics) credit_score = 0 if total_borrowed > 0: repayment_rate = (total_repaid / total_borrowed) * 100 if repayment_rate >= 90: credit_score = 85 elif repayment_rate >= 75: credit_score = 70 elif repayment_rate >= 50: credit_score = 55 else: credit_score = 30 # Get recent loans (limit to 5 for performance) recent_loans = client_loans.select_related('application__loan_product').order_by('-created_at')[:5] loans_data = [] ordinal_names = ['First', 'Second', 'Third', 'Fourth', 'Fifth'] for index, loan in enumerate(recent_loans): loan_name = ordinal_names[index] if index < len(ordinal_names) else f"{index + 1}th" loans_data.append({ 'loan_number': loan.loan_number, 'loan_name': f"{loan_name} Loan", 'product_name': loan.application.loan_product.name if loan.application and loan.application.loan_product else 'N/A', 'principal_amount': float(loan.principal_amount or 0), 'interest_amount': float(loan.interest_amount or 0), 'status': loan.status, 'status_display': loan.get_status_display() if hasattr(loan, 'get_status_display') else loan.status.title(), 'disbursement_date': loan.disbursement_date.strftime('%B %d, %Y') if loan.disbursement_date else 'N/A', 'due_date': loan.due_date.strftime('%B %d, %Y') if loan.due_date else 'N/A', 'amount_paid': float(loan.amount_paid or 0), 'outstanding_amount': float((loan.principal_amount or 0) - (loan.amount_paid or 0)) }) # Simple chart data (much faster than comprehensive analytics) status_counts = client_loans.values('status').annotate(count=Count('id')) chart_data = { 'loan_status_distribution': { 'labels': [item['status'].title() for item in status_counts], 'data': [item['count'] for item in status_counts], 'colors': ['#10B981', '#3B82F6', '#EF4444', '#F59E0B', '#8B5CF6'] }, 'repayment_trend': { 'labels': ['Last 3 Months', 'Last 6 Months', 'Total'], 'data': [total_repaid * 0.3, total_repaid * 0.6, total_repaid] # Simplified trend } } # Check documents (keep existing logic as it's fast) documents = { 'id_document': { 'uploaded': bool(client.id_document), 'url': client.id_document.url if client.id_document else None, 'name': 'National ID Document', 'description': 'Front and back scanned copies' }, 'selfie': { 'uploaded': bool(client.selfie), 'url': client.selfie.url if client.selfie else None, 'name': 'Selfie Photo', 'description': 'Recent photo for verification' }, 'utility_bill': { 'uploaded': bool(client.utility_bill), 'url': client.utility_bill.url if client.utility_bill else None, 'name': 'Utility Bill', 'description': 'Recent utility bill for address verification' }, 'bank_statement': { 'uploaded': bool(client.bank_statement), 'url': client.bank_statement.url if client.bank_statement else None, 'name': 'Bank Statement', 'description': 'Last 6 months bank statements' }, 'business_license': { 'uploaded': bool(client.business_license), 'url': client.business_license.url if client.business_license else None, 'name': 'Business License', 'description': 'Business registration and license documents' }, 'tax_certificate': { 'uploaded': bool(client.tax_certificate), 'url': client.tax_certificate.url if client.tax_certificate else None, 'name': 'Tax Certificate', 'description': 'Tax compliance certificates' }, 'logbook': { 'uploaded': bool(client.logbook), 'url': client.logbook.url if client.logbook else None, 'name': 'Logbook', 'description': 'Vehicle logbook documents' }, 'title_deed': { 'uploaded': bool(client.title_deed), 'url': client.title_deed.url if client.title_deed else None, 'name': 'Title Deed', 'description': 'Property title deed documents' }, 'signature': { 'uploaded': bool(client.signature), 'url': client.signature.url if client.signature else None, 'name': 'Digital Signature', 'description': 'Digital signature image' }, 'other_documents': { 'uploaded': bool(client.other_documents), 'url': client.other_documents.url if client.other_documents else None, 'name': 'Other Documents', 'description': 'Additional supporting documents' } } # Prepare client data (keep existing structure for compatibility) client_data = { # Basic Information 'id': str(client.id), 'full_name': client.get_full_name() if hasattr(client, 'get_full_name') else f"{getattr(client, 'first_name', '')} {getattr(client, 'last_name', '')}".strip(), 'first_name': getattr(client, 'first_name', '') or '', 'last_name': getattr(client, 'last_name', '') or '', 'email': getattr(client, 'email', 'N/A') or 'N/A', 'phone_number': getattr(client, 'phone_number', 'N/A') or 'N/A', 'id_number': getattr(client, 'id_number', 'N/A') or 'N/A', 'date_of_birth': client.date_of_birth.strftime('%B %d, %Y') if hasattr(client, 'date_of_birth') and client.date_of_birth else 'N/A', 'gender': client.get_gender_display() if hasattr(client, 'gender') and client.gender else 'N/A', 'marital_status': client.get_marital_status_display() if hasattr(client, 'marital_status') and client.marital_status else 'N/A', 'nationality': getattr(client, 'nationality', 'N/A') or 'N/A', 'country': getattr(client, 'country', 'N/A') or 'N/A', 'nickname': getattr(client, 'nickname', 'N/A') or 'N/A', 'place_of_birth': getattr(client, 'place_of_birth', 'N/A') or 'N/A', 'domicile': getattr(client, 'domicile', 'N/A') or 'N/A', 'cp_domicile': getattr(client, 'cp_domicile', 'N/A') or 'N/A', 'start_time': client.start_time.strftime('%I:%M %p') if hasattr(client, 'start_time') and client.start_time else 'N/A', 'registration_date': client.registration_date.strftime('%B %d, %Y') if hasattr(client, 'registration_date') and client.registration_date else 'N/A', 'physical_address': getattr(client, 'physical_address', 'N/A') or 'N/A', 'city': getattr(client, 'city', 'N/A') or 'N/A', 'county': getattr(client, 'county', 'N/A') or 'N/A', 'postal_code': getattr(client, 'postal_code', 'N/A') or 'N/A', 'postal_code_business': getattr(client, 'postal_code_business', 'N/A') or 'N/A', 'physical_location': getattr(client, 'physical_location', 'N/A') or 'N/A', 'status': getattr(client, 'status', 'unknown'), 'status_display': client.get_status_display() if hasattr(client, 'get_status_display') else getattr(client, 'status', 'Unknown').title(), 'created_at': client.created_at.strftime('%B %d, %Y') if hasattr(client, 'created_at') else 'N/A', # Business Information 'business_name': getattr(client, 'business_name', 'N/A') or 'N/A', 'business_type': getattr(client, 'business_type', 'N/A') or 'N/A', 'other_business_type': getattr(client, 'other_business_type', 'N/A') or 'N/A', 'business_address': getattr(client, 'business_address', 'N/A') or 'N/A', 'employer': getattr(client, 'employer', 'N/A') or 'N/A', 'monthly_income': float(getattr(client, 'monthly_income', 0)) if getattr(client, 'monthly_income', 0) else 0, 'capital_invested': float(getattr(client, 'capital_invested', 0)) if getattr(client, 'capital_invested', 0) else 0, 'source_of_funds': getattr(client, 'source_of_funds', 'N/A') or 'N/A', 'expected_turnover': float(getattr(client, 'expected_turnover', 0)) if getattr(client, 'expected_turnover', 0) else 0, # Reference Information 'declaration_name': getattr(client, 'declaration_name', 'N/A') or 'N/A', 'personal_pin': getattr(client, 'personal_pin', 'N/A') or 'N/A', 'recommender_name': getattr(client, 'recommender_name', 'N/A') or 'N/A', 'recommender_id': getattr(client, 'recommender_id', 'N/A') or 'N/A', 'recommender_tel': getattr(client, 'recommender_tel', 'N/A') or 'N/A', 'recommender_mobile': getattr(client, 'recommender_mobile', 'N/A') or 'N/A', 'recommender_residence': getattr(client, 'recommender_residence', 'N/A') or 'N/A', 'guarantor_name': getattr(client, 'guarantor_name', 'N/A') or 'N/A', 'guarantor_id': getattr(client, 'guarantor_id', 'N/A') or 'N/A', 'guarantor_tel': getattr(client, 'guarantor_tel', 'N/A') or 'N/A', 'guarantor_mobile': getattr(client, 'guarantor_mobile', 'N/A') or 'N/A', 'guarantor_residence': getattr(client, 'guarantor_residence', 'N/A') or 'N/A', # OPTIMIZED: Lightweight loan statistics 'credit_score': credit_score, 'total_borrowed': total_borrowed, 'total_repaid': total_repaid, 'active_loans': active_loans, 'outstanding_amount': outstanding_amount, 'loans': loans_data, 'documents': documents, # OPTIMIZED: Simplified analytics data 'analytics': { 'risk_analysis': { 'risk_level': 'Low' if credit_score >= 70 else 'Medium' if credit_score >= 50 else 'High', 'repayment_rate': round((total_repaid / total_borrowed * 100) if total_borrowed > 0 else 0, 2) }, 'profitability': { 'total_interest_earned': float(loan_stats['total_interest'] or 0), 'profit_margin': round((float(loan_stats['total_interest'] or 0) / total_borrowed * 100) if total_borrowed > 0 else 0, 2) }, 'operational_metrics': [ {'metric': 'Total Loans', 'value': loan_stats['total_loans'] or 0}, {'metric': 'Active Loans', 'value': active_loans}, {'metric': 'Repayment Rate', 'value': f"{round((total_repaid / total_borrowed * 100) if total_borrowed > 0 else 0, 1)}%"} ], 'recommendations': [ 'Monitor repayment patterns closely' if credit_score < 50 else 'Client shows good repayment history', 'Consider increasing loan limits' if credit_score >= 70 else 'Maintain current loan limits' ], 'executive_summary': f"Client has {loan_stats['total_loans'] or 0} loans with {active_loans} currently active. Total borrowed: ${total_borrowed:,.2f}, repaid: ${total_repaid:,.2f}.", 'portfolio_metrics': { 'total_loans': loan_stats['total_loans'] or 0, 'active_loans': active_loans, 'total_principal': total_borrowed, 'total_repaid': total_repaid }, 'portfolio_trends': { 'repayment_trend': 'Stable' if credit_score >= 60 else 'Needs attention', 'growth_rate': 'Positive' if total_repaid > total_borrowed * 0.8 else 'Moderate' }, 'revenue_metrics': [ {'metric': 'Interest Earned', 'value': f"${float(loan_stats['total_interest'] or 0):,.2f}"}, {'metric': 'Profit Margin', 'value': f"{round((float(loan_stats['total_interest'] or 0) / total_borrowed * 100) if total_borrowed > 0 else 0, 1)}%"} ], 'revenue_trends': { 'monthly_growth': 'Stable', 'profitability': 'Good' if float(loan_stats['total_interest'] or 0) > total_borrowed * 0.1 else 'Moderate' }, 'risk_metrics': [ {'metric': 'Credit Score', 'value': credit_score}, {'metric': 'Outstanding Amount', 'value': f"${outstanding_amount:,.2f}"}, {'metric': 'Risk Level', 'value': 'Low' if credit_score >= 70 else 'Medium' if credit_score >= 50 else 'High'} ], 'risk_factors': [ 'High outstanding amount' if outstanding_amount > total_borrowed * 0.5 else 'Manageable debt level', 'Low repayment rate' if credit_score < 50 else 'Good repayment history' ], }, # Charts 'chart_data': chart_data, # URLs 'update_url': request.build_absolute_uri(reverse('users:client_update', kwargs={'client_id': client.id})), 'delete_url': request.build_absolute_uri(reverse('users:client_delete', kwargs={'client_id': client.id})), 'report_url': request.build_absolute_uri(reverse('users:client_report', kwargs={'client_id': client.id})), 'download_report_url': request.build_absolute_uri(reverse('users:client_download_report', kwargs={'client_id': client.id})), } response_data = {'success': True, 'client': client_data} # Cache the result for 5 minutes (300 seconds) cache.set(cache_key, response_data, 300) return JsonResponse(response_data) except Exception as e: print(f"Error in client_popup view: {str(e)}") return JsonResponse({'success': False, 'error': str(e)}) @login_required @xframe_options_sameorigin def client_report(request, client_id): """Generate comprehensive client report""" try: client = get_object_or_404(CustomUser, id=client_id, role='borrower') # Check if this is a report generation request generate_report = request.GET.get('generate_report', False) download_report = request.GET.get('download', False) # Get loan statistics try: active_loans = Loan.objects.filter(borrower=client, status='active').count() total_loans = Loan.objects.filter(borrower=client).count() total_borrowed = Loan.objects.filter(borrower=client).aggregate( total=Sum('principal_amount') )['total'] or 0 total_repaid = Loan.objects.filter(borrower=client).aggregate( total=Sum('amount_paid') )['total'] or 0 outstanding_amount = total_borrowed - total_repaid # Get loan history loans = Loan.objects.filter(borrower=client).order_by('-created_at') # Get loan performance metrics total_on_time = loans.filter(status='paid', paid_on_time=True).count() total_late = loans.filter(status='paid', paid_on_time=False).count() total_defaulted = loans.filter(status='defaulted').count() # Calculate repayment performance if total_loans > 0: on_time_rate = (total_on_time / total_loans) * 100 late_rate = (total_late / total_loans) * 100 default_rate = (total_defaulted / total_loans) * 100 else: on_time_rate = late_rate = default_rate = 0 # Get monthly loan trends from datetime import datetime, timedelta from django.utils import timezone # Get repayment data for the last 6 months six_months_ago = timezone.now() - timedelta(days=180) repayments = Repayment.objects.filter( loan__borrower=client, payment_date__gte=six_months_ago ).order_by('payment_date') # Group repayments by month monthly_repayments = {} for repayment in repayments: month_key = repayment.payment_date.strftime('%B %Y') if month_key not in monthly_repayments: monthly_repayments[month_key] = 0 monthly_repayments[month_key] += float(repayment.amount) chart_data = { 'repayment_trend': { 'labels': list(monthly_repayments.keys()), 'data': list(monthly_repayments.values()) } } except Exception as e: active_loans = total_loans = 0 total_borrowed = total_repaid = outstanding_amount = 0 on_time_rate = late_rate = default_rate = 0 loans = [] monthly_repayments = {} print(f"Error calculating loan stats for client {client.id}: {str(e)}") # Get credit score try: credit_score = LoanScoring.objects.get(user=client) credit_score_value = credit_score.total_score credit_score_details = { 'payment_history': credit_score.repayment_history_score, 'income_score': credit_score.income_score, 'rollover_frequency': credit_score.rollover_frequency_score, 'employment_stability': credit_score.employment_stability_score, 'total_score': credit_score.total_score } except LoanScoring.DoesNotExist: credit_score_value = 0 credit_score_details = {} # Get document status documents = { 'id_document': bool(client.id_document), 'selfie': bool(client.selfie), 'utility_bill': bool(client.utility_bill), 'bank_statement': bool(client.bank_statement), 'logbook': bool(client.logbook), 'title_deed': bool(client.title_deed), 'signature': bool(client.signature), } # Get recent activities activities = AuditLog.objects.filter( Q(user=client) | Q(object_id=str(client.id), model_name='CustomUser') ).order_by('-created_at')[:10] context = { 'client': client, 'loans': loans, 'loan_stats': { 'active_loans': active_loans, 'total_loans': total_loans, 'total_borrowed': total_borrowed, 'total_repaid': total_repaid, 'outstanding_amount': outstanding_amount, 'on_time_rate': on_time_rate, 'late_rate': late_rate, 'default_rate': default_rate, }, 'monthly_trends': monthly_repayments, # Use monthly_repayments for the chart 'credit_score': credit_score_value, 'credit_score_details': credit_score_details, 'documents': documents, 'activities': activities, 'report_date': timezone.now(), 'is_report': generate_report, } if download_report: # Generate PDF response for download response = HttpResponse(content_type='application/pdf') response['Content-Disposition'] = f'attachment; filename="client_report_{client.id}_{timezone.now().strftime("%Y%m%d_%H%M%S")}.pdf"' # For now, return a simple text response indicating PDF generation # In a real implementation, you would use a library like WeasyPrint or ReportLab response.write(f"Client Report for {client.get_full_name()}\n") response.write(f"Generated on: {timezone.now().strftime('%B %d, %Y at %I:%M %p')}\n\n") response.write("This is a placeholder for the PDF report.\n") response.write("In a production environment, this would generate a comprehensive PDF report.\n") return response elif generate_report: return render(request, 'users/client_report_print.html', context) else: return render(request, 'users/client_report.html', context) except Exception as e: messages.error(request, f'Error generating report: {str(e)}') return redirect('users:client_list') @login_required def client_download_report(request, client_id): """Download client report as PDF""" try: client = get_object_or_404(CustomUser, id=client_id, role='borrower') # Get loan statistics try: active_loans = Loan.objects.filter(borrower=client, status='active').count() total_loans = Loan.objects.filter(borrower=client).count() total_borrowed = Loan.objects.filter(borrower=client).aggregate( total=Sum('principal_amount') )['total'] or 0 total_repaid = Loan.objects.filter(borrower=client).aggregate( total=Sum('amount_paid') )['total'] or 0 outstanding_amount = total_borrowed - total_repaid # Get loan history loans = Loan.objects.filter(borrower=client).order_by('-created_at') except Exception as e: active_loans = 0 total_loans = 0 total_borrowed = 0 total_repaid = 0 outstanding_amount = 0 loans = [] print(f"Error calculating loan stats for client {client.id}: {str(e)}") # Get credit score try: credit_score = LoanScoring.objects.get(user=client) credit_score_value = credit_score.total_score credit_score_details = { 'payment_history': credit_score.repayment_history_score, 'income_score': credit_score.income_score, 'rollover_frequency': credit_score.rollover_frequency_score, 'employment_stability': credit_score.employment_stability_score, 'total_score': credit_score.total_score } except LoanScoring.DoesNotExist: credit_score_value = 0 credit_score_details = {} # Get document status documents = { 'id_document': bool(client.id_document), 'selfie': bool(client.selfie), 'utility_bill': bool(client.utility_bill), 'bank_statement': bool(client.bank_statement), 'logbook': bool(client.logbook), 'title_deed': bool(client.title_deed), 'signature': bool(client.signature), } context = { 'client': client, 'loans': loans, 'loan_stats': { 'active_loans': active_loans, 'total_loans': total_loans, 'total_borrowed': total_borrowed, 'total_repaid': total_repaid, 'outstanding_amount': outstanding_amount, }, 'credit_score': credit_score_value, 'credit_score_details': credit_score_details, 'documents': documents, 'report_date': timezone.now(), } # Generate PDF response response = HttpResponse(content_type='application/pdf') response['Content-Disposition'] = f'attachment; filename="client_report_{client.id}_{timezone.now().strftime("%Y%m%d_%H%M%S")}.pdf"' # For now, return a simple text response indicating PDF generation # In a real implementation, you would use a library like WeasyPrint or ReportLab response.write(f"Client Report for {client.get_full_name()}\n") response.write(f"Generated on: {timezone.now().strftime('%B %d, %Y at %I:%M %p')}\n\n") response.write("This is a placeholder for the PDF report.\n") response.write("In a production environment, this would generate a comprehensive PDF report.\n") return response except Exception as e: messages.error(request, f'Error downloading report: {str(e)}') return redirect('users:client_list') @login_required def client_documents(request, client_id): """View for managing client's KYC documents""" client = get_object_or_404(CustomUser, id=client_id) # Check permissions if not request.user.is_staff and request.user != client: messages.error(request, 'You do not have permission to view these documents') return redirect('users:client_list') # Get documents by type with prefetch for optimization documents = Document.objects.filter( Q(uploaded_by=client) | Q(shared_with=client) ).select_related('uploaded_by').prefetch_related('tags') context = { 'client': client, 'id_documents': documents.filter(document_type='id_document').order_by('-created_at'), 'selfie_documents': documents.filter(document_type='selfie').order_by('-created_at'), 'utility_documents': documents.filter(document_type='utility_bill').order_by('-created_at'), 'bank_documents': documents.filter(document_type='bank_statement').order_by('-created_at'), } return render(request, 'users/client_documents.html', context) @login_required def upload_client_document(request, client_id): """API endpoint for uploading client documents""" if request.method == 'POST': client = get_object_or_404(CustomUser, id=client_id) # Check permissions if not request.user.is_staff and request.user != client: return JsonResponse({ 'status': 'error', 'error': 'Permission denied' }) try: document_type = request.POST.get('document_type') description = request.POST.get('description', '') file = request.FILES.get('file') if not all([document_type, file]): return JsonResponse({ 'status': 'error', 'error': 'Missing required fields' }) # Validate file type allowed_types = ['image/jpeg', 'image/png', 'image/gif', 'application/pdf'] file_type = magic.from_buffer(file.read(1024), mime=True) file.seek(0) # Reset file pointer after reading if file_type not in allowed_types: return JsonResponse({ 'status': 'error', 'error': f'Invalid file type. Allowed types: {", ".join(allowed_types)}' }) # Create document document = Document.objects.create( name=file.name, file=file, document_type=document_type, description=description, uploaded_by=client, mime_type=file_type ) # Generate thumbnail for images if file_type.startswith('image/'): try: img = Image.open(file) img.thumbnail((300, 300)) thumb_io = BytesIO() img.save(thumb_io, format='JPEG', quality=85) thumb_name = f'thumb_{file.name}' document.thumbnail.save(thumb_name, ContentFile(thumb_io.getvalue()), save=True) except Exception as e: print(f"Error generating thumbnail: {e}") # Create audit log AuditLog.objects.create( user=request.user, action='create', model_name='Document', object_id=str(document.id), description=f'Uploaded {document_type} document for client {client.get_full_name()}' ) # Create notification for admins if staff uploaded (only if notifications are enabled) if request.user.is_staff: # Check if notifications are enabled from utils.models import SystemSetting if SystemSetting.get_bool('email_notifications', True): Notification.objects.create( title='New Client Document', message=f'Staff member {request.user.get_full_name()} uploaded a {document_type} for {client.get_full_name()}', priority='medium', user=client ) return JsonResponse({'status': 'success'}) except Exception as e: return JsonResponse({ 'status': 'error', 'error': str(e) }) return JsonResponse({ 'status': 'error', 'error': 'Invalid request method' }) @login_required @require_role('admin') def admin_confirm_delete(request, admin_id): """View for confirming admin deletion""" admin = get_object_or_404(CustomUser, id=admin_id, is_staff=True) if request.method == 'POST': # Create audit log AuditLog.objects.create( user=request.user, action='delete', model_name='CustomUser', object_id=str(admin.id), description=f'Deactivated admin user: {admin.get_full_name()}' ) # Deactivate admin admin.is_active = False admin.save() messages.success(request, f'Admin {admin.get_full_name()} has been deactivated') return redirect('users:admin_list') return render(request, 'users/admin_confirm_delete.html', {'admin': admin}) @login_required @require_role('admin') def admin_confirm_reactivate(request, admin_id): """View for confirming admin reactivation""" admin = get_object_or_404(CustomUser, id=admin_id, is_staff=True) if request.method == 'POST': # Create audit log AuditLog.objects.create( user=request.user, action='update', model_name='CustomUser', object_id=str(admin.id), description=f'Reactivated admin user: {admin.get_full_name()}' ) # Reactivate admin admin.is_active = True admin.save() messages.success(request, f'Admin {admin.get_full_name()} has been reactivated') return redirect('users:admin_list') return render(request, 'users/admin_confirm_reactivate.html', {'admin': admin}) @login_required @require_role('admin') def user_access_logs(request): """View user access logs and audit trails""" # Get filter parameters user_id = request.GET.get('user_id') action = request.GET.get('action') module = request.GET.get('module') date_from = request.GET.get('date_from') date_to = request.GET.get('date_to') # Build query from users.models import UserAccessLog logs = UserAccessLog.objects.select_related('user').all() if user_id: logs = logs.filter(user_id=user_id) if action: logs = logs.filter(action__icontains=action) if module: logs = logs.filter(module__icontains=module) if date_from: logs = logs.filter(accessed_at__date__gte=date_from) if date_to: logs = logs.filter(accessed_at__date__lte=date_to) # Pagination paginator = Paginator(logs.order_by('-accessed_at'), 50) page_number = request.GET.get('page') page_obj = paginator.get_page(page_number) # Get users for filter dropdown users = CustomUser.objects.filter(is_staff=True).order_by('first_name', 'last_name') # Get unique actions and modules for filter dropdowns actions = UserAccessLog.objects.values_list('action', flat=True).distinct().order_by('action') modules = UserAccessLog.objects.values_list('module', flat=True).distinct().order_by('module') context = { 'page_obj': page_obj, 'users': users, 'actions': actions, 'modules': modules, 'filters': { 'user_id': user_id, 'action': action, 'module': module, 'date_from': date_from, 'date_to': date_to, } } return render(request, 'users/access_logs.html', context) @login_required @require_role('admin') def user_permissions(request, user_id): """Manage user permissions""" user = get_object_or_404(CustomUser, id=user_id) if request.method == 'POST': # Update permissions from users.models import RolePermission for module in ['users', 'loans', 'reports', 'documents', 'settings', 'notifications', 'audit', 'kyc', 'payments', 'communications']: for action in ['view', 'create', 'edit', 'delete', 'approve', 'reject', 'export', 'import', 'manage']: is_allowed = request.POST.get(f'perm_{module}_{action}') == 'on' # Update or create permission perm, created = RolePermission.objects.get_or_create( role=user.role, module=module, action=action, defaults={'is_allowed': is_allowed} ) if not created and perm.is_allowed != is_allowed: perm.is_allowed = is_allowed perm.save() # Log the permission update request.user.log_access( action='update_permissions', module='users', object_type='CustomUser', object_id=str(user.id), description=f'Updated permissions for {user.get_full_name()}', request=request ) messages.success(request, f'Permissions updated for {user.get_full_name()}') return redirect('users:admin_list') # Get current permissions from users.models import RolePermission permissions = {} for module in ['users', 'loans', 'reports', 'documents', 'settings', 'notifications', 'audit', 'kyc', 'payments', 'communications']: permissions[module] = {} for action in ['view', 'create', 'edit', 'delete', 'approve', 'reject', 'export', 'import', 'manage']: try: perm = RolePermission.objects.get(role=user.role, module=module, action=action) permissions[module][action] = perm.is_allowed except RolePermission.DoesNotExist: permissions[module][action] = False context = { 'user': user, 'permissions': permissions, } return render(request, 'users/user_permissions.html', context) @login_required def client_list_ajax(request): """AJAX endpoint for client list with filters""" try: # Get filter parameters search = request.GET.get('search', '').strip() status_filter = request.GET.get('status', 'all') business_type_filter = request.GET.get('business_type', 'all') sort_by = request.GET.get('sort_by', 'first_name') sort_order = request.GET.get('sort_order', 'asc') # Start with all clients clients = CustomUser.objects.filter(role='borrower') # Apply search filter if search: clients = clients.filter( Q(first_name__icontains=search) | Q(last_name__icontains=search) | Q(email__icontains=search) | Q(phone_number__icontains=search) | Q(business_name__icontains=search) | Q(id_number__icontains=search) ) # Apply status filter if status_filter != 'all': clients = clients.filter(status=status_filter) # Apply business type filter if business_type_filter != 'all': clients = clients.filter(business_type=business_type_filter) # Apply sorting if sort_by == 'first_name': if sort_order == 'desc': clients = clients.order_by('-first_name', '-last_name') else: clients = clients.order_by('first_name', 'last_name') elif sort_by == 'last_name': if sort_order == 'desc': clients = clients.order_by('-last_name', '-first_name') else: clients = clients.order_by('last_name', 'first_name') elif sort_by == 'business_name': if sort_order == 'desc': clients = clients.order_by('-business_name') else: clients = clients.order_by('business_name') elif sort_by == 'status': if sort_order == 'desc': clients = clients.order_by('-status') else: clients = clients.order_by('status') elif sort_by == 'created_at': if sort_order == 'desc': clients = clients.order_by('-created_at') else: clients = clients.order_by('created_at') else: # Default sorting by first name ascending clients = clients.order_by('first_name', 'last_name') # Annotate with loan statistics clients = clients.annotate( active_loans=Count('loan', filter=Q(loan__status='active')), total_borrowed=Coalesce(Sum('loan__principal_amount'), 0), total_repaid=Coalesce(Sum('loan__amount_paid'), 0) ) # Render the table HTML from django.template.loader import render_to_string html = render_to_string('users/client_list_table.html', { 'clients': clients, 'request': request }, request=request) return JsonResponse({ 'success': True, 'html': html, 'count': clients.count() }) except Exception as e: return JsonResponse({ 'success': False, 'error': str(e) }) @login_required def api_search_clients(request): """API endpoint for searching clients""" from django.http import JsonResponse query = request.GET.get('q', '').strip() # Get all active clients clients = CustomUser.objects.filter( role='borrower', is_active=True ).order_by('first_name', 'last_name') # If there's a search query, filter the results if query: if len(query) < 2: return JsonResponse({'clients': []}) clients = clients.filter( Q(first_name__icontains=query) | Q(last_name__icontains=query) | Q(email__icontains=query) | Q(phone_number__icontains=query) | Q(id_number__icontains=query) )[:10] # Limit search results to 10 else: # Return all clients (limit to 100 for performance) clients = clients[:100] # Format response clients_data = [] for client in clients: clients_data.append({ 'id': str(client.id), 'full_name': client.get_full_name(), 'email': client.email, 'phone_number': client.phone_number, 'id_number': client.id_number, }) return JsonResponse({'clients': clients_data})