diff --git a/backend/problem/tests.py b/backend/problem/tests.py
index f096ad95..2d79e6ad 100644
--- a/backend/problem/tests.py
+++ b/backend/problem/tests.py
@@ -1,13 +1,16 @@
import copy
import hashlib
+import io
import json
import os
+import re
import shutil
from datetime import timedelta
from zipfile import ZipFile
import requests
from django.conf import settings
+from django.core.files.uploadedfile import SimpleUploadedFile
from django.db import DatabaseError
from django.utils import timezone
@@ -182,6 +185,243 @@ def test_upload_test_case_zip(self):
self.assertEqual(f.read(), name + "\n" + name + "\n" + "end")
+class ImportProblemAPITest(APITestCase):
+
+ def setUp(self):
+ self.create_school_fixtures(college_id=1, college_name="Test", department_id=1, department_name="Test")
+ self.url = self.reverse("import_problem_api")
+ self.create_super_admin()
+
+ def tearDown(self):
+ for problem in Problem.objects.all():
+ for path in re.findall(rf"{settings.UPLOAD_PREFIX}/[\w.]+", problem.description + problem.hint):
+ os.remove(os.path.join(settings.UPLOAD_DIR, os.path.basename(path)))
+
+ def problem_json(self, display_id, description, hint=""):
+ return {
+ "display_id": display_id, "title": f"title-{display_id}",
+ "description": {"format": "html", "value": description},
+ "input_description": {"format": "html", "value": "input"},
+ "output_description": {"format": "html", "value": "output"},
+ "hint": {"format": "html", "value": hint},
+ "test_case_score": None, "time_limit": 1000, "memory_limit": 256,
+ "samples": [{"input": "1", "output": "1"}], "template": {}, "spj": None,
+ "rule_type": "ACM", "source": "", "field": 0, "difficulty": "Mid", "answers": [], "tags": ["tag"],
+ }
+
+ def import_zip(self, problems):
+ files = {}
+ for problem_dir, (info, images) in problems.items():
+ files[f"{problem_dir}/problem.json"] = json.dumps(info)
+ files[f"{problem_dir}/testcase/1.in"] = "1"
+ files[f"{problem_dir}/testcase/1.out"] = "1"
+ for name, content in images.items():
+ files[f"{problem_dir}/images/{name}"] = content
+ return self.import_files(files)
+
+ def import_files(self, files):
+ buffer = io.BytesIO()
+ with ZipFile(buffer, "w") as f:
+ for name, content in files.items():
+ f.writestr(name, content)
+ return self.upload(buffer.getvalue())
+
+ def upload(self, content):
+ file = SimpleUploadedFile("problems.zip", content, content_type="application/zip")
+ return self.client.post(self.url, data={"file": file}, format="multipart")
+
+ def import_errors(self, resp):
+ self.assertFailed(resp)
+ self.assertFalse(Problem.objects.exists())
+ return [(error["dir"], error["location"], error["field"], error["message"], error["line"])
+ for error in resp.data["data"]["errors"]]
+
+ def read_upload(self, path):
+ with open(os.path.join(settings.UPLOAD_DIR, os.path.basename(path)), "rb") as f:
+ return f.read()
+
+ def test_import_images_in_zip(self):
+ resp = self.import_zip({
+ "1": (self.problem_json("p1", '
'), {"1.png": b"first"}),
+ "2": (self.problem_json("p2", '
', "
"),
+ {"1.png": b"second", "b.jpg": b"hint", "unused.png": b"unused"}),
+ })
+ self.assertSuccess(resp)
+ self.assertEqual(resp.data["data"]["import_count"], 2)
+
+ pattern = rf"{settings.UPLOAD_PREFIX}/\w+\.(?:png|jpg)"
+ first = Problem.objects.get(_id="p1")
+ second = Problem.objects.get(_id="p2")
+ self.assertNotIn("images/", first.description + second.description + second.hint)
+ self.assertIn('width="50"', second.description)
+ self.assertEqual(self.read_upload(re.search(pattern, first.description).group()), b"first")
+ self.assertEqual(self.read_upload(re.search(pattern, second.description).group()), b"second")
+ self.assertEqual(self.read_upload(re.search(pattern, second.hint).group()), b"hint")
+
+ def test_import_images_with_unusual_html(self):
+ resp = self.import_zip({
+ "1": (self.problem_json("p1", '
x
\n
\n
',
+ '
'),
+ {"a.png": b"a", "b.png": b"b", "a&b.png": b"amp"}),
+ })
+ self.assertSuccess(resp)
+
+ problem = Problem.objects.get(_id="p1")
+ pattern = rf"{settings.UPLOAD_PREFIX}/\w+\.png"
+ self.assertNotIn("images/", problem.description + problem.hint)
+ self.assertEqual([self.read_upload(path) for path in re.findall(pattern, problem.description)], [b"a", b"b"])
+ self.assertEqual(self.read_upload(re.search(pattern, problem.hint).group()), b"amp")
+ self.assertIn('alt="a > b"', problem.description)
+
+ def test_image_errors(self):
+ resp = self.import_zip({
+ "1": (self.problem_json("p1", '
'), {"1.png": b"x"}),
+ "2": (self.problem_json("p2", '
'), {"1.png": b"x"}),
+ "3": (self.problem_json("p3", '
'), {}),
+ })
+ self.assertEqual(self.import_errors(resp), [("3", "images", "description.value", "images/missing.png 파일이 없습니다", 1)])
+ self.assertEqual(set(os.listdir(settings.UPLOAD_DIR)), uploads)
+ self.assertEqual(set(os.listdir(settings.TEST_CASE_DIR)), test_cases)
+
+
class ProblemAdminAPITest(APITestCase):
def setUp(self):
diff --git a/backend/problem/views/admin.py b/backend/problem/views/admin.py
index 1127c47b..5af525fa 100644
--- a/backend/problem/views/admin.py
+++ b/backend/problem/views/admin.py
@@ -1,9 +1,14 @@
import hashlib
+import html
import json
import os
+import re
import shutil
import tempfile
import zipfile
+from html.parser import HTMLParser
+from itertools import islice
+from json.decoder import scanstring
from wsgiref.util import FileWrapper
from django.conf import settings
@@ -11,6 +16,7 @@
from django.db.models import Q
from django.http import StreamingHttpResponse, FileResponse
from django.shortcuts import get_object_or_404
+from django.utils import translation
from account.decorators import problem_permission_required, ensure_created_by
from contest.models import Contest, ContestStatus
@@ -642,10 +648,240 @@ def get(self, request):
return resp
+class ImageTagParser(HTMLParser):
+ def __init__(self):
+ super().__init__(convert_charrefs=True)
+ self.tags = []
+
+ def handle_starttag(self, tag, attrs):
+ if tag == "img":
+ self.tags.append((self.getpos(), self.get_starttag_text(), attrs))
+
+
class ImportProblemAPI(CSRFExemptAPIView, TestCaseZipProcessor):
request_parsers = ()
+ image_fields = ("description", "input_description", "output_description", "hint")
+ image_suffixes = (".gif", ".jpg", ".jpeg", ".bmp", ".png")
+ missing_test_case_limit = 10
+
+ def find_image_tags(self, value):
+ parser = ImageTagParser()
+ parser.feed(value)
+ parser.close()
+ line_starts = [0]
+ for line in value.split("\n")[:-1]:
+ line_starts.append(line_starts[-1] + len(line) + 1)
+ tags = []
+ for (lineno, offset), text, attrs in parser.tags:
+ src = dict(attrs).get("src") or ""
+ if src.startswith("images/"):
+ start = line_starts[lineno - 1] + offset
+ tags.append((start, start + len(text), text, attrs, src))
+ return tags
+
+ json_error_messages = {
+ "Expecting value": "값이 와야 합니다",
+ "Expecting ',' delimiter": "쉼표(,)가 빠졌습니다",
+ "Expecting ':' delimiter": "콜론(:)이 빠졌습니다",
+ "Expecting property name enclosed in double quotes": "키는 큰따옴표로 감싸야 합니다 (마지막 쉼표가 남아 있는지 확인해주세요)",
+ "Unterminated string starting at": "문자열이 닫히지 않았습니다",
+ "Invalid control character at": "문자열 안에 줄바꿈이나 탭 같은 제어 문자가 있습니다",
+ "Invalid \\escape": "잘못된 이스케이프 문자입니다",
+ "Extra data": "JSON이 끝난 뒤에 내용이 더 있습니다",
+ }
+
+ def fail(self, message):
+ raise APIError({"message": message, "errors": [self.error_item("", "", "zip", message)]})
+
+ def error_item(self, problem_dir, title, location, message, field="", line=None, column=None, detail=""):
+ return {"dir": problem_dir, "title": title, "location": location, "field": field, "message": message,
+ "line": line, "column": column, "detail": detail}
+
+ def add_error(self, problem_dir, title, location, message, **kwargs):
+ self.errors.append(self.error_item(problem_dir, title, location, message, **kwargs))
+
+ def add_field_error(self, problem_dir, title, path, message, location="problem.json", target=None):
+ field = "".join(f"[{key}]" if isinstance(key, int) else f".{key}" for key in path).lstrip(".")
+ pos = self.find_json_position(self.problem_text, path)
+ if pos is not None and target and self.problem_text.find(target, pos) != -1:
+ pos = self.problem_text.find(target, pos)
+ line = None if pos is None else self.json_location(self.problem_text, pos)[0]
+ self.add_error(problem_dir, title, location, message, field=field, line=line)
+
+ def flatten_errors(self, errors, path=()):
+ if isinstance(errors, dict):
+ for key, value in errors.items():
+ yield from self.flatten_errors(value, path + (key,))
+ elif errors and all(isinstance(item, str) for item in errors):
+ yield path, errors[0]
+ else:
+ for index, value in enumerate(errors):
+ if value:
+ yield from self.flatten_errors(value, path + (index,))
+
+ def find_json_position(self, text, path):
+ decoder = json.JSONDecoder()
+ whitespace = re.compile(r"\s*")
+
+ def skip(pos):
+ return whitespace.match(text, pos).end()
+
+ pos = skip(0)
+ found = None
+ for key in path:
+ if text[pos] == "{" and isinstance(key, str):
+ pos = skip(pos + 1)
+ while text[pos] == '"':
+ name, pos = scanstring(text, pos + 1)
+ pos = skip(skip(pos) + 1)
+ if name == key:
+ break
+ pos = skip(decoder.raw_decode(text, pos)[1])
+ if text[pos] == ",":
+ pos = skip(pos + 1)
+ else:
+ return found
+ elif text[pos] == "[" and isinstance(key, int):
+ pos = skip(pos + 1)
+ for _ in range(key):
+ if text[pos] == "]":
+ return found
+ pos = skip(decoder.raw_decode(text, pos)[1])
+ if text[pos] == ",":
+ pos = skip(pos + 1)
+ if text[pos] == "]":
+ return found
+ else:
+ return found
+ found = pos
+ return found
+
+ def code_snippet(self, text, lineno, colno, width=100):
+ lines = text.split("\n")
+ start, end = max(lineno - 3, 0), min(lineno + 2, len(lines))
+ digits = len(str(end))
+ snippet = []
+ for number in range(start + 1, end + 1):
+ line = lines[number - 1]
+ offset = 0
+ if len(line) > width:
+ center = colno - 1 if number == lineno else 0
+ offset = max(min(center - width // 2, len(line) - width), 0)
+ line = ("…" if offset else "") + line[offset:offset + width] + ("…" if offset + width < len(line) else "")
+ snippet.append(f"{'>' if number == lineno else ' '} {number:>{digits}} | {line}")
+ if number == lineno:
+ snippet.append(f" {' ' * digits} | {' ' * (colno - 1 - offset + (1 if offset else 0))}^")
+ return "\n".join(snippet)
+
+ def json_location(self, text, pos):
+ lineno = text.count("\n", 0, pos) + 1
+ colno = pos - text.rfind("\n", 0, pos)
+ return lineno, colno
+
+ def collect_images(self, problem_info, problem_dir, name_list):
+ images = {}
+ for field in self.image_fields:
+ for *_, src in self.find_image_tags(problem_info[field]["value"]):
+ if f"{problem_dir}/{src}" not in name_list:
+ self.add_field_error(problem_dir, problem_info["title"], (field, "value"),
+ f"{src} 파일이 없습니다", location="images", target=src)
+ elif os.path.splitext(src)[-1].lower() not in self.image_suffixes:
+ suffixes = ", ".join(suffix[1:] for suffix in self.image_suffixes)
+ self.add_field_error(problem_dir, problem_info["title"], (field, "value"),
+ f"{src}는 지원하지 않는 형식입니다 ({suffixes})", location="images",
+ target=src)
+ else:
+ images[src] = f"{problem_dir}/{src}"
+ return images
+
+ def check_test_cases(self, problem_info, problem_dir, name_list):
+ prefix = f"{problem_dir}/testcase/"
+ spj = problem_info["spj"] is not None
+ suffixes = (".in",) if spj else (".in", ".out")
+ names = {name[len(prefix):] for name in name_list if name.startswith(prefix)}
+ if "1.in" not in names:
+ required = "1.in" if spj else "1.in, 1.out"
+ self.add_error(problem_dir, problem_info["title"], "testcase", f"테스트케이스가 없습니다 ({required}부터 필요)")
+ return set()
+ numbers = sorted({int(name.split(".")[0]) for name in names
+ if re.fullmatch(r"\d+\.(in|out)", name) and name.endswith(suffixes)})
+
+ def missing_names():
+ previous = 0
+ for number in numbers:
+ for gap in range(previous + 1, number):
+ for suffix in suffixes:
+ yield f"{gap}{suffix}"
+ for suffix in suffixes:
+ if f"{number}{suffix}" not in names:
+ yield f"{number}{suffix}"
+ previous = number
+
+ missing_count = sum(f"{number}{suffix}" not in names for number in numbers for suffix in suffixes)
+ missing_count += (numbers[-1] - len(numbers)) * len(suffixes)
+ if missing_count:
+ shown = list(islice(missing_names(), self.missing_test_case_limit))
+ rest = f" 외 {missing_count - len(shown)}개" if missing_count > len(shown) else ""
+ self.add_error(problem_dir, problem_info["title"], "testcase", f"{', '.join(shown)}{rest} 파일이 없습니다")
+ return names
+
+ def check_test_case_score(self, problem_info, problem_dir, test_cases):
+ if problem_info["rule_type"] != ProblemRuleType.OI:
+ return
+ if not problem_info["test_case_score"]:
+ self.add_field_error(problem_dir, problem_info["title"], ("test_case_score",),
+ "OI 문제는 테스트케이스별 점수가 필요합니다")
+ return
+ keys = ("input_name",) if problem_info["spj"] is not None else ("input_name", "output_name")
+ for index, item in enumerate(problem_info["test_case_score"]):
+ for key in keys:
+ if test_cases and item[key] not in test_cases:
+ self.add_field_error(problem_dir, problem_info["title"], ("test_case_score", index, key),
+ f"testcase 폴더에 {item[key]} 파일이 없습니다")
+
+ def check_display_id(self, problem_info, problem_dir, display_ids):
+ display_id = problem_info["display_id"]
+ if display_id in display_ids:
+ self.add_field_error(problem_dir, problem_info["title"], ("display_id",),
+ f"{display_ids[display_id]} 폴더의 문제와 Problem ID '{display_id}'가 겹칩니다")
+ elif Problem.objects.filter(_id=display_id).exists():
+ self.add_field_error(problem_dir, problem_info["title"], ("display_id",),
+ f"Problem ID '{display_id}'인 문제가 이미 있습니다")
+ display_ids.setdefault(display_id, problem_dir)
+
+ def save_images(self, zip_file, problem_info, images):
+ urls = {}
+ for src, path in images.items():
+ name = rand_str(10) + os.path.splitext(src)[-1].lower()
+ target_path = os.path.join(settings.UPLOAD_DIR, name)
+ self.created_files.append(target_path)
+ with zip_file.open(path) as source, open(target_path, "wb") as target:
+ shutil.copyfileobj(source, target)
+ urls[src] = f"{settings.UPLOAD_PREFIX}/{name}"
+ for field in self.image_fields:
+ value = problem_info[field]["value"]
+ for start, end, text, attrs, src in reversed(self.find_image_tags(value)):
+ attrs_html = "".join(
+ f" {key}" if key != "src" and val is None
+ else f' {key}="{html.escape(urls[src] if key == "src" else val)}"'
+ for key, val in attrs)
+ closing = " />" if text.rstrip().endswith("/>") else ">"
+ value = f"{value[:start]}+ 오류가 있으면 zip 안의 어떤 문제도 저장되지 않습니다. +
+