feat: 新增 rebuild_sql_display 管理命令，为存量 SQL 题回填展示数据

生成逻辑从 ProblemBase._build_sql_display 提取为 problem.utils.generate_sql_display，
管理端保存与回填命令共用。命令支持 --dry-run / --force。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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)

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}"

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):