OnlineJudge/contest/views/oj.py
2018-12-15 21:30:25 +08:00

207 lines
9.0 KiB
Python

import io
import xlsxwriter
from django.http import HttpResponse
from django.utils.timezone import now
from django.core.cache import cache
from problem.models import Problem
from utils.api import APIView, validate_serializer
from utils.constants import CacheKey
from utils.shortcuts import datetime2str
from account.models import AdminType
from account.decorators import login_required, check_contest_permission
from utils.constants import ContestRuleType, ContestStatus
from ..models import ContestAnnouncement, Contest, OIContestRank, ACMContestRank
from ..serializers import ContestAnnouncementSerializer
from ..serializers import ContestSerializer, ContestPasswordVerifySerializer
from ..serializers import OIContestRankSerializer, ACMContestRankSerializer
from ..serializers import ContestSimilarResultSerializer
class ContestAnnouncementListAPI(APIView):
@check_contest_permission(check_type="announcements")
def get(self, request):
contest_id = request.GET.get("contest_id")
if not contest_id:
return self.error("Invalid parameter, contest_id is required")
data = ContestAnnouncement.objects.select_related("created_by").filter(contest_id=contest_id, visible=True)
max_id = request.GET.get("max_id")
if max_id:
data = data.filter(id__gt=max_id)
return self.success(ContestAnnouncementSerializer(data, many=True).data)
class ContestAPI(APIView):
def get(self, request):
id = request.GET.get("id")
if not id:
return self.error("Invalid parameter, id is required")
try:
contest = Contest.objects.get(id=id, visible=True)
except Contest.DoesNotExist:
return self.error("Contest does not exist")
data = ContestSerializer(contest).data
data["now"] = datetime2str(now())
return self.success(data)
class ContestListAPI(APIView):
def get(self, request):
contests = Contest.objects.select_related("created_by").filter(visible=True)
keyword = request.GET.get("keyword")
rule_type = request.GET.get("rule_type")
status = request.GET.get("status")
if keyword:
contests = contests.filter(title__contains=keyword)
if rule_type:
contests = contests.filter(rule_type=rule_type)
if status:
cur = now()
if status == ContestStatus.CONTEST_NOT_START:
contests = contests.filter(start_time__gt=cur)
elif status == ContestStatus.CONTEST_ENDED:
contests = contests.filter(end_time__lt=cur)
else:
contests = contests.filter(start_time__lte=cur, end_time__gte=cur)
return self.success(self.paginate_data(request, contests, ContestSerializer))
class ContestPasswordVerifyAPI(APIView):
@validate_serializer(ContestPasswordVerifySerializer)
@login_required
def post(self, request):
data = request.data
try:
contest = Contest.objects.get(id=data["contest_id"], visible=True, password__isnull=False)
except Contest.DoesNotExist:
return self.error("Contest does not exist")
if contest.password != data["password"]:
return self.error("Wrong password")
# password verify OK.
if "accessible_contests" not in request.session:
request.session["accessible_contests"] = []
request.session["accessible_contests"].append(contest.id)
# https://docs.djangoproject.com/en/dev/topics/http/sessions/#when-sessions-are-saved
request.session.modified = True
return self.success(True)
class ContestAccessAPI(APIView):
@login_required
def get(self, request):
contest_id = request.GET.get("contest_id")
if not contest_id:
return self.error()
return self.success({"access": int(contest_id) in request.session.get("accessible_contests", [])})
class ContestRankAPI(APIView):
# filter root or *{username}
def realrank_filter(self, page_qs):
rank_cnt = 1
for t in page_qs["results"]:
if t["user"]["username"][0] == '*':
t["rank"] = '*'
else:
t["rank"] = rank_cnt
rank_cnt = rank_cnt + 1
def get_rank(self):
if self.contest.rule_type == ContestRuleType.ACM:
return ACMContestRank.objects.filter(contest=self.contest,
user__admin_type=AdminType.REGULAR_USER,
user__is_disabled=False).\
select_related("user").order_by("-accepted_number", "total_time")
else:
return OIContestRank.objects.filter(contest=self.contest,
user__admin_type=AdminType.REGULAR_USER,
user__is_disabled=False). \
select_related("user").order_by("-total_score")
def column_string(self, n):
string = ""
while n > 0:
n, remainder = divmod(n - 1, 26)
string = chr(65 + remainder) + string
return string
@check_contest_permission(check_type="ranks")
def get(self, request):
download_csv = request.GET.get("download_csv")
force_refresh = request.GET.get("force_refresh")
is_contest_admin = request.user.is_authenticated() and request.user.is_contest_admin(self.contest)
if self.contest.rule_type == ContestRuleType.OI:
serializer = OIContestRankSerializer
else:
serializer = ACMContestRankSerializer
if force_refresh == "1" and is_contest_admin:
qs = self.get_rank()
else:
cache_key = f"{CacheKey.contest_rank_cache}:{self.contest.id}"
qs = cache.get(cache_key)
if not qs:
qs = self.get_rank()
cache.set(cache_key, qs)
if download_csv:
data = serializer(qs, many=True, is_contest_admin=is_contest_admin).data
contest_problems = Problem.objects.filter(contest=self.contest, visible=True).order_by("_id")
problem_ids = [item.id for item in contest_problems]
f = io.BytesIO()
workbook = xlsxwriter.Workbook(f)
worksheet = workbook.add_worksheet()
worksheet.write("A1", "User ID")
worksheet.write("B1", "Username")
worksheet.write("C1", "Real Name")
if self.contest.rule_type == ContestRuleType.OI:
worksheet.write("D1", "Total Score")
for item in range(contest_problems.count()):
worksheet.write(self.column_string(5 + item) + "1", f"{contest_problems[item].title}")
for index, item in enumerate(data):
worksheet.write_string(index + 1, 0, str(item["user"]["id"]))
worksheet.write_string(index + 1, 1, item["user"]["username"])
worksheet.write_string(index + 1, 2, item["user"]["real_name"] or "")
worksheet.write_string(index + 1, 3, str(item["total_score"]))
for k, v in item["submission_info"].items():
worksheet.write_string(index + 1, 4 + problem_ids.index(int(k)), str(v))
else:
worksheet.write("D1", "AC")
worksheet.write("E1", "Total Submission")
worksheet.write("F1", "Total Time")
for item in range(contest_problems.count()):
worksheet.write(self.column_string(7 + item) + "1", f"{contest_problems[item].title}")
for index, item in enumerate(data):
worksheet.write_string(index + 1, 0, str(item["user"]["id"]))
worksheet.write_string(index + 1, 1, item["user"]["username"])
worksheet.write_string(index + 1, 2, item["user"]["real_name"] or "")
worksheet.write_string(index + 1, 3, str(item["accepted_number"]))
worksheet.write_string(index + 1, 4, str(item["submission_number"]))
worksheet.write_string(index + 1, 5, str(item["total_time"]))
for k, v in item["submission_info"].items():
worksheet.write_string(index + 1, 6 + problem_ids.index(int(k)), str(v["is_ac"]))
workbook.close()
f.seek(0)
response = HttpResponse(f.read())
response["Content-Disposition"] = f"attachment; filename=content-{self.contest.id}-rank.xlsx"
response["Content-Type"] = "application/xlsx"
return response
page_qs = self.paginate_data(request, qs)
page_qs["results"] = serializer(page_qs["results"], many=True, is_contest_admin=is_contest_admin).data
self.realrank_filter(page_qs)
return self.success(page_qs)
class ContestGetSimilarAPI(APIView):
def get(self, request):
contest_id = request.GET.get("contest_id")
contest = Contest.objects.get(id=contest_id, visible=True)
return self.success(ContestSimilarResultSerializer(contest).data)