✨ MOVED LIVE_ENDPOINTS and added color to live-API

classquiz/__init__.py

@@ -10,7 +10,7 @@ from starlette.middleware.sessions import SessionMiddleware
 
 from classquiz.config import settings
 from classquiz.db import database
-from classquiz.routers import users, quiz, utils, stats, storage, search, testing_routes, editor
+from classquiz.routers import users, quiz, utils, stats, storage, search, testing_routes, editor, live
 from classquiz.socket_server import sio
 from classquiz.helpers import meilisearch_init, telemetry_ping
 
@@ -55,7 +55,8 @@ app.include_router(quiz.router, tags=["quiz"], prefix="/api/v1/quiz", include_in
 app.include_router(utils.router, tags=["utils"], prefix="/api/v1/utils", include_in_schema=True)
 app.include_router(stats.router, tags=["stats"], prefix="/api/v1/stats", include_in_schema=True)
 app.include_router(storage.router, tags=["storage"], prefix="/api/v1/storage", include_in_schema=True)
-app.include_router(search.router, tags=["search"], prefix="/api/v1/search", include_in_schema=True)
+app.include_router(search.router, tags=["search"], prefix="/api/v1/search", include_in_schema=True),
+app.include_router(live.router, tags=["live"], prefix="/api/v1/live", include_in_schema=True)
 app.include_router(
     testing_routes.router, tags=["internal", "testing"], prefix="/api/v1/internal/testing", include_in_schema=False
 )

classquiz/routers/live.py

@@ -0,0 +1,169 @@
+#  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
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel
+from classquiz.config import settings, redis
+from classquiz.db.models import (
+    PlayGame,
+    GamePlayer,
+    GameAnswer1,
+    GameAnswer2,
+    GameSession,
+    QuizQuestion,
+    RangeQuizAnswer,
+    ABCDQuizAnswer,
+    QuizQuestionType,
+)
+from classquiz.auth import check_api_key
+from classquiz.socket_server import ReturnQuestion, sio
+
+settings = settings()
+
+router = APIRouter()
+
+
+class _GetLiveDataPlayers(BaseModel):
+    count: int | str
+
+
+class _ABCDQuizAnswer(ABCDQuizAnswer):
+    color_code: str
+
+
+class _QuizQuestion(QuizQuestion):
+    answers: list[ABCDQuizAnswer] | RangeQuizAnswer
+
+
+class _GetLivePlayGame(PlayGame):
+    questions: list[_QuizQuestion]
+
+
+class GetLiveDataResponse(BaseModel):
+    quiz: _GetLivePlayGame
+    data: GameSession
+    players: _GetLiveDataPlayers
+
+
+@router.get("/")
+async def get_live_game_data(
+    game_pin: int, api_key: str, player_count_as_a_string: bool = False, in_array: bool = False
+):
+    user_id = await check_api_key(api_key)
+    redis_res = await redis.get(f"game:{game_pin}")
+    if redis_res is None or user_id is None:
+        raise HTTPException(status_code=404, detail="Game not found or API key not found")
+    game = PlayGame.parse_raw(redis_res)
+    if game.user_id != user_id:
+        raise HTTPException(status_code=404, detail="Game not found or API key not found")
+    for i, question in enumerate(game.questions):
+        if question.type == QuizQuestionType.ABCD:
+            for o, answer in enumerate(question.answers):
+                color = "#00F007"
+                if not answer.right:
+                    color = "#FF0000"
+                game.questions[i].answers[o] = _ABCDQuizAnswer(
+                    right=answer.right, answer=answer.answer, color=answer.color, color_code=color
+                )
+
+    data = GameSession.parse_raw(await redis.get(f"game_session:{game_pin}"))
+    for i in range(0, len(game.questions)):
+        res = await redis.get(f"game_session:{game_pin}:{i}")
+        if res is None:
+            break
+        else:
+            res = json.loads(res)
+            ga_1 = GameAnswer1(id=i, answers=[GameAnswer2.parse_obj(i) for i in res])
+            data.answers.append(ga_1)
+    player_count = await redis.scard(f"game_session:{game_pin}:players")
+    return_obj = None
+    if player_count_as_a_string:
+        return_obj = GetLiveDataResponse(quiz=game, data=data, players=_GetLiveDataPlayers(count=str(player_count)))
+    else:
+        return_obj = GetLiveDataResponse(quiz=game, data=data, players=_GetLiveDataPlayers(count=player_count))
+
+    if in_array:
+        return [return_obj]
+    else:
+        return return_obj
+
+
+@router.get("/user_count")
+async def get_game_user_count(game_pin: int, as_string: bool = False):
+    # if redis_res is None:
+    #     raise HTTPException(status_code=404, detail="Game not found")
+    player_count = await redis.scard(f"game_session:{game_pin}:players")
+    if as_string:
+        return {"players": {"count": str(player_count)}}
+    else:
+        return {"players": {"count": player_count}}
+
+
+class _LivePlayersReturn(BaseModel):
+    # players: list[GamePlayer | None]
+    answers: list[GameAnswer1 | None]
+    players: list[GamePlayer | None]
+
+
+@router.get(
+    "/players",
+)
+async def get_game_session(game_pin: int, api_key: str, in_array: bool = False):
+    user_id = await check_api_key(api_key)
+    redis_res = await redis.get(f"game_session:{game_pin}")
+    if redis_res is None or user_id is None:
+        raise HTTPException(status_code=404, detail="Game not found or API key not found")
+    data = GameSession.parse_raw(redis_res)
+    game = PlayGame.parse_raw(await redis.get(f"game:{game_pin}"))
+    if game.user_id != user_id:
+        raise HTTPException(status_code=404, detail="Game not found or API key not found")
+    for i in range(0, len(game.questions)):
+        res = await redis.get(f"game_session:{game_pin}:{i}")
+        if res is None:
+            break
+        else:
+            res = json.loads(res)
+            ga_1 = GameAnswer1(id=i, answers=[GameAnswer2.parse_obj(i) for i in res])
+            data.answers.append(ga_1)
+    players = await redis.smembers(f"game_session:{game_pin}:players")
+    player_list = []
+    for p in players:
+        player_list.append(GamePlayer.parse_raw(p))
+    if in_array:
+        return [_LivePlayersReturn(answers=data.answers, players=player_list)]
+    else:
+        return _LivePlayersReturn(answers=data.answers, players=player_list)
+
+
+@router.post("/set_question")
+async def set_next_question(game_pin: int, question_number: int, api_key: str):
+    user_id = await check_api_key(api_key)
+    redis_res = await redis.get(f"game:{game_pin}")
+    if redis_res is None or user_id is None:
+        raise HTTPException(status_code=404, detail="Game not found or API key not found")
+    game_data = PlayGame.parse_raw(redis_res)
+    if game_data.user_id != user_id:
+        raise HTTPException(status_code=404, detail="Game not found or API key not found")
+    game_data.current_question = question_number
+    await redis.set(f"game:{game_pin}", game_data.json())
+    await sio.emit(
+        "set_question_number",
+        {
+            "question_index": question_number,
+            "question": ReturnQuestion(**game_data.dict(include={"questions"})["questions"][question_number]).dict(),
+        },
+        room=game_pin,
+    )
+
+
+@router.get("/scores")
+async def get_live_player_scores(game_pin: int):
+    res = await redis.hgetall(f"game_session:{game_pin}:player_scores")
+    return_arr = []
+    for username in res:
+        return_arr.append({"username": username, "score": int(res[username])})
+    return_arr = sorted(return_arr, key=lambda d: d["score"], reverse=True)
+    return return_arr

classquiz/routers/quiz.py

@@ -16,12 +16,11 @@ from fastapi.responses import JSONResponse, StreamingResponse
 from pydantic import ValidationError, BaseModel
 import bleach
 
-from classquiz.auth import get_current_user, check_api_key
+from classquiz.auth import get_current_user
 from classquiz.config import redis, settings, storage, meilisearch
-from classquiz.db.models import Quiz, QuizInput, User, PlayGame, GameSession, GameAnswer1, GameAnswer2, GamePlayer
+from classquiz.db.models import Quiz, QuizInput, User, PlayGame
 from classquiz.kahoot_importer.import_quiz import import_quiz
 import html
-from classquiz.socket_server import sio, ReturnQuestion
 
 settings = settings()
 
@@ -253,125 +252,3 @@ async def export_quiz_answers(export_token: str, game_pin: str):
         media_type="application/vnd.ms-excel",
         headers={"Content-Disposition": f"attachment;filename=ClassQuiz-{quiz.title}.xlsx"},
     )
