diff --git a/problem/management/__init__.py b/problem/management/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/problem/management/commands/__init__.py b/problem/management/commands/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/problem/management/commands/rebuild_sql_display.py b/problem/management/commands/rebuild_sql_display.py new file mode 100644 index 0000000..8151953 --- /dev/null +++ b/problem/management/commands/rebuild_sql_display.py @@ -0,0 +1,36 @@ +from django.core.management.base import BaseCommand + +from problem.models import Problem +from problem.utils import generate_sql_display + + +class Command(BaseCommand): + help = "为存量 SQL 题目重新生成 sql_display 展示数据（新建/编辑题目时会自动生成，此命令用于回填老题）" + + def add_arguments(self, parser): + parser.add_argument("--dry-run", action="store_true", help="只检查并打印结果，不写数据库") + parser.add_argument("--force", action="store_true", help="已有 sql_display 的题目也重新生成（默认跳过）") + + def handle(self, *args, **options): + qs = Problem.objects.filter(sql_config__isnull=False).order_by("id") + ok = failed = skipped = 0 + for p in qs: + label = f"id={p.id} _id={p._id}" + (f" contest={p.contest_id}" if p.contest_id else "") + f" {p.title}" + if p.sql_display is not None and not options["force"]: + skipped += 1 + self.stdout.write(f"SKIP {label}（已有 sql_display，--force 可强制重建）") + continue + display, error = generate_sql_display(p.test_case_id, p.answers, p.sql_config) + if error: + failed += 1 + self.stderr.write(self.style.ERROR(f"FAIL {label}: {error}")) + continue + if not options["dry_run"]: + p.sql_display = display + p.save(update_fields=["sql_display"]) + ok += 1 + self.stdout.write(self.style.SUCCESS(f"OK {label}")) + summary = f"完成：成功 {ok}，失败 {failed}，跳过 {skipped}，共 {qs.count()} 道 SQL 题" + if options["dry_run"]: + summary += "（dry-run，未写库）" + self.stdout.write(summary) diff --git a/problem/utils.py b/problem/utils.py index 8769e8c..92a2a06 100644 --- a/problem/utils.py +++ b/problem/utils.py @@ -1,6 +1,13 @@ +import json +import os import re from functools import lru_cache +from django.conf import settings + +from judge.sql_runner import SQLCaseError, build_display +from utils.shortcuts import natural_sort_key + TEMPLATE_BASE = """//PREPEND BEGIN {} //PREPEND END @@ -19,11 +26,43 @@ def parse_problem_template(template_str): prepend = re.findall(r"//PREPEND BEGIN\n([\s\S]+?)//PREPEND END", template_str) template = re.findall(r"//TEMPLATE BEGIN\n([\s\S]+?)//TEMPLATE END", template_str) append = re.findall(r"//APPEND BEGIN\n([\s\S]+?)//APPEND END", template_str) - return {"prepend": prepend[0] if prepend else "", - "template": template[0] if template else "", - "append": append[0] if append else ""} + return {"prepend": prepend[0] if prepend else "", "template": template[0] if template else "", "append": append[0] if append else ""} @lru_cache(maxsize=100) def build_problem_template(prepend, template, append): return TEMPLATE_BASE.format(prepend, template, append) + + +def generate_sql_display(test_case_id, answers, sql_config): + """SQL 题：用测试点1的初始化脚本 + 标准答案生成题目页展示数据。 + + 返回 (sql_display, error)：成功时 error 为 None，失败时 sql_display 为 None、error 为中文提示。 + """ + test_case_dir = os.path.join(settings.TEST_CASE_DIR, 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): + return None, "测试点信息读取失败，请重新上传测试点" + if not info.get("sql"): + return None, "测试点不是 SQL 类型，请重新上传 SQL 测试点压缩包" + try: + keys = sorted(info["test_cases"].keys(), key=natural_sort_key) + if not keys: + return None, "题目没有任何测试点" + input_name = info["test_cases"][keys[0]]["input_name"] + except (KeyError, AttributeError, TypeError): + return None, "测试点信息损坏，请重新上传测试点" + try: + with open(os.path.join(test_case_dir, input_name), encoding="utf-8") as f: + init_sql = f.read() + except OSError: + return None, f"测试点脚本 {input_name} 读取失败" + ref_sql = next((item["code"] for item in answers or [] if item.get("language") == "SQL" and item.get("code", "").strip()), None) + if ref_sql is None: + return None, "题目缺少 SQL 标准答案" + try: + return build_display(init_sql, ref_sql, sql_config["mode"]), None + except SQLCaseError as e: + return None, f"SQL 展示数据生成失败: {e.message}" diff --git a/problem/views/admin.py b/problem/views/admin.py index 226297b..09e72b0 100644 --- a/problem/views/admin.py +++ b/problem/views/admin.py @@ -13,7 +13,6 @@ from django.http import StreamingHttpResponse from account.decorators import ensure_created_by, problem_permission_required, teacher_admin_required from contest.models import Contest, ContestStatus -from judge.sql_runner import SQLCaseError, build_display from submission.models import Submission from utils.api import APIError, APIView, CSRFExemptAPIView, validate_serializer from utils.openai import get_ai_client @@ -31,6 +30,7 @@ from ..serializers import ( ProblemAdminSerializer, TestCaseUploadForm, ) +from ..utils import generate_sql_display class TestCaseZipProcessor(object): @@ -224,32 +224,11 @@ class ProblemBase(APIView): data["sql_display"] = None def _build_sql_display(self, data): - """SQL 题：用测试点1的初始化脚本 + 标准答案生成题目页展示数据。返回错误信息字符串，成功返回 None。""" - test_case_dir = os.path.join(settings.TEST_CASE_DIR, data["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): - return "测试点信息读取失败，请重新上传测试点" - if not info.get("sql"): - return "测试点不是 SQL 类型，请重新上传 SQL 测试点压缩包" - try: - keys = sorted(info["test_cases"].keys(), key=natural_sort_key) - if not keys: - return "题目没有任何测试点" - input_name = info["test_cases"][keys[0]]["input_name"] - except (KeyError, AttributeError, TypeError): - return "测试点信息损坏，请重新上传测试点" - try: - with open(os.path.join(test_case_dir, input_name), encoding="utf-8") as f: - init_sql = f.read() - except OSError: - return f"测试点脚本 {input_name} 读取失败" - ref_sql = next(item["code"] for item in data["answers"] if item.get("language") == "SQL" and item.get("code", "").strip()) - try: - data["sql_display"] = build_display(init_sql, ref_sql, data["sql_config"]["mode"]) - except SQLCaseError as e: - return f"SQL 展示数据生成失败: {e.message}" + """SQL 题：生成题目页展示数据。返回错误信息字符串，成功返回 None。""" + display, error = generate_sql_display(data["test_case_id"], data["answers"], data["sql_config"]) + if error: + return error + data["sql_display"] = display class ProblemAPI(ProblemBase):