from typing import List from repositories import QuizRepository, UserRepository from schemas import QuizGetSchema from schemas.requests import QuizCreateSchema from schemas.response import UserQuizListResponse, RecomendationResponse from exception import DataNotFoundException from mapper import map_quiz_entity_to_schema from exception import ValidationException class QuizService: def __init__(self, quiz_repository=QuizRepository, user_repository=UserRepository): self.quiz_repository = quiz_repository self.user_repostory = user_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") return map_quiz_entity_to_schema(data) def search_quiz( self, keyword: str, page: int = 1, page_size: int = 10 ) -> tuple[list[RecomendationResponse], int]: if not keyword: raise ValidationException("Keyword cannot be empty.") quizzes = self.quiz_repository.search_by_title_or_category( keyword, page, page_size ) total = self.quiz_repository.count_by_search(keyword) print("quiz len", len(quizzes)) mapped_quizzes = [] for quiz in quizzes: author = self.user_repostory.get_user_by_id(user_id=quiz.author_id) if author is None: print(quiz.author_id, "skipped") continue # or handle default name mapped_quizzes.append( RecomendationResponse( quiz_id=str(quiz.id), author_id=str(author.id), author_name=author.name, title=quiz.title, description=quiz.description, date=quiz.date.strftime("%d-%B-%Y"), ) ) 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) total_user_quiz = self.quiz_repository.count_by_user_id(user_id) return UserQuizListResponse( total=total_user_quiz, quizzes=[map_quiz_entity_to_schema(quiz) for quiz in quizzes], ) def create_quiz(self, quiz_data: QuizCreateSchema): 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." ) return self.quiz_repository.create(quiz_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_recommendation(self, page: int = 1, limit: int = 3): page = page or 1 limit = limit or 3 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( RecomendationResponse( quiz_id=str(quiz.id), author_id=str(author.id), author_name=author.name, title=quiz.title, description=quiz.description, date=quiz.date.strftime("%d-%B-%Y"), ) ) return result