-
-
-class _GetLiveDataPlayers(BaseModel):
-    count: int | str
-
-
-class GetLiveDataResponse(BaseModel):
-    quiz: PlayGame
-    data: GameSession
-    players: _GetLiveDataPlayers
-
-
-@router.get("/live", tags=["live"])
-async def get_live_game_data(
-    game_pin: int, api_key: str, player_count_as_a_string: bool = False, in_array: bool = False
-):
-    user_id = await check_api_key(api_key)
-    redis_res = await redis.get(f"game:{game_pin}")
-    if redis_res is None or user_id is None:
-        raise HTTPException(status_code=404, detail="Game not found or API key not found")
-    game = PlayGame.parse_raw(redis_res)
-    if game.user_id != user_id:
-        raise HTTPException(status_code=404, detail="Game not found or API key not found")
-    data = GameSession.parse_raw(await redis.get(f"game_session:{game_pin}"))
-    for i in range(0, len(game.questions)):
-        res = await redis.get(f"game_session:{game_pin}:{i}")
-        if res is None:
-            break
-        else:
-            res = json.loads(res)
-            ga_1 = GameAnswer1(id=i, answers=[GameAnswer2.parse_obj(i) for i in res])
-            data.answers.append(ga_1)
-    player_count = await redis.scard(f"game_session:{game_pin}:players")
-    return_obj = None
-    if player_count_as_a_string:
-        return_obj = GetLiveDataResponse(quiz=game, data=data, players=_GetLiveDataPlayers(count=str(player_count)))
-    else:
-        return_obj = GetLiveDataResponse(quiz=game, data=data, players=_GetLiveDataPlayers(count=player_count))
-
-    if in_array:
-        return [return_obj]
-    else:
-        return return_obj
-
-
-@router.get("/live/user_count", tags=["live"])
-async def get_game_user_count(game_pin: int, as_string: bool = False):
-    # if redis_res is None:
-    #     raise HTTPException(status_code=404, detail="Game not found")
-    player_count = await redis.scard(f"game_session:{game_pin}:players")
-    if as_string:
-        return {"players": {"count": str(player_count)}}
-    else:
-        return {"players": {"count": player_count}}
-
-
-class _LivePlayersReturn(BaseModel):
-    # players: list[GamePlayer | None]
-    answers: list[GameAnswer1 | None]
-    players: list[GamePlayer | None]
-
-
-@router.get(
-    "/live/players",
-    tags=["live"],
-)
-async def get_game_session(game_pin: int, api_key: str, in_array: bool = False):
-    user_id = await check_api_key(api_key)
-    redis_res = await redis.get(f"game_session:{game_pin}")
-    if redis_res is None or user_id is None:
-        raise HTTPException(status_code=404, detail="Game not found or API key not found")
-    data = GameSession.parse_raw(redis_res)
-    game = PlayGame.parse_raw(await redis.get(f"game:{game_pin}"))
-    if game.user_id != user_id:
-        raise HTTPException(status_code=404, detail="Game not found or API key not found")
-    for i in range(0, len(game.questions)):
-        res = await redis.get(f"game_session:{game_pin}:{i}")
-        if res is None:
-            break
-        else:
-            res = json.loads(res)
-            ga_1 = GameAnswer1(id=i, answers=[GameAnswer2.parse_obj(i) for i in res])
-            data.answers.append(ga_1)
-    players = await redis.smembers(f"game_session:{game_pin}:players")
-    player_list = []
-    for p in players:
-        player_list.append(GamePlayer.parse_raw(p))
-    if in_array:
-        return [_LivePlayersReturn(answers=data.answers, players=player_list)]
-    else:
-        return _LivePlayersReturn(answers=data.answers, players=player_list)
-
-
-@router.post("/live/set_question", tags=["live"])
-async def set_next_question(game_pin: int, question_number: int, api_key: str):
-    user_id = await check_api_key(api_key)
-    redis_res = await redis.get(f"game:{game_pin}")
-    if redis_res is None or user_id is None:
-        raise HTTPException(status_code=404, detail="Game not found or API key not found")
-    game_data = PlayGame.parse_raw(redis_res)
-    if game_data.user_id != user_id:
-        raise HTTPException(status_code=404, detail="Game not found or API key not found")
-    game_data.current_question = question_number
-    await redis.set(f"game:{game_pin}", game_data.json())
-    await sio.emit(
-        "set_question_number",
-        {
-            "question_index": question_number,
-            "question": ReturnQuestion(**game_data.dict(include={"questions"})["questions"][question_number]).dict(),
-        },
-        room=game_pin,
-    )
-
-
-@router.get("/live/scores", tags=["live"])
-async def get_live_player_scores(game_pin: int):
-    res = await redis.hgetall(f"game_session:{game_pin}:player_scores")
-    return_arr = []
-    for username in res:
-        return_arr.append({"username": username, "score": int(res[username])})
-    return_arr = sorted(return_arr, key=lambda d: d["score"], reverse=True)
-    return return_arr