""" REST API Views for Streams Application This module contains Django REST Framework views for managing streaming channels, sessions, configurations, and related operations through a RESTful API interface. """ from rest_framework import viewsets, status, permissions from rest_framework.decorators import action from rest_framework.response import Response from rest_framework.pagination import PageNumberPagination from rest_framework.throttling import UserRateThrottle from django_filters.rest_framework import DjangoFilterBackend from django.shortcuts import get_object_or_404 from django.utils import timezone from django.db.models import Count, Sum, Avg, Q from datetime import timedelta from drf_spectacular.utils import extend_schema, extend_schema_view, OpenApiParameter, OpenApiExample from drf_spectacular.types import OpenApiTypes from core.schema import ( PAGINATION_PARAMETERS, ORDERING_PARAMETER, SEARCH_PARAMETER, CHANNEL_STATUS_PARAMETER, SESSION_STATUS_PARAMETER, CHANNEL_FILTER_PARAMETER, CHANNEL_EXAMPLES, SUCCESS_EXAMPLES, ERROR_EXAMPLES ) from apps.streams.models import Channel, StreamSession, HLSSegment, VideoConfiguration, AudioConfiguration from apps.streams.tasks import start_stream_capture, stop_stream_capture, create_playlist_for_channel from apps.core.throttling import StreamControlThrottle, DetectionAPIThrottle from apps.streams.api.serializers import ( ChannelSerializer, StreamSessionSerializer, HLSSegmentSerializer, VideoConfigurationSerializer, AudioConfigurationSerializer, ChannelStatsSerializer, StreamStartSerializer ) class StandardResultsSetPagination(PageNumberPagination): """ Standard pagination configuration for API responses. Provides consistent pagination across all API endpoints with configurable page sizes and metadata. """ page_size = 20 page_size_query_param = 'page_size' max_page_size = 100 @extend_schema_view( list=extend_schema( summary='List channels', description='Retrieve a paginated list of streaming channels', parameters=PAGINATION_PARAMETERS + [ORDERING_PARAMETER, SEARCH_PARAMETER, CHANNEL_STATUS_PARAMETER], examples=CHANNEL_EXAMPLES, tags=['Channels'] ), create=extend_schema( summary='Create channel', description='Create a new streaming channel', examples=CHANNEL_EXAMPLES, tags=['Channels'] ), retrieve=extend_schema( summary='Get channel', description='Retrieve a specific channel by ID', tags=['Channels'] ), update=extend_schema( summary='Update channel', description='Update an existing channel', tags=['Channels'] ), partial_update=extend_schema( summary='Partially update channel', description='Partially update an existing channel', tags=['Channels'] ), destroy=extend_schema( summary='Delete channel', description='Delete a channel and all associated data', tags=['Channels'] ) ) class ChannelViewSet(viewsets.ModelViewSet): """ ViewSet for managing streaming channels via REST API. Provides CRUD operations for channels along with custom actions for starting/stopping streams and retrieving statistics. """ queryset = Channel.objects.all().select_related('created_by') serializer_class = ChannelSerializer permission_classes = [permissions.IsAuthenticated] pagination_class = StandardResultsSetPagination filter_backends = [DjangoFilterBackend] filterset_fields = ['is_active', 'created_by'] search_fields = ['name', 'slug', 'description'] ordering_fields = ['name', 'created_at', 'updated_at'] ordering = ['-created_at'] def perform_create(self, serializer): """ Set the current user as the channel creator. Args: serializer: Channel serializer instance """ serializer.save(created_by=self.request.user) @extend_schema( summary='Start stream capture', description='Start stream capture for a specific channel with optional video/audio configurations', request=StreamStartSerializer, responses={ 202: OpenApiExample( 'Stream Started', value={ 'success': True, 'message': 'Stream started for News Channel', 'task_id': 'abc123-def456-ghi789', 'channel_id': 'f47ac10b-58cc-4372-a567-0e02b2c3d479' } ), 400: ERROR_EXAMPLES['validation_error'], 500: ERROR_EXAMPLES['not_found'] }, tags=['Channels'] ) @action(detail=True, methods=['post'], throttle_classes=[StreamControlThrottle]) def start_stream(self, request, pk=None): """ Start stream capture for a specific channel. Args: request: HTTP request with optional configuration IDs pk: Channel primary key Returns: Response: Success/error response with task information """ channel = self.get_object() # Check if channel already has an active session if channel.get_active_session(): return Response( {'error': 'Channel already has an active stream session'}, status=status.HTTP_400_BAD_REQUEST ) # Validate request data serializer = StreamStartSerializer(data=request.data) if not serializer.is_valid(): return Response( serializer.errors, status=status.HTTP_400_BAD_REQUEST ) try: # Start the stream capture task task = start_stream_capture.delay( channel_id=str(channel.id), video_config_id=str(serializer.validated_data.get('video_config_id')) if serializer.validated_data.get('video_config_id') else None, audio_config_id=str(serializer.validated_data.get('audio_config_id')) if serializer.validated_data.get('audio_config_id') else None ) return Response({ 'success': True, 'message': f'Stream started for {channel.name}', 'task_id': task.id, 'channel_id': str(channel.id) }, status=status.HTTP_202_ACCEPTED) except Exception as e: return Response( {'error': str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) @extend_schema( summary='Stop stream capture', description='Stop active stream capture for a specific channel', responses={ 202: SUCCESS_EXAMPLES['task_started'], 400: ERROR_EXAMPLES['validation_error'], 500: ERROR_EXAMPLES['not_found'] }, tags=['Channels'] ) @action(detail=True, methods=['post'], throttle_classes=[StreamControlThrottle]) def stop_stream(self, request, pk=None): """ Stop active stream for a specific channel. Args: request: HTTP request pk: Channel primary key Returns: Response: Success/error response with task information """ channel = self.get_object() active_session = channel.get_active_session() if not active_session: return Response( {'error': 'No active stream session found for this channel'}, status=status.HTTP_400_BAD_REQUEST ) try: # Stop the stream capture task task = stop_stream_capture.delay(str(active_session.id)) return Response({ 'success': True, 'message': f'Stream stopped for {channel.name}', 'task_id': task.id, 'session_id': str(active_session.id) }, status=status.HTTP_202_ACCEPTED) except Exception as e: return Response( {'error': str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) @extend_schema( summary='Get channel statistics', description='Retrieve detailed statistics for a specific channel including session counts, durations, and success rates', responses={ 200: ChannelStatsSerializer, 404: ERROR_EXAMPLES['not_found'] }, tags=['Channels'] ) @action(detail=True, methods=['get']) def statistics(self, request, pk=None): """ Get detailed statistics for a specific channel. Args: request: HTTP request pk: Channel primary key Returns: Response: Channel statistics data """ channel = self.get_object() # Calculate statistics sessions = channel.sessions.all() segments = HLSSegment.objects.filter(session__channel=channel) # Success rate calculation total_sessions = sessions.count() successful_sessions = sessions.filter( status__in=['completed', 'active'] ).count() success_rate = (successful_sessions / total_sessions * 100) if total_sessions > 0 else 0 # Duration calculations completed_sessions = sessions.filter(status='completed') total_duration = 0 session_durations = [] for session in completed_sessions: if session.duration(): duration_seconds = int(session.duration().total_seconds()) total_duration += duration_seconds session_durations.append(duration_seconds) avg_session_duration = sum(session_durations) / len(session_durations) if session_durations else 0 stats_data = { 'total_sessions': total_sessions, 'active_sessions': sessions.filter(status__in=['active', 'processing']).count(), 'total_segments': segments.count(), 'total_duration': total_duration, 'success_rate': round(success_rate, 2), 'avg_session_duration': int(avg_session_duration), 'last_session_at': sessions.order_by('-started_at').first().started_at if sessions.exists() else None } serializer = ChannelStatsSerializer(stats_data) return Response(serializer.data) @extend_schema( summary='Get active channels', description='Retrieve all channels that currently have active stream sessions', responses={200: ChannelSerializer(many=True)}, tags=['Channels'] ) @action(detail=False, methods=['get']) def active(self, request): """ Get all channels with active streams. Args: request: HTTP request Returns: Response: List of channels with active sessions """ active_channels = self.queryset.filter( sessions__status__in=['active', 'processing'] ).distinct() serializer = self.get_serializer(active_channels, many=True) return Response(serializer.data) @extend_schema_view( list=extend_schema( summary='List stream sessions', description='Retrieve a paginated list of stream sessions', parameters=PAGINATION_PARAMETERS + [ORDERING_PARAMETER, SESSION_STATUS_PARAMETER, CHANNEL_FILTER_PARAMETER], tags=['Sessions'] ), create=extend_schema( summary='Create stream session', description='Create a new stream session', tags=['Sessions'] ), retrieve=extend_schema( summary='Get stream session', description='Retrieve a specific stream session by ID', tags=['Sessions'] ), update=extend_schema( summary='Update stream session', description='Update an existing stream session', tags=['Sessions'] ), partial_update=extend_schema( summary='Partially update stream session', description='Partially update an existing stream session', tags=['Sessions'] ), destroy=extend_schema( summary='Delete stream session', description='Delete a stream session and all associated data', tags=['Sessions'] ) ) class StreamSessionViewSet(viewsets.ModelViewSet): """ ViewSet for managing stream sessions via REST API. Provides CRUD operations for sessions along with custom actions for stopping sessions and retrieving segments. """ queryset = StreamSession.objects.all().select_related( 'channel', 'video_config', 'audio_config' ) serializer_class = StreamSessionSerializer permission_classes = [permissions.IsAuthenticated] pagination_class = StandardResultsSetPagination filter_backends = [DjangoFilterBackend] filterset_fields = ['status', 'channel'] ordering_fields = ['started_at', 'ended_at', 'status'] ordering = ['-started_at'] @extend_schema( summary='Stop stream session', description='Stop a specific active stream session', responses={ 202: SUCCESS_EXAMPLES['task_started'], 400: ERROR_EXAMPLES['validation_error'], 500: ERROR_EXAMPLES['not_found'] }, tags=['Sessions'] ) @action(detail=True, methods=['post']) def stop(self, request, pk=None): """ Stop a specific stream session. Args: request: HTTP request pk: Session primary key Returns: Response: Success/error response with task information """ session = self.get_object() if not session.is_running(): return Response( {'error': 'Session is not currently running'}, status=status.HTTP_400_BAD_REQUEST ) try: # Stop the stream capture task task = stop_stream_capture.delay(str(session.id)) return Response({ 'success': True, 'message': f'Stream stopped for {session.channel.name}', 'task_id': task.id }, status=status.HTTP_202_ACCEPTED) except Exception as e: return Response( {'error': str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) @extend_schema( summary='Get session segments', description='Retrieve HLS segments for a specific stream session', parameters=PAGINATION_PARAMETERS, responses={200: HLSSegmentSerializer(many=True)}, tags=['Sessions'] ) @action(detail=True, methods=['get']) def segments(self, request, pk=None): """ Get segments for a specific session. Args: request: HTTP request with optional pagination pk: Session primary key Returns: Response: Paginated list of session segments """ session = self.get_object() segments = session.segments.order_by('-processed_at') # Apply pagination paginator = self.pagination_class() page = paginator.paginate_queryset(segments, request) if page is not None: serializer = HLSSegmentSerializer(page, many=True) return paginator.get_paginated_response(serializer.data) serializer = HLSSegmentSerializer(segments, many=True) return Response(serializer.data) @action(detail=False, methods=['get']) def active(self, request): """ Get all active stream sessions. Args: request: HTTP request Returns: Response: List of active sessions """ active_sessions = self.queryset.filter( status__in=['active', 'processing'] ) serializer = self.get_serializer(active_sessions, many=True) return Response(serializer.data) @extend_schema_view( list=extend_schema( summary='List HLS segments', description='Retrieve a paginated list of HLS segments across all sessions', parameters=PAGINATION_PARAMETERS + [ORDERING_PARAMETER], tags=['Segments'] ), retrieve=extend_schema( summary='Get HLS segment', description='Retrieve a specific HLS segment by ID', tags=['Segments'] ) ) class HLSSegmentViewSet(viewsets.ReadOnlyModelViewSet): """ ViewSet for HLS segments (read-only). Provides read-only access to segment information for monitoring and debugging purposes. """ queryset = HLSSegment.objects.all().select_related('session__channel') serializer_class = HLSSegmentSerializer permission_classes = [permissions.IsAuthenticated] pagination_class = StandardResultsSetPagination filter_backends = [DjangoFilterBackend] filterset_fields = ['session', 'is_available'] ordering_fields = ['processed_at', 'sequence_number'] ordering = ['-processed_at'] @extend_schema( summary='Get recent segments', description='Retrieve the 50 most recently processed HLS segments across all sessions', responses={200: HLSSegmentSerializer(many=True)}, tags=['Segments'] ) @action(detail=False, methods=['get']) def recent(self, request): """ Get recently processed segments across all sessions. Args: request: HTTP request Returns: Response: List of recent segments """ recent_segments = self.queryset.order_by('-processed_at')[:50] serializer = self.get_serializer(recent_segments, many=True) return Response(serializer.data) @extend_schema_view( list=extend_schema( summary='List video configurations', description='Retrieve a list of video encoding configurations', parameters=PAGINATION_PARAMETERS + [ORDERING_PARAMETER], tags=['Configurations'] ), create=extend_schema( summary='Create video configuration', description='Create a new video encoding configuration', tags=['Configurations'] ), retrieve=extend_schema( summary='Get video configuration', description='Retrieve a specific video configuration by ID', tags=['Configurations'] ), update=extend_schema( summary='Update video configuration', description='Update an existing video configuration', tags=['Configurations'] ), partial_update=extend_schema( summary='Partially update video configuration', description='Partially update an existing video configuration', tags=['Configurations'] ), destroy=extend_schema( summary='Delete video configuration', description='Delete a video configuration', tags=['Configurations'] ) ) class VideoConfigurationViewSet(viewsets.ModelViewSet): """ ViewSet for video encoding configurations. Provides CRUD operations for video encoding presets that can be used with stream sessions. """ queryset = VideoConfiguration.objects.all() serializer_class = VideoConfigurationSerializer permission_classes = [permissions.IsAuthenticated] pagination_class = StandardResultsSetPagination ordering_fields = ['name', 'created_at'] ordering = ['name'] @extend_schema_view( list=extend_schema( summary='List audio configurations', description='Retrieve a list of audio encoding configurations', parameters=PAGINATION_PARAMETERS + [ORDERING_PARAMETER], tags=['Configurations'] ), create=extend_schema( summary='Create audio configuration', description='Create a new audio encoding configuration', tags=['Configurations'] ), retrieve=extend_schema( summary='Get audio configuration', description='Retrieve a specific audio configuration by ID', tags=['Configurations'] ), update=extend_schema( summary='Update audio configuration', description='Update an existing audio configuration', tags=['Configurations'] ), partial_update=extend_schema( summary='Partially update audio configuration', description='Partially update an existing audio configuration', tags=['Configurations'] ), destroy=extend_schema( summary='Delete audio configuration', description='Delete an audio configuration', tags=['Configurations'] ) ) class AudioConfigurationViewSet(viewsets.ModelViewSet): """ ViewSet for audio encoding configurations. Provides CRUD operations for audio encoding presets that can be used with stream sessions. """ queryset = AudioConfiguration.objects.all() serializer_class = AudioConfigurationSerializer permission_classes = [permissions.IsAuthenticated] pagination_class = StandardResultsSetPagination ordering_fields = ['name', 'created_at'] ordering = ['name']