""" Advanced Filtering and Sorting Service Provides comprehensive filtering, sorting, and search capabilities with persistent settings """ from django.db import models from django.contrib.auth.models import User from django.core.cache import cache from django.db.models import Q, F, Value, Case, When, IntegerField, CharField from django.db.models.functions import Concat, Lower, Cast from typing import Dict, List, Any, Optional, Union, Tuple import json import logging from datetime import datetime, date, timedelta from decimal import Decimal import re logger = logging.getLogger(__name__) class FilterPreset(models.Model): """Model to store user filter presets""" user = models.ForeignKey(User, on_delete=models.CASCADE) name = models.CharField(max_length=100) report_type = models.CharField(max_length=50) filters = models.JSONField() is_default = models.BooleanField(default=False) is_shared = models.BooleanField(default=False) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: unique_together = ['user', 'name', 'report_type'] def __str__(self): return f"{self.user.username} - {self.name} ({self.report_type})" class SavedSearch(models.Model): """Model to store user saved searches""" user = models.ForeignKey(User, on_delete=models.CASCADE) name = models.CharField(max_length=100) report_type = models.CharField(max_length=50) search_criteria = models.JSONField() result_count = models.IntegerField(default=0) last_executed = models.DateTimeField(auto_now=True) created_at = models.DateTimeField(auto_now_add=True) class Meta: unique_together = ['user', 'name', 'report_type'] def __str__(self): return f"{self.user.username} - {self.name} ({self.report_type})" class AdvancedFilteringService: """ Service class for advanced filtering, sorting, and search functionality """ def __init__(self): """Initialize the filtering service""" self.cache_timeout = 3600 # 1 hour # Define filterable fields for each report type self.filterable_fields = { 'loans_due': { 'borrower_name': {'type': 'text', 'label': 'Borrower Name'}, 'loan_number': {'type': 'text', 'label': 'Loan Number'}, 'due_date': {'type': 'date_range', 'label': 'Due Date'}, 'due_amount': {'type': 'number_range', 'label': 'Due Amount'}, 'loan_product': {'type': 'select', 'label': 'Product Type'}, 'repayment_method': {'type': 'select', 'label': 'Repayment Method'}, 'portfolio_manager': {'type': 'select', 'label': 'Portfolio Manager'}, 'branch': {'type': 'select', 'label': 'Branch'}, 'urgency_level': {'type': 'select', 'label': 'Urgency Level'} }, 'loans_in_arrears': { 'borrower_name': {'type': 'text', 'label': 'Borrower Name'}, 'loan_number': {'type': 'text', 'label': 'Loan Number'}, 'arrears_amount': {'type': 'number_range', 'label': 'Arrears Amount'}, 'days_overdue': {'type': 'number_range', 'label': 'Days Overdue'}, 'risk_level': {'type': 'select', 'label': 'Risk Level'}, 'loan_product': {'type': 'select', 'label': 'Product Type'}, 'repayment_method': {'type': 'select', 'label': 'Repayment Method'}, 'last_payment_date': {'type': 'date_range', 'label': 'Last Payment Date'}, 'portfolio_manager': {'type': 'select', 'label': 'Portfolio Manager'} }, 'delinquent_loans': { 'borrower_name': {'type': 'text', 'label': 'Borrower Name'}, 'loan_number': {'type': 'text', 'label': 'Loan Number'}, 'days_overdue': {'type': 'number_range', 'label': 'Days Overdue'}, 'overdue_amount': {'type': 'number_range', 'label': 'Overdue Amount'}, 'aging_bucket': {'type': 'select', 'label': 'Aging Bucket'}, 'collection_status': {'type': 'select', 'label': 'Collection Status'}, 'loan_product': {'type': 'select', 'label': 'Product Type'}, 'portfolio_manager': {'type': 'select', 'label': 'Portfolio Manager'} }, 'processing_fees': { 'borrower_name': {'type': 'text', 'label': 'Borrower Name'}, 'loan_number': {'type': 'text', 'label': 'Loan Number'}, 'fee_amount': {'type': 'number_range', 'label': 'Fee Amount'}, 'fee_date': {'type': 'date_range', 'label': 'Fee Date'}, 'loan_product': {'type': 'select', 'label': 'Product Type'}, 'payment_status': {'type': 'select', 'label': 'Payment Status'}, 'branch': {'type': 'select', 'label': 'Branch'} }, 'interest_income': { 'borrower_name': {'type': 'text', 'label': 'Borrower Name'}, 'loan_number': {'type': 'text', 'label': 'Loan Number'}, 'interest_amount': {'type': 'number_range', 'label': 'Interest Amount'}, 'interest_date': {'type': 'date_range', 'label': 'Interest Date'}, 'loan_product': {'type': 'select', 'label': 'Product Type'}, 'interest_rate': {'type': 'number_range', 'label': 'Interest Rate'}, 'portfolio_manager': {'type': 'select', 'label': 'Portfolio Manager'} } } # Define sortable fields self.sortable_fields = { 'loans_due': [ 'borrower_name', 'due_date', 'due_amount', 'loan_number', 'loan_product', 'days_until_due' ], 'loans_in_arrears': [ 'borrower_name', 'arrears_amount', 'days_overdue', 'loan_number', 'risk_level', 'last_payment_date' ], 'delinquent_loans': [ 'borrower_name', 'days_overdue', 'overdue_amount', 'loan_number', 'collection_status' ], 'processing_fees': [ 'borrower_name', 'fee_amount', 'fee_date', 'loan_number', 'payment_status' ], 'interest_income': [ 'borrower_name', 'interest_amount', 'interest_date', 'loan_number', 'interest_rate' ] } def apply_advanced_filters(self, queryset, filters: Dict[str, Any], report_type: str) -> models.QuerySet: """ Apply advanced filters to queryset Args: queryset: Django queryset to filter filters: Dictionary of filter criteria report_type: Type of report being filtered Returns: Filtered queryset """ try: filterable_fields = self.filterable_fields.get(report_type, {}) for field_name, filter_value in filters.items(): if not filter_value or field_name not in filterable_fields: continue field_config = filterable_fields[field_name] field_type = field_config['type'] if field_type == 'text': queryset = self._apply_text_filter(queryset, field_name, filter_value) elif field_type == 'date_range': queryset = self._apply_date_range_filter(queryset, field_name, filter_value) elif field_type == 'number_range': queryset = self._apply_number_range_filter(queryset, field_name, filter_value) elif field_type == 'select': queryset = self._apply_select_filter(queryset, field_name, filter_value) return queryset except Exception as e: logger.error(f"Error applying advanced filters: {str(e)}") return queryset def _apply_text_filter(self, queryset, field_name: str, filter_value: str) -> models.QuerySet: """Apply text-based filter with multiple search options""" if isinstance(filter_value, dict): search_type = filter_value.get('type', 'contains') search_value = filter_value.get('value', '') else: search_type = 'contains' search_value = str(filter_value) if not search_value: return queryset # Build filter based on search type if search_type == 'exact': return queryset.filter(**{f"{field_name}__iexact": search_value}) elif search_type == 'starts_with': return queryset.filter(**{f"{field_name}__istartswith": search_value}) elif search_type == 'ends_with': return queryset.filter(**{f"{field_name}__iendswith": search_value}) elif search_type == 'regex': return queryset.filter(**{f"{field_name}__iregex": search_value}) else: # contains (default) return queryset.filter(**{f"{field_name}__icontains": search_value}) def _apply_date_range_filter(self, queryset, field_name: str, filter_value: Dict[str, Any]) -> models.QuerySet: """Apply date range filter""" start_date = filter_value.get('start') end_date = filter_value.get('end') if start_date: if isinstance(start_date, str): start_date = datetime.strptime(start_date, '%Y-%m-%d').date() queryset = queryset.filter(**{f"{field_name}__gte": start_date}) if end_date: if isinstance(end_date, str): end_date = datetime.strptime(end_date, '%Y-%m-%d').date() queryset = queryset.filter(**{f"{field_name}__lte": end_date}) return queryset def _apply_number_range_filter(self, queryset, field_name: str, filter_value: Dict[str, Any]) -> models.QuerySet: """Apply number range filter""" min_value = filter_value.get('min') max_value = filter_value.get('max') if min_value is not None: queryset = queryset.filter(**{f"{field_name}__gte": min_value}) if max_value is not None: queryset = queryset.filter(**{f"{field_name}__lte": max_value}) return queryset def _apply_select_filter(self, queryset, field_name: str, filter_value) -> models.QuerySet: """Apply select/choice filter""" if isinstance(filter_value, list): return queryset.filter(**{f"{field_name}__in": filter_value}) else: return queryset.filter(**{field_name: filter_value}) def apply_advanced_sorting(self, queryset, sort_criteria: List[Dict[str, str]], report_type: str) -> models.QuerySet: """ Apply advanced sorting with multiple columns Args: queryset: Django queryset to sort sort_criteria: List of sort criteria [{'field': 'field_name', 'direction': 'asc/desc'}] report_type: Type of report being sorted Returns: Sorted queryset """ try: sortable_fields = self.sortable_fields.get(report_type, []) order_by_fields = [] for criteria in sort_criteria: field_name = criteria.get('field') direction = criteria.get('direction', 'asc') if field_name not in sortable_fields: continue # Add direction prefix if direction == 'desc': field_name = f"-{field_name}" order_by_fields.append(field_name) if order_by_fields: return queryset.order_by(*order_by_fields) return queryset except Exception as e: logger.error(f"Error applying advanced sorting: {str(e)}") return queryset def perform_advanced_search(self, queryset, search_criteria: Dict[str, Any], report_type: str) -> models.QuerySet: """ Perform advanced search with multiple criteria Args: queryset: Django queryset to search search_criteria: Dictionary of search criteria report_type: Type of report being searched Returns: Filtered queryset """ try: search_query = search_criteria.get('query', '') search_fields = search_criteria.get('fields', []) search_operator = search_criteria.get('operator', 'OR') # OR/AND if not search_query or not search_fields: return queryset # Build Q objects for each field q_objects = [] for field in search_fields: q_objects.append(Q(**{f"{field}__icontains": search_query})) # Combine Q objects based on operator if search_operator == 'AND': combined_q = q_objects[0] for q in q_objects[1:]: combined_q &= q else: # OR (default) combined_q = q_objects[0] for q in q_objects[1:]: combined_q |= q return queryset.filter(combined_q) except Exception as e: logger.error(f"Error performing advanced search: {str(e)}") return queryset def save_filter_preset(self, user, name: str, report_type: str, filters: Dict[str, Any], is_default: bool = False) -> FilterPreset: """Save filter preset for user""" try: # Remove existing default if setting new default if is_default: FilterPreset.objects.filter( user=user, report_type=report_type, is_default=True ).update(is_default=False) # Create or update preset preset, created = FilterPreset.objects.update_or_create( user=user, name=name, report_type=report_type, defaults={ 'filters': filters, 'is_default': is_default } ) return preset except Exception as e: logger.error(f"Error saving filter preset: {str(e)}") raise def get_filter_presets(self, user, report_type: str) -> List[FilterPreset]: """Get filter presets for user and report type""" try: return FilterPreset.objects.filter( user=user, report_type=report_type ).order_by('-is_default', 'name') except Exception as e: logger.error(f"Error getting filter presets: {str(e)}") return [] def get_default_filter_preset(self, user, report_type: str) -> Optional[FilterPreset]: """Get default filter preset for user and report type""" try: return FilterPreset.objects.filter( user=user, report_type=report_type, is_default=True ).first() except Exception as e: logger.error(f"Error getting default filter preset: {str(e)}") return None def save_search(self, user, name: str, report_type: str, search_criteria: Dict[str, Any], result_count: int = 0) -> SavedSearch: """Save search criteria for user""" try: search, created = SavedSearch.objects.update_or_create( user=user, name=name, report_type=report_type, defaults={ 'search_criteria': search_criteria, 'result_count': result_count } ) return search except Exception as e: logger.error(f"Error saving search: {str(e)}") raise def get_saved_searches(self, user, report_type: str) -> List[SavedSearch]: """Get saved searches for user and report type""" try: return SavedSearch.objects.filter( user=user, report_type=report_type ).order_by('-last_executed') except Exception as e: logger.error(f"Error getting saved searches: {str(e)}") return [] def get_filter_options(self, report_type: str, field_name: str, queryset=None) -> List[Dict[str, Any]]: """Get available filter options for a field""" try: cache_key = f"filter_options_{report_type}_{field_name}" cached_options = cache.get(cache_key) if cached_options is not None: return cached_options options = [] # Get distinct values for select fields if queryset and field_name in self.filterable_fields.get(report_type, {}): field_config = self.filterable_fields[report_type][field_name] if field_config['type'] == 'select': distinct_values = queryset.values_list(field_name, flat=True).distinct() options = [ {'value': value, 'label': str(value)} for value in distinct_values if value is not None ] # Cache options for 1 hour cache.set(cache_key, options, self.cache_timeout) return options except Exception as e: logger.error(f"Error getting filter options: {str(e)}") return [] def get_persistent_filters(self, user, report_type: str, session_key: str) -> Dict[str, Any]: """Get persistent filters from cache or default preset""" try: # Try to get from cache first cache_key = f"persistent_filters_{user.id}_{report_type}_{session_key}" cached_filters = cache.get(cache_key) if cached_filters is not None: return cached_filters # Fall back to default preset default_preset = self.get_default_filter_preset(user, report_type) if default_preset: return default_preset.filters return {} except Exception as e: logger.error(f"Error getting persistent filters: {str(e)}") return {} def save_persistent_filters(self, user, report_type: str, session_key: str, filters: Dict[str, Any]): """Save filters to cache for persistence across page refreshes""" try: cache_key = f"persistent_filters_{user.id}_{report_type}_{session_key}" cache.set(cache_key, filters, self.cache_timeout) except Exception as e: logger.error(f"Error saving persistent filters: {str(e)}") def get_persistent_sort(self, user, report_type: str, session_key: str) -> List[Dict[str, str]]: """Get persistent sort settings""" try: cache_key = f"persistent_sort_{user.id}_{report_type}_{session_key}" cached_sort = cache.get(cache_key) if cached_sort is not None: return cached_sort # Default sort return [{'field': 'created_at', 'direction': 'desc'}] except Exception as e: logger.error(f"Error getting persistent sort: {str(e)}") return [] def save_persistent_sort(self, user, report_type: str, session_key: str, sort_criteria: List[Dict[str, str]]): """Save sort settings for persistence""" try: cache_key = f"persistent_sort_{user.id}_{report_type}_{session_key}" cache.set(cache_key, sort_criteria, self.cache_timeout) except Exception as e: logger.error(f"Error saving persistent sort: {str(e)}") def build_filter_summary(self, filters: Dict[str, Any], report_type: str) -> str: """Build human-readable filter summary""" try: if not filters: return "No filters applied" filterable_fields = self.filterable_fields.get(report_type, {}) summary_parts = [] for field_name, filter_value in filters.items(): if not filter_value or field_name not in filterable_fields: continue field_config = filterable_fields[field_name] field_label = field_config['label'] if isinstance(filter_value, dict): if 'start' in filter_value or 'end' in filter_value: # Date/number range start = filter_value.get('start') end = filter_value.get('end') if start and end: summary_parts.append(f"{field_label}: {start} to {end}") elif start: summary_parts.append(f"{field_label}: from {start}") elif end: summary_parts.append(f"{field_label}: up to {end}") elif 'min' in filter_value or 'max' in filter_value: # Number range min_val = filter_value.get('min') max_val = filter_value.get('max') if min_val is not None and max_val is not None: summary_parts.append(f"{field_label}: {min_val} to {max_val}") elif min_val is not None: summary_parts.append(f"{field_label}: from {min_val}") elif max_val is not None: summary_parts.append(f"{field_label}: up to {max_val}") else: # Text search value = filter_value.get('value', '') search_type = filter_value.get('type', 'contains') summary_parts.append(f"{field_label} {search_type}: {value}") elif isinstance(filter_value, list): summary_parts.append(f"{field_label}: {', '.join(map(str, filter_value))}") else: summary_parts.append(f"{field_label}: {filter_value}") return "; ".join(summary_parts) if summary_parts else "No filters applied" except Exception as e: logger.error(f"Error building filter summary: {str(e)}") return "Filter summary unavailable" def export_filter_configuration(self, user, report_type: str) -> Dict[str, Any]: """Export user's filter configuration for backup/sharing""" try: presets = self.get_filter_presets(user, report_type) searches = self.get_saved_searches(user, report_type) return { 'report_type': report_type, 'user': user.username, 'exported_at': datetime.now().isoformat(), 'presets': [ { 'name': preset.name, 'filters': preset.filters, 'is_default': preset.is_default, 'created_at': preset.created_at.isoformat() } for preset in presets ], 'searches': [ { 'name': search.name, 'search_criteria': search.search_criteria, 'result_count': search.result_count, 'created_at': search.created_at.isoformat() } for search in searches ] } except Exception as e: logger.error(f"Error exporting filter configuration: {str(e)}") return {} def import_filter_configuration(self, user, config_data: Dict[str, Any]) -> bool: """Import filter configuration from backup""" try: report_type = config_data.get('report_type') if not report_type: return False # Import presets for preset_data in config_data.get('presets', []): self.save_filter_preset( user=user, name=preset_data['name'], report_type=report_type, filters=preset_data['filters'], is_default=preset_data.get('is_default', False) ) # Import searches for search_data in config_data.get('searches', []): self.save_search( user=user, name=search_data['name'], report_type=report_type, search_criteria=search_data['search_criteria'], result_count=search_data.get('result_count', 0) ) return True except Exception as e: logger.error(f"Error importing filter configuration: {str(e)}") return False def create_smart_filter_suggestions(self, user, report_type: str, current_filters: Dict[str, Any]) -> List[Dict[str, Any]]: """Create smart filter suggestions based on user behavior and data patterns""" try: suggestions = [] # Get user's filter history recent_presets = FilterPreset.objects.filter( user=user, report_type=report_type ).order_by('-updated_at')[:5] # Analyze common filter patterns common_filters = {} for preset in recent_presets: for field, value in preset.filters.items(): if field not in common_filters: common_filters[field] = [] common_filters[field].append(value) # Generate suggestions based on patterns for field, values in common_filters.items(): if field not in current_filters: # Suggest most common value for this field most_common = max(set(map(str, values)), key=values.count) if values else None if most_common: suggestions.append({ 'type': 'common_filter', 'field': field, 'value': most_common, 'reason': f'You often filter by {field} = {most_common}' }) # Add contextual suggestions based on current filters if 'days_overdue' in current_filters: days_overdue = current_filters['days_overdue'] if isinstance(days_overdue, dict) and days_overdue.get('min', 0) > 30: suggestions.append({ 'type': 'contextual', 'field': 'risk_level', 'value': ['High Risk', 'Critical'], 'reason': 'Loans overdue >30 days are typically high risk' }) # Add seasonal suggestions current_month = datetime.now().month if current_month in [12, 1, 2]: # Holiday season suggestions.append({ 'type': 'seasonal', 'field': 'collection_status', 'value': 'Follow Up Required', 'reason': 'Holiday season often requires more follow-up' }) return suggestions[:3] # Limit to top 3 suggestions except Exception as e: logger.error(f"Error creating smart filter suggestions: {str(e)}") return [] def apply_bulk_filters(self, queryset, bulk_filters: List[Dict[str, Any]], report_type: str) -> models.QuerySet: """Apply multiple filter sets and combine results""" try: if not bulk_filters: return queryset combined_queryset = None for filter_set in bulk_filters: filters = filter_set.get('filters', {}) operator = filter_set.get('operator', 'OR') # OR/AND with other sets # Apply filters to a copy of the original queryset filtered_qs = self.apply_advanced_filters(queryset, filters, report_type) if combined_queryset is None: combined_queryset = filtered_qs else: if operator == 'AND': combined_queryset = combined_queryset.intersection(filtered_qs) else: # OR combined_queryset = combined_queryset.union(filtered_qs) return combined_queryset or queryset except Exception as e: logger.error(f"Error applying bulk filters: {str(e)}") return queryset def create_dynamic_filter_groups(self, queryset, report_type: str) -> Dict[str, Any]: """Create dynamic filter groups based on data distribution""" try: groups = {} # Analyze data distribution for numeric fields if report_type == 'loans_in_arrears': # Group by arrears amount ranges arrears_ranges = [ {'label': 'Small (< 10K)', 'min': 0, 'max': 10000}, {'label': 'Medium (10K - 50K)', 'min': 10000, 'max': 50000}, {'label': 'Large (50K - 100K)', 'min': 50000, 'max': 100000}, {'label': 'Very Large (> 100K)', 'min': 100000, 'max': None} ] groups['arrears_amount_ranges'] = [] for range_def in arrears_ranges: count = queryset.filter( arrears_amount__gte=range_def['min'], arrears_amount__lt=range_def['max'] if range_def['max'] else float('inf') ).count() if count > 0: groups['arrears_amount_ranges'].append({ 'label': f"{range_def['label']} ({count})", 'filter': { 'arrears_amount': { 'min': range_def['min'], 'max': range_def['max'] } }, 'count': count }) # Group by days overdue ranges overdue_ranges = [ {'label': '1-30 days', 'min': 1, 'max': 30}, {'label': '31-60 days', 'min': 31, 'max': 60}, {'label': '61-90 days', 'min': 61, 'max': 90}, {'label': '90+ days', 'min': 91, 'max': None} ] groups['days_overdue_ranges'] = [] for range_def in overdue_ranges: filter_kwargs = {'days_overdue__gte': range_def['min']} if range_def['max']: filter_kwargs['days_overdue__lte'] = range_def['max'] count = queryset.filter(**filter_kwargs).count() if count > 0: groups['days_overdue_ranges'].append({ 'label': f"{range_def['label']} ({count})", 'filter': { 'days_overdue': { 'min': range_def['min'], 'max': range_def['max'] } }, 'count': count }) return groups except Exception as e: logger.error(f"Error creating dynamic filter groups: {str(e)}") return {} def get_filter_performance_stats(self, user, report_type: str) -> Dict[str, Any]: """Get performance statistics for user's filters""" try: presets = FilterPreset.objects.filter(user=user, report_type=report_type) searches = SavedSearch.objects.filter(user=user, report_type=report_type) stats = { 'total_presets': presets.count(), 'total_searches': searches.count(), 'most_used_preset': None, 'most_effective_search': None, 'filter_usage_frequency': {}, 'average_result_count': 0 } # Find most used preset (assuming usage tracking) if presets.exists(): # In a real implementation, you'd track usage stats['most_used_preset'] = presets.first().name # Find most effective search (highest result count) if searches.exists(): most_effective = searches.order_by('-result_count').first() stats['most_effective_search'] = { 'name': most_effective.name, 'result_count': most_effective.result_count } # Calculate average result count total_results = sum(search.result_count for search in searches) stats['average_result_count'] = total_results / searches.count() # Analyze filter usage frequency all_filters = {} for preset in presets: for field in preset.filters.keys(): all_filters[field] = all_filters.get(field, 0) + 1 stats['filter_usage_frequency'] = dict( sorted(all_filters.items(), key=lambda x: x[1], reverse=True)[:5] ) return stats except Exception as e: logger.error(f"Error getting filter performance stats: {str(e)}") return {} def optimize_filter_performance(self, queryset, filters: Dict[str, Any], report_type: str) -> Tuple[models.QuerySet, Dict[str, Any]]: """Optimize filter performance and provide recommendations""" try: optimization_info = { 'original_count': queryset.count(), 'optimizations_applied': [], 'performance_tips': [] } # Apply filters with performance monitoring start_time = datetime.now() filtered_queryset = self.apply_advanced_filters(queryset, filters, report_type) end_time = datetime.now() filter_time = (end_time - start_time).total_seconds() optimization_info['filter_time_seconds'] = filter_time optimization_info['filtered_count'] = filtered_queryset.count() # Provide performance recommendations if filter_time > 2.0: # Slow filter optimization_info['performance_tips'].append( "Consider adding database indexes for frequently filtered fields" ) if optimization_info['filtered_count'] > 1000: optimization_info['performance_tips'].append( "Large result set - consider adding more specific filters" ) # Suggest filter order optimization if len(filters) > 3: optimization_info['performance_tips'].append( "Apply most selective filters first for better performance" ) return filtered_queryset, optimization_info except Exception as e: logger.error(f"Error optimizing filter performance: {str(e)}") return queryset, {'error': str(e)} def create_filter_analytics_report(self, user, report_type: str, date_range: Dict[str, Any]) -> Dict[str, Any]: """Create analytics report on filter usage patterns""" try: start_date = date_range.get('start', datetime.now() - timedelta(days=30)) end_date = date_range.get('end', datetime.now()) if isinstance(start_date, str): start_date = datetime.strptime(start_date, '%Y-%m-%d') if isinstance(end_date, str): end_date = datetime.strptime(end_date, '%Y-%m-%d') # Get filter activity in date range presets_created = FilterPreset.objects.filter( user=user, report_type=report_type, created_at__range=[start_date, end_date] ) searches_performed = SavedSearch.objects.filter( user=user, report_type=report_type, last_executed__range=[start_date, end_date] ) analytics = { 'period': { 'start': start_date.strftime('%Y-%m-%d'), 'end': end_date.strftime('%Y-%m-%d') }, 'activity_summary': { 'presets_created': presets_created.count(), 'searches_performed': searches_performed.count(), 'unique_filters_used': len(set( field for preset in presets_created for field in preset.filters.keys() )) }, 'most_used_filters': {}, 'search_effectiveness': {}, 'recommendations': [] } # Analyze most used filters filter_usage = {} for preset in presets_created: for field in preset.filters.keys(): filter_usage[field] = filter_usage.get(field, 0) + 1 analytics['most_used_filters'] = dict( sorted(filter_usage.items(), key=lambda x: x[1], reverse=True)[:5] ) # Analyze search effectiveness if searches_performed.exists(): avg_results = sum(s.result_count for s in searches_performed) / searches_performed.count() analytics['search_effectiveness'] = { 'average_results': avg_results, 'most_successful_search': searches_performed.order_by('-result_count').first().name, 'total_searches': searches_performed.count() } # Generate recommendations if analytics['activity_summary']['presets_created'] == 0: analytics['recommendations'].append( "Consider creating filter presets for frequently used filter combinations" ) if len(analytics['most_used_filters']) > 0: most_used = list(analytics['most_used_filters'].keys())[0] analytics['recommendations'].append( f"Consider setting a default filter for '{most_used}' since you use it frequently" ) return analytics except Exception as e: logger.error(f"Error creating filter analytics report: {str(e)}") return {'error': str(e)}