add creation

自由创作：新增 Creation 模型（继承 Task，display 自动递增）与
/api/creation/ 路由，支持增删改和代码保存，只有创作者本人和超管可访问。
自由创作不进入提交/评分体系；prompt 的两个 WebSocket consumer
连接时校验 can_access_task。补充 task/tests.py 覆盖上述行为。

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

.github/workflows/depley.yaml

@@ -12,6 +12,7 @@ jobs:
   deploy:
     runs-on: ubuntu-latest
     strategy:
+      fail-fast: false
       matrix:
         include:
           - name: debian

api/urls.py

@@ -26,6 +26,7 @@ api = NinjaAPI()
 api.add_router("account/", "account.api.router")
 api.add_router("tutorial/", "task.tutorial.router")
 api.add_router("challenge/", "task.challenge.router")
+api.add_router("creation/", "task.creation.router")
 api.add_router("submission/", "submission.api.router")
 api.add_router("upload/", "utils.upload.router")
 api.add_router("assets/", "task.assets.router")

prompt/consumers.py

@@ -2,7 +2,7 @@ import json
 from channels.generic.websocket import AsyncWebsocketConsumer
 from channels.db import database_sync_to_async
 from .models import Message
-from .utils import get_or_create_active_conversation
+from .utils import can_access_task, get_or_create_active_conversation
 from .llm import stream_chat, extract_code, stream_guidance, parse_guidance_response
 
 
@@ -14,6 +14,10 @@ class PromptConsumer(AsyncWebsocketConsumer):
             return
 
         self.task_id = int(self.scope["url_route"]["kwargs"]["task_id"])
+        if not await self.check_access():
+            await self.close()
+            return
+
         self.current_user_message = None
         await self.accept()
 
@@ -77,6 +81,10 @@ class PromptConsumer(AsyncWebsocketConsumer):
                 await self.delete_message(self.current_user_message)
                 self.current_user_message = None
 
+    @database_sync_to_async
+    def check_access(self):
+        return can_access_task(self.user, self.task_id)
+
     @database_sync_to_async
     def get_or_create_conversation(self):
         return get_or_create_active_conversation(self.user, self.task_id)
@@ -129,6 +137,10 @@ class GuidanceConsumer(AsyncWebsocketConsumer):
             return
 
         self.task_id = int(self.scope["url_route"]["kwargs"]["task_id"])
+        if not await self.check_access():
+            await self.close()
+            return
+
         self.current_user_message = None
         self.session_messages = []
         await self.accept()
@@ -185,6 +197,10 @@ class GuidanceConsumer(AsyncWebsocketConsumer):
                 await self.delete_message(self.current_user_message)
                 self.current_user_message = None
 
+    @database_sync_to_async
+    def check_access(self):
+        return can_access_task(self.user, self.task_id)
+
     @database_sync_to_async
     def get_or_create_conversation(self):
         return get_or_create_active_conversation(self.user, self.task_id)

prompt/utils.py

@@ -2,6 +2,19 @@ from django.db.models import Count, Q
 from .models import Conversation
 
 
