from django.shortcuts import render, get_object_or_404, redirect
from django.contrib.auth.decorators import login_required, user_passes_test
from users.decorators import admin_required, staff_required, portfolio_access_required
from django.http import HttpResponse, JsonResponse, Http404
from django.contrib import messages
from django.utils import timezone as django_timezone
from django.db.models import Sum, Count, Q, F
from django.core.files.storage import default_storage
from django.core.files.base import ContentFile
from django.conf import settings
from .models import AuditLog, SystemSetting, Document, DocumentTag, DocumentShare, Notification
from .forms import DocumentUploadForm
from .reports import OptimizedReportGenerator as ReportGenerator
from loans.models import Loan, LoanApplication, Repayment
from users.models import CustomUser
from reports.models import LoanScoring
import json
import os
import mimetypes
from PIL import Image
from io import BytesIO
# Try to import magic, but make it optional
try:
import magic
MAGIC_AVAILABLE = True
except (ImportError, OSError):
MAGIC_AVAILABLE = False
magic = None
import zipfile
from datetime import datetime, timedelta
from django.contrib.auth import get_user_model
from django.core.exceptions import PermissionDenied
from pathlib import Path
from django.core.paginator import Paginator
from utils.filtering import (
apply_branch_and_portfolio_filters,
get_filtered_clients,
get_filtered_loans,
get_filtered_documents,
get_filtered_notifications
)
User = get_user_model()
def is_admin(user):
return user.is_staff
@login_required
@portfolio_access_required
def documents(request):
# Check if user has permission to access documents
if not request.user.has_permission('documents', 'access'):
messages.error(request, 'You do not have permission to access documents.')
return redirect('dashboard')
user = request.user
# Handle document upload
if request.method == 'POST':
if 'upload_document' in request.POST:
form = DocumentUploadForm(request.POST, request.FILES)
if form.is_valid():
document = form.save(commit=False)
document.uploaded_by = user
document.save()
# Handle tags
tags = request.POST.get('tags', '').split(',')
for tag_name in tags:
tag_name = tag_name.strip()
if tag_name:
tag, created = DocumentTag.objects.get_or_create(
name=tag_name,
defaults={'created_by': user}
)
document.tags.add(tag)
# Create audit log
AuditLog.objects.create(
user=user,
action='create',
model_name='Document',
object_id=str(document.id),
description=f'Uploaded document: {document.name}'
)
messages.success(request, 'Document uploaded successfully!')
return redirect('documents')
else:
form = DocumentUploadForm()
# BRANCH FILTERING IMPLEMENTATION - COPY FROM CUSTOMER DOCUMENTS
from users.models import CustomUser
# Get selected branch from session
selected_branch_id = request.session.get('selected_branch_id')
# Step 1: Get Document model documents with branch filtering
documents_queryset = Document.objects.all()
# Apply branch filtering to Document model documents
if selected_branch_id:
documents_queryset = documents_queryset.filter(
Q(uploaded_by__branch_id=selected_branch_id) |
Q(is_public=True) |
Q(shared_with=user)
)
# Step 2: Get KYC documents from CustomUser model with branch filtering
clients = CustomUser.objects.filter(role='borrower').order_by('first_name', 'last_name')
# Apply portfolio-based access control for staff members
if user.role in ['loan_officer', 'team_leader'] and not user.is_superuser:
# Staff members can only see documents from their assigned borrowers
clients = clients.filter(portfolio_manager=user)
elif user.role in ['secretary', 'auditor'] and not user.is_superuser:
# Secretaries and auditors can only see documents from borrowers in their branch
if user.branch:
clients = clients.filter(branch=user.branch)
# Apply branch filtering to clients if a branch is selected
if selected_branch_id:
clients = clients.filter(branch_id=selected_branch_id)
# Step 3: Combine both sources into a unified structure
documents_by_type = {}
# Add Document model documents
for doc in documents_queryset:
if doc.document_type not in documents_by_type:
documents_by_type[doc.document_type] = []
documents_by_type[doc.document_type].append(doc)
# Add KYC documents from CustomUser model
for client in clients:
# ID Documents
if client.id_document:
if 'id_document' not in documents_by_type:
documents_by_type['id_document'] = []
# Create a document-like object for KYC documents
kyc_doc = type('KYCDocument', (), {
'id': f'kyc_id_{client.id}',
'name': f'ID Document - {client.get_full_name()}',
'file': client.id_document,
'created_at': client.created_at,
'document_type': 'id_document',
'source': 'kyc',
'uploaded_by': client,
'description': 'KYC Verification',
'tags': type('Tags', (), {'all': lambda: []})()
})()
documents_by_type['id_document'].append(kyc_doc)
# Selfie Documents
if client.selfie:
if 'selfie' not in documents_by_type:
documents_by_type['selfie'] = []
kyc_doc = type('KYCDocument', (), {
'id': f'kyc_selfie_{client.id}',
'name': f'Selfie Photo - {client.get_full_name()}',
'file': client.selfie,
'created_at': client.created_at,
'document_type': 'selfie',
'source': 'kyc',
'uploaded_by': client,
'description': 'KYC Verification',
'tags': type('Tags', (), {'all': lambda: []})()
})()
documents_by_type['selfie'].append(kyc_doc)
# Utility Bill Documents
if client.utility_bill:
if 'utility_bill' not in documents_by_type:
documents_by_type['utility_bill'] = []
kyc_doc = type('KYCDocument', (), {
'id': f'kyc_utility_{client.id}',
'name': f'Utility Bill - {client.get_full_name()}',
'file': client.utility_bill,
'created_at': client.created_at,
'document_type': 'utility_bill',
'source': 'kyc',
'uploaded_by': client,
'description': 'KYC Verification',
'tags': type('Tags', (), {'all': lambda: []})()
})()
documents_by_type['utility_bill'].append(kyc_doc)
# Bank Statement Documents
if client.bank_statement:
if 'bank_statement' not in documents_by_type:
documents_by_type['bank_statement'] = []
kyc_doc = type('KYCDocument', (), {
'id': f'kyc_bank_{client.id}',
'name': f'Bank Statement - {client.get_full_name()}',
'file': client.bank_statement,
'created_at': client.created_at,
'document_type': 'bank_statement',
'source': 'kyc',
'uploaded_by': client,
'description': 'KYC Verification',
'tags': type('Tags', (), {'all': lambda: []})()
})()
documents_by_type['bank_statement'].append(kyc_doc)
# Business License Documents
if client.business_license:
if 'business_license' not in documents_by_type:
documents_by_type['business_license'] = []
kyc_doc = type('KYCDocument', (), {
'id': f'kyc_business_{client.id}',
'name': f'Business License - {client.get_full_name()}',
'file': client.business_license,
'created_at': client.created_at,
'document_type': 'business_license',
'source': 'kyc',
'uploaded_by': client,
'description': 'KYC Verification',
'tags': type('Tags', (), {'all': lambda: []})()
})()
documents_by_type['business_license'].append(kyc_doc)
# Step 4: Apply search filters
search_query = request.GET.get('search', '')
document_type = request.GET.get('type', '')
tag_filter = request.GET.get('tag', '')
if search_query:
filtered_documents = {}
for doc_type, docs in documents_by_type.items():
filtered_docs = []
for doc in docs:
if (search_query.lower() in doc.name.lower() or
(hasattr(doc, 'description') and doc.description and search_query.lower() in doc.description.lower())):
filtered_docs.append(doc)
if filtered_docs:
filtered_documents[doc_type] = filtered_docs
documents_by_type = filtered_documents
if document_type:
if document_type in documents_by_type:
documents_by_type = {document_type: documents_by_type[document_type]}
else:
documents_by_type = {}
# Step 5: Calculate total documents
total_documents = sum(len(docs) for docs in documents_by_type.values())
# Step 6: Get template data
available_tags = DocumentTag.objects.all()
document_types = Document.DOCUMENT_TYPES
# Add sort options
sort_options = [
('name', 'Name A-Z'),
('-name', 'Name Z-A'),
('-created_at', 'Newest First'),
('created_at', 'Oldest First'),
]
context = {
'documents': documents_by_type,
'form': form,
'available_tags': available_tags,
'document_types': document_types,
'search_query': search_query,
'selected_type': document_type,
'selected_tag': tag_filter,
'total_documents': total_documents,
'sort_options': sort_options,
'current_sort': request.GET.get('sort', '-created_at'),
}
return render(request, 'utils/documents.html', context)
@login_required
def share_document(request):
"""API endpoint to share documents with users"""
# Check if user has permission to share documents
if not request.user.has_permission('documents', 'share'):
return JsonResponse({'success': False, 'error': 'You do not have permission to share documents.'}, status=403)
if request.method == 'POST':
try:
data = json.loads(request.body)
document_ids = data['document_id'].split(',')
user_ids = data['users']
message = data.get('message', '')
for doc_id in document_ids:
document = get_object_or_404(Document, id=doc_id)
# Check if user has permission to share
if document.uploaded_by != request.user and not request.user.is_staff:
return JsonResponse({
'status': 'error',
'error': 'You do not have permission to share this document'
})
# Share with selected users
for user_id in user_ids:
user = get_object_or_404(CustomUser, id=user_id)
DocumentShare.objects.get_or_create(
document=document,
shared_with=user,
defaults={
'shared_by': request.user,
'message': message
}
)
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
def get_document_preview(request, document_id):
"""Get document preview or thumbnail"""
document = get_object_or_404(
Document,
Q(id=document_id) &
(Q(uploaded_by=request.user) | Q(shared_with=request.user) | Q(is_public=True))
)
if document.thumbnail:
return JsonResponse({
'type': 'image',
'url': document.thumbnail.url
})
elif document.mime_type.startswith('image/'):
return JsonResponse({
'type': 'image',
'url': document.file.url
})
else:
return JsonResponse({
'type': 'other',
'mime_type': document.mime_type
})
@login_required
def get_document_tags(request):
"""API endpoint to get all document tags"""
tags = DocumentTag.objects.all().values('id', 'name')
return JsonResponse({'tags': list(tags)})
@login_required
def add_document_tags(request, document_id):
"""API endpoint to add tags to a document"""
if request.method == 'POST':
try:
document = get_object_or_404(Document, id=document_id)
data = json.loads(request.body)
tags = data.get('tags', [])
# Check permission
if document.uploaded_by != request.user and not request.user.is_staff:
return JsonResponse({
'status': 'error',
'error': 'Permission denied'
})
# Process tags
tag_objects = []
for tag_name in tags:
tag, created = DocumentTag.objects.get_or_create(
name=tag_name.strip(),
defaults={'created_by': request.user}
)
tag_objects.append(tag)
# Add tags to document
document.tags.add(*tag_objects)
return JsonResponse({
'status': 'success',
'tags': list(document.tags.values('id', 'name'))
})
except Exception as e:
return JsonResponse({
'status': 'error',
'error': str(e)
})
return JsonResponse({
'status': 'error',
'error': 'Invalid request method'
})
@login_required
def remove_document_tag(request, document_id, tag_id):
"""API endpoint to remove a tag from a document"""
if request.method == 'POST':
try:
document = get_object_or_404(Document, id=document_id)
tag = get_object_or_404(DocumentTag, id=tag_id)
# Check permission
if document.uploaded_by != request.user and not request.user.is_staff:
return JsonResponse({
'status': 'error',
'error': 'Permission denied'
})
document.tags.remove(tag)
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
def bulk_download_documents(request):
"""Download multiple documents as a zip file"""
# Check if user has permission to download documents
if not request.user.has_permission('documents', 'download'):
messages.error(request, 'You do not have permission to download documents.')
return redirect('documents')
if request.method == 'POST':
try:
data = json.loads(request.body)
document_ids = data.get('document_ids', [])
# Create a temporary directory for the zip file
temp_dir = os.path.join(settings.MEDIA_ROOT, 'temp')
os.makedirs(temp_dir, exist_ok=True)
# Create zip file
from utils.datetime_utils import get_current_datetime
zip_filename = f'documents_{get_current_datetime().strftime("%Y%m%d_%H%M%S")}.zip'
zip_path = os.path.join(temp_dir, zip_filename)
# Add files to zip
with zipfile.ZipFile(zip_path, 'w') as zip_file:
for doc_id in document_ids:
document = get_object_or_404(
Document,
Q(id=doc_id) &
(Q(uploaded_by=request.user) | Q(shared_with=request.user) | Q(is_public=True))
)
zip_file.write(document.file.path, document.name)
# Send file
with open(zip_path, 'rb') as f:
response = HttpResponse(f.read(), content_type='application/zip')
response['Content-Disposition'] = f'attachment; filename="{zip_filename}"'
# Clean up
os.remove(zip_path)
return response
except Exception as e:
return JsonResponse({
'status': 'error',
'error': str(e)
})
return JsonResponse({
'status': 'error',
'error': 'Invalid request method'
})
@login_required
@portfolio_access_required
def notifications(request):
# Check if user has permission to access notifications
if not request.user.has_permission('notifications', 'access'):
messages.error(request, 'You do not have permission to access notifications.')
return redirect('dashboard')
user = request.user
# BRANCH FILTERING IMPLEMENTATION - COPY FROM CUSTOMER DOCUMENTS
# Step 1: Get base queryset
notifications = Notification.objects.all()
# Step 2: Apply portfolio-based access control for staff members
if user.role in ['loan_officer', 'team_leader'] and not user.is_superuser:
# Staff members can only see notifications related to their assigned borrowers
# This includes notifications for their assigned clients, their loan applications, and system notifications
notifications = notifications.filter(
Q(user__portfolio_manager=user) | # Direct user notifications
Q(loan_app__borrower__portfolio_manager=user) | # Loan application notifications
Q(related_loan__borrower__portfolio_manager=user) | # Loan notifications
Q(user__isnull=True) # System-wide notifications
)
elif user.role in ['secretary', 'auditor'] and not user.is_superuser:
# Secretaries and auditors can only see notifications from users in their branch
if user.branch:
notifications = notifications.filter(
Q(user__branch=user.branch) | # Direct user notifications
Q(loan_app__borrower__branch=user.branch) | # Loan application notifications
Q(related_loan__borrower__branch=user.branch) | # Loan notifications
Q(user__isnull=True) # System-wide notifications
)
else:
# If no branch assigned, only show system notifications
notifications = notifications.filter(user__isnull=True)
else:
# Admin and superuser can see all notifications
# Apply branch filtering if a branch is selected (only for admin users)
selected_branch_id = request.session.get('selected_branch_id')
if selected_branch_id:
notifications = notifications.filter(
Q(user__branch_id=selected_branch_id) | Q(user__isnull=True)
)
# Step 3: Apply search and filters
search_query = request.GET.get('search', '')
notification_type = request.GET.get('type', '')
priority = request.GET.get('priority', '')
read_status = request.GET.get('read', '')
if search_query:
notifications = notifications.filter(
Q(title__icontains=search_query) |
Q(message__icontains=search_query)
)
if notification_type:
notifications = notifications.filter(notification_type=notification_type)
if priority:
notifications = notifications.filter(priority=priority)
if read_status == 'read':
notifications = notifications.filter(read_at__isnull=False)
elif read_status == 'unread':
notifications = notifications.filter(read_at__isnull=True)
# Step 4: Order and paginate
notifications = notifications.order_by('-created_at')
paginator = Paginator(notifications, 20)
page = request.GET.get('page')
notifications_page = paginator.get_page(page)
# Get filter options
notification_types = Notification.NOTIFICATION_TYPES
priority_levels = Notification.PRIORITY_LEVELS
# Get system alerts (notifications without specific user)
system_alerts_queryset = Notification.objects.filter(user__isnull=True).order_by('-created_at')
system_alerts = system_alerts_queryset[:10]
# Get security alerts (high priority system alerts) - filter before slicing
security_alerts = system_alerts_queryset.filter(priority__in=['urgent', 'high'])[:5]
# Get recent activity from audit logs
recent_activities = []
try:
recent_activities = AuditLog.objects.select_related('user').order_by('-created_at')[:10]
except:
pass # In case AuditLog is not available
# Additional context for admin users
additional_context = {}
if user.is_staff:
from loans.models import Loan, LoanApplication
# Apply branch filtering for non-superuser staff
branch_filter = {}
if not user.is_superuser and hasattr(user, 'branch') and user.branch:
branch_filter = {'borrower__branch': user.branch}
# Get statistics for admin dashboard - exclude rolled-over loans
additional_context.update({
'is_admin': True,
'loan_statistics': {
'total_disbursed': Loan.objects.filter(is_deleted=False, is_rolled_over=False, **branch_filter).exclude(status='rolled_over').aggregate(total=Sum('principal_amount'))['total'] or 0,
'active_loans': Loan.objects.filter(status='active', is_deleted=False, is_rolled_over=False, **branch_filter).exclude(status='rolled_over').count(),
'default_rate': 0, # Calculate if needed
'defaulted_loans': Loan.objects.filter(status='defaulted', is_deleted=False, **branch_filter).count(),
},
'user_statistics': {
'total_users': User.objects.filter(branch=user.branch).count() if user.branch and not user.is_superuser else User.objects.count(),
'active_users': User.objects.filter(is_active=True, branch=user.branch).count() if user.branch and not user.is_superuser else User.objects.filter(is_active=True).count(),
},
'pending_applications': LoanApplication.objects.filter(status='pending', **branch_filter)[:5],
'overdue_loans': Loan.objects.filter(status='overdue', is_deleted=False, **branch_filter)[:5] if hasattr(Loan, 'status') else [],
})
context = {
'notifications': notifications_page,
'notification_types': notification_types,
'priority_levels': priority_levels,
'search_query': search_query,
'selected_type': notification_type,
'selected_priority': priority,
'selected_read_status': read_status,
'system_alerts': system_alerts,
'security_alerts': security_alerts,
'recent_activities': recent_activities,
**additional_context
}
return render(request, 'utils/notifications.html', context)
@login_required
def settings(request):
"""View for managing system settings - read-only for non-admins"""
# Check if user has permission to access settings
if not request.user.has_permission('settings', 'access'):
messages.error(request, 'You do not have permission to access settings.')
return redirect('dashboard')
# Check if user is admin - only admins can modify settings
is_admin_user = request.user.is_superuser or request.user.role == 'admin'
if request.method == 'POST' and is_admin_user:
setting_id = request.POST.get('setting_id')
value = request.POST.get('value')
setting = get_object_or_404(SystemSetting, id=setting_id)
# Validate setting value based on category
try:
if setting.category == 'loan':
value = float(value)
if value < 0:
raise ValueError("Loan settings must be positive numbers")
elif setting.category == 'notification':
value = value.lower() == 'true'
except ValueError as e:
messages.error(request, f'Invalid value for {setting.key}: {str(e)}')
return redirect('utils:settings')
# Update setting
setting.value = str(value)
setting.save()
# Create audit log
AuditLog.objects.create(
user=request.user,
action='update',
model_name='SystemSetting',
object_id=str(setting.id),
description=f'Updated setting {setting.key} to {value}'
)
messages.success(request, f'Setting {setting.key} updated successfully')
return redirect('utils:settings')
elif request.method == 'POST' and not is_admin_user:
messages.error(request, 'You do not have permission to modify system settings.')
return redirect('utils:settings')
# Get settings grouped by category
settings = {}
for category, _ in SystemSetting.CATEGORIES:
settings[category] = SystemSetting.objects.filter(category=category).order_by('key')
# Get default role permissions for settings page
from users.models import DefaultRolePermission, RolePermission
from users.simplified_permissions_views import SIMPLIFIED_PERMISSIONS
default_role_permissions = {}
if is_admin_user:
# Get all roles
roles = ['admin', 'team_leader', 'loan_officer', 'secretary', 'auditor']
for role in roles:
default_role_permissions[role] = {}
# Use actual SIMPLIFIED_PERMISSIONS structure (same as simplified permissions page)
for module_name, actions in SIMPLIFIED_PERMISSIONS.items():
module_code = module_name.lower().replace(' ', '').replace('&', '').replace('-', '')
if module_name not in default_role_permissions[role]:
default_role_permissions[role][module_name] = []
# Get or check default permission for each action
for action_code, action_description in actions.items():
try:
default_perm = DefaultRolePermission.objects.get(
role=role,
module=module_code,
action=action_code
)
is_allowed = default_perm.is_allowed
except DefaultRolePermission.DoesNotExist:
# Check if there's a role permission as fallback
try:
role_perm = RolePermission.objects.get(
role=role,
module=module_code,
action=action_code
)
is_allowed = role_perm.is_allowed
except RolePermission.DoesNotExist:
is_allowed = False # Default to False if not set
default_role_permissions[role][module_name].append({
'action': action_code,
'action_name': action_description, # Add action name for display
'is_allowed': is_allowed,
'description': action_description,
})
context = {
'settings': settings,
'categories': SystemSetting.CATEGORIES,
'is_admin_user': is_admin_user,
'default_role_permissions': default_role_permissions,
'roles': ['admin', 'team_leader', 'loan_officer', 'secretary', 'auditor'] if is_admin_user else [],
}
return render(request, 'utils/settings.html', context)
@login_required
@user_passes_test(is_admin)
def update_default_permission(request):
"""AJAX endpoint for updating default role permissions"""
if request.method != 'POST':
return JsonResponse({'error': 'Method not allowed'}, status=405)
import json
from users.models import DefaultRolePermission, RolePermission
from django.db import transaction
try:
data = json.loads(request.body)
role = data.get('role')
module = data.get('module')
action = data.get('action')
is_allowed = data.get('is_allowed', False)
if not all([role, module, action]):
return JsonResponse({'error': 'Missing required fields'}, status=400)
with transaction.atomic():
# Update or create default permission
default_perm, created = DefaultRolePermission.objects.get_or_create(
role=role,
module=module,
action=action,
defaults={'is_allowed': is_allowed}
)
if not created:
default_perm.is_allowed = is_allowed
default_perm.save()
# Update role permission for all users with this role
RolePermission.objects.filter(
role=role,
module=module,
action=action
).update(is_allowed=is_allowed)
return JsonResponse({'success': True, 'message': 'Permission updated successfully'})
except Exception as e:
return JsonResponse({'error': str(e)}, status=500)
@login_required
@user_passes_test(is_admin)
def update_setting(request):
"""AJAX endpoint for updating settings"""
if request.method != 'POST':
return JsonResponse({'success': False, 'error': 'Method not allowed'}, status=405)
try:
data = json.loads(request.body)
key = data.get('key')
value = data.get('value')
if not key:
return JsonResponse({'success': False, 'error': 'Setting key is required'}, status=400)
# Get the setting
try:
setting = SystemSetting.objects.get(key=key)
except SystemSetting.DoesNotExist:
return JsonResponse({'success': False, 'error': f'Setting {key} not found'}, status=404)
# Validate setting value based on category
try:
if setting.category == 'loan':
if key.endswith('_repayment_methods'):
# Handle JSON array for repayment methods
try:
methods = json.loads(value)
except (json.JSONDecodeError, TypeError):
raise ValueError("Invalid repayment methods format")
if not isinstance(methods, list):
raise ValueError("Repayment methods must be a list")
value = json.dumps(methods)
else:
# Generic loan numeric settings
try:
numeric = float(value)
except (TypeError, ValueError):
raise ValueError("Loan settings must be numeric")
if numeric < 0:
raise ValueError("Loan settings must be positive numbers")
value = str(numeric)
elif setting.category == 'notification':
value = str(value).lower() == 'true'
elif setting.category == 'system':
if 'max_file_size' in key.lower():
try:
size = int(value)
except (TypeError, ValueError):
raise ValueError("File size must be an integer")
if size <= 0:
raise ValueError("File size must be positive")
value = str(size)
elif 'allowed_file_types' in key.lower():
try:
types = json.loads(value)
except (json.JSONDecodeError, TypeError):
raise ValueError("Invalid file types format")
if not isinstance(types, list):
raise ValueError("File types must be a list")
value = json.dumps(types)
except ValueError as e:
return JsonResponse({'success': False, 'error': str(e)}, status=400)
# Update setting
old_value = setting.value
setting.value = str(value)
setting.save()
# Create audit log
AuditLog.objects.create(
user=request.user,
action='update',
model_name='SystemSetting',
object_id=str(setting.id),
description=f'Updated setting {key} from {old_value} to {value}'
)
return JsonResponse({
'success': True,
'message': f'Setting {key} updated successfully',
'new_value': str(value)
})
except json.JSONDecodeError:
return JsonResponse({'success': False, 'error': 'Invalid JSON data'}, status=400)
except Exception as e:
return JsonResponse({'success': False, 'error': str(e)}, status=500)
@login_required
def download_client_report(request, client_id, format='pdf'):
"""Download a client report in PDF or CSV format"""
client = get_object_or_404(CustomUser, id=client_id)
if format == 'pdf':
buffer = OptimizedReportGenerator.generate_client_report_pdf(client)
response = HttpResponse(buffer, content_type='application/pdf')
response['Content-Disposition'] = f'attachment; filename="{client.get_full_name()}_report.pdf"'
else: # CSV
data = OptimizedReportGenerator.generate_client_csv(client)
response = HttpResponse(data, content_type='text/csv')
response['Content-Disposition'] = f'attachment; filename="{client.get_full_name()}_report.csv"'
return response
@login_required
def download_loan_report(request, loan_id, format='pdf'):
"""Download a loan report in PDF or CSV format"""
loan = get_object_or_404(Loan, id=loan_id)
if format == 'pdf':
buffer = OptimizedReportGenerator.generate_loan_report_pdf(loan)
response = HttpResponse(buffer, content_type='application/pdf')
response['Content-Disposition'] = f'attachment; filename="loan_{loan.loan_number}_report.pdf"'
else: # CSV
data = OptimizedReportGenerator.generate_loan_csv(loan)
response = HttpResponse(data, content_type='text/csv')
response['Content-Disposition'] = f'attachment; filename="loan_{loan.loan_number}_report.csv"'
return response
@login_required
@user_passes_test(is_admin)
def download_system_report(request, format='pdf'):
"""Download a system-wide report in PDF or CSV format"""
if format == 'pdf':
buffer = OptimizedReportGenerator.generate_system_report_pdf()
response = HttpResponse(buffer, content_type='application/pdf')
response['Content-Disposition'] = 'attachment; filename="system_report.pdf"'
else: # CSV
data = OptimizedReportGenerator.generate_system_csv()
response = HttpResponse(data, content_type='text/csv')
response['Content-Disposition'] = 'attachment; filename="system_report.csv"'
return response
@login_required
def download_credit_score_report(request, client_id, format='pdf'):
"""Download a credit score report in PDF or CSV format"""
client = get_object_or_404(CustomUser, id=client_id)
if format == 'pdf':
buffer = OptimizedReportGenerator.generate_credit_score_report_pdf(client)
response = HttpResponse(buffer, content_type='application/pdf')
response['Content-Disposition'] = f'attachment; filename="{client.get_full_name()}_credit_score.pdf"'
else: # CSV
data = OptimizedReportGenerator.generate_credit_score_csv(client)
response = HttpResponse(data, content_type='text/csv')
response['Content-Disposition'] = f'attachment; filename="{client.get_full_name()}_credit_score.csv"'
return response
@login_required
def get_document(request, document_type):
"""API endpoint to get documents by type"""
documents = Document.objects.filter(document_type=document_type).order_by('-created_at')
data = [{
'id': doc.id,
'name': doc.name,
'url': doc.file.url,
'uploaded_by': doc.uploaded_by.get_full_name(),
'created_at': doc.created_at.strftime('%Y-%m-%d %H:%M:%S')
} for doc in documents]
return JsonResponse({'documents': data})
@login_required
@user_passes_test(is_admin)
def delete_document(request, document_id):
"""API endpoint to delete a document"""
# Check if user has permission to delete documents
if not request.user.has_permission('documents', 'delete'):
messages.error(request, 'You do not have permission to delete documents.')
return redirect('documents')
document = get_object_or_404(Document, id=document_id)
# Create audit log before deletion
AuditLog.objects.create(
user=request.user,
action='delete',
model_name='Document',
object_id=str(document.id),
description=f'Deleted document: {document.name}'
)
document.delete()
return JsonResponse({'status': 'success'})
@login_required
def get_notifications(request):
"""API endpoint to get all notifications"""
user = request.user
# Apply portfolio-based access control for staff members
if user.role in ['loan_officer', 'team_leader'] and not user.is_superuser:
# Staff members can only see notifications related to their assigned borrowers
notifications = Notification.objects.filter(
Q(user__portfolio_manager=user) | Q(user__isnull=True)
)
elif user.role in ['secretary', 'auditor'] and not user.is_superuser:
# Secretaries and auditors can only see notifications from users in their branch
if user.branch:
notifications = Notification.objects.filter(
Q(user__branch=user.branch) | Q(user__isnull=True)
)
else:
# If no branch assigned, only show system notifications
notifications = Notification.objects.filter(user__isnull=True)
else:
# Admin and superuser can see all notifications
notifications = Notification.objects.filter(
Q(user=user) | Q(user__isnull=True)
)
notifications = notifications.order_by('-created_at')[:5]
return JsonResponse({
'notifications': [{
'id': str(n.id),
'title': n.title,
'message': n.message,
'priority': n.priority,
'is_read': n.read_at is not None,
'created_at': n.created_at.isoformat(),
'action_url': n.action_url or '',
'notification_type': n.notification_type,
'icon': n.icon
} for n in notifications]
})
@login_required
def get_new_notifications(request):
"""API endpoint to get new notifications since last check"""
from datetime import datetime, timedelta
# Get last check time from query parameter (optional)
last_check = request.GET.get('since')
user = request.user
# Base query for user's notifications
if user.role in ['loan_officer', 'team_leader'] and not user.is_superuser:
notifications = Notification.objects.filter(
Q(user__portfolio_manager=user) | Q(user__isnull=True)
)
elif user.role in ['secretary', 'auditor'] and not user.is_superuser:
if user.branch:
notifications = Notification.objects.filter(
Q(user__branch=user.branch) | Q(user__isnull=True)
)
else:
notifications = Notification.objects.filter(user__isnull=True)
else:
notifications = Notification.objects.filter(
Q(user=user) | Q(user__isnull=True)
)
# Filter by time if provided
if last_check:
try:
last_check_time = datetime.fromisoformat(last_check.replace('Z', '+00:00'))
notifications = notifications.filter(created_at__gt=last_check_time)
except:
# Default to last 5 minutes if parsing fails
five_mins_ago = datetime.now() - timedelta(minutes=5)
notifications = notifications.filter(created_at__gte=five_mins_ago)
else:
# Default to last 1 minute
one_min_ago = datetime.now() - timedelta(minutes=1)
notifications = notifications.filter(created_at__gte=one_min_ago)
notifications = notifications.filter(read_at__isnull=True).order_by('-created_at')[:10]
return JsonResponse({
'count': notifications.count(),
'notifications': [{
'id': str(n.id),
'title': n.title,
'message': n.message,
'priority': n.priority,
'is_read': False,
'created_at': n.created_at.isoformat(),
'action_url': n.action_url or '',
'notification_type': n.notification_type,
'icon': n.icon
} for n in notifications]
})
@login_required
def get_unread_count(request):
"""API endpoint to get unread notification count"""
try:
count = Notification.objects.filter(
Q(user=request.user) | Q(user__isnull=True),
read_at__isnull=True # Use read_at instead of read (boolean field doesn't exist)
).count()
return JsonResponse({'count': count})
except Exception as e:
import logging
logger = logging.getLogger(__name__)
logger.error(f"Error getting unread notification count: {str(e)}", exc_info=True)
return JsonResponse({'count': 0, 'error': str(e)}, status=500)
@login_required
def unread_notifications_count(request):
"""Return count of unread notifications for the current user"""
try:
# Assuming you have a Notification model
from utils.models import Notification
count = Notification.objects.filter(
Q(user=request.user) | Q(user__isnull=True),
read_at__isnull=True
).count()
return JsonResponse({'count': count})
except Exception as e:
return JsonResponse({'count': 0, 'error': str(e)})
@login_required
def payment_receipts(request):
"""View all payment receipts"""
from loans.models import Repayment
from django.db.models import Q
# Get selected branch from session
selected_branch_id = request.session.get('selected_branch_id')
# Get filter parameters
search_query = request.GET.get('search', '')
date_from = request.GET.get('date_from', '')
date_to = request.GET.get('date_to', '')
payment_method = request.GET.get('payment_method', '')
# Base queryset
receipts = Repayment.objects.select_related('loan', 'loan__borrower').all()
# Apply branch filtering
if selected_branch_id:
# Filter by selected branch
receipts = receipts.filter(loan__borrower__branch_id=selected_branch_id)
elif hasattr(request.user, 'branch') and request.user.branch and not request.user.is_superuser:
# Filter by user's branch (non-superuser)
receipts = receipts.filter(loan__borrower__branch=request.user.branch)
# Superuser without branch selection sees all receipts
# Apply filters
if search_query:
receipts = receipts.filter(
Q(receipt_number__icontains=search_query) |
Q(loan__borrower__first_name__icontains=search_query) |
Q(loan__borrower__last_name__icontains=search_query) |
Q(loan__loan_number__icontains=search_query)
)
if date_from:
receipts = receipts.filter(payment_date__gte=date_from)
if date_to:
receipts = receipts.filter(payment_date__lte=date_to)
if payment_method:
receipts = receipts.filter(payment_method=payment_method)
# Order by payment date (newest first)
receipts = receipts.order_by('-payment_date')
context = {
'receipts': receipts,
'search_query': search_query,
'date_from': date_from,
'date_to': date_to,
'payment_method': payment_method
}
return render(request, 'utils/payment_receipts.html', context)
@login_required
def mark_notification_read(request, notification_id):
"""API endpoint to mark a notification as read"""
try:
notification = Notification.objects.get(
id=notification_id,
user=request.user
)
notification.mark_as_read()
return JsonResponse({'success': True})
except Notification.DoesNotExist:
return JsonResponse({'success': False, 'error': 'Notification not found'}, status=404)
@login_required
def mark_all_notifications_read(request):
"""API endpoint to mark all notifications as read"""
from django.utils import timezone as django_timezone
Notification.objects.filter(
Q(user=request.user) | Q(user__isnull=True),
read_at__isnull=True
).update(read_at=django_timezone.now())
return JsonResponse({'success': True})
@login_required
def get_notification_stats(request):
"""API endpoint to get notification statistics"""
total_notifications = Notification.objects.filter(
Q(user=request.user) | Q(user__isnull=True)
).count()
unread_notifications = Notification.objects.filter(
Q(user=request.user) | Q(user__isnull=True),
read_at__isnull=True
).count()
# Get notifications by priority
priority_stats = {}
for priority in ['urgent', 'high', 'medium', 'low']:
count = Notification.objects.filter(
Q(user=request.user) | Q(user__isnull=True),
priority=priority
).count()
priority_stats[priority] = count
# Get notifications by type
type_stats = {}
for notification_type, _ in Notification.NOTIFICATION_TYPES:
count = Notification.objects.filter(
Q(user=request.user) | Q(user__isnull=True),
notification_type=notification_type
).count()
if count > 0:
type_stats[notification_type] = count
return JsonResponse({
'total': total_notifications,
'unread': unread_notifications,
'priority_stats': priority_stats,
'type_stats': type_stats
})
@login_required
def get_recent_activity(request):
"""API endpoint to get recent activity for notifications page"""
from django.utils import timezone
from utils.datetime_utils import get_current_datetime, make_datetime_compatible
from datetime import timedelta
# Get recent activity from the last 24 hours
recent_activities = AuditLog.objects.filter(
created_at__gte=get_current_datetime() - timedelta(hours=24)
).select_related('user').order_by('-created_at')[:20]
activities = []
for activity in recent_activities:
activity_type = activity.action
icon = {
'create': 'plus-circle',
'update': 'edit',
'delete': 'trash',
'payment': 'money-bill-wave',
'login': 'sign-in-alt',
'logout': 'sign-out-alt'
}.get(activity_type, 'info')
activities.append({
'type': activity_type,
'user': activity.user.get_full_name() if activity.user else 'System',
'action': activity.description,
'timestamp': activity.created_at.isoformat(),
'icon': icon,
'model_name': activity.model_name,
'object_id': activity.object_id
})
return JsonResponse({'activities': activities})
@login_required
def client_loan_history(request, client_id):
"""API endpoint to get client's loan history for charts"""
loans = Loan.objects.filter(client_id=client_id).order_by('created_at')
data = {
'dates': [loan.created_at.strftime('%Y-%m-%d') for loan in loans],
'amounts': [float(loan.principal_amount) for loan in loans]
}
return JsonResponse(data)
@login_required
def download_receipt(request, receipt_id):
"""Download a payment receipt PDF"""
# Check if user has permission to download payment receipts
if not request.user.has_permission('payment_receipts', 'download'):
messages.error(request, 'You do not have permission to download receipts.')
return redirect('dashboard')
from utils.models import Receipt
from utils.reports import OptimizedReportGenerator
from django.http import HttpResponse
try:
receipt = get_object_or_404(Receipt, id=receipt_id)
# Check if user has permission to view this receipt
if not (request.user.is_staff or request.user == receipt.borrower):
return HttpResponse("Unauthorized", status=403)
# Generate PDF if not exists or regenerate
if not receipt.pdf_file:
pdf_buffer = OptimizedReportGenerator.generate_payment_receipt_pdf(receipt)
# Save PDF to file
from django.core.files.base import ContentFile
pdf_content = ContentFile(pdf_buffer.getvalue())
receipt.pdf_file.save(
f'receipt_{receipt.receipt_number}.pdf',
pdf_content,
save=True
)
# Return PDF response
response = HttpResponse(receipt.pdf_file.read(), content_type='application/pdf')
response['Content-Disposition'] = f'attachment; filename="Receipt_{receipt.receipt_number}.pdf"'
return response
except Exception as e:
return HttpResponse(f"Error generating receipt: {str(e)}", status=500)
@login_required
def view_receipt(request, receipt_id):
"""View a payment receipt PDF in browser"""
from utils.models import Receipt
from utils.reports import OptimizedReportGenerator
from django.http import HttpResponse
try:
receipt = get_object_or_404(Receipt, id=receipt_id)
# Check if user has permission to view this receipt
if not (request.user.is_staff or request.user == receipt.borrower):
return HttpResponse("Unauthorized", status=403)
# Check if thermal printer format is requested
thermal_printer = request.GET.get('thermal', 'false').lower() == 'true'
# Generate PDF
pdf_buffer = OptimizedReportGenerator.generate_payment_receipt_pdf(receipt, thermal_printer=thermal_printer)
# Return PDF for inline viewing
filename_suffix = "_thermal" if thermal_printer else ""
response = HttpResponse(pdf_buffer.getvalue(), content_type='application/pdf')
response['Content-Disposition'] = f'inline; filename="Receipt_{receipt.receipt_number}{filename_suffix}.pdf"'
return response
except Exception as e:
return HttpResponse(f"Error viewing receipt: {str(e)}", status=500)
@login_required
def download_loan_statement(request, loan_id):
"""Download a loan statement PDF"""
from loans.models import Loan
from utils.reports import OptimizedReportGenerator
from django.http import HttpResponse
try:
loan = get_object_or_404(Loan, id=loan_id)
# Check if user has permission to view this loan
if not (request.user.is_staff or request.user == loan.borrower):
return HttpResponse("Unauthorized", status=403)
# Generate PDF
pdf_buffer = OptimizedReportGenerator.generate_loan_statement_pdf(loan)
# Return PDF response
response = HttpResponse(pdf_buffer.getvalue(), content_type='application/pdf')
response['Content-Disposition'] = f'attachment; filename="Statement_{loan.loan_number}.pdf"'
return response
except Exception as e:
return HttpResponse(f"Error generating statement: {str(e)}", status=500)
@login_required
def view_loan_statement(request, loan_id):
"""View a loan statement PDF in browser"""
from loans.models import Loan
from utils.reports import OptimizedReportGenerator
from django.http import HttpResponse
try:
loan = get_object_or_404(Loan, id=loan_id)
# Check if user has permission to view this loan
if not (request.user.is_staff or request.user == loan.borrower):
return HttpResponse("Unauthorized", status=403)
# Generate PDF
pdf_buffer = OptimizedReportGenerator.generate_loan_statement_pdf(loan)
# Return PDF for inline viewing
response = HttpResponse(pdf_buffer.getvalue(), content_type='application/pdf')
response['Content-Disposition'] = f'inline; filename="Statement_{loan.loan_number}.pdf"'
return response
except Exception as e:
return HttpResponse(f"Error viewing statement: {str(e)}", status=500)
@login_required
def receipts_list(request):
"""Enhanced receipts list with statistics and better filtering"""
# Check if user has permission to access payment receipts
if not request.user.has_permission('payment_receipts', 'access'):
messages.error(request, 'You do not have permission to access payment receipts.')
return redirect('dashboard')
from loans.models import Repayment
from django.core.paginator import Paginator
from django.db.models import Q, Sum, Count
from django.utils import timezone
from utils.datetime_utils import get_current_datetime, make_datetime_compatible
from datetime import datetime, timedelta
# BRANCH FILTERING IMPLEMENTATION
# Step 1: Get base queryset - Use Repayment model instead of Receipt
receipts = Repayment.objects.select_related('loan', 'loan__borrower', 'loan__application', 'loan__application__loan_product').all()
# Step 2: Apply branch filtering
selected_branch_id = request.session.get('selected_branch_id')
user = request.user
# Branch filtering logic
if user.is_superuser:
if selected_branch_id:
# Superuser with branch selection - filter by selected branch
receipts = receipts.filter(loan__borrower__branch_id=selected_branch_id)
# Superuser without branch selection sees all receipts (no filter)
elif user.is_staff:
if selected_branch_id:
# Staff with branch selection
receipts = receipts.filter(loan__borrower__branch_id=selected_branch_id)
elif hasattr(user, 'branch') and user.branch:
# Staff with own branch
receipts = receipts.filter(loan__borrower__branch=user.branch)
# Staff without branch sees all receipts
else:
# Regular user
if hasattr(user, 'branch') and user.branch:
# Regular user with branch - see receipts for borrowers in their branch
receipts = receipts.filter(loan__borrower__branch=user.branch)
else:
# Regular user without branch - see only their own receipts
receipts = receipts.filter(loan__borrower=user)
# Calculate statistics before filtering
today = get_current_datetime().date()
week_start = today - timedelta(days=today.weekday())
# Get statistics for the current user's receipts
base_receipts = receipts # Store original queryset for stats
total_stats = base_receipts.aggregate(
total_amount=Sum('amount'),
total_count=Count('id')
)
today_receipts = base_receipts.filter(payment_date__date=today).count()
week_receipts = base_receipts.filter(payment_date__date__gte=week_start).count()
# Search functionality
search = request.GET.get('search')
if search:
receipts = receipts.filter(
Q(receipt_number__icontains=search) |
Q(loan__borrower__first_name__icontains=search) |
Q(loan__borrower__last_name__icontains=search) |
Q(loan__borrower__phone_number__icontains=search) |
Q(loan__loan_number__icontains=search)
)
# Date filter
date_from = request.GET.get('date_from')
date_to = request.GET.get('date_to')
if date_from:
receipts = receipts.filter(payment_date__gte=date_from)
if date_to:
receipts = receipts.filter(payment_date__lte=date_to)
# Payment method filter
payment_method = request.GET.get('payment_method')
if payment_method:
receipts = receipts.filter(payment_method=payment_method)
# Order by date descending
receipts = receipts.order_by('-payment_date')
# Pagination
paginator = Paginator(receipts, 12) # Show 12 per page for better grid layout
page_number = request.GET.get('page')
receipts_page = paginator.get_page(page_number)
context = {
'receipts': receipts_page,
'page_obj': receipts_page,
'is_paginated': receipts_page.has_other_pages(),
'search': search,
'date_from': date_from,
'date_to': date_to,
'payment_method': payment_method,
'today_receipts': today_receipts,
'week_receipts': week_receipts,
'total_amount': total_stats['total_amount'] or 0,
'total_count': total_stats['total_count'] or 0,
}
return render(request, 'utils/receipts_list.html', context)
@login_required
def credit_scores_management(request):
"""Credit scores management page"""
from reports.models import LoanScoring
from django.core.paginator import Paginator
from django.db.models import Q
# Get filter parameters
risk_level = request.GET.get('risk_level')
score_range = request.GET.get('score_range')
search = request.GET.get('search')
# Base queryset
credit_scores = LoanScoring.objects.select_related('user').all()
# Apply filters
if risk_level:
credit_scores = credit_scores.filter(risk_level=risk_level)
if score_range:
if score_range == '80-100':
credit_scores = credit_scores.filter(total_score__gte=80)
elif score_range == '60-79':
credit_scores = credit_scores.filter(total_score__gte=60, total_score__lt=80)
elif score_range == '40-59':
credit_scores = credit_scores.filter(total_score__gte=40, total_score__lt=60)
elif score_range == '0-39':
credit_scores = credit_scores.filter(total_score__lt=40)
if search:
credit_scores = credit_scores.filter(
Q(user__first_name__icontains=search) |
Q(user__last_name__icontains=search) |
Q(user__email__icontains=search)
)
# Order by score descending
credit_scores = credit_scores.order_by('-total_score')
# Pagination
paginator = Paginator(credit_scores, 20)
page_number = request.GET.get('page')
credit_scores_page = paginator.get_page(page_number)
# Statistics
all_scores = LoanScoring.objects.all()
stats = {
'total_borrowers': all_scores.count(),
'low_risk': all_scores.filter(risk_level='low').count(),
'medium_risk': all_scores.filter(risk_level='medium').count(),
'high_risk': all_scores.filter(risk_level='high').count(),
}
context = {
'credit_scores': credit_scores_page,
'stats': stats,
}
return render(request, 'utils/credit_scores.html', context)
@login_required
def client_credit_score(request, client_id):
"""API endpoint to get client's credit score history for charts"""
scores = LoanScoring.objects.filter(client_id=client_id).order_by('created_at')
data = {
'dates': [score.created_at.strftime('%Y-%m-%d') for score in scores],
'scores': [score.total_score for score in scores]
}
return JsonResponse(data)
@login_required
def maintenance(request):
"""System maintenance page"""
return render(request, 'utils/maintenance.html')
@login_required
@portfolio_access_required
def all_customer_documents(request):
"""View all customer documents across the system"""
# Check if user has permission to access customer documents
if not request.user.has_permission('customer_documents', 'access'):
messages.error(request, 'You do not have permission to access customer documents.')
return redirect('dashboard')
from users.models import CustomUser
# Get filter parameters
search_query = request.GET.get('search', '')
document_type = request.GET.get('document_type', '')
client_id = request.GET.get('client_id', '')
date_from = request.GET.get('date_from', '')
date_to = request.GET.get('date_to', '')
status = request.GET.get('status', '')
# Get selected branch from session
selected_branch_id = request.session.get('selected_branch_id')
# Get all clients (borrowers) - filter by branch if selected
clients = CustomUser.objects.filter(role='borrower').order_by('first_name', 'last_name')
# Apply portfolio-based access control for staff members
if request.user.role in ['loan_officer', 'team_leader'] and not request.user.is_superuser:
# Staff members can only see documents from their assigned borrowers
clients = clients.filter(portfolio_manager=request.user)
elif request.user.role in ['secretary', 'auditor'] and not request.user.is_superuser:
# Secretaries and auditors can only see documents from borrowers in their branch
if request.user.branch:
clients = clients.filter(branch=request.user.branch)
# Apply branch filtering if a branch is selected
if selected_branch_id:
clients = clients.filter(branch_id=selected_branch_id)
# Get documents from both CustomUser model and Document model
customer_documents = []
# Process each client
for client in clients:
client_docs = {
'client': client,
'id_documents': [],
'selfie_documents': [],
'utility_documents': [],
'bank_documents': [],
'business_documents': [],
'other_documents': [],
'total_documents': 0
}
# Get documents from CustomUser model (KYC documents)
if client.id_document:
client_docs['id_documents'].append({
'name': 'ID Document',
'file': client.id_document,
'uploaded_at': client.created_at,
'document_type': 'id_document',
'source': 'kyc'
})
client_docs['total_documents'] += 1
if client.selfie:
client_docs['selfie_documents'].append({
'name': 'Selfie Photo',
'file': client.selfie,
'uploaded_at': client.created_at,
'document_type': 'selfie',
'source': 'kyc'
})
client_docs['total_documents'] += 1
if client.utility_bill:
client_docs['utility_documents'].append({
'name': 'Utility Bill',
'file': client.utility_bill,
'uploaded_at': client.created_at,
'document_type': 'utility_bill',
'source': 'kyc'
})
client_docs['total_documents'] += 1
if client.bank_statement:
client_docs['bank_documents'].append({
'name': 'Bank Statement',
'file': client.bank_statement,
'uploaded_at': client.created_at,
'document_type': 'bank_statement',
'source': 'kyc'
})
client_docs['total_documents'] += 1
if client.business_license:
client_docs['business_documents'].append({
'name': 'Business License',
'file': client.business_license,
'uploaded_at': client.created_at,
'document_type': 'business_license',
'source': 'kyc'
})
client_docs['total_documents'] += 1
if client.tax_certificate:
client_docs['business_documents'].append({
'name': 'Tax Certificate',
'file': client.tax_certificate,
'uploaded_at': client.created_at,
'document_type': 'tax_certificate',
'source': 'kyc'
})
client_docs['total_documents'] += 1
# Get documents from Document model
from .models import Document
doc_documents = Document.objects.filter(
Q(uploaded_by=client) | Q(shared_with=client)
).select_related('uploaded_by').prefetch_related('tags')
for doc in doc_documents:
doc_info = {
'name': doc.name,
'file': doc.file,
'uploaded_at': doc.created_at,
'document_type': doc.document_type,
'source': 'document_model',
'description': doc.description,
'tags': doc.tags.all()
}
if doc.document_type == 'id_document':
client_docs['id_documents'].append(doc_info)
elif doc.document_type == 'selfie':
client_docs['selfie_documents'].append(doc_info)
elif doc.document_type == 'utility_bill':
client_docs['utility_documents'].append(doc_info)
elif doc.document_type == 'bank_statement':
client_docs['bank_documents'].append(doc_info)
elif doc.document_type in ['business_license', 'tax_return']:
client_docs['business_documents'].append(doc_info)
else:
client_docs['other_documents'].append(doc_info)
client_docs['total_documents'] += 1
# Apply filters
if search_query:
if search_query.lower() not in client.get_full_name().lower():
continue
if client_id and str(client.id) != client_id:
continue
if status and client.status != status:
continue
# Add to results if client has documents or matches search
if client_docs['total_documents'] > 0 or search_query:
customer_documents.append(client_docs)
# Apply document type filter
if document_type:
filtered_documents = []
for client_docs in customer_documents:
filtered_client_docs = client_docs.copy()
filtered_client_docs['total_documents'] = 0
# Only include documents of the specified type
for doc_type in ['id_documents', 'selfie_documents', 'utility_documents', 'bank_documents', 'business_documents', 'other_documents']:
if doc_type == f'{document_type}_documents':
filtered_client_docs[doc_type] = client_docs[doc_type]
filtered_client_docs['total_documents'] += len(client_docs[doc_type])
else:
filtered_client_docs[doc_type] = []
if filtered_client_docs['total_documents'] > 0:
filtered_documents.append(filtered_client_docs)
customer_documents = filtered_documents
# Apply date filters
if date_from or date_to:
from datetime import datetime
filtered_documents = []
for client_docs in customer_documents:
filtered_client_docs = client_docs.copy()
filtered_client_docs['total_documents'] = 0
for doc_type in ['id_documents', 'selfie_documents', 'utility_documents', 'bank_documents', 'business_documents', 'other_documents']:
filtered_docs = []
for doc in client_docs[doc_type]:
doc_date = doc['uploaded_at']
if date_from:
from_date = datetime.strptime(date_from, '%Y-%m-%d').date()
if doc_date.date() < from_date:
continue
if date_to:
to_date = datetime.strptime(date_to, '%Y-%m-%d').date()
if doc_date.date() > to_date:
continue
filtered_docs.append(doc)
filtered_client_docs[doc_type] = filtered_docs
filtered_client_docs['total_documents'] += len(filtered_docs)
if filtered_client_docs['total_documents'] > 0:
filtered_documents.append(filtered_client_docs)
customer_documents = filtered_documents
# Sort by total documents (descending)
customer_documents.sort(key=lambda x: x['total_documents'], reverse=True)
# Get document type choices for filter
document_types = [
('id_document', 'ID Documents'),
('selfie', 'Selfie Photos'),
('utility_bill', 'Utility Bills'),
('bank_statement', 'Bank Statements'),
('business_license', 'Business Documents'),
('other', 'Other Documents'),
]
context = {
'customer_documents': customer_documents,
'clients': clients,
'document_types': document_types,
'filters': {
'search': search_query,
'document_type': document_type,
'client_id': client_id,
'date_from': date_from,
'date_to': date_to,
'status': status,
},
'total_documents': sum(doc['total_documents'] for doc in customer_documents),
'total_clients': len(customer_documents),
}
return render(request, 'utils/all_customer_documents.html', context)
@login_required
def serve_media_file(request, file_path):
"""
Secure media file serving with permission checks
"""
# Clean the file path to prevent directory traversal
file_path = os.path.normpath(file_path)
if file_path.startswith('..') or file_path.startswith('/'):
raise Http404("Invalid file path")
# Construct full path
full_path = os.path.join(settings.MEDIA_ROOT, file_path)
# Security check: ensure file is within MEDIA_ROOT
if not os.path.abspath(full_path).startswith(os.path.abspath(settings.MEDIA_ROOT)):
raise Http404("File not found")
# Check if file exists
if not os.path.exists(full_path) or not os.path.isfile(full_path):
raise Http404("File not found")
# Permission check based on file type
if 'kyc/' in file_path:
# KYC documents - only allow access to own documents or admin
if not request.user.is_staff:
# For regular users, check if they own the document
# This is a simplified check - you might need to implement more specific logic
if not request.user.is_superuser:
raise PermissionDenied("Access denied")
# Determine content type
content_type, _ = mimetypes.guess_type(full_path)
if content_type is None:
content_type = 'application/octet-stream'
# Serve the file
try:
with open(full_path, 'rb') as f:
response = HttpResponse(f.read(), content_type=content_type)
response['Content-Disposition'] = f'inline; filename="{os.path.basename(full_path)}"'
return response
except IOError:
raise Http404("File not found")
@login_required
@login_required
def test_media_access(request):
"""Test view to verify media file access"""
media_root = settings.MEDIA_ROOT
test_file = os.path.join(media_root, 'test.txt')
# Create a test file
with open(test_file, 'w') as f:
f.write('Media test file')
# Check if file exists
if os.path.exists(test_file):
with open(test_file, 'r') as f:
content = f.read()
# Clean up
os.remove(test_file)
html_content = f"""
Media Test
Media File Test
Media root: {media_root}
Test file content: {content}
Status: SUCCESS - Media files are working
"""
return HttpResponse(html_content)
else:
html_content = f"""
Media Test
Media File Test
Media root: {media_root}
Status: FAILED - Media files are not working
"""
return HttpResponse(html_content)
@login_required
def simple_media_test(request):
"""Simple media test without file operations"""
try:
media_root = settings.MEDIA_ROOT
html_content = f"""
Media Test
Media File Test
Media root: {media_root}
Status: SUCCESS - Django is working
Media directory exists: {os.path.exists(media_root)}
"""
return HttpResponse(html_content)
except Exception as e:
html_content = f"""
Media Test
Media File Test
Error: {str(e)}
Status: FAILED - Django error
"""
return HttpResponse(html_content)
@login_required
def bulk_generate_receipts(request):
"""Generate receipts for all repayments that don't have receipts"""
# Check if user has permission to generate payment receipts
if not request.user.has_permission('payment_receipts', 'generate'):
messages.error(request, 'You do not have permission to generate receipts.')
return redirect('dashboard')
if request.method == 'POST':
# Get all repayments without receipts - fix the select_related
repayments_without_receipts = Repayment.objects.filter(
receipt__isnull=True
).select_related('loan', 'loan__borrower')
generated_count = 0
errors = []
for repayment in repayments_without_receipts:
try:
# Use the existing create_or_update_receipt method
repayment.create_or_update_receipt()
generated_count += 1
except Exception as e:
errors.append(f"Error generating receipt for repayment {repayment.pk}: {str(e)}")
if generated_count > 0:
messages.success(request, f"Successfully generated {generated_count} receipts.")
if errors:
for error in errors[:5]: # Show only first 5 errors
messages.error(request, error)
if len(errors) > 5:
messages.warning(request, f"And {len(errors) - 5} more errors occurred.")
if generated_count == 0 and not errors:
messages.info(request, "All repayments already have receipts generated.")
return redirect('utils:receipts_list')
@login_required
def edit_receipt(request, receipt_id):
"""Edit receipt details"""
# Check if user has permission to edit payment receipts
if not request.user.has_permission('payment_receipts', 'edit'):
messages.error(request, 'You do not have permission to edit receipts.')
return redirect('dashboard')
from utils.models import Receipt
from django.shortcuts import get_object_or_404, redirect
from django.contrib import messages
from django.db.models import Sum
from decimal import Decimal
receipt = get_object_or_404(Receipt, id=receipt_id)
# Check permissions
if not request.user.is_staff and receipt.borrower != request.user:
messages.error(request, 'You do not have permission to edit this receipt.')
return redirect('utils:receipts_list')
if request.method == 'POST':
# Update receipt details
amount_paid = Decimal(request.POST.get('amount_paid', '0'))
payment_method = request.POST.get('payment_method')
# Update the related repayment first
repayment = receipt.repayment
old_amount = repayment.amount
repayment.amount = amount_paid
repayment.payment_method = payment_method
repayment.save() # This will trigger receipt update
# Recalculate loan totals
loan = receipt.loan
loan.amount_paid = loan.repayments.aggregate(
total=Sum('amount')
)['total'] or Decimal('0.00')
loan.save()
messages.success(request, f'Receipt {receipt.receipt_number} updated successfully.')
return redirect('utils:receipts_list')
context = {
'receipt': receipt,
'payment_methods': Receipt.PAYMENT_METHODS,
}
return render(request, 'utils/edit_receipt.html', context)
@login_required
def delete_receipt(request, receipt_id):
"""Delete receipt and related repayment"""
# Check if user has permission to delete payment receipts
if not request.user.has_permission('payment_receipts', 'delete'):
messages.error(request, 'You do not have permission to delete receipts.')
return redirect('dashboard')
from utils.models import Receipt
from django.shortcuts import get_object_or_404, redirect
from django.contrib import messages
from django.db.models import Sum
from decimal import Decimal
receipt = get_object_or_404(Receipt, id=receipt_id)
# Check permissions - only staff can delete receipts
if not request.user.is_staff:
messages.error(request, 'You do not have permission to delete receipts.')
return redirect('utils:receipts_list')
if request.method == 'POST':
loan = receipt.loan
repayment = receipt.repayment
receipt_number = receipt.receipt_number
# Delete the repayment (this will cascade delete the receipt)
repayment.delete()
# Recalculate loan totals
loan.amount_paid = loan.repayments.aggregate(
total=Sum('amount')
)['total'] or Decimal('0.00')
# Update loan status if needed
if loan.amount_paid < loan.total_amount and loan.status == 'paid':
loan.status = 'active'
loan.save()
# Recalculate all remaining receipt balances for this loan
remaining_repayments = loan.repayments.order_by('payment_date', 'created_at')
running_paid = Decimal('0.00')
for repayment in remaining_repayments:
try:
receipt_obj = Receipt.objects.get(repayment=repayment)
previous_balance = loan.total_amount - running_paid
new_balance = previous_balance - repayment.amount
receipt_obj.previous_balance = previous_balance
receipt_obj.new_balance = new_balance
receipt_obj.save()
running_paid += repayment.amount
except Receipt.DoesNotExist:
pass
messages.success(request, f'Receipt {receipt_number} deleted successfully.')
return redirect('utils:receipts_list')
context = {'receipt': receipt}
return render(request, 'utils/delete_receipt_confirm.html', context)
@login_required
def generate_receipt(request, loan_id):
"""Generate receipt for a loan repayment"""
# Check if user has permission to generate payment receipts
if not request.user.has_permission('payment_receipts', 'generate'):
messages.error(request, 'You do not have permission to generate receipts.')
return redirect('dashboard')
from loans.models import Loan
from django.shortcuts import get_object_or_404, redirect
from django.contrib import messages
from decimal import Decimal
loan = get_object_or_404(Loan, id=loan_id)
# Check permissions
if not request.user.is_staff and loan.borrower != request.user:
messages.error(request, 'You do not have permission to generate receipts for this loan.')
return redirect('loans:loan_detail', pk=loan_id)
if request.method == 'POST':
amount = Decimal(request.POST.get('amount', '0'))
payment_method = request.POST.get('payment_method', 'cash')
if amount <= 0:
messages.error(request, 'Payment amount must be greater than zero.')
return redirect('loans:loan_detail', pk=loan_id)
# Create new repayment (this will automatically create receipt)
from loans.models import Repayment
repayment = Repayment.objects.create(
loan=loan,
amount=amount,
payment_method=payment_method,
)
messages.success(request, f'Receipt {repayment.receipt_number} generated successfully.')
return redirect('loans:loan_detail', pk=loan_id)
context = {
'loan': loan,
'payment_methods': [('mpesa', 'M-Pesa'), ('bank', 'Bank Transfer'), ('cash', 'Cash'), ('cheque', 'Cheque')],
}
return render(request, 'utils/generate_receipt.html', context)
@login_required
def reports_dashboard(request):
"""Enhanced reports dashboard with all available reports and statements"""
from .reports import OptimizedReportGenerator, log_report_generation
import logging
# Get selected branch from session
selected_branch_id = request.session.get('selected_branch_id')
# Log dashboard access
log_report_generation(request.user, 'reports_dashboard_access')
# Get quick summary data for dashboard with branch filtering
try:
dashboard_data = OptimizedReportGenerator.generate_dashboard_data(branch_id=selected_branch_id)
executive_summary = OptimizedReportGenerator.generate_executive_summary(30, branch_id=selected_branch_id)
risk_analysis = OptimizedReportGenerator.generate_risk_analysis(branch_id=selected_branch_id)
except Exception as e:
logger = logging.getLogger(__name__)
logger.error(f"Error loading dashboard data: {e}")
dashboard_data = {'error': str(e)}
executive_summary = {'error': str(e)}
risk_analysis = {'error': str(e)}
# Available report types with descriptions
report_types = [
{
'id': 'executive_summary',
'name': 'Executive Summary',
'description': 'High-level overview of loan portfolio performance',
'icon': 'fas fa-chart-line',
'formats': ['pdf', 'excel', 'csv']
},
{
'id': 'performance_trends',
'name': 'Performance Trends',
'description': 'Loan performance trends over time',
'icon': 'fas fa-trending-up',
'formats': ['pdf', 'excel', 'csv']
},
{
'id': 'risk_analysis',
'name': 'Risk Analysis',
'description': 'Portfolio risk assessment and overdue loans',
'icon': 'fas fa-exclamation-triangle',
'formats': ['pdf', 'excel', 'csv']
},
{
'id': 'collection_statement',
'name': 'Collection Statement',
'description': 'Detailed collection report for date range',
'icon': 'fas fa-money-bill-wave',
'formats': ['pdf', 'excel', 'csv']
}
]
# Statement types
statement_types = [
{
'id': 'loan_statement',
'name': 'Loan Statement',
'description': 'Individual loan details and repayment history',
'icon': 'fas fa-file-invoice',
'formats': ['pdf', 'excel', 'csv']
},
{
'id': 'client_statement',
'name': 'Client Statement',
'description': 'Complete client profile and loan history',
'icon': 'fas fa-user-circle',
'formats': ['pdf', 'excel', 'csv']
}
]
context = {
'dashboard_data': dashboard_data,
'executive_summary': executive_summary,
'risk_analysis': risk_analysis,
'report_types': report_types,
'statement_types': statement_types,
'total_reports': len(report_types) + len(statement_types)
}
return render(request, 'utils/reports_dashboard.html', context)
@login_required
def generate_report_api(request):
"""API endpoint to generate reports and statements"""
if request.method != 'POST':
return JsonResponse({'success': False, 'error': 'Method not allowed'}, status=405)
try:
from .reports import generate_report_file, get_report_data, log_report_generation
import json
from utils.datetime_utils import get_current_datetime
data = json.loads(request.body)
report_type = data.get('report_type')
format_type = data.get('format', 'pdf')
parameters = data.get('parameters', {})
# Add user to parameters for logging
parameters['user'] = request.user
# Log the report generation request
log_report_generation(request.user, f'{report_type}_request', parameters)
if format_type in ['pdf', 'excel']:
# Generate file
file_buffer = generate_report_file(report_type, format_type, **parameters)
if file_buffer:
# Prepare file for download
filename = f"{report_type}_{get_current_datetime().strftime('%Y%m%d_%H%M%S')}.{format_type}"
if format_type == 'pdf':
content_type = 'application/pdf'
elif format_type == 'excel':
content_type = 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
filename = filename.replace('.excel', '.xlsx')
# Convert buffer to base64 for JSON response
import base64
file_content = base64.b64encode(file_buffer.getvalue()).decode('utf-8')
return JsonResponse({
'success': True,
'filename': filename,
'content_type': content_type,
'file_content': file_content
})
else:
return JsonResponse({
'success': False,
'error': 'Failed to generate report file'
})
elif format_type == 'json':
# Return JSON data
report_data = get_report_data(report_type, **parameters)
if 'error' in report_data:
return JsonResponse({
'success': False,
'error': report_data['error']
})
return JsonResponse({
'success': True,
'data': report_data
})
else:
return JsonResponse({
'success': False,
'error': f'Unsupported format: {format_type}'
})
except Exception as e:
log_report_generation(request.user, 'report_generation_error', {'error': str(e)}, False, str(e))
return JsonResponse({
'success': False,
'error': str(e)
})
@login_required
def download_report(request, report_type, format_type):
"""Direct download endpoint for reports"""
try:
from .reports import generate_report_file, log_report_generation
from utils.datetime_utils import get_current_datetime
# Get parameters from query string
parameters = dict(request.GET.items())
parameters['user'] = request.user
# Log the download request
log_report_generation(request.user, f'{report_type}_download', parameters)
# Generate the file
file_buffer = generate_report_file(report_type, format_type, **parameters)
if not file_buffer:
messages.error(request, f'Failed to generate {report_type} report')
return redirect('utils:utils_reports_dashboard')
# Prepare response
filename = f"{report_type}_{get_current_datetime().strftime('%Y%m%d_%H%M%S')}"
if format_type == 'pdf':
content_type = 'application/pdf'
filename += '.pdf'
elif format_type == 'excel':
content_type = 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
filename += '.xlsx'
elif format_type == 'csv':
content_type = 'text/csv'
filename += '.csv'
# For CSV, file_buffer is a string, not BytesIO
response = HttpResponse(file_buffer, content_type=content_type)
response['Content-Disposition'] = f'attachment; filename="{filename}"'
return response
# For PDF and Excel
response = HttpResponse(file_buffer.getvalue(), content_type=content_type)
response['Content-Disposition'] = f'attachment; filename="{filename}"'
return response
except Exception as e:
log_report_generation(request.user, f'{report_type}_download_error', {'error': str(e)}, False, str(e))
messages.error(request, f'Error downloading report: {str(e)}')
return redirect('utils:utils_reports_dashboard')
@login_required
def loan_statement_view(request, loan_id):
"""View loan statement with download options"""
try:
from .reports import OptimizedReportGenerator, log_report_generation
loan = get_object_or_404(Loan, id=loan_id)
# Check permissions
if not request.user.is_staff and loan.borrower != request.user:
raise PermissionDenied("You don't have permission to view this loan statement")
# Generate statement data
statement_data = OptimizedReportGenerator.generate_loan_statement(loan_id, request.user)
if 'error' in statement_data:
messages.error(request, f'Error generating statement: {statement_data["error"]}')
return redirect('loans:loan_detail', loan_id=loan_id)
context = {
'loan': loan,
'statement_data': statement_data,
'can_download': True
}
return render(request, 'utils/loan_statement.html', context)
except Exception as e:
log_report_generation(request.user, 'loan_statement_view_error', {'loan_id': loan_id}, False, str(e))
messages.error(request, f'Error loading loan statement: {str(e)}')
return redirect('loans:loans')
@login_required
def client_statement_view(request, client_id):
"""View client statement with download options"""
try:
from .reports import OptimizedReportGenerator, log_report_generation
client = get_object_or_404(CustomUser, id=client_id)
# Check permissions
if not request.user.is_staff and client != request.user:
raise PermissionDenied("You don't have permission to view this client statement")
# Generate statement data
statement_data = OptimizedReportGenerator.generate_client_statement(client_id, request.user)
if 'error' in statement_data:
messages.error(request, f'Error generating statement: {statement_data["error"]}')
return redirect('users:client_detail', client_id=client_id)
context = {
'client': client,
'statement_data': statement_data,
'can_download': True
}
return render(request, 'utils/client_statement.html', context)
except Exception as e:
log_report_generation(request.user, 'client_statement_view_error', {'client_id': client_id}, False, str(e))
messages.error(request, f'Error loading client statement: {str(e)}')
return redirect('users:client_list')
def test_media_serving(request):
"""Test endpoint to verify media file serving"""
import os
from django.conf import settings
from django.http import JsonResponse
# Check if PHOTO_GIDEON.jpeg exists
test_file = os.path.join(settings.MEDIA_ROOT, 'kyc/selfies/PHOTO_GIDEON.jpeg')
return JsonResponse({
'media_root': settings.MEDIA_ROOT,
'media_url': settings.MEDIA_URL,
'test_file_exists': os.path.exists(test_file),
'test_file_path': test_file,
'test_url': f"{settings.MEDIA_URL}kyc/selfies/PHOTO_GIDEON.jpeg"
})