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 []) ]