✨ #41

classquiz/helpers.py

@@ -1,4 +1,8 @@
 from classquiz.db.models import Quiz, User
+import xlsxwriter
+from aiohttp import ClientSession
+from io import BytesIO
+from PIL import Image
 
 
 async def get_meili_data(quiz: Quiz) -> dict:
@@ -9,3 +13,57 @@ async def get_meili_data(quiz: Quiz) -> dict:
         "user": (await User.objects.filter(id=quiz.user_id).first()).username,
         "created_at": int(quiz.created_at.timestamp()),
     }
+
+
+async def generate_spreadsheet(quiz_results: dict, quiz: Quiz, with_images: bool = True) -> BytesIO:
+    storage = BytesIO()
+    workbook = xlsxwriter.Workbook(storage, {"in_memory": True})
+    worksheet = workbook.add_worksheet()
+    worksheet.name = "Questions"
+    worksheet.write(0, 0, "Question")
+    worksheet.write(0, 1, "Time")
+    worksheet.write(0, 2, "Image")
+    worksheet.write(0, 3, "Correct answers")
+    worksheet.write(0, 4, "Correct answers")
+    worksheet.write(0, 5, "Wrong answers")
+    for i in range(len(quiz_results)):
+        question = quiz.questions[i]
+        answer_data = quiz_results[str(i)]
+        worksheet.write(i + 1, 0, question["question"])
+        worksheet.write(i + 1, 1, question["time"])
+        if with_images:
+            if quiz[i]["image"] != "" or question["image"] is not None:
+                async with ClientSession() as session, session.get(question["image"]) as response:
+                    img_data = BytesIO(await response.read())
+                    worksheet.insert_image(i + 1, 2, quiz[i]["image"], {"image_data": img_data})
+                    image = Image.open(img_data)
+                    worksheet.set_row(i + 1, image.height)
+                    worksheet.set_column(2, 2, image.width)
+        answer_amount = len(answer_data)
+        correct_answers = 0
+        wrong_answers = 0
+        for j in answer_data:
+            j_correct = j["right"]
+            if j_correct:
+                correct_answers += 1
+            else:
+                wrong_answers += 1
+        worksheet.write(i + 1, 3, f"{round(correct_answers / answer_amount * 100)}%")
+        worksheet.write(i + 1, 4, correct_answers)
+        worksheet.write(i + 1, 5, wrong_answers)
+
+        ws = workbook.add_worksheet(f"{i + 1}. Question")
+        ws.write(0, 0, "Answer")
+        ws.write(0, 1, "Correct")
+        ws.write(0, 2, "Username")
+        for j in range(len(answer_data)):
+            ws.write(j + 1, 0, answer_data[j]["answer"])
+            if answer_data[j]["right"]:
+                ws.write(j + 1, 1, "True")
+            else:
+                ws.write(j + 1, 1, "False")
+            ws.write(j + 1, 2, answer_data[j]["username"])
+
+    workbook.close()
+    storage.seek(0)
+    return storage

classquiz/routers/quiz.py

@@ -3,9 +3,9 @@ import re
 import uuid
 from datetime import datetime
 from random import randint
-from classquiz.helpers import get_meili_data
+from classquiz.helpers import get_meili_data, generate_spreadsheet
 from fastapi import APIRouter, Depends, HTTPException
-from fastapi.responses import JSONResponse
+from fastapi.responses import JSONResponse, StreamingResponse
 from pydantic import ValidationError, BaseModel
 import bleach
 
@@ -196,3 +196,29 @@ async def delete_quiz(quiz_id: str, user: User = Depends(get_current_user)):
         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)
+    game_data = PlayGame(**json.loads(await redis.get(f"game:{game_pin}")))
+    print(data)
+    quiz = await Quiz.objects.get_or_none(id=game_data.quiz_id)
+    print(quiz)
+    if quiz is None:
+        raise HTTPException(status_code=404, detail="quiz not found")
+    spreadsheet = await generate_spreadsheet(quiz=quiz, quiz_results=data)
+    print(len(spreadsheet.getvalue()))
+
+    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-{quiz.title}.xlsx"},
+    )

classquiz/socket_server/__init__.py

@@ -1,4 +1,5 @@
 import json
+import os
 
 import aiohttp
 import socketio
@@ -10,6 +11,17 @@ sio = socketio.AsyncServer(async_mode="asgi", cors_allowed_origins=[])
 settings = settings()
 
 
