# -*- coding: utf-8 -*- """ Adtlas Activities Views This module contains the views for the activities app, providing comprehensive activity tracking, monitoring, and reporting functionality through both web interfaces and REST API endpoints. Features: - Activity listing and filtering - Real-time activity feeds - Activity analytics and reporting - User activity dashboards - Activity search and export - Category management Author: Adtlas Development Team Version: 1.0.0 Last Updated: 2025-01-27 """ import json from datetime import datetime, timedelta from django.shortcuts import render, get_object_or_404 from django.contrib.auth.decorators import login_required from django.contrib.auth.mixins import LoginRequiredMixin from django.http import JsonResponse, HttpResponse from django.views.generic import ListView, DetailView, TemplateView from django.views.decorators.http import require_http_methods from django.views.decorators.csrf import csrf_exempt from django.utils.decorators import method_decorator from django.utils import timezone from django.db.models import Count, Q, Avg, Max, Min from django.core.paginator import Paginator from django.utils.translation import gettext_lazy as _ from django.contrib.contenttypes.models import ContentType from rest_framework import generics, status, filters from rest_framework.decorators import api_view, permission_classes from rest_framework.response import Response from rest_framework.permissions import IsAuthenticated from rest_framework.pagination import PageNumberPagination from django_filters.rest_framework import DjangoFilterBackend from apps.activities.models import Activity, ActivityCategory, ActivitySummary from apps.activities.api.v1.serializers import ( ActivitySerializer, ActivityCategorySerializer, ActivitySummarySerializer, ActivityCreateSerializer ) from apps.activities.filters import ActivityFilter from apps.activities.permissions import CanViewActivities, CanManageActivities class ActivityPagination(PageNumberPagination): """ Custom pagination class for activity listings. This pagination class provides configurable page sizes and includes additional metadata in the response. """ page_size = 25 page_size_query_param = 'page_size' max_page_size = 100 def get_paginated_response(self, data): """ Return a paginated response with additional metadata. Args: data: Serialized data for the current page Returns: Response: Paginated response with metadata """ return Response({ 'links': { 'next': self.get_next_link(), 'previous': self.get_previous_link() }, 'count': self.page.paginator.count, 'page_size': self.page_size, 'total_pages': self.page.paginator.num_pages, 'current_page': self.page.number, 'results': data }) class ActivityListView(LoginRequiredMixin, ListView): """ Web view for listing activities with filtering and pagination. This view provides a web interface for browsing activities with advanced filtering, search, and export capabilities. Attributes: model: Activity model template_name: Template for rendering the view context_object_name: Name for the activities in template context paginate_by: Number of activities per page """ model = Activity template_name = 'activities/activity_list.html' context_object_name = 'activities' paginate_by = 25 def get_queryset(self): """ Get filtered and optimized queryset for activities. Returns: QuerySet: Filtered activities queryset """ queryset = Activity.objects.select_related( 'user', 'category', 'content_type' ).order_by('-created_at') # Apply filters based on GET parameters user_id = self.request.GET.get('user') if user_id: queryset = queryset.filter(user_id=user_id) category_id = self.request.GET.get('category') if category_id: queryset = queryset.filter(category_id=category_id) action = self.request.GET.get('action') if action: queryset = queryset.filter(action=action) is_successful = self.request.GET.get('is_successful') if is_successful is not None: queryset = queryset.filter(is_successful=is_successful.lower() == 'true') # Date range filtering date_from = self.request.GET.get('date_from') date_to = self.request.GET.get('date_to') if date_from: try: date_from = datetime.strptime(date_from, '%Y-%m-%d').date() queryset = queryset.filter(created_at__date__gte=date_from) except ValueError: pass if date_to: try: date_to = datetime.strptime(date_to, '%Y-%m-%d').date() queryset = queryset.filter(created_at__date__lte=date_to) except ValueError: pass # Search functionality search = self.request.GET.get('search') if search: queryset = queryset.filter( Q(description__icontains=search) | Q(user__email__icontains=search) | Q(user__first_name__icontains=search) | Q(user__last_name__icontains=search) ) return queryset def get_context_data(self, **kwargs): """ Add additional context data for the template. Args: **kwargs: Additional keyword arguments Returns: dict: Template context data """ context = super().get_context_data(**kwargs) # Add filter options context['categories'] = ActivityCategory.objects.filter(is_active=True) context['actions'] = Activity.ACTION_CHOICES # Add current filter values context['current_filters'] = { 'user': self.request.GET.get('user', ''), 'category': self.request.GET.get('category', ''), 'action': self.request.GET.get('action', ''), 'is_successful': self.request.GET.get('is_successful', ''), 'date_from': self.request.GET.get('date_from', ''), 'date_to': self.request.GET.get('date_to', ''), 'search': self.request.GET.get('search', ''), } # Add statistics queryset = self.get_queryset() context['total_count'] = queryset.count() context['success_count'] = queryset.filter(is_successful=True).count() context['failure_count'] = queryset.filter(is_successful=False).count() return context class ActivityDetailView(LoginRequiredMixin, DetailView): """ Web view for displaying detailed activity information. This view shows comprehensive details about a specific activity including metadata, related objects, and context information. Attributes: model: Activity model template_name: Template for rendering the view context_object_name: Name for the activity in template context """ model = Activity template_name = 'activities/activity_detail.html' context_object_name = 'activity' def get_queryset(self): """ Get optimized queryset for activity detail. Returns: QuerySet: Optimized activities queryset """ return Activity.objects.select_related( 'user', 'category', 'content_type' ) class ActivityDashboardView(LoginRequiredMixin, TemplateView): """ Web view for activity analytics dashboard. This view provides a comprehensive dashboard with activity statistics, charts, and analytics for monitoring system usage. Attributes: template_name: Template for rendering the dashboard """ template_name = 'activities/dashboard.html' def get_context_data(self, **kwargs): """ Add dashboard data to the template context. Args: **kwargs: Additional keyword arguments Returns: dict: Template context data with dashboard statistics """ context = super().get_context_data(**kwargs) # Date ranges for statistics now = timezone.now() today = now.date() yesterday = today - timedelta(days=1) week_ago = today - timedelta(days=7) month_ago = today - timedelta(days=30) # Basic statistics context['stats'] = { 'total_activities': Activity.objects.count(), 'today_activities': Activity.objects.filter(created_at__date=today).count(), 'yesterday_activities': Activity.objects.filter(created_at__date=yesterday).count(), 'week_activities': Activity.objects.filter(created_at__date__gte=week_ago).count(), 'month_activities': Activity.objects.filter(created_at__date__gte=month_ago).count(), 'success_rate': self._calculate_success_rate(), 'active_users_today': Activity.objects.filter( created_at__date=today ).values('user').distinct().count(), } # Top categories context['top_categories'] = Activity.objects.filter( created_at__date__gte=week_ago ).values( 'category__name', 'category__color' ).annotate( count=Count('id') ).order_by('-count')[:10] # Top actions context['top_actions'] = Activity.objects.filter( created_at__date__gte=week_ago ).values('action').annotate( count=Count('id') ).order_by('-count')[:10] # Recent activities context['recent_activities'] = Activity.objects.select_related( 'user', 'category' ).order_by('-created_at')[:20] # Activity trends (last 30 days) context['activity_trends'] = self._get_activity_trends(30) return context def _calculate_success_rate(self): """ Calculate overall success rate for activities. Returns: float: Success rate as a percentage """ total = Activity.objects.count() if total == 0: return 0.0 successful = Activity.objects.filter(is_successful=True).count() return (successful / total) * 100 def _get_activity_trends(self, days): """ Get activity trends for the specified number of days. Args: days (int): Number of days to analyze Returns: list: List of daily activity counts """ end_date = timezone.now().date() start_date = end_date - timedelta(days=days) # Get daily activity counts daily_counts = Activity.objects.filter( created_at__date__gte=start_date ).extra( select={'day': 'date(created_at)'} ).values('day').annotate( count=Count('id') ).order_by('day') # Convert to list with all dates included counts_dict = {item['day']: item['count'] for item in daily_counts} trends = [] current_date = start_date while current_date <= end_date: trends.append({ 'date': current_date.strftime('%Y-%m-%d'), 'count': counts_dict.get(current_date, 0) }) current_date += timedelta(days=1) return trends # REST API Views class ActivityListAPIView(generics.ListCreateAPIView): """ REST API view for listing and creating activities. This view provides REST API endpoints for retrieving activities with filtering, pagination, and search capabilities, as well as creating new activity records. Permissions: - List: IsAuthenticated + CanViewActivities - Create: IsAuthenticated + CanManageActivities """ queryset = Activity.objects.select_related('user', 'category', 'content_type') serializer_class = ActivitySerializer permission_classes = [IsAuthenticated, CanViewActivities] pagination_class = ActivityPagination filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] filterset_class = ActivityFilter search_fields = ['description', 'user__email', 'user__first_name', 'user__last_name'] ordering_fields = ['created_at', 'action', 'is_successful', 'duration_ms'] ordering = ['-created_at'] def get_serializer_class(self): """ Return appropriate serializer class based on action. Returns: Serializer class for the current action """ if self.request.method == 'POST': return ActivityCreateSerializer return ActivitySerializer def get_permissions(self): """ Return appropriate permissions based on action. Returns: list: Permission instances for the current action """ if self.request.method == 'POST': return [IsAuthenticated(), CanManageActivities()] return [IsAuthenticated(), CanViewActivities()] def perform_create(self, serializer): """ Perform activity creation with additional context. Args: serializer: Activity serializer instance """ # Add request context if not provided if not serializer.validated_data.get('ip_address'): serializer.validated_data['ip_address'] = self._get_client_ip() if not serializer.validated_data.get('user_agent'): serializer.validated_data['user_agent'] = self.request.META.get('HTTP_USER_AGENT', '') if not serializer.validated_data.get('session_key'): serializer.validated_data['session_key'] = self.request.session.session_key serializer.save() def _get_client_ip(self): """ Get client IP address from request. Returns: str: Client IP address """ x_forwarded_for = self.request.META.get('HTTP_X_FORWARDED_FOR') if x_forwarded_for: ip = x_forwarded_for.split(',')[0] else: ip = self.request.META.get('REMOTE_ADDR') return ip class ActivityDetailAPIView(generics.RetrieveUpdateDestroyAPIView): """ REST API view for retrieving, updating, and deleting activities. This view provides REST API endpoints for detailed activity operations including retrieval, updates, and deletion. Permissions: - Retrieve: IsAuthenticated + CanViewActivities - Update/Delete: IsAuthenticated + CanManageActivities """ queryset = Activity.objects.select_related('user', 'category', 'content_type') serializer_class = ActivitySerializer permission_classes = [IsAuthenticated, CanViewActivities] def get_permissions(self): """ Return appropriate permissions based on action. Returns: list: Permission instances for the current action """ if self.request.method in ['PUT', 'PATCH', 'DELETE']: return [IsAuthenticated(), CanManageActivities()] return [IsAuthenticated(), CanViewActivities()] class ActivityCategoryListAPIView(generics.ListCreateAPIView): """ REST API view for listing and creating activity categories. This view provides REST API endpoints for managing activity categories with filtering and search capabilities. Permissions: - List: IsAuthenticated - Create: IsAuthenticated + CanManageActivities """ queryset = ActivityCategory.objects.filter(is_active=True) serializer_class = ActivityCategorySerializer permission_classes = [IsAuthenticated] filter_backends = [filters.SearchFilter, filters.OrderingFilter] search_fields = ['name', 'code', 'description'] ordering_fields = ['name', 'created_at'] ordering = ['name'] def get_permissions(self): """ Return appropriate permissions based on action. Returns: list: Permission instances for the current action """ if self.request.method == 'POST': return [IsAuthenticated(), CanManageActivities()] return [IsAuthenticated()] class ActivitySummaryListAPIView(generics.ListAPIView): """ REST API view for listing activity summaries. This view provides REST API endpoints for retrieving activity summaries with filtering and aggregation capabilities. Permissions: - List: IsAuthenticated + CanViewActivities """ queryset = ActivitySummary.objects.select_related('user', 'category') serializer_class = ActivitySummarySerializer permission_classes = [IsAuthenticated, CanViewActivities] filter_backends = [DjangoFilterBackend, filters.OrderingFilter] filterset_fields = ['date', 'user', 'category'] ordering_fields = ['date', 'total_activities'] ordering = ['-date'] @api_view(['GET']) @permission_classes([IsAuthenticated, CanViewActivities]) def activity_statistics_api(request): """ REST API endpoint for activity statistics. This endpoint provides comprehensive activity statistics including counts, trends, and performance metrics. Args: request: Django request object Returns: Response: JSON response with activity statistics """ # Date range parameters days = int(request.GET.get('days', 30)) end_date = timezone.now().date() start_date = end_date - timedelta(days=days) # Base queryset for the date range base_queryset = Activity.objects.filter(created_at__date__gte=start_date) # Basic statistics total_activities = base_queryset.count() successful_activities = base_queryset.filter(is_successful=True).count() failed_activities = base_queryset.filter(is_successful=False).count() # User statistics unique_users = base_queryset.values('user').distinct().count() unique_ips = base_queryset.values('ip_address').distinct().count() # Performance statistics duration_stats = base_queryset.filter( duration_ms__isnull=False ).aggregate( avg_duration=Avg('duration_ms'), min_duration=Min('duration_ms'), max_duration=Max('duration_ms') ) # Category breakdown category_stats = base_queryset.values( 'category__name', 'category__color' ).annotate( count=Count('id') ).order_by('-count') # Action breakdown action_stats = base_queryset.values('action').annotate( count=Count('id') ).order_by('-count') # Daily trends daily_trends = [] current_date = start_date while current_date <= end_date: day_count = base_queryset.filter(created_at__date=current_date).count() daily_trends.append({ 'date': current_date.strftime('%Y-%m-%d'), 'count': day_count }) current_date += timedelta(days=1) # Hourly distribution (for last 7 days) hourly_stats = base_queryset.filter( created_at__date__gte=end_date - timedelta(days=7) ).extra( select={'hour': 'extract(hour from created_at)'} ).values('hour').annotate( count=Count('id') ).order_by('hour') return Response({ 'period': { 'start_date': start_date.strftime('%Y-%m-%d'), 'end_date': end_date.strftime('%Y-%m-%d'), 'days': days }, 'totals': { 'total_activities': total_activities, 'successful_activities': successful_activities, 'failed_activities': failed_activities, 'success_rate': (successful_activities / total_activities * 100) if total_activities > 0 else 0, 'unique_users': unique_users, 'unique_ips': unique_ips }, 'performance': { 'avg_duration_ms': duration_stats['avg_duration'], 'min_duration_ms': duration_stats['min_duration'], 'max_duration_ms': duration_stats['max_duration'] }, 'breakdowns': { 'by_category': list(category_stats), 'by_action': list(action_stats), 'by_hour': list(hourly_stats) }, 'trends': { 'daily': daily_trends } }) @api_view(['POST']) @permission_classes([IsAuthenticated]) def log_activity_api(request): """ REST API endpoint for logging activities. This is a convenience endpoint for quickly logging activities with minimal required data. Args: request: Django request object with activity data Returns: Response: JSON response with created activity data """ try: # Extract activity data from request data = request.data.copy() # Set user if not provided if 'user' not in data: data['user'] = request.user.id # Set default success status if 'is_successful' not in data: data['is_successful'] = True # Create activity using the log_activity method activity = Activity.log_activity( user=request.user, action=data.get('action'), description=data.get('description'), category=data.get('category'), request=request, is_successful=data.get('is_successful', True), error_message=data.get('error_message', ''), metadata=data.get('metadata', {}) ) # Serialize and return the created activity serializer = ActivitySerializer(activity) return Response(serializer.data, status=status.HTTP_201_CREATED) except Exception as e: return Response( {'error': str(e)}, status=status.HTTP_400_BAD_REQUEST ) @login_required @require_http_methods(["GET"]) def export_activities_csv(request): """ Export activities as CSV file. This view exports filtered activities as a CSV file for external analysis and reporting. Args: request: Django request object with filter parameters Returns: HttpResponse: CSV file response """ import csv # Get filtered queryset (reuse logic from ListView) view = ActivityListView() view.request = request queryset = view.get_queryset() # Create CSV response response = HttpResponse(content_type='text/csv') response['Content-Disposition'] = 'attachment; filename="activities.csv"' writer = csv.writer(response) # Write header writer.writerow([ 'ID', 'User', 'Action', 'Category', 'Description', 'Related Object', 'IP Address', 'Success', 'Duration (ms)', 'Created At', 'Error Message' ]) # Write data for activity in queryset.select_related('user', 'category', 'content_type'): writer.writerow([ str(activity.id), str(activity.user) if activity.user else 'Anonymous', activity.get_action_display(), activity.category.name if activity.category else '', activity.description, activity.get_object_display(), activity.ip_address or '', 'Yes' if activity.is_successful else 'No', activity.duration_ms or '', activity.created_at.strftime('%Y-%m-%d %H:%M:%S'), activity.error_message ]) return response @api_view(['GET']) @permission_classes([IsAuthenticated, CanViewActivities]) def user_activity_feed_api(request, user_id=None): """ REST API endpoint for user activity feed. This endpoint provides a real-time activity feed for a specific user or the current user if no user_id is provided. Args: request: Django request object user_id: Optional user ID for the feed Returns: Response: JSON response with user activity feed """ # Determine target user if user_id: target_user_id = user_id else: target_user_id = request.user.id # Get recent activities for the user limit = int(request.GET.get('limit', 20)) activities = Activity.objects.filter( user_id=target_user_id ).select_related( 'category', 'content_type' ).order_by('-created_at')[:limit] # Serialize activities serializer = ActivitySerializer(activities, many=True) return Response({ 'user_id': target_user_id, 'count': len(activities), 'activities': serializer.data })