import secrets from datetime import timedelta from decimal import Decimal from asgiref.sync import async_to_sync from channels.layers import get_channel_layer from django.db import transaction from django.db.models import Q from django.utils import timezone from rest_framework.exceptions import ValidationError from accounts.models import User from .models import ( CheatFlag, Contest, ContestAnswer, ContestAttempt, MathGameAttempt, RatingHistory, RealtimeMatch, ) CHALLENGE_CODE_ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" WAITING_MATCH_TTL = timedelta(minutes=10) ATTEMPT_QUESTION_COUNT = 8 def _broadcast_match(match_id, reason): channel_layer = get_channel_layer() if channel_layer is None: return async_to_sync(channel_layer.group_send)( f"match_{match_id}", { "type": "match.state", "reason": reason, }, ) def notify_match_on_commit(match_id, reason): transaction.on_commit(lambda: _broadcast_match(match_id, reason)) def _new_challenge_code(): for _ in range(20): code = "".join(secrets.choice(CHALLENGE_CODE_ALPHABET) for _ in range(6)) if not RealtimeMatch.objects.filter(challenge_code=code).exists(): return code raise ValidationError("暂时无法生成联机码，请稍后重试") def _validate_realtime_contest(contest): if contest.kind != Contest.Kind.REALTIME or contest.status != Contest.Status.PUBLISHED: raise ValidationError("实时比赛不可用") def _validate_game_kind(game_kind): value = str(game_kind or RealtimeMatch.GameKind.QUIZ) if value not in RealtimeMatch.GameKind.values: raise ValidationError({"game_kind": "不支持的实时比赛玩法"}) return value def _cancel_expired_waiting_matches(): RealtimeMatch.objects.filter(status=RealtimeMatch.Status.WAITING).filter( Q(expires_at__isnull=True) | Q(expires_at__lte=timezone.now()) ).update(status=RealtimeMatch.Status.CANCELLED) def _active_match_for(user): return ( RealtimeMatch.objects.filter( Q(player_one=user) | Q(player_two=user), status=RealtimeMatch.Status.ACTIVE, ) .select_related("contest", "player_one", "player_two") .first() ) def _cancel_other_waiting_matches(user, match_type, game_kind): matches = list( RealtimeMatch.objects.filter( player_one=user, status=RealtimeMatch.Status.WAITING, ) .exclude(match_type=match_type, game_kind=game_kind) .values_list("id", flat=True) ) if matches: RealtimeMatch.objects.filter(id__in=matches).update( status=RealtimeMatch.Status.CANCELLED ) for match_id in matches: notify_match_on_commit(match_id, "cancelled") def _select_question_order(contest): question_ids = list( contest.contest_questions.filter(question_version__question__is_active=True) .order_by("id") .values_list("id", flat=True) ) if not question_ids: raise ValidationError("比赛题库为空") count = min(ATTEMPT_QUESTION_COUNT, len(question_ids)) return secrets.SystemRandom().sample(question_ids, count) def _attempt_questions(attempt): queryset = attempt.contest.contest_questions.select_related("question_version") if attempt.question_order: items = {item.id: item for item in queryset.filter(id__in=attempt.question_order)} return [items[item_id] for item_id in attempt.question_order if item_id in items] return list(queryset.all()) def _activate_match(match, user): from .game_services import build_game now = timezone.now() match.player_two = user match.player_two_rating = user.rating match.status = RealtimeMatch.Status.ACTIVE match.started_at = now match.save( update_fields=["player_two", "player_two_rating", "status", "started_at"] ) if match.game_kind == RealtimeMatch.GameKind.QUIZ: question_order = _select_question_order(match.contest) ContestAttempt.objects.bulk_create( [ ContestAttempt( contest=match.contest, user=match.player_one, match=match, question_order=question_order, ), ContestAttempt( contest=match.contest, user=user, match=match, question_order=question_order, ), ] ) match.attempts.update(started_at=now) else: puzzle, solution = build_game( match.game_kind, MathGameAttempt.Difficulty.STANDARD, ) MathGameAttempt.objects.bulk_create( [ MathGameAttempt( user=match.player_one, match=match, kind=match.game_kind, puzzle=puzzle, solution=solution, ), MathGameAttempt( user=user, match=match, kind=match.game_kind, puzzle=puzzle, solution=solution, ), ] ) match.game_attempts.update(started_at=now) notify_match_on_commit(match.id, "matched") return match def normalize_answer(value): text = str(value).strip().lower().replace(" ", "") try: return str(Decimal(text).normalize()) except Exception: return text def attempt_payload(attempt, include_results=False): questions = [] answers = {answer.contest_question_id: answer for answer in attempt.answers.all()} for display_order, item in enumerate(_attempt_questions(attempt), start=1): question = { "order": display_order, "prompt": item.question_version.prompt, "metadata": item.question_version.metadata, "points": item.points, } if include_results: answer = answers.get(item.id) question.update( { "submitted_answer": answer.submitted_answer if answer else "", "is_correct": answer.is_correct if answer else False, "correct_answer": item.question_version.answer, "explanation": item.question_version.explanation, } ) questions.append(question) return { "attempt_id": attempt.id, "match_id": attempt.match_id, "contest": attempt.contest.title, "kind": attempt.contest.kind, "status": attempt.status, "duration_seconds": attempt.contest.duration_seconds, "server_started_at": attempt.started_at, "score": attempt.score, "correct_count": attempt.correct_count, "answer_count": attempt.answer_count, "duration_ms": attempt.duration_ms, "questions": questions, } @transaction.atomic def start_attempt(user, contest): now = timezone.now() if contest.status != Contest.Status.PUBLISHED: raise ValidationError("比赛尚未发布") if contest.starts_at and contest.starts_at > now: raise ValidationError("比赛尚未开始") if contest.ends_at and contest.ends_at <= now: raise ValidationError("比赛已经结束") if contest.kind == Contest.Kind.DAILY: existing = ContestAttempt.objects.filter(user=user, contest=contest).first() if existing: return attempt_payload( existing, include_results=existing.status != ContestAttempt.Status.ACTIVE, ) attempt = ContestAttempt.objects.create( contest=contest, user=user, question_order=_select_question_order(contest), ) return attempt_payload(attempt) @transaction.atomic def submit_attempt(user, attempt_id, raw_answers, submission_key): attempt = ( ContestAttempt.objects.select_for_update() .select_related("contest") .get(id=attempt_id, user=user) ) if attempt.status != ContestAttempt.Status.ACTIVE: if submission_key and attempt.submission_key == submission_key: include_results = not attempt.match_id or ( attempt.match.status == RealtimeMatch.Status.COMPLETED ) return attempt_payload(attempt, include_results=include_results) raise ValidationError("该答题记录已经结算") if not submission_key: raise ValidationError({"Idempotency-Key": "正式提交必须提供幂等键"}) now = timezone.now() duration_ms = max(0, int((now - attempt.started_at).total_seconds() * 1000)) limit_ms = attempt.contest.duration_seconds * 1000 items = _attempt_questions(attempt) if not isinstance(raw_answers, list): raise ValidationError({"answers": "答案必须是数组"}) by_order = {} try: for item in raw_answers: order = int(item["order"]) if order <= 0 or order in by_order: raise ValueError by_order[order] = item.get("answer", "") except (KeyError, TypeError, ValueError) as exc: raise ValidationError({"answers": "答案题号无效或重复"}) from exc score = 0 correct_count = 0 for display_order, contest_question in enumerate(items, start=1): submitted = str(by_order.get(display_order, ""))[:200] correct = normalize_answer(submitted) == normalize_answer( contest_question.question_version.answer ) if correct and duration_ms <= limit_ms: score += contest_question.points correct_count += 1 ContestAnswer.objects.create( attempt=attempt, contest_question=contest_question, submitted_answer=submitted, is_correct=correct and duration_ms <= limit_ms, elapsed_ms=min(duration_ms, limit_ms + 60_000), ) attempt.status = ( ContestAttempt.Status.SUBMITTED if duration_ms <= limit_ms else ContestAttempt.Status.EXPIRED ) attempt.score = score attempt.correct_count = correct_count attempt.answer_count = len(raw_answers) attempt.duration_ms = duration_ms attempt.submission_key = submission_key attempt.submitted_at = now attempt.save() if raw_answers and duration_ms / len(raw_answers) < 150: CheatFlag.objects.create( attempt=attempt, reason="extreme_answer_speed", evidence={ "duration_ms": duration_ms, "answer_count": len(raw_answers), }, ) if attempt.match_id: match = finalize_match(attempt.match_id) if match.status != RealtimeMatch.Status.COMPLETED: notify_match_on_commit(attempt.match_id, "submitted") attempt.match.refresh_from_db() return attempt_payload( attempt, include_results=attempt.match.status == RealtimeMatch.Status.COMPLETED, ) return attempt_payload(attempt, include_results=True) @transaction.atomic def find_match(user, contest, game_kind=RealtimeMatch.GameKind.QUIZ): _validate_realtime_contest(contest) game_kind = _validate_game_kind(game_kind) _cancel_expired_waiting_matches() active = _active_match_for(user) if active: if active.contest_id == contest.id and active.game_kind == game_kind: return active raise ValidationError("你已有一场进行中的实时比赛") _cancel_other_waiting_matches( user, RealtimeMatch.MatchType.RANDOM, game_kind, ) existing = ( RealtimeMatch.objects.filter( Q(player_one=user) | Q(player_two=user), contest=contest, match_type=RealtimeMatch.MatchType.RANDOM, game_kind=game_kind, status__in=[RealtimeMatch.Status.WAITING, RealtimeMatch.Status.ACTIVE], ).filter(Q(status=RealtimeMatch.Status.ACTIVE) | Q(expires_at__gt=timezone.now())) .order_by("-created_at") .first() ) if existing: return existing waiting = ( RealtimeMatch.objects.select_for_update(skip_locked=True) .filter( contest=contest, match_type=RealtimeMatch.MatchType.RANDOM, game_kind=game_kind, status=RealtimeMatch.Status.WAITING, expires_at__gt=timezone.now(), player_one_rating__gte=max(0, user.rating - 300), player_one_rating__lte=user.rating + 300, ) .exclude(player_one=user) .order_by("created_at") .first() ) if waiting is None: return RealtimeMatch.objects.create( contest=contest, match_type=RealtimeMatch.MatchType.RANDOM, game_kind=game_kind, player_one=user, player_one_rating=user.rating, expires_at=timezone.now() + WAITING_MATCH_TTL, ) return _activate_match(waiting, user) @transaction.atomic def create_challenge(user, contest, game_kind=RealtimeMatch.GameKind.QUIZ): _validate_realtime_contest(contest) game_kind = _validate_game_kind(game_kind) _cancel_expired_waiting_matches() active = _active_match_for(user) if active: raise ValidationError("你已有一场进行中的实时比赛") _cancel_other_waiting_matches( user, RealtimeMatch.MatchType.CHALLENGE, game_kind, ) existing = ( RealtimeMatch.objects.filter( player_one=user, contest=contest, match_type=RealtimeMatch.MatchType.CHALLENGE, game_kind=game_kind, status=RealtimeMatch.Status.WAITING, expires_at__gt=timezone.now(), ) .order_by("-created_at") .first() ) if existing: return existing return RealtimeMatch.objects.create( contest=contest, match_type=RealtimeMatch.MatchType.CHALLENGE, game_kind=game_kind, challenge_code=_new_challenge_code(), player_one=user, player_one_rating=user.rating, expires_at=timezone.now() + WAITING_MATCH_TTL, ) @transaction.atomic def join_challenge(user, challenge_code): _cancel_expired_waiting_matches() code = str(challenge_code or "").strip().upper() if len(code) != 6 or any(character not in CHALLENGE_CODE_ALPHABET for character in code): raise ValidationError({"challenge_code": "联机码应为 6 位大写字母或数字"}) try: match = ( RealtimeMatch.objects.select_for_update() .select_related("contest", "player_one") .get( challenge_code=code, match_type=RealtimeMatch.MatchType.CHALLENGE, ) ) except RealtimeMatch.DoesNotExist as exc: raise ValidationError({"challenge_code": "联机码不存在"}) from exc if match.player_one_id == user.id: raise ValidationError({"challenge_code": "不能加入自己创建的约战"}) if match.status != RealtimeMatch.Status.WAITING or ( match.expires_at and match.expires_at <= timezone.now() ): raise ValidationError({"challenge_code": "联机码已失效或已被使用"}) active = _active_match_for(user) if active and active.id != match.id: raise ValidationError("你已有一场进行中的实时比赛") own_waiting_ids = list( RealtimeMatch.objects.filter( player_one=user, status=RealtimeMatch.Status.WAITING, ) .exclude(id=match.id) .values_list("id", flat=True) ) if own_waiting_ids: RealtimeMatch.objects.filter(id__in=own_waiting_ids).update( status=RealtimeMatch.Status.CANCELLED ) for match_id in own_waiting_ids: notify_match_on_commit(match_id, "cancelled") return _activate_match(match, user) @transaction.atomic def cancel_waiting_match(user, match_id): match = RealtimeMatch.objects.select_for_update().get(id=match_id) if match.player_one_id != user.id: raise ValidationError("只有创建者可以取消等待") if match.status != RealtimeMatch.Status.WAITING: raise ValidationError("只能取消等待中的比赛") match.status = RealtimeMatch.Status.CANCELLED match.save(update_fields=["status"]) notify_match_on_commit(match.id, "cancelled") return match def match_payload(match, user): from .game_services import game_payload reveal_results = match.status == RealtimeMatch.Status.COMPLETED if match.game_kind == RealtimeMatch.GameKind.QUIZ: attempts = { attempt.user_id: attempt for attempt in match.attempts.select_related("user", "contest").all() } else: attempts = { attempt.user_id: attempt for attempt in match.game_attempts.select_related("user").all() } attempt = attempts.get(user.id) opponent = match.player_two if match.player_one_id == user.id else match.player_one opponent_attempt = attempts.get(opponent.id) if opponent else None rating_change = ( match.rating_changes.filter(user=user).values("delta", "rating_after").first() if reveal_results else None ) attempt_data = None if attempt: attempt_data = ( attempt_payload(attempt, include_results=reveal_results) if match.game_kind == RealtimeMatch.GameKind.QUIZ else game_payload(attempt) ) return { "match_id": match.id, "match_type": match.match_type, "game_kind": match.game_kind, "game_label": match.get_game_kind_display(), "is_owner": match.player_one_id == user.id, "challenge_code": ( match.challenge_code if match.match_type == RealtimeMatch.MatchType.CHALLENGE and match.status == RealtimeMatch.Status.WAITING else None ), "status": match.status, "contest": ( match.contest.title if match.game_kind == RealtimeMatch.GameKind.QUIZ else match.get_game_kind_display() ), "duration_seconds": match.contest.duration_seconds, "expires_at": match.expires_at, "started_at": match.started_at, "opponent": ( { "nickname": opponent.nickname, "rating": opponent.rating, "status": opponent_attempt.status if opponent_attempt else None, "score": opponent_attempt.score if reveal_results and opponent_attempt else None, "correct_count": ( opponent_attempt.correct_count if reveal_results and opponent_attempt and match.game_kind == RealtimeMatch.GameKind.QUIZ else None ), "duration_ms": ( opponent_attempt.duration_ms if reveal_results and opponent_attempt else None ), } if opponent else None ), "attempt": attempt_data, "result": ( { "winner": ( "draw" if match.winner_id is None else "self" if match.winner_id == user.id else "opponent" ), "rating_delta": rating_change["delta"] if rating_change else 0, "rating_after": rating_change["rating_after"] if rating_change else user.rating, } if reveal_results else None ), "websocket_path": f"/ws/v1/contest/matches/{match.id}/", } @transaction.atomic def refresh_match_state(match_id): match = ( RealtimeMatch.objects.select_for_update() .select_related("contest", "player_one", "player_two") .get(id=match_id) ) if ( match.status == RealtimeMatch.Status.WAITING and match.expires_at and match.expires_at <= timezone.now() ): match.status = RealtimeMatch.Status.CANCELLED match.save(update_fields=["status"]) notify_match_on_commit(match.id, "expired") elif match.status == RealtimeMatch.Status.ACTIVE and match.started_at: deadline = match.started_at + timedelta(seconds=match.contest.duration_seconds) if timezone.now() >= deadline: now = timezone.now() if match.game_kind == RealtimeMatch.GameKind.QUIZ: for attempt in match.attempts.filter( status=ContestAttempt.Status.ACTIVE ): elapsed_ms = max( 0, int((now - attempt.started_at).total_seconds() * 1000), ) attempt.status = ContestAttempt.Status.EXPIRED attempt.duration_ms = elapsed_ms attempt.submitted_at = now attempt.save( update_fields=["status", "duration_ms", "submitted_at"] ) match = finalize_match(match.id) else: elapsed_ms = max( 0, int((now - match.started_at).total_seconds() * 1000), ) match.game_attempts.filter( status=MathGameAttempt.Status.ACTIVE ).update( status=MathGameAttempt.Status.FAILED, duration_ms=elapsed_ms, submitted_at=now, ) match = finalize_game_match(match.id) return match def _elo_delta(rating, opponent_rating, score, k=32): expected = 1 / (1 + 10 ** ((opponent_rating - rating) / 400)) return round(k * (score - expected)) def _settle_match(match, first_result, second_result, winner_id): users = { user.id: user for user in User.objects.select_for_update().filter( id__in=[match.player_one_id, match.player_two_id] ) } player_one = users[match.player_one_id] player_two = users[match.player_two_id] deltas = ( _elo_delta(player_one.rating, player_two.rating, first_result), _elo_delta(player_two.rating, player_one.rating, second_result), ) for user, delta in zip((player_one, player_two), deltas): before = user.rating user.rating = max(0, before + delta) user.save(update_fields=["rating"]) RatingHistory.objects.create( user=user, match=match, rating_before=before, rating_after=user.rating, delta=delta, ) match.winner_id = winner_id match.status = RealtimeMatch.Status.COMPLETED match.completed_at = timezone.now() match.save(update_fields=["winner", "status", "completed_at"]) notify_match_on_commit(match.id, "completed") return match @transaction.atomic def finalize_match(match_id): match = RealtimeMatch.objects.select_for_update().get(id=match_id) if match.status != RealtimeMatch.Status.ACTIVE: return match attempts = list(match.attempts.select_related("user").order_by("user_id")) if len(attempts) != 2 or any( attempt.status == ContestAttempt.Status.ACTIVE for attempt in attempts ): return match first = next(item for item in attempts if item.user_id == match.player_one_id) second = next(item for item in attempts if item.user_id == match.player_two_id) if first.score > second.score: first_result, second_result = 1.0, 0.0 winner_id = first.user_id elif second.score > first.score: first_result, second_result = 0.0, 1.0 winner_id = second.user_id else: diff = first.duration_ms - second.duration_ms if abs(diff) <= 100: first_result = second_result = 0.5 winner_id = None elif diff < 0: first_result, second_result = 1.0, 0.0 winner_id = first.user_id else: first_result, second_result = 0.0, 1.0 winner_id = second.user_id return _settle_match( match, first_result, second_result, winner_id, ) @transaction.atomic def finalize_game_match(match_id): match = RealtimeMatch.objects.select_for_update().get(id=match_id) if match.status != RealtimeMatch.Status.ACTIVE: return match attempts = list(match.game_attempts.select_related("user").order_by("user_id")) if len(attempts) != 2 or any( attempt.status == MathGameAttempt.Status.ACTIVE for attempt in attempts ): return match first = next(item for item in attempts if item.user_id == match.player_one_id) second = next(item for item in attempts if item.user_id == match.player_two_id) first_completed = first.status == MathGameAttempt.Status.COMPLETED second_completed = second.status == MathGameAttempt.Status.COMPLETED if first_completed and not second_completed: first_result, second_result, winner_id = 1.0, 0.0, first.user_id elif second_completed and not first_completed: first_result, second_result, winner_id = 0.0, 1.0, second.user_id elif not first_completed and not second_completed: first_result = second_result = 0.5 winner_id = None else: diff = first.duration_ms - second.duration_ms if abs(diff) <= 100: first_result = second_result = 0.5 winner_id = None elif diff < 0: first_result, second_result, winner_id = 1.0, 0.0, first.user_id else: first_result, second_result, winner_id = 0.0, 1.0, second.user_id return _settle_match( match, first_result, second_result, winner_id, )