from flask_socketio import SocketIO, emit, join_room, leave_room from flask import request from app.services import SessionService import threading import time import json from redis import Redis class RedisRepository: """Small helper wrapper to (de)serialize python objects to Redis.""" def __init__(self, redis: Redis): self.redis = redis def set_data(self, key: str, value): self.redis.set(key, json.dumps(value)) def get_data(self, key: str): data = self.redis.get(key) return json.loads(data) if data else None def delete_key(self, key: str): self.redis.delete(key) class SocketController: def __init__( self, socketio: SocketIO, redis: Redis, session_service: SessionService, ): self.socketio = socketio self.session_service = session_service self.redis_repo = RedisRepository(redis) # Menyimpan SID admin untuk setiap session \u2192 {session_code: sid} self.admin_sids: dict[str, str] = {} self._register_events() # --------------------------------------------------------------------- # Helper utilities # --------------------------------------------------------------------- @staticmethod def _is_correct(user_answer, question: dict) -> bool: """Bandingkan jawaban user dengan kunci jawaban.""" print("user answer", user_answer, "question_target", question["target_answer"]) if question["type"] == "fill_the_blank": return ( str(user_answer).strip().lower() == str(question["target_answer"]).strip().lower() ) elif question["type"] == "true_false": return bool(user_answer) == question["target_answer"] elif question["type"] == "option": # user_answer bisa dikirim sebagai index atau value teks option. # Pastikan di‑cast ke int terlebih dahulu jika memungkinkan. try: return int(user_answer) == question["target_answer"] except ValueError: return ( str(user_answer).strip().lower() == str(question["options"][question["target_answer"]]) .strip() .lower() ) return False def _questions_key(self, session_code: str) -> str: return f"session:{session_code}:questions" def _answers_key(self, session_code: str) -> str: return f"session:{session_code}:answers" def _scores_key(self, session_code: str) -> str: return f"session:{session_code}:scores" # --------------------------------------------------------------------- # Socket.IO event bindings # --------------------------------------------------------------------- def _register_events(self): @self.socketio.on("connect") def on_connect(): print(f"Client connected: {request.sid}") emit("connection_response", {"status": "connected", "sid": request.sid}) @self.socketio.on("disconnect") def on_disconnect(): print(f"Client disconnected: {request.sid}") @self.socketio.on("join_room") def handle_join_room(data): session_code = data.get("session_code") user_id = data.get("user_id") if not session_code or not user_id: emit("error", {"message": "session_code and user_id are required"}) return session = self.session_service.join_session( session_code=session_code, user_id=user_id ) if session is None: emit("error", {"message": "Failed to join session or session inactive"}) return join_room(session_code) # Kalau user ini admin, simpan SID‑nya. if session["is_admin"]: self.admin_sids[session_code] = request.sid message = "Admin has joined the room." else: message = f"User {session['username']} has joined the room." print(message) emit( "room_message", { "type": "join", "message": message, "room": session_code, "argument": "adm_update", "data": session if not session["is_admin"] else None, }, room=session_code, ) @self.socketio.on("submit_answer") def handle_submit_answer(data): session_code = data.get("session_id") # front‑end masih mengirim session_id user_id = data.get("user_id") question_index = data.get("question_index") user_answer = data.get("answer") if not all( [ session_code, user_id, question_index is not None, user_answer is not None, ] ): emit( "error", { "message": "session_id, user_id, question_index, and answer are required" }, ) return # ----- 1. Ambil list pertanyaan untuk mendapatkan kunci jawaban ---------- questions = ( self.redis_repo.get_data(self._questions_key(session_code)) or [] ) question = next( (q for q in questions if q["index"] == question_index), None ) if question is None: emit("error", {"message": "Question not found"}) return is_correct = self._is_correct(user_answer, question) print( f"User {user_id} answered Q{question_index} with '{user_answer}' -> {'✔' if is_correct else '✖'}" ) # ----- 2. Simpan jawaban ke Redis -------------------------------------- answers = self.redis_repo.get_data(self._answers_key(session_code)) or [] answers.append( { "user_id": user_id, "question_index": question_index, "answer": user_answer, "correct": is_correct, } ) self.redis_repo.set_data(self._answers_key(session_code), answers) # ----- 3. Update skor per‑user ----------------------------------------- scores = self.redis_repo.get_data(self._scores_key(session_code)) or {} user_score = scores.get(str(user_id), {"correct": 0, "incorrect": 0}) if is_correct: user_score["correct"] += 1 else: user_score["incorrect"] += 1 scores[str(user_id)] = user_score self.redis_repo.set_data(self._scores_key(session_code), scores) # ----- 4. Beri tahu user (ack) ----------------------------------------- emit( "answer_submitted", { "user_id": user_id, "question_index": question_index, "answer": user_answer, "correct": is_correct, }, room=request.sid, ) # ----- 5. Kirim update skor hanya ke admin ----------------------------- admin_sid = self.admin_sids.get(session_code) if admin_sid: emit("score_update", scores, room=admin_sid) @self.socketio.on("leave_room") def handle_leave_room(data): session_code = data.get("session_id") user_id = data.get("user_id") username = data.get("username", "anonymous") leave_room(session_code) emit( "room_message", { "type": "leave", "message": f"{username} has left the room.", "room": session_code, "data": user_id, }, room=session_code, ) @self.socketio.on("send_message") def on_send_message(data): session_code = data.get("session_id") message = data.get("message") username = data.get("username", "anonymous") emit( "receive_message", {"message": message, "from": username}, room=session_code, ) @self.socketio.on("end_session") def handle_end_session(data): session_code = data.get("session_id") user_id = data.get("user_id") if not session_code or not user_id: emit("error", {"message": "session_id and user_id required"}) return # Validasi user berhak mengakhiri session self.session_service.end_session(session_id=session_code, user_id=user_id) answers = self.redis_repo.get_data(self._answers_key(session_code)) or [] scores = self.redis_repo.get_data(self._scores_key(session_code)) or {} print("\n📦 Final Quiz Data for Session", session_code) print("------------------------------------------------------------") for entry in answers: status = "✔" if entry["correct"] else "✖" print( f"User {entry['user_id']} - Q{entry['question_index']}: {entry['answer']} {status}" ) print("\n🏁 Rekap Skor:") for uid, sc in scores.items(): print(f"User {uid}: Benar {sc['correct']} | Salah {sc['incorrect']}") # Bersihkan semua data session di Redis for key in [ self._answers_key(session_code), self._scores_key(session_code), self._questions_key(session_code), ]: self.redis_repo.delete_key(key) emit( "room_closed", {"message": "Session has ended.", "room": session_code}, room=session_code, ) @self.socketio.on("start_quiz") def handle_start_quiz(data): session_code = data.get("session_code") if not session_code: emit("error", {"message": "session_code is required"}) return emit("quiz_started", {"message": "Quiz has started!"}, room=session_code) threading.Thread( target=self._simulate_quiz_flow, args=(session_code,), daemon=True ).start() # --------------------------------------------------------------------- # Quiz flow simulation ------------------------------------------------- # --------------------------------------------------------------------- def _simulate_quiz_flow(self, session_code: str): """Mengirim list pertanyaan satu per satu secara otomatis (demo).""" questions = [ { "index": 1, "question": "Kerajaan Hindu tertua di Indonesia adalah?", "target_answer": "Kutai", "duration": 30, "type": "fill_the_blank", "options": None, }, { "index": 2, "question": "Apakah benar Majapahit mencapai puncak kejayaan pada masa Hayam Wuruk?", "target_answer": True, "duration": 30, "type": "true_false", "options": None, }, { "index": 3, "question": "Kerajaan maritim terbesar di Asia Tenggara pada abad ke‑7 adalah?", "target_answer": 2, "duration": 30, "type": "option", "options": ["Majapahit", "Tarumanegara", "Sriwijaya", "Mataram Kuno"], }, { "index": 4, "question": "Prasasti Yupa merupakan peninggalan dari kerajaan?", "target_answer": "Kutai", "duration": 30, "type": "fill_the_blank", "options": None, }, { "index": 5, "question": "Apakah Tarumanegara terletak di wilayah Kalimantan Timur?", "target_answer": False, "duration": 30, "type": "true_false", "options": None, }, { "index": 6, "question": "Kitab Negarakertagama ditulis oleh?", "target_answer": 0, "duration": 30, "type": "option", "options": [ "Empu Tantular", "Empu Prapanca", "Hayam Wuruk", "Gajah Mada", ], }, { "index": 7, "question": "Tokoh yang terkenal dengan Sumpah Palapa adalah?", "target_answer": "Gajah Mada", "duration": 30, "type": "fill_the_blank", "options": None, }, { "index": 8, "question": "Candi Borobudur dibangun oleh kerajaan Hindu?", "target_answer": False, "duration": 30, "type": "true_false", "options": None, }, { "index": 9, "question": "Raja terkenal dari Kerajaan Sriwijaya adalah?", "target_answer": 1, "duration": 30, "type": "option", "options": [ "Dapunta Hyang", "Balaputradewa", "Airlangga", "Hayam Wuruk", ], }, { "index": 10, "question": "Candi Prambanan merupakan peninggalan agama?", "target_answer": "Hindu", "duration": 30, "type": "fill_the_blank", "options": None, }, ] # Simpan ke Redis agar bisa dipakai saat evaluasi jawaban self.redis_repo.set_data(self._questions_key(session_code), questions) # Beri sedikit jeda sebelum pertanyaan pertama time.sleep(2) for q in questions: print(f"\n📢 Mengirim pertanyaan {q['index']} ke room {session_code}") q.pop("target_answer") self.socketio.emit("quiz_question", q, room=session_code) time.sleep(q["duration"]) self.socketio.emit( "quiz_done", {"message": "Quiz has ended!"}, room=session_code )