""" Django views for advanced filtering and sorting functionality """ from django.http import JsonResponse, HttpResponse from django.views.decorators.http import require_http_methods from django.contrib.auth.decorators import login_required from django.views.decorators.csrf import csrf_exempt from django.utils.decorators import method_decorator from django.views import View from django.shortcuts import get_object_or_404 from django.core.paginator import Paginator from django.db.models import Q import json import logging from typing import Dict, List, Any, Optional from .advanced_filtering_service import AdvancedFilteringService, FilterPreset, SavedSearch from .export_functions import EnhancedExcelExportService logger = logging.getLogger(__name__) class AdvancedFilteringView(View): """Base view for advanced filtering functionality""" def __init__(self): super().__init__() self.filtering_service = AdvancedFilteringService() @method_decorator(login_required) def dispatch(self, request, *args, **kwargs): return super().dispatch(request, *args, **kwargs) class FilterOptionsView(AdvancedFilteringView): """API endpoint to get filter options for a specific field""" def get(self, request, report_type: str, field_name: str): """Get available filter options for a field""" try: # Get base queryset based on report type queryset = self._get_base_queryset(report_type, request.user) # Get filter options options = self.filtering_service.get_filter_options( report_type, field_name, queryset ) return JsonResponse(options, safe=False) except Exception as e: logger.error(f"Error getting filter options: {str(e)}") return JsonResponse({'error': str(e)}, status=500) def _get_base_queryset(self, report_type: str, user): """Get base queryset for the report type""" # This would be implemented based on your specific models # For now, returning a placeholder from loans.models import Loan # Adjust import based on your models if report_type in ['loans_due', 'loans_in_arrears', 'delinquent_loans']: return Loan.objects.filter( # Add user-specific filtering based on permissions # For example: portfolio_manager=user if user is loan officer ) return Loan.objects.none() class FilteredReportView(AdvancedFilteringView): """API endpoint to get filtered report data""" def get(self, request, report_type: str): """Get filtered report data""" try: # Parse filter parameters filters = json.loads(request.GET.get('filters', '{}')) search = json.loads(request.GET.get('search', '{}')) sort = json.loads(request.GET.get('sort', '[]')) # Get base queryset queryset = self._get_base_queryset(report_type, request.user) # Apply filters if filters: queryset = self.filtering_service.apply_advanced_filters( queryset, filters, report_type ) # Apply search if search.get('query'): queryset = self.filtering_service.perform_advanced_search( queryset, search, report_type ) # Apply sorting if sort: queryset = self.filtering_service.apply_advanced_sorting( queryset, sort, report_type ) # Pagination page = int(request.GET.get('page', 1)) per_page = int(request.GET.get('per_page', 50)) paginator = Paginator(queryset, per_page) page_obj = paginator.get_page(page) # Serialize data data = { 'count': paginator.count, 'num_pages': paginator.num_pages, 'current_page': page, 'results': self._serialize_results(page_obj.object_list, report_type) } # Save persistent filters session_key = request.session.session_key or 'default' self.filtering_service.save_persistent_filters( request.user, report_type, session_key, filters ) self.filtering_service.save_persistent_sort( request.user, report_type, session_key, sort ) return JsonResponse(data) except Exception as e: logger.error(f"Error getting filtered report data: {str(e)}") return JsonResponse({'error': str(e)}, status=500) def _get_base_queryset(self, report_type: str, user): """Get base queryset for the report type""" # Import your models here from loans.models import Loan base_queryset = Loan.objects.all() # Apply user-specific filtering based on permissions # This would depend on your permission system if hasattr(user, 'branch') and user.branch: base_queryset = base_queryset.filter(branch=user.branch) if hasattr(user, 'role') and user.role == 'loan_officer': base_queryset = base_queryset.filter(portfolio_manager=user) return base_queryset def _serialize_results(self, queryset, report_type: str) -> List[Dict[str, Any]]: """Serialize queryset results for JSON response""" results = [] for obj in queryset: if report_type == 'loans_due': results.append({ 'id': obj.id, 'loan_number': getattr(obj, 'loan_number', ''), 'borrower_name': getattr(obj, 'borrower_name', ''), 'borrower_phone': getattr(obj, 'borrower_phone', ''), 'due_date': getattr(obj, 'due_date', None), 'due_amount': float(getattr(obj, 'due_amount', 0)), 'loan_product': getattr(obj, 'loan_product', ''), 'repayment_method': getattr(obj, 'repayment_method', ''), 'portfolio_manager': str(getattr(obj, 'portfolio_manager', '')), }) elif report_type == 'loans_in_arrears': results.append({ 'id': obj.id, 'loan_number': getattr(obj, 'loan_number', ''), 'borrower_name': getattr(obj, 'borrower_name', ''), 'borrower_phone': getattr(obj, 'borrower_phone', ''), 'arrears_amount': float(getattr(obj, 'arrears_amount', 0)), 'days_overdue': getattr(obj, 'days_overdue', 0), 'risk_level': getattr(obj, 'risk_level', ''), 'loan_product': getattr(obj, 'loan_product', ''), 'last_payment_date': getattr(obj, 'last_payment_date', None), }) # Add more report types as needed return results class FilterPresetView(AdvancedFilteringView): """API endpoint for managing filter presets""" def get(self, request, report_type: str = None, preset_id: int = None): """Get filter presets""" try: if preset_id: # Get specific preset preset = get_object_or_404(FilterPreset, id=preset_id, user=request.user) data = { 'id': preset.id, 'name': preset.name, 'filters': preset.filters, 'is_default': preset.is_default, 'created_at': preset.created_at.isoformat() } return JsonResponse(data) elif report_type: # Get presets for report type presets = self.filtering_service.get_filter_presets(request.user, report_type) data = [ { 'id': preset.id, 'name': preset.name, 'filters': preset.filters, 'is_default': preset.is_default, 'is_shared': preset.is_shared, 'created_at': preset.created_at.isoformat() } for preset in presets ] return JsonResponse(data, safe=False) else: return JsonResponse({'error': 'Report type required'}, status=400) except Exception as e: logger.error(f"Error getting filter presets: {str(e)}") return JsonResponse({'error': str(e)}, status=500) def post(self, request): """Create new filter preset""" try: data = json.loads(request.body) preset = self.filtering_service.save_filter_preset( user=request.user, name=data['name'], report_type=data['report_type'], filters=data['filters'], is_default=data.get('is_default', False) ) response_data = { 'id': preset.id, 'name': preset.name, 'filters': preset.filters, 'is_default': preset.is_default, 'created_at': preset.created_at.isoformat() } return JsonResponse(response_data, status=201) except Exception as e: logger.error(f"Error creating filter preset: {str(e)}") return JsonResponse({'error': str(e)}, status=500) def delete(self, request, preset_id: int): """Delete filter preset""" try: preset = get_object_or_404(FilterPreset, id=preset_id, user=request.user) preset.delete() return JsonResponse({'success': True}) except Exception as e: logger.error(f"Error deleting filter preset: {str(e)}") return JsonResponse({'error': str(e)}, status=500) class SavedSearchView(AdvancedFilteringView): """API endpoint for managing saved searches""" def get(self, request, report_type: str = None, search_id: int = None): """Get saved searches""" try: if search_id: # Get specific search search = get_object_or_404(SavedSearch, id=search_id, user=request.user) data = { 'id': search.id, 'name': search.name, 'search_criteria': search.search_criteria, 'result_count': search.result_count, 'last_executed': search.last_executed.isoformat(), 'created_at': search.created_at.isoformat() } return JsonResponse(data) elif report_type: # Get searches for report type searches = self.filtering_service.get_saved_searches(request.user, report_type) data = [ { 'id': search.id, 'name': search.name, 'search_criteria': search.search_criteria, 'result_count': search.result_count, 'last_executed': search.last_executed.isoformat(), 'created_at': search.created_at.isoformat() } for search in searches ] return JsonResponse(data, safe=False) else: return JsonResponse({'error': 'Report type required'}, status=400) except Exception as e: logger.error(f"Error getting saved searches: {str(e)}") return JsonResponse({'error': str(e)}, status=500) def post(self, request): """Create new saved search""" try: data = json.loads(request.body) search = self.filtering_service.save_search( user=request.user, name=data['name'], report_type=data['report_type'], search_criteria=data['search_criteria'], result_count=data.get('result_count', 0) ) response_data = { 'id': search.id, 'name': search.name, 'search_criteria': search.search_criteria, 'result_count': search.result_count, 'created_at': search.created_at.isoformat() } return JsonResponse(response_data, status=201) except Exception as e: logger.error(f"Error creating saved search: {str(e)}") return JsonResponse({'error': str(e)}, status=500) def delete(self, request, search_id: int): """Delete saved search""" try: search = get_object_or_404(SavedSearch, id=search_id, user=request.user) search.delete() return JsonResponse({'success': True}) except Exception as e: logger.error(f"Error deleting saved search: {str(e)}") return JsonResponse({'error': str(e)}, status=500) class ExportFilteredReportView(AdvancedFilteringView): """API endpoint to export filtered report data""" def get(self, request, report_type: str): """Export filtered report data""" try: # Parse filter parameters filters = json.loads(request.GET.get('filters', '{}')) search = json.loads(request.GET.get('search', '{}')) sort = json.loads(request.GET.get('sort', '[]')) export_format = request.GET.get('export', 'excel') # Get base queryset queryset = self._get_base_queryset(report_type, request.user) # Apply filters if filters: queryset = self.filtering_service.apply_advanced_filters( queryset, filters, report_type ) # Apply search if search.get('query'): queryset = self.filtering_service.perform_advanced_search( queryset, search, report_type ) # Apply sorting if sort: queryset = self.filtering_service.apply_advanced_sorting( queryset, sort, report_type ) # Prepare report data report_data = { 'summary': self._generate_summary(queryset, report_type), 'loans': self._serialize_for_export(queryset, report_type), 'filters_applied': filters, 'search_applied': search, 'sort_applied': sort } # Generate export if export_format == 'excel': export_service = EnhancedExcelExportService() return export_service.create_multi_worksheet_export( report_data, report_type, filters ) else: return JsonResponse({'error': 'Unsupported export format'}, status=400) except Exception as e: logger.error(f"Error exporting filtered report: {str(e)}") return JsonResponse({'error': str(e)}, status=500) def _get_base_queryset(self, report_type: str, user): """Get base queryset for the report type""" # Same as FilteredReportView from loans.models import Loan base_queryset = Loan.objects.all() if hasattr(user, 'branch') and user.branch: base_queryset = base_queryset.filter(branch=user.branch) if hasattr(user, 'role') and user.role == 'loan_officer': base_queryset = base_queryset.filter(portfolio_manager=user) return base_queryset def _generate_summary(self, queryset, report_type: str) -> Dict[str, Any]: """Generate summary statistics for the filtered data""" summary = { 'total_loans': queryset.count(), 'total_amount': 0, 'average_amount': 0 } if report_type == 'loans_due': total_due = sum(float(getattr(loan, 'due_amount', 0)) for loan in queryset) summary.update({ 'total_due_amount': total_due, 'average_due_amount': total_due / summary['total_loans'] if summary['total_loans'] > 0 else 0 }) elif report_type == 'loans_in_arrears': total_arrears = sum(float(getattr(loan, 'arrears_amount', 0)) for loan in queryset) summary.update({ 'total_arrears_amount': total_arrears, 'average_arrears_amount': total_arrears / summary['total_loans'] if summary['total_loans'] > 0 else 0 }) return summary def _serialize_for_export(self, queryset, report_type: str) -> List[Dict[str, Any]]: """Serialize queryset for export""" results = [] for obj in queryset: if report_type == 'loans_due': results.append({ 'loan_number': getattr(obj, 'loan_number', ''), 'borrower_name': getattr(obj, 'borrower_name', ''), 'borrower_phone': getattr(obj, 'borrower_phone', ''), 'due_date': str(getattr(obj, 'due_date', '')), 'due_amount': float(getattr(obj, 'due_amount', 0)), 'loan_product': getattr(obj, 'loan_product', ''), 'repayment_method': getattr(obj, 'repayment_method', ''), 'portfolio_manager': str(getattr(obj, 'portfolio_manager', '')), }) elif report_type == 'loans_in_arrears': results.append({ 'loan_number': getattr(obj, 'loan_number', ''), 'borrower_name': getattr(obj, 'borrower_name', ''), 'borrower_phone': getattr(obj, 'borrower_phone', ''), 'arrears_amount': float(getattr(obj, 'arrears_amount', 0)), 'days_overdue': getattr(obj, 'days_overdue', 0), 'risk_level': getattr(obj, 'risk_level', ''), 'loan_product': getattr(obj, 'loan_product', ''), 'last_payment_date': str(getattr(obj, 'last_payment_date', '')), }) return results # URL patterns would be added to urls.py: """ from django.urls import path from .advanced_filtering_views import ( FilterOptionsView, FilteredReportView, FilterPresetView, SavedSearchView, ExportFilteredReportView ) urlpatterns = [ path('api/filter-options///', FilterOptionsView.as_view(), name='filter_options'), path('api/reports//filtered/', FilteredReportView.as_view(), name='filtered_report'), path('api/filter-presets/', FilterPresetView.as_view(), name='filter_presets'), path('api/filter-presets//', FilterPresetView.as_view(), name='filter_presets_by_type'), path('api/filter-presets//', FilterPresetView.as_view(), name='filter_preset_detail'), path('api/saved-searches/', SavedSearchView.as_view(), name='saved_searches'), path('api/saved-searches//', SavedSearchView.as_view(), name='saved_searches_by_type'), path('api/saved-searches//', SavedSearchView.as_view(), name='saved_search_detail'), path('api/reports//export/', ExportFilteredReportView.as_view(), name='export_filtered_report'), path('api/filter-suggestions//', FilterSuggestionsView.as_view(), name='filter_suggestions'), path('api/filter-analytics//', FilterAnalyticsView.as_view(), name='filter_analytics'), path('api/filter-groups//', DynamicFilterGroupsView.as_view(), name='dynamic_filter_groups'), ] """ class FilterSuggestionsView(AdvancedFilteringView): """API endpoint for smart filter suggestions""" def get(self, request, report_type: str): """Get smart filter suggestions""" try: current_filters = json.loads(request.GET.get('filters', '{}')) suggestions = self.filtering_service.create_smart_filter_suggestions( request.user, report_type, current_filters ) return JsonResponse(suggestions, safe=False) except Exception as e: logger.error(f"Error getting filter suggestions: {str(e)}") return JsonResponse({'error': str(e)}, status=500) class FilterAnalyticsView(AdvancedFilteringView): """API endpoint for filter usage analytics""" def get(self, request, report_type: str): """Get filter analytics""" try: # Get date range from query parameters start_date = request.GET.get('start_date') end_date = request.GET.get('end_date') date_range = {} if start_date: date_range['start'] = start_date if end_date: date_range['end'] = end_date # Get performance stats performance_stats = self.filtering_service.get_filter_performance_stats( request.user, report_type ) # Get analytics report if date range provided analytics_report = {} if date_range: analytics_report = self.filtering_service.create_filter_analytics_report( request.user, report_type, date_range ) data = { 'performance_stats': performance_stats, 'analytics_report': analytics_report } return JsonResponse(data) except Exception as e: logger.error(f"Error getting filter analytics: {str(e)}") return JsonResponse({'error': str(e)}, status=500) class DynamicFilterGroupsView(AdvancedFilteringView): """API endpoint for dynamic filter groups""" def get(self, request, report_type: str): """Get dynamic filter groups based on data distribution""" try: # Get base queryset queryset = self._get_base_queryset(report_type, request.user) # Create dynamic filter groups groups = self.filtering_service.create_dynamic_filter_groups( queryset, report_type ) return JsonResponse(groups) except Exception as e: logger.error(f"Error getting dynamic filter groups: {str(e)}") return JsonResponse({'error': str(e)}, status=500) def _get_base_queryset(self, report_type: str, user): """Get base queryset for the report type""" from loans.models import Loan base_queryset = Loan.objects.all() if hasattr(user, 'branch') and user.branch: base_queryset = base_queryset.filter(branch=user.branch) if hasattr(user, 'role') and user.role == 'loan_officer': base_queryset = base_queryset.filter(portfolio_manager=user) return base_queryset class BulkFilterView(AdvancedFilteringView): """API endpoint for applying bulk filters""" def post(self, request, report_type: str): """Apply bulk filters""" try: data = json.loads(request.body) bulk_filters = data.get('bulk_filters', []) # Get base queryset queryset = self._get_base_queryset(report_type, request.user) # Apply bulk filters filtered_queryset = self.filtering_service.apply_bulk_filters( queryset, bulk_filters, report_type ) # Pagination page = int(request.GET.get('page', 1)) per_page = int(request.GET.get('per_page', 50)) paginator = Paginator(filtered_queryset, per_page) page_obj = paginator.get_page(page) # Serialize data response_data = { 'count': paginator.count, 'num_pages': paginator.num_pages, 'current_page': page, 'results': self._serialize_results(page_obj.object_list, report_type) } return JsonResponse(response_data) except Exception as e: logger.error(f"Error applying bulk filters: {str(e)}") return JsonResponse({'error': str(e)}, status=500) def _get_base_queryset(self, report_type: str, user): """Get base queryset for the report type""" from loans.models import Loan base_queryset = Loan.objects.all() if hasattr(user, 'branch') and user.branch: base_queryset = base_queryset.filter(branch=user.branch) if hasattr(user, 'role') and user.role == 'loan_officer': base_queryset = base_queryset.filter(portfolio_manager=user) return base_queryset def _serialize_results(self, queryset, report_type: str) -> List[Dict[str, Any]]: """Serialize queryset results for JSON response""" results = [] for obj in queryset: if report_type == 'loans_due': results.append({ 'id': obj.id, 'loan_number': getattr(obj, 'loan_number', ''), 'borrower_name': getattr(obj, 'borrower_name', ''), 'borrower_phone': getattr(obj, 'borrower_phone', ''), 'due_date': getattr(obj, 'due_date', None), 'due_amount': float(getattr(obj, 'due_amount', 0)), 'loan_product': getattr(obj, 'loan_product', ''), 'repayment_method': getattr(obj, 'repayment_method', ''), 'portfolio_manager': str(getattr(obj, 'portfolio_manager', '')), }) elif report_type == 'loans_in_arrears': results.append({ 'id': obj.id, 'loan_number': getattr(obj, 'loan_number', ''), 'borrower_name': getattr(obj, 'borrower_name', ''), 'borrower_phone': getattr(obj, 'borrower_phone', ''), 'arrears_amount': float(getattr(obj, 'arrears_amount', 0)), 'days_overdue': getattr(obj, 'days_overdue', 0), 'risk_level': getattr(obj, 'risk_level', ''), 'loan_product': getattr(obj, 'loan_product', ''), 'last_payment_date': getattr(obj, 'last_payment_date', None), }) return results class FilterPerformanceView(AdvancedFilteringView): """API endpoint for filter performance optimization""" def post(self, request, report_type: str): """Optimize filter performance""" try: data = json.loads(request.body) filters = data.get('filters', {}) # Get base queryset queryset = self._get_base_queryset(report_type, request.user) # Optimize filters and get performance info optimized_queryset, optimization_info = self.filtering_service.optimize_filter_performance( queryset, filters, report_type ) # Serialize first 10 results for preview preview_results = self._serialize_results( optimized_queryset[:10], report_type ) response_data = { 'optimization_info': optimization_info, 'preview_results': preview_results } return JsonResponse(response_data) except Exception as e: logger.error(f"Error optimizing filter performance: {str(e)}") return JsonResponse({'error': str(e)}, status=500) def _get_base_queryset(self, report_type: str, user): """Get base queryset for the report type""" from loans.models import Loan base_queryset = Loan.objects.all() if hasattr(user, 'branch') and user.branch: base_queryset = base_queryset.filter(branch=user.branch) if hasattr(user, 'role') and user.role == 'loan_officer': base_queryset = base_queryset.filter(portfolio_manager=user) return base_queryset def _serialize_results(self, queryset, report_type: str) -> List[Dict[str, Any]]: """Serialize queryset results for JSON response""" results = [] for obj in queryset: if report_type == 'loans_due': results.append({ 'id': obj.id, 'loan_number': getattr(obj, 'loan_number', ''), 'borrower_name': getattr(obj, 'borrower_name', ''), 'due_amount': float(getattr(obj, 'due_amount', 0)), }) elif report_type == 'loans_in_arrears': results.append({ 'id': obj.id, 'loan_number': getattr(obj, 'loan_number', ''), 'borrower_name': getattr(obj, 'borrower_name', ''), 'arrears_amount': float(getattr(obj, 'arrears_amount', 0)), }) return results