from typing import Optional from uuid import uuid4 from app.repositories import ( SessionRepository, UserRepository, SessionMemoryRepository, QuizRepository, UserAnswerRepository, ) from app.models.entities import SessionEntity from app.helpers import DatetimeUtil from flask_socketio import SocketIO import time class SessionService: def __init__( self, repository_mongo: SessionRepository, repository_redis: SessionMemoryRepository, user_repository: UserRepository, quiz_repository: QuizRepository, answer_repository: UserAnswerRepository, ): self.repository_mongo = repository_mongo self.repository_redis = repository_redis self.user_repository = user_repository self.quiz_repository = quiz_repository self.answer_repository = answer_repository def create_session(self, quiz_id: str, host_id: str, limit_participan: int) -> str: generateed_code = uuid4().hex[:6].upper() session = SessionEntity( session_code=generateed_code, quiz_id=quiz_id, host_id=host_id, created_at=DatetimeUtil.now_iso(), limit_participan=limit_participan, participants=[], current_question_index=0, is_active=True, ) session_id = self.repository_mongo.insert(session) session.id = session_id self.repository_redis.create_session(session_id, session) data = self.quiz_repository.get_by_id(quiz_id=quiz_id) self.repository_redis.set_quiz_for_session(session_id, data) return { "session_id": session_id, "session_code": generateed_code, } def join_session(self, session_code: str, user_id: str) -> dict: user = self.user_repository.get_user_by_id(user_id) session = self.repository_redis.find_session_by_code(session_code) if session is None: return None session_id = session["id"] is_existing_user = any( u["id"] == user_id for u in session.get("participants", []) ) session_quiz = self.repository_redis.get_quiz_for_session(session["id"]) quiz_info = { "title": session_quiz.title, "description": session_quiz.description, "total_quiz": session_quiz.total_quiz, "limit_duration": session_quiz.limit_duration, } if session["host_id"] == user_id: return { "session_id": session_id, "is_admin": True, "message": "admin joined", "session_info": session, "quiz_info": quiz_info, } if is_existing_user: return { "session_id": session_id, "is_admin": False, "user_id": str(user.id), "username": user.name, "user_pic": user.pic_url, "session_info": session, "quiz_info": quiz_info, "new_user": not is_existing_user, } self.repository_redis.add_user_to_session( session_id=session["id"], user_data={ "id": str(user.id), "username": user.name, "user_pic": user.pic_url, }, ) session = self.repository_redis.get_session(session["id"]) response = { "session_id": session_id, "is_admin": False, "user_id": str(user.id), "username": user.name, "user_pic": user.pic_url, "session_info": session if not is_existing_user else None, "quiz_info": quiz_info, "new_user": not is_existing_user, } return response def leave_session(self, session_id: str, user_id: str) -> dict: is_success = self.repository_redis.remove_user_from_session(session_id, user_id) if is_success: participant_left = self.repository_redis.get_user_in_session(session_id) return {"is_success": True, "participants": participant_left} return {"is_success": False} def start_session(self, session_id: str) -> bool: now = DatetimeUtil.now_iso() return self.repository.update(session_id, {"started_at": now}) def end_session(self, session_id: str, user_id: str): session = self.repository.find_by_session_id(session_id) if session and session.host_id == user_id: session.is_active = False self.repository.update(session_id, {"is_active": False}) def advance_question(self, session_id: str) -> Optional[int]: session = self.repository.find_by_session_id(session_id) if not session: return None current_index = session.get("current_question_index", 0) total = session.get("total_questions", 0) if current_index + 1 >= total: self.end_session(session_id) return None new_index = current_index + 1 self.repository.update(session_id, {"current_question_index": new_index}) return new_index def get_session(self, session_id: str) -> Optional[SessionEntity]: session = self.repository.find_by_session_id(session_id) return SessionEntity(**session) if session else None def simulate_quiz_flow(self, session_id: str, socketio: SocketIO): quiz = self.repository_redis.get_quiz_for_session(session_id) questions = quiz.question_listings time.sleep(2) for q in questions: print(f"\nMengirim pertanyaan {q.index} ke room {session_id}") question_to_send = q.model_dump(exclude={"target_answer"}) socketio.emit("quiz_question", question_to_send, room=session_id) time.sleep(q.duration) def generate_quiz_recap(self, session_id: str, socketio: SocketIO): try: answers = self.repository_redis.get_all_answers(session_id) scores = self.repository_redis.get_scores(session_id) quiz = self.repository_redis.get_quiz_for_session(session_id) questions = quiz.question_listings recap_data = { "session_id": session_id, "total_questions": len(questions), "answers": [], "scores": [], "questions": [], } recap_data["questions"] = [ { "index": q["index"], "question": q["question"], "type": q["type"], } for q in questions ] for entry in answers: related_question = next( (q for q in questions if q["index"] == entry["question_index"]), None, ) recap_data["answers"].append( { "user_id": entry["user_id"], "question_index": entry["question_index"], "answer": entry["answer"], "correct": entry["correct"], "question": ( related_question["question"] if related_question else "Pertanyaan tidak ditemukan" ), } ) for uid, sc in scores.items(): recap_data["scores"].append( { "user_id": uid, "correct": sc.get("correct", 0), "incorrect": sc.get("incorrect", 0), "total_score": sc.get("correct", 0) * 10, } ) # Urutkan skor tertinggi ke terendah recap_data["scores"].sort(key=lambda x: x["total_score"], reverse=True) # Emit recap data ke semua peserta socketio.emit( "quiz_recap", { "message": "Kuis telah selesai. Berikut adalah rekap lengkap.", "recap": recap_data, }, room=session_id, ) # Emit informasi bahwa kuis telah berakhir socketio.emit( "quiz_done", {"message": "Kuis telah berakhir.", "session_id": session_id}, room=session_id, ) except Exception as e: # Tangani error dan informasikan ke peserta socketio.emit( "quiz_error", { "message": "Terjadi kesalahan saat membuat rekap kuis.", "error": str(e), }, room=session_id, )