diff --git a/judge/dispatcher.py b/judge/dispatcher.py index 9206249..323d66a 100644 --- a/judge/dispatcher.py +++ b/judge/dispatcher.py @@ -28,6 +28,7 @@ def process_pending_task(): if cache.llen(CacheKey.waiting_queue): # 防止循环引入 from judge.tasks import judge_task + tmp_data = cache.rpop(CacheKey.waiting_queue) if tmp_data: data = json.loads(tmp_data.decode("utf-8")) @@ -42,8 +43,7 @@ class ChooseJudgeServer: with transaction.atomic(): cutoff = timezone.now() - timedelta(seconds=6) server = ( - JudgeServer.objects - .select_for_update(skip_locked=True) + JudgeServer.objects.select_for_update(skip_locked=True) .filter( is_disabled=False, last_heartbeat__gte=cutoff, @@ -78,8 +78,6 @@ class DispatcherBase(object): logger.exception(e) - - class JudgeDispatcher(DispatcherBase): def __init__(self, submission_id, problem_id): super().__init__() @@ -93,6 +91,24 @@ class JudgeDispatcher(DispatcherBase): else: self.problem = Problem.objects.get(id=problem_id) + def _push_status(self, result, status, extra=None): + data = { + "type": "submission_update", + "submission_id": str(self.submission.id), + "result": result, + "status": status, + } + if extra: + data.update(extra) + try: + push_submission_update( + submission_id=str(self.submission.id), + user_id=self.submission.user_id, + data=data, + ) + except Exception as e: + logger.error(f"Failed to push submission update: {str(e)}") + def _compute_statistic_info(self, resp_data): # 用时和内存占用保存为多个测试点中最长的那个 self.submission.statistic_info["time_cost"] = max([x["cpu_time"] for x in resp_data]) @@ -131,7 +147,7 @@ class JudgeDispatcher(DispatcherBase): "max_memory": 1024 * 1024 * self.problem.memory_limit, "test_case_id": self.problem.test_case_id, "output": False, - "io_mode": self.problem.io_mode + "io_mode": self.problem.io_mode, } with ChooseJudgeServer() as server: @@ -139,57 +155,30 @@ class JudgeDispatcher(DispatcherBase): data = {"submission_id": self.submission.id, "problem_id": self.problem.id} cache.lpush(CacheKey.waiting_queue, json.dumps(data)) # 推送排队状态 - try: - push_submission_update( - submission_id=str(self.submission.id), - user_id=self.submission.user_id, - data={ - "type": "submission_update", - "submission_id": str(self.submission.id), - "result": JudgeStatus.PENDING, - "status": "pending", - } - ) - except Exception as e: - logger.error(f"Failed to push submission update: {str(e)}") + self._push_status(JudgeStatus.PENDING, "pending") return Submission.objects.filter(id=self.submission.id).update(result=JudgeStatus.JUDGING) - + # 推送判题中状态 - try: - push_submission_update( - submission_id=str(self.submission.id), - user_id=self.submission.user_id, - data={ - "type": "submission_update", - "submission_id": str(self.submission.id), - "result": JudgeStatus.JUDGING, - "status": "judging", - } - ) - except Exception as e: - logger.error(f"Failed to push submission update: {str(e)}") - + self._push_status(JudgeStatus.JUDGING, "judging") + resp = self._request(urljoin(server.service_url, "/judge"), data=data) if not resp: Submission.objects.filter(id=self.submission.id).update(result=JudgeStatus.SYSTEM_ERROR) # 推送系统错误状态 - try: - push_submission_update( - submission_id=str(self.submission.id), - user_id=self.submission.user_id, - data={ - "type": "submission_update", - "submission_id": str(self.submission.id), - "result": JudgeStatus.SYSTEM_ERROR, - "status": "error", - } - ) - except Exception as e: - logger.error(f"Failed to push submission update: {str(e)}") + self._push_status(JudgeStatus.SYSTEM_ERROR, "error") return + self._process_judge_result(resp) + + def _process_judge_result(self, resp): + """判题结果的统一后处理：状态聚合、AST 钩子、统计、排名、WebSocket 推送。 + + resp 结构与外部 judger 返回一致：{"err": "CompileError"|None, "data": ...}； + SQLJudgeDispatcher 组装同构 resp 后也走这里。 + """ + language = self.submission.language if resp["err"]: self.submission.result = JudgeStatus.COMPILE_ERROR self.submission.statistic_info["err_info"] = resp["data"] @@ -220,31 +209,22 @@ class JudgeDispatcher(DispatcherBase): self.submission.save(update_fields=["result", "info", "statistic_info"]) # 推送判题完成状态 - try: - push_submission_update( - submission_id=str(self.submission.id), - user_id=self.submission.user_id, - data={ - "type": "submission_update", - "submission_id": str(self.submission.id), - "result": self.submission.result, - "status": "finished", - "time_cost": self.submission.statistic_info.get("time_cost"), - "memory_cost": self.submission.statistic_info.get("memory_cost"), - "score": self.submission.statistic_info.get("score", 0), - } - ) - except Exception as e: - logger.error(f"Failed to push submission update: {str(e)}") + self._push_status( + self.submission.result, + "finished", + extra={ + "time_cost": self.submission.statistic_info.get("time_cost"), + "memory_cost": self.submission.statistic_info.get("memory_cost"), + "score": self.submission.statistic_info.get("score", 0), + }, + ) if self.contest_id: # 以提交时刻（而非判题时刻）是否落在比赛时间窗内为准， # 避免临界提交因判题排队延迟到比赛结束后才处理而被丢弃 in_contest = self.contest.start_time <= self.submission.create_time <= self.contest.end_time - if not in_contest or \ - User.objects.get(id=self.submission.user_id).is_contest_admin(self.contest): - logger.info( - "Contest debug mode, id: " + str(self.contest_id) + ", submission id: " + self.submission.id) + if not in_contest or User.objects.get(id=self.submission.user_id).is_contest_admin(self.contest): + logger.info("Contest debug mode, id: " + str(self.contest_id) + ", submission id: " + self.submission.id) return with transaction.atomic(): self.update_contest_problem_status() @@ -286,8 +266,7 @@ class JudgeDispatcher(DispatcherBase): score = self.submission.statistic_info["score"] if not is_accepted(oi_problems_status[problem_id]["status"]): # minus last time score, add this tim score - profile.add_score(this_time_score=score, - last_time_score=oi_problems_status[problem_id]["score"]) + profile.add_score(this_time_score=score, last_time_score=oi_problems_status[problem_id]["score"]) oi_problems_status[problem_id]["score"] = score oi_problems_status[problem_id]["status"] = JudgeStatus.ACCEPTED if is_accepted(self.submission.result) else self.submission.result if is_accepted(self.submission.result): @@ -331,15 +310,12 @@ class JudgeDispatcher(DispatcherBase): score = self.submission.statistic_info["score"] if problem_id not in oi_problems_status: user_profile.add_score(score) - oi_problems_status[problem_id] = {"status": profile_status, - "_id": self.problem._id, - "score": score} + oi_problems_status[problem_id] = {"status": profile_status, "_id": self.problem._id, "score": score} if is_accepted(self.submission.result): user_profile.accepted_number += 1 elif not is_accepted(oi_problems_status[problem_id]["status"]): # minus last time score, add this time score - user_profile.add_score(this_time_score=score, - last_time_score=oi_problems_status[problem_id]["score"]) + user_profile.add_score(this_time_score=score, last_time_score=oi_problems_status[problem_id]["score"]) oi_problems_status[problem_id]["score"] = score oi_problems_status[problem_id]["status"] = profile_status if is_accepted(self.submission.result): @@ -425,4 +401,3 @@ class JudgeDispatcher(DispatcherBase): info["error_number"] = 1 rank.submission_info[str(self.submission.problem_id)] = info rank.save(update_fields=["submission_info", "total_time", "accepted_number", "submission_number"]) - diff --git a/judge/languages.py b/judge/languages.py index c08c1cc..31aa8e0 100644 --- a/judge/languages.py +++ b/judge/languages.py @@ -174,4 +174,6 @@ languages = [ {"config": _py3_lang_config, "name": "Python3", "description": "Python 3.12", "content_type": "text/x-python"}, {"config": _go_lang_config, "name": "Golang", "description": "Golang 1.22", "content_type": "text/x-go"}, {"config": _node_lang_config, "name": "JavaScript", "description": "Node.js 20", "content_type": "text/javascript"}, + # SQL 题不走外部 judger 沙箱（judge/sql_dispatcher.py 在 worker 内用 sqlite3 判题），config 仅占位 + {"config": {"template": ""}, "name": "SQL", "description": "SQLite 3", "content_type": "text/x-sql"}, ] diff --git a/judge/sql_dispatcher.py b/judge/sql_dispatcher.py new file mode 100644 index 0000000..d357b52 --- /dev/null +++ b/judge/sql_dispatcher.py @@ -0,0 +1,101 @@ +"""SQL 题判题调度：不走 JudgeServer 沙箱，在 dramatiq worker 内用 sqlite3 直接判题。 + +复用 JudgeDispatcher 的 _process_judge_result 完成状态聚合、统计、排名和 WebSocket 推送， +因此比赛排名、rejudge、题目统计的语义与沙箱判题完全一致。 +""" + +import json +import logging +import os + +from django.conf import settings + +from judge.dispatcher import JudgeDispatcher +from judge.sql_runner import SQLCaseError, run_case +from submission.models import JudgeStatus, Submission +from utils.shortcuts import natural_sort_key + +logger = logging.getLogger(__name__) + + +class SQLProblemConfigError(Exception): + pass + + +class SQLJudgeDispatcher(JudgeDispatcher): + def judge(self): + Submission.objects.filter(id=self.submission.id).update(result=JudgeStatus.JUDGING) + self._push_status(JudgeStatus.JUDGING, "judging") + + try: + ref_sql, mode, order_sensitive, init_scripts = self._load_problem_config() + except SQLProblemConfigError as e: + self._system_error(str(e)) + return + + cases = [] + for index, init_sql in enumerate(init_scripts, start=1): + try: + case = run_case( + init_sql, + ref_sql, + self.submission.code, + mode=mode, + order_sensitive=order_sensitive, + time_limit_ms=self.problem.time_limit, + memory_limit_mb=self.problem.memory_limit, + ) + except SQLCaseError as e: + # 初始化/标准答案执行失败，属出题配置问题 + self._system_error(e.message) + return + case["test_case"] = str(index) + # 语法错误与数据无关，首个测试点即可确认，整题按编译错误处理（ACM 不罚时，前端展示 err_info） + if index == 1 and case["result"] == JudgeStatus.COMPILE_ERROR: + self._process_judge_result({"err": "CompileError", "data": case["error_message"]}) + return + cases.append(case) + + self._process_judge_result({"err": None, "data": cases}) + + def _system_error(self, message): + logger.error(f"SQL judge system error, submission {self.submission.id}, problem {self.problem.id}: {message}") + Submission.objects.filter(id=self.submission.id).update(result=JudgeStatus.SYSTEM_ERROR, statistic_info={"err_info": message}) + self._push_status(JudgeStatus.SYSTEM_ERROR, "error") + + def _load_problem_config(self): + """校验并加载 SQL 题配置，返回 (标准答案, mode, order_sensitive, 各测试点初始化脚本)。""" + sql_config = self.problem.sql_config or {} + mode = sql_config.get("mode") + if mode not in ("query", "modify"): + raise SQLProblemConfigError("题目缺少 SQL 配置（题型）") + + ref_sql = None + for item in self.problem.answers or []: + if item.get("language") == "SQL" and item.get("code", "").strip(): + ref_sql = item["code"] + break + if not ref_sql: + raise SQLProblemConfigError("题目缺少 SQL 标准答案") + + test_case_dir = os.path.join(settings.TEST_CASE_DIR, self.problem.test_case_id) + try: + with open(os.path.join(test_case_dir, "info"), encoding="utf-8") as f: + info = json.load(f) + except (OSError, json.JSONDecodeError) as e: + raise SQLProblemConfigError(f"测试点信息读取失败: {e}") + if not info.get("sql"): + raise SQLProblemConfigError("测试点不是 SQL 类型，请重新上传 SQL 测试点压缩包") + + init_scripts = [] + # 按 "1","2",… 自然序遍历，保证与 test_case_score 的下标对应（OI 计分依赖顺序） + for key in sorted(info["test_cases"].keys(), key=natural_sort_key): + input_name = info["test_cases"][key]["input_name"] + try: + with open(os.path.join(test_case_dir, input_name), encoding="utf-8") as f: + init_scripts.append(f.read()) + except OSError as e: + raise SQLProblemConfigError(f"测试点脚本 {input_name} 读取失败: {e}") + if not init_scripts: + raise SQLProblemConfigError("题目没有任何测试点") + return ref_sql, mode, sql_config.get("order_sensitive", False), init_scripts diff --git a/judge/sql_runner.py b/judge/sql_runner.py new file mode 100644 index 0000000..f29589d --- /dev/null +++ b/judge/sql_runner.py @@ -0,0 +1,246 @@ +"""SQL 题判题核心：在内存 SQLite 中分别执行标准答案和学生 SQL 并比对结果。 + +查询题（mode="query"）比对最后一条 SELECT 的结果集； +增删改题（mode="modify"）比对执行后所有用户表的最终状态。 +学生 SQL 通过 authorizer（禁 ATTACH/PRAGMA，查询题只读）、 +progress_handler（墙钟超时）和 max_page_count（内存上限）三重防护。 +""" + +import sqlite3 +import time + +from submission.models import JudgeStatus + +# 单结果集/单表最大行数，防 CROSS JOIN 撑爆 worker 内存 +ROW_LIMIT = 10_000 +# progress_handler 检查粒度（SQLite VM 指令数） +PROGRESS_STEP = 1_000 +ERROR_MESSAGE_MAX_LEN = 200 + +# prepare 阶段的语法类错误，映射为 COMPILE_ERROR +_SYNTAX_ERROR_MARKERS = ("syntax error", "unrecognized token", "incomplete input") + +# 两种模式都禁止的授权码：挂载外部库 / 数据库参数 +_DENIED_ALWAYS = {sqlite3.SQLITE_ATTACH, sqlite3.SQLITE_DETACH, sqlite3.SQLITE_PRAGMA} + +# 查询题允许的授权码（白名单外一律拒绝，防先 INSERT 伪造数据再 SELECT） +_QUERY_MODE_ALLOWED = {getattr(sqlite3, name) for name in ("SQLITE_SELECT", "SQLITE_READ", "SQLITE_FUNCTION", "SQLITE_RECURSIVE", "SQLITE_TRANSACTION") if hasattr(sqlite3, name)} + + +class SQLCaseError(Exception): + """携带 JudgeStatus 的判题异常。SYSTEM_ERROR 级别（出题配置问题）会传播到 dispatcher。""" + + def __init__(self, result, message): + super().__init__(message) + self.result = result + self.message = message + + +def _truncate(message): + message = str(message) + if len(message) > ERROR_MESSAGE_MAX_LEN: + return message[:ERROR_MESSAGE_MAX_LEN] + "..." + return message + + +def split_statements(script): + """用 sqlite3.complete_statement 累积切分多条语句，正确处理字符串/注释里的分号；末尾缺分号自动补。""" + statements = [] + buf = "" + for part in script.split(";"): + buf += part + ";" + if sqlite3.complete_statement(buf): + stmt = buf.strip() + buf = "" + if stmt and stmt != ";": + statements.append(stmt) + # 残句（未闭合的引号/注释，或末尾缺分号但上面已补），交给 execute 报错或执行 + remainder = buf.strip() + if remainder and remainder != ";": + statements.append(remainder) + return statements + + +def _canonical_value(v): + """值归一化并打类型标签，防止 NULL/"NULL"、1/"1" 碰撞；数值统一比对（1 == 1.0，浮点 6 位有效数字）。""" + if v is None: + return ("null",) + if isinstance(v, int): + return ("num", str(v)) + if isinstance(v, float): + if v.is_integer() and abs(v) < 2**53: + return ("num", str(int(v))) + return ("num", format(v, ".6g")) + if isinstance(v, bytes): + return ("blob", v.hex()) + return ("str", str(v)) + + +def _canonical_row(row): + return tuple(_canonical_value(v) for v in row) + + +def _new_db(memory_limit_mb): + conn = sqlite3.connect(":memory:", isolation_level=None) # autocommit，脚本行为可预期 + conn.execute("PRAGMA page_size=4096") + # 4096B/页 × 256 页/MB，超限报 "database or disk is full" + conn.execute(f"PRAGMA max_page_count={max(int(memory_limit_mb), 1) * 256}") + return conn + + +def _execute_statements(conn, script, deadline=None): + """逐条执行，返回最后一条产生结果集的语句的 (列数, 行列表)；无结果集返回 None。""" + last_result = None + for stmt in split_statements(script): + if deadline is not None and time.monotonic() > deadline: + raise sqlite3.OperationalError("interrupted") + cursor = conn.execute(stmt) + if cursor.description is not None: + rows = cursor.fetchmany(ROW_LIMIT + 1) + if len(rows) > ROW_LIMIT: + raise SQLCaseError(JudgeStatus.MEMORY_LIMIT_EXCEEDED, f"查询结果超过 {ROW_LIMIT} 行") + last_result = (len(cursor.description), [_canonical_row(r) for r in rows]) + cursor.close() + return last_result + + +def _dump_tables(conn): + """dump 所有用户表：{表名: (列数, 行多重集)}，行内排序，表状态天然无序。""" + cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name") + tables = [r[0] for r in cursor.fetchall()] + state = {} + for table in tables: + quoted = table.replace('"', '""') + cur = conn.execute(f'SELECT * FROM "{quoted}"') + rows = cur.fetchmany(ROW_LIMIT + 1) + if len(rows) > ROW_LIMIT: + raise SQLCaseError(JudgeStatus.MEMORY_LIMIT_EXCEEDED, f"表 {table} 超过 {ROW_LIMIT} 行") + state[table] = (len(cur.description), sorted(_canonical_row(r) for r in rows)) + return state + + +def _execute_trusted(conn, script, deadline, error_prefix): + """执行受信脚本，任何失败都是出题问题 → SYSTEM_ERROR。""" + conn.set_progress_handler(lambda: 1 if time.monotonic() > deadline else 0, PROGRESS_STEP) + try: + return _execute_statements(conn, script, deadline) + except SQLCaseError as e: + raise SQLCaseError(JudgeStatus.SYSTEM_ERROR, f"{error_prefix}: {e.message}") + except sqlite3.Error as e: + raise SQLCaseError(JudgeStatus.SYSTEM_ERROR, f"{error_prefix}: {_truncate(e)}") + finally: + conn.set_progress_handler(None, PROGRESS_STEP) + + +def _run_student(conn, script, mode, deadline): + """带三重防护执行学生 SQL，异常映射为学生级 JudgeStatus。""" + denied_hints = [] + + def authorizer(action, arg1, arg2, db_name, trigger): + if action in _DENIED_ALWAYS: + denied_hints.append("禁止使用 ATTACH/DETACH/PRAGMA 等数据库管理语句") + return sqlite3.SQLITE_DENY + if mode == "query" and action not in _QUERY_MODE_ALLOWED: + denied_hints.append("本题为查询题，禁止修改数据或表结构（INSERT/UPDATE/DELETE/CREATE 等）") + return sqlite3.SQLITE_DENY + return sqlite3.SQLITE_OK + + conn.set_authorizer(authorizer) + conn.set_progress_handler(lambda: 1 if time.monotonic() > deadline else 0, PROGRESS_STEP) + try: + last_result = _execute_statements(conn, script, deadline) + except sqlite3.Error as e: + # 注意：authorizer 拒绝抛的是 DatabaseError 基类而非 OperationalError，统一按消息映射 + msg = str(e) + if "interrupted" in msg: + raise SQLCaseError(JudgeStatus.CPU_TIME_LIMIT_EXCEEDED, "SQL 执行超时") + if "database or disk is full" in msg: + raise SQLCaseError(JudgeStatus.MEMORY_LIMIT_EXCEEDED, "数据量超出内存限制") + if "not authorized" in msg or "prohibited" in msg: + hint = denied_hints[-1] if denied_hints else "本题禁止使用该语句" + raise SQLCaseError(JudgeStatus.RUNTIME_ERROR, hint) + if any(marker in msg for marker in _SYNTAX_ERROR_MARKERS): + raise SQLCaseError(JudgeStatus.COMPILE_ERROR, _truncate(msg)) + raise SQLCaseError(JudgeStatus.RUNTIME_ERROR, _truncate(msg)) + finally: + conn.set_progress_handler(None, PROGRESS_STEP) + conn.set_authorizer(None) + + if mode == "query": + return last_result + # dump 是我们自己的读取，不应吃学生的超时/授权限制（上面已清除） + return _dump_tables(conn) + + +def _compare(expected, actual, mode, order_sensitive): + if mode == "query": + exp_cols, exp_rows = expected + act_cols, act_rows = actual + if exp_cols != act_cols: + return False + if order_sensitive: + return exp_rows == act_rows + return sorted(exp_rows) == sorted(act_rows) + # modify: dump dict 里行已排序 + return expected == actual + + +def run_case(init_sql, ref_sql, student_sql, *, mode, order_sensitive, time_limit_ms, memory_limit_mb): + """判一个测试点，返回与外部 judger 单测试点同构的 dict。 + + 学生错误（CE/WA/TLE/MLE/RE）体现在返回值里； + 出题配置错误（初始化/标准答案失败）抛 SQLCaseError(SYSTEM_ERROR)，由 dispatcher 处理。 + """ + time_limit_s = time_limit_ms / 1000 + # 受信脚本（初始化/标准答案）的运行上限放宽，避免出题数据较大时误报；仍防 worker 永久阻塞 + trusted_limit_s = max(time_limit_s * 5, 10) + + ref_conn = _new_db(memory_limit_mb) + try: + _execute_trusted(ref_conn, init_sql, time.monotonic() + trusted_limit_s, "初始化脚本执行失败") + last_result = _execute_trusted(ref_conn, ref_sql, time.monotonic() + trusted_limit_s, "标准答案执行失败") + if mode == "query": + expected = last_result + else: + try: + expected = _dump_tables(ref_conn) + except SQLCaseError as e: + raise SQLCaseError(JudgeStatus.SYSTEM_ERROR, f"标准答案结果超出限制: {e.message}") + finally: + ref_conn.close() + if mode == "query" and expected is None: + raise SQLCaseError(JudgeStatus.SYSTEM_ERROR, "标准答案未产生查询结果集") + + case = { + "test_case": "", + "result": JudgeStatus.ACCEPTED, + "cpu_time": 0, + "real_time": 0, + "memory": 0, + "signal": 0, + "exit_code": 0, + "error": 0, + "output_md5": "", + "error_message": None, + } + + stu_conn = _new_db(memory_limit_mb) + try: + _execute_trusted(stu_conn, init_sql, time.monotonic() + trusted_limit_s, "初始化脚本执行失败") + start = time.monotonic() + try: + actual = _run_student(stu_conn, student_sql, mode, start + time_limit_s) + except SQLCaseError as e: + elapsed = int((time.monotonic() - start) * 1000) + case.update(result=e.result, error_message=e.message, cpu_time=elapsed, real_time=elapsed) + return case + elapsed = int((time.monotonic() - start) * 1000) + finally: + stu_conn.close() + + case["cpu_time"] = case["real_time"] = elapsed + if mode == "query" and actual is None: + case.update(result=JudgeStatus.WRONG_ANSWER, error_message="提交的 SQL 未产生查询结果集") + elif not _compare(expected, actual, mode, order_sensitive): + case["result"] = JudgeStatus.WRONG_ANSWER + return case diff --git a/judge/tasks.py b/judge/tasks.py index bd13305..1bcee67 100644 --- a/judge/tasks.py +++ b/judge/tasks.py @@ -2,13 +2,18 @@ import dramatiq from account.models import User from judge.dispatcher import JudgeDispatcher +from judge.sql_dispatcher import SQLJudgeDispatcher from submission.models import Submission from utils.shortcuts import DRAMATIQ_WORKER_ARGS @dramatiq.actor(**DRAMATIQ_WORKER_ARGS()) def judge_task(submission_id, problem_id): - uid = Submission.objects.get(id=submission_id).user_id - if User.objects.get(id=uid).is_disabled: + submission = Submission.objects.get(id=submission_id) + if User.objects.get(id=submission.user_id).is_disabled: return - JudgeDispatcher(submission_id, problem_id).judge() + # SQL 题不依赖 JudgeServer 沙箱，在 worker 内用 sqlite3 判题 + if submission.language == "SQL": + SQLJudgeDispatcher(submission_id, problem_id).judge() + else: + JudgeDispatcher(submission_id, problem_id).judge() diff --git a/options/migrations/0002_add_sql_language.py b/options/migrations/0002_add_sql_language.py new file mode 100644 index 0000000..10ab603 --- /dev/null +++ b/options/migrations/0002_add_sql_language.py @@ -0,0 +1,43 @@ +# SysOptions.languages 存在数据库里，_init_option 只在 key 缺失时写入默认值， +# 因此新增 SQL 语言必须用数据迁移追加到已部署库；不用 reset_languages()（会覆盖管理员的自定义配置）。 + +from django.db import migrations + +SQL_LANGUAGE = { + "config": {"template": ""}, + "name": "SQL", + "description": "SQLite 3", + "content_type": "text/x-sql", +} + + +def add_sql_language(apps, schema_editor): + SysOptions = apps.get_model("options", "SysOptions") + try: + option = SysOptions.objects.get(key="languages") + except SysOptions.DoesNotExist: + # 库还没初始化过 languages，留给 _init_option 用代码默认值（已含 SQL）创建 + return + if not any(item.get("name") == "SQL" for item in option.value): + option.value.append(SQL_LANGUAGE) + option.save(update_fields=["value"]) + + +def remove_sql_language(apps, schema_editor): + SysOptions = apps.get_model("options", "SysOptions") + try: + option = SysOptions.objects.get(key="languages") + except SysOptions.DoesNotExist: + return + option.value = [item for item in option.value if item.get("name") != "SQL"] + option.save(update_fields=["value"]) + + +class Migration(migrations.Migration): + dependencies = [ + ("options", "0001_initial"), + ] + + operations = [ + migrations.RunPython(add_sql_language, remove_sql_language), + ] diff --git a/problem/migrations/0011_problem_sql_config.py b/problem/migrations/0011_problem_sql_config.py new file mode 100644 index 0000000..c61db2a --- /dev/null +++ b/problem/migrations/0011_problem_sql_config.py @@ -0,0 +1,18 @@ +# Generated by Django 6.0.4 on 2026-07-02 15:57 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('problem', '0010_problem_ast_rules'), + ] + + operations = [ + migrations.AddField( + model_name='problem', + name='sql_config', + field=models.JSONField(blank=True, default=None, null=True), + ), + ] diff --git a/problem/models.py b/problem/models.py index 829363b..309525f 100644 --- a/problem/models.py +++ b/problem/models.py @@ -86,6 +86,9 @@ class Problem(models.Model): # AST 代码结构检查规则 ast_rules = models.JSONField(null=True, blank=True, default=None) + # SQL 题配置: {"mode": "query"|"modify", "order_sensitive": bool}，非 SQL 题为 None + sql_config = models.JSONField(null=True, blank=True, default=None) + class Meta: db_table = "problem" constraints = [ diff --git a/problem/serializers.py b/problem/serializers.py index 701daab..76b78a3 100644 --- a/problem/serializers.py +++ b/problem/serializers.py @@ -37,6 +37,11 @@ class CreateProblemCodeTemplateSerializer(serializers.Serializer): pass +class SQLConfigSerializer(serializers.Serializer): + mode = serializers.ChoiceField(choices=["query", "modify"]) + order_sensitive = serializers.BooleanField(default=False) + + class ProblemIOModeSerializer(serializers.Serializer): io_mode = serializers.ChoiceField(choices=ProblemIOMode.choices) input = serializers.CharField() @@ -89,6 +94,9 @@ class CreateOrEditProblemSerializer(serializers.Serializer): # AST 规则 ast_rules = serializers.JSONField(required=False, allow_null=True, default=None) + # SQL 题配置 + sql_config = SQLConfigSerializer(required=False, allow_null=True, default=None) + class CreateProblemSerializer(CreateOrEditProblemSerializer): pass diff --git a/problem/views/admin.py b/problem/views/admin.py index 460fbd0..3dabeeb 100644 --- a/problem/views/admin.py +++ b/problem/views/admin.py @@ -33,13 +33,16 @@ from ..serializers import ( class TestCaseZipProcessor(object): - def process_zip(self, uploaded_zip_file, dir=""): + def process_zip(self, uploaded_zip_file, dir="", sql=False): try: zip_file = zipfile.ZipFile(uploaded_zip_file, "r") except zipfile.BadZipFile: raise APIError("Bad zip file") name_list = zip_file.namelist() - test_case_list = self.filter_name_list(name_list, dir=dir) + if sql: + test_case_list = self.filter_sql_name_list(name_list, dir=dir) + else: + test_case_list = self.filter_name_list(name_list, dir=dir) if not test_case_list: raise APIError("Empty file") @@ -62,18 +65,33 @@ class TestCaseZipProcessor(object): info = [] - # ["1.in", "1.out", "2.in", "2.out"] => [("1.in", "1.out"), ("2.in", "2.out")] - test_case_list = zip(*[test_case_list[i::2] for i in range(2)]) - for index, item in enumerate(test_case_list): - data = { - "stripped_output_md5": md5_cache[item[1]], - "input_size": size_cache[item[0]], - "output_size": size_cache[item[1]], - "input_name": item[0], - "output_name": item[1], - } - info.append(data) - test_case_info["test_cases"][str(index + 1)] = data + if sql: + # SQL 题：每个 N.sql 是一个测试点的建表+数据脚本，没有期望输出（判题时跑标准答案生成）。 + # output_name 复用同名、md5 置空，以兼容 CreateTestCaseScoreSerializer 和前端测试点表格。 + test_case_info["sql"] = True + for index, item in enumerate(test_case_list): + data = { + "stripped_output_md5": "", + "input_size": size_cache[item], + "output_size": 0, + "input_name": item, + "output_name": item, + } + info.append(data) + test_case_info["test_cases"][str(index + 1)] = data + else: + # ["1.in", "1.out", "2.in", "2.out"] => [("1.in", "1.out"), ("2.in", "2.out")] + test_case_list = zip(*[test_case_list[i::2] for i in range(2)]) + for index, item in enumerate(test_case_list): + data = { + "stripped_output_md5": md5_cache[item[1]], + "input_size": size_cache[item[0]], + "output_size": size_cache[item[1]], + "input_name": item[0], + "output_name": item[1], + } + info.append(data) + test_case_info["test_cases"][str(index + 1)] = data with open(os.path.join(test_case_dir, "info"), "w", encoding="utf-8") as f: f.write(json.dumps(test_case_info, indent=4)) @@ -97,6 +115,19 @@ class TestCaseZipProcessor(object): else: return sorted(ret, key=natural_sort_key) + def filter_sql_name_list(self, name_list, dir=""): + # SQL 题测试点：连续编号的 1.sql, 2.sql, ... + ret = [] + prefix = 1 + while True: + name = f"{prefix}.sql" + if f"{dir}{name}" in name_list: + ret.append(name) + prefix += 1 + continue + else: + return sorted(ret, key=natural_sort_key) + class TestCaseAPI(CSRFExemptAPIView, TestCaseZipProcessor): request_parsers = () @@ -118,7 +149,19 @@ class TestCaseAPI(CSRFExemptAPIView, TestCaseZipProcessor): test_case_dir = os.path.join(settings.TEST_CASE_DIR, problem.test_case_id) if not os.path.isdir(test_case_dir): return self.error("Test case does not exists") - name_list = self.filter_name_list(os.listdir(test_case_dir)) + # SQL 题的测试点是 N.sql，需按 info 里的类型标记选择文件列表 + is_sql = False + info_path = os.path.join(test_case_dir, "info") + if os.path.isfile(info_path): + try: + with open(info_path, encoding="utf-8") as f: + is_sql = bool(json.load(f).get("sql")) + except (OSError, json.JSONDecodeError): + pass + if is_sql: + name_list = self.filter_sql_name_list(os.listdir(test_case_dir)) + else: + name_list = self.filter_name_list(os.listdir(test_case_dir)) name_list.append("info") file_name = os.path.join(test_case_dir, problem.test_case_id + ".zip") with zipfile.ZipFile(file_name, "w") as file: @@ -140,7 +183,8 @@ class TestCaseAPI(CSRFExemptAPIView, TestCaseZipProcessor): with open(zip_file, "wb") as f: for chunk in file: f.write(chunk) - info, test_case_id = self.process_zip(zip_file) + sql = request.POST.get("sql") in ("1", "true", "True") + info, test_case_id = self.process_zip(zip_file, sql=sql) os.remove(zip_file) return self.success({"id": test_case_id, "info": info}) @@ -158,6 +202,19 @@ class ProblemBase(APIView): data["total_score"] = total_score data["languages"] = list(data["languages"]) + # SQL 题校验：.sql 测试点与 .in/.out 沙箱判题互斥，SQL 必须是唯一语言 + if "SQL" in data["languages"]: + if data["languages"] != ["SQL"]: + return "SQL problem cannot be mixed with other languages" + if not data.get("sql_config"): + return "SQL problem requires sql_config" + has_sql_answer = any(item.get("language") == "SQL" and item.get("code", "").strip() for item in (data.get("answers") or [])) + if not has_sql_answer: + return "SQL problem requires a SQL reference answer" + else: + # 防脏数据：非 SQL 题不应携带 SQL 配置 + data["sql_config"] = None + class ProblemAPI(ProblemBase): @problem_permission_required