+def can_access_task(user, task_id):
+    """自由创作只有创作者本人（和超管）能对话，其他任务对所有登录用户开放。"""
+    from account.models import RoleChoices
+    from task.models import Creation
+
+    owner_id = (
+        Creation.objects.filter(pk=task_id).values_list("owner_id", flat=True).first()
+    )
+    if owner_id is None:
+        return True
+    return owner_id == user.id or user.role == RoleChoices.SUPER
+
+
 def get_active_conversation(user, task_id):
     """Return the conversation with the most messages for this user+task, or None."""
     return (

submission/api.py

@@ -56,7 +56,7 @@ from .schemas import (
 
 from .models import Award, ItemOrdering, Rating, Submission, SubmissionAward
 from .gradebook import GradebookFilters, build_gradebook, gradebook_csv_rows
-from task.models import Task
+from task.models import Task, TaskTypeChoices
 from account.models import RoleChoices, User
 
 router = Router()
@@ -132,6 +132,8 @@ def create_submission(request, payload: SubmissionIn):
     创建一个新的提交
     """
     task = get_object_or_404(Task, id=payload.task_id)
+    if task.task_type == TaskTypeChoices.CREATION:
+        raise HttpError(400, "自由创作不需要提交")
 
     manual_asst_msg = None
     linked_msg = None

task/admin.py

@@ -1,5 +1,6 @@
 from django.contrib import admin
-from .models import Tutorial, Challenge
+from .models import Tutorial, Challenge, Creation
 
 admin.site.register(Tutorial)
 admin.site.register(Challenge)
+admin.site.register(Creation)

task/creation.py

@@ -0,0 +1,88 @@
+from typing import List
+from ninja import Router
+from ninja.errors import HttpError
+from django.shortcuts import get_object_or_404
+from django.contrib.auth.decorators import login_required
+
+from account.models import RoleChoices
+from .models import Creation
+from .schemas import CreationCodeIn, CreationIn, CreationOut, CreationSlim
+
+router = Router()
+
+
+def _clean_title(title: str) -> str:
+    title = (title or "").strip()
+    if not title:
+        raise HttpError(400, "主题名称不能为空")
+    return title[:100]
+
+
+def _get_owned(request, display: int) -> Creation:
+    creation = get_object_or_404(Creation, display=display)
+    if creation.owner_id != request.user.id and request.user.role != RoleChoices.SUPER:
+        raise HttpError(403, "只能操作自己的创作")
+    return creation
+
+
+@router.get("/mine", response=List[CreationSlim])
+@login_required
+def list_mine(request):
+    """
+    当前用户自己创建的所有主题
+    """
+    return Creation.objects.filter(owner=request.user)
+
+
+@router.post("/", response=CreationOut)
+@login_required
+def create(request, payload: CreationIn):
+    """
+    新建一个主题，序号由后端自动分配
+    """
+    return Creation.objects.create(
+        owner=request.user,
+        title=_clean_title(payload.title),
+        content=payload.content,
+    )
+
+
+@router.get("/{display}", response=CreationOut)
+@login_required
+def get(request, display: int):
+    return _get_owned(request, display)
+
+
+@router.put("/{display}", response=CreationOut)
+@login_required
+def update(request, display: int, payload: CreationIn):
+    """
+    修改主题名称和主题描述
+    """
+    creation = _get_owned(request, display)
+    creation.title = _clean_title(payload.title)
+    creation.content = payload.content
+    creation.save(update_fields=["title", "content", "modified"])
+    return creation
+
+
+@router.put("/{display}/code", response=CreationOut)
+@login_required
+def update_code(request, display: int, payload: CreationCodeIn):
+    """
+    保存当前作品代码
+    """
+    creation = _get_owned(request, display)
+    creation.html = payload.html
+    creation.css = payload.css
+    creation.js = payload.js
+    creation.save(update_fields=["html", "css", "js", "modified"])
+    return creation
+
+
+@router.delete("/{display}")
+@login_required
+def remove(request, display: int):
+    creation = _get_owned(request, display)
+    creation.delete()
+    return {"message": "删除成功"}

task/migrations/0010_alter_task_task_type_creation.py

@@ -0,0 +1,37 @@
+# Generated by Django 6.0.1 on 2026-08-14 15:13
+
+import django.db.models.deletion
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('task', '0009_task_example_css_task_example_html_task_example_js'),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='task',
+            name='task_type',
+            field=models.CharField(choices=[('challenge', '挑战'), ('tutorial', '教程'), ('creation', '创作')], editable=False, max_length=20, verbose_name='类型'),
+        ),
+        migrations.CreateModel(
+            name='Creation',
+            fields=[
+                ('task_ptr', models.OneToOneField(auto_created=True, on_delete=django.db.models.deletion.CASCADE, parent_link=True, primary_key=True, serialize=False, to='task.task')),
+                ('html', models.TextField(blank=True, default='', verbose_name='HTML代码')),
+                ('css', models.TextField(blank=True, default='', verbose_name='CSS代码')),
+                ('js', models.TextField(blank=True, default='', verbose_name='JS代码')),
+                ('owner', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='creations', to=settings.AUTH_USER_MODEL, verbose_name='创作者')),
+            ],
+            options={
+                'verbose_name': '创作',
+                'verbose_name_plural': '创作',
+                'ordering': ('-created',),
+            },
+            bases=('task.task',),
+        ),
+    ]

task/models.py

@@ -6,6 +6,7 @@ from django_extensions.db.models import TimeStampedModel
 class TaskTypeChoices(models.TextChoices):
     CHALLENGE = "challenge", "挑战"
     TUTORIAL = "tutorial", "教程"
+    CREATION = "creation", "创作"
 
 
 class Task(TimeStampedModel):
@@ -67,6 +68,44 @@ class Challenge(Task):
         verbose_name_plural = verbose_name
 
 
+class Creation(Task):
+    """
+    用户自己创建的主题：标题和内容都由用户自己写，只有创作者本人能看到和修改
+    """
+
+    owner = models.ForeignKey(
+        settings.AUTH_USER_MODEL,
+        on_delete=models.CASCADE,
+        related_name="creations",
+        verbose_name="创作者",
+    )
+    html = models.TextField(blank=True, default="", verbose_name="HTML代码")
+    css = models.TextField(blank=True, default="", verbose_name="CSS代码")
+    js = models.TextField(blank=True, default="", verbose_name="JS代码")
+
+    @property
+    def owner_name(self):
+        return self.owner.username if self.owner_id else ""
+
+    def save(self, *args, **kwargs):
+        if not self.display:
+            last = (
+                Creation.objects.order_by("-display")
+                .values_list("display", flat=True)
+                .first()
+            )
+            self.display = (last or 0) + 1
+        super().save(*args, **kwargs)
+
+    def __str__(self):
+        return self.title
+
+    class Meta:
+        ordering = ("-created",)
+        verbose_name = "创作"
+        verbose_name_plural = verbose_name
+
+
 def task_asset_upload_to(instance, filename):
     return f"tasks/{instance.task.task_type}/{instance.task.display}/{instance.name}"
 

task/schemas.py

@@ -1,3 +1,4 @@
+from datetime import datetime
 from typing import Optional
 from ninja import Schema, ModelSchema
 from .models import Tutorial, Challenge
@@ -55,3 +56,44 @@ class ChallengeIn(Schema):
     example_html: Optional[str] = None
     example_css: Optional[str] = None
     example_js: Optional[str] = None
+
+
+class CreationSlim(Schema):
+    display: int
+    title: str
+    content: str
+    created: datetime
+    modified: datetime
+    has_code: bool
+
+    @staticmethod
+    def resolve_has_code(obj):
+        return bool(obj.html or obj.css or obj.js)
+
+
+class CreationOut(Schema):
+    task_id: int
+    display: int
+    title: str
+    content: str
+    html: str
+    css: str
+    js: str
+    owner_name: str
+    created: datetime
+    modified: datetime
+
+    @staticmethod
+    def resolve_task_id(obj):
+        return obj.pk
+
+
+class CreationIn(Schema):
+    title: str
+    content: str = ""
+
+
+class CreationCodeIn(Schema):
+    html: str = ""
+    css: str = ""
+    js: str = ""

task/tests.py

@@ -0,0 +1,94 @@
+from django.test import TestCase
+from django.contrib.auth import get_user_model
+
+from prompt.utils import can_access_task
+from task.models import Creation
+
+User = get_user_model()
+
+
+def _make_user(username="student1"):
+    return User.objects.create_user(username=username, password="pw")
+
+
+class CreationApiTest(TestCase):
+    def setUp(self):
+        self.user = _make_user("student1")
+        self.other = _make_user("student2")
+        self.client.force_login(self.user)
+
+    def _create(self, title="我的计时器", content="做一个倒计时"):
+        return self.client.post(
+            "/api/creation/",
+            data={"title": title, "content": content},
+            content_type="application/json",
+        )
+
+    def test_create_assigns_display(self):
+        """新建主题时后端自动分配递增的序号"""
+        first = self._create().json()
+        second = self._create("第二个主题").json()
+        self.assertEqual(second["display"], first["display"] + 1)
+        self.assertEqual(first["owner_name"], "student1")
+        self.assertEqual(first["task_id"], Creation.objects.get(display=first["display"]).pk)
+
+    def test_create_rejects_empty_title(self):
+        resp = self._create(title="   ")
+        self.assertEqual(resp.status_code, 400)
+
+    def test_list_mine_only_returns_own(self):
+        self._create()
+        self.client.force_login(self.other)
+        self._create("别人的主题")
+        resp = self.client.get("/api/creation/mine")
+        self.assertEqual(resp.status_code, 200)
+        titles = [item["title"] for item in resp.json()]
+        self.assertEqual(titles, ["别人的主题"])
+
+    def test_update_and_save_code(self):
+        display = self._create().json()["display"]
+        resp = self.client.put(
+            f"/api/creation/{display}",
+            data={"title": "新名字", "content": "新的描述"},
+            content_type="application/json",
+        )
+        self.assertEqual(resp.status_code, 200)
+        self.assertEqual(resp.json()["title"], "新名字")
+
+        resp = self.client.put(
+            f"/api/creation/{display}/code",
+            data={"html": "<p>hi</p>", "css": "p{color:red}", "js": ""},
+            content_type="application/json",
+        )
+        self.assertEqual(resp.status_code, 200)
+        creation = Creation.objects.get(display=display)
+        self.assertEqual(creation.html, "<p>hi</p>")
+
+        resp = self.client.get("/api/creation/mine")
+        self.assertTrue(resp.json()[0]["has_code"])
+
+    def test_other_user_cannot_read_or_change(self):
+        display = self._create().json()["display"]
+        self.client.force_login(self.other)
+        self.assertEqual(self.client.get(f"/api/creation/{display}").status_code, 403)
+        self.assertEqual(self.client.delete(f"/api/creation/{display}").status_code, 403)
+
+    def test_delete(self):
+        display = self._create().json()["display"]
+        self.assertEqual(self.client.delete(f"/api/creation/{display}").status_code, 200)
+        self.assertFalse(Creation.objects.filter(display=display).exists())
+
+    def test_submission_rejected_for_creation(self):
+        """自由创作不进入提交/评分体系"""
+        task_id = self._create().json()["task_id"]
+        resp = self.client.post(
+            "/api/submission/",
+            data={"task_id": task_id, "html": "<p>hi</p>", "css": "", "js": ""},
+            content_type="application/json",
+        )
+        self.assertEqual(resp.status_code, 400)
+
+    def test_can_access_task_limits_chat_to_owner(self):
+        creation = Creation.objects.create(owner=self.user, title="t", content="")
+        self.assertTrue(can_access_task(self.user, creation.pk))
+        self.assertFalse(can_access_task(self.other, creation.pk))