import logging from datetime import timedelta from django.core.exceptions import ImproperlyConfigured from django.shortcuts import get_object_or_404 from django.utils import timezone from rest_framework import status from rest_framework.generics import GenericAPIView from rest_framework.permissions import IsAuthenticated, AllowAny from rest_framework.authentication import TokenAuthentication from rest_framework.response import Response from drf_yasg.utils import swagger_auto_schema from drf_yasg import openapi import time import jwt from apps.course.models import Course, CourseLiveSession, Participant, LiveSessionRecording, LiveSessionUser from apps.course.serializers import LiveSessionRoomCreateSerializer, LiveSessionTokenSerializer, LiveSessionRecordedFileSerializer, LiveSessionRecordingSerializer from apps.course.services.plugnmeet import PlugNMeetClient, PlugNMeetError from apps.course.services.web_egress import WebEgressClient, WebEgressError from utils.exceptions import AppAPIException from django.conf import settings logger = logging.getLogger(__name__) def get_live_session_subject(course): title = "" course_title = course.title if isinstance(course_title, list): for tr in course_title: if isinstance(tr, dict) and tr.get('language_code') == 'en': val = tr.get('title') or tr.get('text') or tr.get('value') or tr.get('name') if val: title = str(val).strip() break if not title: for tr in course_title: if isinstance(tr, dict) and tr.get('language_code') == 'ru': val = tr.get('title') or tr.get('text') or tr.get('value') or tr.get('name') if val: title = str(val).strip() break if not title: from apps.course.models.course import extract_text_from_json title = extract_text_from_json(course_title) if not title: title = "Course" return f"{title} Live Session" def get_plugnmeet_webhook_url() -> str: base = getattr(settings, 'SITE_DOMAIN', '').rstrip('/') if not base: raise ImproperlyConfigured('SITE_DOMAIN must be configured for PlugNMeet webhook delivery.') return f"{base}/api/courses/plugnmeet/webhook/" def get_online_class_frontend_base() -> str: base = getattr( settings, "ONLINE_CLASS_FRONTEND_DOMAIN", getattr(settings, "SITE_DOMAIN", ""), ).rstrip("/") if base and not base.startswith(("http://", "https://")): base = f"https://{base}" return base def get_web_egress_callback_url() -> str: base = getattr(settings, "SITE_DOMAIN", "").rstrip("/") if not base: raise ImproperlyConfigured( "SITE_DOMAIN must be configured for WebEgress callback delivery." ) return f"{base}/api/courses/online/room/recording/web-egress/callback/" def decode_plugnmeet_access_token(raw_token: str) -> dict: try: return jwt.decode( raw_token, settings.PLUGNMEET_API_SECRET, algorithms=["HS256"], ) except jwt.PyJWTError as exc: raise AppAPIException( {"message": "Invalid or expired meeting access token."}, status_code=status.HTTP_401_UNAUTHORIZED, ) from exc def extract_meeting_token_from_request(request) -> str: auth_header = request.headers.get("Authorization", "").strip() if auth_header.lower().startswith("bearer "): return auth_header[7:].strip() if auth_header.lower().startswith("token "): return auth_header[6:].strip() if auth_header: return auth_header token = request.data.get("access_token") or request.query_params.get("access_token") if token: return token raise AppAPIException( {"message": "Meeting access token is required."}, status_code=status.HTTP_401_UNAUTHORIZED, ) def get_session_from_meeting_token(request, *, require_admin: bool = False): meeting_token = extract_meeting_token_from_request(request) claims = decode_plugnmeet_access_token(meeting_token) room_id = claims.get("room_id") if not room_id: raise AppAPIException( {"message": "Meeting token is missing room information."}, status_code=status.HTTP_400_BAD_REQUEST, ) session = get_object_or_404(CourseLiveSession, room_id=room_id) if require_admin and not claims.get("is_admin"): raise AppAPIException( {"message": "Only moderators can control recording."}, status_code=status.HTTP_403_FORBIDDEN, ) return session, claims def is_web_egress_recording_status(status_value: str) -> bool: return status_value in { CourseLiveSession.WEB_EGRESS_STATUS_STARTING, CourseLiveSession.WEB_EGRESS_STATUS_RECORDING, } def build_web_egress_title(session: CourseLiveSession) -> str: return session.recording_title or f"{session.subject} - Recording" class CourseLiveSessionRoomCreateAPIView(GenericAPIView): permission_classes = [IsAuthenticated] authentication_classes = [TokenAuthentication] serializer_class = LiveSessionRoomCreateSerializer @swagger_auto_schema( operation_description="Create a live session room for a course", tags=["Imam-Javad - Course"], manual_parameters=[ openapi.Parameter( 'slug', openapi.IN_PATH, description="Course slug", type=openapi.TYPE_STRING, required=True ) ], responses={ 201: openapi.Response( description="Live session room created successfully" ) } ) def post(self, request, slug, *args, **kwargs): # 1. Standard Permissions Logic course = get_object_or_404(Course, slug__contains=[{"title": slug}]) if not request.user.can_manage_course(course): raise AppAPIException({'message': 'Permission denied'}, status_code=403) # 2. Setup ID and Metadata subject = get_live_session_subject(course) # 3. Database Logic - Check FIRST before calling PlugNMeet session = CourseLiveSession.objects.filter( course=course, ended_at__isnull=True ).first() needs_room_creation = False if session: room_id = session.room_id logger.info(f"[LiveSession Create] Found active session - session_id={session.id} room_id={room_id}") else: # No active session, always create a new one with a dynamic room_id timestamp = int(time.time()) room_id = f"room-{course.id}-{timestamp}" session = CourseLiveSession.objects.create( course=course, room_id=room_id, subject=subject, started_at=timezone.now() ) needs_room_creation = True logger.info(f"[LiveSession Create] Created new session - session_id={session.id} room_id={room_id}") # 4. Create room in PlugNMeet ONLY if needed if needs_room_creation: metadata = self._build_metadata(subject) try: client = PlugNMeetClient() plugnmeet_response = client.create_room({ 'room_id': room_id, 'empty_timeout': 90, 'metadata': metadata, }) logger.info(f"[LiveSession Create] Room created in PlugNMeet - room_id={room_id}") except Exception as exc: logger.error(f"[LiveSession Create] PlugNMeet Error: {exc}") # If room creation fails, revert the session changes if session.ended_at is None: session.ended_at = timezone.now() session.save(update_fields=['ended_at', 'updated_at']) raise AppAPIException({'message': f'Failed to create room: {str(exc)}'}, status_code=500) else: logger.info(f"[LiveSession Create] Skipping room creation - room already exists - room_id={room_id}") # 5. Generate the JOIN TOKEN (The Entry Ticket) token_payload = { "room_id": room_id, "user_info": { "name": f"{request.user.first_name} {request.user.last_name}", "user_id": str(request.user.id), "is_admin": True, "is_hidden": False } } pnm_token = jwt.encode( { "iss": settings.PLUGNMEET_API_KEY, "exp": int(time.time()) + 3600, "sub": str(request.user.id), **token_payload }, settings.PLUGNMEET_API_SECRET, algorithm="HS256" ) logger.info(f"[LiveSession Create] Success - session_id={session.id} room_id={room_id} user_id={request.user.id}") return Response({ 'success': True, 'session': {'id': session.id, 'room_id': session.room_id}, 'access_token': pnm_token }, status=201) # def post(self, request, slug, *args, **kwargs): # logger.info(f"[LiveSession Create] Request from user_id={request.user.id} for course={slug}") # data = dict(request.data or {}) # if 'metadata' in data: # logger.warning("[LiveSession Create] 'metadata' provided by client will be ignored for security reasons.") # data.pop('metadata', None) # serializer = self.get_serializer(data=data) # serializer.is_valid(raise_exception=True) # course = get_object_or_404(Course, slug=slug) # if not request.user.can_manage_course(course): # logger.warning(f"[LiveSession Create] Permission denied - user_id={request.user.id} course={slug}") # raise AppAPIException({'message': 'You do not have permission to create a live session for this course.'}, status_code=status.HTTP_403_FORBIDDEN) # logger.info(f"[LiveSession Create] Permission granted for user_id={request.user.id} course={slug}") # subject = serializer.validated_data.get('subject') or f"{course.title} Live Session" # room_id = self._build_room_id(course) # metadata = self._build_metadata(subject) # payload = { # 'room_id': room_id, # 'metadata': metadata, # } # logger.info(f"[LiveSession Create] Calling PlugNMeet API - room_id={room_id} course={slug}") # try: # client = PlugNMeetClient() # plugnmeet_response = client.create_room(payload) # logger.info(f"[LiveSession Create] PlugNMeet room created successfully - room_id={room_id}") # except ImproperlyConfigured as exc: # logger.error(f"[LiveSession Create] Configuration error - {str(exc)}") # raise AppAPIException({'message': str(exc)}, status_code=status.HTTP_500_INTERNAL_SERVER_ERROR) # except PlugNMeetError as exc: # logger.error(f"[LiveSession Create] PlugNMeet API error - room_id={room_id} error={str(exc)}") # detail = exc.response_data or {'message': str(exc)} # status_code = exc.status_code or status.HTTP_502_BAD_GATEWAY # raise AppAPIException(detail, status_code=status_code) # session, created = CourseLiveSession.objects.get_or_create( # course=course, # room_id=room_id, # defaults={ # 'subject': subject, # 'started_at': timezone.now(), # }, # ) # if created: # logger.info(f"[LiveSession Create] New session created - session_id={session.id} room_id={room_id} course={slug}") # else: # logger.info(f"[LiveSession Create] Existing session reactivated - session_id={session.id} room_id={room_id} course={slug}") # updates = {} # if session.subject != subject: # session.subject = subject # updates['subject'] = subject # if session.room_id != room_id: # session.room_id = room_id # updates['room_id'] = room_id # if session.started_at is None: # session.started_at = timezone.now() # updates['started_at'] = session.started_at # if updates: # session.save(update_fields=list(updates.keys())) # logger.info(f"[LiveSession Create] Session updated - session_id={session.id} fields={list(updates.keys())}") # logger.info(f"[LiveSession Create] Success - session_id={session.id} room_id={room_id} course={slug} user_id={request.user.id}") # return Response({ # 'session': { # 'id': session.id, # 'room_id': session.room_id, # 'subject': session.subject, # 'started_at': session.started_at, # }, # 'plugnmeet': plugnmeet_response, # }, status=status.HTTP_201_CREATED if created else status.HTTP_200_OK) @staticmethod def _build_room_id(course: Course) -> str: return f"room-{course.id}-imamjavad" def _build_metadata(self, subject: str) -> dict: # Build secured, centralized metadata. Client overrides are NOT allowed. return { 'room_title': subject, 'webhook_url': get_plugnmeet_webhook_url(), 'default_lock_settings': { 'lock_microphone': True, 'lock_webcam': True, 'lock_screen_sharing': True, 'lock_whiteboard': True, 'lock_shared_notepad': False, 'lock_chat': False, 'lock_chat_send_message': False, 'lock_chat_file_share': False, 'lock_private_chat': False, }, 'room_features': { 'allow_webcams': True, 'mute_on_start': True, 'allow_screen_sharing': True, 'allow_recording': True, 'allow_rtmp': False, 'allow_view_other_webcams': True, 'allow_view_other_participants_list': True, 'admin_only_webcams': False, 'allow_polls': True, 'room_duration': 0, 'chat_features': { 'allow_chat': True, 'allow_file_upload': True, }, 'shared_note_pad_features': { 'allowed_shared_note_pad': True, }, 'whiteboard_features': { 'allowed_whiteboard': True, }, 'breakout_room_features': { 'is_allow': True, 'allowed_number_rooms': 6, }, 'waiting_room_features': { 'is_active': False, }, 'recording_features': { 'is_allow': True, 'is_allow_cloud': True, 'is_allow_local': False, 'enable_auto_cloud_recording': False, 'only_record_admin_webcams': False, }, }, } def _deep_update(self, base: dict, overrides: dict) -> dict: for key, value in overrides.items(): if isinstance(value, dict) and isinstance(base.get(key), dict): base[key] = self._deep_update(base.get(key, {}), value) else: base[key] = value return base class CourseLiveSessionTokenAPIView(GenericAPIView): permission_classes = [IsAuthenticated] authentication_classes = [TokenAuthentication] serializer_class = LiveSessionTokenSerializer @swagger_auto_schema( operation_description="Generate access token for live session", tags=["Imam-Javad - Course"], responses={ 200: openapi.Response( description="Live session token generated successfully" ) } ) def post(self, request, *args, **kwargs): serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) course_slug = serializer.validated_data['course_slug'] user = request.user logger.info(f"[LiveSession Token] Request from user_id={user.id} for course={course_slug}") try: course = Course.objects.filter(slug__contains=[{"title": course_slug}]).first() if not course: raise Course.DoesNotExist except Course.DoesNotExist: logger.warning(f"[LiveSession Token] Course not found - course={course_slug} user_id={user.id}") raise AppAPIException({'message': 'Course not found.'}, status_code=status.HTTP_404_NOT_FOUND) if not course.is_online: logger.warning(f"[LiveSession Token] Course not configured for online - course={course_slug} user_id={user.id}") raise AppAPIException({'message': 'Course is not configured for online sessions.'}, status_code=status.HTTP_400_BAD_REQUEST) try: session = CourseLiveSession.objects.select_related('course').get( course=course, ended_at__isnull=True ) logger.info(f"[LiveSession Token] Active session found in DB - session_id={session.id} room_id={session.room_id} course={course_slug}") except CourseLiveSession.DoesNotExist: logger.warning(f"[LiveSession Token] No active session found - course={course_slug} user_id={user.id}") raise AppAPIException({'message': 'No active live session found for this course.'}, status_code=status.HTTP_404_NOT_FOUND) # Check user role first to determine permissions is_admin = user.can_manage_course(course) user_role = "professor" if is_admin else "student" logger.info(f"[LiveSession Token] User role determined - user_id={user.id} role={user_role} course={course_slug}") # CRITICAL: Verify the room is actually active in PlugNMeet before issuing token # This prevents issuing tokens for rooms that have crashed or ended without webhook notification room_id = session.room_id room_is_active = self._verify_room_is_active(session) if not room_is_active: # Room is not active in PlugNMeet but we have a session record if is_admin: # For professors: Auto-recreate the room in PlugNMeet logger.info(f"[LiveSession Token] Room inactive but professor requesting - recreating room - room_id={room_id} session_id={session.id}") try: self._recreate_room_in_plugnmeet(course, session) logger.info(f"[LiveSession Token] Room recreated successfully - room_id={room_id}") except Exception as e: logger.error(f"[LiveSession Token] Failed to recreate room - room_id={room_id} error={str(e)}") raise AppAPIException({ 'status': 'False', 'message': f'Failed to recreate room: {str(e)}', 'msg': f'Failed to recreate room: {str(e)}' }, status_code=status.HTTP_500_INTERNAL_SERVER_ERROR) else: # For students: Refuse token - they cannot create rooms logger.error(f"[LiveSession Token] Room not active and user is student - refusing token - room_id={room_id} user_id={user.id}") raise AppAPIException({ 'status': 'False', 'message': 'room is not active. Please wait for the professor to start the class.', 'msg': 'room is not active. Please wait for the professor to start the class.' }, status_code=status.HTTP_400_BAD_REQUEST) if not is_admin and not Participant.objects.filter(course=course, student_id=user.id, is_active=True).exists(): logger.warning(f"[LiveSession Token] Access denied - user_id={user.id} not enrolled in course={course_slug}") raise AppAPIException({'message': 'You do not have access to this live session.'}, status_code=status.HTTP_403_FORBIDDEN) user_info = { 'user_id': str(user.id), 'name': user.get_full_name() or user.email or user.username or f"user-{user.id}", 'is_admin': is_admin, } user_metadata = {} profile_pic = self._build_profile_url(request, user) logger.info(f"[LiveSession Token] Profile pic URL - user_id={user.id} url={profile_pic}") if profile_pic: user_metadata['profilePic'] = profile_pic if not is_admin: user_metadata['lock_settings'] = { 'lock_microphone': True, 'lock_screen_sharing': True, 'lock_webcam': True, 'lock_whiteboard': True, 'lock_shared_notepad': False, 'lock_chat': False, 'lock_chat_send_message': False, 'lock_chat_file_share': False, 'lock_private_chat': False, } else: user_metadata['is_hidden'] = False if user_metadata: user_info['user_metadata'] = user_metadata payload = { 'room_id': room_id, 'user_info': user_info, } logger.info(f"[LiveSession Token] Requesting token from PlugNMeet - room_id={room_id} user_id={user.id} role={user_role}") try: client = PlugNMeetClient() plugnmeet_response = client.get_join_token(payload) logger.info(f"[LiveSession Token] Token generated successfully - room_id={room_id} user_id={user.id}") # Fail-safe participant registration in database try: role = 'moderator' if is_admin else 'participant' LiveSessionUser.objects.update_or_create( session=session, user=user, defaults={ 'role': role, 'is_online': True, 'exited_at': None, 'entered_at': timezone.now() } ) logger.info(f"[LiveSession Token] Fail-safe registered user entry - session_id={session.id} user_id={user.id}") except Exception as e: logger.error(f"[LiveSession Token] Failed to register fail-safe user entry: {e}") except ImproperlyConfigured as exc: logger.error(f"[LiveSession Token] Configuration error - {str(exc)}") raise AppAPIException({'message': str(exc)}, status_code=status.HTTP_500_INTERNAL_SERVER_ERROR) except PlugNMeetError as exc: logger.error(f"[LiveSession Token] PlugNMeet API error - room_id={room_id} user_id={user.id} error={str(exc)}") detail = exc.response_data or {'message': str(exc)} status_code = exc.status_code or status.HTTP_502_BAD_GATEWAY raise AppAPIException(detail, status_code=status_code) logger.info(f"[LiveSession Token] Success - room_id={room_id} user_id={user.id} role={user_role} course={course_slug}") return Response({ 'room_id': room_id, 'token': plugnmeet_response.get('token'), 'plugnmeet': plugnmeet_response, }) def _build_metadata(self, subject: str) -> dict: # Build secured, centralized metadata. Client overrides are NOT allowed. return { 'room_title': subject, 'default_lock_settings': { 'lock_microphone': True, 'lock_webcam': True, 'lock_screen_sharing': True, 'lock_whiteboard': True, 'lock_shared_notepad': False, 'lock_chat': False, 'lock_chat_send_message': False, 'lock_chat_file_share': False, 'lock_private_chat': False, }, 'room_features': { 'allow_webcams': True, 'mute_on_start': True, 'allow_screen_sharing': True, 'allow_recording': True, 'allow_rtmp': False, 'allow_view_other_webcams': True, 'allow_view_other_participants_list': True, 'admin_only_webcams': False, 'allow_polls': True, 'room_duration': 0, 'chat_features': { 'allow_chat': True, 'allow_file_upload': True, }, 'shared_note_pad_features': { 'allowed_shared_note_pad': True, }, 'whiteboard_features': { 'allowed_whiteboard': True, }, 'breakout_room_features': { 'is_allow': True, 'allowed_number_rooms': 6, }, 'waiting_room_features': { 'is_active': False, }, 'recording_features': { 'is_allow': True, 'is_allow_cloud': True, 'is_allow_local': False, 'enable_auto_cloud_recording': False, 'only_record_admin_webcams': False, }, }, } @staticmethod def _verify_room_is_active(session: CourseLiveSession) -> bool: """ Verify that the room is actually active in PlugNMeet. Args: session: The CourseLiveSession to verify Returns: bool: True if room is active in PlugNMeet, False otherwise Side effects: - Closes session in database if PlugNMeet reports room is inactive """ if not session.room_id: logger.warning(f"[Room Verify] Session has no room_id - session_id={session.id}") return False try: client = PlugNMeetClient() response = client.is_room_active(session.room_id) # Debug: Log full response logger.debug(f"[Room Verify] PlugNMeet response - room_id={session.room_id} response={response}") # Handle isActive as boolean or string is_active_raw = response.get('isActive', False) is_active = is_active_raw if isinstance(is_active_raw, bool) else str(is_active_raw).lower() == 'true' response_msg = response.get('msg', 'unknown') response_status = response.get('status', False) # Trust status and msg if they indicate active room if response_status and 'active' in response_msg.lower() and 'not' not in response_msg.lower(): is_active = True if is_active: logger.debug(f"[Room Verify] ✓ Room is active - room_id={session.room_id} session_id={session.id}") return True else: logger.warning(f"[Room Verify] ✗ Room is NOT active - room_id={session.room_id} session_id={session.id} msg={response_msg}") # Auto-close the session since room is not active now = timezone.now() session.ended_at = now session.save(update_fields=['ended_at', 'updated_at']) # Also set all users offline try: updated_count = LiveSessionUser.objects.filter( session=session, is_online=True ).update(is_online=False, exited_at=now, updated_at=now) logger.info(f"[Room Verify] Set {updated_count} users offline for session {session.id}") except Exception as e: logger.error(f"[Room Verify] Failed to set users offline: {e}") logger.info(f"[Room Verify] Session auto-closed - session_id={session.id} room_id={session.room_id}") return False except PlugNMeetError as e: error_msg = str(e) logger.error(f"[Room Verify] PlugNMeet API error - room_id={session.room_id} error={error_msg}") # If room not found, close the session if 'not found' in error_msg.lower() or 'does not exist' in error_msg.lower(): now = timezone.now() session.ended_at = now session.save(update_fields=['ended_at', 'updated_at']) # Also set all users offline try: updated_count = LiveSessionUser.objects.filter( session=session, is_online=True ).update(is_online=False, exited_at=now, updated_at=now) logger.info(f"[Room Verify] Set {updated_count} users offline (room not found) for session {session.id}") except Exception as ex: logger.error(f"[Room Verify] Failed to set users offline: {ex}") logger.warning(f"[Room Verify] Room not found - session closed - session_id={session.id}") return False except Exception as e: logger.error(f"[Room Verify] Unexpected error - room_id={session.room_id} error={type(e).__name__}: {str(e)}") return False def _recreate_room_in_plugnmeet(self, course: Course, session: CourseLiveSession) -> None: """ Recreate a room in PlugNMeet when session exists but room is inactive. This happens when: - Webhook failed to notify us of room closure - PlugNMeet server restarted - Room was manually ended Args: course: The course for which to recreate the room session: The existing session to reactivate Raises: PlugNMeetError: If room creation fails """ subject = session.subject or get_live_session_subject(course) room_id = session.room_id metadata = self._build_metadata(subject) payload = { 'room_id': room_id, 'metadata': metadata, } logger.info(f"[Room Recreate] Recreating room in PlugNMeet - room_id={room_id} session_id={session.id}") try: client = PlugNMeetClient() plugnmeet_response = client.create_room(payload) logger.info(f"[Room Recreate] Room recreated successfully - room_id={room_id} response={plugnmeet_response}") # Reset session ended_at to mark it as active again session.ended_at = None session.save(update_fields=['ended_at', 'updated_at']) logger.info(f"[Room Recreate] Session reactivated - session_id={session.id}") except PlugNMeetError as exc: logger.error(f"[Room Recreate] Failed to recreate room - room_id={room_id} error={str(exc)}") raise @staticmethod def _build_profile_url(request, user): avatar = getattr(user, 'avatar', None) if avatar and getattr(avatar, 'url', None): return request.build_absolute_uri(avatar.url) return None class CourseLiveSessionRecordingAPIView(GenericAPIView): permission_classes = [AllowAny] authentication_classes = [] @swagger_auto_schema( operation_description="Get or control WebEgress recording state for the current room.", tags=["Imam-Javad - Course"], responses={200: openapi.Response(description="Recording state returned successfully")}, ) def get(self, request, *args, **kwargs): session, _ = get_session_from_meeting_token(request, require_admin=False) self._refresh_status_from_service(session) return Response(self._build_status_payload(session), status=status.HTTP_200_OK) @swagger_auto_schema( operation_description="Start or stop WebEgress recording for the current room.", tags=["Imam-Javad - Course"], request_body=openapi.Schema( type=openapi.TYPE_OBJECT, required=["action"], properties={ "action": openapi.Schema( type=openapi.TYPE_STRING, enum=["start", "stop"], description="Recording control action.", ), }, ), responses={200: openapi.Response(description="Recording action completed successfully")}, ) def post(self, request, *args, **kwargs): if not getattr(settings, "ONLINE_CLASS_WEB_EGRESS_ENABLED", False): raise AppAPIException( {"message": "WebEgress recording is not enabled."}, status_code=status.HTTP_503_SERVICE_UNAVAILABLE, ) session, claims = get_session_from_meeting_token(request, require_admin=True) action = str(request.data.get("action") or "").strip().lower() if action not in {"start", "stop"}: raise AppAPIException( {"message": "Invalid recording action."}, status_code=status.HTTP_400_BAD_REQUEST, ) if action == "start": payload = self._start_recording(session, claims) else: payload = self._stop_recording(session) return Response(payload, status=status.HTTP_200_OK) def _start_recording(self, session: CourseLiveSession, claims: dict) -> dict: if is_web_egress_recording_status(session.web_egress_status): return self._build_status_payload( session, message="Recording is already active or being processed.", ) client = WebEgressClient() now = timezone.now() session.web_egress_status = CourseLiveSession.WEB_EGRESS_STATUS_STARTING session.web_egress_started_at = now session.web_egress_stopped_at = None session.web_egress_error = None session.save( update_fields=[ "web_egress_status", "web_egress_started_at", "web_egress_stopped_at", "web_egress_error", "updated_at", ] ) try: service_response = client.start_recording( { "session_id": session.id, "course_id": session.course_id, "room_id": session.room_id, "room_title": session.subject, "recording_title": build_web_egress_title(session), "join_url": self._build_web_egress_join_url(session), "callback_url": get_web_egress_callback_url(), "callback_token": getattr( settings, "ONLINE_CLASS_WEB_EGRESS_CALLBACK_TOKEN", "", ), "requested_by": { "user_id": str(claims.get("user_id") or claims.get("sub") or ""), "name": claims.get("name") or "", }, } ) except (ImproperlyConfigured, WebEgressError) as exc: session.web_egress_status = CourseLiveSession.WEB_EGRESS_STATUS_FAILED session.web_egress_error = str(exc) session.save( update_fields=[ "web_egress_status", "web_egress_error", "updated_at", ] ) raise AppAPIException( {"message": str(exc)}, status_code=status.HTTP_502_BAD_GATEWAY, ) session.web_egress_id = service_response.get("egress_id") or session.web_egress_id session.web_egress_status = CourseLiveSession.WEB_EGRESS_STATUS_RECORDING session.web_egress_error = None session.save( update_fields=[ "web_egress_id", "web_egress_status", "web_egress_error", "updated_at", ] ) return self._build_status_payload( session, message=service_response.get("message") or "Recording started successfully.", ) def _stop_recording(self, session: CourseLiveSession) -> dict: if not is_web_egress_recording_status(session.web_egress_status): return self._build_status_payload( session, message="Recording is not active.", ) client = WebEgressClient() try: service_response = client.stop_recording( { "session_id": session.id, "room_id": session.room_id, "egress_id": session.web_egress_id, } ) except (ImproperlyConfigured, WebEgressError) as exc: session.web_egress_status = CourseLiveSession.WEB_EGRESS_STATUS_FAILED session.web_egress_error = str(exc) session.save( update_fields=[ "web_egress_status", "web_egress_error", "updated_at", ] ) raise AppAPIException( {"message": str(exc)}, status_code=status.HTTP_502_BAD_GATEWAY, ) session.web_egress_status = CourseLiveSession.WEB_EGRESS_STATUS_STOPPING session.web_egress_stopped_at = timezone.now() session.web_egress_error = None session.save( update_fields=[ "web_egress_status", "web_egress_stopped_at", "web_egress_error", "updated_at", ] ) return self._build_status_payload( session, message=service_response.get("message") or "Recording stop requested successfully.", ) def _refresh_status_from_service(self, session: CourseLiveSession) -> None: if not getattr(settings, "ONLINE_CLASS_WEB_EGRESS_ENABLED", False): return if not session.web_egress_id and session.web_egress_status in { CourseLiveSession.WEB_EGRESS_STATUS_IDLE, CourseLiveSession.WEB_EGRESS_STATUS_COMPLETED, CourseLiveSession.WEB_EGRESS_STATUS_FAILED, }: return try: client = WebEgressClient() service_response = client.get_status( room_id=session.room_id, session_id=session.id, ) except (ImproperlyConfigured, WebEgressError) as exc: logger.warning( "[WebEgress] Failed to refresh status for session=%s room_id=%s error=%s", session.id, session.room_id, str(exc), ) return normalized_status = self._normalize_service_status(service_response) update_fields = ["updated_at"] if service_response.get("egress_id") is not None and session.web_egress_id != service_response.get("egress_id"): session.web_egress_id = service_response.get("egress_id") update_fields.append("web_egress_id") if normalized_status and session.web_egress_status != normalized_status: session.web_egress_status = normalized_status update_fields.append("web_egress_status") error_message = service_response.get("error") or service_response.get("message") if normalized_status == CourseLiveSession.WEB_EGRESS_STATUS_FAILED: session.web_egress_error = error_message or session.web_egress_error update_fields.append("web_egress_error") if normalized_status == CourseLiveSession.WEB_EGRESS_STATUS_RECORDING and not session.web_egress_started_at: session.web_egress_started_at = timezone.now() update_fields.append("web_egress_started_at") if normalized_status in { CourseLiveSession.WEB_EGRESS_STATUS_STOPPING, CourseLiveSession.WEB_EGRESS_STATUS_PROCESSING, CourseLiveSession.WEB_EGRESS_STATUS_COMPLETED, CourseLiveSession.WEB_EGRESS_STATUS_FAILED, } and not session.web_egress_stopped_at: session.web_egress_stopped_at = timezone.now() update_fields.append("web_egress_stopped_at") if len(update_fields) > 1: session.save(update_fields=update_fields) def _normalize_service_status(self, payload: dict) -> str: raw = str(payload.get("state") or payload.get("status") or "").strip().lower() if raw in {"recording", "active", "started"}: return CourseLiveSession.WEB_EGRESS_STATUS_RECORDING if raw in {"starting", "pending"}: return CourseLiveSession.WEB_EGRESS_STATUS_STARTING if raw in {"stopping"}: return CourseLiveSession.WEB_EGRESS_STATUS_STOPPING if raw in {"processing", "finalizing", "uploading"}: return CourseLiveSession.WEB_EGRESS_STATUS_PROCESSING if raw in {"completed", "done", "finished"}: return CourseLiveSession.WEB_EGRESS_STATUS_COMPLETED if raw in {"failed", "error"}: return CourseLiveSession.WEB_EGRESS_STATUS_FAILED if payload.get("is_recording") is True: return CourseLiveSession.WEB_EGRESS_STATUS_RECORDING if payload.get("is_recording") is False: return CourseLiveSession.WEB_EGRESS_STATUS_IDLE return "" def _build_web_egress_join_url(self, session: CourseLiveSession) -> str: client = PlugNMeetClient() join_response = client.get_join_token( { "room_id": session.room_id, "user_info": { "user_id": f"web-egress-{session.id}", "name": "Session Recorder", "is_admin": True, "is_hidden": True, }, } ) access_token = join_response.get("token") if not access_token: raise WebEgressError("PlugNMeet did not return a recorder access token.") frontend_base = get_online_class_frontend_base() return ( f"{frontend_base}/?access_token={access_token}" f"&web_egress=1&room_id={session.room_id}" ) def _build_status_payload(self, session: CourseLiveSession, message: str | None = None) -> dict: return { "success": True, "room_id": session.room_id, "session_id": session.id, "status": session.web_egress_status, "is_recording": is_web_egress_recording_status(session.web_egress_status), "egress_id": session.web_egress_id, "started_at": session.web_egress_started_at.isoformat() if session.web_egress_started_at else None, "stopped_at": session.web_egress_stopped_at.isoformat() if session.web_egress_stopped_at else None, "error": session.web_egress_error, "message": message, } class CourseLiveSessionWebEgressCallbackAPIView(GenericAPIView): permission_classes = [AllowAny] authentication_classes = [] @swagger_auto_schema( operation_description="Receive final WebEgress recording callback and save the final file.", tags=["Imam-Javad - Course"], responses={201: openapi.Response(description="Recording callback processed successfully")}, ) def post(self, request, *args, **kwargs): self._verify_callback_token(request) room_id = request.data.get("room_id") session_id = request.data.get("session_id") status_value = str(request.data.get("status") or "completed").strip().lower() egress_id = request.data.get("egress_id") error_message = request.data.get("error") or "" if not room_id and not session_id: raise AppAPIException( {"message": "room_id or session_id is required."}, status_code=status.HTTP_400_BAD_REQUEST, ) session = self._get_target_session(room_id=room_id, session_id=session_id) uploaded_file = request.FILES.get("file") duration_seconds = request.data.get("file_time_seconds") recording = None if uploaded_file: file_duration = None if duration_seconds not in (None, ""): try: file_duration = timedelta(seconds=float(duration_seconds)) except (TypeError, ValueError): file_duration = None recording = LiveSessionRecording.objects.create( session=session, file=uploaded_file, title=request.data.get("title") or build_web_egress_title(session), recording_type=request.data.get("recording_type") or "video", file_time=file_duration, ) session.web_egress_id = egress_id or session.web_egress_id session.web_egress_stopped_at = session.web_egress_stopped_at or timezone.now() session.web_egress_error = error_message or None session.web_egress_status = self._normalize_callback_status( status_value=status_value, has_file=bool(uploaded_file), ) session.save( update_fields=[ "web_egress_id", "web_egress_stopped_at", "web_egress_error", "web_egress_status", "updated_at", ] ) response_status = status.HTTP_201_CREATED if recording else status.HTTP_200_OK return Response( { "success": True, "session_id": session.id, "room_id": session.room_id, "recording_id": recording.id if recording else None, "status": session.web_egress_status, }, status=response_status, ) def _verify_callback_token(self, request) -> None: expected = getattr(settings, "ONLINE_CLASS_WEB_EGRESS_CALLBACK_TOKEN", "") if not expected: raise AppAPIException( {"message": "WebEgress callback token is not configured."}, status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, ) received = ( request.headers.get("X-Web-Egress-Token") or request.data.get("callback_token") or "" ) if received != expected: raise AppAPIException( {"message": "Invalid WebEgress callback token."}, status_code=status.HTTP_403_FORBIDDEN, ) def _get_target_session(self, *, room_id=None, session_id=None) -> CourseLiveSession: queryset = CourseLiveSession.objects.order_by("-started_at", "-id") if session_id: try: return queryset.get(id=int(session_id)) except (ValueError, CourseLiveSession.DoesNotExist): pass if room_id: return get_object_or_404(queryset, room_id=room_id) raise AppAPIException( {"message": "Unable to resolve target live session."}, status_code=status.HTTP_404_NOT_FOUND, ) @staticmethod def _normalize_callback_status(*, status_value: str, has_file: bool) -> str: if has_file: return CourseLiveSession.WEB_EGRESS_STATUS_COMPLETED if status_value in {"failed", "error"}: return CourseLiveSession.WEB_EGRESS_STATUS_FAILED if status_value in {"processing", "uploading", "stopping"}: return CourseLiveSession.WEB_EGRESS_STATUS_PROCESSING return CourseLiveSession.WEB_EGRESS_STATUS_COMPLETED class CourseLiveSessionRecordedFileAPIView(GenericAPIView): permission_classes = [AllowAny] authentication_classes = [TokenAuthentication] serializer_class = LiveSessionRecordingSerializer @swagger_auto_schema( operation_description="Update recorded file for live session", tags=["Imam-Javad - Course"], manual_parameters=[ openapi.Parameter( 'room_slug', openapi.IN_PATH, description="Room Slug (e.g. room-16)", type=openapi.TYPE_STRING, required=True ) ], responses={ 200: openapi.Response( description="Recorded file updated successfully" ) } ) def patch(self, request, room_slug, *args, **kwargs): logger.info(f"[LiveSession Recorded File] Request headers: {dict(request.headers)}") user_id = request.user.id if request.user else None logger.info(f"[LiveSession Recorded File] Request from user_id={user_id} for room_slug={room_slug}") import re match = re.search(r'\d+', room_slug) if not match: logger.error(f"[LiveSession Recorded File] Invalid room_slug: {room_slug}") raise AppAPIException({'message': f'Invalid room slug: {room_slug}'}, status_code=status.HTTP_400_BAD_REQUEST) course_id = int(match.group()) logger.info(f"[LiveSession Recorded File] Extracted course_id={course_id} from room_slug={room_slug}") course = get_object_or_404(Course, id=course_id) # If user is authenticated, enforce permissions. # Otherwise, allow the background recording service to proceed. if request.user and request.user.is_authenticated: if not request.user.can_manage_course(course): logger.warning(f"[LiveSession Recorded File] Permission denied - user_id={request.user.id} course_id={course_id}") raise AppAPIException({'message': 'You do not have permission to update this course.'}, status_code=status.HTTP_403_FORBIDDEN) else: logger.info(f"[LiveSession Recorded File] Allowing unauthenticated recording registration for room_slug={room_slug}") try: session = course.live_sessions.latest('-started_at') except CourseLiveSession.DoesNotExist: logger.warning(f"[LiveSession Recorded File] No active session found - course_id={course_id}") raise AppAPIException({'message': 'No live session found for this course.'}, status_code=status.HTTP_404_NOT_FOUND) logger.info(f"[LiveSession Recorded File] Latest session found - session_id={session.id} course_id={course_id}") logger.info(f"[LiveSession Recorded File] Request data: {request.data}") serializer = self.get_serializer(data=request.data) if not serializer.is_valid(): logger.error(f"[LiveSession Recorded File] Validation errors: {serializer.errors}") raise AppAPIException(serializer.errors, status_code=status.HTTP_400_BAD_REQUEST) recording = LiveSessionRecording.objects.create( session=session, file=serializer.validated_data['file'], title=serializer.validated_data.get('title') or f"{session.subject} Recording", recording_type=serializer.validated_data.get('recording_type', 'video'), file_time=serializer.validated_data.get('file_time'), ) logger.info(f"[LiveSession Recorded File] Recording created successfully - recording_id={recording.id} session_id={session.id}") return Response({ 'id': recording.id, 'session_id': session.id, 'title': recording.title, 'file': request.build_absolute_uri(recording.file.url), 'recording_type': recording.recording_type, 'file_time': recording.file_time, 'is_active': recording.is_active, }, status=status.HTTP_201_CREATED) class CourseLiveSessionSetRecordingTitleAPIView(GenericAPIView): permission_classes = [AllowAny] authentication_classes = [] @swagger_auto_schema( operation_description="Set the recording title for the active live session", tags=["Imam-Javad - Course"], request_body=openapi.Schema( type=openapi.TYPE_OBJECT, required=['room_id', 'title'], properties={ 'room_id': openapi.Schema(type=openapi.TYPE_STRING, description='Room ID'), 'title': openapi.Schema(type=openapi.TYPE_STRING, description='Custom Recording Title'), } ), responses={ 200: openapi.Response( description="Recording title updated successfully" ) } ) def post(self, request, *args, **kwargs): room_id = request.data.get('room_id') title = request.data.get('title') if not room_id or not title: raise AppAPIException({'message': 'room_id and title are required.'}, status_code=status.HTTP_400_BAD_REQUEST) try: session = CourseLiveSession.objects.get(room_id=room_id, ended_at__isnull=True) except CourseLiveSession.DoesNotExist: raise AppAPIException({'message': 'Active session not found.'}, status_code=status.HTTP_404_NOT_FOUND) session.recording_title = title.strip() session.save(update_fields=['recording_title', 'updated_at']) logger.info(f"[LiveSession Title] Set recording title for room_id={room_id} to '{title}'") return Response({'success': True, 'message': 'Recording title updated successfully.'}, status=status.HTTP_200_OK)