v0.2 Preview: Construction Plan.
CI / test (push) Failing after 1m31s

This commit is contained in:
2026-08-08 21:25:27 +08:00
parent b88d15698a
commit 1f98f7152d
822 changed files with 96959 additions and 8 deletions
View File
+35
View File
@@ -0,0 +1,35 @@
from django.contrib import admin
from django.contrib.auth.admin import UserAdmin
from .models import (
AuditLog,
InviteCode,
InviteCodeUsage,
User,
UserSession,
VisitorMigration,
)
@admin.register(User)
class HulumathUserAdmin(UserAdmin):
fieldsets = UserAdmin.fieldsets + (
("葫芦数学", {"fields": ("nickname", "avatar_url", "bio", "track", "rating")}),
)
add_fieldsets = UserAdmin.add_fieldsets + (
("葫芦数学", {"fields": ("nickname", "track")}),
)
list_display = ("username", "nickname", "track", "rating", "is_active", "date_joined")
@admin.register(InviteCode)
class InviteCodeAdmin(admin.ModelAdmin):
list_display = ("code", "group", "used_count", "max_uses", "expires_at", "is_active")
list_filter = ("is_active", "group")
search_fields = ("code", "group")
admin.site.register(InviteCodeUsage)
admin.site.register(UserSession)
admin.site.register(VisitorMigration)
admin.site.register(AuditLog)
+6
View File
@@ -0,0 +1,6 @@
from django.apps import AppConfig
class AccountsConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'accounts'
+104
View File
@@ -0,0 +1,104 @@
from django.db import transaction
from django.utils import timezone
from content.models import ContentInteraction, ContentItem
from math_life.models import (
MathBTIAssessment,
MathBTIResult,
MathIdentity,
Story,
StoryRun,
StoryVersion,
)
from math_life.services import score_mathbti
from progression.models import Card, UserCard
from progression.services import initialize_math_identity
from .models import VisitorMigration
def migration_summary(payload):
mathbti = 1 if isinstance(payload.get("mathbti"), dict) else 0
return {
"mathbti_results": mathbti,
"story_saves": min(len(payload.get("story_saves", [])), 20),
"favorites": min(len(payload.get("favorites", [])), 200),
"cards": min(len(payload.get("cards", [])), 100),
}
@transaction.atomic
def apply_visitor_migration(user):
migration = VisitorMigration.objects.select_for_update().get(user=user)
if migration.status == VisitorMigration.Status.APPLIED:
return migration
payload = migration.payload
mathbti = payload.get("mathbti")
if isinstance(mathbti, dict):
assessment = MathBTIAssessment.objects.filter(
version=mathbti.get("version"),
is_published=True,
).first()
if assessment:
code, scores = score_mathbti(assessment.definition, mathbti.get("answers", []))
identity = MathIdentity.objects.get(code=code)
MathBTIResult.objects.create(
user=user,
assessment=assessment,
identity=identity,
answers=mathbti["answers"],
axis_scores=scores,
)
initialize_math_identity(user, identity)
for item in payload.get("story_saves", [])[:20]:
if not isinstance(item, dict):
continue
story = Story.objects.filter(slug=item.get("story_slug")).first()
if story is None:
continue
version = StoryVersion.objects.filter(
story=story,
version=item.get("version"),
is_published=True,
).first()
node_id = item.get("current_node")
if version is None or node_id not in version.content.get("nodes", {}):
continue
StoryRun.objects.get_or_create(
user=user,
story_version=version,
status=StoryRun.Status.ACTIVE,
defaults={
"current_node": node_id,
"state": item.get("state", {}) if isinstance(item.get("state"), dict) else {},
},
)
favorites = ContentItem.objects.filter(
slug__in=payload.get("favorites", [])[:200],
is_published=True,
)
ContentInteraction.objects.bulk_create(
[
ContentInteraction(
user=user,
content=item,
action=ContentInteraction.Action.FAVORITE,
)
for item in favorites
],
ignore_conflicts=True,
)
cards = Card.objects.filter(slug__in=payload.get("cards", [])[:100])
UserCard.objects.bulk_create(
[UserCard(user=user, card=card, source="visitor_migration") for card in cards],
ignore_conflicts=True,
)
migration.status = VisitorMigration.Status.APPLIED
migration.applied_at = timezone.now()
migration.save(update_fields=["status", "applied_at"])
return migration
+104
View File
@@ -0,0 +1,104 @@
# Generated by Django 4.2.23 on 2026-08-08 11:15
from django.conf import settings
import django.contrib.auth.models
import django.contrib.auth.validators
from django.db import migrations, models
import django.db.models.deletion
import django.utils.timezone
import uuid
class Migration(migrations.Migration):
initial = True
dependencies = [
('auth', '0012_alter_user_first_name_max_length'),
]
operations = [
migrations.CreateModel(
name='User',
fields=[
('password', models.CharField(max_length=128, verbose_name='password')),
('last_login', models.DateTimeField(blank=True, null=True, verbose_name='last login')),
('is_superuser', models.BooleanField(default=False, help_text='Designates that this user has all permissions without explicitly assigning them.', verbose_name='superuser status')),
('username', models.CharField(error_messages={'unique': 'A user with that username already exists.'}, help_text='Required. 150 characters or fewer. Letters, digits and @/./+/-/_ only.', max_length=150, unique=True, validators=[django.contrib.auth.validators.UnicodeUsernameValidator()], verbose_name='username')),
('first_name', models.CharField(blank=True, max_length=150, verbose_name='first name')),
('last_name', models.CharField(blank=True, max_length=150, verbose_name='last name')),
('email', models.EmailField(blank=True, max_length=254, verbose_name='email address')),
('is_staff', models.BooleanField(default=False, help_text='Designates whether the user can log into this admin site.', verbose_name='staff status')),
('is_active', models.BooleanField(default=True, help_text='Designates whether this user should be treated as active. Unselect this instead of deleting accounts.', verbose_name='active')),
('date_joined', models.DateTimeField(default=django.utils.timezone.now, verbose_name='date joined')),
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('nickname', models.CharField(max_length=40)),
('avatar_url', models.URLField(blank=True)),
('bio', models.CharField(blank=True, max_length=200)),
('track', models.CharField(choices=[('beginner', '入门'), ('standard', '标准'), ('advanced', '进阶'), ('open', 'Open')], default='standard', max_length=16)),
('rating', models.PositiveIntegerField(default=1000)),
('groups', models.ManyToManyField(blank=True, help_text='The groups this user belongs to. A user will get all permissions granted to each of their groups.', related_name='user_set', related_query_name='user', to='auth.group', verbose_name='groups')),
('user_permissions', models.ManyToManyField(blank=True, help_text='Specific permissions for this user.', related_name='user_set', related_query_name='user', to='auth.permission', verbose_name='user permissions')),
],
options={
'verbose_name': 'user',
'verbose_name_plural': 'users',
'abstract': False,
},
managers=[
('objects', django.contrib.auth.models.UserManager()),
],
),
migrations.CreateModel(
name='InviteCode',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('code', models.CharField(max_length=32, unique=True)),
('group', models.CharField(blank=True, max_length=80)),
('max_uses', models.PositiveIntegerField(default=1)),
('used_count', models.PositiveIntegerField(default=0)),
('expires_at', models.DateTimeField(blank=True, null=True)),
('is_active', models.BooleanField(default=True)),
('created_at', models.DateTimeField(auto_now_add=True)),
('created_by', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='invite_codes', to=settings.AUTH_USER_MODEL)),
],
),
migrations.CreateModel(
name='UserSession',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('session_key', models.CharField(max_length=40, unique=True)),
('user_agent', models.CharField(blank=True, max_length=300)),
('ip_address', models.GenericIPAddressField(blank=True, null=True)),
('last_seen_at', models.DateTimeField(auto_now=True)),
('revoked_at', models.DateTimeField(blank=True, null=True)),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='login_sessions', to=settings.AUTH_USER_MODEL)),
],
),
migrations.CreateModel(
name='InviteCodeUsage',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('used_at', models.DateTimeField(auto_now_add=True)),
('ip_address', models.GenericIPAddressField(blank=True, null=True)),
('invite_code', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='usages', to='accounts.invitecode')),
('user', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='invite_usage', to=settings.AUTH_USER_MODEL)),
],
),
migrations.CreateModel(
name='AuditLog',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('action', models.CharField(max_length=80)),
('target_type', models.CharField(max_length=80)),
('target_id', models.CharField(blank=True, max_length=80)),
('reason', models.CharField(blank=True, max_length=300)),
('metadata', models.JSONField(blank=True, default=dict)),
('created_at', models.DateTimeField(auto_now_add=True)),
('actor', models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, to=settings.AUTH_USER_MODEL)),
],
options={
'ordering': ['-created_at'],
},
),
]
@@ -0,0 +1,28 @@
# Generated by Django 4.2.23 on 2026-08-08 11:28
from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion
import uuid
class Migration(migrations.Migration):
dependencies = [
('accounts', '0001_initial'),
]
operations = [
migrations.CreateModel(
name='VisitorMigration',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('payload', models.JSONField(default=dict)),
('summary', models.JSONField(default=dict)),
('status', models.CharField(choices=[('pending', '待确认'), ('applied', '已迁移')], default='pending', max_length=16)),
('created_at', models.DateTimeField(auto_now_add=True)),
('applied_at', models.DateTimeField(blank=True, null=True)),
('user', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='visitor_migration', to=settings.AUTH_USER_MODEL)),
],
),
]
+90
View File
@@ -0,0 +1,90 @@
import uuid
from django.contrib.auth.models import AbstractUser
from django.db import models
from django.utils import timezone
class User(AbstractUser):
class Track(models.TextChoices):
BEGINNER = "beginner", "入门"
STANDARD = "standard", "标准"
ADVANCED = "advanced", "进阶"
OPEN = "open", "Open"
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
nickname = models.CharField(max_length=40)
avatar_url = models.URLField(blank=True)
bio = models.CharField(max_length=200, blank=True)
track = models.CharField(max_length=16, choices=Track.choices, default=Track.STANDARD)
rating = models.PositiveIntegerField(default=1000)
def __str__(self):
return self.nickname or self.username
class InviteCode(models.Model):
code = models.CharField(max_length=32, unique=True)
group = models.CharField(max_length=80, blank=True)
max_uses = models.PositiveIntegerField(default=1)
used_count = models.PositiveIntegerField(default=0)
expires_at = models.DateTimeField(null=True, blank=True)
is_active = models.BooleanField(default=True)
created_by = models.ForeignKey(
User, null=True, blank=True, on_delete=models.SET_NULL, related_name="invite_codes"
)
created_at = models.DateTimeField(auto_now_add=True)
@property
def is_available(self):
return (
self.is_active
and self.used_count < self.max_uses
and (self.expires_at is None or self.expires_at > timezone.now())
)
def __str__(self):
return self.code
class InviteCodeUsage(models.Model):
invite_code = models.ForeignKey(InviteCode, on_delete=models.PROTECT, related_name="usages")
user = models.OneToOneField(User, on_delete=models.CASCADE, related_name="invite_usage")
used_at = models.DateTimeField(auto_now_add=True)
ip_address = models.GenericIPAddressField(null=True, blank=True)
class UserSession(models.Model):
user = models.ForeignKey(User, on_delete=models.CASCADE, related_name="login_sessions")
session_key = models.CharField(max_length=40, unique=True)
user_agent = models.CharField(max_length=300, blank=True)
ip_address = models.GenericIPAddressField(null=True, blank=True)
last_seen_at = models.DateTimeField(auto_now=True)
revoked_at = models.DateTimeField(null=True, blank=True)
class VisitorMigration(models.Model):
class Status(models.TextChoices):
PENDING = "pending", "待确认"
APPLIED = "applied", "已迁移"
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
user = models.OneToOneField(User, on_delete=models.CASCADE, related_name="visitor_migration")
payload = models.JSONField(default=dict)
summary = models.JSONField(default=dict)
status = models.CharField(max_length=16, choices=Status.choices, default=Status.PENDING)
created_at = models.DateTimeField(auto_now_add=True)
applied_at = models.DateTimeField(null=True, blank=True)
class AuditLog(models.Model):
actor = models.ForeignKey(User, null=True, on_delete=models.SET_NULL)
action = models.CharField(max_length=80)
target_type = models.CharField(max_length=80)
target_id = models.CharField(max_length=80, blank=True)
reason = models.CharField(max_length=300, blank=True)
metadata = models.JSONField(default=dict, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
ordering = ["-created_at"]
+72
View File
@@ -0,0 +1,72 @@
import json
from django.contrib.auth.password_validation import validate_password
from django.db import transaction
from rest_framework import serializers
from .migration_service import migration_summary
from .models import InviteCode, InviteCodeUsage, User, VisitorMigration
class UserSerializer(serializers.ModelSerializer):
class Meta:
model = User
fields = ("id", "username", "nickname", "avatar_url", "bio", "track", "rating")
read_only_fields = ("id", "rating")
class RegisterSerializer(serializers.Serializer):
invite_code = serializers.CharField(max_length=32)
username = serializers.RegexField(r"^[A-Za-z0-9_]{3,30}$")
password = serializers.CharField(write_only=True, min_length=8)
nickname = serializers.CharField(max_length=40)
track = serializers.ChoiceField(choices=User.Track.choices, default=User.Track.STANDARD)
visitor_data = serializers.JSONField(required=False, default=dict, write_only=True)
def validate_username(self, value):
if User.objects.filter(username__iexact=value).exists():
raise serializers.ValidationError("用户名已存在")
return value
def validate_password(self, value):
validate_password(value)
return value
def validate_visitor_data(self, value):
if not isinstance(value, dict):
raise serializers.ValidationError("游客数据必须是对象")
for field in ("story_saves", "favorites", "cards"):
if field in value and not isinstance(value[field], list):
raise serializers.ValidationError(f"{field} 必须是数组")
if "mathbti" in value and not isinstance(value["mathbti"], dict):
raise serializers.ValidationError("mathbti 必须是对象")
if len(json.dumps(value, ensure_ascii=False).encode("utf-8")) > 256 * 1024:
raise serializers.ValidationError("游客数据不能超过 256KB")
return value
@transaction.atomic
def create(self, validated_data):
code_text = validated_data.pop("invite_code")
visitor_data = validated_data.pop("visitor_data", {})
try:
invite = InviteCode.objects.select_for_update().get(code__iexact=code_text)
except InviteCode.DoesNotExist as exc:
raise serializers.ValidationError({"invite_code": "邀请码无效"}) from exc
if not invite.is_available:
raise serializers.ValidationError({"invite_code": "邀请码已过期或已用完"})
user = User.objects.create_user(**validated_data)
InviteCodeUsage.objects.create(
invite_code=invite,
user=user,
ip_address=self.context.get("ip_address"),
)
invite.used_count += 1
invite.save(update_fields=["used_count"])
if visitor_data:
VisitorMigration.objects.create(
user=user,
payload=visitor_data,
summary=migration_summary(visitor_data),
)
return user
@@ -0,0 +1,99 @@
import pytest
from accounts.migration_service import apply_visitor_migration, migration_summary
from accounts.models import User, VisitorMigration
from content.models import ContentInteraction, ContentItem
from math_life.models import (
MathBTIAssessment,
MathBTIResult,
MathIdentity,
Story,
StoryRun,
StoryVersion,
)
from progression.models import Card, UserCard
@pytest.mark.django_db
def test_apply_visitor_migration_确认后事务化导入且可重复调用():
user = User.objects.create_user(
username="visitor_user",
password="StrongPass_2026",
nickname="游客用户",
)
definition = {
"axes": [{"id": "style"}],
"scoring": {"axes": ["style"], "cutoff": 0},
"questions": [
{"id": 1, "axis": "style", "options": [{"score": 0}, {"score": 1}]}
],
}
MathBTIAssessment.objects.create(
version="test-1",
title="测试 MathBTI",
definition=definition,
is_published=True,
)
MathIdentity.objects.create(
code="1",
name="测试身份",
clan="信仰者",
mathematician="高斯",
)
story = Story.objects.create(slug="visitor-story", title="游客人生")
StoryVersion.objects.create(
story=story,
version=1,
is_published=True,
content={
"start_node": "start",
"nodes": {
"start": {"scene": "开始", "choices": []},
"saved": {"scene": "存档", "choices": []},
},
},
)
content = ContentItem.objects.create(
slug="favorite-content",
title="收藏内容",
kind=ContentItem.Kind.KNOWLEDGE,
is_published=True,
)
legacy_card = Card.objects.create(
slug="legacy-card",
name="旧人物卡",
mathematician="欧拉",
description="游客阶段获得",
)
payload = {
"mathbti": {"version": "test-1", "answers": [1]},
"story_saves": [
{
"story_slug": "visitor-story",
"version": 1,
"current_node": "saved",
"state": {"energy": 3},
}
],
"favorites": ["favorite-content"],
"cards": ["legacy-card"],
}
VisitorMigration.objects.create(
user=user,
payload=payload,
summary=migration_summary(payload),
)
first = apply_visitor_migration(user)
second = apply_visitor_migration(user)
assert first.status == VisitorMigration.Status.APPLIED
assert second.status == VisitorMigration.Status.APPLIED
assert MathBTIResult.objects.filter(user=user).count() == 1
assert StoryRun.objects.get(user=user).current_node == "saved"
assert ContentInteraction.objects.filter(
user=user,
content=content,
action=ContentInteraction.Action.FAVORITE,
).count() == 1
assert UserCard.objects.filter(user=user, card=legacy_card).count() == 1
+76
View File
@@ -0,0 +1,76 @@
from datetime import timedelta
import pytest
from django.utils import timezone
from rest_framework import serializers
from accounts.models import InviteCode, InviteCodeUsage, User
from accounts.serializers import RegisterSerializer
def registration_data(code="VALID-CODE", username="math_user"):
return {
"invite_code": code,
"username": username,
"password": "StrongPass_2026",
"nickname": "数学少年",
"track": User.Track.STANDARD,
}
@pytest.mark.django_db
def test_register_serializer_有效邀请码创建用户并原子消费():
invite = InviteCode.objects.create(code="VALID-CODE", max_uses=1)
serializer = RegisterSerializer(
data=registration_data(),
context={"ip_address": "127.0.0.1"},
)
assert serializer.is_valid(), serializer.errors
user = serializer.save()
invite.refresh_from_db()
assert invite.used_count == 1
assert user.check_password("StrongPass_2026")
usage = InviteCodeUsage.objects.get(user=user)
assert usage.invite_code == invite
assert usage.ip_address == "127.0.0.1"
@pytest.mark.django_db
def test_register_serializer_过期邀请码拒绝创建用户():
InviteCode.objects.create(
code="EXPIRED",
max_uses=1,
expires_at=timezone.now() - timedelta(seconds=1),
)
serializer = RegisterSerializer(data=registration_data(code="EXPIRED"))
assert serializer.is_valid(), serializer.errors
with pytest.raises(serializers.ValidationError, match="已过期或已用完"):
serializer.save()
assert User.objects.count() == 0
@pytest.mark.django_db
def test_register_serializer_用户名大小写重复时校验失败():
User.objects.create_user(
username="Math_User",
password="StrongPass_2026",
nickname="已有用户",
)
serializer = RegisterSerializer(data=registration_data(username="math_user"))
assert not serializer.is_valid()
assert "用户名已存在" in str(serializer.errors["username"])
@pytest.mark.django_db
def test_register_serializer_游客迁移字段类型错误时拒绝():
InviteCode.objects.create(code="VALID-CODE", max_uses=1)
data = registration_data()
data["visitor_data"] = {"favorites": None}
serializer = RegisterSerializer(data=data)
assert not serializer.is_valid()
assert "favorites 必须是数组" in str(serializer.errors["visitor_data"])
+3
View File
@@ -0,0 +1,3 @@
from django.test import TestCase
# Create your tests here.
+12
View File
@@ -0,0 +1,12 @@
from django.urls import path
from .views import LoginView, LogoutView, MeView, RegisterView, VisitorMigrationView
urlpatterns = [
path("register/", RegisterView.as_view(), name="register"),
path("login/", LoginView.as_view(), name="login"),
path("logout/", LogoutView.as_view(), name="logout"),
path("me/", MeView.as_view(), name="me"),
path("visitor-migration/", VisitorMigrationView.as_view(), name="visitor-migration"),
]
+111
View File
@@ -0,0 +1,111 @@
from django.contrib.auth import authenticate, login, logout
from rest_framework import permissions, status
from rest_framework.response import Response
from rest_framework.views import APIView
from .migration_service import apply_visitor_migration
from .models import UserSession, VisitorMigration
from .serializers import RegisterSerializer, UserSerializer
def client_ip(request):
forwarded = request.META.get("HTTP_X_FORWARDED_FOR")
return forwarded.split(",")[0].strip() if forwarded else request.META.get("REMOTE_ADDR")
class RegisterView(APIView):
permission_classes = [permissions.AllowAny]
def post(self, request):
serializer = RegisterSerializer(
data=request.data,
context={"ip_address": client_ip(request)},
)
serializer.is_valid(raise_exception=True)
user = serializer.save()
login(request, user)
self._record_session(request, user)
payload = UserSerializer(user).data
payload["visitor_migration_pending"] = hasattr(user, "visitor_migration")
return Response(payload, status=status.HTTP_201_CREATED)
@staticmethod
def _record_session(request, user):
if not request.session.session_key:
request.session.save()
UserSession.objects.update_or_create(
session_key=request.session.session_key,
defaults={
"user": user,
"user_agent": request.headers.get("User-Agent", "")[:300],
"ip_address": client_ip(request),
},
)
class LoginView(APIView):
permission_classes = [permissions.AllowAny]
def post(self, request):
user = authenticate(
request,
username=request.data.get("username", ""),
password=request.data.get("password", ""),
)
if user is None or not user.is_active:
return Response(
{"error": {"code": "invalid_credentials", "message": "用户名或密码错误"}},
status=status.HTTP_400_BAD_REQUEST,
)
login(request, user)
RegisterView._record_session(request, user)
return Response(UserSerializer(user).data)
class LogoutView(APIView):
def post(self, request):
if request.session.session_key:
UserSession.objects.filter(session_key=request.session.session_key).delete()
logout(request)
return Response(status=status.HTTP_204_NO_CONTENT)
class MeView(APIView):
def get(self, request):
return Response(UserSerializer(request.user).data)
def patch(self, request):
serializer = UserSerializer(request.user, data=request.data, partial=True)
serializer.is_valid(raise_exception=True)
serializer.save()
return Response(serializer.data)
class VisitorMigrationView(APIView):
def get(self, request):
migration = VisitorMigration.objects.filter(user=request.user).first()
if migration is None:
return Response({"pending": False, "summary": {}})
return Response(
{
"pending": migration.status == VisitorMigration.Status.PENDING,
"status": migration.status,
"summary": migration.summary,
}
)
def post(self, request):
migration = VisitorMigration.objects.filter(user=request.user).first()
if migration is None:
return Response(
{"error": {"code": "no_visitor_data", "message": "没有待迁移的游客数据"}},
status=status.HTTP_404_NOT_FOUND,
)
migration = apply_visitor_migration(request.user)
return Response(
{
"pending": False,
"status": migration.status,
"summary": migration.summary,
}
)
View File
+3
View File
@@ -0,0 +1,3 @@
from django.contrib import admin
# Register your models here.
+18
View File
@@ -0,0 +1,18 @@
from rest_framework.views import exception_handler as drf_exception_handler
def exception_handler(exc, context):
response = drf_exception_handler(exc, context)
if response is None:
return response
request = context.get("request")
response.data = {
"error": {
"code": getattr(exc, "default_code", "request_error"),
"message": "请求未能完成",
"details": response.data,
"request_id": getattr(request, "request_id", None),
}
}
return response
+6
View File
@@ -0,0 +1,6 @@
from django.apps import AppConfig
class CommonConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'common'
+17
View File
@@ -0,0 +1,17 @@
from channels.generic.websocket import AsyncJsonWebsocketConsumer
class HealthConsumer(AsyncJsonWebsocketConsumer):
group_name = "healthcheck"
async def connect(self):
try:
await self.channel_layer.group_add(self.group_name, self.channel_name)
await self.channel_layer.group_discard(self.group_name, self.channel_name)
except Exception:
await self.close(code=1011)
return
await self.accept()
await self.send_json({"status": "ok", "channel_layer": "ok"})
await self.close()
+11
View File
@@ -0,0 +1,11 @@
import contextvars
import logging
request_id_context = contextvars.ContextVar("request_id", default="-")
class RequestIDFilter(logging.Filter):
def filter(self, record):
record.request_id = request_id_context.get()
return True
+1
View File
@@ -0,0 +1 @@
@@ -0,0 +1 @@
@@ -0,0 +1,63 @@
import re
from django.core.management.base import BaseCommand, CommandError
from django.db import connection
MINIMUM_VERSION = (8, 0, 35)
class Command(BaseCommand):
help = "检查生产 MySQL 的版本、字符集、引擎、严格模式和事务隔离级别"
def handle(self, *args, **options):
if connection.vendor != "mysql":
raise CommandError(f"数据库必须是 MySQL,当前为 {connection.vendor}")
with connection.cursor() as cursor:
cursor.execute(
"""
SELECT VERSION(),
@@character_set_server,
@@collation_server,
@@sql_mode,
@@transaction_isolation,
@@default_storage_engine
"""
)
(
version_text,
charset,
collation,
sql_mode,
isolation,
storage_engine,
) = cursor.fetchone()
match = re.match(r"^(\d+)\.(\d+)\.(\d+)", version_text)
if match is None:
raise CommandError(f"无法解析 MySQL 版本: {version_text}")
version = tuple(int(part) for part in match.groups())
if version < MINIMUM_VERSION:
minimum = ".".join(str(part) for part in MINIMUM_VERSION)
raise CommandError(f"MySQL 版本过低: {version_text},最低要求 {minimum}")
if charset.lower() != "utf8mb4":
raise CommandError(f"character_set_server 必须是 utf8mb4,当前为 {charset}")
if not collation.lower().startswith("utf8mb4_"):
raise CommandError(f"collation_server 必须属于 utf8mb4,当前为 {collation}")
modes = {mode.strip().upper() for mode in sql_mode.split(",")}
if not {"STRICT_TRANS_TABLES", "STRICT_ALL_TABLES"} & modes:
raise CommandError(f"MySQL 未启用严格模式: {sql_mode}")
if isolation.upper().replace("_", "-") != "READ-COMMITTED":
raise CommandError(f"事务隔离级别必须是 READ-COMMITTED,当前为 {isolation}")
if storage_engine.lower() != "innodb":
raise CommandError(f"默认存储引擎必须是 InnoDB,当前为 {storage_engine}")
self.stdout.write(
self.style.SUCCESS(
"MySQL 检查通过: "
f"version={version_text}, charset={charset}, "
f"collation={collation}, isolation={isolation}, engine={storage_engine}"
)
)
+19
View File
@@ -0,0 +1,19 @@
import uuid
from .logging import request_id_context
class RequestIDMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
request_id = request.headers.get("X-Request-ID") or str(uuid.uuid4())
request.request_id = request_id
token = request_id_context.set(request_id)
try:
response = self.get_response(request)
response["X-Request-ID"] = request_id
return response
finally:
request_id_context.reset(token)
+3
View File
@@ -0,0 +1,3 @@
from django.db import models
# Create your models here.
+3
View File
@@ -0,0 +1,3 @@
from django.test import TestCase
# Create your tests here.
+14
View File
@@ -0,0 +1,14 @@
from django.db import connection
from django.http import JsonResponse
from django.shortcuts import render
def home(request):
return render(request, "index.html")
def health(request):
with connection.cursor() as cursor:
cursor.execute("SELECT 1")
cursor.fetchone()
return JsonResponse({"status": "ok", "database": "ok"})
View File
+29
View File
@@ -0,0 +1,29 @@
import os
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")
from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from django.core.asgi import get_asgi_application
from django.urls import path
django_asgi_application = get_asgi_application()
from common.consumers import HealthConsumer
from contest.routing import websocket_urlpatterns
application = ProtocolTypeRouter(
{
"http": django_asgi_application,
"websocket": AuthMiddlewareStack(
URLRouter(
[
path("ws/health/", HealthConsumer.as_asgi()),
*websocket_urlpatterns,
]
)
),
}
)
+196
View File
@@ -0,0 +1,196 @@
import os
from pathlib import Path
import dj_database_url
from django.core.exceptions import ImproperlyConfigured
BASE_DIR = Path(__file__).resolve().parent.parent
PROJECT_ROOT = BASE_DIR.parent
SECRET_KEY = os.getenv("DJANGO_SECRET_KEY", "dev-only-change-before-production")
DEBUG = os.getenv("DJANGO_DEBUG", "true").lower() == "true"
if not DEBUG and SECRET_KEY == "dev-only-change-before-production":
raise ImproperlyConfigured("生产环境必须设置 DJANGO_SECRET_KEY")
ALLOWED_HOSTS = [
host.strip()
for host in os.getenv("DJANGO_ALLOWED_HOSTS", "localhost,127.0.0.1,testserver").split(",")
if host.strip()
]
INSTALLED_APPS = [
"django.contrib.admin",
"django.contrib.auth",
"django.contrib.contenttypes",
"django.contrib.sessions",
"django.contrib.messages",
"django.contrib.staticfiles",
"corsheaders",
"rest_framework",
"channels",
"accounts",
"math_life",
"contest",
"progression",
"latex_lab",
"content",
"engagement",
"common",
]
MIDDLEWARE = [
"django.middleware.security.SecurityMiddleware",
"whitenoise.middleware.WhiteNoiseMiddleware",
"corsheaders.middleware.CorsMiddleware",
"django.contrib.sessions.middleware.SessionMiddleware",
"django.middleware.common.CommonMiddleware",
"django.middleware.csrf.CsrfViewMiddleware",
"django.contrib.auth.middleware.AuthenticationMiddleware",
"django.contrib.messages.middleware.MessageMiddleware",
"django.middleware.clickjacking.XFrameOptionsMiddleware",
"common.middleware.RequestIDMiddleware",
]
ROOT_URLCONF = "config.urls"
TEMPLATES = [
{
"BACKEND": "django.template.backends.django.DjangoTemplates",
"DIRS": [BASE_DIR / "templates"],
"APP_DIRS": True,
"OPTIONS": {
"context_processors": [
"django.template.context_processors.debug",
"django.template.context_processors.request",
"django.contrib.auth.context_processors.auth",
"django.contrib.messages.context_processors.messages",
],
},
},
]
WSGI_APPLICATION = "config.wsgi.application"
ASGI_APPLICATION = "config.asgi.application"
DATABASE_URL = os.getenv("DATABASE_URL")
if not DEBUG and not DATABASE_URL:
raise ImproperlyConfigured("生产环境必须设置 MySQL DATABASE_URL")
if not DEBUG and not DATABASE_URL.startswith("mysql://"):
raise ImproperlyConfigured("生产环境 DATABASE_URL 必须使用 MySQL")
DATABASES = {
"default": dj_database_url.parse(
DATABASE_URL or f"sqlite:///{BASE_DIR / 'db.sqlite3'}",
conn_max_age=60,
conn_health_checks=True,
)
}
if DATABASES["default"]["ENGINE"] == "django.db.backends.mysql":
DATABASES["default"]["OPTIONS"] = {
"charset": "utf8mb4",
"init_command": (
"SET sql_mode='STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,"
"NO_ENGINE_SUBSTITUTION'"
),
"isolation_level": "read committed",
}
AUTH_PASSWORD_VALIDATORS = [
{"NAME": "django.contrib.auth.password_validation.UserAttributeSimilarityValidator"},
{"NAME": "django.contrib.auth.password_validation.MinimumLengthValidator"},
{"NAME": "django.contrib.auth.password_validation.CommonPasswordValidator"},
{"NAME": "django.contrib.auth.password_validation.NumericPasswordValidator"},
]
AUTH_USER_MODEL = "accounts.User"
LANGUAGE_CODE = "zh-hans"
TIME_ZONE = "Asia/Shanghai"
USE_I18N = True
USE_TZ = True
STATIC_URL = "/static/"
STATIC_ROOT = BASE_DIR / "staticfiles"
STATICFILES_DIRS = [BASE_DIR / "static"]
STORAGES = {
"default": {"BACKEND": "django.core.files.storage.FileSystemStorage"},
"staticfiles": {"BACKEND": "whitenoise.storage.CompressedManifestStaticFilesStorage"},
}
DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField"
REST_FRAMEWORK = {
"DEFAULT_AUTHENTICATION_CLASSES": [
"rest_framework.authentication.SessionAuthentication",
],
"DEFAULT_PERMISSION_CLASSES": [
"rest_framework.permissions.IsAuthenticated",
],
"DEFAULT_PAGINATION_CLASS": "rest_framework.pagination.PageNumberPagination",
"PAGE_SIZE": 20,
"EXCEPTION_HANDLER": "common.api.exception_handler",
"DEFAULT_THROTTLE_CLASSES": [
"rest_framework.throttling.AnonRateThrottle",
"rest_framework.throttling.UserRateThrottle",
],
"DEFAULT_THROTTLE_RATES": {
"anon": os.getenv("API_ANON_RATE", "120/minute"),
"user": os.getenv("API_USER_RATE", "600/minute"),
},
}
CORS_ALLOWED_ORIGINS = [
origin.strip()
for origin in os.getenv("CORS_ALLOWED_ORIGINS", "http://localhost:3000").split(",")
if origin.strip()
]
CORS_ALLOW_CREDENTIALS = True
CSRF_TRUSTED_ORIGINS = [
origin.strip()
for origin in os.getenv("CSRF_TRUSTED_ORIGINS", "").split(",")
if origin.strip()
]
REDIS_URL = os.getenv("REDIS_URL")
if not DEBUG and not REDIS_URL:
raise ImproperlyConfigured("生产环境必须设置 Redis REDIS_URL")
if REDIS_URL:
CHANNEL_LAYERS = {
"default": {
"BACKEND": "channels_redis.core.RedisChannelLayer",
"CONFIG": {"hosts": [REDIS_URL]},
}
}
else:
CHANNEL_LAYERS = {
"default": {"BACKEND": "channels.layers.InMemoryChannelLayer"}
}
SESSION_COOKIE_HTTPONLY = True
SESSION_COOKIE_SAMESITE = "Lax"
CSRF_COOKIE_SAMESITE = "Lax"
SECURE_PROXY_SSL_HEADER = ("HTTP_X_FORWARDED_PROTO", "https")
SECURE_CONTENT_TYPE_NOSNIFF = True
X_FRAME_OPTIONS = "DENY"
SECURE_SSL_REDIRECT = not DEBUG
SESSION_COOKIE_SECURE = not DEBUG
CSRF_COOKIE_SECURE = not DEBUG
SECURE_HSTS_SECONDS = int(os.getenv("SECURE_HSTS_SECONDS", "31536000" if not DEBUG else "0"))
SECURE_HSTS_INCLUDE_SUBDOMAINS = not DEBUG
SECURE_HSTS_PRELOAD = not DEBUG
LOGGING = {
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"standard": {
"format": "%(asctime)s %(levelname)s %(name)s request_id=%(request_id)s %(message)s"
}
},
"filters": {"request_id": {"()": "common.logging.RequestIDFilter"}},
"handlers": {
"console": {
"class": "logging.StreamHandler",
"formatter": "standard",
"filters": ["request_id"],
}
},
"root": {"handlers": ["console"], "level": "INFO"},
}
+21
View File
@@ -0,0 +1,21 @@
from django.contrib import admin
from django.urls import include, path
from common.views import health, home
admin.site.site_header = "葫芦数学运营后台"
admin.site.site_title = "葫芦数学"
admin.site.index_title = "内容与运营"
urlpatterns = [
path("", home, name="home"),
path("health/", health, name="health"),
path("admin/", admin.site.urls),
path("api/v1/accounts/", include("accounts.urls")),
path("api/v1/math-life/", include("math_life.urls")),
path("api/v1/contests/", include("contest.urls")),
path("api/v1/latex/", include("latex_lab.urls")),
path("api/v1/content/", include("content.urls")),
path("api/v1/progression/", include("progression.urls")),
]
+16
View File
@@ -0,0 +1,16 @@
"""
WSGI config for config project.
It exposes the WSGI callable as a module-level variable named ``application``.
For more information on this file, see
https://docs.djangoproject.com/en/4.2/howto/deployment/wsgi/
"""
import os
from django.core.wsgi import get_wsgi_application
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')
application = get_wsgi_application()
View File
+21
View File
@@ -0,0 +1,21 @@
from django.contrib import admin
from .models import ContentInteraction, ContentItem, VideoProgress
@admin.register(ContentItem)
class ContentItemAdmin(admin.ModelAdmin):
list_display = (
"title",
"kind",
"ability_dimension",
"discipline",
"duration_seconds",
"is_published",
)
list_filter = ("kind", "ability_dimension", "discipline", "is_published")
search_fields = ("title", "summary", "body")
admin.site.register(ContentInteraction)
admin.site.register(VideoProgress)
+6
View File
@@ -0,0 +1,6 @@
from django.apps import AppConfig
class ContentConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'content'
@@ -0,0 +1,51 @@
# Generated by Django 4.2.23 on 2026-08-08 11:15
from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
initial = True
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.CreateModel(
name='ContentItem',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('slug', models.SlugField(unique=True)),
('title', models.CharField(max_length=160)),
('kind', models.CharField(choices=[('video', '视频'), ('knowledge', '知识卡片'), ('person', '数学人物')], max_length=16)),
('summary', models.TextField(blank=True)),
('body', models.TextField(blank=True)),
('cover_url', models.URLField(blank=True)),
('media_url', models.URLField(blank=True)),
('topics', models.JSONField(blank=True, default=list)),
('source', models.CharField(blank=True, max_length=300)),
('is_published', models.BooleanField(default=False)),
('published_at', models.DateTimeField(blank=True, null=True)),
],
options={
'ordering': ['-published_at'],
},
),
migrations.CreateModel(
name='ContentInteraction',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('action', models.CharField(choices=[('view', '浏览'), ('favorite', '收藏'), ('complete', '完成')], max_length=16)),
('created_at', models.DateTimeField(auto_now_add=True)),
('content', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='interactions', to='content.contentitem')),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
],
),
migrations.AddConstraint(
model_name='contentinteraction',
constraint=models.UniqueConstraint(fields=('user', 'content', 'action'), name='unique_content_interaction'),
),
]
@@ -0,0 +1,78 @@
# Generated by Django 4.2.23 on 2026-08-08 11:47
from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
('content', '0001_initial'),
]
operations = [
migrations.AddField(
model_name='contentitem',
name='ability_dimension',
field=models.CharField(blank=True, choices=[('vision', '数学眼光'), ('humanities', '数学人文'), ('detection', '数学侦探'), ('modeling', '数学建模'), ('connection', '数学联结')], max_length=20),
),
migrations.AddField(
model_name='contentitem',
name='author',
field=models.CharField(blank=True, max_length=120),
),
migrations.AddField(
model_name='contentitem',
name='comment_count',
field=models.PositiveIntegerField(default=0),
),
migrations.AddField(
model_name='contentitem',
name='discipline',
field=models.CharField(blank=True, max_length=80),
),
migrations.AddField(
model_name='contentitem',
name='discipline_icon',
field=models.CharField(blank=True, max_length=10),
),
migrations.AddField(
model_name='contentitem',
name='duration_seconds',
field=models.PositiveIntegerField(default=0),
),
migrations.AddField(
model_name='contentitem',
name='module',
field=models.CharField(blank=True, max_length=40),
),
migrations.AddField(
model_name='contentitem',
name='sub_category',
field=models.CharField(blank=True, max_length=80),
),
migrations.AddField(
model_name='contentitem',
name='view_count',
field=models.PositiveIntegerField(default=0),
),
migrations.CreateModel(
name='VideoProgress',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('position_seconds', models.PositiveIntegerField(default=0)),
('completed', models.BooleanField(default=False)),
('reward_granted', models.BooleanField(default=False)),
('updated_at', models.DateTimeField(auto_now=True)),
('completed_at', models.DateTimeField(blank=True, null=True)),
('content', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='video_progress', to='content.contentitem')),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='video_progress', to=settings.AUTH_USER_MODEL)),
],
),
migrations.AddConstraint(
model_name='videoprogress',
constraint=models.UniqueConstraint(fields=('user', 'content'), name='unique_user_video_progress'),
),
]
+85
View File
@@ -0,0 +1,85 @@
from django.conf import settings
from django.db import models
class ContentItem(models.Model):
class Ability(models.TextChoices):
VISION = "vision", "数学眼光"
HUMANITIES = "humanities", "数学人文"
DETECTION = "detection", "数学侦探"
MODELING = "modeling", "数学建模"
CONNECTION = "connection", "数学联结"
class Kind(models.TextChoices):
VIDEO = "video", "视频"
KNOWLEDGE = "knowledge", "知识卡片"
PERSON = "person", "数学人物"
slug = models.SlugField(unique=True)
title = models.CharField(max_length=160)
kind = models.CharField(max_length=16, choices=Kind.choices)
summary = models.TextField(blank=True)
body = models.TextField(blank=True)
cover_url = models.URLField(blank=True)
media_url = models.URLField(blank=True)
topics = models.JSONField(default=list, blank=True)
source = models.CharField(max_length=300, blank=True)
ability_dimension = models.CharField(
max_length=20, choices=Ability.choices, blank=True
)
module = models.CharField(max_length=40, blank=True)
sub_category = models.CharField(max_length=80, blank=True)
discipline = models.CharField(max_length=80, blank=True)
discipline_icon = models.CharField(max_length=10, blank=True)
author = models.CharField(max_length=120, blank=True)
duration_seconds = models.PositiveIntegerField(default=0)
view_count = models.PositiveIntegerField(default=0)
comment_count = models.PositiveIntegerField(default=0)
is_published = models.BooleanField(default=False)
published_at = models.DateTimeField(null=True, blank=True)
class Meta:
ordering = ["-published_at"]
def __str__(self):
return self.title
class ContentInteraction(models.Model):
class Action(models.TextChoices):
VIEW = "view", "浏览"
FAVORITE = "favorite", "收藏"
COMPLETE = "complete", "完成"
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
content = models.ForeignKey(ContentItem, on_delete=models.CASCADE, related_name="interactions")
action = models.CharField(max_length=16, choices=Action.choices)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
constraints = [
models.UniqueConstraint(
fields=("user", "content", "action"), name="unique_content_interaction"
)
]
class VideoProgress(models.Model):
user = models.ForeignKey(
settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="video_progress"
)
content = models.ForeignKey(
ContentItem, on_delete=models.CASCADE, related_name="video_progress"
)
position_seconds = models.PositiveIntegerField(default=0)
completed = models.BooleanField(default=False)
reward_granted = models.BooleanField(default=False)
updated_at = models.DateTimeField(auto_now=True)
completed_at = models.DateTimeField(null=True, blank=True)
class Meta:
constraints = [
models.UniqueConstraint(
fields=("user", "content"), name="unique_user_video_progress"
)
]
+68
View File
@@ -0,0 +1,68 @@
from django.db import transaction
from django.utils import timezone
from rest_framework.exceptions import ValidationError
from progression.models import RewardTransaction, UserAbility
from .models import ContentInteraction, ContentItem, VideoProgress
ABILITY_TO_DIMENSION = {
ContentItem.Ability.VISION: UserAbility.Dimension.VISION,
ContentItem.Ability.HUMANITIES: UserAbility.Dimension.HUMANITIES,
ContentItem.Ability.DETECTION: UserAbility.Dimension.DETECTION,
ContentItem.Ability.MODELING: UserAbility.Dimension.MODELING,
ContentItem.Ability.CONNECTION: UserAbility.Dimension.CONNECTION,
}
@transaction.atomic
def complete_video(user, content):
if content.kind != ContentItem.Kind.VIDEO:
raise ValidationError("只有视频内容可以提交观看完成")
progress, _ = VideoProgress.objects.select_for_update().get_or_create(
user=user,
content=content,
)
progress.position_seconds = max(progress.position_seconds, content.duration_seconds)
progress.completed = True
progress.completed_at = progress.completed_at or timezone.now()
reward = None
dimension = ABILITY_TO_DIMENSION.get(content.ability_dimension)
if dimension and not progress.reward_granted:
transaction_key = f"video-complete:{content.id}"
reward_tx, created = RewardTransaction.objects.get_or_create(
user=user,
idempotency_key=transaction_key,
defaults={
"source": "video_complete",
"rewards": {"ability": dimension, "fragments": 1},
},
)
if created:
ability, _ = UserAbility.objects.select_for_update().get_or_create(
user=user,
dimension=dimension,
)
ability.fragments += 1
ability.save(update_fields=["fragments"])
progress.reward_granted = True
reward = reward_tx.rewards
progress.save(
update_fields=[
"position_seconds",
"completed",
"completed_at",
"reward_granted",
"updated_at",
]
)
ContentInteraction.objects.get_or_create(
user=user,
content=content,
action=ContentInteraction.Action.COMPLETE,
)
return progress, reward
+64
View File
@@ -0,0 +1,64 @@
import pytest
from rest_framework.exceptions import ValidationError
from accounts.models import User
from content.models import ContentInteraction, ContentItem, VideoProgress
from content.services import complete_video
from progression.models import RewardTransaction, UserAbility
@pytest.mark.django_db
def test_complete_video_重复提交只发放一次能力碎片():
user = User.objects.create_user(
username="video_user",
password="StrongPass_2026",
nickname="视频用户",
)
video = ContentItem.objects.create(
slug="video-modeling",
title="建模视频",
kind=ContentItem.Kind.VIDEO,
ability_dimension=ContentItem.Ability.MODELING,
duration_seconds=360,
is_published=True,
)
first, first_reward = complete_video(user, video)
second, second_reward = complete_video(user, video)
ability = UserAbility.objects.get(
user=user,
dimension=UserAbility.Dimension.MODELING,
)
assert first.completed is True
assert second.reward_granted is True
assert first_reward == {"ability": "modeling", "fragments": 1}
assert second_reward is None
assert ability.fragments == 1
assert RewardTransaction.objects.filter(user=user).count() == 1
assert VideoProgress.objects.filter(user=user, content=video).count() == 1
assert ContentInteraction.objects.filter(
user=user,
content=video,
action=ContentInteraction.Action.COMPLETE,
).count() == 1
@pytest.mark.django_db
def test_complete_video_非视频内容拒绝完成():
user = User.objects.create_user(
username="knowledge_user",
password="StrongPass_2026",
nickname="知识用户",
)
content = ContentItem.objects.create(
slug="knowledge-item",
title="知识卡片",
kind=ContentItem.Kind.KNOWLEDGE,
is_published=True,
)
with pytest.raises(ValidationError, match="只有视频"):
complete_video(user, content)
assert RewardTransaction.objects.count() == 0
+3
View File
@@ -0,0 +1,3 @@
from django.test import TestCase
# Create your tests here.
+18
View File
@@ -0,0 +1,18 @@
from django.urls import path
from .views import (
ContentDetailView,
ContentListView,
FavoriteView,
VideoCatalogView,
VideoCompleteView,
)
urlpatterns = [
path("", ContentListView.as_view(), name="content-list"),
path("videos/catalog/", VideoCatalogView.as_view(), name="video-catalog"),
path("<slug:slug>/", ContentDetailView.as_view(), name="content-detail"),
path("<slug:slug>/favorite/", FavoriteView.as_view(), name="content-favorite"),
path("<slug:slug>/complete/", VideoCompleteView.as_view(), name="video-complete"),
]
+181
View File
@@ -0,0 +1,181 @@
from django.db.models import Count
from django.shortcuts import get_object_or_404
from rest_framework import permissions, status
from rest_framework.response import Response
from rest_framework.views import APIView
from .models import ContentInteraction, ContentItem, VideoProgress
from .services import complete_video
ABILITY_META = {
ContentItem.Ability.VISION: {"label": "数学眼光", "icon": "", "color": "#5d73e8"},
ContentItem.Ability.HUMANITIES: {"label": "数学人文", "icon": "", "color": "#b76d38"},
ContentItem.Ability.DETECTION: {"label": "数学侦探", "icon": "", "color": "#287f78"},
ContentItem.Ability.MODELING: {"label": "数学建模", "icon": "", "color": "#d98628"},
ContentItem.Ability.CONNECTION: {"label": "数学联结", "icon": "", "color": "#8667b5"},
}
def serialize_item(item, progress=None):
return {
"slug": item.slug,
"title": item.title,
"kind": item.kind,
"summary": item.summary,
"body": item.body,
"cover_url": item.cover_url,
"media_url": item.media_url,
"topics": item.topics,
"source": item.source,
"ability_dimension": item.ability_dimension,
"ability": ABILITY_META.get(item.ability_dimension),
"module": item.module,
"sub_category": item.sub_category,
"discipline": item.discipline,
"discipline_icon": item.discipline_icon,
"author": item.author,
"duration_seconds": item.duration_seconds,
"view_count": item.view_count,
"comment_count": item.comment_count,
"published_at": item.published_at,
"progress": (
{
"position_seconds": progress.position_seconds,
"completed": progress.completed,
"reward_granted": progress.reward_granted,
}
if progress
else None
),
}
class ContentListView(APIView):
permission_classes = [permissions.AllowAny]
def get(self, request):
queryset = ContentItem.objects.filter(is_published=True)
kind = request.query_params.get("kind")
ability = request.query_params.get("ability")
discipline = request.query_params.get("discipline")
if kind:
queryset = queryset.filter(kind=kind)
if ability:
queryset = queryset.filter(ability_dimension=ability)
if discipline:
queryset = queryset.filter(discipline=discipline)
progress_by_content = {}
if request.user.is_authenticated:
progress_by_content = {
item.content_id: item
for item in VideoProgress.objects.filter(
user=request.user,
content__in=queryset,
)
}
return Response(
[
serialize_item(item, progress_by_content.get(item.id))
for item in queryset[:100]
]
)
class VideoCatalogView(APIView):
permission_classes = [permissions.AllowAny]
def get(self, request):
videos = ContentItem.objects.filter(
kind=ContentItem.Kind.VIDEO,
is_published=True,
)
counts = {
row["ability_dimension"]: row["count"]
for row in videos.values("ability_dimension").annotate(count=Count("id"))
}
abilities = [
{
"id": ability,
**meta,
"count": counts.get(ability, 0),
}
for ability, meta in ABILITY_META.items()
]
disciplines = [
{
"name": row["discipline"],
"icon": row["discipline_icon"],
"count": row["count"],
}
for row in videos.values("discipline", "discipline_icon")
.annotate(count=Count("id"))
.order_by("discipline")
]
return Response(
{
"total": videos.count(),
"abilities": abilities,
"disciplines": disciplines,
}
)
class ContentDetailView(APIView):
permission_classes = [permissions.AllowAny]
def get(self, request, slug):
item = get_object_or_404(ContentItem, slug=slug, is_published=True)
progress = None
if request.user.is_authenticated:
ContentInteraction.objects.get_or_create(
user=request.user,
content=item,
action=ContentInteraction.Action.VIEW,
)
progress, _ = VideoProgress.objects.get_or_create(
user=request.user,
content=item,
)
return Response(serialize_item(item, progress))
class VideoCompleteView(APIView):
def post(self, request, slug):
item = get_object_or_404(
ContentItem,
slug=slug,
kind=ContentItem.Kind.VIDEO,
is_published=True,
)
progress, reward = complete_video(request.user, item)
return Response(
{
"completed": progress.completed,
"reward_granted": progress.reward_granted,
"reward": reward,
}
)
class FavoriteView(APIView):
def post(self, request, slug):
item = get_object_or_404(ContentItem, slug=slug, is_published=True)
_, created = ContentInteraction.objects.get_or_create(
user=request.user,
content=item,
action=ContentInteraction.Action.FAVORITE,
)
return Response(
{"favorite": True},
status=status.HTTP_201_CREATED if created else status.HTTP_200_OK,
)
def delete(self, request, slug):
ContentInteraction.objects.filter(
user=request.user,
content__slug=slug,
action=ContentInteraction.Action.FAVORITE,
).delete()
return Response(status=status.HTTP_204_NO_CONTENT)
View File
+47
View File
@@ -0,0 +1,47 @@
from django.contrib import admin
from .models import (
CheatFlag,
Contest,
ContestAnswer,
ContestAttempt,
ContestQuestion,
LeaderboardSnapshot,
Question,
QuestionVersion,
RatingHistory,
RealtimeMatch,
)
class ContestQuestionInline(admin.TabularInline):
model = ContestQuestion
extra = 1
@admin.register(Contest)
class ContestAdmin(admin.ModelAdmin):
list_display = ("title", "kind", "track", "status", "starts_at", "ends_at")
list_filter = ("kind", "track", "status")
inlines = [ContestQuestionInline]
@admin.register(ContestAttempt)
class ContestAttemptAdmin(admin.ModelAdmin):
list_display = ("id", "user", "contest", "status", "score", "duration_ms", "started_at")
list_filter = ("status", "contest__kind", "contest__track")
readonly_fields = ("started_at", "submitted_at")
@admin.register(CheatFlag)
class CheatFlagAdmin(admin.ModelAdmin):
list_display = ("attempt", "reason", "status", "created_at")
list_filter = ("status", "reason")
admin.site.register(Question)
admin.site.register(QuestionVersion)
admin.site.register(ContestAnswer)
admin.site.register(RealtimeMatch)
admin.site.register(RatingHistory)
admin.site.register(LeaderboardSnapshot)
+6
View File
@@ -0,0 +1,6 @@
from django.apps import AppConfig
class ContestConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'contest'
+54
View File
@@ -0,0 +1,54 @@
from channels.db import database_sync_to_async
from channels.generic.websocket import AsyncJsonWebsocketConsumer
from .models import RealtimeMatch
class MatchConsumer(AsyncJsonWebsocketConsumer):
async def connect(self):
self.match_id = self.scope["url_route"]["kwargs"]["match_id"]
self.group_name = f"match_{self.match_id}"
user = self.scope["user"]
if not user.is_authenticated or not await self._is_participant(user.id):
await self.close(code=4403)
return
await self.channel_layer.group_add(self.group_name, self.channel_name)
await self.accept()
await self.send_json({"type": "connected", "match_id": str(self.match_id)})
async def disconnect(self, close_code):
if hasattr(self, "group_name"):
await self.channel_layer.group_discard(self.group_name, self.channel_name)
async def receive_json(self, content, **kwargs):
event_type = content.get("type")
if event_type == "ping":
await self.send_json({"type": "pong"})
return
if event_type == "progress":
answered_count = max(0, min(int(content.get("answered_count", 0)), 100))
await self.channel_layer.group_send(
self.group_name,
{
"type": "match.progress",
"user_id": str(self.scope["user"].id),
"answered_count": answered_count,
},
)
async def match_progress(self, event):
await self.send_json(
{
"type": "progress",
"user_id": event["user_id"],
"answered_count": event["answered_count"],
}
)
@database_sync_to_async
def _is_participant(self, user_id):
return RealtimeMatch.objects.filter(id=self.match_id).filter(
player_one_id=user_id
).exists() or RealtimeMatch.objects.filter(
id=self.match_id, player_two_id=user_id
).exists()
+1
View File
@@ -0,0 +1 @@
@@ -0,0 +1 @@
@@ -0,0 +1,112 @@
from django.core.management.base import BaseCommand
from contest.models import Contest, ContestQuestion, Question, QuestionVersion
QUESTIONS = {
Question.Track.BEGINNER: [
("b-12-plus-19", "12 + 19", "31"),
("b-8-times-7", "8 × 7", "56"),
("b-90-minus-37", "90 - 37", "53"),
("b-144-div-12", "144 ÷ 12", "12"),
("b-25-times-4", "25 × 4", "100"),
],
Question.Track.STANDARD: [
("s-17-times-23", "17 × 23", "391"),
("s-625-div-25", "625 ÷ 25", "25"),
("s-48-times-15", "48 × 15", "720"),
("s-1000-minus-387", "1000 - 387", "613"),
("s-35-squared", "35²", "1225"),
],
Question.Track.ADVANCED: [
("a-mod-2pow10", "2¹⁰ 除以 7 的余数", "2"),
("a-sum-1-50", "1 到 50 的整数和", "1275"),
("a-15-choose-2", "C(15,2)", "105"),
("a-sqrt-2025", "√2025", "45"),
("a-3pow6", "3⁶", "729"),
],
}
class Command(BaseCommand):
help = "创建首批分层题目、实时 1v1、每日赛和单人练习"
def handle(self, *args, **options):
versions = {}
for track, questions in QUESTIONS.items():
versions[track] = []
for slug, prompt, answer in questions:
question, _ = Question.objects.update_or_create(
slug=slug,
defaults={"track": track, "tags": ["口算"], "is_active": True},
)
version, _ = QuestionVersion.objects.update_or_create(
question=question,
version=1,
defaults={
"prompt": prompt,
"answer": answer,
"explanation": f"答案为 {answer}",
},
)
versions[track].append(version)
contest_specs = []
for track, label in (
(Question.Track.BEGINNER, "入门"),
(Question.Track.STANDARD, "标准"),
(Question.Track.ADVANCED, "进阶"),
):
contest_specs.extend(
[
(
f"realtime-{track}",
f"{label}实时 1v1",
Contest.Kind.REALTIME,
track,
60,
),
(
f"daily-{track}",
f"{label}今日挑战",
Contest.Kind.DAILY,
track,
180,
),
(
f"practice-{track}",
f"{label}单人闯关",
Contest.Kind.PRACTICE,
track,
300,
),
]
)
for slug, title, kind, track, duration in contest_specs:
contest, _ = Contest.objects.update_or_create(
slug=slug,
defaults={
"title": title,
"kind": kind,
"track": track,
"duration_seconds": duration,
"status": Contest.Status.PUBLISHED,
},
)
ContestQuestion.objects.filter(contest=contest).delete()
ContestQuestion.objects.bulk_create(
[
ContestQuestion(
contest=contest,
question_version=version,
order=index,
points=100,
)
for index, version in enumerate(versions[track], start=1)
]
)
self.stdout.write(
self.style.SUCCESS(
f"已创建 {sum(map(len, QUESTIONS.values()))} 道题和 {len(contest_specs)} 场比赛"
)
)
+166
View File
@@ -0,0 +1,166 @@
# Generated by Django 4.2.23 on 2026-08-08 11:15
from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion
import uuid
class Migration(migrations.Migration):
initial = True
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.CreateModel(
name='Contest',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('slug', models.SlugField(unique=True)),
('title', models.CharField(max_length=120)),
('kind', models.CharField(choices=[('realtime', '实时 1v1'), ('daily', '今日挑战'), ('practice', '单人闯关'), ('weekly', '主题周赛')], max_length=16)),
('track', models.CharField(choices=[('beginner', '入门'), ('standard', '标准'), ('advanced', '进阶'), ('open', 'Open')], max_length=16)),
('status', models.CharField(choices=[('draft', '草稿'), ('published', '已发布'), ('closed', '已结束')], default='draft', max_length=16)),
('duration_seconds', models.PositiveIntegerField(default=60)),
('starts_at', models.DateTimeField(blank=True, null=True)),
('ends_at', models.DateTimeField(blank=True, null=True)),
],
),
migrations.CreateModel(
name='Question',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('slug', models.SlugField(unique=True)),
('track', models.CharField(choices=[('beginner', '入门'), ('standard', '标准'), ('advanced', '进阶'), ('open', 'Open')], max_length=16)),
('tags', models.JSONField(blank=True, default=list)),
('is_active', models.BooleanField(default=True)),
],
),
migrations.CreateModel(
name='RealtimeMatch',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('status', models.CharField(choices=[('waiting', '等待对手'), ('active', '进行中'), ('completed', '已完成'), ('cancelled', '已取消')], default='waiting', max_length=16)),
('player_one_rating', models.PositiveIntegerField()),
('player_two_rating', models.PositiveIntegerField(blank=True, null=True)),
('created_at', models.DateTimeField(auto_now_add=True)),
('started_at', models.DateTimeField(blank=True, null=True)),
('completed_at', models.DateTimeField(blank=True, null=True)),
('contest', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, to='contest.contest')),
('player_one', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='matches_as_player_one', to=settings.AUTH_USER_MODEL)),
('player_two', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, related_name='matches_as_player_two', to=settings.AUTH_USER_MODEL)),
('winner', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, related_name='won_matches', to=settings.AUTH_USER_MODEL)),
],
),
migrations.CreateModel(
name='RatingHistory',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('rating_before', models.PositiveIntegerField()),
('rating_after', models.PositiveIntegerField()),
('delta', models.SmallIntegerField()),
('created_at', models.DateTimeField(auto_now_add=True)),
('match', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='rating_changes', to='contest.realtimematch')),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='rating_history', to=settings.AUTH_USER_MODEL)),
],
),
migrations.CreateModel(
name='QuestionVersion',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('version', models.PositiveIntegerField()),
('prompt', models.TextField()),
('answer', models.CharField(max_length=200)),
('explanation', models.TextField(blank=True)),
('metadata', models.JSONField(blank=True, default=dict)),
('created_at', models.DateTimeField(auto_now_add=True)),
('question', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='versions', to='contest.question')),
],
),
migrations.CreateModel(
name='LeaderboardSnapshot',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('track', models.CharField(choices=[('beginner', '入门'), ('standard', '标准'), ('advanced', '进阶'), ('open', 'Open')], max_length=16)),
('period', models.CharField(max_length=40)),
('rankings', models.JSONField(default=list)),
('generated_at', models.DateTimeField(auto_now_add=True)),
('contest', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='contest.contest')),
],
),
migrations.CreateModel(
name='ContestQuestion',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('order', models.PositiveIntegerField()),
('points', models.PositiveIntegerField(default=100)),
('contest', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='contest_questions', to='contest.contest')),
('question_version', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, to='contest.questionversion')),
],
options={
'ordering': ['order'],
},
),
migrations.CreateModel(
name='ContestAttempt',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('status', models.CharField(choices=[('active', '进行中'), ('submitted', '已提交'), ('expired', '已超时')], default='active', max_length=16)),
('score', models.PositiveIntegerField(default=0)),
('correct_count', models.PositiveIntegerField(default=0)),
('answer_count', models.PositiveIntegerField(default=0)),
('duration_ms', models.PositiveIntegerField(default=0)),
('submission_key', models.CharField(blank=True, max_length=80, null=True)),
('started_at', models.DateTimeField(auto_now_add=True)),
('submitted_at', models.DateTimeField(blank=True, null=True)),
('contest', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='attempts', to='contest.contest')),
('match', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, related_name='attempts', to='contest.realtimematch')),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='contest_attempts', to=settings.AUTH_USER_MODEL)),
],
options={
'ordering': ['-started_at'],
},
),
migrations.CreateModel(
name='ContestAnswer',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('submitted_answer', models.CharField(max_length=200)),
('is_correct', models.BooleanField()),
('elapsed_ms', models.PositiveIntegerField()),
('answered_at', models.DateTimeField(auto_now_add=True)),
('attempt', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='answers', to='contest.contestattempt')),
('contest_question', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, to='contest.contestquestion')),
],
),
migrations.CreateModel(
name='CheatFlag',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('reason', models.CharField(max_length=120)),
('evidence', models.JSONField(default=dict)),
('status', models.CharField(choices=[('open', '待处理'), ('dismissed', '已忽略'), ('confirmed', '已确认')], default='open', max_length=16)),
('created_at', models.DateTimeField(auto_now_add=True)),
('attempt', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='cheat_flags', to='contest.contestattempt')),
],
),
migrations.AddConstraint(
model_name='questionversion',
constraint=models.UniqueConstraint(fields=('question', 'version'), name='unique_question_version'),
),
migrations.AddConstraint(
model_name='contestquestion',
constraint=models.UniqueConstraint(fields=('contest', 'order'), name='unique_contest_question_order'),
),
migrations.AddConstraint(
model_name='contestattempt',
constraint=models.UniqueConstraint(fields=('user', 'submission_key'), name='unique_user_contest_submission'),
),
migrations.AddConstraint(
model_name='contestanswer',
constraint=models.UniqueConstraint(fields=('attempt', 'contest_question'), name='unique_attempt_question_answer'),
),
]
@@ -0,0 +1,17 @@
# Generated by Django 4.2.23 on 2026-08-08 13:02
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('contest', '0001_initial'),
]
operations = [
migrations.AddIndex(
model_name='realtimematch',
index=models.Index(fields=['contest', 'status', 'player_one_rating', 'created_at'], name='matchmaking_lookup_idx'),
),
]
+194
View File
@@ -0,0 +1,194 @@
import uuid
from django.conf import settings
from django.db import models
class Question(models.Model):
class Track(models.TextChoices):
BEGINNER = "beginner", "入门"
STANDARD = "standard", "标准"
ADVANCED = "advanced", "进阶"
OPEN = "open", "Open"
slug = models.SlugField(unique=True)
track = models.CharField(max_length=16, choices=Track.choices)
tags = models.JSONField(default=list, blank=True)
is_active = models.BooleanField(default=True)
def __str__(self):
return self.slug
class QuestionVersion(models.Model):
question = models.ForeignKey(Question, on_delete=models.CASCADE, related_name="versions")
version = models.PositiveIntegerField()
prompt = models.TextField()
answer = models.CharField(max_length=200)
explanation = models.TextField(blank=True)
metadata = models.JSONField(default=dict, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
constraints = [
models.UniqueConstraint(fields=("question", "version"), name="unique_question_version")
]
def __str__(self):
return f"{self.question.slug} v{self.version}"
class Contest(models.Model):
class Kind(models.TextChoices):
REALTIME = "realtime", "实时 1v1"
DAILY = "daily", "今日挑战"
PRACTICE = "practice", "单人闯关"
WEEKLY = "weekly", "主题周赛"
class Status(models.TextChoices):
DRAFT = "draft", "草稿"
PUBLISHED = "published", "已发布"
CLOSED = "closed", "已结束"
slug = models.SlugField(unique=True)
title = models.CharField(max_length=120)
kind = models.CharField(max_length=16, choices=Kind.choices)
track = models.CharField(max_length=16, choices=Question.Track.choices)
status = models.CharField(max_length=16, choices=Status.choices, default=Status.DRAFT)
duration_seconds = models.PositiveIntegerField(default=60)
starts_at = models.DateTimeField(null=True, blank=True)
ends_at = models.DateTimeField(null=True, blank=True)
def __str__(self):
return self.title
class ContestQuestion(models.Model):
contest = models.ForeignKey(Contest, on_delete=models.CASCADE, related_name="contest_questions")
question_version = models.ForeignKey(QuestionVersion, on_delete=models.PROTECT)
order = models.PositiveIntegerField()
points = models.PositiveIntegerField(default=100)
class Meta:
constraints = [
models.UniqueConstraint(fields=("contest", "order"), name="unique_contest_question_order")
]
ordering = ["order"]
class RealtimeMatch(models.Model):
class Status(models.TextChoices):
WAITING = "waiting", "等待对手"
ACTIVE = "active", "进行中"
COMPLETED = "completed", "已完成"
CANCELLED = "cancelled", "已取消"
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
contest = models.ForeignKey(Contest, on_delete=models.PROTECT)
player_one = models.ForeignKey(
settings.AUTH_USER_MODEL, on_delete=models.PROTECT, related_name="matches_as_player_one"
)
player_two = models.ForeignKey(
settings.AUTH_USER_MODEL,
null=True,
blank=True,
on_delete=models.PROTECT,
related_name="matches_as_player_two",
)
status = models.CharField(max_length=16, choices=Status.choices, default=Status.WAITING)
player_one_rating = models.PositiveIntegerField()
player_two_rating = models.PositiveIntegerField(null=True, blank=True)
winner = models.ForeignKey(
settings.AUTH_USER_MODEL,
null=True,
blank=True,
on_delete=models.PROTECT,
related_name="won_matches",
)
created_at = models.DateTimeField(auto_now_add=True)
started_at = models.DateTimeField(null=True, blank=True)
completed_at = models.DateTimeField(null=True, blank=True)
class Meta:
indexes = [
models.Index(
fields=("contest", "status", "player_one_rating", "created_at"),
name="matchmaking_lookup_idx",
)
]
class ContestAttempt(models.Model):
class Status(models.TextChoices):
ACTIVE = "active", "进行中"
SUBMITTED = "submitted", "已提交"
EXPIRED = "expired", "已超时"
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
contest = models.ForeignKey(Contest, on_delete=models.PROTECT, related_name="attempts")
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="contest_attempts")
match = models.ForeignKey(
RealtimeMatch, null=True, blank=True, on_delete=models.PROTECT, related_name="attempts"
)
status = models.CharField(max_length=16, choices=Status.choices, default=Status.ACTIVE)
score = models.PositiveIntegerField(default=0)
correct_count = models.PositiveIntegerField(default=0)
answer_count = models.PositiveIntegerField(default=0)
duration_ms = models.PositiveIntegerField(default=0)
submission_key = models.CharField(max_length=80, null=True, blank=True)
started_at = models.DateTimeField(auto_now_add=True)
submitted_at = models.DateTimeField(null=True, blank=True)
class Meta:
constraints = [
models.UniqueConstraint(
fields=("user", "submission_key"), name="unique_user_contest_submission"
)
]
ordering = ["-started_at"]
class ContestAnswer(models.Model):
attempt = models.ForeignKey(ContestAttempt, on_delete=models.CASCADE, related_name="answers")
contest_question = models.ForeignKey(ContestQuestion, on_delete=models.PROTECT)
submitted_answer = models.CharField(max_length=200)
is_correct = models.BooleanField()
elapsed_ms = models.PositiveIntegerField()
answered_at = models.DateTimeField(auto_now_add=True)
class Meta:
constraints = [
models.UniqueConstraint(
fields=("attempt", "contest_question"), name="unique_attempt_question_answer"
)
]
class RatingHistory(models.Model):
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="rating_history")
match = models.ForeignKey(RealtimeMatch, on_delete=models.PROTECT, related_name="rating_changes")
rating_before = models.PositiveIntegerField()
rating_after = models.PositiveIntegerField()
delta = models.SmallIntegerField()
created_at = models.DateTimeField(auto_now_add=True)
class LeaderboardSnapshot(models.Model):
contest = models.ForeignKey(Contest, on_delete=models.CASCADE)
track = models.CharField(max_length=16, choices=Question.Track.choices)
period = models.CharField(max_length=40)
rankings = models.JSONField(default=list)
generated_at = models.DateTimeField(auto_now_add=True)
class CheatFlag(models.Model):
class Status(models.TextChoices):
OPEN = "open", "待处理"
DISMISSED = "dismissed", "已忽略"
CONFIRMED = "confirmed", "已确认"
attempt = models.ForeignKey(ContestAttempt, on_delete=models.CASCADE, related_name="cheat_flags")
reason = models.CharField(max_length=120)
evidence = models.JSONField(default=dict)
status = models.CharField(max_length=16, choices=Status.choices, default=Status.OPEN)
created_at = models.DateTimeField(auto_now_add=True)
+8
View File
@@ -0,0 +1,8 @@
from django.urls import path
from .consumers import MatchConsumer
websocket_urlpatterns = [
path("ws/v1/contest/matches/<uuid:match_id>/", MatchConsumer.as_asgi()),
]
+279
View File
@@ -0,0 +1,279 @@
from decimal import Decimal
from django.db import transaction
from django.db.models import Q
from django.utils import timezone
from rest_framework.exceptions import ValidationError
from accounts.models import User
from .models import (
CheatFlag,
Contest,
ContestAnswer,
ContestAttempt,
RatingHistory,
RealtimeMatch,
)
def normalize_answer(value):
text = str(value).strip().lower().replace(" ", "")
try:
return str(Decimal(text).normalize())
except Exception:
return text
def attempt_payload(attempt, include_results=False):
questions = []
answers = {answer.contest_question_id: answer for answer in attempt.answers.all()}
for item in attempt.contest.contest_questions.select_related("question_version").all():
question = {
"order": item.order,
"prompt": item.question_version.prompt,
"metadata": item.question_version.metadata,
"points": item.points,
}
if include_results and item.id in answers:
answer = answers[item.id]
question.update(
{
"submitted_answer": answer.submitted_answer,
"is_correct": answer.is_correct,
"correct_answer": item.question_version.answer,
"explanation": item.question_version.explanation,
}
)
questions.append(question)
return {
"attempt_id": attempt.id,
"contest": attempt.contest.title,
"kind": attempt.contest.kind,
"status": attempt.status,
"duration_seconds": attempt.contest.duration_seconds,
"server_started_at": attempt.started_at,
"score": attempt.score,
"correct_count": attempt.correct_count,
"answer_count": attempt.answer_count,
"duration_ms": attempt.duration_ms,
"questions": questions,
}
@transaction.atomic
def start_attempt(user, contest):
now = timezone.now()
if contest.status != Contest.Status.PUBLISHED:
raise ValidationError("比赛尚未发布")
if contest.starts_at and contest.starts_at > now:
raise ValidationError("比赛尚未开始")
if contest.ends_at and contest.ends_at <= now:
raise ValidationError("比赛已经结束")
if contest.kind == Contest.Kind.DAILY:
existing = ContestAttempt.objects.filter(user=user, contest=contest).first()
if existing:
return attempt_payload(
existing,
include_results=existing.status != ContestAttempt.Status.ACTIVE,
)
attempt = ContestAttempt.objects.create(contest=contest, user=user)
return attempt_payload(attempt)
@transaction.atomic
def submit_attempt(user, attempt_id, raw_answers, submission_key):
attempt = (
ContestAttempt.objects.select_for_update()
.select_related("contest")
.get(id=attempt_id, user=user)
)
if attempt.status != ContestAttempt.Status.ACTIVE:
if submission_key and attempt.submission_key == submission_key:
return attempt_payload(attempt, include_results=True)
raise ValidationError("该答题记录已经结算")
if not submission_key:
raise ValidationError({"Idempotency-Key": "正式提交必须提供幂等键"})
now = timezone.now()
duration_ms = max(0, int((now - attempt.started_at).total_seconds() * 1000))
limit_ms = attempt.contest.duration_seconds * 1000
items = list(
attempt.contest.contest_questions.select_related("question_version").all()
)
if not isinstance(raw_answers, list):
raise ValidationError({"answers": "答案必须是数组"})
by_order = {}
try:
for item in raw_answers:
order = int(item["order"])
if order <= 0 or order in by_order:
raise ValueError
by_order[order] = item.get("answer", "")
except (KeyError, TypeError, ValueError) as exc:
raise ValidationError({"answers": "答案题号无效或重复"}) from exc
score = 0
correct_count = 0
for contest_question in items:
submitted = str(by_order.get(contest_question.order, ""))[:200]
correct = normalize_answer(submitted) == normalize_answer(
contest_question.question_version.answer
)
if correct and duration_ms <= limit_ms:
score += contest_question.points
correct_count += 1
ContestAnswer.objects.create(
attempt=attempt,
contest_question=contest_question,
submitted_answer=submitted,
is_correct=correct and duration_ms <= limit_ms,
elapsed_ms=min(duration_ms, limit_ms + 60_000),
)
attempt.status = (
ContestAttempt.Status.SUBMITTED
if duration_ms <= limit_ms
else ContestAttempt.Status.EXPIRED
)
attempt.score = score
attempt.correct_count = correct_count
attempt.answer_count = len(raw_answers)
attempt.duration_ms = duration_ms
attempt.submission_key = submission_key
attempt.submitted_at = now
attempt.save()
if raw_answers and duration_ms / len(raw_answers) < 150:
CheatFlag.objects.create(
attempt=attempt,
reason="extreme_answer_speed",
evidence={
"duration_ms": duration_ms,
"answer_count": len(raw_answers),
},
)
if attempt.match_id:
finalize_match(attempt.match_id)
return attempt_payload(attempt, include_results=True)
@transaction.atomic
def find_match(user, contest):
if contest.kind != Contest.Kind.REALTIME or contest.status != Contest.Status.PUBLISHED:
raise ValidationError("实时比赛不可用")
existing = (
RealtimeMatch.objects.filter(
Q(player_one=user) | Q(player_two=user),
contest=contest,
status__in=[RealtimeMatch.Status.WAITING, RealtimeMatch.Status.ACTIVE],
)
.order_by("-created_at")
.first()
)
if existing:
return existing
waiting = (
RealtimeMatch.objects.select_for_update(skip_locked=True)
.filter(
contest=contest,
status=RealtimeMatch.Status.WAITING,
player_one_rating__gte=max(0, user.rating - 300),
player_one_rating__lte=user.rating + 300,
)
.exclude(player_one=user)
.order_by("created_at")
.first()
)
if waiting is None:
return RealtimeMatch.objects.create(
contest=contest,
player_one=user,
player_one_rating=user.rating,
)
waiting.player_two = user
waiting.player_two_rating = user.rating
waiting.status = RealtimeMatch.Status.ACTIVE
waiting.started_at = timezone.now()
waiting.save(
update_fields=["player_two", "player_two_rating", "status", "started_at"]
)
ContestAttempt.objects.bulk_create(
[
ContestAttempt(contest=contest, user=waiting.player_one, match=waiting),
ContestAttempt(contest=contest, user=user, match=waiting),
]
)
return waiting
def match_payload(match, user):
attempt = match.attempts.filter(user=user).first()
opponent = match.player_two if match.player_one_id == user.id else match.player_one
return {
"match_id": match.id,
"status": match.status,
"opponent": (
{"nickname": opponent.nickname, "rating": opponent.rating}
if opponent
else None
),
"attempt": attempt_payload(attempt) if attempt else None,
"websocket_path": f"/ws/v1/contest/matches/{match.id}/",
}
def _elo_delta(rating, opponent_rating, score, k=32):
expected = 1 / (1 + 10 ** ((opponent_rating - rating) / 400))
return round(k * (score - expected))
@transaction.atomic
def finalize_match(match_id):
match = RealtimeMatch.objects.select_for_update().get(id=match_id)
if match.status != RealtimeMatch.Status.ACTIVE:
return match
attempts = list(match.attempts.select_related("user").order_by("user_id"))
if len(attempts) != 2 or any(
attempt.status == ContestAttempt.Status.ACTIVE for attempt in attempts
):
return match
first = next(item for item in attempts if item.user_id == match.player_one_id)
second = next(item for item in attempts if item.user_id == match.player_two_id)
if first.score > second.score:
first_result, second_result = 1.0, 0.0
match.winner_id = first.user_id
elif second.score > first.score:
first_result, second_result = 0.0, 1.0
match.winner_id = second.user_id
else:
first_result = second_result = 0.5
users = {
user.id: user
for user in User.objects.select_for_update().filter(
id__in=[match.player_one_id, match.player_two_id]
)
}
player_one = users[match.player_one_id]
player_two = users[match.player_two_id]
deltas = (
_elo_delta(player_one.rating, player_two.rating, first_result),
_elo_delta(player_two.rating, player_one.rating, second_result),
)
for user, delta in zip((player_one, player_two), deltas):
before = user.rating
user.rating = max(0, before + delta)
user.save(update_fields=["rating"])
RatingHistory.objects.create(
user=user,
match=match,
rating_before=before,
rating_after=user.rating,
delta=delta,
)
match.status = RealtimeMatch.Status.COMPLETED
match.completed_at = timezone.now()
match.save(update_fields=["winner", "status", "completed_at"])
return match
+203
View File
@@ -0,0 +1,203 @@
from datetime import timedelta
import pytest
from django.utils import timezone
from rest_framework.exceptions import ValidationError
from accounts.models import User
from contest.models import (
Contest,
ContestAttempt,
ContestQuestion,
Question,
QuestionVersion,
RatingHistory,
RealtimeMatch,
)
from contest.services import (
finalize_match,
find_match,
normalize_answer,
start_attempt,
submit_attempt,
)
@pytest.fixture
def user(db):
return User.objects.create_user(
username="contest_user",
password="StrongPass_2026",
nickname="比赛用户",
)
@pytest.fixture
def daily_contest(db):
question = Question.objects.create(
slug="sum-question",
track=Question.Track.STANDARD,
)
version = QuestionVersion.objects.create(
question=question,
version=1,
prompt="17 + 25",
answer="42",
explanation="相加得 42",
)
contest = Contest.objects.create(
slug="daily-test",
title="测试今日赛",
kind=Contest.Kind.DAILY,
track=Question.Track.STANDARD,
status=Contest.Status.PUBLISHED,
duration_seconds=60,
)
ContestQuestion.objects.create(
contest=contest,
question_version=version,
order=1,
points=100,
)
return contest
@pytest.mark.parametrize(
"raw, expected",
[
pytest.param(" 1.0 ", "1", id="小数标准化"),
pytest.param("ABC ", "abc", id="文本去空格并转小写"),
pytest.param("-0", "-0", id="保留十进制负零表示"),
],
)
def test_normalize_answer_标准化输入(raw, expected):
assert normalize_answer(raw) == expected
@pytest.mark.django_db
def test_start_attempt_每日赛重复进入复用同一记录(user, daily_contest):
first = start_attempt(user, daily_contest)
second = start_attempt(user, daily_contest)
assert first["attempt_id"] == second["attempt_id"]
assert ContestAttempt.objects.count() == 1
@pytest.mark.django_db
def test_submit_attempt_服务端判分且幂等重放(user, daily_contest):
started = start_attempt(user, daily_contest)
result = submit_attempt(
user,
started["attempt_id"],
[{"order": 1, "answer": "42.0"}],
"submission-1",
)
replay = submit_attempt(
user,
started["attempt_id"],
[{"order": 1, "answer": "0"}],
"submission-1",
)
assert result["status"] == ContestAttempt.Status.SUBMITTED
assert result["score"] == 100
assert result["correct_count"] == 1
assert result["questions"][0]["correct_answer"] == "42"
assert replay["score"] == 100
@pytest.mark.django_db
def test_submit_attempt_缺少幂等键时拒绝(user, daily_contest):
started = start_attempt(user, daily_contest)
with pytest.raises(ValidationError, match="幂等键"):
submit_attempt(
user,
started["attempt_id"],
[{"order": 1, "answer": "42"}],
None,
)
@pytest.mark.django_db
def test_submit_attempt_畸形题号返回校验错误且记录保持进行中(user, daily_contest):
started = start_attempt(user, daily_contest)
with pytest.raises(ValidationError):
submit_attempt(
user,
started["attempt_id"],
[{"order": "first", "answer": "42"}],
"malformed-order",
)
attempt = ContestAttempt.objects.get(id=started["attempt_id"])
assert attempt.status == ContestAttempt.Status.ACTIVE
assert attempt.answers.count() == 0
@pytest.mark.django_db
def test_submit_attempt_超过服务端时限不计分(user, daily_contest):
started = start_attempt(user, daily_contest)
ContestAttempt.objects.filter(id=started["attempt_id"]).update(
started_at=timezone.now() - timedelta(seconds=61)
)
result = submit_attempt(
user,
started["attempt_id"],
[{"order": 1, "answer": "42"}],
"late-submit",
)
assert result["status"] == ContestAttempt.Status.EXPIRED
assert result["score"] == 0
assert result["correct_count"] == 0
@pytest.mark.django_db
def test_find_match_同分玩家配对并由_finalize_match_唯一结算_rating():
first = User.objects.create_user(
username="player_one",
password="StrongPass_2026",
nickname="玩家一",
)
second = User.objects.create_user(
username="player_two",
password="StrongPass_2026",
nickname="玩家二",
)
contest = Contest.objects.create(
slug="realtime-test",
title="测试实时赛",
kind=Contest.Kind.REALTIME,
track=Question.Track.STANDARD,
status=Contest.Status.PUBLISHED,
)
waiting = find_match(first, contest)
active = find_match(second, contest)
active.refresh_from_db()
assert waiting.id == active.id
assert active.status == RealtimeMatch.Status.ACTIVE
assert active.attempts.count() == 2
active.attempts.filter(user=first).update(
status=ContestAttempt.Status.SUBMITTED,
score=200,
)
active.attempts.filter(user=second).update(
status=ContestAttempt.Status.SUBMITTED,
score=100,
)
finalized = finalize_match(active.id)
first.refresh_from_db()
second.refresh_from_db()
assert finalized.status == RealtimeMatch.Status.COMPLETED
assert finalized.winner == first
assert first.rating == 1016
assert second.rating == 984
assert RatingHistory.objects.filter(match=active).count() == 2
+3
View File
@@ -0,0 +1,3 @@
from django.test import TestCase
# Create your tests here.
+20
View File
@@ -0,0 +1,20 @@
from django.urls import path
from .views import (
AttemptStartView,
AttemptSubmitView,
ContestListView,
LeaderboardView,
MatchmakingView,
MatchStateView,
)
urlpatterns = [
path("", ContestListView.as_view(), name="contest-list"),
path("<slug:slug>/start/", AttemptStartView.as_view(), name="attempt-start"),
path("<slug:slug>/matchmaking/", MatchmakingView.as_view(), name="matchmaking"),
path("<slug:slug>/leaderboard/", LeaderboardView.as_view(), name="leaderboard"),
path("attempts/<uuid:attempt_id>/submit/", AttemptSubmitView.as_view(), name="attempt-submit"),
path("matches/<uuid:match_id>/", MatchStateView.as_view(), name="match-state"),
]
+100
View File
@@ -0,0 +1,100 @@
from django.shortcuts import get_object_or_404
from rest_framework import permissions, status
from rest_framework.response import Response
from rest_framework.views import APIView
from .models import Contest, ContestAttempt, RealtimeMatch
from .services import (
find_match,
match_payload,
start_attempt,
submit_attempt,
)
class ContestListView(APIView):
permission_classes = [permissions.AllowAny]
def get(self, request):
contests = Contest.objects.filter(status=Contest.Status.PUBLISHED).order_by(
"kind", "title"
)
return Response(
[
{
"slug": contest.slug,
"title": contest.title,
"kind": contest.kind,
"track": contest.track,
"duration_seconds": contest.duration_seconds,
"starts_at": contest.starts_at,
"ends_at": contest.ends_at,
}
for contest in contests
]
)
class AttemptStartView(APIView):
def post(self, request, slug):
contest = get_object_or_404(Contest, slug=slug)
return Response(start_attempt(request.user, contest), status=status.HTTP_201_CREATED)
class AttemptSubmitView(APIView):
def post(self, request, attempt_id):
get_object_or_404(ContestAttempt, id=attempt_id, user=request.user)
payload = submit_attempt(
user=request.user,
attempt_id=attempt_id,
raw_answers=request.data.get("answers", []),
submission_key=request.headers.get("Idempotency-Key"),
)
return Response(payload)
class MatchmakingView(APIView):
def post(self, request, slug):
contest = get_object_or_404(Contest, slug=slug)
match = find_match(request.user, contest)
return Response(match_payload(match, request.user), status=status.HTTP_202_ACCEPTED)
class MatchStateView(APIView):
def get(self, request, match_id):
match = get_object_or_404(
RealtimeMatch.objects.select_related("player_one", "player_two"),
id=match_id,
)
if request.user.id not in (match.player_one_id, match.player_two_id):
return Response(status=status.HTTP_403_FORBIDDEN)
return Response(match_payload(match, request.user))
class LeaderboardView(APIView):
permission_classes = [permissions.AllowAny]
def get(self, request, slug):
contest = get_object_or_404(Contest, slug=slug)
attempts = (
ContestAttempt.objects.filter(
contest=contest,
status=ContestAttempt.Status.SUBMITTED,
cheat_flags__isnull=True,
)
.select_related("user")
.order_by("-score", "duration_ms", "submitted_at")[:100]
)
return Response(
[
{
"rank": index,
"nickname": attempt.user.nickname,
"track": attempt.user.track,
"score": attempt.score,
"correct_count": attempt.correct_count,
"duration_ms": attempt.duration_ms,
}
for index, attempt in enumerate(attempts, start=1)
]
)
View File
+3
View File
@@ -0,0 +1,3 @@
from django.contrib import admin
# Register your models here.
+6
View File
@@ -0,0 +1,6 @@
from django.apps import AppConfig
class EngagementConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'engagement'
@@ -0,0 +1,44 @@
# Generated by Django 4.2.23 on 2026-08-08 11:15
from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
initial = True
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.CreateModel(
name='Notification',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('category', models.CharField(max_length=40)),
('title', models.CharField(max_length=120)),
('body', models.TextField()),
('is_read', models.BooleanField(default=False)),
('created_at', models.DateTimeField(auto_now_add=True)),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='notifications', to=settings.AUTH_USER_MODEL)),
],
),
migrations.CreateModel(
name='DailyCheckIn',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('date', models.DateField()),
('streak', models.PositiveIntegerField(default=1)),
('reward', models.JSONField(default=dict)),
('created_at', models.DateTimeField(auto_now_add=True)),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='checkins', to=settings.AUTH_USER_MODEL)),
],
),
migrations.AddConstraint(
model_name='dailycheckin',
constraint=models.UniqueConstraint(fields=('user', 'date'), name='unique_user_daily_checkin'),
),
]
+24
View File
@@ -0,0 +1,24 @@
from django.conf import settings
from django.db import models
class DailyCheckIn(models.Model):
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="checkins")
date = models.DateField()
streak = models.PositiveIntegerField(default=1)
reward = models.JSONField(default=dict)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
constraints = [
models.UniqueConstraint(fields=("user", "date"), name="unique_user_daily_checkin")
]
class Notification(models.Model):
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="notifications")
category = models.CharField(max_length=40)
title = models.CharField(max_length=120)
body = models.TextField()
is_read = models.BooleanField(default=False)
created_at = models.DateTimeField(auto_now_add=True)
+3
View File
@@ -0,0 +1,3 @@
from django.test import TestCase
# Create your tests here.
+3
View File
@@ -0,0 +1,3 @@
from django.shortcuts import render
# Create your views here.
View File
+29
View File
@@ -0,0 +1,29 @@
from django.contrib import admin
from .models import (
FormulaDocument,
FormulaRevision,
HandwritingRecognitionJob,
LatexAttempt,
LatexCourse,
LatexExercise,
LatexLesson,
)
class LatexLessonInline(admin.StackedInline):
model = LatexLesson
extra = 1
@admin.register(LatexCourse)
class LatexCourseAdmin(admin.ModelAdmin):
list_display = ("title", "order", "is_published")
inlines = [LatexLessonInline]
admin.site.register(FormulaDocument)
admin.site.register(FormulaRevision)
admin.site.register(LatexExercise)
admin.site.register(LatexAttempt)
admin.site.register(HandwritingRecognitionJob)
+6
View File
@@ -0,0 +1,6 @@
from django.apps import AppConfig
class LatexLabConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'latex_lab'
@@ -0,0 +1,103 @@
# Generated by Django 4.2.23 on 2026-08-08 11:15
from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion
import uuid
class Migration(migrations.Migration):
initial = True
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.CreateModel(
name='FormulaDocument',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('title', models.CharField(max_length=120)),
('source', models.TextField(blank=True)),
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='formulas', to=settings.AUTH_USER_MODEL)),
],
options={
'ordering': ['-updated_at'],
},
),
migrations.CreateModel(
name='LatexCourse',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('slug', models.SlugField(unique=True)),
('title', models.CharField(max_length=120)),
('description', models.TextField(blank=True)),
('order', models.PositiveIntegerField(default=0)),
('is_published', models.BooleanField(default=False)),
],
),
migrations.CreateModel(
name='LatexLesson',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('slug', models.SlugField()),
('title', models.CharField(max_length=120)),
('content', models.TextField()),
('example_source', models.TextField(blank=True)),
('order', models.PositiveIntegerField(default=0)),
('course', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='lessons', to='latex_lab.latexcourse')),
],
options={
'ordering': ['order'],
},
),
migrations.CreateModel(
name='LatexExercise',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('prompt', models.TextField()),
('expected_source', models.TextField()),
('explanation', models.TextField(blank=True)),
('order', models.PositiveIntegerField(default=0)),
('lesson', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='exercises', to='latex_lab.latexlesson')),
],
),
migrations.CreateModel(
name='LatexAttempt',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('submitted_source', models.TextField()),
('is_correct', models.BooleanField()),
('created_at', models.DateTimeField(auto_now_add=True)),
('exercise', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, to='latex_lab.latexexercise')),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='latex_attempts', to=settings.AUTH_USER_MODEL)),
],
),
migrations.CreateModel(
name='HandwritingRecognitionJob',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('status', models.CharField(choices=[('pending', '等待'), ('complete', '完成'), ('failed', '失败')], default='pending', max_length=16)),
('result', models.JSONField(blank=True, default=dict)),
('created_at', models.DateTimeField(auto_now_add=True)),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
],
),
migrations.CreateModel(
name='FormulaRevision',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('source', models.TextField()),
('created_at', models.DateTimeField(auto_now_add=True)),
('document', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='revisions', to='latex_lab.formuladocument')),
],
),
migrations.AddConstraint(
model_name='latexlesson',
constraint=models.UniqueConstraint(fields=('course', 'slug'), name='unique_course_lesson'),
),
]
+77
View File
@@ -0,0 +1,77 @@
import uuid
from django.conf import settings
from django.db import models
class FormulaDocument(models.Model):
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="formulas")
title = models.CharField(max_length=120)
source = models.TextField(blank=True)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
ordering = ["-updated_at"]
class FormulaRevision(models.Model):
document = models.ForeignKey(FormulaDocument, on_delete=models.CASCADE, related_name="revisions")
source = models.TextField()
created_at = models.DateTimeField(auto_now_add=True)
class LatexCourse(models.Model):
slug = models.SlugField(unique=True)
title = models.CharField(max_length=120)
description = models.TextField(blank=True)
order = models.PositiveIntegerField(default=0)
is_published = models.BooleanField(default=False)
def __str__(self):
return self.title
class LatexLesson(models.Model):
course = models.ForeignKey(LatexCourse, on_delete=models.CASCADE, related_name="lessons")
slug = models.SlugField()
title = models.CharField(max_length=120)
content = models.TextField()
example_source = models.TextField(blank=True)
order = models.PositiveIntegerField(default=0)
class Meta:
constraints = [
models.UniqueConstraint(fields=("course", "slug"), name="unique_course_lesson")
]
ordering = ["order"]
class LatexExercise(models.Model):
lesson = models.ForeignKey(LatexLesson, on_delete=models.CASCADE, related_name="exercises")
prompt = models.TextField()
expected_source = models.TextField()
explanation = models.TextField(blank=True)
order = models.PositiveIntegerField(default=0)
class LatexAttempt(models.Model):
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="latex_attempts")
exercise = models.ForeignKey(LatexExercise, on_delete=models.PROTECT)
submitted_source = models.TextField()
is_correct = models.BooleanField()
created_at = models.DateTimeField(auto_now_add=True)
class HandwritingRecognitionJob(models.Model):
class Status(models.TextChoices):
PENDING = "pending", "等待"
COMPLETE = "complete", "完成"
FAILED = "failed", "失败"
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
status = models.CharField(max_length=16, choices=Status.choices, default=Status.PENDING)
result = models.JSONField(default=dict, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
+28
View File
@@ -0,0 +1,28 @@
from rest_framework import serializers
from .models import FormulaDocument, FormulaRevision
class FormulaDocumentSerializer(serializers.ModelSerializer):
class Meta:
model = FormulaDocument
fields = ("id", "title", "source", "created_at", "updated_at")
read_only_fields = ("id", "created_at", "updated_at")
def validate_source(self, value):
if len(value) > 50_000:
raise serializers.ValidationError("公式源码不能超过 50000 个字符")
return value
def create(self, validated_data):
document = FormulaDocument.objects.create(
user=self.context["request"].user,
**validated_data,
)
FormulaRevision.objects.create(document=document, source=document.source)
return document
def update(self, instance, validated_data):
instance = super().update(instance, validated_data)
FormulaRevision.objects.create(document=instance, source=instance.source)
return instance
+15
View File
@@ -0,0 +1,15 @@
from latex_lab.views import ExerciseSubmitView
def test_normalize_忽略公式结构外的空白():
left = ExerciseSubmitView.normalize(r"\frac { a } { b }")
right = ExerciseSubmitView.normalize(r"\frac{a}{b}")
assert left == right
def test_normalize_保留_text_命令内部的语义空格():
with_space = ExerciseSubmitView.normalize(r"\text{a b}")
without_space = ExerciseSubmitView.normalize(r"\text{ab}")
assert with_space != without_space
+3
View File
@@ -0,0 +1,3 @@
from django.test import TestCase
# Create your tests here.
+18
View File
@@ -0,0 +1,18 @@
from django.urls import include, path
from rest_framework.routers import DefaultRouter
from .views import CourseListView, ExerciseSubmitView, FormulaDocumentViewSet
router = DefaultRouter()
router.register("documents", FormulaDocumentViewSet, basename="formula-document")
urlpatterns = [
path("", include(router.urls)),
path("courses/", CourseListView.as_view(), name="latex-course-list"),
path(
"exercises/<int:exercise_id>/submit/",
ExerciseSubmitView.as_view(),
name="latex-exercise-submit",
),
]
+106
View File
@@ -0,0 +1,106 @@
from django.shortcuts import get_object_or_404
from rest_framework import permissions, status, viewsets
from rest_framework.response import Response
from rest_framework.views import APIView
from .models import LatexAttempt, LatexCourse, LatexExercise
from .serializers import FormulaDocumentSerializer
class FormulaDocumentViewSet(viewsets.ModelViewSet):
serializer_class = FormulaDocumentSerializer
def get_queryset(self):
return self.request.user.formulas.all()
class CourseListView(APIView):
permission_classes = [permissions.AllowAny]
def get(self, request):
courses = LatexCourse.objects.filter(is_published=True).prefetch_related(
"lessons__exercises"
)
return Response(
[
{
"slug": course.slug,
"title": course.title,
"description": course.description,
"lessons": [
{
"slug": lesson.slug,
"title": lesson.title,
"content": lesson.content,
"example_source": lesson.example_source,
"exercise_count": lesson.exercises.count(),
}
for lesson in course.lessons.all()
],
}
for course in courses
]
)
class ExerciseSubmitView(APIView):
@staticmethod
def normalize(source):
source = source or ""
text_commands = (r"\text", r"\mbox", r"\textrm", r"\textsf", r"\texttt")
normalized = []
index = 0
while index < len(source):
command = next(
(item for item in text_commands if source.startswith(item, index)),
None,
)
if command is None:
if not source[index].isspace():
normalized.append(source[index])
index += 1
continue
command_end = index + len(command)
group_start = command_end
while group_start < len(source) and source[group_start].isspace():
group_start += 1
if group_start >= len(source) or source[group_start] != "{":
normalized.append(command)
index = command_end
continue
depth = 0
group_end = group_start
while group_end < len(source):
if source[group_end] == "{":
depth += 1
elif source[group_end] == "}":
depth -= 1
if depth == 0:
group_end += 1
break
group_end += 1
normalized.append(command)
normalized.append(source[group_start:group_end])
index = group_end
return "".join(normalized)
def post(self, request, exercise_id):
exercise = get_object_or_404(LatexExercise, id=exercise_id)
submitted = str(request.data.get("source", ""))[:50_000]
is_correct = self.normalize(submitted) == self.normalize(exercise.expected_source)
LatexAttempt.objects.create(
user=request.user,
exercise=exercise,
submitted_source=submitted,
is_correct=is_correct,
)
return Response(
{
"is_correct": is_correct,
"expected_source": exercise.expected_source if not is_correct else None,
"explanation": exercise.explanation,
},
status=status.HTTP_201_CREATED,
)
+22
View File
@@ -0,0 +1,22 @@
#!/usr/bin/env python
"""Django's command-line utility for administrative tasks."""
import os
import sys
def main():
"""Run administrative tasks."""
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')
try:
from django.core.management import execute_from_command_line
except ImportError as exc:
raise ImportError(
"Couldn't import Django. Are you sure it's installed and "
"available on your PYTHONPATH environment variable? Did you "
"forget to activate a virtual environment?"
) from exc
execute_from_command_line(sys.argv)
if __name__ == '__main__':
main()
View File
+45
View File
@@ -0,0 +1,45 @@
from django.contrib import admin
from django.core.exceptions import ValidationError
from .models import (
Character,
MathBTIAssessment,
MathBTIResult,
MathIdentity,
SkillPackage,
Story,
StoryChoice,
StoryRun,
StoryVersion,
UserRelationship,
)
from .services import validate_story_content
@admin.register(StoryVersion)
class StoryVersionAdmin(admin.ModelAdmin):
list_display = ("story", "version", "is_published", "published_at", "created_at")
list_filter = ("is_published", "story__kind")
def save_model(self, request, obj, form, change):
errors = validate_story_content(obj.content)
if errors:
raise ValidationError(errors)
super().save_model(request, obj, form, change)
@admin.register(StoryRun)
class StoryRunAdmin(admin.ModelAdmin):
list_display = ("id", "user", "story_version", "status", "current_node", "updated_at")
list_filter = ("status", "story_version__story")
readonly_fields = ("started_at", "updated_at", "completed_at")
admin.site.register(MathIdentity)
admin.site.register(MathBTIAssessment)
admin.site.register(MathBTIResult)
admin.site.register(Character)
admin.site.register(Story)
admin.site.register(StoryChoice)
admin.site.register(UserRelationship)
admin.site.register(SkillPackage)
+6
View File
@@ -0,0 +1,6 @@
from django.apps import AppConfig
class MathLifeConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'math_life'
+1
View File
@@ -0,0 +1 @@
@@ -0,0 +1 @@
@@ -0,0 +1,357 @@
import json
from pathlib import Path
from django.conf import settings
from django.core.management.base import BaseCommand, CommandError
from django.utils import timezone
from accounts.models import InviteCode
from content.models import ContentItem
from latex_lab.models import LatexCourse, LatexExercise, LatexLesson
from math_life.models import (
MathBTIAssessment,
MathIdentity,
SkillPackage,
Story,
StoryVersion,
)
from math_life.services import validate_story_content
DISCIPLINE_ICONS = {
"人工智能": "🤖",
"计算机": "💻",
"电子信息": "📡",
"微电子": "🔌",
"机械工程": "",
"自动化": "",
"能源动力": "🔥",
"化工材料": "",
"生物医学": "",
"经济管理": "",
"环境工程": "",
"数学建模": "📐",
}
def load_json(path):
with path.open(encoding="utf-8") as source:
return json.load(source)
def video_ability(video):
title = video["title"]
module = video["module"]
discipline = video["discipline"]
if module == "数思" or title == "你被平均数骗过吗?":
return ContentItem.Ability.VISION
if module == "数说" or title == "消费者行为预测的回归分析":
return ContentItem.Ability.HUMANITIES
if discipline in {"生物医学", "自动化"}:
return ContentItem.Ability.DETECTION
if discipline in {"机械工程", "化工材料", "环境工程", "能源动力", "数学建模"}:
return ContentItem.Ability.MODELING
if title in {"金融里的数学:为什么风险可以算出来", "博弈论与市场竞争"}:
return ContentItem.Ability.MODELING
return ContentItem.Ability.CONNECTION
def skill_content(title, person, dilemma):
return {
"title": title,
"start_node": "opening",
"nodes": {
"opening": {
"character": "旁白",
"scene": f"你成为青年时期的{person}{dilemma}",
"choices": [
{
"text": "接受风险,争取更大的可能",
"next": "risk",
"effects": {"time": -2, "reputation": 1},
},
{
"text": "先保住当下,再等待机会",
"next": "steady",
"effects": {"money": 2, "reputation": -1},
},
],
},
"risk": {
"character": person,
"scene": "选择带来了压力,也让你接触到原本看不见的问题。",
"choices": [
{
"text": "把有限时间投入研究",
"next": "crossroads",
"effects": {"energy": -2, "research": 3},
}
],
},
"steady": {
"character": person,
"scene": "稳定让你积累了资源,但窗口正在逐渐关闭。",
"choices": [
{
"text": "用积累换一次尝试",
"next": "crossroads",
"effects": {"money": -1, "research": 2},
}
],
},
"crossroads": {
"character": "同伴",
"scene": "同伴提供了不完整的消息。你必须决定是独自推进,还是公开方法。",
"choices": [
{
"text": "先完成证明,再公开",
"next": "ending_scholar",
"effects": {"research": 2, "cooperation": -1},
},
{
"text": "邀请同伴共同验证",
"next": "ending_bridge",
"effects": {"cooperation": 3, "reputation": 1},
},
],
},
"ending_scholar": {
"character": "旁白",
"scene": "你守住了方法的完整性,也承担了独行的代价。",
"choices": [],
},
"ending_bridge": {
"character": "旁白",
"scene": "成果不再只属于一个人,你让更多人能够继续向前。",
"choices": [],
},
},
}
class Command(BaseCommand):
help = "导入 MathBTI、信仰者主线和两个人物 Skill 样板"
def handle(self, *args, **options):
docs = Path(settings.PROJECT_ROOT) / "docs"
mathbti_path = docs / "old_scripts" / "seed_mathbti.json"
story_path = docs / "数学少年线_story.json"
videos_path = docs / "old_scripts" / "seed_videos.json"
if not mathbti_path.exists() or not story_path.exists() or not videos_path.exists():
raise CommandError("缺少 docs 中的初始内容文件")
definition = load_json(mathbti_path)
assessment, _ = MathBTIAssessment.objects.update_or_create(
version=definition["version"],
defaults={
"title": definition["title"],
"definition": definition,
"is_published": True,
"published_at": timezone.now(),
},
)
for code, result in definition["results"].items():
MathIdentity.objects.update_or_create(
code=code,
defaults={
"name": result["name"],
"clan": result["clan_name"],
"mathematician": result["mathematician"],
"description": result["description"],
"initial_abilities": result.get("stats5", {}),
"portrait": result.get("portrait", ""),
},
)
story_document = load_json(story_path)
errors = validate_story_content(story_document)
if errors:
raise CommandError("; ".join(errors))
flagship, _ = Story.objects.update_or_create(
slug="believer-math-teen",
defaults={
"title": story_document["title"],
"summary": story_document.get("description", ""),
"kind": Story.Kind.FLAGSHIP,
"estimated_minutes": 120,
"is_visible": True,
},
)
StoryVersion.objects.update_or_create(
story=flagship,
version=1,
defaults={
"content": story_document,
"is_published": True,
"published_at": timezone.now(),
},
)
samples = [
(
"hua-luogeng-skill",
"华罗庚:从自学到远行",
"华罗庚",
"你没有完整的学院路径,却收到一次改变研究方向的机会。",
["《华罗庚传》", "中国科学院公开人物资料"],
),
(
"su-buqing-skill",
"苏步青:选择回国",
"苏步青",
"海外研究条件优越,故乡却需要从零建设数学教育。",
["复旦大学校史资料", "中国科学院公开人物资料"],
),
]
for slug, title, person, dilemma, sources in samples:
story, _ = Story.objects.update_or_create(
slug=slug,
defaults={
"title": title,
"summary": dilemma,
"kind": Story.Kind.SKILL,
"estimated_minutes": 20,
"is_visible": True,
},
)
document = skill_content(title, person, dilemma)
StoryVersion.objects.update_or_create(
story=story,
version=1,
defaults={
"content": document,
"is_published": True,
"published_at": timezone.now(),
},
)
SkillPackage.objects.update_or_create(
story=story,
defaults={
"real_person": person,
"historical_context": dilemma,
"fact_sources": sources,
"fictional_scope": "人生节点基于公开资料,具体对话与选择为艺术加工。",
"creator": "葫芦数学内容组",
},
)
InviteCode.objects.get_or_create(
code="HULU2026",
defaults={"group": "本地首发体验", "max_uses": 100, "is_active": True},
)
course, _ = LatexCourse.objects.update_or_create(
slug="latex-from-zero",
defaults={
"title": "LaTeX 零基础表达",
"description": "从上下标到完整数学证明的六步课程。",
"order": 1,
"is_published": True,
},
)
lesson_specs = [
("basics", "上标、下标和基础运算", r"x^2 + y_1", r"x^2+y_1"),
("fractions", "分式、根式和括号", r"\frac{a}{b}+\sqrt{x}", r"\frac{a}{b}+\sqrt{x}"),
("calculus", "求和、积分和极限", r"\sum_{i=1}^{n}i", r"\sum_{i=1}^{n}i"),
("matrix", "矩阵和方程组", r"\begin{matrix}a&b\\c&d\end{matrix}", r"\begin{matrix}a&b\\c&d\end{matrix}"),
("alignment", "多行公式与对齐", r"\begin{aligned}a&=b\\&=c\end{aligned}", r"\begin{aligned}a&=b\\&=c\end{aligned}"),
("proof", "完整解答与证明排版", r"\because a=b,\ \therefore a+c=b+c", r"\because a=b,\therefore a+c=b+c"),
]
for order, (slug, title, example, expected) in enumerate(lesson_specs, start=1):
lesson, _ = LatexLesson.objects.update_or_create(
course=course,
slug=slug,
defaults={
"title": title,
"content": f"本节通过可运行示例学习{title}",
"example_source": example,
"order": order,
},
)
LatexExercise.objects.update_or_create(
lesson=lesson,
order=1,
defaults={
"prompt": f"输入与示例等价的 {title} 公式。",
"expected_source": expected,
"explanation": "注意命令、花括号和环境闭合。",
},
)
content_specs = [
(
"why-proof-matters",
"为什么数学家坚持证明",
ContentItem.Kind.KNOWLEDGE,
"答案正确并不等于我们知道它为什么正确。",
["证明", "数学精神"],
),
(
"gauss-17-gon",
"高斯与正十七边形",
ContentItem.Kind.PERSON,
"一个十九岁少年的发现,如何连接古典几何与代数。",
["高斯", "几何"],
),
(
"model-is-not-world",
"模型不是现实本身",
ContentItem.Kind.KNOWLEDGE,
"建模从选择变量开始,也从那一刻开始承担遗漏的代价。",
["建模", "应用"],
),
]
for slug, title, kind, summary, topics in content_specs:
ContentItem.objects.update_or_create(
slug=slug,
defaults={
"title": title,
"kind": kind,
"summary": summary,
"body": summary,
"topics": topics,
"source": "葫芦数学首发内容",
"is_published": True,
"published_at": timezone.now(),
},
)
videos = load_json(videos_path)
for index, video in enumerate(videos, start=1):
ContentItem.objects.update_or_create(
slug=f"legacy-video-{index:03d}",
defaults={
"title": video["title"],
"kind": ContentItem.Kind.VIDEO,
"summary": video.get("description", ""),
"body": video.get("description", ""),
"cover_url": video.get("cover_url", ""),
"media_url": video.get("video_url", ""),
"topics": [
video.get("module", ""),
video.get("sub_category", ""),
video.get("discipline", ""),
],
"source": "老版志愿者视频流",
"ability_dimension": video_ability(video),
"module": video.get("module", ""),
"sub_category": video.get("sub_category", ""),
"discipline": video.get("discipline", ""),
"discipline_icon": DISCIPLINE_ICONS.get(
video.get("discipline"),
video.get("discipline_icon") or "",
),
"author": video.get("author", ""),
"duration_seconds": int(video.get("duration_min", 0)) * 60,
"view_count": 800 + index * 137,
"comment_count": 20 + (index * 29) % 760,
"is_published": True,
"published_at": timezone.now(),
},
)
self.stdout.write(
self.style.SUCCESS(
f"已导入 MathBTI {assessment.version}、人生内容、LaTeX 课程和 {len(videos)} 条视频"
)
)
@@ -0,0 +1,170 @@
# Generated by Django 4.2.23 on 2026-08-08 11:15
from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion
import uuid
class Migration(migrations.Migration):
initial = True
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.CreateModel(
name='Character',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('slug', models.SlugField(unique=True)),
('name', models.CharField(max_length=80)),
('profile', models.TextField(blank=True)),
('metadata', models.JSONField(blank=True, default=dict)),
],
),
migrations.CreateModel(
name='MathBTIAssessment',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('version', models.CharField(max_length=30, unique=True)),
('title', models.CharField(max_length=120)),
('definition', models.JSONField()),
('is_published', models.BooleanField(default=False)),
('published_at', models.DateTimeField(blank=True, null=True)),
],
),
migrations.CreateModel(
name='MathIdentity',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('code', models.CharField(max_length=8, unique=True)),
('name', models.CharField(max_length=80)),
('clan', models.CharField(max_length=40)),
('mathematician', models.CharField(max_length=80)),
('description', models.TextField(blank=True)),
('initial_abilities', models.JSONField(blank=True, default=dict)),
('portrait', models.CharField(blank=True, max_length=200)),
],
),
migrations.CreateModel(
name='Story',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('slug', models.SlugField(unique=True)),
('title', models.CharField(max_length=120)),
('summary', models.TextField(blank=True)),
('kind', models.CharField(choices=[('flagship', '旗舰人生'), ('skill', '人物 Skill'), ('special', '特别篇')], default='flagship', max_length=16)),
('estimated_minutes', models.PositiveIntegerField(default=20)),
('is_visible', models.BooleanField(default=True)),
],
),
migrations.CreateModel(
name='UserRelationship',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('trust', models.SmallIntegerField(default=0)),
('rivalry', models.SmallIntegerField(default=0)),
('debt', models.SmallIntegerField(default=0)),
('cooperation', models.SmallIntegerField(default=0)),
('metadata', models.JSONField(blank=True, default=dict)),
('character', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='math_life.character')),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
],
),
migrations.CreateModel(
name='StoryVersion',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('version', models.PositiveIntegerField()),
('content', models.JSONField()),
('is_published', models.BooleanField(default=False)),
('published_at', models.DateTimeField(blank=True, null=True)),
('created_at', models.DateTimeField(auto_now_add=True)),
('story', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='versions', to='math_life.story')),
],
options={
'ordering': ['story', '-version'],
},
),
migrations.CreateModel(
name='StoryRun',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('current_node', models.CharField(max_length=100)),
('state', models.JSONField(default=dict)),
('status', models.CharField(choices=[('active', '进行中'), ('completed', '已完成'), ('abandoned', '已放弃')], default='active', max_length=16)),
('ending_code', models.CharField(blank=True, max_length=80)),
('started_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
('completed_at', models.DateTimeField(blank=True, null=True)),
('story_version', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='runs', to='math_life.storyversion')),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='story_runs', to=settings.AUTH_USER_MODEL)),
],
options={
'ordering': ['-updated_at'],
},
),
migrations.CreateModel(
name='StoryChoice',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('sequence', models.PositiveIntegerField()),
('idempotency_key', models.CharField(blank=True, max_length=80, null=True)),
('node_id', models.CharField(max_length=100)),
('choice_index', models.PositiveIntegerField()),
('choice_text', models.CharField(max_length=300)),
('effects', models.JSONField(blank=True, default=dict)),
('created_at', models.DateTimeField(auto_now_add=True)),
('run', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='choices', to='math_life.storyrun')),
],
options={
'ordering': ['sequence'],
},
),
migrations.CreateModel(
name='SkillPackage',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('real_person', models.CharField(max_length=120)),
('historical_context', models.TextField()),
('fact_sources', models.JSONField(default=list)),
('fictional_scope', models.TextField()),
('creator', models.CharField(max_length=120)),
('story', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='skill_package', to='math_life.story')),
],
),
migrations.CreateModel(
name='MathBTIResult',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('answers', models.JSONField(default=list)),
('axis_scores', models.JSONField(default=dict)),
('created_at', models.DateTimeField(auto_now_add=True)),
('assessment', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, to='math_life.mathbtiassessment')),
('identity', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, to='math_life.mathidentity')),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='mathbti_results', to=settings.AUTH_USER_MODEL)),
],
options={
'ordering': ['-created_at'],
},
),
migrations.AddConstraint(
model_name='userrelationship',
constraint=models.UniqueConstraint(fields=('user', 'character'), name='unique_user_character'),
),
migrations.AddConstraint(
model_name='storyversion',
constraint=models.UniqueConstraint(fields=('story', 'version'), name='unique_story_version'),
),
migrations.AddConstraint(
model_name='storychoice',
constraint=models.UniqueConstraint(fields=('run', 'sequence'), name='unique_run_choice_sequence'),
),
migrations.AddConstraint(
model_name='storychoice',
constraint=models.UniqueConstraint(fields=('run', 'idempotency_key'), name='unique_run_choice_idempotency'),
),
]
+151
View File
@@ -0,0 +1,151 @@
import uuid
from django.conf import settings
from django.db import models
class MathIdentity(models.Model):
code = models.CharField(max_length=8, unique=True)
name = models.CharField(max_length=80)
clan = models.CharField(max_length=40)
mathematician = models.CharField(max_length=80)
description = models.TextField(blank=True)
initial_abilities = models.JSONField(default=dict, blank=True)
portrait = models.CharField(max_length=200, blank=True)
def __str__(self):
return f"{self.code} {self.name}"
class MathBTIAssessment(models.Model):
version = models.CharField(max_length=30, unique=True)
title = models.CharField(max_length=120)
definition = models.JSONField()
is_published = models.BooleanField(default=False)
published_at = models.DateTimeField(null=True, blank=True)
def __str__(self):
return self.title
class MathBTIResult(models.Model):
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="mathbti_results")
assessment = models.ForeignKey(MathBTIAssessment, on_delete=models.PROTECT)
identity = models.ForeignKey(MathIdentity, on_delete=models.PROTECT)
answers = models.JSONField(default=list)
axis_scores = models.JSONField(default=dict)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
ordering = ["-created_at"]
class Character(models.Model):
slug = models.SlugField(unique=True)
name = models.CharField(max_length=80)
profile = models.TextField(blank=True)
metadata = models.JSONField(default=dict, blank=True)
def __str__(self):
return self.name
class Story(models.Model):
class Kind(models.TextChoices):
FLAGSHIP = "flagship", "旗舰人生"
SKILL = "skill", "人物 Skill"
SPECIAL = "special", "特别篇"
slug = models.SlugField(unique=True)
title = models.CharField(max_length=120)
summary = models.TextField(blank=True)
kind = models.CharField(max_length=16, choices=Kind.choices, default=Kind.FLAGSHIP)
estimated_minutes = models.PositiveIntegerField(default=20)
is_visible = models.BooleanField(default=True)
def __str__(self):
return self.title
class StoryVersion(models.Model):
story = models.ForeignKey(Story, on_delete=models.CASCADE, related_name="versions")
version = models.PositiveIntegerField()
content = models.JSONField()
is_published = models.BooleanField(default=False)
published_at = models.DateTimeField(null=True, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
constraints = [
models.UniqueConstraint(fields=("story", "version"), name="unique_story_version")
]
ordering = ["story", "-version"]
def __str__(self):
return f"{self.story} v{self.version}"
class StoryRun(models.Model):
class Status(models.TextChoices):
ACTIVE = "active", "进行中"
COMPLETED = "completed", "已完成"
ABANDONED = "abandoned", "已放弃"
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="story_runs")
story_version = models.ForeignKey(StoryVersion, on_delete=models.PROTECT, related_name="runs")
current_node = models.CharField(max_length=100)
state = models.JSONField(default=dict)
status = models.CharField(max_length=16, choices=Status.choices, default=Status.ACTIVE)
ending_code = models.CharField(max_length=80, blank=True)
started_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
completed_at = models.DateTimeField(null=True, blank=True)
class Meta:
ordering = ["-updated_at"]
class StoryChoice(models.Model):
run = models.ForeignKey(StoryRun, on_delete=models.CASCADE, related_name="choices")
sequence = models.PositiveIntegerField()
idempotency_key = models.CharField(max_length=80, null=True, blank=True)
node_id = models.CharField(max_length=100)
choice_index = models.PositiveIntegerField()
choice_text = models.CharField(max_length=300)
effects = models.JSONField(default=dict, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
constraints = [
models.UniqueConstraint(fields=("run", "sequence"), name="unique_run_choice_sequence"),
models.UniqueConstraint(
fields=("run", "idempotency_key"), name="unique_run_choice_idempotency"
),
]
ordering = ["sequence"]
class UserRelationship(models.Model):
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
character = models.ForeignKey(Character, on_delete=models.CASCADE)
trust = models.SmallIntegerField(default=0)
rivalry = models.SmallIntegerField(default=0)
debt = models.SmallIntegerField(default=0)
cooperation = models.SmallIntegerField(default=0)
metadata = models.JSONField(default=dict, blank=True)
class Meta:
constraints = [
models.UniqueConstraint(fields=("user", "character"), name="unique_user_character")
]
class SkillPackage(models.Model):
story = models.OneToOneField(Story, on_delete=models.CASCADE, related_name="skill_package")
real_person = models.CharField(max_length=120)
historical_context = models.TextField()
fact_sources = models.JSONField(default=list)
fictional_scope = models.TextField()
creator = models.CharField(max_length=120)
+192
View File
@@ -0,0 +1,192 @@
from copy import deepcopy
from django.db import transaction
from django.utils import timezone
from rest_framework.exceptions import ValidationError
from .models import StoryChoice, StoryRun, StoryVersion
def score_mathbti(definition, answer_indexes):
questions = definition.get("questions", [])
if len(answer_indexes) != len(questions):
raise ValidationError({"answers": "答案数量与题目数量不一致"})
axis_scores = {axis["id"]: 0 for axis in definition.get("axes", [])}
for question, raw_index in zip(questions, answer_indexes):
try:
option_index = int(raw_index)
options = question["options"]
if option_index < 0 or option_index >= len(options):
raise IndexError
option = options[option_index]
except (IndexError, KeyError, TypeError, ValueError) as exc:
raise ValidationError({"answers": f"题目 {question.get('id')} 的答案无效"}) from exc
axis_scores[question["axis"]] += int(option["score"])
cutoff = int(definition.get("scoring", {}).get("cutoff", 4))
axis_order = definition.get("scoring", {}).get("axes", list(axis_scores))
identity_code = "".join("1" if axis_scores[axis] > cutoff else "0" for axis in axis_order)
return identity_code, axis_scores
def validate_story_content(content):
nodes = content.get("nodes")
start = content.get("start_node")
errors = []
if not isinstance(nodes, dict) or not nodes:
return ["nodes 必须是非空对象"]
if start not in nodes:
errors.append("start_node 不存在")
referenced = set()
for node_id, node in nodes.items():
if not isinstance(node.get("choices", []), list):
errors.append(f"{node_id}.choices 必须是数组")
continue
for choice in node.get("choices", []):
target = choice.get("next")
if not target:
errors.append(f"{node_id} 存在缺少 next 的选项")
elif target not in nodes:
errors.append(f"{node_id} 引用了不存在的节点 {target}")
else:
referenced.add(target)
if start in nodes:
reachable = set()
pending = [start]
while pending:
node_id = pending.pop()
if node_id in reachable:
continue
reachable.add(node_id)
pending.extend(
choice.get("next")
for choice in nodes[node_id].get("choices", [])
if choice.get("next") in nodes
)
unreachable = sorted(set(nodes) - reachable)
if unreachable:
errors.append(f"存在不可达节点: {', '.join(unreachable[:10])}")
return errors
def _merge_effects(state, effects):
result = deepcopy(state)
for key, value in effects.items():
if isinstance(value, dict):
result[key] = _merge_effects(result.get(key, {}), value)
elif isinstance(value, (int, float)):
result[key] = result.get(key, 0) + value
else:
result[key] = value
return result
def _node_payload(run):
node = run.story_version.content["nodes"][run.current_node]
choices = [
{"index": index, "text": choice.get("text", "")}
for index, choice in enumerate(node.get("choices", []))
]
return {
"run_id": run.id,
"story": run.story_version.story.title,
"version": run.story_version.version,
"status": run.status,
"current_node": run.current_node,
"state": run.state,
"node": {
"scene": node.get("scene", ""),
"character": node.get("character", ""),
"sprite": node.get("sprite", ""),
"choices": choices,
},
}
@transaction.atomic
def start_story(user, story):
version = (
StoryVersion.objects.select_related("story")
.filter(story=story, is_published=True)
.order_by("-version")
.first()
)
if version is None:
raise ValidationError("该人生尚未发布")
active = StoryRun.objects.filter(
user=user, story_version=version, status=StoryRun.Status.ACTIVE
).first()
if active:
return _node_payload(active)
run = StoryRun.objects.create(
user=user,
story_version=version,
current_node=version.content["start_node"],
state={},
)
return _node_payload(run)
def get_run_payload(run):
return _node_payload(run)
@transaction.atomic
def make_choice(*, run_id, user, choice_index, idempotency_key=None):
run = (
StoryRun.objects.select_for_update()
.select_related("story_version__story")
.get(id=run_id, user=user)
)
if idempotency_key:
existing = StoryChoice.objects.filter(
run=run, idempotency_key=idempotency_key
).first()
if existing:
return _node_payload(run)
if run.status != StoryRun.Status.ACTIVE:
raise ValidationError("该人生已经结束")
node_id = run.current_node
node = run.story_version.content["nodes"][node_id]
choices = node.get("choices", [])
try:
normalized_index = int(choice_index)
if normalized_index < 0 or normalized_index >= len(choices):
raise IndexError
choice = choices[normalized_index]
except (IndexError, TypeError, ValueError) as exc:
raise ValidationError({"choice_index": "选项不存在"}) from exc
sequence = run.choices.count() + 1
effects = choice.get("effects", {})
StoryChoice.objects.create(
run=run,
sequence=sequence,
idempotency_key=idempotency_key,
node_id=node_id,
choice_index=normalized_index,
choice_text=choice.get("text", ""),
effects=effects,
)
run.state = _merge_effects(run.state, effects)
run.current_node = choice["next"]
next_node = run.story_version.content["nodes"][run.current_node]
if not next_node.get("choices"):
run.status = StoryRun.Status.COMPLETED
run.ending_code = run.current_node
run.completed_at = timezone.now()
run.save(
update_fields=[
"state",
"current_node",
"status",
"ending_code",
"completed_at",
"updated_at",
]
)
return _node_payload(run)
+144
View File
@@ -0,0 +1,144 @@
import pytest
from rest_framework.exceptions import ValidationError
from accounts.models import User
from math_life.models import Story, StoryChoice, StoryRun, StoryVersion
from math_life.services import (
make_choice,
score_mathbti,
start_story,
validate_story_content,
)
@pytest.fixture
def mathbti_definition():
return {
"axes": [{"id": "style"}, {"id": "purpose"}],
"scoring": {"axes": ["style", "purpose"], "cutoff": 1},
"questions": [
{
"id": 1,
"axis": "style",
"options": [{"score": 0}, {"score": 2}],
},
{
"id": 2,
"axis": "purpose",
"options": [{"score": 0}, {"score": 2}],
},
],
}
def test_score_mathbti_按轴汇总并生成身份编码(mathbti_definition):
code, scores = score_mathbti(mathbti_definition, [1, 0])
assert code == "10"
assert scores == {"style": 2, "purpose": 0}
def test_score_mathbti_答案数量不一致时拒绝(mathbti_definition):
with pytest.raises(ValidationError, match="答案数量"):
score_mathbti(mathbti_definition, [1])
def test_score_mathbti_负数选项索引必须拒绝(mathbti_definition):
with pytest.raises(ValidationError, match="答案无效"):
score_mathbti(mathbti_definition, [-1, 0])
def test_validate_story_content_报告缺失引用与不可达节点():
document = {
"start_node": "start",
"nodes": {
"start": {"choices": [{"text": "错误入口", "next": "missing"}]},
"orphan": {"choices": []},
},
}
errors = validate_story_content(document)
assert any("不存在的节点 missing" in error for error in errors)
assert any("不可达节点: orphan" in error for error in errors)
@pytest.fixture
def story_setup(db):
user = User.objects.create_user(
username="story_user",
password="StrongPass_2026",
nickname="剧情用户",
)
story = Story.objects.create(slug="test-story", title="测试人生")
version = StoryVersion.objects.create(
story=story,
version=1,
is_published=True,
content={
"start_node": "start",
"nodes": {
"start": {
"scene": "起点",
"choices": [
{
"text": "向左",
"next": "left_end",
"effects": {"energy": -1, "favorability": {"高斯": 2}},
},
{
"text": "向右",
"next": "right_end",
"effects": {"energy": 2},
},
],
},
"left_end": {"scene": "左结局", "choices": []},
"right_end": {"scene": "右结局", "choices": []},
},
},
)
return user, story, version
@pytest.mark.django_db
def test_make_choice_应用嵌套效果完成结局且幂等(story_setup):
user, story, _ = story_setup
started = start_story(user, story)
result = make_choice(
run_id=started["run_id"],
user=user,
choice_index=0,
idempotency_key="choice-1",
)
replay = make_choice(
run_id=started["run_id"],
user=user,
choice_index=0,
idempotency_key="choice-1",
)
assert result["status"] == StoryRun.Status.COMPLETED
assert result["current_node"] == "left_end"
assert result["state"] == {"energy": -1, "favorability": {"高斯": 2}}
assert replay["current_node"] == "left_end"
assert StoryChoice.objects.count() == 1
@pytest.mark.django_db
def test_make_choice_负数索引必须拒绝且存档不变(story_setup):
user, story, _ = story_setup
started = start_story(user, story)
with pytest.raises(ValidationError, match="选项不存在"):
make_choice(
run_id=started["run_id"],
user=user,
choice_index=-1,
idempotency_key="invalid-choice",
)
run = StoryRun.objects.get(id=started["run_id"])
assert run.current_node == "start"
assert StoryChoice.objects.count() == 0
+3
View File
@@ -0,0 +1,3 @@
from django.test import TestCase
# Create your tests here.
+20
View File
@@ -0,0 +1,20 @@
from django.urls import path
from .views import (
MathBTIAssessmentView,
MathBTISubmitView,
StoryChoiceView,
StoryListView,
StoryRunView,
StoryStartView,
)
urlpatterns = [
path("mathbti/", MathBTIAssessmentView.as_view(), name="mathbti-assessment"),
path("mathbti/submit/", MathBTISubmitView.as_view(), name="mathbti-submit"),
path("stories/", StoryListView.as_view(), name="story-list"),
path("stories/<slug:slug>/start/", StoryStartView.as_view(), name="story-start"),
path("runs/<uuid:run_id>/", StoryRunView.as_view(), name="story-run"),
path("runs/<uuid:run_id>/choice/", StoryChoiceView.as_view(), name="story-choice"),
]
+129
View File
@@ -0,0 +1,129 @@
from copy import deepcopy
from django.shortcuts import get_object_or_404
from rest_framework import permissions, status
from rest_framework.response import Response
from rest_framework.views import APIView
from progression.services import initialize_math_identity
from .models import (
MathBTIAssessment,
MathBTIResult,
MathIdentity,
Story,
StoryRun,
)
from .services import get_run_payload, make_choice, score_mathbti, start_story
class MathBTIAssessmentView(APIView):
permission_classes = [permissions.AllowAny]
def get(self, request):
assessment = (
MathBTIAssessment.objects.filter(is_published=True).order_by("-published_at").first()
)
if assessment is None:
return Response(
{"error": {"code": "not_published", "message": "MathBTI 尚未发布"}},
status=status.HTTP_404_NOT_FOUND,
)
definition = deepcopy(assessment.definition)
for question in definition.get("questions", []):
for option in question.get("options", []):
option.pop("score", None)
definition.pop("results", None)
return Response({"version": assessment.version, "definition": definition})
class MathBTISubmitView(APIView):
permission_classes = [permissions.AllowAny]
def post(self, request):
assessment = get_object_or_404(
MathBTIAssessment,
version=request.data.get("version"),
is_published=True,
)
identity_code, axis_scores = score_mathbti(
assessment.definition,
request.data.get("answers", []),
)
identity = get_object_or_404(MathIdentity, code=identity_code)
if request.user.is_authenticated:
result = MathBTIResult.objects.create(
user=request.user,
assessment=assessment,
identity=identity,
answers=request.data.get("answers", []),
axis_scores=axis_scores,
)
initialize_math_identity(request.user, identity)
result_id = result.id
else:
result_id = None
return Response(
{
"id": result_id,
"identity": {
"code": identity.code,
"name": identity.name,
"clan": identity.clan,
"mathematician": identity.mathematician,
"description": identity.description,
"portrait": identity.portrait,
"initial_abilities": identity.initial_abilities,
},
"axis_scores": axis_scores,
},
status=status.HTTP_201_CREATED,
)
class StoryListView(APIView):
permission_classes = [permissions.AllowAny]
def get(self, request):
stories = Story.objects.filter(is_visible=True).order_by("kind", "title")
return Response(
[
{
"slug": story.slug,
"title": story.title,
"summary": story.summary,
"kind": story.kind,
"estimated_minutes": story.estimated_minutes,
"available": story.versions.filter(is_published=True).exists(),
}
for story in stories
]
)
class StoryStartView(APIView):
def post(self, request, slug):
story = get_object_or_404(Story, slug=slug, is_visible=True)
return Response(start_story(request.user, story), status=status.HTTP_201_CREATED)
class StoryRunView(APIView):
def get(self, request, run_id):
run = get_object_or_404(
StoryRun.objects.select_related("story_version__story"),
id=run_id,
user=request.user,
)
return Response(get_run_payload(run))
class StoryChoiceView(APIView):
def post(self, request, run_id):
get_object_or_404(StoryRun, id=run_id, user=request.user)
payload = make_choice(
run_id=run_id,
user=request.user,
choice_index=request.data.get("choice_index"),
idempotency_key=request.headers.get("Idempotency-Key"),
)
return Response(payload)
View File
+10
View File
@@ -0,0 +1,10 @@
from django.contrib import admin
from .models import Card, RewardTransaction, UserAbility, UserCard, UserPet
admin.site.register(UserPet)
admin.site.register(UserAbility)
admin.site.register(Card)
admin.site.register(UserCard)
admin.site.register(RewardTransaction)
+6
View File
@@ -0,0 +1,6 @@
from django.apps import AppConfig
class ProgressionConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'progression'
@@ -0,0 +1,83 @@
# Generated by Django 4.2.23 on 2026-08-08 11:15
from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
initial = True
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.CreateModel(
name='Card',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('slug', models.SlugField(unique=True)),
('name', models.CharField(max_length=80)),
('mathematician', models.CharField(max_length=80)),
('description', models.TextField()),
('image', models.CharField(blank=True, max_length=200)),
],
),
migrations.CreateModel(
name='UserPet',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('name', models.CharField(default='小葫芦', max_length=40)),
('level', models.PositiveIntegerField(default=1)),
('experience', models.PositiveIntegerField(default=0)),
('appearance', models.JSONField(blank=True, default=dict)),
('created_at', models.DateTimeField(auto_now_add=True)),
('user', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='pet', to=settings.AUTH_USER_MODEL)),
],
),
migrations.CreateModel(
name='UserCard',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('source', models.CharField(max_length=80)),
('acquired_at', models.DateTimeField(auto_now_add=True)),
('card', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, to='progression.card')),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='cards', to=settings.AUTH_USER_MODEL)),
],
),
migrations.CreateModel(
name='UserAbility',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('dimension', models.CharField(choices=[('vision', '眼光'), ('humanities', '人文'), ('detection', '侦探'), ('modeling', '建模'), ('connection', '联结')], max_length=20)),
('level', models.PositiveIntegerField(default=0)),
('fragments', models.PositiveIntegerField(default=0)),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='abilities', to=settings.AUTH_USER_MODEL)),
],
),
migrations.CreateModel(
name='RewardTransaction',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('idempotency_key', models.CharField(max_length=100)),
('source', models.CharField(max_length=80)),
('rewards', models.JSONField(default=dict)),
('created_at', models.DateTimeField(auto_now_add=True)),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='rewards', to=settings.AUTH_USER_MODEL)),
],
),
migrations.AddConstraint(
model_name='usercard',
constraint=models.UniqueConstraint(fields=('user', 'card'), name='unique_user_card'),
),
migrations.AddConstraint(
model_name='userability',
constraint=models.UniqueConstraint(fields=('user', 'dimension'), name='unique_user_ability'),
),
migrations.AddConstraint(
model_name='rewardtransaction',
constraint=models.UniqueConstraint(fields=('user', 'idempotency_key'), name='unique_user_reward_transaction'),
),
]

Some files were not shown because too many files have changed in this diff Show More