diff --git a/classquiz/db/models.py b/classquiz/db/models.py index 57ab111..c0ab082 100644 --- a/classquiz/db/models.py +++ b/classquiz/db/models.py @@ -1,16 +1,26 @@ # SPDX-FileCopyrightText: 2023 Marlon W (Mawoka) # # SPDX-License-Identifier: MPL-2.0 +from __future__ import annotations import os import uuid from datetime import datetime -from typing import Optional +from typing import Optional, Self +from classquiz.config import redis +import json import ormar from ormar import ReferentialAction -from pydantic import BaseModel, Json, field_validator, ConfigDict, RootModel, ValidationInfo +from pydantic import ( + BaseModel, + Json, + field_validator, + ConfigDict, + RootModel, + ValidationInfo, +) from enum import Enum from . import metadata, database from .quiztivity import QuizTivityPage @@ -242,11 +252,34 @@ class PlayGame(BaseModel): custom_field: str | None = None question_show: bool = False + @classmethod + async def get_from_redis(self, game_pin: str) -> Self: + redis_data: str = await redis.get(f"game:{game_pin}") + data = self.model_validate_json(redis_data) + return data + + async def save(self, game_pin: str, ex: int = 7200): + await redis.set(f"game:{game_pin}", self.model_dump_json(), ex=ex) + + def to_player_data(self) -> dict: + return ( + { + **json.loads(self.model_dump_json(exclude={"quiz_id", "questions", "user_id"})), + "question_count": len(self.questions), + }, + ) + class GamePlayer(BaseModel): username: str sid: str | None = None + async def to_player_stack(self, game_pin: str): + await redis.sadd( + f"game_session:{game_pin}:players", + self.model_dump_json(), + ) + class GameAnswer2(BaseModel): username: str @@ -265,6 +298,18 @@ class GameSession(BaseModel): # players: list[GamePlayer | None] answers: list[GameAnswer1 | None] + @classmethod + async def get_from_redis(self, game_pin: str) -> Self: + redis_data = await redis.get(f"game_session:{game_pin}") + return self.model_validate_json(redis_data) + + async def save(self, game_pin: str, ex: int = 7200): + await redis.set( + f"game_session:{game_pin}", + self.model_dump_json(), + ex=7200, + ) + class UpdatePassword(BaseModel): old_password: str @@ -294,6 +339,14 @@ class AnswerDataList(RootModel): def __len__(self) -> int: return len(self.root) + @classmethod + async def get_redis_or_empty(self, game_pin: str, question_number: str) -> Self: + redis_res = await redis.get(f"game_session:{game_pin}:{question_number}") + if redis_res is None: + return self([]) + else: + return self.model_validate_json(redis_res) + class GameInLobby(BaseModel): game_pin: str diff --git a/classquiz/kahoot_importer/import_quiz.py b/classquiz/kahoot_importer/import_quiz.py index 87cf75d..f093da4 100644 --- a/classquiz/kahoot_importer/import_quiz.py +++ b/classquiz/kahoot_importer/import_quiz.py @@ -94,6 +94,8 @@ async def import_quiz(quiz_id: str, user: User) -> Quiz | int: img_obj = await handle_image_upload(quiz.kahoot.cover, user) uploaded_images.append(img_obj) cover = img_obj.id.hex + if quiz.kahoot.description is None or quiz.kahoot.description == "": + quiz.kahoot.description = "Description Missing!" quiz_data = Quiz( id=quiz_id, public=False, diff --git a/classquiz/oauth/github.py b/classquiz/oauth/github.py index c95a618..3d0f53b 100644 --- a/classquiz/oauth/github.py +++ b/classquiz/oauth/github.py @@ -51,7 +51,7 @@ class GitHubOauthResponse(BaseModel): events_url: Optional[str] = None received_events_url: Optional[str] = None type: Optional[str] = None - site_admin: Optional[str] = None + site_admin: Optional[bool] = None name: Optional[str] = None company: Optional[str] = None blog: Optional[str] = None @@ -103,7 +103,8 @@ async def auth(request: Request, response: Response): except authlib.integrations.base_client.OAuthError: return RedirectResponse("/account/oauth-error") resp = await oauth.github.get("user", token=token) - user_data = GitHubOauthResponse(**resp.json()) + data = resp.json() + user_data = GitHubOauthResponse(**data) if user_data.email is None: return RedirectResponse("/account/oauth-error?error=email") user_in_db = await User.objects.get_or_none(email=user_data.email) @@ -140,7 +141,10 @@ async def auth(request: Request, response: Response): else: raise HTTPException(status_code=500, detail=str(e)) user = await User.objects.get_or_none( - email=user_data.email, username=user_data.login, auth_type=UserAuthTypes.GITHUB, verified=True + email=user_data.email, + username=user_data.login, + auth_type=UserAuthTypes.GITHUB, + verified=True, ) await log_user_in(user=user, request=request, response=response) diff --git a/classquiz/oauth/init_oauth.py b/classquiz/oauth/init_oauth.py index 2da4c8f..91c6bcf 100644 --- a/classquiz/oauth/init_oauth.py +++ b/classquiz/oauth/init_oauth.py @@ -4,15 +4,37 @@ from authlib.integrations.starlette_client import OAuth -from classquiz.config import settings +from classquiz.config import settings, redis from functools import lru_cache +import json settings = settings() +class RedisCache: + def __init__(self) -> None: + pass + + async def get(self, key: str) -> str | None: + value = await redis.get(f"authlib:{key}") + if value is None: + return None + try: + return json.loads(value) + except json.JSONDecodeError: + return value + + async def set(self, key: str, value: str, expires: int | None = None): + data = json.dumps(value) + await redis.set(f"authlib:{key}", data, ex=expires) + + async def delete(self, key: str) -> None: + await redis.delete(key) + + @lru_cache() def init_oauth() -> OAuth: - oauth = OAuth() + oauth = OAuth(cache=RedisCache()) if settings.google_client_secret is not None and settings.google_client_id is not None: oauth.register( name="google", diff --git a/classquiz/routers/editor.py b/classquiz/routers/editor.py index cf167d7..5828674 100644 --- a/classquiz/routers/editor.py +++ b/classquiz/routers/editor.py @@ -11,15 +11,27 @@ import asyncpg.exceptions import bleach from fastapi import APIRouter, HTTPException, Depends from pydantic import BaseModel +import html -from classquiz.config import settings, redis, storage, meilisearch, ALLOWED_TAGS_FOR_QUIZ, arq +from classquiz.config import ( + settings, + redis, + storage, + meilisearch, + ALLOWED_TAGS_FOR_QUIZ, + arq, +) from classquiz.db.models import Quiz, QuizInput, User, QuizQuestionType, StorageItem from classquiz.auth import get_current_user import os from datetime import datetime from uuid import UUID -from classquiz.helpers import get_meili_data, check_image_string, extract_image_ids_from_quiz +from classquiz.helpers import ( + get_meili_data, + check_image_string, + extract_image_ids_from_quiz, +) from classquiz.storage.errors import DeletionFailedError settings = settings() @@ -79,15 +91,15 @@ async def finish_edit(edit_id: str, quiz_input: QuizInput): quiz_input.background_color = bleach.clean(quiz_input.background_color, tags=[], strip=True) for i, question in enumerate(quiz_input.questions): - if question.type == QuizQuestionType.ABCD: + if question.type == QuizQuestionType.ABCD or question.type == QuizQuestionType.VOTING: for i2, answer in enumerate(question.answers): if answer.color is not None: quiz_input.questions[i].answers[i2].color = bleach.clean(answer.color, tags=[], strip=True) if answer.answer == "": quiz_input.questions[i].answers[i2].answer = None if answer.answer is not None: - quiz_input.questions[i].answers[i2].answer = bleach.clean( - answer.answer, tags=ALLOWED_TAGS_FOR_QUIZ, strip=True + quiz_input.questions[i].answers[i2].answer = html.unescape( + bleach.clean(answer.answer, tags=ALLOWED_TAGS_FOR_QUIZ, strip=True) ) images_to_delete = [] diff --git a/classquiz/socket_server/__init__.py b/classquiz/socket_server/__init__.py index 755ce80..52875cc 100644 --- a/classquiz/socket_server/__init__.py +++ b/classquiz/socket_server/__init__.py @@ -9,7 +9,6 @@ import json import os import random -import aiohttp import socketio from cryptography.fernet import Fernet @@ -19,14 +18,28 @@ from classquiz.db.models import ( QuizQuestionType, GameSession, GamePlayer, - QuizQuestion, VotingQuizAnswer, AnswerDataList, AnswerData, ) -from pydantic import BaseModel, ValidationError, field_validator, ValidationInfo +from pydantic import BaseModel, ValidationError from datetime import datetime +from classquiz.socket_server.helpers import ( + check_answer, + check_captcha, + has_already_answered, +) +from .models import ( + RejoinGameData, + JoinGameData, + ReturnQuestion, + SubmitAnswerData, + RegisterAsAdminData, + KickPlayerInput, + ConnectSessionIdEvent, +) + from classquiz.socket_server.export_helpers import save_quiz_to_storage from classquiz.socket_server.session import get_session, save_session @@ -74,19 +87,6 @@ async def set_answer(answers, game_pin: str, q_index: int, data: AnswerData) -> return answers -class _JoinGameData(BaseModel): - username: str - game_pin: str - captcha: str | None = None - custom_field: str | None = None - - -class _RejoinGameData(BaseModel): - old_sid: str - game_pin: str - username: str - - @sio.event async def rejoin_game(sid: str, data: dict): redis_res = await redis.get(f"game:{data['game_pin']}") @@ -94,7 +94,7 @@ async def rejoin_game(sid: str, data: dict): await sio.emit("game_not_found", room=sid) return try: - data = _RejoinGameData(**data) + data = RejoinGameData(**data) except ValidationError as e: await sio.emit("error", room=sid) print(e) @@ -106,10 +106,12 @@ async def rejoin_game(sid: str, data: dict): await sio.emit("time_sync", encrypted_datetime, room=sid) await redis.set(redis_sid_key, sid) await redis.srem( - f"game_session:{data.game_pin}:players", GamePlayer(username=data.username, sid=data.old_sid).model_dump_json() + f"game_session:{data.game_pin}:players", + GamePlayer(username=data.username, sid=data.old_sid).model_dump_json(), ) await redis.sadd( - f"game_session:{data.game_pin}:players", GamePlayer(username=data.username, sid=sid).model_dump_json() + f"game_session:{data.game_pin}:players", + GamePlayer(username=data.username, sid=sid).model_dump_json(), ) game_data = PlayGame.model_validate_json(redis_res) session = { @@ -122,10 +124,7 @@ async def rejoin_game(sid: str, data: dict): await sio.enter_room(sid, data.game_pin) await sio.emit( "rejoined_game", - { - **json.loads(game_data.model_dump_json(exclude={"quiz_id", "questions", "user_id"})), - "question_count": len(game_data.questions), - }, + game_data.to_player_data(), room=sid, ) @@ -137,7 +136,7 @@ async def join_game(sid: str, data: dict): await sio.emit("game_not_found", room=sid) return try: - data = _JoinGameData(**data) + data = JoinGameData(**data) except ValidationError as e: await sio.emit("error", room=sid) print(e) @@ -148,42 +147,9 @@ async def join_game(sid: str, data: dict): return # +++ START checking captcha +++ if game_data.captcha_enabled: - async with aiohttp.ClientSession() as session: - try: - if settings.hcaptcha_key is not None: - try: - async with session.post( - "https://hcaptcha.com/siteverify", - data={ - "response": data.captcha, - "secret": settings.hcaptcha_key, - }, - ) as resp: - resp_data = await resp.model_dump_json() - if not resp_data["success"]: - print("CAPTCHA FAILED") - return - except KeyError: - print("CAPTCHA FAILED") - return - elif settings.recaptcha_key is not None: - async with session.post( - "https://www.google.com/recaptcha/api/siteverify", - data={ - "secret": settings.recaptcha_key, - "response": data.captcha, - }, - ) as resp: - try: - resp_data = await resp.model_dump_json() - if not resp_data["success"]: - print("CAPTCHA FAILED") - return - except KeyError: - print("CAPTCHA FAILED") - return - except TypeError: - pass + captcha_res = check_captcha(data.captcha) + if not captcha_res: + return # --- END checking captcha --- if await redis.get(f"game_session:{data.game_pin}:players:{data.username}") is not None: await sio.emit("username_already_exists", room=sid) @@ -198,18 +164,12 @@ async def join_game(sid: str, data: dict): await save_session(sid, sio, session) await sio.emit( "joined_game", - { - **json.loads(game_data.model_dump_json(exclude={"quiz_id", "questions", "user_id"})), - "question_count": len(game_data.questions), - }, + game_data.to_player_data(), room=sid, ) - redis_res = await redis.get(f"game_session:{data.game_pin}") - redis_res = GameSession.model_validate_json(redis_res) await redis.set(f"game_session:{data.game_pin}:players:{data.username}", sid, ex=7200) - await redis.sadd( - f"game_session:{data.game_pin}:players", GamePlayer(username=data.username, sid=sid).model_dump_json() - ) + await GamePlayer(username=data.username, sid=sid).to_player_stack(data.game_pin) + if data.custom_field == "": data.custom_field = None if data.custom_field is not None: @@ -236,49 +196,36 @@ async def start_game(sid: str, _data: dict): session = await get_session(sid, sio) if not session["admin"]: return - game_data = PlayGame.model_validate_json(await redis.get(f"game:{session['game_pin']}")) + game_data = await PlayGame.get_from_redis(session["game_pin"]) game_data.started = True - await redis.set(f"game:{session['game_pin']}", game_data.model_dump_json(), ex=7200) + await game_data.save(session["game_pin"]) await redis.delete(f"game_in_lobby:{game_data.user_id.hex}") await sio.emit("start_game", room=session["game_pin"]) -class _RegisterAsAdminData(BaseModel): - game_pin: str - game_id: str - - @sio.event async def register_as_admin(sid: str, data: dict): try: - data = _RegisterAsAdminData(**data) + data = RegisterAsAdminData(**data) except ValidationError as e: await sio.emit("error", room=sid) print(e) return game_pin = data.game_pin game_id = data.game_id - if (await redis.get(f"game_session:{game_pin}")) is None: - await redis.set( - f"game_session:{game_pin}", - GameSession(admin=sid, game_id=game_id, answers=[]).model_dump_json(), - ex=7200, - ) - - await sio.emit( - "registered_as_admin", - {"game_id": game_id, "game": await redis.get(f"game:{game_pin}")}, - room=sid, - ) - session = {} - session["game_pin"] = game_pin - session["admin"] = True - session["remote"] = False - await save_session(sid, sio, session) - await sio.enter_room(sid, game_pin) - await sio.enter_room(sid, f"admin:{data.game_pin}") - else: + if await redis.get(f"game_session:{game_pin}") is not None: await sio.emit("already_registered_as_admin", room=sid) + return + await GameSession(admin=sid, game_id=game_id, answers=[]).save(game_pin) + await sio.emit( + "registered_as_admin", + {"game_id": game_id, "game": await redis.get(f"game:{game_pin}")}, + room=sid, + ) + session = {"game_pin": game_pin, "admin": True, "remote": False} + await save_session(sid, sio, session) + await sio.enter_room(sid, game_pin) + await sio.enter_room(sid, f"admin:{data.game_pin}") @sio.event @@ -286,57 +233,25 @@ async def get_question_results(sid: str, data: dict): session = await get_session(sid, sio) if not session["admin"]: return - - redis_res = await redis.get(f"game_session:{session['game_pin']}:{data['question_number']}") - if redis_res is None: - redis_res = [] - else: - redis_res = AnswerDataList.model_validate_json(redis_res).model_dump() - game_data = PlayGame.model_validate_json(await redis.get(f"game:{session['game_pin']}")) - game_data.question_show = False - await redis.set(f"game:{session['game_pin']}", game_data.model_dump_json()) game_pin = session["game_pin"] - - await sio.emit("question_results", redis_res, room=game_pin) - - -class ABCDQuizAnswerWithoutSolution(BaseModel): - answer: str - color: str | None = None - - -class RangeQuizAnswerWithoutSolution(BaseModel): - min: int - max: int - - -class ReturnQuestion(QuizQuestion): - answers: list[ABCDQuizAnswerWithoutSolution] | RangeQuizAnswerWithoutSolution | list[VotingQuizAnswer] - type: QuizQuestionType = QuizQuestionType.ABCD - - @field_validator("answers") - def answers_not_none_if_abcd_type(cls, v, info: ValidationInfo): - if info.data["type"] == QuizQuestionType.ABCD and type(v[0]) is not ABCDQuizAnswerWithoutSolution: - raise ValueError("Answers can't be none if type is ABCD") - if info.data["type"] == QuizQuestionType.RANGE and type(v) is not RangeQuizAnswerWithoutSolution: - raise ValueError("Answer must be from type RangeQuizAnswer if type is RANGE") - # skipcq: PTC-W0047 - if info.data["type"] == QuizQuestionType.VOTING and type(v[0]) is not VotingQuizAnswer: - pass - return v + answer_data_list = await AnswerDataList.get_redis_or_empty(game_pin, data["question_number"]) + game_data = await PlayGame.get_from_redis(game_pin) + game_data.question_show = False + await game_data.save(game_pin) + await sio.emit("question_results", answer_data_list.model_dump(), room=game_pin) @sio.event -async def set_question_number(sid, data: str): +async def set_question_number(sid: str, data: str): # data is just a number (as a str) of the question session = await get_session(sid, sio) if not session["admin"]: return game_pin = session["game_pin"] - game_data = PlayGame.model_validate_json(await redis.get(f"game:{session['game_pin']}")) + game_data = await PlayGame.get_from_redis(session["game_pin"]) game_data.current_question = int(float(data)) game_data.question_show = True - await redis.set(f"game:{session['game_pin']}", game_data.model_dump_json(), ex=7200) + await game_data.save(session["game_pin"]) await redis.set(f"game:{session['game_pin']}:current_time", datetime.now().isoformat(), ex=7200) temp_return = game_data.model_dump(include={"questions"})["questions"][int(float(data))] if game_data.questions[int(float(data))].type == QuizQuestionType.SLIDE: @@ -364,93 +279,39 @@ async def set_question_number(sid, data: str): ) -class _SubmitAnswerDataOrderType(BaseModel): - answer: str - - -class _SubmitAnswerData(BaseModel): - question_index: int - answer: str | int - complex_answer: list[_SubmitAnswerDataOrderType] | None = None - - @sio.event async def submit_answer(sid: str, data: dict): now = datetime.now() try: - data = _SubmitAnswerData(**data) + data = SubmitAnswerData(**data) except ValidationError as e: await sio.emit("error", room=sid) print(e) return data.answer = str(data.answer) session = await get_session(sid, sio) - game_data = PlayGame.model_validate_json(await redis.get(f"game:{session['game_pin']}")) - answer_right = False - if game_data.questions[int(float(data.question_index))].type == QuizQuestionType.ABCD: - for answer in game_data.questions[int(float(data.question_index))].answers: - if answer.answer == data.answer and answer.right: - answer_right = True - break - elif game_data.questions[int(float(data.question_index))].type == QuizQuestionType.RANGE: - if ( - game_data.questions[int(float(data.question_index))].answers.min_correct - <= int(float(data.answer)) - <= game_data.questions[int(float(data.question_index))].answers.max_correct - ): - answer_right = True - elif game_data.questions[int(float(data.question_index))].type == QuizQuestionType.VOTING: - answer_right = False - elif game_data.questions[int(float(data.question_index))].type == QuizQuestionType.ORDER: - if data.complex_answer is None: - answer_right = False - else: - question = game_data.questions[int(float(data.question_index))] - correct_answers = [] - for a in question.answers: - correct_answers.append({"answer": a.answer}) - answer_order = [] - for a in data.model_dump()["complex_answer"]: - answer_order.append(a["answer"]) - data.answer = ", ".join(answer_order) - if correct_answers == data.model_dump()["complex_answer"]: - answer_right = True - - elif game_data.questions[int(float(data.question_index))].type == QuizQuestionType.TEXT: - answer_right = False - for q in game_data.questions[int(float(data.question_index))].answers: - if q.case_sensitive: - if data.answer == q.answer: - answer_right = True - break - else: - if data.answer.lower() == q.answer.lower(): - answer_right = True - break - elif game_data.questions[int(data.question_index)].type == QuizQuestionType.CHECK: - correct_string = "" - for i, a in enumerate(game_data.questions[int(float(data.question_index))].answers): - if a.right: - correct_string += str(i) - answer_right = bool(correct_string == data.answer) - else: - raise NotImplementedError - latency = int(float((await get_session(sid, sio))["ping"])) + question_index = int(float(data.question_index)) + game_data = await PlayGame.get_from_redis(session["game_pin"]) + already_answered = await has_already_answered(session["game_pin"], question_index, session["username"]) + if already_answered: + await sio.emit("already_replied", room=sid) + return + answer_right, answer = check_answer(game_data, data) + latency = int(float(session["ping"])) time_q_started = datetime.fromisoformat(await redis.get(f"game:{session['game_pin']}:current_time")) - diff = (time_q_started - now).total_seconds() * 1000 # - timedelta(milliseconds=latency) score = 0 if answer_right: score = calculate_score( abs(diff) - latency, - int(float(game_data.questions[int(float(data.question_index))].time)), + int(float(game_data.questions[question_index].time)), ) if score > 1000: score = 1000 await redis.hincrby(f"game_session:{session['game_pin']}:player_scores", session["username"], score) answer_data = AnswerData( username=session["username"], - answer=data.answer, + answer=answer, right=answer_right, time_taken=abs(diff) - latency, score=score, @@ -465,26 +326,18 @@ async def submit_answer(sid: str, data: dict): player_count = await redis.scard(f"game_session:{session['game_pin']}:players") await sio.emit("player_answer", {}) if len(answers) == player_count: - # await sio.emit( - # "question_results", - # await redis.get(f"game_session:{session['game_pin']}:{data.question_index}"), - # room=session["game_pin"], - # ) - game_data = PlayGame.model_validate_json(await redis.get(f"game:{session['game_pin']}")) + game_data = await PlayGame.get_from_redis(session["game_pin"]) game_data.question_show = False - await redis.set(f"game:{session['game_pin']}", game_data.model_dump_json()) + await game_data.save(session["game_pin"]) await sio.emit("everyone_answered", {}) -# await redis.set(f"game_data:{session['game_pin']}", json.dumps(data)) - - @sio.event async def get_final_results(sid: str, _data: dict): session: dict = await get_session(sid, sio) - game_data = PlayGame(**json.loads(await redis.get(f"game:{session['game_pin']}"))) if not session["admin"]: return + game_data = await PlayGame.get_from_redis(session["game_pin"]) results = await generate_final_results(game_data, session["game_pin"]) await sio.emit("final_results", results, room=session["game_pin"]) @@ -494,7 +347,7 @@ async def get_export_token(sid: str): session = await get_session(sid, sio) if not session["admin"]: return - game_data = PlayGame(**json.loads(await redis.get(f"game:{session['game_pin']}"))) + game_data = await PlayGame.get_from_redis(session["game_pin"]) results = await generate_final_results(game_data, session["game_pin"]) token = os.urandom(32).hex() await redis.set(f"export_token:{token}", json.dumps(results), ex=7200) @@ -504,10 +357,14 @@ async def get_export_token(sid: str): @sio.event async def show_solutions(sid: str, _data: dict): session: dict = await get_session(sid, sio) - game_data = PlayGame(**json.loads(await redis.get(f"game:{session['game_pin']}"))) + game_data = await PlayGame.get_from_redis(session["game_pin"]) if not session["admin"]: return - await sio.emit("solutions", game_data.questions[game_data.current_question].model_dump(), room=session["game_pin"]) + await sio.emit( + "solutions", + game_data.questions[game_data.current_question].model_dump(), + room=session["game_pin"], + ) @sio.event @@ -521,14 +378,10 @@ async def echo_time_sync(sid: str, data: str): await save_session(sid, sio, session) -class _KickPlayerInput(BaseModel): - username: str - - @sio.event async def kick_player(sid: str, data: dict): try: - data = _KickPlayerInput(**data) + data = KickPlayerInput(**data) except ValidationError as e: await sio.emit("error", room=sid) print(e) @@ -604,10 +457,6 @@ async def save_quiz(sid: str): await sio.emit("results_saved_successfully") -class ConnectSessionIdEvent(BaseModel): - session_id: str - - @sio.event async def connect(sid: str, _environ, _auth): session_id = os.urandom(16).hex() diff --git a/classquiz/socket_server/helpers.py b/classquiz/socket_server/helpers.py new file mode 100644 index 0000000..d4a2ed9 --- /dev/null +++ b/classquiz/socket_server/helpers.py @@ -0,0 +1,139 @@ +# SPDX-FileCopyrightText: 2025 Marlon W (Mawoka) +# +# SPDX-License-Identifier: MPL-2.0 +import aiohttp +from classquiz.config import settings +from classquiz.db.models import ( + PlayGame, + QuizQuestionType, + TextQuizAnswer, + ABCDQuizAnswer, + VotingQuizAnswer, + RangeQuizAnswer, + AnswerDataList, +) +from classquiz.socket_server.models import SubmitAnswerData +from .models import SubmitAnswerDataOrderType + + +async def check_captcha(captcha_data: str) -> bool: + async with aiohttp.ClientSession() as session: + try: + if settings.hcaptcha_key is not None: + try: + async with session.post( + "https://hcaptcha.com/siteverify", + data={ + "response": captcha_data, + "secret": settings.hcaptcha_key, + }, + ) as resp: + resp_data = await resp.model_dump_json() + if not resp_data["success"]: + return + except KeyError: + return False + elif settings.recaptcha_key is not None: + async with session.post( + "https://www.google.com/recaptcha/api/siteverify", + data={ + "secret": settings.recaptcha_key, + "response": captcha_data, + }, + ) as resp: + try: + resp_data = await resp.model_dump_json() + if not resp_data["success"]: + return False + except KeyError: + return False + except TypeError: + pass + return False + return True + + +def check_answer(game_data: PlayGame, data: SubmitAnswerData) -> (bool, str): + q_i = int(float(data.question_index)) + q_type = game_data.questions[q_i].type + q_answers = game_data.questions[q_i].answers + q_answer = data.answer + if q_type == QuizQuestionType.ABCD: + return (check_abcd_question(q_answer, q_answers), data.answer) + elif q_type == QuizQuestionType.RANGE: + return ( + check_range_question(q_answer, q_answers), + q_answer, + ) + elif q_type == QuizQuestionType.VOTING: + return (False, q_answer) + elif q_type == QuizQuestionType.ORDER: + return check_order_question(data.complex_answer, q_answer, q_answers) + elif q_type == QuizQuestionType.TEXT: + return ( + check_text_question(q_answer, q_answers), + q_answer, + ) + + elif q_type == QuizQuestionType.CHECK: + return ( + check_check_question(q_answer, q_answers), + q_answer, + ) + else: + return (False, q_answer) + return (False, q_answer) + + +def check_abcd_question(answer: str, answers: ABCDQuizAnswer) -> bool: + for a in answers: + if a.answer == answer and a.right: + return True + return False + + +def check_range_question(answer: str, answers: RangeQuizAnswer) -> bool: + if answers.min_correct <= int(float(answer)) <= answers.max_correct: + return answers.min_correct <= int(float(answer)) <= answers.max_correct + + +def check_order_question( + complex_answer: list[SubmitAnswerDataOrderType] | None, + answer: str, + answers: list[VotingQuizAnswer], +) -> (bool, str): + if complex_answer is None: + return (False, answer) + correct_answers = [{"answer": a.answer} for a in answers] + submitted_answers = [{"answer": a.answer} for a in complex_answer] + answer_str = ", ".join(a["answer"] for a in submitted_answers) + is_correct = submitted_answers == correct_answers + return is_correct, answer_str + + +def check_text_question(answer: str, answers: list[TextQuizAnswer]) -> bool: + for q in answers: + if q.case_sensitive: + if answer == q.answer: + return True + else: + if answer.lower() == q.answer.lower(): + return True + return False + + +def check_check_question(answer: str, answers: list[ABCDQuizAnswer]) -> bool: + correct_string = "" + for i, a in enumerate(answers): + if a.right: + correct_string += str(i) + return bool(correct_string == answer) + + +async def has_already_answered(game_pin: str, q_index: int, username: str) -> bool: + answers = await AnswerDataList.get_redis_or_empty(game_pin, q_index) + if answers is None: + return False + else: + answers = list(filter(lambda a: a.username == username, answers.root)) + return len(answers) > 0 diff --git a/classquiz/socket_server/models.py b/classquiz/socket_server/models.py new file mode 100644 index 0000000..4bae14a --- /dev/null +++ b/classquiz/socket_server/models.py @@ -0,0 +1,68 @@ +# SPDX-FileCopyrightText: 2025 Marlon W (Mawoka) +# +# SPDX-License-Identifier: MPL-2.0 + +from pydantic import BaseModel, field_validator, ValidationInfo +from classquiz.db.models import QuizQuestion, QuizQuestionType, VotingQuizAnswer + + +class JoinGameData(BaseModel): + username: str + game_pin: str + captcha: str | None = None + custom_field: str | None = None + + +class RejoinGameData(BaseModel): + old_sid: str + game_pin: str + username: str + + +class RegisterAsAdminData(BaseModel): + game_pin: str + game_id: str + + +class ABCDQuizAnswerWithoutSolution(BaseModel): + answer: str + color: str | None = None + + +class RangeQuizAnswerWithoutSolution(BaseModel): + min: int + max: int + + +class ReturnQuestion(QuizQuestion): + answers: list[ABCDQuizAnswerWithoutSolution] | RangeQuizAnswerWithoutSolution | list[VotingQuizAnswer] + type: QuizQuestionType = QuizQuestionType.ABCD + + @field_validator("answers") + def answers_not_none_if_abcd_type(cls, v, info: ValidationInfo): + if info.data["type"] == QuizQuestionType.ABCD and type(v[0]) is not ABCDQuizAnswerWithoutSolution: + raise ValueError("Answers can't be none if type is ABCD") + if info.data["type"] == QuizQuestionType.RANGE and type(v) is not RangeQuizAnswerWithoutSolution: + raise ValueError("Answer must be from type RangeQuizAnswer if type is RANGE") + # skipcq: PTC-W0047 + if info.data["type"] == QuizQuestionType.VOTING and type(v[0]) is not VotingQuizAnswer: + pass + return v + + +class SubmitAnswerDataOrderType(BaseModel): + answer: str + + +class SubmitAnswerData(BaseModel): + question_index: int + answer: str | int + complex_answer: list[SubmitAnswerDataOrderType] | None = None + + +class KickPlayerInput(BaseModel): + username: str + + +class ConnectSessionIdEvent(BaseModel): + session_id: str diff --git a/frontend/src/lib/components/buttons/brown.svelte b/frontend/src/lib/components/buttons/brown.svelte index b9f4962..7a2846a 100644 --- a/frontend/src/lib/components/buttons/brown.svelte +++ b/frontend/src/lib/components/buttons/brown.svelte @@ -5,19 +5,14 @@ SPDX-License-Identifier: MPL-2.0 --> @@ -39,7 +35,7 @@ SPDX-License-Identifier: MPL-2.0 class:cursor-not-allowed={disabled} class:pointer-events-none={disabled} class="text-black hover:bg-bg-[#B07156]/80 w-full px-4 py-2 leading-5 transition-all duration-200 transform bg-[#B07156] rounded-sm text-center outline-hidden hover:cursor-pointer" - onclick={bubble('click')} + {onclick} class:flex class:justify-center={flex} > @@ -50,7 +46,7 @@ SPDX-License-Identifier: MPL-2.0 {disabled} {type} class="text-black hover:cursor-pointer hover:opacity-80 w-full px-4 py-2 leading-5 transition-all duration-200 transform bg-[#B07156] rounded-sm text-center focus:outline-hidden disabled:cursor-not-allowed disabled:opacity-50 outline-hidden" - onclick={bubble('click')} + {onclick} class:flex class:justify-center={flex} > diff --git a/frontend/src/lib/components/buttons/gray.svelte b/frontend/src/lib/components/buttons/gray.svelte index 65705a6..3d94a99 100644 --- a/frontend/src/lib/components/buttons/gray.svelte +++ b/frontend/src/lib/components/buttons/gray.svelte @@ -5,16 +5,13 @@ SPDX-License-Identifier: MPL-2.0 --> @@ -31,7 +29,7 @@ SPDX-License-Identifier: MPL-2.0 {href} {target} class="w-full px-4 py-2 leading-5 text-black dark:text-white transition-colors duration-200 transform bg-gray-50 dark:bg-gray-700 rounded-sm text-center hover:bg-gray-300 focus:outline-hidden disabled:cursor-not-allowed disabled:opacity-50 dark:hover:bg-gray-600" - onclick={bubble('click')} + {onclick} class:flex class:block={!flex} class:justify-center={flex} @@ -42,7 +40,7 @@ SPDX-License-Identifier: MPL-2.0