# -*- coding: utf-8 -*- """ Analytics Filters ================ Django filter classes for the Adtlas Analytics module. Provides advanced filtering, searching, and data querying capabilities for analytics data and reports. Filter Categories: - Date Range Filters: Time-based data filtering - Campaign Filters: Campaign-specific data filtering - Channel Filters: Channel-based data filtering - Performance Filters: Metric-based filtering - Geographic Filters: Location-based filtering - Audience Filters: Demographic filtering - Status Filters: State-based filtering - Custom Filters: Business logic filtering Key Features: - Advanced date range filtering - Multi-field search capabilities - Performance metric filtering - Geographic and demographic filtering - Custom business logic filters - Optimized database queries - Cache-friendly filtering - Export-ready filter sets Filter Classes: - BaseAnalyticsFilter: Base filter functionality - SfrAnalyticsFilter: SFR analytics data filtering - BouyguesAnalyticsFilter: Bouygues analytics filtering - ImpressionFilter: Impression data filtering - VastResponseFilter: VAST response filtering - PerformanceMetricFilter: Performance metrics filtering - AnalyticsReportFilter: Report filtering - DashboardFilter: Dashboard data filtering Usage Examples: - Apply to ViewSets for API filtering - Use in admin interface for data management - Integrate with search functionality - Support export and reporting features Author: Adtlas Development Team Version: 1.0.0 Last Updated: 2024 """ import logging from datetime import datetime, timedelta from typing import Dict, List, Optional, Any, Union from django.db import models from django.db.models import Q, Count, Sum, Avg, Max, Min from django.utils import timezone from django.contrib.auth import get_user_model import django_filters from django_filters import rest_framework as filters from django_filters.constants import EMPTY_VALUES from apps.campaigns.models import Campaign from apps.channels.models import Channel from apps.advertisers.models import Brand from .models import ( SfrAnalytics, BouyguesAnalytics, Impression, VastResponse, PerformanceMetric, AnalyticsReport ) from .utils import ( parse_date_range, validate_date_range, get_cache_key ) # Configure logging logger = logging.getLogger(__name__) # Get User model User = get_user_model() class BaseAnalyticsFilter(filters.FilterSet): """ Base filter class for analytics data. Provides common filtering functionality for all analytics models: - Date range filtering - User-based filtering - Performance optimization - Cache integration Common Filters: - date_from: Start date for filtering - date_to: End date for filtering - date_range: Predefined date ranges - created_after: Records created after date - created_before: Records created before date - user: User-based filtering - search: General search functionality Features: - Optimized database queries - Date validation - User permission integration - Cache-friendly filtering """ # Date range filters date_from = filters.DateTimeFilter( field_name='timestamp', lookup_expr='gte', help_text='Filter records from this date (inclusive)' ) date_to = filters.DateTimeFilter( field_name='timestamp', lookup_expr='lte', help_text='Filter records to this date (inclusive)' ) date_range = filters.ChoiceFilter( method='filter_date_range', choices=[ ('today', 'Today'), ('yesterday', 'Yesterday'), ('last_7_days', 'Last 7 days'), ('last_30_days', 'Last 30 days'), ('last_90_days', 'Last 90 days'), ('this_week', 'This week'), ('last_week', 'Last week'), ('this_month', 'This month'), ('last_month', 'Last month'), ('this_quarter', 'This quarter'), ('last_quarter', 'Last quarter'), ('this_year', 'This year'), ('last_year', 'Last year') ], help_text='Predefined date ranges for quick filtering' ) # Creation date filters created_after = filters.DateTimeFilter( field_name='created_at', lookup_expr='gte', help_text='Filter records created after this date' ) created_before = filters.DateTimeFilter( field_name='created_at', lookup_expr='lte', help_text='Filter records created before this date' ) # Search filter search = filters.CharFilter( method='filter_search', help_text='General search across relevant fields' ) # Ordering ordering = filters.OrderingFilter( fields=( ('timestamp', 'timestamp'), ('created_at', 'created_at'), ('updated_at', 'updated_at'), ), field_labels={ 'timestamp': 'Timestamp', 'created_at': 'Created Date', 'updated_at': 'Updated Date', } ) class Meta: abstract = True def filter_date_range(self, queryset, name, value): """ Filter by predefined date ranges. Args: queryset: Django QuerySet name: Filter field name value: Selected date range Returns: Filtered QuerySet """ try: if value in EMPTY_VALUES: return queryset now = timezone.now() if value == 'today': start_date = now.replace(hour=0, minute=0, second=0, microsecond=0) end_date = now.replace(hour=23, minute=59, second=59, microsecond=999999) elif value == 'yesterday': yesterday = now - timedelta(days=1) start_date = yesterday.replace(hour=0, minute=0, second=0, microsecond=0) end_date = yesterday.replace(hour=23, minute=59, second=59, microsecond=999999) elif value == 'last_7_days': start_date = now - timedelta(days=7) end_date = now elif value == 'last_30_days': start_date = now - timedelta(days=30) end_date = now elif value == 'last_90_days': start_date = now - timedelta(days=90) end_date = now elif value == 'this_week': start_date = now - timedelta(days=now.weekday()) start_date = start_date.replace(hour=0, minute=0, second=0, microsecond=0) end_date = now elif value == 'last_week': end_of_last_week = now - timedelta(days=now.weekday() + 1) start_date = end_of_last_week - timedelta(days=6) start_date = start_date.replace(hour=0, minute=0, second=0, microsecond=0) end_date = end_of_last_week.replace(hour=23, minute=59, second=59, microsecond=999999) elif value == 'this_month': start_date = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0) end_date = now elif value == 'last_month': first_day_this_month = now.replace(day=1) last_day_last_month = first_day_this_month - timedelta(days=1) start_date = last_day_last_month.replace(day=1, hour=0, minute=0, second=0, microsecond=0) end_date = last_day_last_month.replace(hour=23, minute=59, second=59, microsecond=999999) elif value == 'this_quarter': quarter = (now.month - 1) // 3 + 1 start_month = (quarter - 1) * 3 + 1 start_date = now.replace(month=start_month, day=1, hour=0, minute=0, second=0, microsecond=0) end_date = now elif value == 'last_quarter': quarter = (now.month - 1) // 3 + 1 if quarter == 1: # Last quarter of previous year start_date = now.replace(year=now.year-1, month=10, day=1, hour=0, minute=0, second=0, microsecond=0) end_date = now.replace(year=now.year-1, month=12, day=31, hour=23, minute=59, second=59, microsecond=999999) else: start_month = (quarter - 2) * 3 + 1 end_month = start_month + 2 start_date = now.replace(month=start_month, day=1, hour=0, minute=0, second=0, microsecond=0) # Get last day of end_month if end_month == 12: end_date = now.replace(month=12, day=31, hour=23, minute=59, second=59, microsecond=999999) else: next_month = now.replace(month=end_month+1, day=1) end_date = (next_month - timedelta(days=1)).replace(hour=23, minute=59, second=59, microsecond=999999) elif value == 'this_year': start_date = now.replace(month=1, day=1, hour=0, minute=0, second=0, microsecond=0) end_date = now elif value == 'last_year': start_date = now.replace(year=now.year-1, month=1, day=1, hour=0, minute=0, second=0, microsecond=0) end_date = now.replace(year=now.year-1, month=12, day=31, hour=23, minute=59, second=59, microsecond=999999) else: return queryset return queryset.filter( timestamp__gte=start_date, timestamp__lte=end_date ) except Exception as e: logger.error(f"Error filtering by date range: {e}") return queryset def filter_search(self, queryset, name, value): """ General search filter - override in subclasses. Args: queryset: Django QuerySet name: Filter field name value: Search term Returns: Filtered QuerySet """ # Base implementation - override in subclasses return queryset def filter_queryset(self, queryset): """ Apply filters to queryset with optimization. Args: queryset: Django QuerySet Returns: Filtered and optimized QuerySet """ try: # Apply parent filters queryset = super().filter_queryset(queryset) # Apply user-based filtering if needed queryset = self._apply_user_filtering(queryset) # Apply performance optimizations queryset = self._optimize_queryset(queryset) return queryset except Exception as e: logger.error(f"Error filtering queryset: {e}") return queryset def _apply_user_filtering(self, queryset): """ Apply user-based filtering based on permissions. Args: queryset: Django QuerySet Returns: Filtered QuerySet """ try: request = getattr(self, 'request', None) if not request or not hasattr(request, 'user'): return queryset user = request.user # Superusers can see all data if user.is_superuser: return queryset # Apply user-specific filtering based on model if hasattr(queryset.model, 'campaign_id'): # Filter by accessible campaigns accessible_campaigns = self._get_accessible_campaigns(user) if accessible_campaigns is not None: queryset = queryset.filter(campaign_id__in=accessible_campaigns) if hasattr(queryset.model, 'advertiser_id'): # Filter by accessible advertisers accessible_advertisers = self._get_accessible_advertisers(user) if accessible_advertisers is not None: queryset = queryset.filter(advertiser_id__in=accessible_advertisers) return queryset except Exception as e: logger.error(f"Error applying user filtering: {e}") return queryset def _get_accessible_campaigns(self, user): """ Get campaigns accessible to user. Args: user: User object Returns: List of campaign IDs or None for no restrictions """ try: if user.is_superuser: return None # No restrictions # Get campaigns based on user's advertiser if hasattr(user, 'advertiser') and user.advertiser: return list(Campaign.objects.filter( advertiser=user.advertiser ).values_list('id', flat=True)) # Get campaigns user has explicit access to return list(Campaign.objects.filter( Q(created_by=user) | Q(team_members=user) ).values_list('id', flat=True)) except Exception as e: logger.error(f"Error getting accessible campaigns: {e}") return [] def _get_accessible_advertisers(self, user): """ Get advertisers accessible to user. Args: user: User object Returns: List of advertiser IDs or None for no restrictions """ try: if user.is_superuser: return None # No restrictions # Get user's advertiser if hasattr(user, 'advertiser') and user.advertiser: return [user.advertiser.id] # Get advertisers user has explicit access to return list(Advertiser.objects.filter( users=user ).values_list('id', flat=True)) except Exception as e: logger.error(f"Error getting accessible advertisers: {e}") return [] def _optimize_queryset(self, queryset): """ Apply performance optimizations to queryset. Args: queryset: Django QuerySet Returns: Optimized QuerySet """ try: # Add select_related for foreign keys if hasattr(queryset.model, 'campaign'): queryset = queryset.select_related('campaign') if hasattr(queryset.model, 'channel'): queryset = queryset.select_related('channel') if hasattr(queryset.model, 'advertiser'): queryset = queryset.select_related('advertiser') # Add prefetch_related for many-to-many fields # (Override in subclasses as needed) return queryset except Exception as e: logger.error(f"Error optimizing queryset: {e}") return queryset class SfrAnalyticsFilter(BaseAnalyticsFilter): """ Filter class for SFR analytics data. Provides SFR-specific filtering capabilities: - Channel filtering - Program filtering - Audience metrics filtering - Rating range filtering - Share percentage filtering SFR-Specific Filters: - channel: Channel filtering - program_name: Program name search - audience_min: Minimum audience count - audience_max: Maximum audience count - rating_min: Minimum rating - rating_max: Maximum rating - share_min: Minimum share percentage - share_max: Maximum share percentage """ # Channel filtering channel = filters.ModelChoiceFilter( queryset=Channel.objects.all(), help_text='Filter by specific channel' ) channel_name = filters.CharFilter( field_name='channel__name', lookup_expr='icontains', help_text='Filter by channel name (partial match)' ) # Program filtering program_name = filters.CharFilter( field_name='program_name', lookup_expr='icontains', help_text='Filter by program name (partial match)' ) # Audience metrics filtering audience_min = filters.NumberFilter( field_name='audience_count', lookup_expr='gte', help_text='Minimum audience count' ) audience_max = filters.NumberFilter( field_name='audience_count', lookup_expr='lte', help_text='Maximum audience count' ) # Rating filtering rating_min = filters.NumberFilter( field_name='rating', lookup_expr='gte', help_text='Minimum rating' ) rating_max = filters.NumberFilter( field_name='rating', lookup_expr='lte', help_text='Maximum rating' ) # Share filtering share_min = filters.NumberFilter( field_name='share', lookup_expr='gte', help_text='Minimum share percentage' ) share_max = filters.NumberFilter( field_name='share', lookup_expr='lte', help_text='Maximum share percentage' ) # Time slot filtering time_slot = filters.ChoiceFilter( method='filter_time_slot', choices=[ ('morning', 'Morning (6-12)'), ('afternoon', 'Afternoon (12-18)'), ('evening', 'Evening (18-24)'), ('night', 'Night (0-6)'), ('prime_time', 'Prime Time (20-23)'), ('day_time', 'Day Time (6-18)'), ], help_text='Filter by time slot' ) class Meta: model = SfrAnalytics fields = { 'id': ['exact'], 'timestamp': ['exact', 'gte', 'lte'], 'audience_count': ['exact', 'gte', 'lte'], 'rating': ['exact', 'gte', 'lte'], 'share': ['exact', 'gte', 'lte'], 'program_name': ['exact', 'icontains'], 'created_at': ['exact', 'gte', 'lte'], 'updated_at': ['exact', 'gte', 'lte'], } def filter_search(self, queryset, name, value): """ Search across SFR analytics fields. Args: queryset: Django QuerySet name: Filter field name value: Search term Returns: Filtered QuerySet """ try: if value in EMPTY_VALUES: return queryset return queryset.filter( Q(program_name__icontains=value) | Q(channel__name__icontains=value) | Q(channel__description__icontains=value) ) except Exception as e: logger.error(f"Error in SFR analytics search: {e}") return queryset def filter_time_slot(self, queryset, name, value): """ Filter by time slot. Args: queryset: Django QuerySet name: Filter field name value: Time slot value Returns: Filtered QuerySet """ try: if value in EMPTY_VALUES: return queryset hour_ranges = { 'morning': (6, 12), 'afternoon': (12, 18), 'evening': (18, 24), 'night': (0, 6), 'prime_time': (20, 23), 'day_time': (6, 18), } if value not in hour_ranges: return queryset start_hour, end_hour = hour_ranges[value] if start_hour < end_hour: return queryset.filter( timestamp__hour__gte=start_hour, timestamp__hour__lt=end_hour ) else: # Handle overnight ranges (like night: 0-6) return queryset.filter( Q(timestamp__hour__gte=start_hour) | Q(timestamp__hour__lt=end_hour) ) except Exception as e: logger.error(f"Error filtering by time slot: {e}") return queryset class BouyguesAnalyticsFilter(BaseAnalyticsFilter): """ Filter class for Bouygues analytics data. Provides Bouygues-specific filtering capabilities: - Channel filtering - Program filtering - Audience demographics filtering - Geographic filtering - Device type filtering Bouygues-Specific Filters: - channel: Channel filtering - program_name: Program name search - audience_min: Minimum audience count - audience_max: Maximum audience count - device_type: Device type filtering - region: Geographic region filtering """ # Channel filtering channel = filters.ModelChoiceFilter( queryset=Channel.objects.all(), help_text='Filter by specific channel' ) channel_name = filters.CharFilter( field_name='channel__name', lookup_expr='icontains', help_text='Filter by channel name (partial match)' ) # Program filtering program_name = filters.CharFilter( field_name='program_name', lookup_expr='icontains', help_text='Filter by program name (partial match)' ) # Audience metrics filtering audience_min = filters.NumberFilter( field_name='audience_count', lookup_expr='gte', help_text='Minimum audience count' ) audience_max = filters.NumberFilter( field_name='audience_count', lookup_expr='lte', help_text='Maximum audience count' ) # Device type filtering device_type = filters.ChoiceFilter( choices=[ ('tv', 'Television'), ('mobile', 'Mobile'), ('tablet', 'Tablet'), ('desktop', 'Desktop'), ('streaming', 'Streaming Device'), ], help_text='Filter by device type' ) # Geographic filtering region = filters.CharFilter( field_name='region', lookup_expr='icontains', help_text='Filter by geographic region' ) class Meta: model = BouyguesAnalytics fields = { 'id': ['exact'], 'timestamp': ['exact', 'gte', 'lte'], 'audience_count': ['exact', 'gte', 'lte'], 'program_name': ['exact', 'icontains'], 'device_type': ['exact'], 'region': ['exact', 'icontains'], 'created_at': ['exact', 'gte', 'lte'], 'updated_at': ['exact', 'gte', 'lte'], } def filter_search(self, queryset, name, value): """ Search across Bouygues analytics fields. Args: queryset: Django QuerySet name: Filter field name value: Search term Returns: Filtered QuerySet """ try: if value in EMPTY_VALUES: return queryset return queryset.filter( Q(program_name__icontains=value) | Q(channel__name__icontains=value) | Q(region__icontains=value) | Q(device_type__icontains=value) ) except Exception as e: logger.error(f"Error in Bouygues analytics search: {e}") return queryset class ImpressionFilter(BaseAnalyticsFilter): """ Filter class for impression data. Provides impression-specific filtering capabilities: - Campaign filtering - Ad spot filtering - Performance metrics filtering - Conversion tracking filtering - Geographic filtering Impression-Specific Filters: - campaign: Campaign filtering - ad_spot: Ad spot filtering - click_through: Click-through filtering - conversion: Conversion filtering - completion_rate_min: Minimum completion rate - completion_rate_max: Maximum completion rate """ # Campaign filtering campaign = filters.ModelChoiceFilter( queryset=Campaign.objects.all(), help_text='Filter by specific campaign' ) campaign_name = filters.CharFilter( field_name='campaign__name', lookup_expr='icontains', help_text='Filter by campaign name (partial match)' ) # Channel filtering channel = filters.ModelChoiceFilter( queryset=Channel.objects.all(), help_text='Filter by specific channel' ) # Performance filtering click_through = filters.BooleanFilter( help_text='Filter by click-through status' ) conversion = filters.BooleanFilter( help_text='Filter by conversion status' ) completion_rate_min = filters.NumberFilter( field_name='completion_rate', lookup_expr='gte', help_text='Minimum completion rate' ) completion_rate_max = filters.NumberFilter( field_name='completion_rate', lookup_expr='lte', help_text='Maximum completion rate' ) # Geographic filtering country = filters.CharFilter( field_name='country', lookup_expr='icontains', help_text='Filter by country' ) region = filters.CharFilter( field_name='region', lookup_expr='icontains', help_text='Filter by region' ) city = filters.CharFilter( field_name='city', lookup_expr='icontains', help_text='Filter by city' ) class Meta: model = Impression fields = { 'id': ['exact'], 'timestamp': ['exact', 'gte', 'lte'], 'click_through': ['exact'], 'conversion': ['exact'], 'completion_rate': ['exact', 'gte', 'lte'], 'country': ['exact', 'icontains'], 'region': ['exact', 'icontains'], 'city': ['exact', 'icontains'], 'created_at': ['exact', 'gte', 'lte'], 'updated_at': ['exact', 'gte', 'lte'], } def filter_search(self, queryset, name, value): """ Search across impression fields. Args: queryset: Django QuerySet name: Filter field name value: Search term Returns: Filtered QuerySet """ try: if value in EMPTY_VALUES: return queryset return queryset.filter( Q(campaign__name__icontains=value) | Q(channel__name__icontains=value) | Q(country__icontains=value) | Q(region__icontains=value) | Q(city__icontains=value) ) except Exception as e: logger.error(f"Error in impression search: {e}") return queryset class VastResponseFilter(BaseAnalyticsFilter): """ Filter class for VAST response data. Provides VAST-specific filtering capabilities: - Campaign filtering - Status code filtering - Response time filtering - Error filtering - Integration filtering VAST-Specific Filters: - campaign: Campaign filtering - status_code: HTTP status code filtering - response_time_min: Minimum response time - response_time_max: Maximum response time - has_error: Error status filtering - integration_type: Integration type filtering """ # Campaign filtering campaign = filters.ModelChoiceFilter( queryset=Campaign.objects.all(), help_text='Filter by specific campaign' ) # Status filtering status_code = filters.NumberFilter( help_text='Filter by HTTP status code' ) status_code_range = filters.ChoiceFilter( method='filter_status_code_range', choices=[ ('success', 'Success (200-299)'), ('client_error', 'Client Error (400-499)'), ('server_error', 'Server Error (500-599)'), ('error', 'All Errors (400+)'), ], help_text='Filter by status code range' ) # Response time filtering response_time_min = filters.NumberFilter( field_name='response_time', lookup_expr='gte', help_text='Minimum response time (ms)' ) response_time_max = filters.NumberFilter( field_name='response_time', lookup_expr='lte', help_text='Maximum response time (ms)' ) # Error filtering has_error = filters.BooleanFilter( method='filter_has_error', help_text='Filter by error presence' ) error_type = filters.CharFilter( field_name='error_message', lookup_expr='icontains', help_text='Filter by error type/message' ) class Meta: model = VastResponse fields = { 'id': ['exact'], 'timestamp': ['exact', 'gte', 'lte'], 'status_code': ['exact', 'gte', 'lte'], 'response_time': ['exact', 'gte', 'lte'], 'created_at': ['exact', 'gte', 'lte'], 'updated_at': ['exact', 'gte', 'lte'], } def filter_status_code_range(self, queryset, name, value): """ Filter by status code ranges. Args: queryset: Django QuerySet name: Filter field name value: Status code range Returns: Filtered QuerySet """ try: if value in EMPTY_VALUES: return queryset if value == 'success': return queryset.filter(status_code__gte=200, status_code__lt=300) elif value == 'client_error': return queryset.filter(status_code__gte=400, status_code__lt=500) elif value == 'server_error': return queryset.filter(status_code__gte=500, status_code__lt=600) elif value == 'error': return queryset.filter(status_code__gte=400) return queryset except Exception as e: logger.error(f"Error filtering by status code range: {e}") return queryset def filter_has_error(self, queryset, name, value): """ Filter by error presence. Args: queryset: Django QuerySet name: Filter field name value: Boolean indicating error presence Returns: Filtered QuerySet """ try: if value in EMPTY_VALUES: return queryset if value: return queryset.filter( Q(status_code__gte=400) | Q(error_message__isnull=False) & ~Q(error_message='') ) else: return queryset.filter( status_code__lt=400, Q(error_message__isnull=True) | Q(error_message='') ) except Exception as e: logger.error(f"Error filtering by error presence: {e}") return queryset def filter_search(self, queryset, name, value): """ Search across VAST response fields. Args: queryset: Django QuerySet name: Filter field name value: Search term Returns: Filtered QuerySet """ try: if value in EMPTY_VALUES: return queryset return queryset.filter( Q(campaign__name__icontains=value) | Q(error_message__icontains=value) | Q(vast_url__icontains=value) ) except Exception as e: logger.error(f"Error in VAST response search: {e}") return queryset class PerformanceMetricFilter(BaseAnalyticsFilter): """ Filter class for performance metrics. Provides performance metric filtering capabilities: - Metric name filtering - Value range filtering - Threshold filtering - Category filtering - Trend analysis filtering Performance-Specific Filters: - metric_name: Metric name filtering - metric_category: Metric category filtering - value_min: Minimum metric value - value_max: Maximum metric value - threshold_exceeded: Threshold status filtering """ # Metric filtering metric_name = filters.CharFilter( lookup_expr='icontains', help_text='Filter by metric name (partial match)' ) metric_category = filters.ChoiceFilter( choices=[ ('performance', 'Performance'), ('availability', 'Availability'), ('quality', 'Quality'), ('user_experience', 'User Experience'), ('business', 'Business'), ('technical', 'Technical'), ], help_text='Filter by metric category' ) # Value filtering value_min = filters.NumberFilter( field_name='value', lookup_expr='gte', help_text='Minimum metric value' ) value_max = filters.NumberFilter( field_name='value', lookup_expr='lte', help_text='Maximum metric value' ) # Threshold filtering threshold_exceeded = filters.BooleanFilter( method='filter_threshold_exceeded', help_text='Filter by threshold exceeded status' ) class Meta: model = PerformanceMetric fields = { 'id': ['exact'], 'timestamp': ['exact', 'gte', 'lte'], 'metric_name': ['exact', 'icontains'], 'value': ['exact', 'gte', 'lte'], 'created_at': ['exact', 'gte', 'lte'], 'updated_at': ['exact', 'gte', 'lte'], } def filter_threshold_exceeded(self, queryset, name, value): """ Filter by threshold exceeded status. Args: queryset: Django QuerySet name: Filter field name value: Boolean indicating threshold status Returns: Filtered QuerySet """ try: if value in EMPTY_VALUES: return queryset # This would need threshold configuration # For now, return queryset as-is return queryset except Exception as e: logger.error(f"Error filtering by threshold: {e}") return queryset def filter_search(self, queryset, name, value): """ Search across performance metric fields. Args: queryset: Django QuerySet name: Filter field name value: Search term Returns: Filtered QuerySet """ try: if value in EMPTY_VALUES: return queryset return queryset.filter( Q(metric_name__icontains=value) | Q(description__icontains=value) ) except Exception as e: logger.error(f"Error in performance metric search: {e}") return queryset class AnalyticsReportFilter(BaseAnalyticsFilter): """ Filter class for analytics reports. Provides report-specific filtering capabilities: - Report type filtering - Status filtering - User filtering - Date range filtering - Export format filtering Report-Specific Filters: - report_type: Report type filtering - status: Report status filtering - generated_by: User filtering - export_format: Export format filtering - has_data: Data availability filtering """ # Report type filtering report_type = filters.ChoiceFilter( choices=[ ('campaign_performance', 'Campaign Performance'), ('channel_analytics', 'Channel Analytics'), ('audience_insights', 'Audience Insights'), ('revenue_analysis', 'Revenue Analysis'), ('custom', 'Custom Report'), ], help_text='Filter by report type' ) # Status filtering status = filters.ChoiceFilter( choices=[ ('pending', 'Pending'), ('processing', 'Processing'), ('completed', 'Completed'), ('failed', 'Failed'), ('cancelled', 'Cancelled'), ], help_text='Filter by report status' ) # User filtering generated_by = filters.ModelChoiceFilter( queryset=User.objects.all(), help_text='Filter by report generator' ) generated_by_me = filters.BooleanFilter( method='filter_generated_by_me', help_text='Filter reports generated by current user' ) # Export format filtering export_format = filters.ChoiceFilter( choices=[ ('pdf', 'PDF'), ('excel', 'Excel'), ('csv', 'CSV'), ('json', 'JSON'), ], help_text='Filter by export format' ) # Data availability filtering has_data = filters.BooleanFilter( method='filter_has_data', help_text='Filter by data availability' ) class Meta: model = AnalyticsReport fields = { 'id': ['exact'], 'report_type': ['exact'], 'status': ['exact'], 'export_format': ['exact'], 'created_at': ['exact', 'gte', 'lte'], 'updated_at': ['exact', 'gte', 'lte'], } def filter_generated_by_me(self, queryset, name, value): """ Filter reports generated by current user. Args: queryset: Django QuerySet name: Filter field name value: Boolean indicating current user filter Returns: Filtered QuerySet """ try: if value in EMPTY_VALUES: return queryset request = getattr(self, 'request', None) if not request or not hasattr(request, 'user'): return queryset if value: return queryset.filter(generated_by=request.user) else: return queryset.exclude(generated_by=request.user) except Exception as e: logger.error(f"Error filtering by current user: {e}") return queryset def filter_has_data(self, queryset, name, value): """ Filter by data availability. Args: queryset: Django QuerySet name: Filter field name value: Boolean indicating data availability Returns: Filtered QuerySet """ try: if value in EMPTY_VALUES: return queryset if value: return queryset.filter( status='completed', file_path__isnull=False ).exclude(file_path='') else: return queryset.filter( Q(status__in=['pending', 'processing', 'failed']) | Q(file_path__isnull=True) | Q(file_path='') ) except Exception as e: logger.error(f"Error filtering by data availability: {e}") return queryset def filter_search(self, queryset, name, value): """ Search across report fields. Args: queryset: Django QuerySet name: Filter field name value: Search term Returns: Filtered QuerySet """ try: if value in EMPTY_VALUES: return queryset return queryset.filter( Q(title__icontains=value) | Q(description__icontains=value) | Q(generated_by__username__icontains=value) | Q(generated_by__first_name__icontains=value) | Q(generated_by__last_name__icontains=value) ) except Exception as e: logger.error(f"Error in report search: {e}") return queryset class DashboardFilter(BaseAnalyticsFilter): """ Filter class for dashboard data aggregation. Provides dashboard-specific filtering for: - Multi-model data aggregation - Real-time data filtering - Summary statistics filtering - Widget-specific filtering Dashboard-Specific Filters: - widget_type: Dashboard widget filtering - aggregation_level: Data aggregation level - real_time: Real-time data filtering - summary_type: Summary statistics type """ # Widget filtering widget_type = filters.ChoiceFilter( choices=[ ('campaign_overview', 'Campaign Overview'), ('channel_performance', 'Channel Performance'), ('audience_metrics', 'Audience Metrics'), ('revenue_summary', 'Revenue Summary'), ('real_time_stats', 'Real-time Statistics'), ], help_text='Filter by dashboard widget type' ) # Aggregation level aggregation_level = filters.ChoiceFilter( choices=[ ('hourly', 'Hourly'), ('daily', 'Daily'), ('weekly', 'Weekly'), ('monthly', 'Monthly'), ('quarterly', 'Quarterly'), ('yearly', 'Yearly'), ], help_text='Data aggregation level' ) # Real-time filtering real_time = filters.BooleanFilter( help_text='Filter for real-time data only' ) # Summary type summary_type = filters.ChoiceFilter( choices=[ ('totals', 'Totals'), ('averages', 'Averages'), ('trends', 'Trends'), ('comparisons', 'Comparisons'), ], help_text='Summary statistics type' ) class Meta: # Dashboard filter doesn't have a specific model # It's used for aggregating data from multiple models fields = {} # Filter utility functions def get_filter_class_for_model(model_class): """ Get appropriate filter class for a model. Args: model_class: Django model class Returns: Filter class or None """ filter_mapping = { SfrAnalytics: SfrAnalyticsFilter, BouyguesAnalytics: BouyguesAnalyticsFilter, Impression: ImpressionFilter, VastResponse: VastResponseFilter, PerformanceMetric: PerformanceMetricFilter, AnalyticsReport: AnalyticsReportFilter, } return filter_mapping.get(model_class) def apply_common_filters(queryset, filters_dict): """ Apply common filters to any queryset. Args: queryset: Django QuerySet filters_dict: Dictionary of filter parameters Returns: Filtered QuerySet """ try: # Apply date range filters if 'date_from' in filters_dict: queryset = queryset.filter(timestamp__gte=filters_dict['date_from']) if 'date_to' in filters_dict: queryset = queryset.filter(timestamp__lte=filters_dict['date_to']) # Apply search filter if 'search' in filters_dict and filters_dict['search']: search_term = filters_dict['search'] # This would need to be customized per model if hasattr(queryset.model, 'name'): queryset = queryset.filter(name__icontains=search_term) return queryset except Exception as e: logger.error(f"Error applying common filters: {e}") return queryset # Export filter classes __all__ = [ 'BaseAnalyticsFilter', 'SfrAnalyticsFilter', 'BouyguesAnalyticsFilter', 'ImpressionFilter', 'VastResponseFilter', 'PerformanceMetricFilter', 'AnalyticsReportFilter', 'DashboardFilter', 'get_filter_class_for_model', 'apply_common_filters' ]