# -*- coding: utf-8 -*- """ Adtlas Activities Models This module contains the activity tracking and audit logging models for the Adtlas DAI Management System. It provides comprehensive activity monitoring, user action tracking, and audit trail functionality. Features: - User activity tracking with detailed context - System event logging and monitoring - Activity categorization and filtering - Audit trail management with retention policies - Activity analytics and reporting - Real-time activity feeds Author: Adtlas Development Team Version: 1.0.0 Last Updated: 2025-01-27 """ from django.db import models from django.urls import reverse from django.utils import timezone from django.core.validators import RegexValidator from django.utils.translation import gettext_lazy as _ from django.contrib.contenttypes.models import ContentType from django.contrib.contenttypes.fields import GenericForeignKey from apps.common.models import UUIDModel, TimestampedModel, SoftDeleteModel from apps.activities.managers import ActivityManager, ActivityCategoryManager class ActivityCategory(UUIDModel, TimestampedModel, SoftDeleteModel): """ Activity Category model for organizing and classifying different types of activities. This model allows for better organization of activities by grouping them into logical categories such as authentication, content management, system administration, etc. Attributes: name (str): Human-readable name of the category code (str): Unique code for programmatic access description (str): Detailed description of the category color (str): Hex color code for UI representation icon (str): CSS icon class for UI representation is_system (bool): Whether this is a system-defined category is_active (bool): Whether this category is currently active retention_days (int): Number of days to retain activities in this category parent (ForeignKey): Parent category for hierarchical organization """ # Category identification name = models.CharField( max_length=100, unique=True, verbose_name=_("Category Name"), help_text=_("Human-readable name of the activity category") ) code = models.CharField( max_length=50, unique=True, validators=[RegexValidator(r"^[a-z_]+$", "Code must be lowercase with underscores only")], verbose_name=_("Category Code"), help_text=_("Unique code for programmatic access (e.g., 'auth', 'content')") ) # Category details description = models.TextField( blank=True, verbose_name=_("Description"), help_text=_("Detailed description of what activities belong to this category") ) # UI representation color = models.CharField( max_length=7, default="#007bff", validators=[RegexValidator(r"^#[0-9A-Fa-f]{6}$", "Color must be a valid hex code")], verbose_name=_("Color"), help_text=_("Hex color code for UI representation (e.g., '#007bff')") ) icon = models.CharField( max_length=50, default="fas fa-activity", verbose_name=_("Icon"), help_text=_("CSS icon class for UI representation (e.g., 'fas fa-user')") ) # Category properties is_system = models.BooleanField( default=False, verbose_name=_("Is System Category"), help_text=_("Whether this is a system-defined category that cannot be deleted") ) is_active = models.BooleanField( default=True, verbose_name=_("Is Active"), help_text=_("Whether this category is currently active") ) # Data retention retention_days = models.PositiveIntegerField( default=365, verbose_name=_("Retention Days"), help_text=_("Number of days to retain activities in this category (0 = forever)") ) # Hierarchical structure parent = models.ForeignKey( "self", on_delete=models.CASCADE, null=True, blank=True, related_name="children", verbose_name=_("Parent Category"), help_text=_("Parent category for hierarchical organization") ) class Meta: db_table = "activities_categories" ordering = ["name"] verbose_name = _("Activity Category") verbose_name_plural = _("Activity Categories") indexes = [ models.Index(fields=["code"]), models.Index(fields=["is_active"]), models.Index(fields=["is_system"]), ] # Custom managers objects = models.Manager() active = ActivityCategoryManager() def __str__(self): """String representation of the activity category.""" return self.name def get_absolute_url(self): """Get the absolute URL for this category.""" return reverse("activities:category_detail", kwargs={"pk": self.pk}) def get_all_children(self): """ Get all child categories recursively. Returns: list: List of all child ActivityCategory instances """ children = list(self.children.filter(is_active=True)) for child in self.children.filter(is_active=True): children.extend(child.get_all_children()) return children def get_activity_count(self, days=None): """ Get the count of activities in this category. Args: days (int, optional): Number of days to look back. If None, count all activities. Returns: int: Number of activities in this category """ queryset = self.activities.all() if days is not None: cutoff_date = timezone.now() - timezone.timedelta(days=days) queryset = queryset.filter(created_at__gte=cutoff_date) return queryset.count() class Activity(UUIDModel, TimestampedModel, SoftDeleteModel): """ Activity model for tracking user actions and system events. This model provides comprehensive activity tracking with support for categorization, detailed context, and audit trail functionality. Attributes: user (ForeignKey): User who performed the activity category (ForeignKey): Category this activity belongs to action (str): Type of action performed description (str): Human-readable description of the activity content_type (ForeignKey): Content type of the related object object_id (int): ID of the related object content_object (GenericForeignKey): Generic relation to any model ip_address (GenericIPAddressField): IP address of the user user_agent (str): User agent string from the request session_key (str): Session key for tracking user sessions request_path (str): URL path of the request request_method (str): HTTP method of the request is_successful (bool): Whether the activity completed successfully error_message (str): Error message if the activity failed metadata (JSONField): Additional metadata about the activity duration_ms (int): Duration of the activity in milliseconds """ # Action type choices ACTION_CHOICES = [ # Authentication actions ('LOGIN', _('Login')), ('LOGOUT', _('Logout')), ('PASSWORD_CHANGE', _('Password Change')), ('PASSWORD_RESET', _('Password Reset')), ('ACCOUNT_LOCKED', _('Account Locked')), ('ACCOUNT_UNLOCKED', _('Account Unlocked')), # CRUD operations ('CREATE', _('Create')), ('READ', _('Read')), ('UPDATE', _('Update')), ('DELETE', _('Delete')), ('RESTORE', _('Restore')), # File operations ('UPLOAD', _('Upload')), ('DOWNLOAD', _('Download')), ('EXPORT', _('Export')), ('IMPORT', _('Import')), # Campaign operations ('CAMPAIGN_START', _('Campaign Start')), ('CAMPAIGN_STOP', _('Campaign Stop')), ('CAMPAIGN_PAUSE', _('Campaign Pause')), ('CAMPAIGN_RESUME', _('Campaign Resume')), # System operations ('SYSTEM_BACKUP', _('System Backup')), ('SYSTEM_RESTORE', _('System Restore')), ('SYSTEM_MAINTENANCE', _('System Maintenance')), ('CONFIGURATION_CHANGE', _('Configuration Change')), # Security events ('PERMISSION_GRANTED', _('Permission Granted')), ('PERMISSION_REVOKED', _('Permission Revoked')), ('ROLE_ASSIGNED', _('Role Assigned')), ('ROLE_REMOVED', _('Role Removed')), ('SECURITY_VIOLATION', _('Security Violation')), # Analytics and reporting ('REPORT_GENERATED', _('Report Generated')), ('ANALYTICS_VIEWED', _('Analytics Viewed')), ('DATA_EXPORTED', _('Data Exported')), # Other ('OTHER', _('Other')), ] # HTTP method choices HTTP_METHOD_CHOICES = [ ('GET', 'GET'), ('POST', 'POST'), ('PUT', 'PUT'), ('PATCH', 'PATCH'), ('DELETE', 'DELETE'), ('HEAD', 'HEAD'), ('OPTIONS', 'OPTIONS'), ] # Core activity fields user = models.ForeignKey( "accounts.User", on_delete=models.SET_NULL, null=True, blank=True, related_name="activities", verbose_name=_("User"), help_text=_("User who performed this activity") ) category = models.ForeignKey( ActivityCategory, on_delete=models.SET_NULL, null=True, blank=True, related_name="activities", verbose_name=_("Category"), help_text=_("Category this activity belongs to") ) action = models.CharField( max_length=50, choices=ACTION_CHOICES, verbose_name=_("Action"), help_text=_("Type of action that was performed") ) description = models.TextField( verbose_name=_("Description"), help_text=_("Human-readable description of what happened") ) # Generic foreign key for relating to any model content_type = models.ForeignKey( ContentType, on_delete=models.SET_NULL, null=True, blank=True, verbose_name=_("Content Type"), help_text=_("Type of object this activity relates to") ) object_id = models.PositiveIntegerField( null=True, blank=True, verbose_name=_("Object ID"), help_text=_("ID of the object this activity relates to") ) content_object = GenericForeignKey('content_type', 'object_id') # Request context ip_address = models.GenericIPAddressField( null=True, blank=True, verbose_name=_("IP Address"), help_text=_("IP address from which the activity was performed") ) user_agent = models.TextField( blank=True, verbose_name=_("User Agent"), help_text=_("Browser/client user agent string") ) session_key = models.CharField( max_length=40, blank=True, verbose_name=_("Session Key"), help_text=_("Session key for tracking user sessions") ) request_path = models.CharField( max_length=500, blank=True, verbose_name=_("Request Path"), help_text=_("URL path where the activity occurred") ) request_method = models.CharField( max_length=10, choices=HTTP_METHOD_CHOICES, blank=True, verbose_name=_("Request Method"), help_text=_("HTTP method used for the request") ) # Activity outcome is_successful = models.BooleanField( default=True, verbose_name=_("Is Successful"), help_text=_("Whether the activity completed successfully") ) error_message = models.TextField( blank=True, verbose_name=_("Error Message"), help_text=_("Error message if the activity failed") ) # Additional data metadata = models.JSONField( default=dict, blank=True, verbose_name=_("Metadata"), help_text=_("Additional structured data about the activity") ) # Performance tracking duration_ms = models.BigIntegerField( null=True, blank=True, verbose_name=_("Duration (ms)"), help_text=_("Duration of the activity in milliseconds") ) # Custom manager objects = ActivityManager() class Meta: db_table = "activities_activities" ordering = ["-created_at"] verbose_name = _("Activity") verbose_name_plural = _("Activities") indexes = [ models.Index(fields=["user", "-created_at"]), models.Index(fields=["category", "-created_at"]), models.Index(fields=["action", "-created_at"]), models.Index(fields=["content_type", "object_id"]), models.Index(fields=["ip_address"]), models.Index(fields=["session_key"]), models.Index(fields=["is_successful"]), models.Index(fields=["-created_at"]), ] @property def duration_seconds(self): """Calculate duration in seconds.""" if self.duration_ms: return self.duration_ms / 1000.0 return None @property def is_recent(self): """Check if activity is recent (within last 24 hours).""" from django.utils import timezone cutoff = timezone.now() - timezone.timedelta(hours=24) return self.created_at >= cutoff def get_content_object(self): """Get the related content object.""" if self.content_type and self.object_id: try: return self.content_type.get_object_for_this_type(pk=self.object_id) except: return None return None def __str__(self): """String representation of the activity.""" user_str = str(self.user) if self.user else "Anonymous" formatted_date = self.created_at.strftime("%Y-%m-%d %H:%M") return f"{user_str} - {self.action} - {formatted_date}" def get_absolute_url(self): """Get the absolute URL for this activity.""" return reverse("activities:activity_detail", kwargs={"pk": self.pk}) def get_object_display(self): """ Get a human-readable representation of the related object. Returns: str: String representation of the related object """ if self.content_object: return str(self.content_object) elif self.content_type: return f"{self.content_type.name} (ID: {self.object_id})" return "No related object" def get_duration_display(self): """ Get a human-readable representation of the activity duration. Returns: str: Formatted duration string """ if self.duration_ms is None: return "Unknown" if self.duration_ms < 1000: return f"{self.duration_ms}ms" elif self.duration_ms < 60000: return f"{self.duration_ms / 1000:.1f}s" else: minutes = self.duration_ms // 60000 seconds = (self.duration_ms % 60000) / 1000 return f"{minutes}m {seconds:.1f}s" def add_metadata(self, key, value): """ Add a key-value pair to the activity metadata. Args: key (str): Metadata key value: Metadata value (must be JSON serializable) """ if not isinstance(self.metadata, dict): self.metadata = {} self.metadata[key] = value self.save(update_fields=['metadata']) def get_metadata(self, key, default=None): """ Get a value from the activity metadata. Args: key (str): Metadata key default: Default value if key doesn't exist Returns: Metadata value or default """ if not isinstance(self.metadata, dict): return default return self.metadata.get(key, default) @classmethod def log_activity(cls, user, action, description="", category=None, content_object=None, ip_address=None, user_agent=None, session_key=None, is_successful=True, duration_ms=None): """Log an activity.""" from django.contrib.contenttypes.models import ContentType activity_data = { "user": user, "action": action, "description": description, "category": category, "ip_address": ip_address, "user_agent": user_agent, "session_key": session_key, "is_successful": is_successful, "duration_ms": duration_ms, } if content_object: activity_data["content_type"] = ContentType.objects.get_for_model(content_object) activity_data["object_id"] = content_object.pk return cls.objects.create(**activity_data) @classmethod def log_activity(cls, user=None, action=None, description=None, category=None, content_object=None, request=None, is_successful=True, error_message=None, metadata=None, **kwargs): """ Convenience method to log an activity. Args: user: User who performed the activity action (str): Action type description (str): Activity description category: ActivityCategory instance or code content_object: Related model instance request: Django request object is_successful (bool): Whether the activity was successful error_message (str): Error message if failed metadata (dict): Additional metadata **kwargs: Additional fields to set on the activity Returns: Activity: Created activity instance """ # Handle category parameter if isinstance(category, str): try: category = ActivityCategory.objects.get(code=category) except ActivityCategory.DoesNotExist: category = None # Extract request information request_data = {} if request: request_data.update({ 'ip_address': cls._get_client_ip(request), 'user_agent': request.META.get('HTTP_USER_AGENT', ''), 'session_key': request.session.session_key, 'request_path': request.path, 'request_method': request.method, }) # Create the activity activity_data = { 'user': user, 'action': action, 'description': description, 'category': category, 'is_successful': is_successful, 'error_message': error_message or '', 'metadata': metadata or {}, **request_data, **kwargs } # Set content object if provided if content_object: activity_data['content_object'] = content_object return cls.objects.create(**activity_data) @staticmethod def _get_client_ip(request): """ Get the client IP address from the request. Args: request: Django request object Returns: str: Client IP address """ x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR') if x_forwarded_for: ip = x_forwarded_for.split(',')[0] else: ip = request.META.get('REMOTE_ADDR') return ip class ActivitySummary(UUIDModel, TimestampedModel): """ Activity Summary model for storing aggregated activity statistics. This model stores pre-calculated activity statistics for improved performance when displaying activity analytics and reports. Attributes: date (DateField): Date for which the summary is calculated user (ForeignKey): User for user-specific summaries (optional) category (ForeignKey): Category for category-specific summaries (optional) total_activities (int): Total number of activities successful_activities (int): Number of successful activities failed_activities (int): Number of failed activities unique_users (int): Number of unique users (for global summaries) unique_ips (int): Number of unique IP addresses avg_duration_ms (float): Average activity duration in milliseconds peak_hour (int): Hour with the most activity (0-23) summary_data (JSONField): Additional summary statistics """ # Summary scope date = models.DateField( verbose_name=_("Date"), help_text=_("Date for which this summary is calculated") ) user = models.ForeignKey( "accounts.User", on_delete=models.CASCADE, null=True, blank=True, related_name="activity_summaries", verbose_name=_("User"), help_text=_("User for user-specific summaries (null for global summaries)") ) category = models.ForeignKey( ActivityCategory, on_delete=models.CASCADE, null=True, blank=True, related_name="activity_summaries", verbose_name=_("Category"), help_text=_("Category for category-specific summaries (null for all categories)") ) # Activity counts total_activities = models.PositiveIntegerField( default=0, verbose_name=_("Total Activities"), help_text=_("Total number of activities for this date/scope") ) successful_activities = models.PositiveIntegerField( default=0, verbose_name=_("Successful Activities"), help_text=_("Number of successful activities") ) failed_activities = models.PositiveIntegerField( default=0, verbose_name=_("Failed Activities"), help_text=_("Number of failed activities") ) # Unique counts unique_users = models.PositiveIntegerField( default=0, verbose_name=_("Unique Users"), help_text=_("Number of unique users (for global summaries)") ) unique_ips = models.PositiveIntegerField( default=0, verbose_name=_("Unique IPs"), help_text=_("Number of unique IP addresses") ) # Performance metrics avg_duration_ms = models.FloatField( null=True, blank=True, verbose_name=_("Average Duration (ms)"), help_text=_("Average activity duration in milliseconds") ) peak_hour = models.PositiveSmallIntegerField( null=True, blank=True, verbose_name=_("Peak Hour"), help_text=_("Hour with the most activity (0-23)") ) # Additional data summary_data = models.JSONField( default=dict, blank=True, verbose_name=_("Summary Data"), help_text=_("Additional summary statistics and breakdowns") ) class Meta: db_table = "activities_summaries" ordering = ["-date"] verbose_name = _("Activity Summary") verbose_name_plural = _("Activity Summaries") unique_together = ["date", "user", "category"] indexes = [ models.Index(fields=["date"]), models.Index(fields=["user", "date"]), models.Index(fields=["category", "date"]), ] def __str__(self): """String representation of the activity summary.""" scope_parts = [str(self.date)] if self.user: scope_parts.append(f"User: {self.user}") if self.category: scope_parts.append(f"Category: {self.category}") return f"Activity Summary - {' - '.join(scope_parts)}" @property def success_rate(self): """ Calculate the success rate as a percentage. Returns: float: Success rate as a percentage (0-100) """ if self.total_activities == 0: return 0.0 return (self.successful_activities / self.total_activities) * 100 @property def failure_rate(self): """ Calculate the failure rate as a percentage. Returns: float: Failure rate as a percentage (0-100) """ return 100.0 - self.success_rate