TIF_E41211115_Genso_quiz_ba.../app/services/quiz_service.py

153 lines
4.9 KiB
Python

from app.repositories import (
QuizRepository,
UserRepository,
SubjectRepository,
UserAnswerRepository,
)
from app.schemas.requests import QuizCreateSchema
from app.schemas.response import (
UserQuizListResponse,
QuizGetSchema,
ListingQuizResponse,
)
from app.exception import DataNotFoundException, ValidationException
from app.mapper import QuizMapper
from app.helpers import DatetimeUtil
from flask import current_app
class QuizService:
def __init__(
self,
quiz_repository: QuizRepository,
user_repository: UserRepository,
subject_repository: SubjectRepository,
answer_repository: UserAnswerRepository,
):
self.quiz_repository = quiz_repository
self.user_repostory = user_repository
self.subject_repository = subject_repository
self.answer_repository = answer_repository
def get_quiz(self, quiz_id) -> QuizGetSchema:
data = self.quiz_repository.get_by_id(quiz_id)
if data is None:
raise DataNotFoundException("Quiz not found")
quiz_subject = self.subject_repository.get_by_id(data.subject_id)
return QuizMapper.map_quiz_entity_to_schema(data, quiz_subject)
def search_quiz(
self, keyword: str, subject_id: str, page: int = 1, page_size: int = 10
) -> tuple[list[ListingQuizResponse], int]:
quizzes = self.quiz_repository.search_by_title_or_category(
keyword=keyword,
page=page,
page_size=page_size,
subject_id=subject_id,
)
total = self.quiz_repository.count_by_search(keyword)
mapped_quizzes = []
for quiz in quizzes:
author = self.user_repostory.get_user_by_id(user_id=quiz.author_id)
if author is None:
continue
mapped_quizzes.append(
QuizMapper.quiz_to_populer_mapper(
quiz_entity=quiz,
user_entity=author,
)
)
return mapped_quizzes, total
def get_user_quiz(
self, user_id: str, page: int = 1, page_size: int = 10
) -> UserQuizListResponse:
quizzes = self.quiz_repository.get_by_user_id(user_id, page, page_size)
if not quizzes:
return UserQuizListResponse(total=0, quizzes=[])
total_user_quiz = self.quiz_repository.count_by_user_id(user_id)
user = self.user_repostory.get_user_by_id(user_id)
quiz_data = [QuizMapper.quiz_to_populer_mapper(quiz, user) for quiz in quizzes]
return UserQuizListResponse(total=total_user_quiz, quizzes=quiz_data)
def create_quiz(self, quiz_data: QuizCreateSchema):
total_time = 0
for question in quiz_data.question_listings:
if question.type == "option" and (
not question.options or len(question.options) != 4
):
raise ValidationException(
"Option type questions must have exactly 4 options."
)
total_time += question.duration
datetime_now = DatetimeUtil.now_iso()
data = QuizMapper.map_quiz_schema_to_entity(
schema=quiz_data,
datetime=datetime_now,
total_duration=total_time,
)
return self.quiz_repository.create(data)
def update_quiz(self, quiz_id, quiz_data):
return self.quiz_repository.update(quiz_id, quiz_data)
def delete_quiz(self, quiz_id):
return self.quiz_repository.delete(quiz_id)
def get_quiz_populer(self, page: int, limit: int):
data = self.quiz_repository.get_top_played_quizzes(page=page, limit=limit)
if not data:
raise DataNotFoundException("Quiz not found")
result = []
for quiz in data:
author = self.user_repostory.get_user_by_id(user_id=quiz.author_id)
result.append(
QuizMapper.quiz_to_populer_mapper(
quiz_entity=quiz,
user_entity=author,
)
)
return result
def get_quiz_recommendation(self, user_id: str, page: int, limit: int):
user_answer = self.answer_repository.get_by_user(user_id=user_id)
if not user_answer:
return []
quiz_ids = list({answer.quiz_id for answer in user_answer})
quiz_work = self.quiz_repository.get_by_ids(quiz_ids)
if not quiz_work:
return []
quiz_subjects = list({quiz.subject_id for quiz in quiz_work})
quiz_data = self.quiz_repository.get_random_quizzes_by_subjects(
subject_ids=quiz_subjects, limit=limit
)
result = []
for quiz in quiz_data:
author = self.user_repostory.get_user_by_id(user_id=quiz.author_id)
result.append(
QuizMapper.quiz_to_populer_mapper(
quiz_entity=quiz,
user_entity=author,
)
)
return result