# -*- coding: utf-8 -*- """ Adtlas Activities Serializers This module contains the Django REST Framework serializers for the activities app, providing comprehensive data serialization and validation for activity tracking, category management, and analytics functionality. Features: - Activity serialization with nested relationships - Category management serialization - Activity summary serialization - Custom validation and field handling - Optimized queries for performance Author: Adtlas Development Team Version: 1.0.0 Last Updated: 2025-01-27 """ from django.utils import timezone from rest_framework import serializers from django.contrib.auth import get_user_model from django.utils.translation import gettext_lazy as _ from django.contrib.contenttypes.models import ContentType from apps.activities.models import Activity, ActivityCategory, ActivitySummary User = get_user_model() class UserBasicSerializer(serializers.ModelSerializer): """ Basic user serializer for activity relationships. This serializer provides essential user information for activity records without exposing sensitive data. Fields: - id: User ID - email: User email address - first_name: User first name - last_name: User last name - full_name: Computed full name """ full_name = serializers.SerializerMethodField() class Meta: model = User fields = ['id', 'email', 'first_name', 'last_name', 'full_name'] read_only_fields = ['id', 'email', 'first_name', 'last_name'] def get_full_name(self, obj): """ Get user's full name. Args: obj: User instance Returns: str: User's full name or email if name not available """ if obj.first_name and obj.last_name: return f"{obj.first_name} {obj.last_name}" elif obj.first_name: return obj.first_name elif obj.last_name: return obj.last_name else: return obj.email class ActivityCategorySerializer(serializers.ModelSerializer): """ Serializer for activity categories. This serializer handles activity category data including validation, creation, and updates with proper field handling. Fields: - All ActivityCategory model fields - activity_count: Number of activities in this category """ activity_count = serializers.SerializerMethodField() class Meta: model = ActivityCategory fields = [ 'id', 'name', 'code', 'description', 'color', 'icon', 'is_active', 'created_at', 'updated_at', 'activity_count' ] read_only_fields = ['id', 'created_at', 'updated_at', 'activity_count'] def get_activity_count(self, obj): """ Get the number of activities in this category. Args: obj: ActivityCategory instance Returns: int: Number of activities in this category """ return getattr(obj, 'activity_count', obj.activities.count()) def validate_code(self, value): """ Validate category code uniqueness. Args: value: Category code value Returns: str: Validated code value Raises: ValidationError: If code already exists """ if value: # Check for existing code (excluding current instance during updates) queryset = ActivityCategory.objects.filter(code=value) if self.instance: queryset = queryset.exclude(pk=self.instance.pk) if queryset.exists(): raise serializers.ValidationError( _("A category with this code already exists.") ) return value def validate_color(self, value): """ Validate color format. Args: value: Color value Returns: str: Validated color value Raises: ValidationError: If color format is invalid """ if value and not value.startswith('#'): raise serializers.ValidationError( _("Color must be in hex format (e.g., #FF5733).") ) if value and len(value) not in [4, 7]: # #RGB or #RRGGBB raise serializers.ValidationError( _("Color must be in valid hex format (#RGB or #RRGGBB).") ) return value class ContentTypeSerializer(serializers.ModelSerializer): """ Serializer for content types in activity relationships. This serializer provides content type information for activities that reference other model instances. Fields: - id: Content type ID - app_label: Application label - model: Model name - name: Human-readable name """ class Meta: model = ContentType fields = ['id', 'app_label', 'model', 'name'] read_only_fields = ['id', 'app_label', 'model', 'name'] class ActivitySerializer(serializers.ModelSerializer): """ Comprehensive serializer for activities. This serializer handles activity data with nested relationships, computed fields, and optimized queries for performance. Fields: - All Activity model fields - user: Nested user information - category: Nested category information - content_type: Nested content type information - object_display: Human-readable object representation - duration_display: Human-readable duration - time_ago: Relative time display """ user = UserBasicSerializer(read_only=True) category = ActivityCategorySerializer(read_only=True) content_type = ContentTypeSerializer(read_only=True) object_display = serializers.SerializerMethodField() duration_display = serializers.SerializerMethodField() time_ago = serializers.SerializerMethodField() action_display = serializers.CharField(source='get_action_display', read_only=True) class Meta: model = Activity fields = [ 'id', 'user', 'action', 'action_display', 'category', 'description', 'content_type', 'object_id', 'object_display', 'ip_address', 'user_agent', 'session_key', 'is_successful', 'error_message', 'duration_ms', 'duration_display', 'metadata', 'created_at', 'time_ago' ] read_only_fields = [ 'id', 'user', 'category', 'content_type', 'object_display', 'duration_display', 'time_ago', 'action_display', 'created_at' ] def get_object_display(self, obj): """ Get human-readable representation of the related object. Args: obj: Activity instance Returns: str: Human-readable object representation """ return obj.get_object_display() def get_duration_display(self, obj): """ Get human-readable duration display. Args: obj: Activity instance Returns: str: Human-readable duration or None """ if obj.duration_ms is None: return None if obj.duration_ms < 1000: return f"{obj.duration_ms}ms" elif obj.duration_ms < 60000: return f"{obj.duration_ms / 1000:.1f}s" else: minutes = obj.duration_ms // 60000 seconds = (obj.duration_ms % 60000) / 1000 return f"{minutes}m {seconds:.1f}s" def get_time_ago(self, obj): """ Get relative time display for activity creation. Args: obj: Activity instance Returns: str: Relative time display """ from django.utils.timesince import timesince return timesince(obj.created_at, timezone.now()) def to_representation(self, instance): """ Customize the serialized representation. Args: instance: Activity instance Returns: dict: Serialized data """ data = super().to_representation(instance) # Add success status display data['success_display'] = 'Success' if instance.is_successful else 'Failed' # Format metadata for better readability if instance.metadata: try: # Ensure metadata is properly formatted if isinstance(instance.metadata, str): import json data['metadata'] = json.loads(instance.metadata) except (json.JSONDecodeError, TypeError): # Keep original metadata if parsing fails pass return data class ActivityCreateSerializer(serializers.ModelSerializer): """ Serializer for creating new activities. This serializer handles activity creation with validation and automatic field population from request context. Fields: - Required: action, description - Optional: category, content_type, object_id, metadata - Auto-populated: user, ip_address, user_agent, session_key """ category_id = serializers.IntegerField(required=False, allow_null=True) content_type_id = serializers.IntegerField(required=False, allow_null=True) class Meta: model = Activity fields = [ 'action', 'description', 'category_id', 'content_type_id', 'object_id', 'is_successful', 'error_message', 'duration_ms', 'metadata', 'ip_address', 'user_agent', 'session_key' ] def validate_category_id(self, value): """ Validate category exists and is active. Args: value: Category ID Returns: int: Validated category ID Raises: ValidationError: If category doesn't exist or is inactive """ if value is not None: try: category = ActivityCategory.objects.get(id=value, is_active=True) return value except ActivityCategory.DoesNotExist: raise serializers.ValidationError( _("Invalid category ID or category is not active.") ) return value def validate_content_type_id(self, value): """ Validate content type exists. Args: value: Content type ID Returns: int: Validated content type ID Raises: ValidationError: If content type doesn't exist """ if value is not None: try: ContentType.objects.get(id=value) return value except ContentType.DoesNotExist: raise serializers.ValidationError( _("Invalid content type ID.") ) return value def validate(self, attrs): """ Perform cross-field validation. Args: attrs: Validated attributes Returns: dict: Validated attributes Raises: ValidationError: If validation fails """ # If content_type is provided, object_id should also be provided if attrs.get('content_type_id') and not attrs.get('object_id'): raise serializers.ValidationError( _("object_id is required when content_type is specified.") ) # If object_id is provided, content_type should also be provided if attrs.get('object_id') and not attrs.get('content_type_id'): raise serializers.ValidationError( _("content_type is required when object_id is specified.") ) return attrs def create(self, validated_data): """ Create activity with proper field mapping. Args: validated_data: Validated data Returns: Activity: Created activity instance """ # Handle category and content_type foreign keys category_id = validated_data.pop('category_id', None) content_type_id = validated_data.pop('content_type_id', None) if category_id: validated_data['category'] = ActivityCategory.objects.get(id=category_id) if content_type_id: validated_data['content_type'] = ContentType.objects.get(id=content_type_id) return Activity.objects.create(**validated_data) class ActivitySummarySerializer(serializers.ModelSerializer): """ Serializer for activity summaries. This serializer handles activity summary data with nested relationships and computed fields for analytics. Fields: - All ActivitySummary model fields - user: Nested user information - category: Nested category information - success_rate: Computed success rate - avg_duration_display: Human-readable average duration """ user = UserBasicSerializer(read_only=True) category = ActivityCategorySerializer(read_only=True) success_rate = serializers.SerializerMethodField() avg_duration_display = serializers.SerializerMethodField() class Meta: model = ActivitySummary fields = [ 'id', 'date', 'user', 'category', 'total_activities', 'successful_activities', 'failed_activities', 'success_rate', 'avg_duration_ms', 'avg_duration_display', 'unique_ips', 'created_at', 'updated_at' ] read_only_fields = [ 'id', 'user', 'category', 'success_rate', 'avg_duration_display', 'created_at', 'updated_at' ] def get_success_rate(self, obj): """ Calculate success rate percentage. Args: obj: ActivitySummary instance Returns: float: Success rate as percentage """ if obj.total_activities == 0: return 0.0 return (obj.successful_activities / obj.total_activities) * 100 def get_avg_duration_display(self, obj): """ Get human-readable average duration display. Args: obj: ActivitySummary instance Returns: str: Human-readable duration or None """ if obj.avg_duration_ms is None: return None if obj.avg_duration_ms < 1000: return f"{obj.avg_duration_ms:.0f}ms" elif obj.avg_duration_ms < 60000: return f"{obj.avg_duration_ms / 1000:.1f}s" else: minutes = obj.avg_duration_ms // 60000 seconds = (obj.avg_duration_ms % 60000) / 1000 return f"{minutes:.0f}m {seconds:.1f}s" class ActivityBulkCreateSerializer(serializers.Serializer): """ Serializer for bulk activity creation. This serializer handles bulk creation of activities for batch processing and import operations. Fields: - activities: List of activity data """ activities = ActivityCreateSerializer(many=True) def create(self, validated_data): """ Create multiple activities in bulk. Args: validated_data: Validated data with activities list Returns: list: List of created activity instances """ activities_data = validated_data['activities'] activities = [] for activity_data in activities_data: # Handle foreign key relationships category_id = activity_data.pop('category_id', None) content_type_id = activity_data.pop('content_type_id', None) if category_id: activity_data['category'] = ActivityCategory.objects.get(id=category_id) if content_type_id: activity_data['content_type'] = ContentType.objects.get(id=content_type_id) activities.append(Activity(**activity_data)) # Bulk create for performance return Activity.objects.bulk_create(activities) def validate_activities(self, value): """ Validate activities list. Args: value: List of activity data Returns: list: Validated activities data Raises: ValidationError: If validation fails """ if not value: raise serializers.ValidationError( _("At least one activity is required.") ) if len(value) > 100: # Limit bulk operations raise serializers.ValidationError( _("Cannot create more than 100 activities at once.") ) return value class ActivityExportSerializer(serializers.ModelSerializer): """ Serializer for activity export operations. This serializer provides a flattened representation of activities suitable for CSV/Excel export. Fields: - Flattened activity data with related object information """ user_email = serializers.CharField(source='user.email', read_only=True) user_name = serializers.SerializerMethodField() category_name = serializers.CharField(source='category.name', read_only=True) content_type_name = serializers.CharField(source='content_type.name', read_only=True) object_display = serializers.SerializerMethodField() action_display = serializers.CharField(source='get_action_display', read_only=True) success_display = serializers.SerializerMethodField() duration_display = serializers.SerializerMethodField() class Meta: model = Activity fields = [ 'id', 'user_email', 'user_name', 'action', 'action_display', 'category_name', 'description', 'content_type_name', 'object_id', 'object_display', 'ip_address', 'is_successful', 'success_display', 'duration_ms', 'duration_display', 'error_message', 'created_at' ] def get_user_name(self, obj): """ Get user's full name for export. Args: obj: Activity instance Returns: str: User's full name or email """ if not obj.user: return 'Anonymous' if obj.user.first_name and obj.user.last_name: return f"{obj.user.first_name} {obj.user.last_name}" elif obj.user.first_name: return obj.user.first_name elif obj.user.last_name: return obj.user.last_name else: return obj.user.email def get_object_display(self, obj): """ Get object display for export. Args: obj: Activity instance Returns: str: Object display string """ return obj.get_object_display() def get_success_display(self, obj): """ Get success status display for export. Args: obj: Activity instance Returns: str: Success status display """ return 'Success' if obj.is_successful else 'Failed' def get_duration_display(self, obj): """ Get duration display for export. Args: obj: Activity instance Returns: str: Duration display """ if obj.duration_ms is None: return '' if obj.duration_ms < 1000: return f"{obj.duration_ms}ms" elif obj.duration_ms < 60000: return f"{obj.duration_ms / 1000:.1f}s" else: minutes = obj.duration_ms // 60000 seconds = (obj.duration_ms % 60000) / 1000 return f"{minutes}m {seconds:.1f}s"