LobotomyLabs/classquiz-ai
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Further cleanup

Browse Source
This commit is contained in:
Mawoka
2025-11-01 15:57:10 +01:00
parent abd2155469
commit 02c917a28f
16 changed files with 215 additions and 336 deletions
Download Patch File Download Diff File Expand all files Collapse all files
classquiz/helpers/__init__.py
+43 -57
View File
@@ -6,14 +6,13 @@
import asyncio
import uuid
from datetime import datetime
from typing import Optional, BinaryIO
from typing import BinaryIO, Any
from fastapi import HTTPException
from openpyxl import load_workbook
import ormar.exceptions
from classquiz.db.models import Quiz, User, InstanceData, QuizQuestion, ABCDQuizAnswer
from classquiz.db.models import Quiz, User, QuizQuestion, ABCDQuizAnswer
import xlsxwriter
from aiohttp import ClientSession
from io import BytesIO
@@ -24,7 +23,7 @@ from classquiz.helpers.hashcash import check as hc_check
settings = settings()
async def get_meili_data(quiz: Quiz) -> dict:
async def get_meili_data(quiz: Quiz) -> dict[str, Any]:
return {
"id": str(quiz.id),
"title": quiz.title,
@@ -35,14 +34,19 @@ async def get_meili_data(quiz: Quiz) -> dict:
}
async def generate_spreadsheet(quiz_results: dict, quiz: Quiz, player_fields: dict, player_scores: dict) -> BytesIO:
async def generate_spreadsheet(
quiz_results: dict[str, Any],
quiz: Quiz,
player_fields: dict[str, Any],
player_scores: dict[str, Any],
) -> BytesIO:
storage = BytesIO()
workbook = xlsxwriter.Workbook(storage, {"in_memory": True})
player_worksheet = workbook.add_worksheet()
player_worksheet.name = "Players"
player_worksheet.write(0, 0, "Username")
player_worksheet.write(0, 1, "Score")
player_worksheet.write(0, 2, "Custom-Field")
_ = player_worksheet.write(0, 0, "Username")
_ = player_worksheet.write(0, 1, "Score")
_ = player_worksheet.write(0, 2, "Custom-Field")
for i, player in enumerate(player_scores.keys()):
player_worksheet.write(i + 1, 0, player)
player_worksheet.write(i + 1, 1, player_scores[player])
@@ -53,33 +57,33 @@ async def generate_spreadsheet(quiz_results: dict, quiz: Quiz, player_fields: di
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")
_ = 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 enumerate(quiz_results):
question = quiz.questions[i]
# print(quiz_results)
try:
answer_data = quiz_results[str(i)]
except KeyError:
continue
worksheet.write(i + 1, 0, question["question"])
worksheet.write(i + 1, 1, question["time"])
_ = worksheet.write(i + 1, 0, question["question"])
_ = worksheet.write(i + 1, 1, question["time"])
try:
async with (
ClientSession() as session,
session.get(f"{settings.root_address}/api/v1/storage/download/{question['image']}") as response,
):
if "image" in response.headers.get("Content-Type"):
content_type = response.headers.get("Content-Type")
if content_type is not None and "image" in content_type:
img_data = BytesIO(await response.read())
worksheet.insert_image(i + 1, 2, question["image"], {"image_data": img_data})
_ = worksheet.insert_image(i + 1, 2, question["image"], {"image_data": img_data})
image = Image.open(img_data)
worksheet.set_row(i + 1, image.height)
worksheet.set_column(2, 2, image.width)
_ = worksheet.set_row(i + 1, image.height)
_ = worksheet.set_column(2, 2, image.width)
except TypeError:
pass
answer_amount = len(answer_data)
@@ -91,24 +95,24 @@ async def generate_spreadsheet(quiz_results: dict, quiz: Quiz, player_fields: di
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)
_ = 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")
_ = ws.write(0, 0, "Answer")
_ = ws.write(0, 1, "Correct")
_ = ws.write(0, 2, "Username")
for j, _ in enumerate(answer_data):
ws.write(j + 1, 0, answer_data[j]["answer"])
_ = ws.write(j + 1, 0, answer_data[j]["answer"])
if answer_data[j]["right"]:
ws.write(j + 1, 1, "True")
_ = ws.write(j + 1, 1, "True")
else:
ws.write(j + 1, 1, "False")
ws.write(j + 1, 2, answer_data[j]["username"])
_ = ws.write(j + 1, 1, "False")
_ = ws.write(j + 1, 2, answer_data[j]["username"])
workbook.close()
storage.seek(0)
_ = storage.seek(0)
return storage
@@ -118,9 +122,11 @@ async def handle_import_from_excel(data: BinaryIO, user: User) -> Quiz:
except KeyError:
raise HTTPException(status_code=400, detail="File not in excel format")
ws = wb.active
if ws is None:
raise Exception("Workbook is None")
questions: list[dict] = []
title = ws["C5"].value
description = ws["C6"].value
title: str | None = ws["C5"].value
description: str | None = ws["C6"].value
if title is None:
raise HTTPException(status_code=400, detail="Title missing")
if description is None:
@@ -182,12 +188,12 @@ async def handle_import_from_excel(data: BinaryIO, user: User) -> Quiz:
questions=questions,
imported_from_kahoot=False,
)
await quiz.save()
_ = await quiz.save()
else:
existing_quiz.questions = [*existing_quiz.questions, *questions]
existing_quiz.updated_at = datetime.now()
existing_quiz.mod_rating = None
await existing_quiz.update()
_ = await existing_quiz.update()
quiz = existing_quiz
return quiz
@@ -227,27 +233,7 @@ async def meilisearch_init():
LOGGER.info("Finished MeiliSearch synchronisation")
async def telemetry_ping():
try:
instance_data = await InstanceData.objects.first()
except ormar.exceptions.NoMatch:
instance_data = InstanceData()
await instance_data.save()
async with (
ClientSession() as session,
session.post(
f"https://cit.mawoka.eu.org/public/{instance_data.instance_id}",
json={
"public_quizzes": await Quiz.objects.filter(public=True).count(),
"private_quizzes": await Quiz.objects.filter(public=False).count(),
"users": await User.objects.count(),
},
),
):
return
def check_hashcash(data: str, input_data: str, claim_in: Optional[str] = "19") -> bool:
def check_hashcash(data: str, input_data: str, claim_in: str | None = "19") -> bool:
"""
It checks that the hashcash is valid, and if it is, it returns True
classquiz/kahoot_importer/README.md
+3 -3
View File
@@ -15,8 +15,8 @@ This library is also tested.
```python
from classquiz.kahoot_importer.get import get, _Response
from asyncio import run
# _Response ia a pydantic-object, so you have access to
# .dict() or .json(exclude={"kahoot"})
# _Response is a pydantic-object, so you have access to
# .dict() or .model_dump_json(exclude={"kahoot"})
async def main():
kahoot_quiz: _Response = await get("GAME_ID")
@@ -31,7 +31,7 @@ run(main())
from classquiz.kahoot_importer.search import search, _Response
from asyncio import run
# _Response ia a pydantic-object, so you have access to
# .dict() or .json(exclude={"kahoot"})
# .dict() or .model_dump_json(exclude={"kahoot"})
async def main():
kahoot_quizzes: _Response = await search("QUERY")
classquiz/kahoot_importer/__init__.py
+22 -23
View File
@@ -3,7 +3,7 @@
# SPDX-License-Identifier: MPL-2.0
from typing import List, Any, Optional
from typing import Any
from uuid import UUID
from pydantic import BaseModel
@@ -33,10 +33,10 @@ class _LastEdit(BaseModel):
class _ImageMetadata(BaseModel):
id: UUID | None = None
content_type: Optional[str] = None
width: Optional[int] = None
height: Optional[int] = None
resources: Optional[str] = None
content_type: str | None = None
width: int | None = None
height: int | None = None
resources: str | None = None
class _SampleQuestion(BaseModel):
@@ -48,11 +48,11 @@ class _SampleQuestion(BaseModel):
class _Access(BaseModel):
groupRead: List[Any | None]
folderGroupIds: List[Any | None]
groupRead: list[Any | None]
folderGroupIds: list[Any | None]
class _Card(BaseModel):
class Card(BaseModel):
type: str
title: str
description: str
@@ -60,12 +60,12 @@ class _Card(BaseModel):
cover: str | None = None
coverMetadata: _CoverMetadata | dict[None, None] | None = None
draftExists: bool
inventoryItemIds: List[Any] = None
inventoryItemIds: list[Any] | None = None
number_of_questions: int
creator: UUID
creator_username: str
creator_avatar: _CreatorAvatar | dict[None, None] | None = None
badges: List[str]
badges: list[str]
visibility: int
locked: bool
writeProtection: bool
@@ -76,11 +76,11 @@ class _Card(BaseModel):
draft: bool
combined: bool
compatibility_level: int
sample_questions: List[_SampleQuestion]
sample_questions: list[_SampleQuestion]
number_of_plays: int
number_of_players: int
total_favourites: int
question_types: List[str]
question_types: list[str]
created: int
modified: int
access: _Access
@@ -89,7 +89,7 @@ class _Card(BaseModel):
class _Entity(BaseModel):
card: _Card
card: Card
class _Origin(BaseModel):
@@ -130,8 +130,8 @@ class _Video(BaseModel):
startTime: float
endTime: float
service: str
full_url: Optional[str] = None
id: Optional[str] = None
full_url: str | None = None
id: str | None = None
class _Question(BaseModel):
@@ -140,17 +140,17 @@ class _Question(BaseModel):
time: int
points: bool
pointsMultiplier: int
choices: List[_Choice]
choices: list[_Choice]
image: str | None = None
imageMetadata: _ImageMetadata | None = None
resources: Optional[str] = None
resources: str | None = None
video: _Video
questionFormat: int
languageInfo: _LanguageInfo | None = None
media: List[Any]
media: list[Any]
class _Kahoot(BaseModel):
class Kahoot(BaseModel):
uuid: UUID
language: str
creator: UUID
@@ -161,20 +161,19 @@ class _Kahoot(BaseModel):
visibility: int
difficulty: int | None = None
audience: str
audience: str
title: str
description: str
quizType: str
tags: str | None | List[str] = None
tags: str | None | list[str] = None
cover: str | None = None
coverMetadata: _CoverMetadata | dict[None, None] | None = None
questions: List[_Question]
questions: list[_Question]
metadata: _Metadata
parent: _Parent | None = None
resources: str | None = None
slug: str
languageInfo: _LanguageInfo | None = None
inventoryItemIds: List[Any]
inventoryItemIds: list[Any]
type: str
created: int
modified: int
classquiz/kahoot_importer/get.py
+3 -3
View File
@@ -6,12 +6,12 @@
from aiohttp import ClientSession
from pydantic import BaseModel
from classquiz.kahoot_importer import _Card, _Kahoot
from classquiz.kahoot_importer import Card, Kahoot
class _Response(BaseModel):
card: _Card
kahoot: _Kahoot
card: Card
kahoot: Kahoot
async def get(game_id: str) -> _Response | int:
classquiz/kahoot_importer/import_quiz.py
+2 -2
View File
@@ -58,7 +58,7 @@ async def import_quiz(quiz_id: str, user: User) -> Quiz | int:
if type(quiz) is int:
return quiz
quiz_questions: list[dict] = []
quiz_id = uuid.uuid4()
new_quiz_id = uuid.uuid4()
meilisearch.delete_index(settings.meilisearch_index)
meilisearch.create_index(settings.meilisearch_index)
uploaded_images: list[StorageItem] = []
@@ -97,7 +97,7 @@ async def import_quiz(quiz_id: str, user: User) -> Quiz | int:
if quiz.kahoot.description is None or quiz.kahoot.description == "":
quiz.kahoot.description = "Description Missing!"
quiz_data = Quiz(
id=quiz_id,
id=new_quiz_id,
public=False,
title=bleach.clean(quiz.kahoot.title, tags=ALLOWED_TAGS_FOR_QUIZ, strip=True),
description=bleach.clean(quiz.kahoot.description, tags=ALLOWED_TAGS_FOR_QUIZ, strip=True),
classquiz/kahoot_importer/search.py
+2 -4
View File
@@ -3,8 +3,6 @@
# SPDX-License-Identifier: MPL-2.0
from typing import List
from aiohttp import ClientSession
from pydantic import BaseModel
@@ -13,7 +11,7 @@ from classquiz.kahoot_importer import _Entity
# noqa : E501
class _Response(BaseModel):
entities: List[_Entity]
entities: list[_Entity]
totalHits: int
cursor: int | None = None
pageTimestamp: int
@@ -28,7 +26,7 @@ async def search(
) -> _Response:
"""
:param inventory_item_id: I dkon't know
:param inventory_item_id: I don't know
:param search_cluster: Doesn't seeem to matter
:param cursor: The position in the result-list (page)
:param query: The search query
classquiz/oauth/__init__.py
+3 -3
View File
@@ -22,7 +22,7 @@ router.include_router(github.router, prefix="/github")
router.include_router(custom.router, prefix="/custom")
async def rememberme_middleware(request: Request, call_next):
async def rememberme_middleware(request: Request, call_next) -> Response:
rememberme_cookie = request.cookies.get("rememberme_token")
bearer_token = request.cookies.get("access_token")
conditions_to_handle_met = True
@@ -41,13 +41,13 @@ async def rememberme_middleware(request: Request, call_next):
# Verifying the bearer
try:
jwt.decode(
_ = jwt.decode(
param, settings.secret_key, algorithms=["HS256"]
) # checking if the token is valid, throws error if not
conditions_to_handle_met = False
except JWTError:
try:
jws.verify(
_ = jws.verify(
param, settings.secret_key, algorithms=["HS256"]
) # Verifying only the signature of the jwt, throws error if signature is invalid
except JWSError:
classquiz/oauth/authenticate_user.py
+11 -6
View File
@@ -14,18 +14,19 @@ from classquiz.auth import create_access_token
settings = settings()
async def log_user_in(user: User, request: Request, response: Response):
async def log_user_in(user: User | None, request: Request, response: Response):
if user is None:
raise HTTPException(status_code=401, detail="User not matched!")
remote_ip = None
if request.headers.get("X-Forwarded-For") is None:
forwarded_for_header = request.headers.get("X-Forwarded-For")
if forwarded_for_header is None:
remote_ip = request.client.host
else:
if "," in request.headers.get("X-Forwarded-For"):
remote_ip = request.headers.get("X-Forwarded-For").split(", ")[0]
if "," in forwarded_for_header:
remote_ip = forwarded_for_header.split(", ")[0]
else:
remote_ip = request.headers.get("X-Forwarded-For")
remote_ip = forwarded_for_header
session_key = os.urandom(32).hex()
user_session = UserSession(
user=user,
@@ -47,7 +48,11 @@ async def log_user_in(user: User, request: Request, response: Response):
max_age=60 * 60 * 24 * 365,
)
response.set_cookie(
key="rememberme_token", value=session_key, httponly=True, samesite="lax", max_age=60 * 60 * 24 * 365
key="rememberme_token",
value=session_key,
httponly=True,
samesite="lax",
max_age=60 * 60 * 24 * 365,
)
return {"access_token": access_token, "token_type": "bearer"}
classquiz/oauth/custom.py
+20 -36
View File
@@ -77,43 +77,27 @@ async def auth(request: Request, response: Response):
except (TypeError, ValidationError):
raise HTTPException(status_code=401, detail="Something went wrong.")
user_in_db = await User.objects.get_or_none(email=user_data.email)
if user_in_db is None:
# REGISTER USER
try:
await User.objects.create(
id=uuid.uuid4(),
email=user_data.email,
username=user_data.preferred_username,
verified=user_data.email_verified,
auth_type=UserAuthTypes.CUSTOM,
google_uid=user_data.sub.hex,
avatar=gzipped_user_avatar(),
)
# skipcq: PYL-W0703
except Exception as e:
if type(e) is asyncpg.exceptions.UniqueViolationError:
error = True
counter = 1
while error:
try:
await User.objects.create(
id=uuid.uuid4(),
email=user_data.email,
username=f"{user_data.preferred_username}{counter}",
verified=user_data.email_verified,
auth_type=UserAuthTypes.CUSTOM,
google_uid=user_data.sub.hex,
avatar=gzipped_user_avatar(),
)
error = False
except asyncpg.exceptions.UniqueViolationError:
counter += 1
error = True
else:
raise HTTPException(status_code=500, detail=str(e))
try:
await User.objects.create(
id=uuid.uuid4(),
email=user_data.email,
username=user_data.preferred_username,
verified=user_data.email_verified,
auth_type=UserAuthTypes.CUSTOM,
google_uid=user_data.sub.hex,
avatar=gzipped_user_avatar(),
)
# skipcq: PYL-W0703
except asyncpg.exceptions.UniqueViolationError:
# Most likely a duplicate email/username, not UUID.
raise HTTPException(status_code=400, detail="User already exists.")
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
user = await User.objects.get_or_none(
email=user_data.email, google_uid=user_data.sub.hex, auth_type=UserAuthTypes.CUSTOM, verified=True
email=user_data.email,
google_uid=user_data.sub.hex,
auth_type=UserAuthTypes.CUSTOM,
verified=True,
)
print(user_data)
classquiz/oauth/github.py
+48 -68
View File
@@ -4,7 +4,6 @@
import uuid
from typing import Optional
import asyncpg
import authlib.integrations.base_client
@@ -35,43 +34,43 @@ class Plan(BaseModel):
class GitHubOauthResponse(BaseModel):
login: str
id: int
email: Optional[str] = None
node_id: Optional[str] = None
avatar_url: Optional[str] = None
gravatar_id: Optional[str] = None
url: Optional[str] = None
html_url: Optional[str] = None
followers_url: Optional[str] = None
following_url: Optional[str] = None
gists_url: Optional[str] = None
starred_url: Optional[str] = None
subscriptions_url: Optional[str] = None
organizations_url: Optional[str] = None
repos_url: Optional[str] = None
events_url: Optional[str] = None
received_events_url: Optional[str] = None
type: Optional[str] = None
site_admin: Optional[bool] = None
name: Optional[str] = None
company: Optional[str] = None
blog: Optional[str] = None
location: Optional[str] = None
hireable: Optional[bool] = None
bio: Optional[str] = None
twitter_username: Optional[str] = None
public_repos: Optional[int] = None
public_gists: Optional[int] = None
followers: Optional[int] = None
following: Optional[int] = None
email: str | None = None
node_id: str | None = None
avatar_url: str | None = None
gravatar_id: str | None = None
url: str | None = None
html_url: str | None = None
followers_url: str | None = None
following_url: str | None = None
gists_url: str | None = None
starred_url: str | None = None
subscriptions_url: str | None = None
organizations_url: str | None = None
repos_url: str | None = None
events_url: str | None = None
received_events_url: str | None = None
type: str | None = None
site_admin: bool | None = None
name: str | None = None
company: str | None = None
blog: str | None = None
location: str | None = None
hireable: bool | None = None
bio: str | None = None
twitter_username: str | None = None
public_repos: int | None = None
public_gists: int | None = None
followers: int | None = None
following: int | None = None
created_at: datetime
updated_at: datetime
private_gists: Optional[int] = None
total_private_repos: Optional[int] = None
owned_private_repos: Optional[int] = None
disk_usage: Optional[int] = None
collaborators: Optional[int] = None
two_factor_authentication: Optional[bool] = None
plan: Optional[Plan] = None
private_gists: int | None = None
total_private_repos: int | None = None
owned_private_repos: int | None = None
disk_usage: int | None = None
collaborators: int | None = None
two_factor_authentication: bool | None = None
plan: Plan | None = None
@router.get("/login")
@@ -107,39 +106,20 @@ async def auth(request: Request, response: Response):
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)
if user_in_db is None:
# REGISTER USER
try:
await User.objects.create(
id=uuid.uuid4(),
email=user_data.email,
username=user_data.login,
verified=True,
auth_type=UserAuthTypes.GITHUB,
avatar=gzipped_user_avatar(),
)
# skipcq: PYL-W0703
except Exception as e:
if type(e) is asyncpg.exceptions.UniqueViolationError:
error = True
counter = 1
while error:
try:
await User.objects.create(
id=uuid.uuid4(),
email=user_data.email,
username=user_data.login,
verified=True,
auth_type=UserAuthTypes.GITHUB,
avatar=gzipped_user_avatar(),
)
error = False
except asyncpg.exceptions.UniqueViolationError:
counter += 1
error = True
else:
raise HTTPException(status_code=500, detail=str(e))
try:
await User.objects.create(
id=uuid.uuid4(),
email=user_data.email,
username=user_data.login,
verified=True,
auth_type=UserAuthTypes.GITHUB,
avatar=gzipped_user_avatar(),
)
except asyncpg.exceptions.UniqueViolationError:
raise HTTPException(status_code=400, detail="User already exists.")
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
user = await User.objects.get_or_none(
email=user_data.email,
username=user_data.login,
classquiz/oauth/google.py
+18 -35
View File
@@ -79,43 +79,26 @@ async def auth(request: Request, response: Response):
user_data = OauthGoogleResponse(**user_data).userinfo
except (TypeError, ValidationError):
raise HTTPException(status_code=401, detail="Something went wrong.")
user_in_db = await User.objects.get_or_none(email=user_data.email)
if user_in_db is None:
# REGISTER USER
try:
await User.objects.create(
id=uuid.uuid4(),
email=user_data.email,
username=user_data.name,
verified=user_data.email_verified,
auth_type=UserAuthTypes.GOOGLE,
google_uid=user_data.sub,
avatar=gzipped_user_avatar(),
)
# skipcq: PYL-W0703
except Exception as e:
if type(e) is asyncpg.exceptions.UniqueViolationError:
error = True
counter = 1
while error:
try:
await User.objects.create(
id=uuid.uuid4(),
email=user_data.email,
username=f"{user_data.name}{counter}",
verified=user_data.email_verified,
auth_type=UserAuthTypes.GOOGLE,
google_uid=user_data.sub,
avatar=gzipped_user_avatar(),
)
error = False
except asyncpg.exceptions.UniqueViolationError:
counter += 1
error = True
else:
raise HTTPException(status_code=500, detail=str(e))
try:
await User.objects.create(
id=uuid.uuid4(),
email=user_data.email,
username=user_data.name,
verified=user_data.email_verified,
auth_type=UserAuthTypes.GOOGLE,
google_uid=user_data.sub,
avatar=gzipped_user_avatar(),
)
except asyncpg.exceptions.UniqueViolationError:
raise HTTPException(status_code=400, detail="User already exists.")
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
user = await User.objects.get_or_none(
email=user_data.email, google_uid=user_data.sub, auth_type=UserAuthTypes.GOOGLE, verified=True
email=user_data.email,
google_uid=user_data.sub,
auth_type=UserAuthTypes.GOOGLE,
verified=True,
)
await log_user_in(user=user, request=request, response=response)
classquiz/routers/admin.py
+3 -3
View File
@@ -14,15 +14,15 @@ router = APIRouter()
@router.delete("/user/id")
async def delete_user_by_id(user_id: UUID, user: User = Depends(get_admin_user)):
async def delete_user_by_id(user_id: UUID, _: User = Depends(get_admin_user)):
return {"deleted": await User.objects.delete(id=user_id)}
@router.delete("/user/username")
async def delete_user_by_username(username: str, user: User = Depends(get_admin_user)):
async def delete_user_by_username(username: str, _: User = Depends(get_admin_user)):
return {"deleted": await User.objects.delete(username=username)}
@router.delete("/user/email")
async def delete_user_by_email(email: str, user: User = Depends(get_admin_user)):
async def delete_user_by_email(email: str, _: User = Depends(get_admin_user)):
return {"deleted": await User.objects.delete(email=email)}
classquiz/routers/avatar.py
-13
View File
@@ -83,19 +83,6 @@ async def get_customized_avatar(
clothe_graphic_type=clothe_graphic_type,
).render_svg()
# skipcq: PY-W0069
# print(f"skin_color: {len(AvatarItemsAsList.skin_color)},")
# print(f"hair_color: {len(AvatarItemsAsList.hair_color)},")
# print(f"facial_hair_type: {len(AvatarItemsAsList.facial_hair_type)},")
# print(f"facial_hair_color: {len(AvatarItemsAsList.facial_hair_color)},")
# print(f"top_type: {len(AvatarItemsAsList.top_type)},")
# print(f"hat_color: {len(AvatarItemsAsList.hat_color)},")
# print(f"mouth_type: {len(AvatarItemsAsList.mouth_type)},")
# print(f"eyebrow_type: {len(AvatarItemsAsList.eyebrow_type)},")
# print(f"nose_type: {len(AvatarItemsAsList.nose_type)},")
# print(f"accessories_type: {len(AvatarItemsAsList.accessories_type)},")
# print(f"clothe_type: {len(AvatarItemsAsList.clothe_type)},")
# print(f"clothe_color: {len(AvatarItemsAsList.clothe_color)},")
# print(f"clothe_graphic_type: {len(AvatarItemsAsList.clothe_graphic_type)},")
return avatar
classquiz/routers/community.py
+17 -5
View File
@@ -17,18 +17,24 @@ router = APIRouter()
#
@router.get("/user/{user_id}", response_model_include={"username", "created_at", "id"}, response_model=User)
@router.get(
"/user/{user_id}",
response_model_include={"username", "created_at", "id"},
response_model=User,
)
async def get_user_by_user_id(user_id: UUID):
user = await User.objects.get_or_none(id=user_id)
# .select_related("quizs")
# print(user)
if user is None:
raise HTTPException(status_code=404, detail="user not found")
else:
return user
@router.get("/quizzes/{user_id}", response_model_exclude={"questions", "user_id"}, response_model=list[Quiz])
@router.get(
"/quizzes/{user_id}",
response_model_exclude={"questions", "user_id"},
response_model=list[Quiz],
)
async def get_quizzes_from_user(user_id: UUID, imported: bool | None = None):
if imported is None:
quizzes = await Quiz.objects.all(user_id=user_id, public=True)
@@ -68,7 +74,13 @@ async def rate_quiz(data: RateQuizInput, quiz_id: uuid.UUID, user: User = Depend
quiz.likes -= 1
else:
quiz.dislikes -= 1
rating = Rating(id=uuid.uuid4(), user=user, positive=positive, quiz=quiz, created_at=datetime.now())
rating = Rating(
id=uuid.uuid4(),
user=user,
positive=positive,
quiz=quiz,
created_at=datetime.now(),
)
await rating.save()
if positive:
quiz.likes += 1
classquiz/routers/eximport.py
+2 -2
View File
@@ -37,7 +37,7 @@ class UUIDEncoder(json.JSONEncoder):
@router.get("/{quiz_id}")
async def export_quiz(quiz_id: uuid.UUID, user: User = Depends(get_current_user)):
async def export_quiz(quiz_id: uuid.UUID, _: User = Depends(get_current_user)):
try:
quiz: Quiz = await Quiz.objects.filter(Quiz.id == quiz_id).first()
except ormar.exceptions.NoMatch:
@@ -138,7 +138,7 @@ async def import_quiz(file: UploadFile = File(), user: User = Depends(get_curren
@router.get("/excel/{quiz_id}")
async def export_quiz_as_excel(quiz_id: uuid.UUID, user: User = Depends(get_current_user)):
async def export_quiz_as_excel(quiz_id: uuid.UUID, _: User = Depends(get_current_user)):
try:
quiz: Quiz = await Quiz.objects.filter(Quiz.id == quiz_id).first()
except ormar.exceptions.NoMatch:
classquiz/routers/users/__init__.py
+18 -73
View File
@@ -10,22 +10,18 @@ from datetime import datetime
import ormar
import pydantic
from email_validator import validate_email, EmailNotValidError
from fastapi import APIRouter, Response, HTTPException, Request, Depends, status
from fastapi import APIRouter, Response, HTTPException, Request, Depends
from fastapi.background import BackgroundTasks
from fastapi.responses import JSONResponse, RedirectResponse, PlainTextResponse
from fastapi.security import OAuth2PasswordRequestForm
from jose import jwt, JWTError
from classquiz import oauth
from classquiz.helpers.avatar import gzipped_user_avatar
import base64
from classquiz.oauth.authenticate_user import rememberme_check, log_user_in
from classquiz.auth import (
get_password_hash,
verify_password,
authenticate_user,
get_current_user,
)
from classquiz.cache import clear_cache_for_account
@@ -33,7 +29,7 @@ from classquiz.config import redis, settings, meilisearch
import uuid
import bleach
from pydantic import BaseModel
from classquiz.db.models import User, UserSession, UpdatePassword, Token, Quiz, ApiKey
from classquiz.db.models import User, UserSession, UpdatePassword, Quiz, ApiKey
from classquiz.emails import send_register_email, send_forgotten_password_email
from classquiz.routers.users import webauthn, twofa
@@ -66,7 +62,12 @@ router.include_router(oauth.router, tags=["users", "oauth"], prefix="/oauth")
async def create_user(user: RouteUser, background_task: BackgroundTasks) -> User | JSONResponse:
if settings.registration_disabled:
raise HTTPException(status_code=423)
user = User(**user.model_dump(), id=uuid.uuid4(), avatar=gzipped_user_avatar(), created_at=datetime.now())
user: User = User(
**user.model_dump(),
id=uuid.uuid4(),
avatar=gzipped_user_avatar(),
created_at=datetime.now(),
)
try:
validate_email(user.email)
except EmailNotValidError as e:
@@ -91,31 +92,6 @@ async def create_user(user: RouteUser, background_task: BackgroundTasks) -> User
return user
@router.post("/token/cookie", response_model=Token, deprecated=True)
async def login_for_cookie_access_token(
request: Request,
response: Response,
form_data: OAuth2PasswordRequestForm = Depends(),
):
user = await authenticate_user(form_data.username, form_data.password)
user = await User.objects.select_related("fidocredentialss").get(id=user.id)
if not user or user.totp_secret is not None or user or len(user.fidocredentialss) != 0:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
)
return await log_user_in(response=response, request=request, user=user)
@router.get("/token/rememberme")
async def rememberme_token(request: Request, response: Response):
rememberme_token_lol = request.cookies.get("rememberme_token")
if rememberme_token_lol is None:
raise HTTPException(status_code=400, detail="No rememberme cookie")
return await rememberme_check(rememberme_token=rememberme_token_lol, response=response)
@router.get("/logout")
async def logout(request: Request, response: Response):
remember_token = request.cookies.get("rememberme_token")
@@ -229,24 +205,26 @@ async def reset_password_with_token(reset_password: ResetPassword, response: Res
return {"message": "Password updated successfully"}
@router.get("/sessions/list", response_model=list[UserSession], response_model_exclude={"user", "session_key", "quizs"})
@router.get(
"/sessions/list",
response_model=list[UserSession],
response_model_exclude={"user", "session_key", "quizs"},
)
async def list_sessions(user: User = Depends(get_current_user)):
sessions = await UserSession.objects.filter(user=user).all()
return [session.model_dump() for session in sessions]
@router.delete("/sessions/{session_id}")
async def delete_session(session_id: str, user: User = Depends(get_current_user)):
try:
session_id = uuid.UUID(session_id)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid session id")
async def delete_session(session_id: uuid.UUID, user: User = Depends(get_current_user)):
await UserSession.objects.filter(user=user, id=session_id).delete()
return {"message": "Session deleted"}
@router.get(
"/session", response_model=UserSession, response_model_exclude={"user": ..., "session_key": ..., "quizs": ...}
"/session",
response_model=UserSession,
response_model_exclude={"user": ..., "session_key": ..., "quizs": ...},
)
async def get_session(request: Request, user: User = Depends(get_current_user)):
try:
@@ -300,43 +278,10 @@ class InternalAuthData(BaseModel):
jwt: str | None = None
@router.post("/auth/internal")
async def internal_auth(data: InternalAuthData, resp: Response):
try:
data.jwt = data.jwt.replace("Bearer ", "")
except AttributeError:
pass
if data.jwt is not None:
try:
payload = jwt.decode(data.jwt, settings.secret_key, algorithms=["HS256"])
email: str = payload.get("sub")
if email is None:
resp.status_code = 401
return resp
except JWTError:
resp.status_code = 401
return resp
else:
return await rememberme_check(data.rememberme, resp)
class GetEmailFromJWT(BaseModel):
jwt: str
@router.post("/auth/internal/email")
async def get_email_from_jwt(data: GetEmailFromJWT):
try:
data.jwt = data.jwt.replace("Bearer ", "")
except AttributeError:
pass
try:
payload = jwt.decode(data.jwt, settings.secret_key, algorithms=["HS256"])
return payload.get("sub")
except JWTError:
raise HTTPException(status_code=401)
@router.post("/api_keys", response_model=ApiKey, response_model_include={"key"})
async def generate_api_key(user: User = Depends(get_current_user)):
key = ApiKey(key=os.urandom(24).hex(), user=user)
@@ -351,7 +296,7 @@ async def list_api_keys(user: User = Depends(get_current_user)):
@router.delete("/api_keys")
async def delete_api_key(api_key: str, user: User = Depends(get_current_user)):
async def delete_api_key(api_key: str, _: User = Depends(get_current_user)):
key = await ApiKey.objects.get_or_none(key=api_key)
if key is None:
raise HTTPException(status_code=404, detail="Key not found")