You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

713 lines
26 KiB

from rest_framework import serializers
# from dj_filer.admin import get_thumbs
from utils import get_thumbs
from apps.course.models import Course, CourseCategory, Attachment, Glossary, LessonCompletion, Participant, Lesson, CourseAttachment, CourseGlossary, CourseLesson
from apps.course.models.course import (
get_localized_field,
get_weekday_label,
normalize_level,
normalize_status,
normalize_weekday_to_key,
)
from apps.chat.models import RoomMessage
from apps.account.serializers import UserProfileSerializer
from decimal import Decimal
import ast
import json
def resolve_request_lang(request, default='en'):
language_code = getattr(request, 'LANGUAGE_CODE', None)
if language_code:
return str(language_code).lower()
if not request:
return default
header_value = ''
if hasattr(request, 'headers'):
header_value = request.headers.get('Accept-Language', '')
if not header_value and hasattr(request, 'META'):
header_value = request.META.get('HTTP_ACCEPT_LANGUAGE', '')
if header_value:
return header_value.split(',')[0].strip().lower()
return default
class CourseCategorySerializer(serializers.ModelSerializer):
course_count = serializers.SerializerMethodField()
name = serializers.SerializerMethodField()
slug = serializers.SerializerMethodField()
class Meta:
model = CourseCategory
fields = ['id', 'name', 'slug', 'icon', 'course_count']
def _lang(self):
request = self.context.get('request')
return resolve_request_lang(request)
def get_name(self, obj):
return get_localized_field(self._lang(), obj.name)
def get_slug(self, obj):
return get_localized_field(self._lang(), obj.slug)
def get_course_count(self, obj):
return obj.course_count
def to_representation(self, instance):
ret = super().to_representation(instance)
icon = ret.get('icon')
if icon and icon.startswith('/'):
request = self.context.get('request')
if request:
ret['icon'] = request.build_absolute_uri(icon)
else:
from django.conf import settings
ret['icon'] = f"{settings.SITE_DOMAIN}{icon}"
return ret
class CourseListSerializer(serializers.ModelSerializer):
category = CourseCategorySerializer()
thumbnail = serializers.SerializerMethodField()
participant_count = serializers.SerializerMethodField()
lessons_count = serializers.SerializerMethodField()
price = serializers.SerializerMethodField()
discount_percentage = serializers.SerializerMethodField()
final_price = serializers.SerializerMethodField()
currency = serializers.SerializerMethodField()
is_free = serializers.SerializerMethodField()
title = serializers.SerializerMethodField()
slug = serializers.SerializerMethodField()
level = serializers.SerializerMethodField()
status = serializers.SerializerMethodField()
short_description = serializers.SerializerMethodField()
class Meta:
model = Course
fields = [
'id',
'title',
'slug',
'participant_count',
'category',
'thumbnail',
'is_online',
'online_link',
'level',
'duration',
'lessons_count',
'short_description',
'status',
'is_free',
'price',
'discount_percentage',
'final_price',
'currency',
]
def _lang(self):
request = self.context.get('request')
return resolve_request_lang(request)
def get_title(self, obj):
return get_localized_field(self._lang(), obj.title)
def get_slug(self, obj):
return get_localized_field(self._lang(), obj.slug)
def get_level(self, obj):
return normalize_level(obj.level)
def get_status(self, obj):
return normalize_status(obj.status)
def get_short_description(self, obj):
return get_localized_field(self._lang(), obj.short_description)
def get_thumbnail(self, obj):
return get_thumbs(obj.thumbnail, self.context.get('request'))
def get_participant_count(self, obj):
return obj.participants.count()
def get_lessons_count(self, obj):
# Use prefetched chapters and lessons if available
if hasattr(obj, 'chapters') and obj.chapters.all():
lessons_count = 0
for chapter in obj.chapters.all():
if chapter.is_active:
lessons_count += sum(1 for lesson in chapter.lessons.all() if lesson.is_active)
return max(lessons_count, obj.lessons_count)
# Fallback to direct query
from apps.course.models.lesson import CourseLesson
lessons_count = CourseLesson.objects.filter(chapter__course=obj, is_active=True, chapter__is_active=True).count()
return max(lessons_count, obj.lessons_count)
def get_price(self, obj):
amount = obj.get_price_for_language(self._lang())
if obj.is_free or amount == 0:
return "0.00"
return str(amount)
def get_discount_percentage(self, obj):
if obj.is_free or obj.get_price_for_language(self._lang()) == 0:
return 0
return obj.discount_percentage
def get_final_price(self, obj):
amount = obj.get_final_price_for_language(self._lang())
if obj.is_free or amount == 0:
return "0.00"
return str(amount)
def get_currency(self, obj):
return obj.get_currency_for_language(self._lang())
def get_is_free(self, obj):
return obj.is_free or (Decimal(obj.price or 0) == 0 and Decimal(obj.price_rub or 0) == 0)
class CourseDetailSerializer(serializers.ModelSerializer):
category = CourseCategorySerializer()
professor = serializers.SerializerMethodField()
thumbnail = serializers.SerializerMethodField()
participant_count = serializers.SerializerMethodField()
access = serializers.SerializerMethodField()
lessons_complated_count = serializers.SerializerMethodField()
lessons_count = serializers.SerializerMethodField()
last_lesson_id = serializers.SerializerMethodField()
room_id = serializers.SerializerMethodField()
user_transaction_status = serializers.SerializerMethodField()
price = serializers.SerializerMethodField()
discount_percentage = serializers.SerializerMethodField()
final_price = serializers.SerializerMethodField()
currency = serializers.SerializerMethodField()
is_free = serializers.SerializerMethodField()
is_professor = serializers.SerializerMethodField()
title = serializers.SerializerMethodField()
slug = serializers.SerializerMethodField()
level = serializers.SerializerMethodField()
status = serializers.SerializerMethodField()
timing = serializers.SerializerMethodField()
features = serializers.SerializerMethodField()
description = serializers.SerializerMethodField()
short_description = serializers.SerializerMethodField()
class Meta:
model = Course
fields = [
'id',
'title',
'slug',
'category',
'access',
'participant_count',
'professor',
'is_professor',
'thumbnail',
'video_type',
'video_file',
'video_link',
'is_online',
'online_link',
'level',
'description',
'duration',
'lessons_count',
'lessons_complated_count',
'short_description',
'status',
'is_free',
'price',
'discount_percentage',
'final_price',
'currency',
'timing',
'features',
'last_lesson_id',
'room_id',
'user_transaction_status',
'is_group_chat_locked',
'is_professor_chat_locked'
]
def get_room_id(self, obj):
# Use prefetched room_messages if available
if hasattr(obj, 'room_messages') and obj.room_messages.all():
return obj.room_messages.first().id
# Fallback to direct query if not prefetched
room_message = RoomMessage.objects.filter(course=obj).first()
if room_message:
return room_message.id
return None
def get_user_transaction_status(self, obj):
from apps.transaction.models import TransactionParticipant
if student := self._get_authenticated_user():
latest_transaction = TransactionParticipant.objects.filter(
user=student,
course=obj,
is_deleted=False
).order_by('-created_at').first()
if latest_transaction:
return latest_transaction.status
return None
def get_last_lesson_id(self, obj):
request = self.context.get('request')
if request and request.user.is_authenticated:
user = request.user
# Gather all active lessons across all active chapters
all_active_lessons = []
if hasattr(obj, 'chapters') and obj.chapters.all():
for chapter in obj.chapters.all():
if chapter.is_active:
all_active_lessons.extend([l for l in chapter.lessons.all() if l.is_active])
# Sort them by chapter priority first, then lesson priority
all_active_lessons.sort(key=lambda x: (x.chapter.priority, x.priority))
completed_lessons = []
for lesson in all_active_lessons:
if hasattr(lesson, 'completions') and lesson.completions.all():
if any(completion.student_id == user.id for completion in lesson.completions.all()):
completed_lessons.append(lesson)
if completed_lessons:
# Find the last completed lesson
last_completed = completed_lessons[-1]
# Find next lesson
# We look for lessons that come AFTER the last completed one in our sorted list
last_completed_index = all_active_lessons.index(last_completed)
if last_completed_index + 1 < len(all_active_lessons):
return all_active_lessons[last_completed_index + 1].id
# If no completed lessons or no next lesson, return first lesson
if all_active_lessons:
return all_active_lessons[0].id
# Fallback to direct queries if not prefetched
last_completed_lesson = LessonCompletion.objects.filter(
student=user,
course_lesson__chapter__course=obj,
course_lesson__is_active=True,
course_lesson__chapter__is_active=True
).order_by('-completed_at').first()
from apps.course.models.lesson import CourseLesson
if last_completed_lesson:
cl = last_completed_lesson.course_lesson
# Try to find next lesson in same chapter
next_lesson = CourseLesson.objects.filter(
chapter=cl.chapter,
priority__gt=cl.priority,
is_active=True
).order_by('priority').first()
# If no more lessons in this chapter, find first lesson in next chapter
if not next_lesson:
next_lesson = CourseLesson.objects.filter(
chapter__course=obj,
chapter__priority__gt=cl.chapter.priority,
chapter__is_active=True,
is_active=True
).order_by('chapter__priority', 'priority').first()
if next_lesson:
return next_lesson.id
else:
# No completions at all, just get the very first lesson
first_lesson = CourseLesson.objects.filter(
chapter__course=obj,
chapter__is_active=True,
is_active=True
).order_by('chapter__priority', 'priority').first()
if first_lesson:
return first_lesson.id
return None
def get_access(self, obj):
if user := self._get_authenticated_user():
return self._has_access(user, obj)
return False
def get_professor(self, obj):
"""Return the course professor's profile using UserProfileSerializer"""
if obj.professor:
return UserProfileSerializer(obj.professor, context=self.context).data
return None
def get_is_professor(self, obj):
if professor := self._get_authenticated_user():
return obj.professor == professor
return False
def get_lessons_count(self, obj):
if hasattr(obj, 'chapters') and obj.chapters.all():
lessons_count = 0
for chapter in obj.chapters.all():
if chapter.is_active:
lessons_count += sum(1 for lesson in chapter.lessons.all() if lesson.is_active)
return max(lessons_count, obj.lessons_count)
from apps.course.models.lesson import CourseLesson
lessons_count = CourseLesson.objects.filter(chapter__course=obj, is_active=True, chapter__is_active=True).count()
return max(lessons_count, obj.lessons_count)
def get_lessons_complated_count(self, obj):
if student := self._get_authenticated_user():
if not self._is_participant(student, obj):
return None
completed_count = self._get_completed_lessons_count(student, obj)
# Ensure completed count doesn't exceed total lessons count
total_lessons = self.get_lessons_count(obj)
return min(completed_count, total_lessons)
return None
def _has_access(self, user, course):
"""
Check if the user has access to the course content.
Access is granted if:
1. User is the professor of the course.
2. User is staff (admin).
3. User is an active participant in the course.
"""
if user.is_staff or user.is_superuser:
return True
if course.professor_id == user.id:
return True
return Participant.objects.filter(
student_id=user.id,
course=course,
is_active=True
).exists()
def _is_participant(self, student, course):
"""Deprecated: use _has_access instead. Kept for backward compatibility if needed."""
return self._has_access(student, course)
def _get_authenticated_user(self):
"""Helper method to retrieve the authenticated user from the context."""
request = self.context.get('request')
return request.user if request and request.user.is_authenticated else None
def _get_completed_lessons_count(self, student, course):
if hasattr(course, 'chapters') and course.chapters.all():
completed_count = 0
for chapter in course.chapters.all():
if chapter.is_active:
for lesson in chapter.lessons.all():
if lesson.is_active and hasattr(lesson, 'completions') and lesson.completions.all():
if any(completion.student_id == student.id for completion in lesson.completions.all()):
completed_count += 1
return completed_count
return LessonCompletion.objects.filter(
student=student,
course_lesson__chapter__course=course,
course_lesson__is_active=True,
course_lesson__chapter__is_active=True
).count()
def get_thumbnail(self, obj):
return get_thumbs(obj.thumbnail, self.context.get('request'))
def get_participant_count(self, obj):
# Use prefetched participants if available
if hasattr(obj, 'participants') and obj.participants.all():
return len(obj.participants.all())
# Fallback to direct query
return obj.participants.count()
def get_price(self, obj):
amount = obj.get_price_for_language(self._lang())
if obj.is_free or amount == 0:
return "0.00"
return str(amount)
def get_discount_percentage(self, obj):
if obj.is_free or obj.get_price_for_language(self._lang()) == 0:
return 0
return obj.discount_percentage
def get_final_price(self, obj):
amount = obj.get_final_price_for_language(self._lang())
if obj.is_free or amount == 0:
return "0.00"
return str(amount)
def get_currency(self, obj):
return obj.get_currency_for_language(self._lang())
def get_is_free(self, obj):
return obj.is_free or (Decimal(obj.price or 0) == 0 and Decimal(obj.price_rub or 0) == 0)
def _lang(self):
request = self.context.get('request')
return resolve_request_lang(request)
def get_title(self, obj):
return get_localized_field(self._lang(), obj.title)
def get_slug(self, obj):
return get_localized_field(self._lang(), obj.slug)
def get_level(self, obj):
return normalize_level(obj.level)
def get_status(self, obj):
return normalize_status(obj.status)
def get_description(self, obj):
return get_localized_field(self._lang(), obj.description)
def get_short_description(self, obj):
return get_localized_field(self._lang(), obj.short_description)
def _parse_builder_payload(self, value):
if value is None:
return None
if isinstance(value, str):
for parser in (json.loads, ast.literal_eval):
try:
parsed = parser(value)
return parsed
except Exception:
continue
return value
return value
def _extract_builder_list(self, value):
"""
Return the raw builder list for the current language.
Supports both:
- direct lists like [{'day': 'monday', 'timing': '12:22'}, ...]
- multilingual wrappers like [{'language_code': 'en', 'title': "[...]"}, ...]
"""
parsed = self._parse_builder_payload(value)
if parsed is None:
return []
if isinstance(parsed, list) and parsed:
if all(isinstance(item, dict) and 'language_code' in item for item in parsed):
current_lang = self._lang()
selected = next(
(
item for item in parsed
if isinstance(item, dict) and item.get('language_code') == current_lang
),
parsed[0],
)
parsed = self._parse_builder_payload(
selected.get('title')
or selected.get('text')
or selected.get('value')
or selected.get('name')
or []
)
if isinstance(parsed, list):
return parsed
if isinstance(parsed, dict):
if isinstance(parsed.get('list'), list):
return parsed['list']
if isinstance(parsed.get('items'), list):
return parsed['items']
return []
def get_timing(self, obj):
localized_rows = []
for item in self._extract_builder_list(obj.timing):
if not isinstance(item, dict):
continue
day_key = normalize_weekday_to_key(item.get('day'))
localized_rows.append({
"day": get_weekday_label(day_key, self._lang()),
"time": item.get('time') or item.get('timing') or "",
})
return localized_rows
def get_features(self, obj):
return self._extract_builder_list(obj.features)
class MyCourseListSerializer(serializers.ModelSerializer):
category = CourseCategorySerializer()
thumbnail = serializers.SerializerMethodField()
lessons_count = serializers.SerializerMethodField()
lessons_complated_count = serializers.SerializerMethodField()
title = serializers.SerializerMethodField()
slug = serializers.SerializerMethodField()
status = serializers.SerializerMethodField()
short_description = serializers.SerializerMethodField()
class Meta:
model = Course
fields = [
'id',
'title',
'slug',
'category',
'thumbnail',
'lessons_count',
'lessons_complated_count',
'short_description',
'status',
]
def get_thumbnail(self, obj):
return get_thumbs(obj.thumbnail, self.context.get('request'))
def get_lessons_count(self, obj):
"""Get the actual count of active lessons"""
# Use prefetched lessons if available
if hasattr(obj, 'lessons') and obj.lessons.all():
lessons_count = sum(1 for lesson in obj.lessons.all() if lesson.is_active)
return max(lessons_count, obj.lessons_count)
# Fallback to direct query
lessons_count = obj.lessons.filter(is_active=True).count()
return max(lessons_count, obj.lessons_count)
def get_lessons_complated_count(self, obj):
if student := self._get_authenticated_user():
if not self._is_participant(student, obj):
return None
completed_count = self._get_completed_lessons_count(student, obj)
# Ensure completed count doesn't exceed total lessons count
total_lessons = self.get_lessons_count(obj)
return min(completed_count, total_lessons)
return None
def _is_participant(self, student, course):
"""Helper method to check if a student is a participant in the given course."""
if student.is_staff or student.is_superuser:
return True
# اگر کاربر استاد دوره است، دسترسی کامل دارد
if course.professor_id == student.id:
return True
# در غیر این صورت چک می‌کنیم که آیا participant است یا خیر
return Participant.objects.filter(
student_id=student.id,
course=course,
is_active=True
).exists()
def _get_authenticated_user(self):
"""Helper method to retrieve the authenticated user from the context."""
request = self.context.get('request')
return request.user if request and request.user.is_authenticated else None
def _get_completed_lessons_count(self, student, course):
"""Helper method to count completed lessons for the student in the given course."""
# Use prefetched completions if available
if hasattr(course, 'lessons') and course.lessons.all():
completed_count = 0
for lesson in course.lessons.all():
if hasattr(lesson, 'completions') and lesson.completions.all():
if any(completion.student_id == student.id for completion in lesson.completions.all()):
completed_count += 1
return completed_count
# Fallback to direct query if not prefetched
return LessonCompletion.objects.filter(
student=student,
course_lesson__course=course
).count()
def _lang(self):
request = self.context.get('request')
return getattr(request, 'LANGUAGE_CODE', None) or 'en'
def get_title(self, obj):
return get_localized_field(self._lang(), obj.title)
def get_slug(self, obj):
return get_localized_field(self._lang(), obj.slug)
def get_status(self, obj):
return normalize_status(obj.status)
def get_short_description(self, obj):
return get_localized_field(self._lang(), obj.short_description)
class AttachmentSerializer(serializers.ModelSerializer):
title = serializers.SerializerMethodField()
class Meta:
model = Attachment
fields = ['id', 'title', 'file', 'file_size']
def _lang(self):
request = self.context.get('request')
return getattr(request, 'LANGUAGE_CODE', None) or 'en'
def get_title(self, obj):
return get_localized_field(self._lang(), obj.title)
class CourseAttachmentSerializer(serializers.ModelSerializer):
title = serializers.SerializerMethodField()
file = serializers.FileField(source='attachment.file', read_only=True)
file_size = serializers.IntegerField(source='attachment.file_size', read_only=True)
class Meta:
model = CourseAttachment
fields = ['id', 'title', 'file', 'file_size']
def _lang(self):
request = self.context.get('request')
return getattr(request, 'LANGUAGE_CODE', None) or 'en'
def get_title(self, obj):
return get_localized_field(self._lang(), obj.attachment.title)
class GlossarySerializer(serializers.ModelSerializer):
description = serializers.SerializerMethodField()
class Meta:
model = Glossary
fields = ['id', 'title', 'description']
def _lang(self):
request = self.context.get('request')
return getattr(request, 'LANGUAGE_CODE', None) or 'en'
def get_description(self, obj):
return get_localized_field(self._lang(), obj.description)
class CourseGlossarySerializer(serializers.ModelSerializer):
title = serializers.CharField(source='glossary.title', read_only=True)
description = serializers.SerializerMethodField()
class Meta:
model = CourseGlossary
fields = ['id', 'title', 'description']
def _lang(self):
request = self.context.get('request')
return getattr(request, 'LANGUAGE_CODE', None) or 'en'
def get_description(self, obj):
return get_localized_field(self._lang(), obj.glossary.description)