Files
devplacepy/devplacepy/routers/quizzes/index.py
T
2026-07-26 16:46:41 +02:00

364 lines
12 KiB
Python

# 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_SCOREBOARD_LIMIT
from devplacepy.content import (
canonical_redirect,
create_content_item,
delete_content_item,
detail_context,
load_detail,
)
from devplacepy.database import (
get_reactions_by_targets,
get_recent_comments_by_target_uids,
get_user_bookmarks,
get_users_by_uids,
get_user_votes,
mark_notifications_read_by_target,
resolve_object_url,
)
from devplacepy.dependencies import json_or_form
from devplacepy.models import QuizForm, QuizImportForm
from devplacepy.responses import action_result, respond
from devplacepy.schemas import (
QuizDetailOut,
QuizDocumentOut,
QuizFormPageOut,
QuizLeaderboardOut,
QuizScoreboardOut,
QuizzesOut,
)
from devplacepy.seo import quiz_schema, site_url, website_schema
from devplacepy.services.audit import record as audit
from devplacepy.services.quiz import QuizError, store
from devplacepy.utils import (
award_rewards,
get_current_user,
require_user,
time_ago,
track_action,
XP_QUIZ,
XP_QUIZ_PUBLISH,
)
from ._shared import action_error, load_quiz, notify_quiz, quiz_seo, quiz_url, require_editor
router = APIRouter()
def _list_items(rows: list[dict], user: dict | None) -> list[dict]:
if not rows:
return []
uids = [row["uid"] for row in rows]
authors = get_users_by_uids([row["user_uid"] for row in rows])
counts = store.comment_counts(uids)
reactions = get_reactions_by_targets("quiz", uids, user)
recent = get_recent_comments_by_target_uids("quiz", uids, 3, user)
bookmarks = get_user_bookmarks(user["uid"], "quiz", uids) if user else set()
votes = get_user_votes(user["uid"], uids) if user else {}
states = store.attempt_states_for(user["uid"], uids) if user else {}
items = []
for row in rows:
state = states.get(row["uid"], {})
items.append(
{
"uid": row["uid"],
"slug": row.get("slug") or row["uid"],
"url": quiz_url(row),
"title": row.get("title") or "",
"description": row.get("description") or "",
"status": row.get("status") or "published",
"created_at": row.get("created_at") or "",
"time_ago": time_ago(row.get("created_at") or ""),
"author": authors.get(row["user_uid"]),
"question_count": int(row.get("question_count") or 0),
"total_points": int(row.get("total_points") or 0),
"attempt_count": int(row.get("attempt_count") or 0),
"time_limit_seconds": int(row.get("time_limit_seconds") or 0),
"pass_percent": int(row.get("pass_percent") or 0),
"stars": int(row.get("stars") or 0),
"my_vote": votes.get(row["uid"], 0),
"comment_count": counts.get(row["uid"], 0),
"bookmarked": row["uid"] in bookmarks,
"reactions": reactions.get(row["uid"], {"counts": {}, "mine": []}),
"recent_comments": recent.get(row["uid"], []),
"viewer_state": state.get("state", "todo"),
"viewer_best_percent": state.get("best_percent", 0.0),
"viewer_best_points": state.get("best_points", 0.0),
"viewer_attempt_uid": state.get("attempt_uid", ""),
}
)
return items
@router.get("", response_class=HTMLResponse)
async def quizzes_page(
request: Request, filter: str = "all", search: str = "", page: int = 1
):
user = get_current_user(request)
current_filter = filter if filter in store.FILTERS else "all"
rows, pagination = store.list_quizzes(
viewer=user, quiz_filter=current_filter, search=search, page=max(1, page)
)
standing = store.standing_for(user["uid"]) if user else None
seo_ctx = quiz_seo(
request,
"Quizzes",
"Author quizzes, play them, and climb the DevPlace quiz scoreboard.",
schemas=[website_schema(site_url(request))],
)
return respond(
request,
"quizzes.html",
{
**seo_ctx,
"request": request,
"user": user,
"quizzes": _list_items(rows, user),
"search": search,
"filter": current_filter,
"current_filter": current_filter,
"counts": store.filter_counts(user, search),
"pagination": pagination,
"scoreboard": store.scoreboard(QUIZ_SCOREBOARD_LIMIT),
"viewer_standing": standing,
"viewer_progress": store.progress_for(user["uid"]) if user else {},
"viewer_can_create": bool(user),
},
model=QuizzesOut,
)
@router.get("/scoreboard")
async def quizzes_scoreboard(request: Request, limit: int = QUIZ_SCOREBOARD_LIMIT):
user = get_current_user(request)
bounded = max(1, min(100, int(limit or QUIZ_SCOREBOARD_LIMIT)))
return JSONResponse(
QuizScoreboardOut(
scoreboard=store.scoreboard(bounded),
viewer_standing=store.standing_for(user["uid"]) if user else None,
limit=bounded,
).model_dump(mode="json")
)
@router.get("/new", response_class=HTMLResponse)
async def quiz_new_page(request: Request):
user = require_user(request)
seo_ctx = quiz_seo(
request,
"New quiz",
"Create a quiz on DevPlace.",
robots="noindex,follow",
)
return respond(
request,
"quiz_new.html",
{**seo_ctx, "request": request, "user": user, "viewer_can_create": True},
model=QuizFormPageOut,
)
@router.post("/create")
async def create_quiz(
request: Request, data: Annotated[QuizForm, Depends(json_or_form(QuizForm))]
):
user = require_user(request)
fields = store.quiz_fields(data)
uid, slug = create_content_item(
"quizzes",
"quiz",
user,
fields,
fields["title"],
XP_QUIZ,
"First Quiz",
fields["description"],
None,
request,
)
url = f"/quizzes/{slug}/edit"
return action_result(request, url, data={"uid": uid, "slug": slug, "url": url})
@router.post("/import")
async def import_quiz(
request: Request, data: Annotated[QuizImportForm, Depends(json_or_form(QuizImportForm))]
):
user = require_user(request)
document = data.document
fields = {
"title": document.title.strip(),
"description": document.description.strip(),
"status": "draft",
"published_at": "",
"question_count": 0,
"total_points": 0,
"attempt_count": 0,
**store.document_settings(document),
}
uid, slug = create_content_item(
"quizzes",
"quiz",
user,
fields,
fields["title"],
XP_QUIZ,
"First Quiz",
fields["description"],
None,
request,
)
imported = store.import_questions(uid, document)
audit.record(
request,
"quiz.import",
user=user,
target_type="quiz",
target_uid=uid,
target_label=fields["title"],
summary=f"{user['username']} imported quiz {fields['title']} with {imported} questions",
metadata={"question_count": imported},
links=[audit.target("quiz", uid, fields["title"])],
)
url = f"/quizzes/{slug}/edit"
return action_result(
request,
url,
data={"uid": uid, "slug": slug, "url": url, "question_count": imported},
)
@router.get("/{slug}", response_class=HTMLResponse)
async def quiz_detail(request: Request, slug: str):
user = get_current_user(request)
quiz = load_quiz(slug, user)
redirect = canonical_redirect("quizzes", quiz, slug)
if redirect:
return redirect
detail = load_detail("quizzes", "quiz", quiz["uid"], user)
if not detail:
return action_error(request, "Quiz not found", "/quizzes")
if user:
mark_notifications_read_by_target(user["uid"], resolve_object_url("quiz", quiz["uid"]))
states = store.attempt_states_for(user["uid"], [quiz["uid"]]) if user else {}
state = states.get(quiz["uid"], {})
author = detail["author"]
seo_ctx = quiz_seo(
request,
quiz.get("title") or "Quiz",
quiz.get("description") or "",
robots="index,follow" if quiz.get("status") == "published" else "noindex,nofollow",
seo_target=("quiz", quiz["uid"]),
og_type="article",
breadcrumbs=[{"name": quiz.get("title") or "Quiz", "url": quiz_url(quiz)}],
schemas=[quiz_schema(quiz, author, site_url(request))],
)
context = detail_context(
request,
user,
detail,
"quiz_row",
seo_ctx,
{
"quiz": store.serialize_quiz(quiz, user, author),
"questions": store.serialize_builder(quiz, user)
if store.is_quiz_owner(quiz, user)
else [],
"leaderboard": store.quiz_leaderboard(quiz["uid"]),
"viewer_state": state.get("state", "todo"),
"viewer_attempt_uid": state.get("attempt_uid", ""),
"target_type": "quiz",
"target_uid": quiz["uid"],
},
)
return respond(request, "quiz.html", context, model=QuizDetailOut)
@router.get("/{slug}/export")
async def export_quiz(request: Request, slug: str):
user = get_current_user(request)
quiz = load_quiz(slug, user)
include_answers = store.is_quiz_owner(quiz, user)
document = store.export_document(quiz["uid"], include_answers)
return JSONResponse(QuizDocumentOut.model_validate(document).model_dump(mode="json"))
@router.get("/{slug}/leaderboard")
async def quiz_leaderboard(request: Request, slug: str, limit: int = 25):
user = get_current_user(request)
quiz = load_quiz(slug, user)
return JSONResponse(
QuizLeaderboardOut(
quiz_uid=quiz["uid"],
entries=store.quiz_leaderboard(quiz["uid"], max(1, min(100, int(limit or 25)))),
).model_dump(mode="json")
)
@router.post("/edit/{slug}")
async def edit_quiz(
request: Request, slug: str, data: Annotated[QuizForm, Depends(json_or_form(QuizForm))]
):
user = require_user(request)
quiz = require_editor(load_quiz(slug, user), user)
try:
updated = store.edit_quiz(quiz["uid"], store.settings_update(data))
except QuizError as exc:
return action_error(request, str(exc), quiz_url(quiz))
audit.record(
request,
"quiz.edit",
user=user,
target_type="quiz",
target_uid=quiz["uid"],
target_label=updated.get("title") or quiz["uid"],
summary=f"{user['username']} edited quiz {updated.get('title')}",
links=[audit.target("quiz", quiz["uid"], updated.get("title") or "")],
)
await notify_quiz(quiz["uid"])
url = quiz_url(updated)
return action_result(request, url, data={"uid": quiz["uid"], "url": url})
@router.post("/delete/{slug}")
async def delete_quiz(request: Request, slug: str):
user = require_user(request)
response = delete_content_item(request, "quizzes", "quiz", user, slug, "/quizzes")
store.clear_cache()
return response
@router.post("/{slug}/publish")
async def publish_quiz(request: Request, slug: str):
user = require_user(request)
quiz = require_editor(load_quiz(slug, user), user)
try:
published = store.publish_quiz(quiz["uid"])
except QuizError as exc:
return action_error(request, str(exc), f"/quizzes/{slug}/edit")
if published.get("publish_won"):
award_rewards(user["uid"], XP_QUIZ_PUBLISH, "Quiz Author")
track_action(user["uid"], "quiz_publish")
audit.record(
request,
"quiz.publish",
user=user,
target_type="quiz",
target_uid=quiz["uid"],
target_label=published.get("title") or quiz["uid"],
summary=f"{user['username']} published quiz {published.get('title')}",
links=[audit.target("quiz", quiz["uid"], published.get("title") or "")],
)
store.clear_cache()
await notify_quiz(quiz["uid"])
url = quiz_url(published)
return action_result(
request, url, data={"uid": quiz["uid"], "url": url, "status": published.get("status")}
)