+async def generate_final_results(game_data: PlayGame, game_pin: str) -> dict:
+    results = {}
+    for i in range(len(game_data.questions)):
+        redis_res = await redis.get(f"game_session:{game_pin}:{i}")
+        if redis_res is None:
+            break
+        else:
+            results[str(i)] = json.loads(redis_res)
+    return results
+
+
 @sio.event
 async def join_game(sid: str, data: dict):
     async with aiohttp.ClientSession() as session:
@@ -156,15 +168,24 @@ async def submit_answer(sid: str, data: dict):
 
 @sio.event
 async def get_final_results(sid: str, _data: dict):
-    session = await sio.get_session(sid)
+    session: dict = await sio.get_session(sid)
     game_data = PlayGame(**json.loads(await redis.get(f"game:{session['game_pin']}")))
     results = {}
     if not session["admin"]:
         return
-    for i in range(len(game_data.questions)):
-        redis_res = await redis.get(f"game_session:{session['game_pin']}:{i}")
-        if redis_res is None:
-            break
-        else:
-            results[str(i)] = json.loads(redis_res)
+    results = await generate_final_results(game_data, session["game_pin"])
     await sio.emit("final_results", results, room=session["game_pin"])
+
+
+@sio.event
+async def get_export_token(sid):
+    session = await sio.get_session(sid)
+    print("generating export token")
+    if not session["admin"]:
+        return
+    game_data = PlayGame(**json.loads(await redis.get(f"game:{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))
+    print("generated export token", token)
+    await sio.emit("export_token", token, room=sid)

classquiz/tests/__init__.py

@@ -53,6 +53,8 @@ example_quiz = {
 test_user_email = "sth@byom.de"
 test_user_password = "test"
 
+# mock_test_results = {'0': [{'username': 'Player 1', 'answer': 'Byte, Bit, KB, MB, GB, TB', 'right': False}, {'username': 'Player 2', 'answer': 'Byte, Bit, KB, MB, GB, TB', 'right': False}, {'username': 'Player 3', 'answer': 'Bit, Byte, KB, MB, GB, TB', 'right': True}], '1': [{'username': 'Player 3', 'answer': 'CPU', 'right': True}, {'username': 'Player 2', 'answer': 'CPU', 'right': True}, {'username': 'Player 1', 'answer': 'CPU', 'right': True}], '2': [{'username': 'Player 1', 'answer': 'Peripheriegeräte', 'right': True}, {'username': 'Player 2', 'answer': 'Speichergeräte', 'right': False}, {'username': 'Player 3', 'answer': 'Kommunikationsgeräte', 'right': False}], '3': [{'username': 'Player 1', 'answer': 'die Taktfrequenz der CPU', 'right': True}, {'username': 'Player 2', 'answer': 'die Taktfrequenz der CPU', 'right': True}, {'username': 'Player 3', 'answer': 'die Taktfrequenz der CPU', 'right': True}], '4': [{'username': 'Player 3', 'answer': 'Drucker', 'right': True}, {'username': 'Player 2', 'answer': 'Drucker', 'right': True}, {'username': 'Player 1', 'answer': 'Tabellenkalkulation', 'right': False}], '5': [{'username': 'Player 1', 'answer': 'CPU (Central Processing Unit)', 'right': False}, {'username': 'Player 2', 'answer': 'RAM (Random Access Memory)', 'right': True}, {'username': 'Player 3', 'answer': 'RAM (Random Access Memory)', 'right': True}], '6': [{'username': 'Player 3', 'answer': 'RAM', 'right': True}, {'username': 'Player 2', 'answer': 'ROM', 'right': False}, {'username': 'Player 1', 'answer': 'RAM', 'right': True}], '7': [{'username': 'Player 1', 'answer': '1 TB (Terabyte)', 'right': True}, {'username': 'Player 2', 'answer': '500 KB (Kilobyte)', 'right': False}, {'username': 'Player 3', 'answer': '500 TB (Terabyte)', 'right': False}], '8': [{'username': 'Player 3', 'answer': 'Tabellenkalkulationsprogramm', 'right': True}, {'username': 'Player 2', 'answer': 'Tabellenkalkulationsprogramm', 'right': True}, {'username': 'Player 1', 'answer': 'Linux', 'right': False}], '9': [{'username': 'Player 1', 'answer': 'Anwendungsprogramme', 'right': True}, {'username': 'Player 2', 'answer': 'Lernprogramme', 'right': False}, {'username': 'Player 3', 'answer': 'Anwendungsprogramme', 'right': True}]}
+
 
 @pytest.fixture(scope="module")
 def test_client() -> Generator: