from django.db.models import Q from rest_framework import viewsets from rest_framework.permissions import IsAuthenticated from rest_framework.authentication import TokenAuthentication from rest_framework.decorators import action from rest_framework.response import Response from rest_framework.exceptions import PermissionDenied from utils.pagination import StandardResultsSetPagination from apps.account.permissions import IsPanelUser, IsSuperAdminOrReadOnlyForProfessor from apps.course.models import ( Course, CourseCategory, CourseChapter, CourseLesson, Lesson, CourseAttachment, Attachment, CourseGlossary, Glossary, CourseLiveSession, LiveSessionUser, LiveSessionRecording ) from apps.course.serializers.admin import ( AdminCourseCategorySerializer, AdminCourseListSerializer, AdminCourseDetailSerializer, AdminCourseChapterSerializer, AdminCourseLessonSerializer, AdminCourseAttachmentSerializer, AdminCourseGlossarySerializer, AdminLessonSerializer, AdminGlossarySerializer, AdminAttachmentSerializer, AdminLiveSessionListSerializer, AdminLiveSessionDetailSerializer ) def is_professor(request): return getattr(request.user, 'user_type', None) == 'professor' class AdminCourseViewSet(viewsets.ModelViewSet): """ Admin CRUD ViewSet for Course management. Professors can only access/modify their own courses. """ permission_classes = [IsAuthenticated, IsPanelUser] authentication_classes = [TokenAuthentication] pagination_class = StandardResultsSetPagination def get_serializer_class(self): if self.action == 'list': return AdminCourseListSerializer return AdminCourseDetailSerializer def get_queryset(self): queryset = Course.objects.all().select_related('category', 'professor') # Professors can only see their own courses if is_professor(self.request): queryset = queryset.filter(professor_id=self.request.user.id) # Search Query search_query = self.request.query_params.get('search', None) has_search = False if search_query: has_search = True from django.db.models import Case, When, Value, IntegerField queryset = queryset.filter( Q(title__icontains=search_query) | Q(description__icontains=search_query) | Q(short_description__icontains=search_query) ).annotate( search_relevance=Case( When(title__icontains=search_query, then=Value(2)), default=Value(1), output_field=IntegerField() ) ) # Filters category_id = self.request.query_params.get('category', None) if category_id: queryset = queryset.filter(category_id=category_id) level = self.request.query_params.get('level', None) if level: queryset = queryset.filter(level__contains=[{"title": level}]) status_param = self.request.query_params.get('status', None) if status_param: queryset = queryset.filter(status__contains=[{"title": status_param}]) is_free_param = self.request.query_params.get('is_free', None) if is_free_param is not None: if is_free_param.lower() == 'true': queryset = queryset.filter(is_free=True) elif is_free_param.lower() == 'false': queryset = queryset.filter(is_free=False) student_id = self.request.query_params.get('student_id', None) if student_id: queryset = queryset.filter(participants__student_id=student_id) if has_search: return queryset.order_by('-search_relevance', '-id') return queryset.order_by('-id') def perform_create(self, serializer): # If the logged-in user is a professor, force the professor field to themselves if is_professor(self.request): serializer.save(professor_id=self.request.user.id) else: serializer.save() def perform_update(self, serializer): # Prevent professors from reassigning a course to another professor if is_professor(self.request): instance = self.get_object() if instance.professor_id != self.request.user.id: raise PermissionDenied("You can only edit your own courses.") serializer.save(professor_id=self.request.user.id) else: serializer.save() class AdminChapterViewSet(viewsets.ModelViewSet): """ Admin CRUD ViewSet for Course Chapters. Professors can only access chapters of their own courses. """ permission_classes = [IsAuthenticated, IsPanelUser] authentication_classes = [TokenAuthentication] serializer_class = AdminCourseChapterSerializer pagination_class = None def get_queryset(self): queryset = CourseChapter.objects.all() # Professors can only see chapters of their own courses if is_professor(self.request): queryset = queryset.filter(course__professor_id=self.request.user.id) course_id = self.request.query_params.get('course', None) if course_id: queryset = queryset.filter(course_id=course_id) return queryset.order_by('priority', 'id') @action(detail=False, methods=['post']) def reorder(self, request): """ Reorders chapters in bulk. Format: {"chapter_ids": [id1, id2, ...]} """ chapter_ids = request.data.get('chapter_ids', []) if not chapter_ids: return Response({'error': 'chapter_ids list is required'}, status=400) from django.db import transaction with transaction.atomic(): for idx, c_id in enumerate(chapter_ids): CourseChapter.objects.filter(id=c_id).update(priority=idx + 1) return Response({'status': 'success'}) class AdminCourseLessonViewSet(viewsets.ModelViewSet): """ Admin CRUD ViewSet for Lessons associated with a Course/Chapter. Professors can only access lessons in their own courses. """ permission_classes = [IsAuthenticated, IsPanelUser] authentication_classes = [TokenAuthentication] serializer_class = AdminCourseLessonSerializer pagination_class = None def get_queryset(self): queryset = CourseLesson.objects.all().select_related('lesson', 'chapter') # Professors can only see lessons of their own courses if is_professor(self.request): queryset = queryset.filter(course__professor_id=self.request.user.id) course_id = self.request.query_params.get('course', None) if course_id: queryset = queryset.filter(course_id=course_id) chapter_id = self.request.query_params.get('chapter', None) if chapter_id: queryset = queryset.filter(chapter_id=chapter_id) return queryset.order_by('priority', 'id') def perform_destroy(self, instance): # Prevent orphans: delete the base Lesson object along with the junction model lesson = instance.lesson instance.delete() if lesson: lesson.delete() @action(detail=False, methods=['post']) def reorder(self, request): """ Reorders lessons in bulk. Format: {"lesson_ids": [id1, id2, ...]} """ lesson_ids = request.data.get('lesson_ids', []) if not lesson_ids: return Response({'error': 'lesson_ids list is required'}, status=400) from django.db import transaction with transaction.atomic(): for idx, l_id in enumerate(lesson_ids): CourseLesson.objects.filter(id=l_id).update(priority=idx + 1) return Response({'status': 'success'}) class AdminCourseAttachmentViewSet(viewsets.ModelViewSet): """ Admin CRUD ViewSet for Course Attachments. Professors can only access attachments of their own courses. """ permission_classes = [IsAuthenticated, IsPanelUser] authentication_classes = [TokenAuthentication] serializer_class = AdminCourseAttachmentSerializer pagination_class = None def get_queryset(self): queryset = CourseAttachment.objects.all().select_related('attachment') # Professors can only see attachments of their own courses if is_professor(self.request): queryset = queryset.filter(course__professor_id=self.request.user.id) course_id = self.request.query_params.get('course', None) if course_id: queryset = queryset.filter(course_id=course_id) return queryset.order_by('-id') def perform_destroy(self, instance): # Prevent orphans: delete base Attachment along with junction model attachment = instance.attachment instance.delete() if attachment: attachment.delete() class AdminCourseGlossaryViewSet(viewsets.ModelViewSet): """ Admin CRUD ViewSet for Course Glossary terms. Professors can only access glossaries of their own courses. """ permission_classes = [IsAuthenticated, IsPanelUser] authentication_classes = [TokenAuthentication] serializer_class = AdminCourseGlossarySerializer pagination_class = None def get_queryset(self): queryset = CourseGlossary.objects.all().select_related('glossary') # Professors can only see glossaries of their own courses if is_professor(self.request): queryset = queryset.filter(course__professor_id=self.request.user.id) course_id = self.request.query_params.get('course', None) if course_id: queryset = queryset.filter(course_id=course_id) return queryset.order_by('-id') def perform_destroy(self, instance): # Prevent orphans: delete base Glossary along with junction model glossary = instance.glossary instance.delete() if glossary: glossary.delete() class AdminCourseCategoryViewSet(viewsets.ModelViewSet): """ Admin CRUD ViewSet for Course Categories. Professors have read-only access. """ permission_classes = [IsAuthenticated, IsSuperAdminOrReadOnlyForProfessor] authentication_classes = [TokenAuthentication] serializer_class = AdminCourseCategorySerializer pagination_class = StandardResultsSetPagination def get_queryset(self): queryset = CourseCategory.objects.all().order_by('-id') search = self.request.query_params.get('search', None) if search: queryset = queryset.filter(Q(name__icontains=search) | Q(slug__icontains=search)) return queryset class AdminLessonViewSet(viewsets.ModelViewSet): """ Admin CRUD ViewSet for base Lesson (Repository). Professors can only see lessons linked to their own courses. """ permission_classes = [IsAuthenticated, IsPanelUser] authentication_classes = [TokenAuthentication] serializer_class = AdminLessonSerializer pagination_class = StandardResultsSetPagination def get_queryset(self): queryset = Lesson.objects.all().order_by('-id') search = self.request.query_params.get('search', None) if search: queryset = queryset.filter(title__icontains=search) content_type = self.request.query_params.get('content_type', None) if content_type: queryset = queryset.filter(content_type__contains=[{"title": content_type}]) return queryset class AdminGlossaryViewSet(viewsets.ModelViewSet): """ Admin CRUD ViewSet for base Glossary (Repository). Professors can only see glossaries linked to their own courses. """ permission_classes = [IsAuthenticated, IsPanelUser] authentication_classes = [TokenAuthentication] serializer_class = AdminGlossarySerializer pagination_class = StandardResultsSetPagination def get_queryset(self): queryset = Glossary.objects.all().order_by('-id') search = self.request.query_params.get('search', None) if search: queryset = queryset.filter(Q(title__icontains=search) | Q(description__icontains=search)) return queryset class AdminAttachmentViewSet(viewsets.ModelViewSet): """ Admin CRUD ViewSet for base Attachment (Repository). Professors can only see attachments linked to their own courses. """ permission_classes = [IsAuthenticated, IsPanelUser] authentication_classes = [TokenAuthentication] serializer_class = AdminAttachmentSerializer pagination_class = StandardResultsSetPagination def get_queryset(self): queryset = Attachment.objects.all().order_by('-id') search = self.request.query_params.get('search', None) if search: queryset = queryset.filter(title__icontains=search) file_type = self.request.query_params.get('type', None) if file_type: image_extensions = ['.jpg', '.jpeg', '.png', '.gif', '.webp', '.svg', '.bmp'] video_extensions = ['.mp4', '.webm', '.ogg', '.mkv', '.mov', '.avi'] audio_extensions = ['.mp3', '.wav', '.ogg', '.m4a', '.aac', '.flac'] if file_type == 'image': q_objs = Q() for ext in image_extensions: q_objs |= Q(file__iendswith=ext) queryset = queryset.filter(q_objs) elif file_type == 'video': q_objs = Q() for ext in video_extensions: q_objs |= Q(file__iendswith=ext) queryset = queryset.filter(q_objs) elif file_type == 'audio': q_objs = Q() for ext in audio_extensions: q_objs |= Q(file__iendswith=ext) queryset = queryset.filter(q_objs) elif file_type == 'other': q_objs = Q() for ext in image_extensions + video_extensions + audio_extensions: q_objs |= Q(file__iendswith=ext) queryset = queryset.exclude(q_objs) return queryset class AdminLiveSessionViewSet(viewsets.ModelViewSet): """ Admin CRUD ViewSet for Course Live Sessions. Professors can only access live sessions of their own courses. """ permission_classes = [IsAuthenticated, IsPanelUser] authentication_classes = [TokenAuthentication] pagination_class = StandardResultsSetPagination def get_serializer_class(self): if self.action == 'list': return AdminLiveSessionListSerializer return AdminLiveSessionDetailSerializer def get_queryset(self): queryset = CourseLiveSession.objects.all().select_related('course') # Professors can only see live sessions of their own courses if is_professor(self.request): queryset = queryset.filter(course__professor_id=self.request.user.id) search = self.request.query_params.get('search', None) if search: queryset = queryset.filter( Q(subject__icontains=search) | Q(room_id__icontains=search) | Q(course__title__icontains=search) ) course_id = self.request.query_params.get('course', None) if course_id: queryset = queryset.filter(course_id=course_id) started_after = self.request.query_params.get('started_after', None) if started_after: queryset = queryset.filter(started_at__gte=started_after) started_before = self.request.query_params.get('started_before', None) if started_before: queryset = queryset.filter(started_at__lte=started_before) ended_after = self.request.query_params.get('ended_after', None) if ended_after: queryset = queryset.filter(ended_at__gte=ended_after) ended_before = self.request.query_params.get('ended_before', None) if ended_before: queryset = queryset.filter(ended_at__lte=ended_before) return queryset.order_by('-started_at')