193 lines
6.7 KiB
Python
193 lines
6.7 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import re
|
||
|
|
import secrets
|
||
|
|
from collections import defaultdict
|
||
|
|
|
||
|
|
from sqlalchemy import select
|
||
|
|
from sqlalchemy.orm import Session, selectinload
|
||
|
|
|
||
|
|
from app.models import Favourite, Question, Quiz, QuizAttempt, QuizVote, User
|
||
|
|
from app.schemas import (
|
||
|
|
GlobalBestScoreOut,
|
||
|
|
QuestionOptionSchema,
|
||
|
|
QuizSummaryOut,
|
||
|
|
ScoreSummaryOut,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def normalize_write_in(text: str) -> str:
|
||
|
|
return " ".join(text.strip().lower().split())
|
||
|
|
|
||
|
|
|
||
|
|
def generate_share_slug(db: Session, title: str) -> str:
|
||
|
|
stem = re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-")[:22] or "quiz"
|
||
|
|
while True:
|
||
|
|
token = secrets.token_hex(3)
|
||
|
|
slug = f"{stem}-{token}"
|
||
|
|
exists = db.scalar(select(Quiz.id).where(Quiz.share_slug == slug))
|
||
|
|
if exists is None:
|
||
|
|
return slug
|
||
|
|
|
||
|
|
|
||
|
|
def compute_quiz_mode(questions: list[Question]) -> str:
|
||
|
|
if not questions:
|
||
|
|
return "empty"
|
||
|
|
kinds = {q.question_type for q in questions}
|
||
|
|
if kinds == {"multiple_choice"}:
|
||
|
|
return "multiple_choice"
|
||
|
|
if kinds == {"write_in"}:
|
||
|
|
return "write_in"
|
||
|
|
return "mixed"
|
||
|
|
|
||
|
|
|
||
|
|
def grade_question(question: Question, submitted_raw: str) -> tuple[bool, str]:
|
||
|
|
submitted = (submitted_raw or "").strip()
|
||
|
|
if question.question_type == "multiple_choice":
|
||
|
|
option_lookup = {
|
||
|
|
str(opt.get("id", "")): str(opt.get("text", ""))
|
||
|
|
for opt in (question.options or [])
|
||
|
|
}
|
||
|
|
correct_display = option_lookup.get(question.correct_answer, question.correct_answer)
|
||
|
|
is_correct = submitted == question.correct_answer
|
||
|
|
return is_correct, correct_display
|
||
|
|
|
||
|
|
accepted = {normalize_write_in(question.correct_answer)}
|
||
|
|
for alias in question.accepted_answers or []:
|
||
|
|
if alias and alias.strip():
|
||
|
|
accepted.add(normalize_write_in(alias))
|
||
|
|
is_correct = bool(submitted) and (normalize_write_in(submitted) in accepted)
|
||
|
|
return is_correct, question.correct_answer
|
||
|
|
|
||
|
|
|
||
|
|
def enrich_quizzes(
|
||
|
|
db: Session,
|
||
|
|
quizzes: list[Quiz],
|
||
|
|
current_user: User | None,
|
||
|
|
) -> list[QuizSummaryOut]:
|
||
|
|
if not quizzes:
|
||
|
|
return []
|
||
|
|
|
||
|
|
quiz_ids = [q.id for q in quizzes]
|
||
|
|
|
||
|
|
# Votes aggregation
|
||
|
|
votes = db.scalars(select(QuizVote).where(QuizVote.quiz_id.in_(quiz_ids))).all()
|
||
|
|
upvotes_map: dict[int, int] = defaultdict(int)
|
||
|
|
downvotes_map: dict[int, int] = defaultdict(int)
|
||
|
|
user_vote_map: dict[int, int] = defaultdict(int)
|
||
|
|
for vote in votes:
|
||
|
|
if vote.value > 0:
|
||
|
|
upvotes_map[vote.quiz_id] += 1
|
||
|
|
elif vote.value < 0:
|
||
|
|
downvotes_map[vote.quiz_id] += 1
|
||
|
|
if current_user and vote.user_id == current_user.id:
|
||
|
|
user_vote_map[vote.quiz_id] = vote.value
|
||
|
|
|
||
|
|
# Attempts aggregation (global best + current user latest & best)
|
||
|
|
attempts = db.scalars(
|
||
|
|
select(QuizAttempt)
|
||
|
|
.options(selectinload(QuizAttempt.user))
|
||
|
|
.where(QuizAttempt.quiz_id.in_(quiz_ids))
|
||
|
|
.order_by(QuizAttempt.completed_at.asc(), QuizAttempt.id.asc())
|
||
|
|
).all()
|
||
|
|
|
||
|
|
global_best_map: dict[int, QuizAttempt] = {}
|
||
|
|
my_latest_map: dict[int, QuizAttempt] = {}
|
||
|
|
my_best_map: dict[int, QuizAttempt] = {}
|
||
|
|
|
||
|
|
for att in attempts:
|
||
|
|
# Global best: strictly greater score_percent (ties go to earliest achiever)
|
||
|
|
prev_global = global_best_map.get(att.quiz_id)
|
||
|
|
if prev_global is None or att.score_percent > prev_global.score_percent:
|
||
|
|
global_best_map[att.quiz_id] = att
|
||
|
|
|
||
|
|
if current_user and att.user_id == current_user.id:
|
||
|
|
# Ordered ascending by completed_at, id -> last seen is latest
|
||
|
|
my_latest_map[att.quiz_id] = att
|
||
|
|
prev_best = my_best_map.get(att.quiz_id)
|
||
|
|
if prev_best is None or att.score_percent > prev_best.score_percent:
|
||
|
|
my_best_map[att.quiz_id] = att
|
||
|
|
|
||
|
|
# Favourites lookup
|
||
|
|
fav_quiz_ids: set[int] = set()
|
||
|
|
if current_user:
|
||
|
|
fav_rows = db.scalars(
|
||
|
|
select(Favourite.quiz_id).where(
|
||
|
|
Favourite.user_id == current_user.id,
|
||
|
|
Favourite.quiz_id.in_(quiz_ids),
|
||
|
|
)
|
||
|
|
).all()
|
||
|
|
fav_quiz_ids = set(fav_rows)
|
||
|
|
|
||
|
|
summaries: list[QuizSummaryOut] = []
|
||
|
|
for q in quizzes:
|
||
|
|
gb = global_best_map.get(q.id)
|
||
|
|
ml = my_latest_map.get(q.id)
|
||
|
|
mb = my_best_map.get(q.id)
|
||
|
|
|
||
|
|
global_best_out = (
|
||
|
|
GlobalBestScoreOut(
|
||
|
|
score_count=gb.score_count,
|
||
|
|
total_questions=gb.total_questions,
|
||
|
|
score_percent=round(gb.score_percent, 1),
|
||
|
|
holder_nickname=gb.user.nickname if gb.user else "student",
|
||
|
|
holder_avatar_animal=gb.user.avatar_animal if gb.user else "fox",
|
||
|
|
holder_avatar_color=gb.user.avatar_color if gb.user else "avatar-1",
|
||
|
|
)
|
||
|
|
if gb is not None
|
||
|
|
else None
|
||
|
|
)
|
||
|
|
my_latest_out = (
|
||
|
|
ScoreSummaryOut(
|
||
|
|
score_count=ml.score_count,
|
||
|
|
total_questions=ml.total_questions,
|
||
|
|
score_percent=round(ml.score_percent, 1),
|
||
|
|
completed_at=ml.completed_at,
|
||
|
|
)
|
||
|
|
if ml is not None
|
||
|
|
else None
|
||
|
|
)
|
||
|
|
my_best_out = (
|
||
|
|
ScoreSummaryOut(
|
||
|
|
score_count=mb.score_count,
|
||
|
|
total_questions=mb.total_questions,
|
||
|
|
score_percent=round(mb.score_percent, 1),
|
||
|
|
completed_at=mb.completed_at,
|
||
|
|
)
|
||
|
|
if mb is not None
|
||
|
|
else None
|
||
|
|
)
|
||
|
|
|
||
|
|
summaries.append(
|
||
|
|
QuizSummaryOut(
|
||
|
|
id=q.id,
|
||
|
|
share_slug=q.share_slug,
|
||
|
|
title=q.title,
|
||
|
|
description=q.description,
|
||
|
|
is_public=q.is_public,
|
||
|
|
author_id=q.author_id,
|
||
|
|
author_nickname=q.author.nickname if q.author else "student",
|
||
|
|
author_avatar_animal=q.author.avatar_animal if q.author else "fox",
|
||
|
|
author_avatar_color=q.author.avatar_color if q.author else "avatar-1",
|
||
|
|
question_count=len(q.questions),
|
||
|
|
quiz_mode=compute_quiz_mode(q.questions), # type: ignore[arg-type]
|
||
|
|
upvotes=upvotes_map[q.id],
|
||
|
|
downvotes=downvotes_map[q.id],
|
||
|
|
user_vote=user_vote_map[q.id],
|
||
|
|
my_latest_score=my_latest_out,
|
||
|
|
my_best_score=my_best_out,
|
||
|
|
global_best_score=global_best_out,
|
||
|
|
is_favourited=(q.id in fav_quiz_ids),
|
||
|
|
created_at=q.created_at,
|
||
|
|
updated_at=q.updated_at,
|
||
|
|
)
|
||
|
|
)
|
||
|
|
return summaries
|
||
|
|
|
||
|
|
|
||
|
|
def serialize_play_options(options_raw: list[dict]) -> list[QuestionOptionSchema]:
|
||
|
|
return [
|
||
|
|
QuestionOptionSchema(id=str(o.get("id", "")), text=str(o.get("text", "")))
|
||
|
|
for o in (options_raw or [])
|
||
|
|
]
|