from django.shortcuts import render, get_object_or_404, redirect
from django.contrib.auth.decorators import login_required, user_passes_test
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 .reports import 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
import magic
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
User = get_user_model()
def is_admin(user):
return user.is_staff
@login_required
def documents(request):
"""View for managing documents with actual document counts and sorting"""
if request.method == 'POST':
document_type = request.POST.get('document_type')
files = request.FILES.getlist('files')
tags = request.POST.get('tags', '').split(',')
# Get file size limit from settings
max_file_size = SystemSetting.get_int('max_file_size_mb', 10) * 1024 * 1024 # Convert MB to bytes
allowed_types = SystemSetting.get_setting('allowed_file_types', '.pdf,.doc,.docx,.jpg,.jpeg,.png').split(',')
# Process tags
tag_objects = []
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': request.user}
)
tag_objects.append(tag)
for file in files:
# Validate file size
if file.size > max_file_size:
messages.error(request, f'File {file.name} exceeds maximum size of {max_file_size/1024/1024}MB')
continue
# Validate file type
ext = f".{file.name.split('.')[-1]}" if '.' in file.name else ''
if ext.lower() not in allowed_types:
messages.error(request, f'File type {ext} not allowed. Allowed types: {", ".join(allowed_types)}')
continue
# Create document
document = Document.objects.create(
name=file.name,
file=file,
document_type=document_type,
uploaded_by=request.user,
mime_type=magic.from_buffer(file.read(1024), mime=True)
)
# Add tags
document.tags.add(*tag_objects)
# Generate thumbnail for images
if document.mime_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: {file.name}'
)
# Create notification for admins (only if notifications are enabled)
if document_type in ['loan_agreement', 'contract']:
# Check if notifications are enabled
from utils.models import SystemSetting
if SystemSetting.get_bool('email_notifications', True):
Notification.objects.create(
title='New Important Document',
message=f'A new {document_type} has been uploaded: {file.name}',
priority='high'
)
messages.success(request, f'Document {file.name} uploaded successfully')
# Get filter parameters
sort_by = request.GET.get('sort', 'newest') # newest, oldest, largest, smallest, name
search_query = request.GET.get('search', '')
# Get all documents from Document model
documents = {}
total_documents = 0
# Get documents from Document model
for doc_type, doc_name in Document.DOCUMENT_TYPES:
queryset = Document.objects.filter(
Q(document_type=doc_type) &
(Q(uploaded_by=request.user) | Q(shared_with=request.user) | Q(is_public=True))
).prefetch_related('tags').select_related('uploaded_by')
# Apply search filter
if search_query:
queryset = queryset.filter(
Q(name__icontains=search_query) |
Q(description__icontains=search_query) |
Q(tags__name__icontains=search_query)
).distinct()
# Apply sorting
if sort_by == 'newest':
queryset = queryset.order_by('-created_at')
elif sort_by == 'oldest':
queryset = queryset.order_by('created_at')
elif sort_by == 'largest':
queryset = queryset.order_by('-file_size')
elif sort_by == 'smallest':
queryset = queryset.order_by('file_size')
elif sort_by == 'name':
queryset = queryset.order_by('name')
documents[doc_type] = queryset
total_documents += queryset.count()
# Get KYC documents from CustomUser model (for admin users)
kyc_documents = {}
if request.user.is_staff:
from users.models import CustomUser
# Get all borrowers with KYC documents
borrowers = CustomUser.objects.filter(role='borrower')
for borrower in borrowers:
# Check each KYC field
kyc_fields = {
'id_document': borrower.id_document,
'selfie': borrower.selfie,
'utility_bill': borrower.utility_bill,
'bank_statement': borrower.bank_statement,
'business_license': borrower.business_license,
'tax_certificate': borrower.tax_certificate,
}
for field_name, field_value in kyc_fields.items():
if field_value:
if field_name not in kyc_documents:
kyc_documents[field_name] = []
# Create a document-like object for KYC documents
kyc_doc = {
'id': f'kyc_{borrower.id}_{field_name}',
'name': f'{borrower.get_full_name()} - {field_name.replace("_", " ").title()}',
'file': field_value,
'uploaded_by': borrower,
'created_at': borrower.created_at,
'document_type': field_name,
'source': 'kyc',
'client_name': borrower.get_full_name(),
'client_id': borrower.id
}
kyc_documents[field_name].append(kyc_doc)
total_documents += 1
# Combine Document model documents with KYC documents
combined_documents = {}
for doc_type, doc_name in Document.DOCUMENT_TYPES:
combined_documents[doc_type] = list(documents[doc_type])
# Add KYC documents if they match the document type
if doc_type in kyc_documents:
combined_documents[doc_type].extend(kyc_documents[doc_type])
context = {
'documents': combined_documents,
'document_types': Document.DOCUMENT_TYPES,
'tags': DocumentTag.objects.all(),
'total_documents': total_documents,
'sort_options': [
('newest', 'Newest First'),
('oldest', 'Oldest First'),
('largest', 'Largest First'),
('smallest', 'Smallest First'),
('name', 'Name A-Z'),
],
'current_sort': sort_by,
'search_query': search_query,
}
return render(request, 'utils/documents.html', context)
@login_required
def share_document(request):
"""API endpoint to share documents with users"""
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"""
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
zip_filename = f'documents_{timezone.now().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
def notifications(request):
"""View for managing notifications with comprehensive system information"""
if request.method == 'POST':
notification_id = request.POST.get('notification_id')
if notification_id == 'all':
Notification.objects.filter(user=request.user, read=False).update(read=True)
messages.success(request, 'All notifications marked as read')
else:
notification = get_object_or_404(Notification, id=notification_id, user=request.user)
notification.read = True
notification.save()
messages.success(request, 'Notification marked as read')
return redirect('utils:notifications')
# Get user notifications and mark them as read
user_notifications = Notification.objects.filter(
Q(user=request.user) | Q(user__isnull=True)
).order_by('-created_at')
# Mark all unread notifications as read
user_notifications.filter(read_at__isnull=True).update(read_at=django_timezone.now())
# Get comprehensive system information for admins
system_alerts = []
overdue_loans = []
pending_applications = []
recent_activities = []
loan_statistics = {}
user_statistics = {}
security_alerts = []
system_health = {}
if request.user.is_staff:
from datetime import timedelta
from loans.models import Loan, LoanApplication, Repayment
from users.models import CustomUser
# Overdue loans
overdue_loans = Loan.objects.filter(
status='active',
due_date__lt=django_timezone.now()
).select_related('borrower', 'application__loan_product')
# Pending loan applications
pending_applications = LoanApplication.objects.filter(
status='pending'
).select_related('borrower')
# Recent system activities (last 24 hours)
recent_activities = AuditLog.objects.filter(
created_at__gte=django_timezone.now() - timedelta(hours=24)
).select_related('user').order_by('-created_at')[:20]
# Loan statistics
total_loans = Loan.objects.count()
active_loans = Loan.objects.filter(status='active').count()
defaulted_loans = Loan.objects.filter(status='defaulted').count()
total_disbursed = Loan.objects.aggregate(total=Sum('principal_amount'))['total'] or 0
total_collected = Repayment.objects.aggregate(total=Sum('amount'))['total'] or 0
loan_statistics = {
'total_loans': total_loans,
'active_loans': active_loans,
'defaulted_loans': defaulted_loans,
'total_disbursed': total_disbursed,
'total_collected': total_collected,
'outstanding_amount': total_disbursed - total_collected,
'default_rate': (defaulted_loans / total_loans * 100) if total_loans > 0 else 0
}
# User statistics
total_users = CustomUser.objects.filter(role='borrower').count()
active_users = CustomUser.objects.filter(role='borrower', is_active=True).count()
new_users_today = CustomUser.objects.filter(
role='borrower',
date_joined__gte=django_timezone.now().replace(hour=0, minute=0, second=0, microsecond=0)
).count()
user_statistics = {
'total_users': total_users,
'active_users': active_users,
'new_users_today': new_users_today,
'inactive_users': total_users - active_users
}
# Generate comprehensive system alerts
if overdue_loans.exists():
system_alerts.append({
'type': 'overdue_loans',
'title': f'{overdue_loans.count()} Overdue Loans',
'message': f'There are {overdue_loans.count()} loans that are overdue for payment',
'priority': 'urgent',
'icon': 'fa-exclamation-triangle',
'count': overdue_loans.count(),
'action_url': '/loans/'
})
if pending_applications.exists():
system_alerts.append({
'type': 'pending_applications',
'title': f'{pending_applications.count()} Pending Applications',
'message': f'There are {pending_applications.count()} loan applications awaiting review',
'priority': 'high',
'icon': 'fa-clock',
'count': pending_applications.count(),
'action_url': '/loans/applications/'
})
# Check for low income users
low_income_users = CustomUser.objects.filter(
monthly_income__lt=50000
).count()
if low_income_users > 0:
system_alerts.append({
'type': 'low_income_users',
'title': f'{low_income_users} Low Income Users',
'message': f'{low_income_users} users have monthly income below KES 50,000',
'priority': 'medium',
'icon': 'fa-user',
'count': low_income_users,
'action_url': '/clients/'
})
# Check for recent failed login attempts
recent_failed_logins = AuditLog.objects.filter(
action='login',
created_at__gte=django_timezone.now() - timedelta(hours=24)
).count()
if recent_failed_logins > 5:
system_alerts.append({
'type': 'failed_logins',
'title': f'{recent_failed_logins} Failed Login Attempts',
'message': f'{recent_failed_logins} failed login attempts in the last 24 hours',
'priority': 'high',
'icon': 'fa-shield-alt',
'count': recent_failed_logins,
'action_url': '/utils/access-logs/'
})
# Security alerts
# Check for multiple login attempts from same IP
from django.db.models import Count
suspicious_ips = AuditLog.objects.filter(
action='login',
created_at__gte=django_timezone.now() - timedelta(hours=1)
).values('ip_address').annotate(
attempt_count=Count('id')
).filter(attempt_count__gt=10)
for ip_data in suspicious_ips:
if ip_data['ip_address']:
security_alerts.append({
'type': 'suspicious_ip',
'title': f'Suspicious Activity from {ip_data["ip_address"]}',
'message': f'{ip_data["attempt_count"]} login attempts from this IP in the last hour',
'priority': 'urgent',
'icon': 'fa-exclamation-triangle',
'timestamp': django_timezone.now()
})
# System health checks
# Check for loans with high default risk
high_risk_loans = Loan.objects.filter(
status='active',
due_date__lt=django_timezone.now() + timedelta(days=7)
).count()
if high_risk_loans > 0:
system_alerts.append({
'type': 'high_risk_loans',
'title': f'{high_risk_loans} High Risk Loans',
'message': f'{high_risk_loans} loans are due within the next 7 days',
'priority': 'medium',
'icon': 'fa-exclamation-circle',
'count': high_risk_loans,
'action_url': '/loans/'
})
# Check for system performance
total_audit_logs = AuditLog.objects.count()
recent_audit_logs = AuditLog.objects.filter(
created_at__gte=django_timezone.now() - timedelta(hours=1)
).count()
system_health = {
'total_audit_logs': total_audit_logs,
'recent_audit_logs': recent_audit_logs,
'system_load': 'normal' if recent_audit_logs < 100 else 'high',
'database_size': 'normal',
'last_backup': django_timezone.now() - timedelta(days=1)
}
# Check for document upload issues
from utils.models import Document
recent_documents = Document.objects.filter(
created_at__gte=django_timezone.now() - timedelta(hours=24)
).count()
if recent_documents == 0:
system_alerts.append({
'type': 'no_documents',
'title': 'No Document Uploads',
'message': 'No documents have been uploaded in the last 24 hours',
'priority': 'low',
'icon': 'fa-file-upload',
'count': 0,
'action_url': '/utils/documents/'
})
# Notification types for filter
notification_types = Notification.NOTIFICATION_TYPES
return render(request, 'utils/notifications.html', {
'notifications': user_notifications,
'notification_types': notification_types,
'system_alerts': system_alerts,
'overdue_loans': overdue_loans,
'pending_applications': pending_applications,
'recent_activities': recent_activities,
'loan_statistics': loan_statistics,
'user_statistics': user_statistics,
'security_alerts': security_alerts,
'system_health': system_health,
'is_admin': request.user.is_staff
})
@login_required
@user_passes_test(is_admin)
def settings(request):
"""View for managing system settings"""
if request.method == 'POST':
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')
# Get settings grouped by category
settings = {}
for category, _ in SystemSetting.CATEGORIES:
settings[category] = SystemSetting.objects.filter(category=category).order_by('key')
context = {
'settings': settings,
'categories': SystemSetting.CATEGORIES
}
return render(request, 'utils/settings.html', context)
@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)
if not isinstance(methods, list):
raise ValueError("Repayment methods must be a list")
value = json.dumps(methods)
except json.JSONDecodeError:
raise ValueError("Invalid repayment methods format")
else:
# Handle numeric values
value = float(value)
if value < 0:
raise ValueError("Loan settings must be positive numbers")
elif setting.category == 'notification':
value = str(value).lower() == 'true'
elif setting.category == 'system':
if 'max_file_size' in key.lower():
value = int(value)
if value <= 0:
raise ValueError("File size must be positive")
elif 'allowed_file_types' in key.lower():
try:
types = json.loads(value)
if not isinstance(types, list):
raise ValueError("File types must be a list")
value = json.dumps(types)
except json.JSONDecodeError:
raise ValueError("Invalid file types format")
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 = ReportGenerator.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 = ReportGenerator.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 = ReportGenerator.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 = ReportGenerator.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 = ReportGenerator.generate_system_report_pdf()
response = HttpResponse(buffer, content_type='application/pdf')
response['Content-Disposition'] = 'attachment; filename="system_report.pdf"'
else: # CSV
data = ReportGenerator.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 = ReportGenerator.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 = ReportGenerator.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"""
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"""
notifications = Notification.objects.filter(
Q(user=request.user) | Q(user__isnull=True)
).order_by('-created_at')[:5]
return JsonResponse({
'notifications': [{
'id': n.id,
'title': n.title,
'message': n.message,
'priority': n.priority,
'read': n.read,
'created_at': n.created_at.isoformat(),
'action_url': n.action_url
} for n in notifications]
})
@login_required
def get_unread_count(request):
"""API endpoint to get unread notification count"""
count = Notification.objects.filter(
Q(user=request.user) | Q(user__isnull=True),
read=False
).count()
return JsonResponse({'count': count})
@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"""
Notification.objects.filter(
Q(user=request.user) | Q(user__isnull=True),
read=False
).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=False
).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 datetime import timedelta
# Get recent activity from the last 24 hours
recent_activities = AuditLog.objects.filter(
created_at__gte=timezone.now() - 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 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
def all_customer_documents(request):
"""Comprehensive view for all customer documents uploaded by admins"""
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 all clients (borrowers)
clients = CustomUser.objects.filter(role='borrower').order_by('first_name', 'last_name')
# 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)