diff --git a/.env.prod b/.env.prod index 84bda00..de1b1ad 100644 --- a/.env.prod +++ b/.env.prod @@ -25,3 +25,9 @@ captcha_public_key="6LdgCjseAAAAAIwg41-kyyulmwDtqD2Gk3THIwy2" captcha_private_key="6LdgCjseAAAAAPHMsIHuQgYAGTJ7_QlhqG4G0NyS" ONLINE_CLASS_FRONTEND_DOMAIN="meet.imamjavad.online" FCM_API_KEY="" + +ONLINE_CLASS_WEB_EGRESS_ENABLED=True +ONLINE_CLASS_WEB_EGRESS_SERVICE_URL=https://habibmeet.nwhco.ir +ONLINE_CLASS_WEB_EGRESS_SERVICE_TOKEN=9b7f2b3e6c1a4d8f9e2c7a1b5d3f6e8c +ONLINE_CLASS_WEB_EGRESS_CALLBACK_TOKEN=3f9c7e1a8d2b6c4f5a1e9d7b2c8f6a3e +ONLINE_CLASS_WEB_EGRESS_TIMEOUT=20 \ No newline at end of file diff --git a/apps/course/migrations/0014_courselivesession_web_egress_fields.py b/apps/course/migrations/0014_courselivesession_web_egress_fields.py new file mode 100644 index 0000000..af51f5e --- /dev/null +++ b/apps/course/migrations/0014_courselivesession_web_egress_fields.py @@ -0,0 +1,78 @@ +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("course", "0013_courselivesession_auto_close_fields"), + ] + + operations = [ + migrations.AddField( + model_name="courselivesession", + name="web_egress_error", + field=models.TextField( + blank=True, + help_text="Last WebEgress error message, if any.", + null=True, + verbose_name="Web Egress Error", + ), + ), + migrations.AddField( + model_name="courselivesession", + name="web_egress_id", + field=models.CharField( + blank=True, + help_text="Active LiveKit WebEgress identifier for this session.", + max_length=255, + null=True, + verbose_name="Web Egress ID", + ), + ), + migrations.AddField( + model_name="courselivesession", + name="web_egress_started_at", + field=models.DateTimeField( + blank=True, + help_text="Timestamp when WebEgress recording started.", + null=True, + verbose_name="Web Egress Started At", + ), + ), + migrations.AddField( + model_name="courselivesession", + name="web_egress_status", + field=models.CharField( + choices=[ + ("idle", "Idle"), + ("starting", "Starting"), + ("recording", "Recording"), + ("stopping", "Stopping"), + ("processing", "Processing"), + ("completed", "Completed"), + ("failed", "Failed"), + ], + default="idle", + help_text="Current WebEgress recording status for this session.", + max_length=20, + verbose_name="Web Egress Status", + ), + ), + migrations.AddField( + model_name="courselivesession", + name="web_egress_stopped_at", + field=models.DateTimeField( + blank=True, + help_text="Timestamp when WebEgress recording stopped.", + null=True, + verbose_name="Web Egress Stopped At", + ), + ), + migrations.AddIndex( + model_name="courselivesession", + index=models.Index( + fields=["web_egress_status"], + name="course_cour_web_egr_5d55cb_idx", + ), + ), + ] diff --git a/apps/course/models/live_session.py b/apps/course/models/live_session.py index 9e67bf4..da859a2 100644 --- a/apps/course/models/live_session.py +++ b/apps/course/models/live_session.py @@ -6,6 +6,24 @@ from apps.account.models import User class CourseLiveSession(models.Model): + WEB_EGRESS_STATUS_IDLE = "idle" + WEB_EGRESS_STATUS_STARTING = "starting" + WEB_EGRESS_STATUS_RECORDING = "recording" + WEB_EGRESS_STATUS_STOPPING = "stopping" + WEB_EGRESS_STATUS_PROCESSING = "processing" + WEB_EGRESS_STATUS_COMPLETED = "completed" + WEB_EGRESS_STATUS_FAILED = "failed" + + WEB_EGRESS_STATUS_CHOICES = ( + (WEB_EGRESS_STATUS_IDLE, _("Idle")), + (WEB_EGRESS_STATUS_STARTING, _("Starting")), + (WEB_EGRESS_STATUS_RECORDING, _("Recording")), + (WEB_EGRESS_STATUS_STOPPING, _("Stopping")), + (WEB_EGRESS_STATUS_PROCESSING, _("Processing")), + (WEB_EGRESS_STATUS_COMPLETED, _("Completed")), + (WEB_EGRESS_STATUS_FAILED, _("Failed")), + ) + course = models.ForeignKey( Course, on_delete=models.CASCADE, @@ -55,6 +73,38 @@ class CourseLiveSession(models.Model): null=True, blank=True, ) + web_egress_id = models.CharField( + max_length=255, + verbose_name=_("Web Egress ID"), + help_text=_("Active LiveKit WebEgress identifier for this session."), + null=True, + blank=True, + ) + web_egress_status = models.CharField( + max_length=20, + choices=WEB_EGRESS_STATUS_CHOICES, + default=WEB_EGRESS_STATUS_IDLE, + verbose_name=_("Web Egress Status"), + help_text=_("Current WebEgress recording status for this session."), + ) + web_egress_started_at = models.DateTimeField( + verbose_name=_("Web Egress Started At"), + help_text=_("Timestamp when WebEgress recording started."), + null=True, + blank=True, + ) + web_egress_stopped_at = models.DateTimeField( + verbose_name=_("Web Egress Stopped At"), + help_text=_("Timestamp when WebEgress recording stopped."), + null=True, + blank=True, + ) + web_egress_error = models.TextField( + verbose_name=_("Web Egress Error"), + help_text=_("Last WebEgress error message, if any."), + null=True, + blank=True, + ) recorded_file = models.FileField( upload_to="live_session_recordings/", verbose_name=_("Recorded File"), @@ -77,6 +127,7 @@ class CourseLiveSession(models.Model): models.Index(fields=["course", "created_at"]), models.Index(fields=["room_id"]), models.Index(fields=["ended_at", "auto_close_after_moderator_exit_at"]), + models.Index(fields=["web_egress_status"]), ] diff --git a/apps/course/services/__init__.py b/apps/course/services/__init__.py index 2dc8b89..cbecc3f 100644 --- a/apps/course/services/__init__.py +++ b/apps/course/services/__init__.py @@ -1,3 +1,8 @@ from .plugnmeet import PlugNMeetClient, PlugNMeetError -__all__ = ['PlugNMeetClient', 'PlugNMeetError'] +__all__ = [ + 'PlugNMeetClient', + 'PlugNMeetError', + 'WebEgressClient', + 'WebEgressError', +] diff --git a/apps/course/services/web_egress.py b/apps/course/services/web_egress.py new file mode 100644 index 0000000..1692f5c --- /dev/null +++ b/apps/course/services/web_egress.py @@ -0,0 +1,109 @@ +from __future__ import annotations + +from typing import Any, Dict, Optional + +import requests +from django.conf import settings +from django.core.exceptions import ImproperlyConfigured + + +class WebEgressError(Exception): + def __init__( + self, + message: str, + *, + status_code: Optional[int] = None, + response_data: Optional[Dict[str, Any]] = None, + ): + super().__init__(message) + self.status_code = status_code + self.response_data = response_data or {} + + +class WebEgressClient: + def __init__( + self, + *, + base_url: Optional[str] = None, + service_token: Optional[str] = None, + timeout: Optional[float] = None, + ): + self.base_url = ( + base_url + or getattr(settings, "ONLINE_CLASS_WEB_EGRESS_SERVICE_URL", "") + ).rstrip("/") + self.service_token = ( + service_token + or getattr(settings, "ONLINE_CLASS_WEB_EGRESS_SERVICE_TOKEN", "") + ) + self.timeout = timeout or getattr( + settings, "ONLINE_CLASS_WEB_EGRESS_TIMEOUT", 20.0 + ) + + if not self.base_url: + raise ImproperlyConfigured( + "ONLINE_CLASS_WEB_EGRESS_SERVICE_URL must be configured." + ) + + def start_recording(self, payload: Dict[str, Any]) -> Dict[str, Any]: + return self._request("POST", "/api/web-egress/start/", json=payload) + + def stop_recording(self, payload: Dict[str, Any]) -> Dict[str, Any]: + return self._request("POST", "/api/web-egress/stop/", json=payload) + + def get_status(self, *, room_id: str, session_id: int) -> Dict[str, Any]: + return self._request( + "GET", + "/api/web-egress/status/", + params={"room_id": room_id, "session_id": session_id}, + ) + + def _request(self, method: str, path: str, **kwargs) -> Dict[str, Any]: + headers = kwargs.pop("headers", {}) + headers["Content-Type"] = "application/json" + if self.service_token: + headers["Authorization"] = f"Bearer {self.service_token}" + + try: + response = requests.request( + method, + f"{self.base_url}{path}", + headers=headers, + timeout=self.timeout, + **kwargs, + ) + except requests.RequestException as exc: + raise WebEgressError("Failed to reach WebEgress service.") from exc + + data = self._safe_json(response) + if response.status_code >= 400: + message = ( + (data or {}).get("message") + or (data or {}).get("detail") + or response.text + or "WebEgress service request failed." + ) + raise WebEgressError( + message, + status_code=response.status_code, + response_data=data, + ) + + if data is None: + raise WebEgressError("WebEgress service returned an invalid response.") + + if isinstance(data, dict) and data.get("status") is False: + raise WebEgressError( + data.get("message") or data.get("detail") or "WebEgress failed.", + status_code=response.status_code, + response_data=data, + ) + + return data + + @staticmethod + def _safe_json(response: requests.Response) -> Optional[Dict[str, Any]]: + try: + return response.json() + except ValueError: + return None diff --git a/apps/course/urls.py b/apps/course/urls.py index b7f8ef1..9604841 100644 --- a/apps/course/urls.py +++ b/apps/course/urls.py @@ -31,6 +31,8 @@ urlpatterns = [ path('online/token/validate/', views.CourseOnlineClassTokenValidateAPIView.as_view(), name='course-online-token-validate'), re_path(r'(?P[\w-]+)/online/room/create/$', views.CourseLiveSessionRoomCreateAPIView.as_view(), name='course-live-session-room-create'), path('online/room/token/', views.CourseLiveSessionTokenAPIView.as_view(), name='course-live-session-token'), + path('online/room/recording/', views.CourseLiveSessionRecordingAPIView.as_view(), name='course-live-session-recording'), + path('online/room/recording/web-egress/callback/', views.CourseLiveSessionWebEgressCallbackAPIView.as_view(), name='course-live-session-web-egress-callback'), path('online/room/set-recording-title/', views.CourseLiveSessionSetRecordingTitleAPIView.as_view(), name='course-live-session-set-recording-title'), path('live-sessions//recorded-file/', views.CourseLiveSessionRecordedFileAPIView.as_view(), name='course-live-session-recorded-file'), diff --git a/apps/course/views/live_session.py b/apps/course/views/live_session.py index 3cb6c77..3812549 100644 --- a/apps/course/views/live_session.py +++ b/apps/course/views/live_session.py @@ -1,4 +1,5 @@ import logging +from datetime import timedelta from django.core.exceptions import ImproperlyConfigured from django.shortcuts import get_object_or_404 @@ -16,6 +17,7 @@ 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__) @@ -53,6 +55,89 @@ def get_plugnmeet_webhook_url() -> str: 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] @@ -656,6 +741,420 @@ class CourseLiveSessionTokenAPIView(GenericAPIView): 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] diff --git a/apps/course/views/webhook.py b/apps/course/views/webhook.py index 605cca1..90e2b01 100644 --- a/apps/course/views/webhook.py +++ b/apps/course/views/webhook.py @@ -210,7 +210,18 @@ class PlugNMeetWebhookAPIView(APIView): session = CourseLiveSession.objects.get(room_id=room_id, ended_at__isnull=True) now = timezone.now() session.ended_at = now - session.save(update_fields=['ended_at', 'updated_at']) + update_fields = ['ended_at', 'updated_at'] + if session.web_egress_status in { + CourseLiveSession.WEB_EGRESS_STATUS_STARTING, + CourseLiveSession.WEB_EGRESS_STATUS_RECORDING, + CourseLiveSession.WEB_EGRESS_STATUS_STOPPING, + }: + session.web_egress_status = CourseLiveSession.WEB_EGRESS_STATUS_PROCESSING + update_fields.append('web_egress_status') + if not session.web_egress_stopped_at: + session.web_egress_stopped_at = now + update_fields.append('web_egress_stopped_at') + session.save(update_fields=update_fields) # Close active user sessions updated_count = LiveSessionUser.objects.filter( diff --git a/config/settings/base.py b/config/settings/base.py index e480ba0..00f228c 100644 --- a/config/settings/base.py +++ b/config/settings/base.py @@ -314,6 +314,23 @@ SITE_DOMAIN = "https://imamjavad.online" DOVODI_DOMAIN = "https://dovodi.newhorizonco.uk" ONLINE_CLASS_FRONTEND_DOMAIN = env('ONLINE_CLASS_FRONTEND_DOMAIN', default=SITE_DOMAIN) ONLINE_CLASS_TOKEN_TTL = env.int('ONLINE_CLASS_TOKEN_TTL', default=3000) +ONLINE_CLASS_WEB_EGRESS_ENABLED = env.bool('ONLINE_CLASS_WEB_EGRESS_ENABLED', default=False) +ONLINE_CLASS_WEB_EGRESS_SERVICE_URL = env( + 'ONLINE_CLASS_WEB_EGRESS_SERVICE_URL', + default='', +) +ONLINE_CLASS_WEB_EGRESS_SERVICE_TOKEN = env( + 'ONLINE_CLASS_WEB_EGRESS_SERVICE_TOKEN', + default='', +) +ONLINE_CLASS_WEB_EGRESS_CALLBACK_TOKEN = env( + 'ONLINE_CLASS_WEB_EGRESS_CALLBACK_TOKEN', + default='', +) +ONLINE_CLASS_WEB_EGRESS_TIMEOUT = env.float( + 'ONLINE_CLASS_WEB_EGRESS_TIMEOUT', + default=20.0, +) PLUGNMEET_SERVER_URL ='https://meet.imamjavad.online' PLUGNMEET_API_KEY ='habibmeet_api_key_2024' PLUGNMEET_API_SECRET ='habibmeet_secret_zumyyYWqv7KR2kUqvYdq4z4sXg7XTBD2ljT6_2024'