# This Source Code Form is subject to the terms of the Mozilla Public # License, v. 2.0. If a copy of the MPL was not distributed with this # file, You can obtain one at https://mozilla.org/MPL/2.0/. import json import random import re import uuid from datetime import datetime from random import randint import ormar.exceptions from classquiz.helpers import generate_spreadsheet from fastapi import APIRouter, Depends, HTTPException from fastapi.responses import JSONResponse, StreamingResponse from pydantic import ValidationError, BaseModel from classquiz.auth import get_current_user from classquiz.config import redis, settings, storage, meilisearch from classquiz.db.models import Quiz, User, PlayGame, GameInLobby, QuizQuestion from classquiz.kahoot_importer.import_quiz import import_quiz import urllib.parse settings = settings() router = APIRouter() @router.get("/get/{quiz_id}") async def get_quiz_from_id(quiz_id: str, user: User | None = Depends(get_current_user)): try: quiz_id = uuid.UUID(quiz_id) except ValueError: raise HTTPException(status_code=400, detail="badly formed quiz id") if user is None: quiz = await Quiz.objects.get_or_none(id=quiz_id, public=True) else: quiz = await Quiz.objects.get_or_none(id=quiz_id, user_id=user.id) if quiz is None: public_quiz = await Quiz.objects.get_or_none(id=quiz_id, public=True) if public_quiz is None: return JSONResponse(status_code=404, content={"detail": "quiz not found"}) else: return public_quiz else: return quiz class PublicQuizResponseUser(BaseModel): username: str id: uuid.UUID class PublicQuizResponse(Quiz.get_pydantic()): user_id: PublicQuizResponseUser questions: list[QuizQuestion] @router.get("/get/public/{quiz_id}") async def get_public_quiz(quiz_id: str): try: quiz_id = uuid.UUID(quiz_id) except ValueError: raise HTTPException(status_code=400, detail="badly formed quiz id") quiz = await Quiz.objects.select_related("user_id").get_or_none(id=quiz_id) if quiz is None: return JSONResponse(status_code=404, content={"detail": "quiz not found"}) else: return PublicQuizResponse(**quiz.dict()) def generate_code() -> str: specified_length = 6 buttons = [ "B", "b", "G", "g", "Y", "y", "R", "r", ] # Capital stands for long press, lowercase letter for short press resulting_code = "" for _ in range(specified_length): resulting_code += random.choice(buttons) return resulting_code @router.post("/start/{quiz_id}") async def start_quiz( quiz_id: str, game_mode: str, captcha_enabled: bool = True, custom_field: str | None = None, cqcs_enabled: bool = False, user: User = Depends(get_current_user), ): try: quiz_id = uuid.UUID(quiz_id) except ValueError: raise HTTPException(status_code=400, detail="badly formed quiz id") quiz = await Quiz.objects.get_or_none(id=quiz_id, user_id=user.id) if quiz is None: quiz = await Quiz.objects.get_or_none(id=quiz_id, public=True) if quiz is None: return JSONResponse(status_code=404, content={"detail": "quiz not found"}) game_pin = randint(100000, 999999) if custom_field == "": custom_field = None game = await redis.get(f"game:{game_pin}") while game is not None: game_pin = randint(100000, 999999) game = await redis.get(f"game:{game_pin}") game = PlayGame( quiz_id=quiz_id, game_pin=str(game_pin), questions=quiz.questions, game_id=uuid.uuid4(), title=quiz.title, description=quiz.description, captcha_enabled=captcha_enabled, cover_image=quiz.cover_image, game_mode=game_mode, user_id=user.id, background_color=quiz.background_color, custom_field=custom_field, background_image=quiz.background_image, ) code = None if cqcs_enabled: code = generate_code() await redis.set(f"game:cqc:code:{code}", game_pin, ex=3600) await redis.set(f"game:{str(game.game_pin)}", game.json(), ex=18000) await redis.set(f"game_pin:{user.id}:{quiz_id}", game_pin, ex=18000) await redis.set( f"game_in_lobby:{user.id.hex}", GameInLobby(game_id=game.game_id, game_pin=str(game_pin), quiz_title=quiz.title).json(), ex=900, ) return {**quiz.dict(exclude={"id"}), **game.dict(exclude={"questions"}), "cqc_code": code} class CheckIfCaptchaEnabledResponse(BaseModel): enabled: bool game_mode: str | None custom_field: str | None @router.get("/play/check_captcha/{game_pin}", response_model=CheckIfCaptchaEnabledResponse) async def check_if_captcha_enabled(game_pin: str): game = await redis.get(f"game:{game_pin}") if game is None: return JSONResponse(status_code=404, content={"detail": "game not found"}) game = PlayGame.parse_raw(game) if game.captcha_enabled: return CheckIfCaptchaEnabledResponse(enabled=True, game_mode=game.game_mode, custom_field=game.custom_field) else: return CheckIfCaptchaEnabledResponse(enabled=False, game_mode=game.game_mode, custom_field=game.custom_field) @router.get("/join/{game_pin}", deprecated=True) async def get_game_id(game_pin: str): redis_res = await redis.get(f"game:{game_pin}") if redis_res is None: raise HTTPException(status_code=404, detail="game not found") else: return json.loads(redis_res)["game_id"] @router.get("/list") async def get_quiz_list(user: User = Depends(get_current_user), page_size: int | None = 10, page: int | None = 1): try: return ( await Quiz.objects.order_by(Quiz.updated_at.desc()) .filter(user_id=user.id) .paginate(page, page_size=page_size) .all() ) except ormar.exceptions.QueryDefinitionError: raise HTTPException(status_code=400, detail="Invalid page(size). page(size) have to be greater than 0.") @router.post("/import/{quiz_id}") async def import_quiz_route(quiz_id: str, user: User = Depends(get_current_user)): if user.storage_used > settings.free_storage_limit: raise HTTPException(status_code=409, detail="Storage limit reached") try: return await import_quiz(quiz_id, user) except ValidationError as e: print(e) raise HTTPException(status_code=400, detail="This quiz isn't (yet) supported") @router.delete("/delete/{quiz_id}") async def delete_quiz(quiz_id: str, user: User = Depends(get_current_user)): try: quiz_id = uuid.UUID(quiz_id) except ValueError: raise HTTPException(status_code=400, detail="badly formed quiz id") quiz = await Quiz.objects.get_or_none(id=quiz_id, user_id=user.id) if quiz is None: return JSONResponse(status_code=404, content={"detail": "quiz not found"}) pics_to_delete = [] pic_name_regex = re.compile("^.*/(.{36}--.{36})$") for question in quiz.questions: try: if question["image"] is not None and not str(question["image"]).startswith("https://i.imgur.com/"): old_image_to_delete = pic_name_regex.match(question["image"]) if old_image_to_delete is not None: pics_to_delete.append(old_image_to_delete.group(1)) except KeyError: pass if len(pics_to_delete) != 0: await storage.delete(pics_to_delete) meilisearch.index(settings.meilisearch_index).delete_document(str(quiz.id)) return await quiz.delete() @router.get("/export_data/{export_token}", response_class=StreamingResponse) async def export_quiz_answers(export_token: str, game_pin: str): data = await redis.get(f"export_token:{export_token}") if data is None: raise HTTPException(status_code=404, detail="export token not found") data = json.loads(data) data2 = await redis.get(f"game:{game_pin}") game_data = PlayGame.parse_raw(data2) quiz = await Quiz.objects.get_or_none(id=game_data.quiz_id) if quiz is None: raise HTTPException(status_code=404, detail="quiz not found") player_fields = await redis.hgetall(f"game:{game_pin}:players:custom_fields") score_data = await redis.hgetall(f"game_session:{game_pin}:player_scores") spreadsheet = await generate_spreadsheet( quiz=quiz, quiz_results=data, player_fields=player_fields, player_scores=score_data ) def iter_file(): yield from spreadsheet await redis.delete(f"export_token:{export_token}") return StreamingResponse( iter_file(), media_type="application/vnd.ms-excel", headers={ "Content-Disposition": f"attachment;filename=ClassQuiz-{urllib.parse.quote(quiz.title)}-{datetime.now().strftime('%m-%d-%Y')}.xlsx" # noqa: E501 }, )