# retoor <retoor@molodetz.nl>
from typing import Annotated
from fastapi import APIRouter, Depends, Request
from fastapi.responses import HTMLResponse, JSONResponse
from devplacepy.config import QUIZ_ANSWER_MAX_CHARS
from devplacepy.database import get_users_by_uids
from devplacepy.dependencies import json_or_form
from devplacepy.models import QuizAnswerForm
from devplacepy.responses import action_result, respond, wants_json
from devplacepy.schemas import (
QuizAnswerResultOut,
QuizAttemptPageOut,
QuizResultOut,
)
from devplacepy.services.audit import record as audit
from devplacepy.services.quiz import QuizError, store
from devplacepy.utils import (
award_rewards,
create_notification,
is_admin,
not_found,
require_user,
track_action,
XP_QUIZ_COMPLETE,
)
from ._shared import action_error, load_quiz, notify_quiz, quiz_seo, quiz_url
router = APIRouter()
def _load_attempt(quiz: dict, attempt_uid: str, user: dict, allow_admin: bool = False):
attempt = store.get_attempt(attempt_uid)
if not attempt or attempt.get("quiz_uid") != quiz["uid"]:
raise not_found("Attempt not found")
if attempt.get("user_uid") != user["uid"] and not (allow_admin and is_admin(user)):
raise not_found("Attempt not found")
return attempt
def _author(quiz: dict):
return get_users_by_uids([quiz["user_uid"]]).get(quiz["user_uid"])
@router.post("/{slug}/attempts")
async def start_attempt(request: Request, slug: str):
user = require_user(request)
quiz = load_quiz(slug, user)
if quiz.get("status") != "published":
return action_error(request, "This quiz is not published yet", quiz_url(quiz))
try:
attempt = store.start_attempt(user, quiz)
except QuizError as exc:
return action_error(request, str(exc), quiz_url(quiz))
audit.record(
request,
"quiz.attempt.start",
user=user,
target_type="quiz",
target_uid=quiz["uid"],
target_label=quiz.get("title") or quiz["uid"],
summary=f"{user['username']} started an attempt on quiz {quiz.get('title')}",
metadata={"attempt_uid": attempt["uid"]},
links=[audit.target("quiz", quiz["uid"], quiz.get("title") or "")],
)
url = f"{quiz_url(quiz)}/attempts/{attempt['uid']}"
return action_result(
request, url, data={"uid": attempt["uid"], "url": url, "status": attempt["status"]}
)
@router.get("/{slug}/attempts/{attempt_uid}", response_class=HTMLResponse)
async def attempt_page(request: Request, slug: str, attempt_uid: str):
user = require_user(request)
quiz = load_quiz(slug, user)
attempt = _load_attempt(quiz, attempt_uid, user)
payload = store.serialize_attempt(quiz, attempt, user)
seo_ctx = quiz_seo(
request,
quiz.get("title") or "Quiz",
"Play this quiz on DevPlace.",
robots="noindex,follow",
breadcrumbs=[{"name": quiz.get("title") or "Quiz", "url": quiz_url(quiz)}],
)
return respond(
request,
"quiz_play.html",
{
**seo_ctx,
"request": request,
"user": user,
"quiz": store.serialize_quiz(quiz, user, _author(quiz)),
"attempt": payload,
"answer_max_chars": QUIZ_ANSWER_MAX_CHARS,
"quiz_error": request.query_params.get("error", ""),
},
model=QuizAttemptPageOut,
)
@router.post("/{slug}/attempts/{attempt_uid}/answer")
async def submit_answer(
request: Request,
slug: str,
attempt_uid: str,
data: Annotated[QuizAnswerForm, Depends(json_or_form(QuizAnswerForm))],
):
user = require_user(request)
quiz = load_quiz(slug, user)
attempt = _load_attempt(quiz, attempt_uid, user)
attempt_url = f"{quiz_url(quiz)}/attempts/{attempt_uid}"
try:
answer, updated, result = await store.answer(
user,
quiz,
attempt,
data.question_uid,
data.submission(),
)
except QuizError as exc:
return action_error(request, str(exc), attempt_url)
if result.graded_by == "fallback":
audit.record_system(
"quiz.grade.failed",
result="failure",
target_type="quiz",
target_uid=quiz["uid"],
summary=(
f"AI grading unavailable for question {data.question_uid}, "
"deterministic fallback used"
),
metadata={"attempt_uid": attempt_uid, "question_uid": data.question_uid},
)
audit.record(
request,
"quiz.attempt.answer",
user=user,
target_type="quiz",
target_uid=quiz["uid"],
target_label=quiz.get("title") or quiz["uid"],
summary=(
f"{user['username']} answered a question on quiz {quiz.get('title')} "
f"for {answer.get('awarded_points')} points"
),
metadata={
"attempt_uid": attempt_uid,
"question_uid": data.question_uid,
"graded_by": result.graded_by,
},
links=[audit.target("quiz", quiz["uid"], quiz.get("title") or "")],
)
if not wants_json(request):
return action_result(request, attempt_url)
return JSONResponse(
QuizAnswerResultOut(
answer=store.serialize_answer(answer),
attempt=store.serialize_attempt(quiz, updated, user),
).model_dump(mode="json")
)
@router.post("/{slug}/attempts/{attempt_uid}/finish")
async def finish_attempt(request: Request, slug: str, attempt_uid: str):
user = require_user(request)
quiz = load_quiz(slug, user)
attempt = _load_attempt(quiz, attempt_uid, user)
finished, won = store.finish(quiz, attempt)
results_url = f"{quiz_url(quiz)}/attempts/{attempt_uid}/results"
if won:
store.clear_cache()
award_rewards(user["uid"], XP_QUIZ_COMPLETE, "Quiz Taker")
track_action(user["uid"], "quiz_complete")
if float(finished.get("score_percent") or 0.0) >= 100.0:
track_action(user["uid"], "quiz_perfect")
if quiz["user_uid"] != user["uid"]:
create_notification(
quiz["user_uid"],
"quiz_attempt",
f"{user['username']} completed your quiz {quiz.get('title')}",
user["uid"],
quiz_url(quiz),
)
audit.record(
request,
"quiz.attempt.finish",
user=user,
target_type="quiz",
target_uid=quiz["uid"],
target_label=quiz.get("title") or quiz["uid"],
summary=(
f"{user['username']} finished quiz {quiz.get('title')} with "
f"{finished.get('score_percent')}%"
),
metadata={
"attempt_uid": attempt_uid,
"score_percent": finished.get("score_percent"),
"passed": finished.get("passed"),
},
links=[audit.target("quiz", quiz["uid"], quiz.get("title") or "")],
)
await notify_quiz(quiz["uid"])
if not wants_json(request):
return action_result(request, results_url)
result = store.serialize_result(quiz, finished, user)
return JSONResponse(
QuizResultOut(
quiz=store.serialize_quiz(quiz, user, _author(quiz)),
attempt=result,
review=result["review"],
fallback_count=result["fallback_count"],
).model_dump(mode="json")
)
@router.get("/{slug}/attempts/{attempt_uid}/results", response_class=HTMLResponse)
async def attempt_results(request: Request, slug: str, attempt_uid: str):
user = require_user(request)
quiz = load_quiz(slug, user)
attempt = _load_attempt(quiz, attempt_uid, user, allow_admin=True)
result = store.serialize_result(quiz, attempt, user)
seo_ctx = quiz_seo(
request,
f"{quiz.get('title') or 'Quiz'} results",
"Your quiz result on DevPlace.",
robots="noindex,follow",
breadcrumbs=[{"name": quiz.get("title") or "Quiz", "url": quiz_url(quiz)}],
)
return respond(
request,
"quiz_results.html",
{
**seo_ctx,
"request": request,
"user": user,
"quiz": store.serialize_quiz(quiz, user, _author(quiz)),
"attempt": result,
"review": result["review"],
"fallback_count": result["fallback_count"],
},
model=QuizResultOut,
)
Reference in New Issue View Git Blame Copy Permalink