import hashlib
import json
import os
import re
import shutil
import yaml
import zipfile
from django.conf import settings
from django.core.files.base import ContentFile
from django.core.files.storage import FileSystemStorage
from django.urls import reverse
from django.utils.translation import gettext as _
from django.core.cache import cache
VALIDATOR_TEMPLATE_PATH = "validator_template/template.py"
if os.altsep:
def split_path_first(
path, repath=re.compile("[%s]" % re.escape(os.sep + os.altsep))
):
return repath.split(path, 1)
else:
def split_path_first(path):
return path.split(os.sep, 1)
class ProblemDataStorage(FileSystemStorage):
def __init__(self):
super(ProblemDataStorage, self).__init__(settings.DMOJ_PROBLEM_DATA_ROOT)
def url(self, name):
path = split_path_first(name)
if len(path) != 2:
raise ValueError("This file is not accessible via a URL.")
return reverse("problem_data_file", args=path)
def _save(self, name, content):
if self.exists(name):
self.delete(name)
return super(ProblemDataStorage, self)._save(name, content)
def get_available_name(self, name, max_length=None):
return name
def rename(self, old, new):
return os.rename(self.path(old), self.path(new))
class ProblemDataError(Exception):
def __init__(self, message):
super(ProblemDataError, self).__init__(message)
self.message = message
class ProblemDataCompiler(object):
def __init__(self, problem, data, cases, files):
self.problem = problem
self.data = data
self.cases = cases
self.files = files
self.generator = data.generator
def make_init(self):
cases = []
batch = None
def end_batch():
if not batch["batched"]:
raise ProblemDataError(_("Empty batches not allowed."))
cases.append(batch)
def make_checker_for_validator(case):
checker_name = "cppvalidator.py"
validator_path = split_path_first(case.custom_validator.name)
if len(validator_path) != 2:
raise ProblemDataError(
_("How did you corrupt the custom checker path?")
)
checker = os.path.join(
settings.DMOJ_PROBLEM_DATA_ROOT, validator_path[0], checker_name
)
validator_name = validator_path[1]
shutil.copy(VALIDATOR_TEMPLATE_PATH, checker)
# replace {{filecpp}} and {{problemid}} in checker file
filedata = open(checker, "r").read()
filedata = filedata.replace("{{filecpp}}", "'%s'" % validator_name)
filedata = filedata.replace("{{problemid}}", "'%s'" % validator_path[0])
open(checker, "w").write(filedata)
return checker_name
def make_checker(case):
if case.checker == "custom":
custom_checker_path = split_path_first(case.custom_checker.name)
if len(custom_checker_path) != 2:
raise ProblemDataError(
_("How did you corrupt the custom checker path?")
)
return custom_checker_path[1]
if case.checker == "customval":
return make_checker_for_validator(case)
if case.checker_args:
return {
"name": case.checker,
"args": json.loads(case.checker_args),
}
return case.checker
for i, case in enumerate(self.cases, 1):
if case.type == "C":
data = {}
if batch:
case.points = None
case.is_pretest = batch["is_pretest"]
else:
if case.points is None:
raise ProblemDataError(
_("Points must be defined for non-batch case #%d.") % i
)
data["is_pretest"] = case.is_pretest
if not self.generator:
if case.input_file not in self.files:
raise ProblemDataError(
_("Input file for case %d does not exist: %s")
% (i, case.input_file)
)
if case.output_file not in self.files:
raise ProblemDataError(
_("Output file for case %d does not exist: %s")
% (i, case.output_file)
)
if case.input_file:
data["in"] = case.input_file
if case.output_file:
data["out"] = case.output_file
if case.points is not None:
data["points"] = case.points
if case.generator_args:
data["generator_args"] = case.generator_args.splitlines()
if case.output_limit is not None:
data["output_limit_length"] = case.output_limit
if case.output_prefix is not None:
data["output_prefix_length"] = case.output_prefix
if case.checker:
data["checker"] = make_checker(case)
else:
case.checker_args = ""
case.save(update_fields=("checker_args", "is_pretest"))
(batch["batched"] if batch else cases).append(data)
elif case.type == "S":
if batch:
end_batch()
if case.points is None:
raise ProblemDataError(
_("Batch start case #%d requires points.") % i
)
batch = {
"points": case.points,
"batched": [],
"is_pretest": case.is_pretest,
}
if case.generator_args:
batch["generator_args"] = case.generator_args.splitlines()
if case.output_limit is not None:
batch["output_limit_length"] = case.output_limit
if case.output_prefix is not None:
batch["output_prefix_length"] = case.output_prefix
if case.checker:
batch["checker"] = make_checker(case)
else:
case.checker_args = ""
case.input_file = ""
case.output_file = ""
case.save(update_fields=("checker_args", "input_file", "output_file"))
elif case.type == "E":
if not batch:
raise ProblemDataError(
_("Attempt to end batch outside of one in case #%d") % i
)
case.is_pretest = batch["is_pretest"]
case.input_file = ""
case.output_file = ""
case.generator_args = ""
case.checker = ""
case.checker_args = ""
case.save()
end_batch()
batch = None
if batch:
end_batch()
init = {}
if self.data.zipfile:
zippath = split_path_first(self.data.zipfile.name)
if len(zippath) != 2:
raise ProblemDataError(_("How did you corrupt the zip path?"))
init["archive"] = zippath[1]
if self.generator:
generator_path = split_path_first(self.generator.name)
if len(generator_path) != 2:
raise ProblemDataError(_("How did you corrupt the generator path?"))
init["generator"] = generator_path[1]
pretests = [case for case in cases if case["is_pretest"]]
for case in cases:
del case["is_pretest"]
if pretests:
init["pretest_test_cases"] = pretests
if cases:
init["test_cases"] = cases
if self.data.output_limit is not None:
init["output_limit_length"] = self.data.output_limit
if self.data.output_prefix is not None:
init["output_prefix_length"] = self.data.output_prefix
if self.data.checker:
if self.data.checker == "interact":
init["interactive"] = {
"files": split_path_first(self.data.interactive_judge.name)[1],
"feedback": True,
}
init["unbuffered"] = True
else:
init["checker"] = make_checker(self.data)
else:
self.data.checker_args = ""
return init
def compile(self):
from judge.models import problem_data_storage
yml_file = "%s/init.yml" % self.problem.code
try:
init = yaml.safe_dump(self.make_init())
except ProblemDataError as e:
self.data.feedback = e.message
self.data.save()
problem_data_storage.delete(yml_file)
else:
self.data.feedback = ""
self.data.save()
problem_data_storage.save(yml_file, ContentFile(init))
@classmethod
def generate(cls, *args, **kwargs):
self = cls(*args, **kwargs)
self.compile()
def get_visible_content(data):
data = data or b""
data = data.replace(b"\r\n", b"\r").replace(b"\r", b"\n")
data = data.decode("utf-8")
if len(data) > settings.TESTCASE_VISIBLE_LENGTH:
data = data[: settings.TESTCASE_VISIBLE_LENGTH]
data += "." * 3
return data
def get_file_cachekey(file):
return hashlib.sha1(file.encode()).hexdigest()
def get_problem_case(problem, files):
result = {}
uncached_files = []
for file in files:
cache_key = "problem_archive:%s:%s" % (problem.code, get_file_cachekey(file))
qs = cache.get(cache_key)
if qs is None:
uncached_files.append(file)
else:
result[file] = qs
if not uncached_files:
return result
archive_path = os.path.join(
settings.DMOJ_PROBLEM_DATA_ROOT, str(problem.data_files.zipfile)
)
if not os.path.exists(archive_path):
raise Exception('archive file "%s" does not exist' % archive_path)
try:
archive = zipfile.ZipFile(archive_path, "r")
except zipfile.BadZipfile:
raise Exception('bad archive: "%s"' % archive_path)
for file in uncached_files:
cache_key = "problem_archive:%s:%s" % (problem.code, get_file_cachekey(file))
with archive.open(file) as f:
s = f.read(settings.TESTCASE_VISIBLE_LENGTH + 3)
# add this so there are no characters left behind (ex, 'รก' = 2 utf-8 chars)
while True:
try:
s.decode("utf-8")
break
except UnicodeDecodeError:
next_char = f.read(1)
if next_char:
s += next_char
else:
raise Exception("File %s is not able to decode in utf-8" % file)
qs = get_visible_content(s)
cache.set(cache_key, qs, 86400)
result[file] = qs
return result