yuetsh
4ebcecfab4
这些是做 Bun 重写时逐行对照发现的，旧后端现在还在线上跑，学生今天就在踩。 每条都在本地起 Django + postgres 实跑验证过，**正反两面都验**：越权要挡住、 正常操作要还能用。合计 14 条断言全过。 ## 越权（跨教师） **ACMContestHelper.put** — 只按 pk 取 rank，任何老师都能改**别人比赛**里的 检查标记。而 `contest_id` 明明在 ACMContesHelperSerializer 里、客户端一直在传， 视图完全没用它。改成按 (pk, contest_id) 取 + ensure_created_by。 同类的 get 本来就有 ensure_created_by，是 put 漏了。 **AddContestProblemAPI.post** — 一个权限装饰器都没有，导致两件事： - 少 problem_permission_required：problem_permission=None 的学生管理员也能建题 - 少 ensure_created_by：任何管理员都能往**别人的比赛**里塞题 另外源题只按 id 取，能把别人比赛里的赛题（或别人未公开的草稿）拷进自己的比赛， 连 answers 参考答案一起拿到。改成源题必须是公开题库题，不可见的要求归属。 （中间件 AdminRoleRequiredMiddleware 只保证"是管理员身份"，挡不住这些。） ## 会 500 **ProblemVisibleAPI.put** — `self.error(...)` 漏写 return，题不存在时继续执行到 未赋值的 problem，抛 UnboundLocalError → 500。 ## 数据正确性 **ProblemSetAdminAPI.get** — 后台列表写死 filter(visible=True)，而列表里就有 可见性开关：一旦关掉，题单从后台消失，教师再也找不到、也无法在界面上打开回来。 后台列表不该过滤 visible，前台该。 **增删题单题目不重算进度** — 题目数是进度的分母。不重算的话学生进度百分比全是 虚高的，原本标"已完成"的人会一直是已完成。更麻烦的是那个手动同步接口 （ProblemSetSyncAPI）前端没有调用点，等于**界面上根本没办法修**。 改成增删之后直接 sync_all_progress_for_problemset。 **移除用户留下孤儿奖章** — 只删了 ProblemSetProgress，UserBadge 还在。学生个人页 会一直挂着一个自己已不在的题单的奖章，而且重新加入时因唯一约束拿不到第二次。 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
306 lines
12 KiB
Python
306 lines
12 KiB
Python
import copy
|
||
import os
|
||
import zipfile
|
||
from datetime import datetime, timedelta
|
||
from ipaddress import ip_network
|
||
|
||
from django.http import FileResponse
|
||
from django.utils.timezone import now
|
||
|
||
from account.decorators import ensure_created_by, super_admin_required, teacher_admin_required
|
||
from account.models import User
|
||
from problem.models import Problem
|
||
from submission.models import JudgeStatus, Submission
|
||
from utils.api import APIView, validate_serializer
|
||
from utils.shortcuts import rand_str
|
||
from utils.tasks import delete_files
|
||
|
||
from ..models import ACMContestRank, Contest, ContestAnnouncement
|
||
from ..serializers import (
|
||
ACMContesHelperSerializer,
|
||
ContestAdminSerializer,
|
||
ContestAnnouncementSerializer,
|
||
ContestCloneSerializer,
|
||
CreateConetestSeriaizer,
|
||
CreateContestAnnouncementSerializer,
|
||
EditConetestSeriaizer,
|
||
EditContestAnnouncementSerializer,
|
||
)
|
||
|
||
|
||
class ContestAPI(APIView):
|
||
@validate_serializer(CreateConetestSeriaizer)
|
||
@teacher_admin_required
|
||
def post(self, request):
|
||
data = request.data
|
||
# DRF 的 DateTimeField 已经校验并规范化过，这里拿到的必定是 ISO 8601 字符串
|
||
data["start_time"] = datetime.fromisoformat(data["start_time"])
|
||
data["end_time"] = datetime.fromisoformat(data["end_time"])
|
||
data["created_by"] = request.user
|
||
if data["end_time"] <= data["start_time"]:
|
||
return self.error("Start time must occur earlier than end time")
|
||
if data.get("password") and data["password"] == "":
|
||
data["password"] = None
|
||
for ip_range in data["allowed_ip_ranges"]:
|
||
try:
|
||
ip_network(ip_range, strict=False)
|
||
except ValueError:
|
||
return self.error(f"{ip_range} is not a valid cidr network")
|
||
contest = Contest.objects.create(**data)
|
||
return self.success(ContestAdminSerializer(contest).data)
|
||
|
||
@validate_serializer(EditConetestSeriaizer)
|
||
@teacher_admin_required
|
||
def put(self, request):
|
||
data = request.data
|
||
try:
|
||
contest = Contest.objects.get(id=data.pop("id"))
|
||
except Contest.DoesNotExist:
|
||
return self.error("Contest does not exist")
|
||
ensure_created_by(contest, request.user)
|
||
# DRF 的 DateTimeField 已经校验并规范化过，这里拿到的必定是 ISO 8601 字符串
|
||
data["start_time"] = datetime.fromisoformat(data["start_time"])
|
||
data["end_time"] = datetime.fromisoformat(data["end_time"])
|
||
if data["end_time"] <= data["start_time"]:
|
||
return self.error("Start time must occur earlier than end time")
|
||
if not data["password"]:
|
||
data["password"] = None
|
||
for ip_range in data["allowed_ip_ranges"]:
|
||
try:
|
||
ip_network(ip_range, strict=False)
|
||
except ValueError:
|
||
return self.error(f"{ip_range} is not a valid cidr network")
|
||
for k, v in data.items():
|
||
setattr(contest, k, v)
|
||
contest.save()
|
||
return self.success(ContestAdminSerializer(contest).data)
|
||
|
||
@teacher_admin_required
|
||
def get(self, request):
|
||
contest_id = request.GET.get("id")
|
||
if contest_id:
|
||
try:
|
||
contest = Contest.objects.get(id=contest_id)
|
||
ensure_created_by(contest, request.user)
|
||
return self.success(ContestAdminSerializer(contest).data)
|
||
except Contest.DoesNotExist:
|
||
return self.error("Contest does not exist")
|
||
|
||
contests = Contest.objects.all().order_by("-create_time")
|
||
if not request.user.is_super_admin():
|
||
contests = contests.filter(created_by=request.user)
|
||
|
||
keyword = request.GET.get("keyword")
|
||
if keyword:
|
||
contests = contests.filter(title__contains=keyword)
|
||
return self.success(self.paginate_data(request, contests, ContestAdminSerializer))
|
||
|
||
|
||
# DEPRECATED: 前端未调用 (2026-05-26)
|
||
class ContestAnnouncementAPI(APIView):
|
||
@validate_serializer(CreateContestAnnouncementSerializer)
|
||
@super_admin_required
|
||
def post(self, request):
|
||
"""
|
||
Create one contest_announcement.
|
||
"""
|
||
data = request.data
|
||
try:
|
||
contest = Contest.objects.get(id=data.pop("contest_id"))
|
||
data["contest"] = contest
|
||
data["created_by"] = request.user
|
||
except Contest.DoesNotExist:
|
||
return self.error("Contest does not exist")
|
||
announcement = ContestAnnouncement.objects.create(**data)
|
||
return self.success(ContestAnnouncementSerializer(announcement).data)
|
||
|
||
@validate_serializer(EditContestAnnouncementSerializer)
|
||
@super_admin_required
|
||
def put(self, request):
|
||
"""
|
||
update contest_announcement
|
||
"""
|
||
data = request.data
|
||
try:
|
||
contest_announcement = ContestAnnouncement.objects.get(id=data.pop("id"))
|
||
except ContestAnnouncement.DoesNotExist:
|
||
return self.error("Contest announcement does not exist")
|
||
for k, v in data.items():
|
||
setattr(contest_announcement, k, v)
|
||
contest_announcement.save()
|
||
return self.success()
|
||
|
||
@super_admin_required
|
||
def delete(self, request):
|
||
"""
|
||
Delete one contest_announcement.
|
||
"""
|
||
contest_announcement_id = request.GET.get("id")
|
||
if contest_announcement_id:
|
||
ContestAnnouncement.objects.filter(id=contest_announcement_id).delete()
|
||
return self.success()
|
||
|
||
@super_admin_required
|
||
def get(self, request):
|
||
"""
|
||
Get one contest_announcement or contest_announcement list.
|
||
"""
|
||
contest_announcement_id = request.GET.get("id")
|
||
if contest_announcement_id:
|
||
try:
|
||
contest_announcement = ContestAnnouncement.objects.get(id=contest_announcement_id)
|
||
return self.success(ContestAnnouncementSerializer(contest_announcement).data)
|
||
except ContestAnnouncement.DoesNotExist:
|
||
return self.error("Contest announcement does not exist")
|
||
|
||
contest_id = request.GET.get("contest_id")
|
||
if not contest_id:
|
||
return self.error("Parameter error")
|
||
contest_announcements = ContestAnnouncement.objects.filter(contest_id=contest_id)
|
||
keyword = request.GET.get("keyword")
|
||
if keyword:
|
||
contest_announcements = contest_announcements.filter(title__contains=keyword)
|
||
return self.success(ContestAnnouncementSerializer(contest_announcements, many=True).data)
|
||
|
||
|
||
class ACMContestHelper(APIView):
|
||
@teacher_admin_required
|
||
def get(self, request):
|
||
contest_id = request.GET.get("contest_id")
|
||
if not contest_id:
|
||
return self.error("Parameter error, contest_id is required")
|
||
try:
|
||
contest = Contest.objects.get(id=contest_id, visible=True)
|
||
except Contest.DoesNotExist:
|
||
return self.error("Contest does not exist")
|
||
ensure_created_by(contest, request.user)
|
||
|
||
problems = Problem.objects.filter(contest=contest).values("id", "_id")
|
||
problem_id_map = {str(p["id"]): p["_id"] for p in problems}
|
||
|
||
ranks = ACMContestRank.objects.filter(contest=contest, accepted_number__gt=0).values("id", "user__username", "user__userprofile__real_name", "submission_info")
|
||
results = []
|
||
for rank in ranks:
|
||
for problem_id, info in rank["submission_info"].items():
|
||
if info["is_ac"]:
|
||
results.append(
|
||
{
|
||
"id": rank["id"],
|
||
"username": rank["user__username"],
|
||
"real_name": rank["user__userprofile__real_name"],
|
||
"problem_id": problem_id,
|
||
"problem_display_id": problem_id_map.get(problem_id, problem_id),
|
||
"ac_info": info,
|
||
"checked": info.get("checked", False),
|
||
}
|
||
)
|
||
results.sort(key=lambda x: -x["ac_info"]["ac_time"])
|
||
return self.success(results)
|
||
|
||
@teacher_admin_required
|
||
@validate_serializer(ACMContesHelperSerializer)
|
||
def put(self, request):
|
||
data = request.data
|
||
# 原来只按 pk 取，contest_id 明明在序列化器里、客户端一直在传，却完全没用上。
|
||
# 于是任何老师都能改别人比赛里的检查标记 —— teacher_admin_required 只保证
|
||
# "是老师"。同类的 get 是有 ensure_created_by 的，这里补齐。
|
||
try:
|
||
rank = ACMContestRank.objects.get(pk=data["rank_id"], contest_id=data["contest_id"])
|
||
except ACMContestRank.DoesNotExist:
|
||
return self.error("Rank id does not exist")
|
||
ensure_created_by(rank.contest, request.user)
|
||
problem_rank_status = rank.submission_info.get(data["problem_id"])
|
||
if not problem_rank_status:
|
||
return self.error("Problem id does not exist")
|
||
problem_rank_status["checked"] = data["checked"]
|
||
rank.save(update_fields=("submission_info",))
|
||
return self.success()
|
||
|
||
|
||
# DEPRECATED: 前端未调用 (2026-05-26)
|
||
class DownloadContestSubmissions(APIView):
|
||
def _dump_submissions(self, contest, exclude_admin=True):
|
||
problem_ids = contest.problem_set.all().values_list("id", "_id")
|
||
id2display_id = {k[0]: k[1] for k in problem_ids}
|
||
ac_map = {k[0]: False for k in problem_ids}
|
||
submissions = Submission.objects.filter(contest=contest, result__in=[JudgeStatus.ACCEPTED, JudgeStatus.AST_CHECK_FAILED]).order_by("-create_time")
|
||
user_ids = submissions.values_list("user_id", flat=True)
|
||
users = User.objects.filter(id__in=user_ids)
|
||
path = f"/tmp/{rand_str()}.zip"
|
||
with zipfile.ZipFile(path, "w") as zip_file:
|
||
for user in users:
|
||
if user.is_admin_role() and exclude_admin:
|
||
continue
|
||
user_ac_map = copy.deepcopy(ac_map)
|
||
user_submissions = submissions.filter(user_id=user.id)
|
||
for submission in user_submissions:
|
||
problem_id = submission.problem_id
|
||
if user_ac_map[problem_id]:
|
||
continue
|
||
file_name = f"{user.username}_{id2display_id[submission.problem_id]}.txt"
|
||
compression = zipfile.ZIP_DEFLATED
|
||
zip_file.writestr(
|
||
zinfo_or_arcname=f"{file_name}",
|
||
data=submission.code,
|
||
compress_type=compression,
|
||
)
|
||
user_ac_map[problem_id] = True
|
||
return path
|
||
|
||
@super_admin_required
|
||
def get(self, request):
|
||
contest_id = request.GET.get("contest_id")
|
||
if not contest_id:
|
||
return self.error("Parameter error")
|
||
try:
|
||
contest = Contest.objects.get(id=contest_id)
|
||
except Contest.DoesNotExist:
|
||
return self.error("Contest does not exist")
|
||
|
||
exclude_admin = request.GET.get("exclude_admin") == "1"
|
||
zip_path = self._dump_submissions(contest, exclude_admin)
|
||
delete_files.send_with_options(args=(zip_path,), delay=300_000)
|
||
resp = FileResponse(open(zip_path, "rb"))
|
||
resp["Content-Type"] = "application/zip"
|
||
resp["Content-Disposition"] = f"attachment;filename={os.path.basename(zip_path)}"
|
||
return resp
|
||
|
||
|
||
class ContestCloneAPI(APIView):
|
||
@validate_serializer(ContestCloneSerializer)
|
||
@teacher_admin_required
|
||
def post(self, request):
|
||
try:
|
||
original = Contest.objects.get(id=request.data["contest_id"])
|
||
except Contest.DoesNotExist:
|
||
return self.error("Contest does not exist")
|
||
|
||
duration = original.end_time - original.start_time
|
||
new_start = now() + timedelta(minutes=10)
|
||
new_end = new_start + duration
|
||
|
||
new_contest = Contest.objects.create(
|
||
title=original.title,
|
||
description=original.description,
|
||
tag=original.tag,
|
||
password=original.password,
|
||
visible=False,
|
||
allowed_ip_ranges=original.allowed_ip_ranges,
|
||
start_time=new_start,
|
||
end_time=new_end,
|
||
created_by=request.user,
|
||
)
|
||
|
||
for problem in Problem.objects.filter(contest=original):
|
||
tags = problem.tags.all()
|
||
problem.pk = None
|
||
problem.contest = new_contest
|
||
problem.submission_number = 0
|
||
problem.accepted_number = 0
|
||
problem.statistic_info = {}
|
||
problem.created_by = request.user
|
||
problem.save()
|
||
problem.tags.set(tags)
|
||
|
||
return self.success(ContestAdminSerializer(new_contest).data)
|