Compare commits
10
Commits
4e4cdf31ac
...
306068bc22
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
306068bc22 | ||
|
|
147107501b | ||
|
|
f37571f395 | ||
|
|
3944e11330 | ||
|
|
8d440b9e4a | ||
|
|
16e4f97e74 | ||
|
|
d6503bb46f | ||
|
|
8fbbd2e6b0 | ||
|
|
b04de9f5c1 | ||
|
|
ec2e629db2 |
@@ -20,7 +20,18 @@ OSS_ACCESS_KEY_ID=
|
||||
OSS_SECRET_ACCESS_KEY=
|
||||
OSS_BUCKET_NAME=
|
||||
OSS_ENDPOINT_URL=
|
||||
# 绑定到 OSS/COS 源站的 CDN 加速域名(阿里云 CDN / 腾讯云 CDN / 又拍云),留空则直接使用 OSS 源站地址
|
||||
OSS_CDN_DOMAIN=
|
||||
|
||||
# 发布后前端 ISR 缓存失效 Webhook
|
||||
FRONTEND_REVALIDATE_URL=https://www.example.com/api/revalidate
|
||||
REVALIDATE_SECRET=change-me
|
||||
|
||||
# 前端站点基础 URL(用于邮件中的链接,如 PIPL 数据删除确认链接)
|
||||
FRONTEND_BASE_URL=https://www.example.com
|
||||
|
||||
# 部署引导(python manage.py bootstrap_deployment):首次部署时自动创建的超级管理员账号
|
||||
# 三项均设置且该用户名不存在时才会创建;账号已存在则跳过,不会重置密码。
|
||||
DJANGO_SUPERUSER_USERNAME=
|
||||
DJANGO_SUPERUSER_EMAIL=
|
||||
DJANGO_SUPERUSER_PASSWORD=
|
||||
|
||||
+46
-1
@@ -1,13 +1,58 @@
|
||||
# ---- Python ----
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
*$py.class
|
||||
*.egg-info/
|
||||
.eggs/
|
||||
.Python
|
||||
build/
|
||||
dist/
|
||||
pip-wheel-metadata/
|
||||
|
||||
# ---- Django / Wagtail ----
|
||||
*.sqlite3
|
||||
*.sqlite3-journal
|
||||
.env
|
||||
.env.*
|
||||
!.env.example
|
||||
/media/
|
||||
/staticfiles/
|
||||
/static/
|
||||
|
||||
# ---- 虚拟环境 ----
|
||||
.venv/
|
||||
venv/
|
||||
*.egg-info/
|
||||
env/
|
||||
ENV/
|
||||
|
||||
# ---- 测试 / 覆盖率 ----
|
||||
.pytest_cache/
|
||||
.coverage
|
||||
.coverage.*
|
||||
htmlcov/
|
||||
.tox/
|
||||
.mypy_cache/
|
||||
.ruff_cache/
|
||||
|
||||
# ---- 日志 / 本地临时文件 ----
|
||||
*.log
|
||||
*.tmp
|
||||
.git_commit_msg.txt
|
||||
pytest_output.txt
|
||||
pytest_full_output.txt
|
||||
|
||||
# ---- IDE / 编辑器 ----
|
||||
.vscode/
|
||||
.idea/
|
||||
*.swp
|
||||
*.swo
|
||||
|
||||
# ---- 操作系统 ----
|
||||
.DS_Store
|
||||
Thumbs.db
|
||||
desktop.ini
|
||||
|
||||
# ---- 前端(node_modules/.next 以防万一在根目录出现,正常情况下由 frontend/ 独立仓库管理)----
|
||||
node_modules/
|
||||
.next/
|
||||
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
"""apps.api 单元测试:预览专用接口 `page_preview`(apps/api/urls.py 的
|
||||
PagePreviewAPIViewSet),覆盖 §2.7/§2.15 描述的"仅凭有效 content_type+token
|
||||
才能读取草稿快照"行为。"""
|
||||
import pytest
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def preview_token(home_page):
|
||||
"""为 home_page 生成一条预览快照记录,返回可用于查询接口的 token。"""
|
||||
preview = home_page.create_page_preview()
|
||||
preview.save()
|
||||
return preview.token
|
||||
|
||||
|
||||
def test_page_preview_returns_draft_snapshot_for_valid_token(home_page, preview_token):
|
||||
client = APIClient()
|
||||
response = client.get(
|
||||
"/api/v2/page_preview/",
|
||||
{"content_type": "home.homepage", "token": preview_token},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.data["title"] == home_page.title
|
||||
|
||||
|
||||
def test_page_preview_returns_404_for_unknown_token(home_page):
|
||||
client = APIClient()
|
||||
response = client.get(
|
||||
"/api/v2/page_preview/",
|
||||
{"content_type": "home.homepage", "token": "does-not-exist"},
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_page_preview_returns_404_for_mismatched_content_type(home_page, preview_token):
|
||||
"""token 是为 home.homepage 生成的,用错误的 content_type 查询应查不到记录。"""
|
||||
client = APIClient()
|
||||
response = client.get(
|
||||
"/api/v2/page_preview/",
|
||||
{"content_type": "blog.blogpage", "token": preview_token},
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
@@ -2,13 +2,61 @@
|
||||
Wagtail API v2 路由配置。
|
||||
详见 documents/设计方案分析与完善版.md §2.7 Headless API 设计规范。
|
||||
"""
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
from django.http import Http404
|
||||
from rest_framework.response import Response
|
||||
from wagtail.api.v2.views import PagesAPIViewSet
|
||||
from wagtail.api.v2.router import WagtailAPIRouter
|
||||
from wagtail.images.api.v2.views import ImagesAPIViewSet
|
||||
from wagtail.documents.api.v2.views import DocumentsAPIViewSet
|
||||
from wagtail_headless_preview.models import PagePreview
|
||||
|
||||
api_router = WagtailAPIRouter("wagtailapi")
|
||||
|
||||
api_router.register_endpoint("pages", PagesAPIViewSet)
|
||||
api_router.register_endpoint("images", ImagesAPIViewSet)
|
||||
api_router.register_endpoint("documents", DocumentsAPIViewSet)
|
||||
|
||||
|
||||
class PagePreviewAPIViewSet(PagesAPIViewSet):
|
||||
"""
|
||||
预览模式专用只读接口:根据 `content_type`(app_label.model,小写)+ `token`
|
||||
从 `wagtail_headless_preview` 存储的草稿快照中取出对应页面实例并序列化返回。
|
||||
`token` 由编辑器在 Wagtail 后台点击"预览"时生成(详见 §2.15/§2.7),
|
||||
未知或已过期的 token 会返回 404,不会暴露任何非预览用途的草稿数据。
|
||||
"""
|
||||
|
||||
known_query_parameters = PagesAPIViewSet.known_query_parameters.union(
|
||||
["content_type", "token"]
|
||||
)
|
||||
|
||||
def listing_view(self, request):
|
||||
# 统一走 detail_view,保证序列化格式与详情接口一致;pk 仅占位,实际查找依赖 content_type+token。
|
||||
self.action = "detail_view"
|
||||
return self.detail_view(request, 0)
|
||||
|
||||
def detail_view(self, request, pk):
|
||||
page = self.get_object()
|
||||
serializer = self.get_serializer(page)
|
||||
return Response(serializer.data)
|
||||
|
||||
def get_object(self):
|
||||
try:
|
||||
app_label, model = self.request.GET["content_type"].split(".")
|
||||
content_type = ContentType.objects.get(app_label=app_label, model=model)
|
||||
page_preview = PagePreview.objects.get(
|
||||
content_type=content_type, token=self.request.GET["token"]
|
||||
)
|
||||
except (KeyError, ValueError, ContentType.DoesNotExist, PagePreview.DoesNotExist):
|
||||
# 缺少参数/content_type 格式错误/未知或已过期的 token,统一返回 404,
|
||||
# 不暴露内部异常细节(详见类文档字符串)。
|
||||
raise Http404("No matching page preview found.")
|
||||
page = page_preview.as_page()
|
||||
if not page.pk:
|
||||
# 新建(尚未保存)页面的预览没有真实主键,填充占位值避免 API 路由生成 URL 时报错
|
||||
page.pk = 0
|
||||
return page
|
||||
|
||||
|
||||
api_router.register_endpoint("page_preview", PagePreviewAPIViewSet)
|
||||
|
||||
|
||||
File diff suppressed because one or more lines are too long
@@ -0,0 +1,85 @@
|
||||
"""apps.blog 单元测试:页面树结构、标签与列表排序。"""
|
||||
import pytest
|
||||
from django.utils import timezone
|
||||
|
||||
from apps.blog.models import BlogIndexPage, BlogPage
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def blog_index(home_page):
|
||||
index = BlogIndexPage(title="博客", slug="blog", intro="技术与行业洞察")
|
||||
home_page.add_child(instance=index)
|
||||
return index
|
||||
|
||||
|
||||
def test_blog_page_can_be_created_under_index(blog_index):
|
||||
post = BlogPage(
|
||||
title="文章一",
|
||||
slug="post-1",
|
||||
published_at=timezone.now(),
|
||||
intro="摘要一",
|
||||
)
|
||||
blog_index.add_child(instance=post)
|
||||
|
||||
assert BlogPage.objects.live().descendant_of(blog_index).count() == 1
|
||||
|
||||
|
||||
def test_blog_page_supports_tags(blog_index):
|
||||
post = BlogPage(
|
||||
title="文章一",
|
||||
slug="post-1",
|
||||
published_at=timezone.now(),
|
||||
)
|
||||
blog_index.add_child(instance=post)
|
||||
|
||||
post.tags.add("Django", "Wagtail")
|
||||
post.save()
|
||||
|
||||
post.refresh_from_db()
|
||||
assert {t.name for t in post.tags.all()} == {"Django", "Wagtail"}
|
||||
|
||||
|
||||
def test_blog_index_context_orders_by_published_at_desc(blog_index):
|
||||
older = BlogPage(
|
||||
title="较早的文章",
|
||||
slug="older-post",
|
||||
published_at=timezone.now() - timezone.timedelta(days=5),
|
||||
)
|
||||
blog_index.add_child(instance=older)
|
||||
|
||||
newer = BlogPage(
|
||||
title="较新的文章",
|
||||
slug="newer-post",
|
||||
published_at=timezone.now(),
|
||||
)
|
||||
blog_index.add_child(instance=newer)
|
||||
|
||||
context = blog_index.get_context(request=None)
|
||||
titles = [p.title for p in context["posts"]]
|
||||
|
||||
assert titles == ["较新的文章", "较早的文章"]
|
||||
|
||||
|
||||
def test_blog_index_context_excludes_draft_posts(blog_index):
|
||||
live_post = BlogPage(
|
||||
title="已发布文章",
|
||||
slug="live-post",
|
||||
published_at=timezone.now(),
|
||||
)
|
||||
blog_index.add_child(instance=live_post)
|
||||
|
||||
draft_post = BlogPage(
|
||||
title="草稿文章",
|
||||
slug="draft-post",
|
||||
published_at=timezone.now(),
|
||||
live=False,
|
||||
)
|
||||
blog_index.add_child(instance=draft_post)
|
||||
|
||||
context = blog_index.get_context(request=None)
|
||||
titles = [p.title for p in context["posts"]]
|
||||
|
||||
assert "已发布文章" in titles
|
||||
assert "草稿文章" not in titles
|
||||
File diff suppressed because one or more lines are too long
@@ -0,0 +1,67 @@
|
||||
"""
|
||||
apps.core 全局 Snippet 只读接口。挂载在 /api/v1/custom/core/
|
||||
(详见 apps/core/urls.py 与 documents/设计方案分析与完善版.md §2.7)。
|
||||
公开只读数据,限流走 DRF ScopedRateThrottle 的 "public" scope(100/min)。
|
||||
"""
|
||||
from rest_framework.response import Response
|
||||
from rest_framework.throttling import ScopedRateThrottle
|
||||
from rest_framework.views import APIView
|
||||
|
||||
from .models import NavigationMenu, Partner, SiteSettings, TeamMember, Testimonial
|
||||
from .serializers import (
|
||||
NavigationMenuSerializer,
|
||||
PartnerSerializer,
|
||||
SiteSettingsSerializer,
|
||||
TeamMemberSerializer,
|
||||
TestimonialSerializer,
|
||||
)
|
||||
|
||||
|
||||
class TeamMemberListView(APIView):
|
||||
throttle_classes = [ScopedRateThrottle]
|
||||
throttle_scope = "public"
|
||||
|
||||
def get(self, request, *args, **kwargs):
|
||||
members = TeamMember.objects.all()
|
||||
return Response({"items": TeamMemberSerializer(members, many=True).data})
|
||||
|
||||
|
||||
class TestimonialListView(APIView):
|
||||
throttle_classes = [ScopedRateThrottle]
|
||||
throttle_scope = "public"
|
||||
|
||||
def get(self, request, *args, **kwargs):
|
||||
testimonials = Testimonial.objects.all()
|
||||
return Response({"items": TestimonialSerializer(testimonials, many=True).data})
|
||||
|
||||
|
||||
class PartnerListView(APIView):
|
||||
throttle_classes = [ScopedRateThrottle]
|
||||
throttle_scope = "public"
|
||||
|
||||
def get(self, request, *args, **kwargs):
|
||||
partners = Partner.objects.all()
|
||||
return Response({"items": PartnerSerializer(partners, many=True).data})
|
||||
|
||||
|
||||
class NavigationMenuView(APIView):
|
||||
"""根据 ?name= 查询指定导航菜单(如 main / footer)。"""
|
||||
|
||||
throttle_classes = [ScopedRateThrottle]
|
||||
throttle_scope = "public"
|
||||
|
||||
def get(self, request, *args, **kwargs):
|
||||
name = request.query_params.get("name", "main")
|
||||
menu = NavigationMenu.objects.prefetch_related("items").filter(name=name).first()
|
||||
if menu is None:
|
||||
return Response({"error": {"code": "NOT_FOUND", "message": "导航菜单不存在"}}, status=404)
|
||||
return Response(NavigationMenuSerializer(menu).data)
|
||||
|
||||
|
||||
class SiteSettingsView(APIView):
|
||||
throttle_classes = [ScopedRateThrottle]
|
||||
throttle_scope = "public"
|
||||
|
||||
def get(self, request, *args, **kwargs):
|
||||
settings_obj = SiteSettings.for_request(request)
|
||||
return Response(SiteSettingsSerializer(settings_obj).data)
|
||||
@@ -7,6 +7,8 @@ from wagtail import blocks
|
||||
from wagtail.images.blocks import ImageChooserBlock
|
||||
from wagtail.snippets.blocks import SnippetChooserBlock
|
||||
|
||||
from .serializers import serialize_image
|
||||
|
||||
|
||||
class CTAButtonBlock(blocks.StructBlock):
|
||||
text = blocks.CharBlock(max_length=50, label=_("按钮文案"))
|
||||
@@ -150,6 +152,113 @@ class FormBlock(blocks.StructBlock):
|
||||
}
|
||||
|
||||
|
||||
class ProductCardItemBlock(blocks.StructBlock):
|
||||
title = blocks.CharBlock(max_length=100, label=_("标题"))
|
||||
description = blocks.TextBlock(required=False, label=_("描述"))
|
||||
image = ImageChooserBlock(required=False, label=_("图片"))
|
||||
link = blocks.URLBlock(required=False, label=_("链接"))
|
||||
|
||||
|
||||
class ProductCardBlock(blocks.StructBlock):
|
||||
heading = blocks.CharBlock(max_length=100, required=False, label=_("模块标题"))
|
||||
items = blocks.ListBlock(ProductCardItemBlock(), label=_("产品列表"))
|
||||
|
||||
class Meta:
|
||||
icon = "grip"
|
||||
label = _("产品矩阵")
|
||||
|
||||
|
||||
class PricingPlanBlock(blocks.StructBlock):
|
||||
name = blocks.CharBlock(max_length=50, label=_("方案名称"))
|
||||
price = blocks.CharBlock(max_length=50, label=_("价格"))
|
||||
features = blocks.ListBlock(
|
||||
blocks.CharBlock(max_length=100), label=_("功能列表")
|
||||
)
|
||||
highlighted = blocks.BooleanBlock(
|
||||
required=False, default=False, label=_("突出显示")
|
||||
)
|
||||
button = CTAButtonBlock(required=False, label=_("按钮"))
|
||||
|
||||
|
||||
class PricingBlock(blocks.StructBlock):
|
||||
heading = blocks.CharBlock(max_length=100, required=False, label=_("模块标题"))
|
||||
plans = blocks.ListBlock(PricingPlanBlock(), label=_("定价方案"))
|
||||
|
||||
class Meta:
|
||||
icon = "tag"
|
||||
label = _("定价")
|
||||
|
||||
|
||||
class TimelineItemBlock(blocks.StructBlock):
|
||||
year = blocks.CharBlock(max_length=20, label=_("年份/时间"))
|
||||
event = blocks.TextBlock(label=_("事件描述"))
|
||||
|
||||
|
||||
class TimelineBlock(blocks.StructBlock):
|
||||
heading = blocks.CharBlock(max_length=100, required=False, label=_("模块标题"))
|
||||
items = blocks.ListBlock(TimelineItemBlock(), label=_("时间线条目"))
|
||||
|
||||
class Meta:
|
||||
icon = "date"
|
||||
label = _("发展历程")
|
||||
|
||||
|
||||
class TeamMemberChooserBlock(SnippetChooserBlock):
|
||||
"""选择 apps.core.TeamMember Snippet,展开完整字段供前端渲染(而非仅主键)。"""
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
super().__init__("core.TeamMember", **kwargs)
|
||||
|
||||
def get_api_representation(self, value, context=None):
|
||||
if value is None:
|
||||
return None
|
||||
return {
|
||||
"id": value.pk,
|
||||
"name": value.name,
|
||||
"role": value.role,
|
||||
"bio": value.bio,
|
||||
"photo": serialize_image(value.photo),
|
||||
}
|
||||
|
||||
|
||||
class TeamBlock(blocks.StructBlock):
|
||||
heading = blocks.CharBlock(max_length=100, required=False, label=_("模块标题"))
|
||||
members = blocks.ListBlock(TeamMemberChooserBlock(), label=_("团队成员"))
|
||||
|
||||
class Meta:
|
||||
icon = "group"
|
||||
label = _("团队展示")
|
||||
|
||||
|
||||
class TechStackItemBlock(blocks.StructBlock):
|
||||
icon = blocks.CharBlock(max_length=50, required=False, label=_("图标"))
|
||||
label = blocks.CharBlock(max_length=50, label=_("名称"))
|
||||
|
||||
|
||||
class TechStackBlock(blocks.StructBlock):
|
||||
heading = blocks.CharBlock(max_length=100, required=False, label=_("模块标题"))
|
||||
items = blocks.ListBlock(TechStackItemBlock(), label=_("技术项"))
|
||||
|
||||
class Meta:
|
||||
icon = "cogs"
|
||||
label = _("技术架构")
|
||||
|
||||
|
||||
class VideoBlock(blocks.StructBlock):
|
||||
heading = blocks.CharBlock(max_length=100, required=False, label=_("标题"))
|
||||
video_url = blocks.URLBlock(
|
||||
label=_("视频地址"),
|
||||
help_text=_(
|
||||
"建议使用自建 OSS/COS 存储的 mp4 直链,或腾讯视频/哔哩哔哩等国内平台的"
|
||||
"嵌入地址,避免使用 YouTube/Vimeo(详见 §2.15 国内可访问性要求)"
|
||||
),
|
||||
)
|
||||
poster = ImageChooserBlock(required=False, label=_("封面图"))
|
||||
|
||||
class Meta:
|
||||
icon = "media"
|
||||
label = _("视频")
|
||||
|
||||
|
||||
COMMON_BLOCKS = [
|
||||
("hero", HeroBlock()),
|
||||
@@ -160,5 +269,11 @@ COMMON_BLOCKS = [
|
||||
("logo_cloud", LogoCloudBlock()),
|
||||
("case_study", CaseStudyBlock()),
|
||||
("form", FormBlock()),
|
||||
("product_card", ProductCardBlock()),
|
||||
("pricing", PricingBlock()),
|
||||
("timeline", TimelineBlock()),
|
||||
("team", TeamBlock()),
|
||||
("tech_stack", TechStackBlock()),
|
||||
("video", VideoBlock()),
|
||||
("richtext", blocks.RichTextBlock(label=_("富文本"))),
|
||||
]
|
||||
|
||||
@@ -0,0 +1,56 @@
|
||||
"""
|
||||
部署引导命令:首次上线(或每次部署)时自动完成"能登录后台 + 拥有正确权限组"这两件事,
|
||||
避免每次部署都要人工登录服务器手动执行 `createsuperuser`。
|
||||
|
||||
用法:
|
||||
python manage.py bootstrap_deployment
|
||||
|
||||
行为(幂等,可安全重复执行,适合放进部署脚本/容器启动脚本的 migrate 之后):
|
||||
1. 若环境变量 DJANGO_SUPERUSER_USERNAME / DJANGO_SUPERUSER_EMAIL /
|
||||
DJANGO_SUPERUSER_PASSWORD 均已设置,且该用户名尚不存在,则创建一个超级管理员账号。
|
||||
- 若用户名已存在,跳过创建(不会重置密码,避免每次部署都覆盖已被人工修改过的密码)。
|
||||
- 若三个环境变量未完整设置,跳过此步骤并给出提示(适用于已手动创建过超管、
|
||||
或本地开发环境无需自动建号的场景)。
|
||||
2. 调用 `setup_rbac_groups` 命令初始化/重置 RBAC 权限组(详见该命令的说明)。
|
||||
|
||||
安全说明:
|
||||
- 密码只应通过环境变量/密钥管理服务注入,不会被写入日志。
|
||||
- 本命令不会修改已存在用户的密码或权限,避免误覆盖人工调整过的账号状态。
|
||||
"""
|
||||
import os
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.core.management import call_command
|
||||
from django.core.management.base import BaseCommand
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
help = "部署引导:按需创建首个超级管理员账号 + 初始化 RBAC 权限组,可重复执行。"
|
||||
|
||||
def handle(self, *args, **options):
|
||||
self._bootstrap_superuser()
|
||||
call_command("setup_rbac_groups")
|
||||
|
||||
def _bootstrap_superuser(self):
|
||||
username = os.environ.get("DJANGO_SUPERUSER_USERNAME")
|
||||
email = os.environ.get("DJANGO_SUPERUSER_EMAIL")
|
||||
password = os.environ.get("DJANGO_SUPERUSER_PASSWORD")
|
||||
|
||||
if not (username and email and password):
|
||||
self.stdout.write(
|
||||
self.style.WARNING(
|
||||
"未完整设置 DJANGO_SUPERUSER_USERNAME / DJANGO_SUPERUSER_EMAIL / "
|
||||
"DJANGO_SUPERUSER_PASSWORD,跳过自动创建超级管理员账号。"
|
||||
)
|
||||
)
|
||||
return
|
||||
|
||||
User = get_user_model()
|
||||
if User.objects.filter(username=username).exists():
|
||||
self.stdout.write(
|
||||
self.style.SUCCESS(f"超级管理员账号 '{username}' 已存在,跳过创建。")
|
||||
)
|
||||
return
|
||||
|
||||
User.objects.create_superuser(username=username, email=email, password=password)
|
||||
self.stdout.write(self.style.SUCCESS(f"已创建超级管理员账号 '{username}'。"))
|
||||
@@ -0,0 +1,201 @@
|
||||
"""初始化 RBAC 权限组。
|
||||
|
||||
根据 documents/设计方案分析与完善版.md §2.8 的角色矩阵,使用 Wagtail 原生
|
||||
`Group` + `GroupPagePermission` + `GroupCollectionPermission` + Django
|
||||
`Permission`(Snippet/用户管理),不引入额外的 RBAC 框架。
|
||||
|
||||
角色与 Django/Wagtail 概念的映射:
|
||||
- Super Admin -> Django `is_superuser=True`(不对应任何 Group,超级用户天然拥有全部权限)
|
||||
- Tenant/Site Admin -> Group "站点管理员"
|
||||
- Editor -> Group "编辑"
|
||||
- Reviewer -> Group "审核员"
|
||||
- Author -> Group "作者"
|
||||
- Viewer -> Group "查看者"
|
||||
|
||||
简化与取舍说明:
|
||||
1. 审批发布:设计方案 §2.9 的 Workflow 审核模块尚未实现,因此本命令暂不配置
|
||||
Wagtail Workflow/TaskState。而是通过“编辑者只有 add/change、审核员和站点管理员
|
||||
才有 publish”的权限差异,天然形成“创建草稿 -> 等待发布”的审核闸门。待 Workflow
|
||||
模块落地后,可在现有 Group 基础上叠加 GroupApprovalTask,无需重新设计权限组。
|
||||
2. Author“仅自己”:利用 Wagtail 内置的 OwnershipPermissionPolicy —— 只授予
|
||||
`add_page`(不授予 `change_page`)时,用户可以创建新页面,且只能编辑/删除自己
|
||||
拥有(owner)的页面,无需额外自定义 wagtail_hooks。
|
||||
3. Reviewer 的“页面编辑”矩阵标注为“查看”,但由于 Wagtail 的页面编辑视图需要
|
||||
`change` 权限才能打开并点击“发布”按钮,因此本命令额外授予审核员 `change_page`,
|
||||
属于为了让“审批发布”可用而做的必要放宽,已在此处明确说明。
|
||||
4. Lead(线索)Snippet 涉及 PIPL 个人信息,出于数据保护考虑做了比矩阵更严格的收紧:
|
||||
仅“站点管理员”可管理,“审核员”仅可查看,“编辑/作者”不授予任何 Lead 权限。
|
||||
5. 权限统一挂载在 Wagtail 页面树根节点(depth=1)与根 Collection 上,适用于当前
|
||||
单站点部署;如未来引入多站点/多租户,需要按 Site 对应的根页面/Collection 分别授权。
|
||||
6. 用户管理(新建/编辑/删除/查看 Django 用户)仅授予“站点管理员”,对应矩阵中
|
||||
“用户管理”列;Django Admin 后台访问不通过 Group 授予,只能由拥有
|
||||
`is_staff=True` 的账号(通常是超级用户)访问。
|
||||
|
||||
本命令幂等:每次运行都会将各 Group 的权限重置为矩阵定义的目标状态,可安全重复执行。
|
||||
"""
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.contrib.auth.models import Group, Permission
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
from django.core.management.base import BaseCommand
|
||||
|
||||
from wagtail.models import Collection, GroupCollectionPermission, GroupPagePermission, Page
|
||||
|
||||
from apps.core.models import NavigationMenu, Partner, SiteSettings, TeamMember, Testimonial
|
||||
from apps.forms.models import FormDefinition, Lead
|
||||
|
||||
SITE_ADMIN = "站点管理员"
|
||||
EDITOR = "编辑"
|
||||
REVIEWER = "审核员"
|
||||
AUTHOR = "作者"
|
||||
VIEWER = "查看者"
|
||||
|
||||
ALL_GROUP_NAMES = [SITE_ADMIN, EDITOR, REVIEWER, AUTHOR, VIEWER]
|
||||
|
||||
# 页面权限:值为 GroupPagePermission 的 permission_type 简写(自动拼接为 "<type>_page")
|
||||
PAGE_PERMISSIONS = {
|
||||
SITE_ADMIN: ["add", "change", "publish", "delete", "bulk_delete", "lock", "unlock", "view"],
|
||||
EDITOR: ["add", "change", "view"],
|
||||
REVIEWER: ["change", "publish", "view"],
|
||||
AUTHOR: ["add", "view"],
|
||||
VIEWER: ["view"],
|
||||
}
|
||||
|
||||
# Snippet 权限:{ Model: { group_name: [action, ...] } },未列出的角色即不授予任何权限
|
||||
SNIPPET_PERMISSIONS = {
|
||||
TeamMember: {
|
||||
SITE_ADMIN: ["add", "change", "delete", "view"],
|
||||
EDITOR: ["add", "change", "delete", "view"],
|
||||
REVIEWER: ["view"],
|
||||
},
|
||||
Testimonial: {
|
||||
SITE_ADMIN: ["add", "change", "delete", "view"],
|
||||
EDITOR: ["add", "change", "delete", "view"],
|
||||
REVIEWER: ["view"],
|
||||
},
|
||||
Partner: {
|
||||
SITE_ADMIN: ["add", "change", "delete", "view"],
|
||||
EDITOR: ["add", "change", "delete", "view"],
|
||||
REVIEWER: ["view"],
|
||||
},
|
||||
NavigationMenu: {
|
||||
SITE_ADMIN: ["add", "change", "delete", "view"],
|
||||
EDITOR: ["add", "change", "delete", "view"],
|
||||
REVIEWER: ["view"],
|
||||
},
|
||||
FormDefinition: {
|
||||
SITE_ADMIN: ["add", "change", "delete", "view"],
|
||||
EDITOR: ["add", "change", "delete", "view"],
|
||||
REVIEWER: ["view"],
|
||||
},
|
||||
# Lead 涉及 PIPL 个人信息,收紧至站点管理员可管理、审核员仅查看
|
||||
Lead: {
|
||||
SITE_ADMIN: ["add", "change", "delete", "view"],
|
||||
REVIEWER: ["view"],
|
||||
},
|
||||
# SiteSettings 是站点级单例配置,仅站点管理员可修改
|
||||
SiteSettings: {
|
||||
SITE_ADMIN: ["change", "view"],
|
||||
},
|
||||
}
|
||||
|
||||
# Collection(图片/文档)权限:{ app_label: model } 用于定位对应的 ContentType
|
||||
COLLECTION_CONTENT_TYPES = {
|
||||
"image": "wagtailimages",
|
||||
"document": "wagtaildocs",
|
||||
}
|
||||
COLLECTION_PERMISSIONS = {
|
||||
SITE_ADMIN: ["add", "change", "delete", "choose", "view"],
|
||||
EDITOR: ["add", "change", "choose", "view"],
|
||||
AUTHOR: ["add", "change", "choose", "view"],
|
||||
REVIEWER: ["choose", "view"],
|
||||
}
|
||||
|
||||
# 用户管理权限(Django auth.User 的增删改查)仅授予站点管理员
|
||||
USER_MANAGEMENT_GROUPS = [SITE_ADMIN]
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
help = "根据 RBAC 角色矩阵(设计方案 §2.8)创建/重置 Wagtail Group 权限,可重复执行。"
|
||||
|
||||
def handle(self, *args, **options):
|
||||
groups = {name: Group.objects.get_or_create(name=name)[0] for name in ALL_GROUP_NAMES}
|
||||
|
||||
self._grant_admin_access(groups)
|
||||
self._setup_page_permissions(groups)
|
||||
self._setup_snippet_permissions(groups)
|
||||
self._setup_collection_permissions(groups)
|
||||
self._setup_user_management_permissions(groups)
|
||||
|
||||
self.stdout.write(
|
||||
self.style.SUCCESS(f"RBAC 权限组已初始化/更新:{', '.join(ALL_GROUP_NAMES)}")
|
||||
)
|
||||
|
||||
def _grant_admin_access(self, groups):
|
||||
"""所有角色都需要能登录 Wagtail 后台。"""
|
||||
access_admin = Permission.objects.get(
|
||||
content_type__app_label="wagtailadmin", codename="access_admin"
|
||||
)
|
||||
for group in groups.values():
|
||||
group.permissions.add(access_admin)
|
||||
|
||||
def _setup_page_permissions(self, groups):
|
||||
root_page = Page.objects.get(depth=1)
|
||||
for name in ALL_GROUP_NAMES:
|
||||
group = groups[name]
|
||||
GroupPagePermission.objects.filter(group=group, page=root_page).delete()
|
||||
for action in PAGE_PERMISSIONS.get(name, []):
|
||||
GroupPagePermission.objects.create(
|
||||
group=group, page=root_page, permission_type=action
|
||||
)
|
||||
|
||||
def _setup_snippet_permissions(self, groups):
|
||||
for model, group_actions in SNIPPET_PERMISSIONS.items():
|
||||
content_type = ContentType.objects.get_for_model(model)
|
||||
model_name = model._meta.model_name
|
||||
all_perms_for_model = Permission.objects.filter(content_type=content_type)
|
||||
for name in ALL_GROUP_NAMES:
|
||||
group = groups[name]
|
||||
group.permissions.remove(*all_perms_for_model)
|
||||
actions = group_actions.get(name, [])
|
||||
if not actions:
|
||||
continue
|
||||
codenames = [f"{action}_{model_name}" for action in actions]
|
||||
perms = Permission.objects.filter(
|
||||
content_type=content_type, codename__in=codenames
|
||||
)
|
||||
group.permissions.add(*perms)
|
||||
|
||||
def _setup_collection_permissions(self, groups):
|
||||
root_collection = Collection.get_first_root_node()
|
||||
content_types = {
|
||||
key: ContentType.objects.get(app_label=app_label, model=key)
|
||||
for key, app_label in COLLECTION_CONTENT_TYPES.items()
|
||||
}
|
||||
all_perms = Permission.objects.filter(content_type__in=content_types.values())
|
||||
|
||||
for name in ALL_GROUP_NAMES:
|
||||
group = groups[name]
|
||||
GroupCollectionPermission.objects.filter(
|
||||
group=group, collection=root_collection, permission__in=all_perms
|
||||
).delete()
|
||||
|
||||
actions = COLLECTION_PERMISSIONS.get(name, [])
|
||||
if not actions:
|
||||
continue
|
||||
for key, content_type in content_types.items():
|
||||
codenames = [f"{action}_{key}" for action in actions]
|
||||
perms = Permission.objects.filter(
|
||||
content_type=content_type, codename__in=codenames
|
||||
)
|
||||
for perm in perms:
|
||||
GroupCollectionPermission.objects.get_or_create(
|
||||
group=group, collection=root_collection, permission=perm
|
||||
)
|
||||
|
||||
def _setup_user_management_permissions(self, groups):
|
||||
user_content_type = ContentType.objects.get_for_model(get_user_model())
|
||||
all_user_perms = Permission.objects.filter(content_type=user_content_type)
|
||||
for name in ALL_GROUP_NAMES:
|
||||
groups[name].permissions.remove(*all_user_perms)
|
||||
for name in USER_MANAGEMENT_GROUPS:
|
||||
groups[name].permissions.add(*all_user_perms)
|
||||
@@ -0,0 +1,110 @@
|
||||
# Generated by Django 6.0.5 on 2026-08-07 03:14
|
||||
|
||||
import django.db.models.deletion
|
||||
import modelcluster.fields
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('core', '0001_initial'),
|
||||
('wagtailcore', '0097_baselogentry_uuid_action_timestamp_indexes'),
|
||||
('wagtailimages', '0027_image_description'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='NavigationMenu',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('name', models.CharField(help_text='前端按此标识获取菜单,如 main(主导航)、footer(页脚导航)', max_length=50, unique=True, verbose_name='菜单标识')),
|
||||
],
|
||||
options={
|
||||
'verbose_name': '导航菜单',
|
||||
'verbose_name_plural': '导航菜单',
|
||||
},
|
||||
),
|
||||
migrations.CreateModel(
|
||||
name='NavigationMenuItem',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('sort_order', models.IntegerField(blank=True, editable=False, null=True)),
|
||||
('label', models.CharField(max_length=50, verbose_name='文案')),
|
||||
('link_url', models.CharField(blank=True, help_text='留空则使用内部页面链接', max_length=255, verbose_name='外部链接')),
|
||||
('open_in_new_tab', models.BooleanField(default=False, verbose_name='新标签页打开')),
|
||||
('link_page', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='+', to='wagtailcore.page', verbose_name='内部页面')),
|
||||
('menu', modelcluster.fields.ParentalKey(on_delete=django.db.models.deletion.CASCADE, related_name='items', to='core.navigationmenu')),
|
||||
],
|
||||
options={
|
||||
'ordering': ['sort_order'],
|
||||
'abstract': False,
|
||||
},
|
||||
),
|
||||
migrations.CreateModel(
|
||||
name='Partner',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('name', models.CharField(max_length=100, verbose_name='名称')),
|
||||
('website_url', models.URLField(blank=True, verbose_name='官网链接')),
|
||||
('order', models.IntegerField(default=0, verbose_name='排序')),
|
||||
('logo', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='+', to='wagtailimages.image', verbose_name='Logo')),
|
||||
],
|
||||
options={
|
||||
'verbose_name': '合作伙伴',
|
||||
'verbose_name_plural': '合作伙伴',
|
||||
'ordering': ['order', 'id'],
|
||||
},
|
||||
),
|
||||
migrations.CreateModel(
|
||||
name='SiteSettings',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('company_name', models.CharField(blank=True, max_length=100, verbose_name='公司全称')),
|
||||
('contact_phone', models.CharField(blank=True, max_length=30, verbose_name='联系电话')),
|
||||
('contact_email', models.EmailField(blank=True, max_length=254, verbose_name='联系邮箱')),
|
||||
('address', models.CharField(blank=True, max_length=255, verbose_name='公司地址')),
|
||||
('icp_number', models.CharField(blank=True, max_length=50, verbose_name='ICP备案号')),
|
||||
('icp_url', models.URLField(default='https://beian.miit.gov.cn/', verbose_name='备案查询链接')),
|
||||
('wechat_account', models.CharField(blank=True, max_length=100, verbose_name='微信公众号名称')),
|
||||
('weibo_url', models.URLField(blank=True, verbose_name='微博主页')),
|
||||
('site', models.OneToOneField(editable=False, on_delete=django.db.models.deletion.CASCADE, to='wagtailcore.site')),
|
||||
('wechat_qrcode', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='+', to='wagtailimages.image', verbose_name='微信公众号二维码')),
|
||||
],
|
||||
options={
|
||||
'verbose_name': '站点设置',
|
||||
},
|
||||
),
|
||||
migrations.CreateModel(
|
||||
name='TeamMember',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('name', models.CharField(max_length=50, verbose_name='姓名')),
|
||||
('role', models.CharField(blank=True, max_length=100, verbose_name='职位')),
|
||||
('bio', models.TextField(blank=True, verbose_name='简介')),
|
||||
('order', models.IntegerField(default=0, verbose_name='排序')),
|
||||
('photo', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='+', to='wagtailimages.image', verbose_name='照片')),
|
||||
],
|
||||
options={
|
||||
'verbose_name': '团队成员',
|
||||
'verbose_name_plural': '团队成员',
|
||||
'ordering': ['order', 'id'],
|
||||
},
|
||||
),
|
||||
migrations.CreateModel(
|
||||
name='Testimonial',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('quote', models.TextField(verbose_name='推荐语')),
|
||||
('author_name', models.CharField(max_length=50, verbose_name='客户姓名')),
|
||||
('author_title', models.CharField(blank=True, max_length=100, verbose_name='客户职位/公司')),
|
||||
('order', models.IntegerField(default=0, verbose_name='排序')),
|
||||
('author_photo', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='+', to='wagtailimages.image', verbose_name='客户头像')),
|
||||
],
|
||||
options={
|
||||
'verbose_name': '客户证言',
|
||||
'verbose_name_plural': '客户证言',
|
||||
'ordering': ['order', 'id'],
|
||||
},
|
||||
),
|
||||
]
|
||||
+235
-3
@@ -1,14 +1,23 @@
|
||||
"""
|
||||
公共抽象模型。所有业务页面应继承 SEOablePage 而非直接继承 wagtail.models.Page,
|
||||
以统一获得 SEO 字段(详见 documents/设计方案分析与完善版.md §2.5)。
|
||||
|
||||
本文件同时存放跨页面复用的全局 Snippet:TeamMember/Testimonial/Partner/
|
||||
NavigationMenu/SiteSettings(详见 documents/设计方案分析与完善版.md §2.5、§2.16)。
|
||||
"""
|
||||
from django.db import models
|
||||
from wagtail.admin.panels import FieldPanel
|
||||
from modelcluster.fields import ParentalKey
|
||||
from modelcluster.models import ClusterableModel
|
||||
from wagtail.admin.panels import FieldPanel, InlinePanel, MultiFieldPanel
|
||||
from wagtail.contrib.settings.models import BaseSiteSetting
|
||||
from wagtail.contrib.settings.registry import register_setting
|
||||
from wagtail.fields import RichTextField
|
||||
from wagtail.models import Page
|
||||
from wagtail.models import Orderable, Page
|
||||
from wagtail.snippets.models import register_snippet
|
||||
from wagtail_headless_preview.models import HeadlessPreviewMixin
|
||||
|
||||
|
||||
class SEOablePage(Page):
|
||||
class SEOablePage(HeadlessPreviewMixin, Page):
|
||||
seo_title_override = models.CharField(
|
||||
max_length=70, blank=True, verbose_name="SEO 标题",
|
||||
help_text="留空则使用页面标题",
|
||||
@@ -53,3 +62,226 @@ class SimpleContentPage(SEOablePage):
|
||||
|
||||
class Meta:
|
||||
verbose_name = "通用内容页"
|
||||
|
||||
|
||||
@register_snippet
|
||||
class TeamMember(models.Model):
|
||||
"""团队成员,用于关于我们/团队介绍等场景复用。"""
|
||||
|
||||
name = models.CharField(max_length=50, verbose_name="姓名")
|
||||
role = models.CharField(max_length=100, blank=True, verbose_name="职位")
|
||||
photo = models.ForeignKey(
|
||||
"wagtailimages.Image",
|
||||
null=True,
|
||||
blank=True,
|
||||
on_delete=models.SET_NULL,
|
||||
related_name="+",
|
||||
verbose_name="照片",
|
||||
)
|
||||
bio = models.TextField(blank=True, verbose_name="简介")
|
||||
order = models.IntegerField(default=0, verbose_name="排序")
|
||||
|
||||
panels = [
|
||||
FieldPanel("name"),
|
||||
FieldPanel("role"),
|
||||
FieldPanel("photo"),
|
||||
FieldPanel("bio"),
|
||||
FieldPanel("order"),
|
||||
]
|
||||
|
||||
class Meta:
|
||||
ordering = ["order", "id"]
|
||||
verbose_name = "团队成员"
|
||||
verbose_name_plural = "团队成员"
|
||||
|
||||
def __str__(self):
|
||||
return self.name
|
||||
|
||||
|
||||
@register_snippet
|
||||
class Testimonial(models.Model):
|
||||
"""客户证言/推荐语,用于首页、案例页等场景复用。"""
|
||||
|
||||
quote = models.TextField(verbose_name="推荐语")
|
||||
author_name = models.CharField(max_length=50, verbose_name="客户姓名")
|
||||
author_title = models.CharField(
|
||||
max_length=100, blank=True, verbose_name="客户职位/公司"
|
||||
)
|
||||
author_photo = models.ForeignKey(
|
||||
"wagtailimages.Image",
|
||||
null=True,
|
||||
blank=True,
|
||||
on_delete=models.SET_NULL,
|
||||
related_name="+",
|
||||
verbose_name="客户头像",
|
||||
)
|
||||
order = models.IntegerField(default=0, verbose_name="排序")
|
||||
|
||||
panels = [
|
||||
FieldPanel("quote"),
|
||||
FieldPanel("author_name"),
|
||||
FieldPanel("author_title"),
|
||||
FieldPanel("author_photo"),
|
||||
FieldPanel("order"),
|
||||
]
|
||||
|
||||
class Meta:
|
||||
ordering = ["order", "id"]
|
||||
verbose_name = "客户证言"
|
||||
verbose_name_plural = "客户证言"
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.author_name} - {self.quote[:20]}"
|
||||
|
||||
|
||||
@register_snippet
|
||||
class Partner(models.Model):
|
||||
"""合作伙伴/客户 Logo,用于 Logo 墙展示。"""
|
||||
|
||||
name = models.CharField(max_length=100, verbose_name="名称")
|
||||
logo = models.ForeignKey(
|
||||
"wagtailimages.Image",
|
||||
null=True,
|
||||
blank=True,
|
||||
on_delete=models.SET_NULL,
|
||||
related_name="+",
|
||||
verbose_name="Logo",
|
||||
)
|
||||
website_url = models.URLField(blank=True, verbose_name="官网链接")
|
||||
order = models.IntegerField(default=0, verbose_name="排序")
|
||||
|
||||
panels = [
|
||||
FieldPanel("name"),
|
||||
FieldPanel("logo"),
|
||||
FieldPanel("website_url"),
|
||||
FieldPanel("order"),
|
||||
]
|
||||
|
||||
class Meta:
|
||||
ordering = ["order", "id"]
|
||||
verbose_name = "合作伙伴"
|
||||
verbose_name_plural = "合作伙伴"
|
||||
|
||||
def __str__(self):
|
||||
return self.name
|
||||
|
||||
|
||||
@register_snippet
|
||||
class NavigationMenu(ClusterableModel):
|
||||
"""导航菜单容器,通过 name 区分主导航/页脚导航等多套菜单。"""
|
||||
|
||||
name = models.CharField(
|
||||
max_length=50,
|
||||
unique=True,
|
||||
verbose_name="菜单标识",
|
||||
help_text="前端按此标识获取菜单,如 main(主导航)、footer(页脚导航)",
|
||||
)
|
||||
|
||||
panels = [
|
||||
FieldPanel("name"),
|
||||
InlinePanel("items", label="菜单项"),
|
||||
]
|
||||
|
||||
class Meta:
|
||||
verbose_name = "导航菜单"
|
||||
verbose_name_plural = "导航菜单"
|
||||
|
||||
def __str__(self):
|
||||
return self.name
|
||||
|
||||
|
||||
class NavigationMenuItem(Orderable):
|
||||
"""导航菜单项:可链接到内部页面,也可填写外部链接。"""
|
||||
|
||||
menu = ParentalKey(
|
||||
NavigationMenu, on_delete=models.CASCADE, related_name="items"
|
||||
)
|
||||
label = models.CharField(max_length=50, verbose_name="文案")
|
||||
link_page = models.ForeignKey(
|
||||
Page,
|
||||
null=True,
|
||||
blank=True,
|
||||
on_delete=models.SET_NULL,
|
||||
related_name="+",
|
||||
verbose_name="内部页面",
|
||||
)
|
||||
link_url = models.CharField(
|
||||
max_length=255,
|
||||
blank=True,
|
||||
verbose_name="外部链接",
|
||||
help_text="留空则使用内部页面链接",
|
||||
)
|
||||
open_in_new_tab = models.BooleanField(default=False, verbose_name="新标签页打开")
|
||||
|
||||
panels = [
|
||||
FieldPanel("label"),
|
||||
FieldPanel("link_page"),
|
||||
FieldPanel("link_url"),
|
||||
FieldPanel("open_in_new_tab"),
|
||||
]
|
||||
|
||||
@property
|
||||
def url(self):
|
||||
if self.link_url:
|
||||
return self.link_url
|
||||
if self.link_page_id:
|
||||
return self.link_page.url
|
||||
return "#"
|
||||
|
||||
def __str__(self):
|
||||
return self.label
|
||||
|
||||
|
||||
@register_setting(icon="cogs")
|
||||
class SiteSettings(BaseSiteSetting):
|
||||
"""站点全局配置:公司信息/联系方式/备案号/社交账号,供 Footer 等全局位置消费。"""
|
||||
|
||||
company_name = models.CharField(max_length=100, blank=True, verbose_name="公司全称")
|
||||
contact_phone = models.CharField(max_length=30, blank=True, verbose_name="联系电话")
|
||||
contact_email = models.EmailField(blank=True, verbose_name="联系邮箱")
|
||||
address = models.CharField(max_length=255, blank=True, verbose_name="公司地址")
|
||||
|
||||
icp_number = models.CharField(max_length=50, blank=True, verbose_name="ICP备案号")
|
||||
icp_url = models.URLField(
|
||||
default="https://beian.miit.gov.cn/", verbose_name="备案查询链接"
|
||||
)
|
||||
|
||||
wechat_qrcode = models.ForeignKey(
|
||||
"wagtailimages.Image",
|
||||
null=True,
|
||||
blank=True,
|
||||
on_delete=models.SET_NULL,
|
||||
related_name="+",
|
||||
verbose_name="微信公众号二维码",
|
||||
)
|
||||
wechat_account = models.CharField(
|
||||
max_length=100, blank=True, verbose_name="微信公众号名称"
|
||||
)
|
||||
weibo_url = models.URLField(blank=True, verbose_name="微博主页")
|
||||
|
||||
panels = [
|
||||
MultiFieldPanel(
|
||||
[
|
||||
FieldPanel("company_name"),
|
||||
FieldPanel("contact_phone"),
|
||||
FieldPanel("contact_email"),
|
||||
FieldPanel("address"),
|
||||
],
|
||||
heading="联系方式",
|
||||
),
|
||||
MultiFieldPanel(
|
||||
[FieldPanel("icp_number"), FieldPanel("icp_url")],
|
||||
heading="备案信息",
|
||||
),
|
||||
MultiFieldPanel(
|
||||
[
|
||||
FieldPanel("wechat_qrcode"),
|
||||
FieldPanel("wechat_account"),
|
||||
FieldPanel("weibo_url"),
|
||||
],
|
||||
heading="社交账号",
|
||||
),
|
||||
]
|
||||
|
||||
class Meta:
|
||||
verbose_name = "站点设置"
|
||||
|
||||
@@ -0,0 +1,74 @@
|
||||
"""
|
||||
apps.core 全局 Snippet 的只读序列化器,供自定义业务接口
|
||||
(/api/v1/custom/core/...)使用。图片字段统一序列化为
|
||||
{"url": ..., "title": ...},与前端 frontend/types/wagtail.ts
|
||||
中既有的 background_image/logos 字段结构保持一致。
|
||||
"""
|
||||
from rest_framework import serializers
|
||||
|
||||
|
||||
def serialize_image(image):
|
||||
if not image:
|
||||
return None
|
||||
return {"url": image.file.url, "title": image.title}
|
||||
|
||||
|
||||
class TeamMemberSerializer(serializers.Serializer):
|
||||
id = serializers.IntegerField()
|
||||
name = serializers.CharField()
|
||||
role = serializers.CharField()
|
||||
bio = serializers.CharField()
|
||||
photo = serializers.SerializerMethodField()
|
||||
|
||||
def get_photo(self, obj):
|
||||
return serialize_image(obj.photo)
|
||||
|
||||
|
||||
class TestimonialSerializer(serializers.Serializer):
|
||||
id = serializers.IntegerField()
|
||||
quote = serializers.CharField()
|
||||
author_name = serializers.CharField()
|
||||
author_title = serializers.CharField()
|
||||
author_photo = serializers.SerializerMethodField()
|
||||
|
||||
def get_author_photo(self, obj):
|
||||
return serialize_image(obj.author_photo)
|
||||
|
||||
|
||||
class PartnerSerializer(serializers.Serializer):
|
||||
id = serializers.IntegerField()
|
||||
name = serializers.CharField()
|
||||
website_url = serializers.CharField()
|
||||
logo = serializers.SerializerMethodField()
|
||||
|
||||
def get_logo(self, obj):
|
||||
return serialize_image(obj.logo)
|
||||
|
||||
|
||||
class NavigationMenuItemSerializer(serializers.Serializer):
|
||||
label = serializers.CharField()
|
||||
url = serializers.CharField()
|
||||
open_in_new_tab = serializers.BooleanField()
|
||||
|
||||
|
||||
class NavigationMenuSerializer(serializers.Serializer):
|
||||
name = serializers.CharField()
|
||||
items = serializers.SerializerMethodField()
|
||||
|
||||
def get_items(self, obj):
|
||||
return NavigationMenuItemSerializer(obj.items.all(), many=True).data
|
||||
|
||||
|
||||
class SiteSettingsSerializer(serializers.Serializer):
|
||||
company_name = serializers.CharField()
|
||||
contact_phone = serializers.CharField()
|
||||
contact_email = serializers.CharField()
|
||||
address = serializers.CharField()
|
||||
icp_number = serializers.CharField()
|
||||
icp_url = serializers.CharField()
|
||||
wechat_account = serializers.CharField()
|
||||
weibo_url = serializers.CharField()
|
||||
wechat_qrcode = serializers.SerializerMethodField()
|
||||
|
||||
def get_wechat_qrcode(self, obj):
|
||||
return serialize_image(obj.wechat_qrcode)
|
||||
+459
-2
@@ -1,8 +1,18 @@
|
||||
"""apps.core 基础单元测试:SEOablePage 抽象基类字段与公共 Block。"""
|
||||
import pytest
|
||||
from django.contrib.auth import get_user_model
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from apps.core.blocks import FormBlock
|
||||
from apps.core.models import SimpleContentPage
|
||||
from apps.core.blocks import FormBlock, TeamMemberChooserBlock
|
||||
from apps.core.models import (
|
||||
NavigationMenu,
|
||||
NavigationMenuItem,
|
||||
Partner,
|
||||
SimpleContentPage,
|
||||
SiteSettings,
|
||||
TeamMember,
|
||||
Testimonial,
|
||||
)
|
||||
from apps.forms.models import FormDefinition, FormDefinitionField
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
@@ -75,3 +85,450 @@ def test_simple_content_page_can_be_created_under_home(home_page):
|
||||
page.refresh_from_db()
|
||||
assert SimpleContentPage.objects.live().descendant_of(home_page).count() == 1
|
||||
assert "个人信息保护" in page.body
|
||||
|
||||
|
||||
def test_team_member_default_ordering():
|
||||
"""团队成员应按 order 字段升序排列。"""
|
||||
TeamMember.objects.create(name="李四", order=2)
|
||||
TeamMember.objects.create(name="张三", order=1)
|
||||
names = list(TeamMember.objects.values_list("name", flat=True))
|
||||
assert names == ["张三", "李四"]
|
||||
|
||||
|
||||
def test_testimonial_str_includes_author_and_quote():
|
||||
testimonial = Testimonial.objects.create(
|
||||
quote="非常专业的服务,效果超出预期。", author_name="王经理"
|
||||
)
|
||||
assert "王经理" in str(testimonial)
|
||||
|
||||
|
||||
def test_partner_str_returns_name():
|
||||
partner = Partner.objects.create(name="示例合作伙伴")
|
||||
assert str(partner) == "示例合作伙伴"
|
||||
|
||||
|
||||
def test_navigation_menu_item_url_prefers_external_link_over_page(home_page):
|
||||
menu = NavigationMenu.objects.create(name="main")
|
||||
item = NavigationMenuItem.objects.create(
|
||||
menu=menu, label="首页", link_page=home_page, link_url="https://example.com"
|
||||
)
|
||||
assert item.url == "https://example.com"
|
||||
|
||||
|
||||
def test_navigation_menu_item_url_falls_back_to_page_url(home_page):
|
||||
menu = NavigationMenu.objects.create(name="main")
|
||||
item = NavigationMenuItem.objects.create(menu=menu, label="首页", link_page=home_page)
|
||||
assert item.url == home_page.url
|
||||
|
||||
|
||||
def test_navigation_menu_item_url_defaults_to_hash_when_empty():
|
||||
menu = NavigationMenu.objects.create(name="main")
|
||||
item = NavigationMenuItem.objects.create(menu=menu, label="占位")
|
||||
assert item.url == "#"
|
||||
|
||||
|
||||
def test_site_settings_for_site_creates_singleton_with_defaults():
|
||||
from wagtail.models import Site
|
||||
|
||||
site = Site.objects.get(is_default_site=True)
|
||||
settings_obj = SiteSettings.for_site(site)
|
||||
assert settings_obj.icp_url == "https://beian.miit.gov.cn/"
|
||||
# 再次获取应复用同一条记录而非重复创建
|
||||
assert SiteSettings.for_site(site).pk == settings_obj.pk
|
||||
|
||||
|
||||
def test_team_api_returns_ordered_members():
|
||||
TeamMember.objects.create(name="李四", role="工程师", order=2)
|
||||
TeamMember.objects.create(name="张三", role="总经理", order=1)
|
||||
|
||||
client = APIClient()
|
||||
response = client.get("/api/v1/custom/core/team/")
|
||||
|
||||
assert response.status_code == 200
|
||||
items = response.data["items"]
|
||||
assert [item["name"] for item in items] == ["张三", "李四"]
|
||||
assert items[0]["role"] == "总经理"
|
||||
|
||||
|
||||
def test_testimonial_api_returns_list():
|
||||
Testimonial.objects.create(quote="效果很好", author_name="客户A")
|
||||
|
||||
client = APIClient()
|
||||
response = client.get("/api/v1/custom/core/testimonials/")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.data["items"][0]["author_name"] == "客户A"
|
||||
|
||||
|
||||
def test_team_member_chooser_block_api_representation_expands_fields():
|
||||
"""TeamMemberChooserBlock.get_api_representation 应展开 TeamMember 的完整字段,
|
||||
而不是仅返回 SnippetChooserBlock 默认的主键。"""
|
||||
member = TeamMember.objects.create(name="王五", role="产品经理", bio="十年行业经验")
|
||||
|
||||
block = TeamMemberChooserBlock()
|
||||
result = block.get_api_representation(member)
|
||||
|
||||
assert result["id"] == member.pk
|
||||
assert result["name"] == "王五"
|
||||
assert result["role"] == "产品经理"
|
||||
assert result["bio"] == "十年行业经验"
|
||||
assert result["photo"] is None
|
||||
|
||||
|
||||
def test_team_member_chooser_block_api_representation_handles_none():
|
||||
block = TeamMemberChooserBlock()
|
||||
assert block.get_api_representation(None) is None
|
||||
|
||||
|
||||
def test_partner_api_returns_list():
|
||||
Partner.objects.create(name="伙伴A", website_url="https://partner-a.example.com")
|
||||
|
||||
client = APIClient()
|
||||
response = client.get("/api/v1/custom/core/partners/")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.data["items"][0]["name"] == "伙伴A"
|
||||
|
||||
|
||||
def test_navigation_menu_api_returns_items_by_name(home_page):
|
||||
menu = NavigationMenu.objects.create(name="main")
|
||||
NavigationMenuItem.objects.create(menu=menu, label="首页", link_page=home_page, sort_order=0)
|
||||
NavigationMenuItem.objects.create(menu=menu, label="外部链接", link_url="https://example.com", sort_order=1)
|
||||
|
||||
client = APIClient()
|
||||
response = client.get("/api/v1/custom/core/navigation/?name=main")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.data["name"] == "main"
|
||||
assert len(response.data["items"]) == 2
|
||||
assert response.data["items"][1]["url"] == "https://example.com"
|
||||
|
||||
|
||||
def test_navigation_menu_api_404_when_missing():
|
||||
client = APIClient()
|
||||
response = client.get("/api/v1/custom/core/navigation/?name=does-not-exist")
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_site_settings_api_returns_defaults():
|
||||
client = APIClient()
|
||||
response = client.get("/api/v1/custom/core/site-settings/")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.data["icp_url"] == "https://beian.miit.gov.cn/"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# RBAC 权限组(setup_rbac_groups 管理命令)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_setup_rbac_groups_creates_all_expected_groups():
|
||||
from django.contrib.auth.models import Group
|
||||
from django.core.management import call_command
|
||||
|
||||
call_command("setup_rbac_groups")
|
||||
|
||||
expected_names = {"站点管理员", "编辑", "审核员", "作者", "查看者"}
|
||||
assert expected_names <= set(Group.objects.values_list("name", flat=True))
|
||||
|
||||
|
||||
def test_setup_rbac_groups_grants_access_admin_to_all_groups():
|
||||
from django.contrib.auth.models import Group
|
||||
from django.core.management import call_command
|
||||
|
||||
call_command("setup_rbac_groups")
|
||||
|
||||
for name in ["站点管理员", "编辑", "审核员", "作者", "查看者"]:
|
||||
group = Group.objects.get(name=name)
|
||||
assert group.permissions.filter(codename="access_admin").exists()
|
||||
|
||||
|
||||
def test_setup_rbac_groups_page_permissions_match_role_matrix():
|
||||
from django.contrib.auth.models import Group
|
||||
from django.core.management import call_command
|
||||
from wagtail.models import GroupPagePermission, Page
|
||||
|
||||
call_command("setup_rbac_groups")
|
||||
|
||||
root_page = Page.objects.get(depth=1)
|
||||
|
||||
def page_codenames(group_name):
|
||||
group = Group.objects.get(name=group_name)
|
||||
return set(
|
||||
GroupPagePermission.objects.filter(group=group, page=root_page).values_list(
|
||||
"permission__codename", flat=True
|
||||
)
|
||||
)
|
||||
|
||||
# 站点管理员拥有全部页面权限
|
||||
assert page_codenames("站点管理员") == {
|
||||
"add_page", "change_page", "publish_page", "delete_page",
|
||||
"bulk_delete_page", "lock_page", "unlock_page", "view_page",
|
||||
}
|
||||
# 编辑可以创建/编辑但不能发布
|
||||
editor_perms = page_codenames("编辑")
|
||||
assert editor_perms == {"add_page", "change_page", "view_page"}
|
||||
# 审核员可以编辑并发布(用于审批发布场景)
|
||||
assert page_codenames("审核员") == {"change_page", "publish_page", "view_page"}
|
||||
# 作者只有 add + view:只能创建页面,仅能编辑自己拥有的页面(Wagtail 内置 ownership 策略)
|
||||
assert page_codenames("作者") == {"add_page", "view_page"}
|
||||
assert "change_page" not in page_codenames("作者")
|
||||
# 查看者仅能查看
|
||||
assert page_codenames("查看者") == {"view_page"}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 健康检查端点(apps/core/views.py,供 K8s liveness/readiness 探针使用)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_healthz_returns_ok(client):
|
||||
response = client.get("/healthz")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"status": "ok"}
|
||||
|
||||
|
||||
def test_readyz_returns_ok_when_database_available(client):
|
||||
response = client.get("/readyz")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"status": "ok"}
|
||||
|
||||
|
||||
def test_readyz_returns_503_when_database_unavailable(client, monkeypatch):
|
||||
from django.db import connection
|
||||
|
||||
def broken_cursor(*args, **kwargs):
|
||||
raise Exception("simulated db outage")
|
||||
|
||||
monkeypatch.setattr(connection, "cursor", broken_cursor)
|
||||
|
||||
response = client.get("/readyz")
|
||||
|
||||
assert response.status_code == 503
|
||||
assert response.json()["status"] == "error"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 发布后前端 revalidate webhook(apps/core/signals.py)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_notify_frontend_revalidate_skips_when_url_not_configured(
|
||||
settings, monkeypatch, home_page
|
||||
):
|
||||
from wagtail.signals import page_published
|
||||
|
||||
settings.FRONTEND_REVALIDATE_URL = ""
|
||||
calls = []
|
||||
monkeypatch.setattr(
|
||||
"apps.core.signals.requests.post",
|
||||
lambda *args, **kwargs: calls.append((args, kwargs)),
|
||||
)
|
||||
|
||||
page_published.send(sender=type(home_page), instance=home_page)
|
||||
|
||||
assert calls == []
|
||||
|
||||
|
||||
def test_notify_frontend_revalidate_posts_webhook_when_configured(
|
||||
settings, monkeypatch, home_page
|
||||
):
|
||||
from wagtail.signals import page_published
|
||||
|
||||
settings.FRONTEND_REVALIDATE_URL = "https://frontend.example.com/api/revalidate"
|
||||
settings.REVALIDATE_SECRET = "test-secret"
|
||||
calls = []
|
||||
|
||||
def fake_post(url, json=None, headers=None, timeout=None):
|
||||
calls.append({"url": url, "json": json, "headers": headers})
|
||||
|
||||
class FakeResponse:
|
||||
status_code = 200
|
||||
|
||||
return FakeResponse()
|
||||
|
||||
monkeypatch.setattr("apps.core.signals.requests.post", fake_post)
|
||||
|
||||
page_published.send(sender=type(home_page), instance=home_page)
|
||||
|
||||
assert len(calls) == 1
|
||||
assert calls[0]["url"] == "https://frontend.example.com/api/revalidate"
|
||||
assert calls[0]["headers"] == {"Authorization": "Bearer test-secret"}
|
||||
assert f"page:{home_page.id}" in calls[0]["json"]["tags"]
|
||||
|
||||
|
||||
def test_notify_frontend_revalidate_swallows_request_exception(
|
||||
settings, monkeypatch, home_page
|
||||
):
|
||||
import requests as requests_module
|
||||
from wagtail.signals import page_published
|
||||
|
||||
settings.FRONTEND_REVALIDATE_URL = "https://frontend.example.com/api/revalidate"
|
||||
|
||||
def raise_request_exception(*args, **kwargs):
|
||||
raise requests_module.RequestException("network error")
|
||||
|
||||
monkeypatch.setattr("apps.core.signals.requests.post", raise_request_exception)
|
||||
|
||||
# 不应向外抛出异常——webhook 失败不能影响页面发布流程。
|
||||
page_published.send(sender=type(home_page), instance=home_page)
|
||||
|
||||
|
||||
def test_setup_rbac_groups_author_can_only_edit_own_pages(home_page):
|
||||
"""验证“Author 仅能编辑自己的页面”这一矩阵要求,由 Wagtail 内置的
|
||||
OwnershipPermissionPolicy(仅 add 权限)自然实现,无需自定义 hook。"""
|
||||
from django.contrib.auth.models import Group
|
||||
from django.core.management import call_command
|
||||
|
||||
call_command("setup_rbac_groups")
|
||||
|
||||
User = get_user_model()
|
||||
author = User.objects.create_user(username="author-user", password="pass1234")
|
||||
other_author = User.objects.create_user(username="other-author", password="pass1234")
|
||||
author.groups.add(Group.objects.get(name="作者"))
|
||||
other_author.groups.add(Group.objects.get(name="作者"))
|
||||
|
||||
own_page = SimpleContentPage(title="自己的页面", slug="own-page", owner=author)
|
||||
home_page.add_child(instance=own_page)
|
||||
|
||||
others_page = SimpleContentPage(title="他人的页面", slug="others-page", owner=other_author)
|
||||
home_page.add_child(instance=others_page)
|
||||
|
||||
assert own_page.permissions_for_user(author).can_edit() is True
|
||||
assert others_page.permissions_for_user(author).can_edit() is False
|
||||
|
||||
|
||||
def test_setup_rbac_groups_snippet_permissions_restrict_lead_to_admin_and_reviewer():
|
||||
from django.contrib.auth.models import Group
|
||||
from django.core.management import call_command
|
||||
|
||||
call_command("setup_rbac_groups")
|
||||
|
||||
def lead_codenames(group_name):
|
||||
group = Group.objects.get(name=group_name)
|
||||
return set(
|
||||
group.permissions.filter(content_type__app_label="forms", codename__endswith="lead")
|
||||
.values_list("codename", flat=True)
|
||||
)
|
||||
|
||||
assert lead_codenames("站点管理员") == {"add_lead", "change_lead", "delete_lead", "view_lead"}
|
||||
assert lead_codenames("审核员") == {"view_lead"}
|
||||
assert lead_codenames("编辑") == set()
|
||||
assert lead_codenames("作者") == set()
|
||||
assert lead_codenames("查看者") == set()
|
||||
|
||||
|
||||
def test_setup_rbac_groups_only_site_admin_has_user_management_permissions():
|
||||
from django.contrib.auth.models import Group
|
||||
from django.core.management import call_command
|
||||
|
||||
call_command("setup_rbac_groups")
|
||||
|
||||
User = get_user_model()
|
||||
user_ct_app_label = User._meta.app_label
|
||||
|
||||
for name in ["编辑", "审核员", "作者", "查看者"]:
|
||||
group = Group.objects.get(name=name)
|
||||
assert not group.permissions.filter(
|
||||
content_type__app_label=user_ct_app_label, content_type__model="user"
|
||||
).exists()
|
||||
|
||||
site_admin = Group.objects.get(name="站点管理员")
|
||||
site_admin_codenames = set(
|
||||
site_admin.permissions.filter(
|
||||
content_type__app_label=user_ct_app_label, content_type__model="user"
|
||||
).values_list("codename", flat=True)
|
||||
)
|
||||
assert site_admin_codenames == {"add_user", "change_user", "delete_user", "view_user"}
|
||||
|
||||
|
||||
def test_setup_rbac_groups_is_idempotent():
|
||||
from django.contrib.auth.models import Group
|
||||
from django.core.management import call_command
|
||||
from wagtail.models import GroupPagePermission, Page
|
||||
|
||||
call_command("setup_rbac_groups")
|
||||
call_command("setup_rbac_groups")
|
||||
|
||||
assert Group.objects.filter(name="站点管理员").count() == 1
|
||||
group = Group.objects.get(name="站点管理员")
|
||||
root_page = Page.objects.get(depth=1)
|
||||
# 重复运行不应产生重复的页面权限记录
|
||||
assert GroupPagePermission.objects.filter(
|
||||
group=group, page=root_page, permission__codename="add_page"
|
||||
).count() == 1
|
||||
assert group.permissions.filter(codename="access_admin").count() == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 部署引导(bootstrap_deployment 管理命令)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_bootstrap_deployment_creates_superuser_from_env_vars(monkeypatch):
|
||||
from django.core.management import call_command
|
||||
|
||||
monkeypatch.setenv("DJANGO_SUPERUSER_USERNAME", "deploy-admin")
|
||||
monkeypatch.setenv("DJANGO_SUPERUSER_EMAIL", "deploy-admin@example.com")
|
||||
monkeypatch.setenv("DJANGO_SUPERUSER_PASSWORD", "s3cret-pass")
|
||||
|
||||
call_command("bootstrap_deployment")
|
||||
|
||||
User = get_user_model()
|
||||
user = User.objects.get(username="deploy-admin")
|
||||
assert user.is_superuser is True
|
||||
assert user.is_staff is True
|
||||
assert user.email == "deploy-admin@example.com"
|
||||
assert user.check_password("s3cret-pass") is True
|
||||
|
||||
|
||||
def test_bootstrap_deployment_skips_superuser_creation_when_env_vars_missing(monkeypatch):
|
||||
from django.core.management import call_command
|
||||
|
||||
monkeypatch.delenv("DJANGO_SUPERUSER_USERNAME", raising=False)
|
||||
monkeypatch.delenv("DJANGO_SUPERUSER_EMAIL", raising=False)
|
||||
monkeypatch.delenv("DJANGO_SUPERUSER_PASSWORD", raising=False)
|
||||
|
||||
User = get_user_model()
|
||||
count_before = User.objects.count()
|
||||
|
||||
call_command("bootstrap_deployment")
|
||||
|
||||
assert User.objects.count() == count_before
|
||||
|
||||
|
||||
def test_bootstrap_deployment_does_not_overwrite_existing_superuser_password(monkeypatch):
|
||||
from django.core.management import call_command
|
||||
|
||||
User = get_user_model()
|
||||
User.objects.create_superuser(
|
||||
username="deploy-admin", email="old@example.com", password="original-pass"
|
||||
)
|
||||
|
||||
monkeypatch.setenv("DJANGO_SUPERUSER_USERNAME", "deploy-admin")
|
||||
monkeypatch.setenv("DJANGO_SUPERUSER_EMAIL", "new@example.com")
|
||||
monkeypatch.setenv("DJANGO_SUPERUSER_PASSWORD", "new-pass")
|
||||
|
||||
call_command("bootstrap_deployment")
|
||||
|
||||
user = User.objects.get(username="deploy-admin")
|
||||
assert user.email == "old@example.com"
|
||||
assert user.check_password("original-pass") is True
|
||||
assert user.check_password("new-pass") is False
|
||||
|
||||
|
||||
def test_bootstrap_deployment_also_initializes_rbac_groups(monkeypatch):
|
||||
from django.contrib.auth.models import Group
|
||||
from django.core.management import call_command
|
||||
|
||||
monkeypatch.delenv("DJANGO_SUPERUSER_USERNAME", raising=False)
|
||||
monkeypatch.delenv("DJANGO_SUPERUSER_EMAIL", raising=False)
|
||||
monkeypatch.delenv("DJANGO_SUPERUSER_PASSWORD", raising=False)
|
||||
|
||||
call_command("bootstrap_deployment")
|
||||
|
||||
expected_names = {"站点管理员", "编辑", "审核员", "作者", "查看者"}
|
||||
assert expected_names <= set(Group.objects.values_list("name", flat=True))
|
||||
|
||||
@@ -0,0 +1,17 @@
|
||||
from django.urls import path
|
||||
|
||||
from .api import (
|
||||
NavigationMenuView,
|
||||
PartnerListView,
|
||||
SiteSettingsView,
|
||||
TeamMemberListView,
|
||||
TestimonialListView,
|
||||
)
|
||||
|
||||
urlpatterns = [
|
||||
path("team/", TeamMemberListView.as_view(), name="team-list"),
|
||||
path("testimonials/", TestimonialListView.as_view(), name="testimonial-list"),
|
||||
path("partners/", PartnerListView.as_view(), name="partner-list"),
|
||||
path("navigation/", NavigationMenuView.as_view(), name="navigation-menu"),
|
||||
path("site-settings/", SiteSettingsView.as_view(), name="site-settings"),
|
||||
]
|
||||
@@ -0,0 +1,28 @@
|
||||
"""
|
||||
LeadDeletionRequest 通过 Django Admin(非 Wagtail)暴露为只读审计视图,
|
||||
供合规/客服人员查看 PIPL 数据删除申请处理记录,详见
|
||||
documents/设计方案分析与完善版.md §2.15。
|
||||
"""
|
||||
from django.contrib import admin
|
||||
|
||||
from .models import LeadDeletionRequest
|
||||
|
||||
|
||||
@admin.register(LeadDeletionRequest)
|
||||
class LeadDeletionRequestAdmin(admin.ModelAdmin):
|
||||
list_display = (
|
||||
"contact",
|
||||
"status",
|
||||
"requested_at",
|
||||
"completed_at",
|
||||
"deleted_count",
|
||||
)
|
||||
list_filter = ("status",)
|
||||
search_fields = ("contact",)
|
||||
readonly_fields = [f.name for f in LeadDeletionRequest._meta.fields]
|
||||
|
||||
def has_add_permission(self, request):
|
||||
return False
|
||||
|
||||
def has_change_permission(self, request, obj=None):
|
||||
return False
|
||||
@@ -0,0 +1,32 @@
|
||||
# Generated by Django 6.0.5 on 2026-08-07 05:13
|
||||
|
||||
import uuid
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('forms', '0001_initial'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='LeadDeletionRequest',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('contact', models.EmailField(max_length=254, verbose_name='联系邮箱')),
|
||||
('token', models.UUIDField(default=uuid.uuid4, editable=False, unique=True, verbose_name='确认令牌')),
|
||||
('status', models.CharField(choices=[('pending', '待确认'), ('completed', '已完成'), ('expired', '已过期')], default='pending', max_length=20, verbose_name='状态')),
|
||||
('requested_at', models.DateTimeField(auto_now_add=True, verbose_name='申请时间')),
|
||||
('confirmed_at', models.DateTimeField(blank=True, null=True, verbose_name='确认时间')),
|
||||
('completed_at', models.DateTimeField(blank=True, null=True, verbose_name='完成时间')),
|
||||
('deleted_count', models.PositiveIntegerField(default=0, verbose_name='已删除记录数')),
|
||||
],
|
||||
options={
|
||||
'verbose_name': '数据删除申请(PIPL)',
|
||||
'verbose_name_plural': '数据删除申请(PIPL)',
|
||||
'ordering': ['-requested_at'],
|
||||
},
|
||||
),
|
||||
]
|
||||
@@ -2,8 +2,14 @@
|
||||
线索表单:FormDefinition 是可复用的表单定义 Snippet,通过 core.blocks.FormBlock
|
||||
嵌入任意 StreamField 页面;提交结果保存为 Lead。
|
||||
详见 documents/设计方案分析与完善版.md §2.6(FormBlock)与 B2B 官网核心转化路径。
|
||||
|
||||
LeadDeletionRequest 支持 PIPL 数据主体删除权(详见 §2.15)。
|
||||
"""
|
||||
import uuid
|
||||
from datetime import timedelta
|
||||
|
||||
from django.db import models
|
||||
from django.utils import timezone
|
||||
from modelcluster.fields import ParentalKey
|
||||
from modelcluster.models import ClusterableModel
|
||||
from wagtail.admin.panels import FieldPanel, InlinePanel
|
||||
@@ -111,3 +117,50 @@ class Lead(models.Model):
|
||||
def __str__(self):
|
||||
form_name = self.form.name if self.form_id else "(表单已删除)"
|
||||
return f"{form_name} - {self.created_at:%Y-%m-%d %H:%M}"
|
||||
|
||||
|
||||
class LeadDeletionRequest(models.Model):
|
||||
"""PIPL 数据主体删除权申请记录(合规审计用途)。
|
||||
|
||||
匿名用户凭提交表单时留下的邮箱发起申请,邮件确认后才会删除匹配的 Lead 记录,
|
||||
避免任意人凭他人邮箱发起删除。仅支持邮箱验证(暂未接入短信网关,
|
||||
手机号验证需待国内短信/验证码服务落地后再支持)。
|
||||
"""
|
||||
|
||||
STATUS_PENDING = "pending"
|
||||
STATUS_COMPLETED = "completed"
|
||||
STATUS_EXPIRED = "expired"
|
||||
STATUS_CHOICES = [
|
||||
(STATUS_PENDING, "待确认"),
|
||||
(STATUS_COMPLETED, "已完成"),
|
||||
(STATUS_EXPIRED, "已过期"),
|
||||
]
|
||||
|
||||
TOKEN_VALID_HOURS = 24
|
||||
|
||||
contact = models.EmailField(verbose_name="联系邮箱")
|
||||
token = models.UUIDField(
|
||||
default=uuid.uuid4, unique=True, editable=False, verbose_name="确认令牌"
|
||||
)
|
||||
status = models.CharField(
|
||||
max_length=20,
|
||||
choices=STATUS_CHOICES,
|
||||
default=STATUS_PENDING,
|
||||
verbose_name="状态",
|
||||
)
|
||||
requested_at = models.DateTimeField(auto_now_add=True, verbose_name="申请时间")
|
||||
confirmed_at = models.DateTimeField(null=True, blank=True, verbose_name="确认时间")
|
||||
completed_at = models.DateTimeField(null=True, blank=True, verbose_name="完成时间")
|
||||
deleted_count = models.PositiveIntegerField(default=0, verbose_name="已删除记录数")
|
||||
|
||||
class Meta:
|
||||
verbose_name = "数据删除申请(PIPL)"
|
||||
verbose_name_plural = "数据删除申请(PIPL)"
|
||||
ordering = ["-requested_at"]
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.contact} - {self.get_status_display()}"
|
||||
|
||||
@property
|
||||
def is_expired(self):
|
||||
return timezone.now() > self.requested_at + timedelta(hours=self.TOKEN_VALID_HOURS)
|
||||
|
||||
@@ -5,3 +5,18 @@ class LeadSubmitSerializer(serializers.Serializer):
|
||||
form_id = serializers.IntegerField()
|
||||
data = serializers.DictField()
|
||||
source_url = serializers.URLField(required=False, allow_blank=True, default="")
|
||||
# 蜂蜜陷阱反垃圾注册字段:正常用户不会看到/填写该字段,机器人填写则视为垃圾提交。
|
||||
# 无需接入极验/腾讯云等第三方验证码服务即可降低自动化垃圾提交,详见设计文档 §2.15。
|
||||
website = serializers.CharField(required=False, allow_blank=True, default="")
|
||||
|
||||
|
||||
class LeadDeletionRequestSerializer(serializers.Serializer):
|
||||
"""PIPL 数据删除权:发起申请时仅需提交联系邮箱。"""
|
||||
|
||||
contact = serializers.EmailField()
|
||||
|
||||
|
||||
class LeadDeletionConfirmSerializer(serializers.Serializer):
|
||||
"""PIPL 数据删除权:确认时提交邮件中的 token。"""
|
||||
|
||||
token = serializers.UUIDField()
|
||||
|
||||
+158
-1
@@ -1,12 +1,28 @@
|
||||
"""apps.forms 单元测试:线索提交 API(B2B 官网核心转化路径,测试覆盖率要求较高)。"""
|
||||
from datetime import timedelta
|
||||
|
||||
import pytest
|
||||
from django.core.cache import cache
|
||||
from django.utils import timezone
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from apps.forms.models import FormDefinition, FormDefinitionField, Lead
|
||||
from apps.forms.models import (
|
||||
FormDefinition,
|
||||
FormDefinitionField,
|
||||
Lead,
|
||||
LeadDeletionRequest,
|
||||
)
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clear_throttle_cache():
|
||||
"""ScopedRateThrottle 依赖默认缓存记录请求次数,避免测试间相互影响(forms scope 限流 5/min)。"""
|
||||
cache.clear()
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def contact_form():
|
||||
form_def = FormDefinition.objects.create(
|
||||
@@ -62,3 +78,144 @@ def test_lead_submit_requires_form_id():
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_lead_submit_honeypot_filled_silently_rejects_without_creating_lead(
|
||||
contact_form,
|
||||
):
|
||||
"""蜜罐字段非空视为机器人:返回“成功”假象但不落库、不发通知邮件。"""
|
||||
client = APIClient()
|
||||
response = client.post(
|
||||
"/api/v1/custom/leads/",
|
||||
{
|
||||
"form_id": contact_form.pk,
|
||||
"data": {"name": "张三"},
|
||||
"website": "http://spam.example.com",
|
||||
},
|
||||
format="json",
|
||||
)
|
||||
|
||||
assert response.status_code == 201
|
||||
assert response.data["success"] is True
|
||||
assert not Lead.objects.exists()
|
||||
|
||||
|
||||
# --- PIPL 数据删除权(详见设计文档 §2.15) ---
|
||||
|
||||
|
||||
def test_deletion_request_returns_generic_message_and_sends_email(
|
||||
contact_form, mailoutbox
|
||||
):
|
||||
Lead.objects.create(
|
||||
form=contact_form,
|
||||
data={"name": "张三", "email": "zhangsan@example.com"},
|
||||
)
|
||||
|
||||
client = APIClient()
|
||||
response = client.post(
|
||||
"/api/v1/custom/leads/deletion-requests/",
|
||||
{"contact": "zhangsan@example.com"},
|
||||
format="json",
|
||||
)
|
||||
|
||||
assert response.status_code == 202
|
||||
assert LeadDeletionRequest.objects.filter(contact="zhangsan@example.com").exists()
|
||||
assert len(mailoutbox) == 1
|
||||
assert mailoutbox[0].to == ["zhangsan@example.com"]
|
||||
|
||||
|
||||
def test_deletion_request_returns_same_generic_message_for_unknown_contact(mailoutbox):
|
||||
"""未提交过表单的邮箱也返回相同提示,避免探测。"""
|
||||
client = APIClient()
|
||||
response_known = client.post(
|
||||
"/api/v1/custom/leads/deletion-requests/",
|
||||
{"contact": "known@example.com"},
|
||||
format="json",
|
||||
)
|
||||
response_unknown = client.post(
|
||||
"/api/v1/custom/leads/deletion-requests/",
|
||||
{"contact": "unknown@example.com"},
|
||||
format="json",
|
||||
)
|
||||
|
||||
assert response_known.status_code == response_unknown.status_code == 202
|
||||
assert response_known.data["message"] == response_unknown.data["message"]
|
||||
|
||||
|
||||
def test_deletion_confirm_deletes_matching_leads_only(contact_form):
|
||||
matching_lead = Lead.objects.create(
|
||||
form=contact_form,
|
||||
data={"name": "张三", "email": "zhangsan@example.com"},
|
||||
)
|
||||
other_lead = Lead.objects.create(
|
||||
form=contact_form,
|
||||
data={"name": "李四", "email": "lisi@example.com"},
|
||||
)
|
||||
deletion_request = LeadDeletionRequest.objects.create(contact="zhangsan@example.com")
|
||||
|
||||
client = APIClient()
|
||||
response = client.post(
|
||||
"/api/v1/custom/leads/deletion-requests/confirm/",
|
||||
{"token": str(deletion_request.token)},
|
||||
format="json",
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.data["success"] is True
|
||||
assert response.data["deleted_count"] == 1
|
||||
assert not Lead.objects.filter(pk=matching_lead.pk).exists()
|
||||
assert Lead.objects.filter(pk=other_lead.pk).exists()
|
||||
|
||||
deletion_request.refresh_from_db()
|
||||
assert deletion_request.status == LeadDeletionRequest.STATUS_COMPLETED
|
||||
assert deletion_request.deleted_count == 1
|
||||
assert deletion_request.completed_at is not None
|
||||
|
||||
|
||||
def test_deletion_confirm_invalid_token_returns_404():
|
||||
client = APIClient()
|
||||
response = client.post(
|
||||
"/api/v1/custom/leads/deletion-requests/confirm/",
|
||||
{"token": "00000000-0000-0000-0000-000000000000"},
|
||||
format="json",
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert response.data["error"]["code"] == "INVALID_TOKEN"
|
||||
|
||||
|
||||
def test_deletion_confirm_expired_token_returns_400():
|
||||
deletion_request = LeadDeletionRequest.objects.create(contact="zhangsan@example.com")
|
||||
LeadDeletionRequest.objects.filter(pk=deletion_request.pk).update(
|
||||
requested_at=timezone.now() - timedelta(hours=25)
|
||||
)
|
||||
|
||||
client = APIClient()
|
||||
response = client.post(
|
||||
"/api/v1/custom/leads/deletion-requests/confirm/",
|
||||
{"token": str(deletion_request.token)},
|
||||
format="json",
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert response.data["error"]["code"] == "TOKEN_EXPIRED"
|
||||
|
||||
deletion_request.refresh_from_db()
|
||||
assert deletion_request.status == LeadDeletionRequest.STATUS_EXPIRED
|
||||
|
||||
|
||||
def test_deletion_confirm_already_processed_returns_400():
|
||||
deletion_request = LeadDeletionRequest.objects.create(
|
||||
contact="zhangsan@example.com",
|
||||
status=LeadDeletionRequest.STATUS_COMPLETED,
|
||||
)
|
||||
|
||||
client = APIClient()
|
||||
response = client.post(
|
||||
"/api/v1/custom/leads/deletion-requests/confirm/",
|
||||
{"token": str(deletion_request.token)},
|
||||
format="json",
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert response.data["error"]["code"] == "ALREADY_PROCESSED"
|
||||
|
||||
+11
-1
@@ -1,7 +1,17 @@
|
||||
from django.urls import path
|
||||
|
||||
from .views import LeadSubmitView
|
||||
from .views import LeadDeletionConfirmView, LeadDeletionRequestView, LeadSubmitView
|
||||
|
||||
urlpatterns = [
|
||||
path("leads/", LeadSubmitView.as_view(), name="lead-submit"),
|
||||
path(
|
||||
"leads/deletion-requests/",
|
||||
LeadDeletionRequestView.as_view(),
|
||||
name="lead-deletion-request",
|
||||
),
|
||||
path(
|
||||
"leads/deletion-requests/confirm/",
|
||||
LeadDeletionConfirmView.as_view(),
|
||||
name="lead-deletion-confirm",
|
||||
),
|
||||
]
|
||||
|
||||
+129
-2
@@ -2,17 +2,27 @@
|
||||
线索提交接口。挂载在 /api/v1/custom/leads/(详见 apps/forms/urls.py 与
|
||||
documents/设计方案分析与完善版.md §2.7 自定义业务接口设计)。
|
||||
公开表单提交接口,限流走 DRF ScopedRateThrottle 的 "forms" scope(5/min,防刷)。
|
||||
|
||||
LeadDeletionRequestView/LeadDeletionConfirmView 支持 PIPL 数据主体删除权(§2.15)。
|
||||
"""
|
||||
import json
|
||||
import logging
|
||||
|
||||
from django.conf import settings
|
||||
from django.core.mail import send_mail
|
||||
from django.utils import timezone
|
||||
from rest_framework.response import Response
|
||||
from rest_framework.throttling import ScopedRateThrottle
|
||||
from rest_framework.views import APIView
|
||||
|
||||
from .models import FormDefinition, Lead
|
||||
from .serializers import LeadSubmitSerializer
|
||||
from .models import FormDefinition, Lead, LeadDeletionRequest
|
||||
from .serializers import (
|
||||
LeadDeletionConfirmSerializer,
|
||||
LeadDeletionRequestSerializer,
|
||||
LeadSubmitSerializer,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class LeadSubmitView(APIView):
|
||||
@@ -31,6 +41,13 @@ class LeadSubmitView(APIView):
|
||||
{"error": {"code": "NOT_FOUND", "message": "表单不存在"}}, status=404
|
||||
)
|
||||
|
||||
# 蜜罐反垃圾:honeypot 字段非空说明是自动化脚本提交,直接返回“成功”假象,
|
||||
# 不落库、不发通知邮件,避免打草惊蛇(详见设计文档 §2.15 验证码/反垃圾方案)。
|
||||
if serializer.validated_data.get("website"):
|
||||
return Response(
|
||||
{"success": True, "message": form_def.success_message}, status=201
|
||||
)
|
||||
|
||||
lead = Lead.objects.create(
|
||||
form=form_def,
|
||||
data=serializer.validated_data["data"],
|
||||
@@ -49,3 +66,113 @@ class LeadSubmitView(APIView):
|
||||
return Response(
|
||||
{"success": True, "message": form_def.success_message}, status=201
|
||||
)
|
||||
|
||||
|
||||
# PIPL 数据删除权:出于隐私考虑,无论邮箱是否存在关联数据,均返回相同的通用提示,
|
||||
# 避免被用来探测某个邮箱是否曾提交过表单。
|
||||
GENERIC_DELETION_REQUEST_MESSAGE = (
|
||||
"如果我们持有与该邮箱关联的信息,确认邮件将发送至该邮箱,请查收并点击链接完成删除确认。"
|
||||
)
|
||||
|
||||
|
||||
class LeadDeletionRequestView(APIView):
|
||||
"""发起 PIPL 数据删除申请:POST {"contact": "user@example.com"}。"""
|
||||
|
||||
throttle_classes = [ScopedRateThrottle]
|
||||
throttle_scope = "forms"
|
||||
|
||||
def post(self, request, *args, **kwargs):
|
||||
serializer = LeadDeletionRequestSerializer(data=request.data)
|
||||
serializer.is_valid(raise_exception=True)
|
||||
contact = serializer.validated_data["contact"]
|
||||
|
||||
deletion_request = LeadDeletionRequest.objects.create(contact=contact)
|
||||
confirm_url = (
|
||||
f"{settings.FRONTEND_BASE_URL.rstrip('/')}/privacy-policy/delete-confirm"
|
||||
f"?token={deletion_request.token}"
|
||||
)
|
||||
try:
|
||||
send_mail(
|
||||
subject="确认删除您的个人信息",
|
||||
message=(
|
||||
"我们收到了删除您个人信息的申请。如果这不是您本人操作,请忽略此邮件。\n\n"
|
||||
f"请在 {LeadDeletionRequest.TOKEN_VALID_HOURS} 小时内点击以下链接确认删除:\n"
|
||||
f"{confirm_url}"
|
||||
),
|
||||
from_email=getattr(settings, "DEFAULT_FROM_EMAIL", None),
|
||||
recipient_list=[contact],
|
||||
fail_silently=False,
|
||||
)
|
||||
except Exception:
|
||||
logger.exception("发送删除确认邮件失败:%s", contact)
|
||||
|
||||
return Response({"message": GENERIC_DELETION_REQUEST_MESSAGE}, status=202)
|
||||
|
||||
|
||||
class LeadDeletionConfirmView(APIView):
|
||||
"""确认 PIPL 数据删除申请:POST {"token": "..."},校验通过后删除匹配的 Lead 记录。"""
|
||||
|
||||
throttle_classes = [ScopedRateThrottle]
|
||||
throttle_scope = "forms"
|
||||
|
||||
def post(self, request, *args, **kwargs):
|
||||
serializer = LeadDeletionConfirmSerializer(data=request.data)
|
||||
serializer.is_valid(raise_exception=True)
|
||||
token = serializer.validated_data["token"]
|
||||
|
||||
try:
|
||||
deletion_request = LeadDeletionRequest.objects.get(token=token)
|
||||
except LeadDeletionRequest.DoesNotExist:
|
||||
return Response(
|
||||
{"error": {"code": "INVALID_TOKEN", "message": "删除链接无效。"}},
|
||||
status=404,
|
||||
)
|
||||
|
||||
if deletion_request.status != LeadDeletionRequest.STATUS_PENDING:
|
||||
return Response(
|
||||
{
|
||||
"error": {
|
||||
"code": "ALREADY_PROCESSED",
|
||||
"message": "该删除申请已处理或已过期。",
|
||||
}
|
||||
},
|
||||
status=400,
|
||||
)
|
||||
|
||||
if deletion_request.is_expired:
|
||||
deletion_request.status = LeadDeletionRequest.STATUS_EXPIRED
|
||||
deletion_request.save(update_fields=["status"])
|
||||
return Response(
|
||||
{"error": {"code": "TOKEN_EXPIRED", "message": "删除链接已过期,请重新申请。"}},
|
||||
status=400,
|
||||
)
|
||||
|
||||
contact = deletion_request.contact.strip().lower()
|
||||
matched_ids = [
|
||||
lead.pk
|
||||
for lead in Lead.objects.all()
|
||||
if any(
|
||||
isinstance(value, str) and value.strip().lower() == contact
|
||||
for value in lead.data.values()
|
||||
)
|
||||
]
|
||||
deleted_count = len(matched_ids)
|
||||
if matched_ids:
|
||||
Lead.objects.filter(pk__in=matched_ids).delete()
|
||||
|
||||
now = timezone.now()
|
||||
deletion_request.status = LeadDeletionRequest.STATUS_COMPLETED
|
||||
deletion_request.confirmed_at = now
|
||||
deletion_request.completed_at = now
|
||||
deletion_request.deleted_count = deleted_count
|
||||
deletion_request.save(
|
||||
update_fields=["status", "confirmed_at", "completed_at", "deleted_count"]
|
||||
)
|
||||
|
||||
return Response(
|
||||
{
|
||||
"success": True,
|
||||
"deleted_count": deleted_count,
|
||||
"message": "已成功删除与该邮箱关联的个人信息。",
|
||||
}
|
||||
)
|
||||
|
||||
File diff suppressed because one or more lines are too long
@@ -0,0 +1,53 @@
|
||||
"""apps.home 单元测试:HomePage 页面树结构。
|
||||
|
||||
test_home_page_subpage_types_allows_all_top_level_sections 是一条回归测试:
|
||||
此前 HomePage.subpage_types 曾经只包含 blog.BlogIndexPage,导致 products/cases/
|
||||
solutions 的 IndexPage 无法通过 Wagtail 管理后台"添加子页面"界面创建(subpage_types/
|
||||
parent_page_types 只影响后台创建校验,不影响 ORM add_child(),问题一度被脚本化测试掩盖)。
|
||||
"""
|
||||
import pytest
|
||||
|
||||
from apps.blog.models import BlogIndexPage
|
||||
from apps.cases.models import CaseStudyIndexPage
|
||||
from apps.core.models import SimpleContentPage
|
||||
from apps.home.models import HomePage
|
||||
from apps.products.models import ProductIndexPage
|
||||
from apps.solutions.models import SolutionIndexPage
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
def test_home_page_subpage_types_allows_all_top_level_sections():
|
||||
expected_types = {
|
||||
"blog.BlogIndexPage",
|
||||
"products.ProductIndexPage",
|
||||
"solutions.SolutionIndexPage",
|
||||
"cases.CaseStudyIndexPage",
|
||||
"core.SimpleContentPage",
|
||||
}
|
||||
|
||||
assert expected_types.issubset(set(HomePage.subpage_types))
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"index_model,kwargs",
|
||||
[
|
||||
(BlogIndexPage, {"title": "博客", "slug": "blog"}),
|
||||
(ProductIndexPage, {"title": "产品", "slug": "products"}),
|
||||
(SolutionIndexPage, {"title": "解决方案", "slug": "solutions"}),
|
||||
(CaseStudyIndexPage, {"title": "案例", "slug": "cases"}),
|
||||
(SimpleContentPage, {"title": "隐私政策", "slug": "privacy-policy"}),
|
||||
],
|
||||
)
|
||||
def test_each_top_level_section_can_be_created_under_home_in_admin(
|
||||
home_page, index_model, kwargs
|
||||
):
|
||||
"""`can_create_at` 是 Wagtail 管理后台"添加子页面"按钮实际使用的判断(综合
|
||||
parent.subpage_types 与 child.parent_page_types 两个方向),此前的 bug 正是
|
||||
仅靠 ORM `add_child()` 无法暴露的——它只影响管理后台,不影响直接调用 ORM。"""
|
||||
assert index_model.can_create_at(home_page) is True
|
||||
|
||||
child = index_model(**kwargs)
|
||||
home_page.add_child(instance=child)
|
||||
|
||||
assert index_model.objects.child_of(home_page).count() == 1
|
||||
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
@@ -0,0 +1,39 @@
|
||||
"""apps.solutions 单元测试:页面树结构与列表上下文(与 apps.products/apps.cases 同构)。"""
|
||||
import pytest
|
||||
|
||||
from apps.solutions.models import SolutionIndexPage, SolutionPage
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def solution_index(home_page):
|
||||
index = SolutionIndexPage(title="解决方案", slug="solutions", intro="行业解决方案")
|
||||
home_page.add_child(instance=index)
|
||||
return index
|
||||
|
||||
|
||||
def test_solution_page_can_be_created_under_index(solution_index):
|
||||
solution = SolutionPage(
|
||||
title="制造业解决方案",
|
||||
slug="manufacturing",
|
||||
industry="制造业",
|
||||
summary="面向制造业的数字化转型方案",
|
||||
)
|
||||
solution_index.add_child(instance=solution)
|
||||
|
||||
assert SolutionPage.objects.live().descendant_of(solution_index).count() == 1
|
||||
|
||||
|
||||
def test_solution_index_context_excludes_draft_solutions(solution_index):
|
||||
live_solution = SolutionPage(title="方案 A", slug="solution-a")
|
||||
solution_index.add_child(instance=live_solution)
|
||||
|
||||
draft_solution = SolutionPage(title="方案 B", slug="solution-b", live=False)
|
||||
solution_index.add_child(instance=draft_solution)
|
||||
|
||||
context = solution_index.get_context(request=None)
|
||||
titles = [s.title for s in context["solutions"]]
|
||||
|
||||
assert "方案 A" in titles
|
||||
assert "方案 B" not in titles
|
||||
@@ -0,0 +1,34 @@
|
||||
# OpenSearch + 中文分词(IK 插件)本地开发 PoC 环境。
|
||||
#
|
||||
# 用途:验证 Wagtail 原生 opensearch2 后端 + infinilabs/analysis-ik 中文分词插件的可行性,
|
||||
# 详见 documents/设计方案分析与完善版.md §2.10 "OpenSearch + 中文分词调研"。
|
||||
# 这是一个独立的 compose 文件,不与主应用的部署编排耦合,按需单独启停:
|
||||
#
|
||||
# docker-compose -f docker-compose.opensearch.yml up -d
|
||||
# curl http://localhost:9200/_cat/plugins # 确认 analysis-ik 已加载
|
||||
# docker-compose -f docker-compose.opensearch.yml down -v # 清理(-v 连数据卷一起删)
|
||||
#
|
||||
# 注意:plugins.security.disabled=true 仅用于本地免证书调试,生产环境必须启用安全插件并配置证书。
|
||||
|
||||
services:
|
||||
opensearch:
|
||||
build:
|
||||
context: ./docker/opensearch
|
||||
container_name: wagtailcms-opensearch-dev
|
||||
environment:
|
||||
- discovery.type=single-node
|
||||
- plugins.security.disabled=true
|
||||
- OPENSEARCH_JAVA_OPTS=-Xms512m -Xmx512m
|
||||
ports:
|
||||
- "9200:9200"
|
||||
- "9600:9600"
|
||||
volumes:
|
||||
- opensearch-dev-data:/usr/share/opensearch/data
|
||||
healthcheck:
|
||||
test: ["CMD-SHELL", "curl -sf http://localhost:9200/_cluster/health || exit 1"]
|
||||
interval: 10s
|
||||
timeout: 5s
|
||||
retries: 10
|
||||
|
||||
volumes:
|
||||
opensearch-dev-data:
|
||||
@@ -0,0 +1,11 @@
|
||||
# 本地开发/调研用镜像:在官方 OpenSearch 2.x 基础上安装 IK 中文分词插件(infinilabs/analysis-ik)。
|
||||
# 仅用于本地开发环境的 PoC 验证;生产环境建议评估阿里云"开放搜索 OpenSearch 版"等托管方案,
|
||||
# 或在自建时用同样的方式基于官方镜像打包(见 documents/设计方案分析与完善版.md §2.10)。
|
||||
FROM opensearchproject/opensearch:2.19.6
|
||||
|
||||
# ⚠️ IK 插件版本必须与 OpenSearch 核心版本完全一致才能加载成功。
|
||||
# 使用前请先到 https://release.infinilabs.com/ 确认是否存在与上面 FROM 版本号匹配的构建;
|
||||
# 如没有完全匹配的版本,需要把 FROM 的 OpenSearch 版本号和下面的插件版本号同时改成
|
||||
# release.infinilabs.com 上实际可用的最新匹配版本,本 Dockerfile 尚未在本地实际构建验证过。
|
||||
RUN opensearch-plugin install --batch \
|
||||
https://get.infini.cloud/opensearch/analysis-ik/2.19.6
|
||||
+63
-22
@@ -219,15 +219,15 @@ Snippet(非页面树内容,用于跨页面复用):`TeamMember`、`Testim
|
||||
| HeroBlock | 首屏 | title, subtitle, cta_buttons(list), background_image/video | ✅ 已实现 |
|
||||
| StatsBlock | 数据展示 | items: [{value, label}] | ✅ 已实现 |
|
||||
| FeatureGridBlock | 能力/优势 | icon, title, description | ✅ 已实现(对应原 FeatureBlock) |
|
||||
| ProductCardBlock | 产品矩阵 | title, description, image, link | ⬜ 待开发 |
|
||||
| ProductCardBlock | 产品矩阵 | title, description, image, link | ✅ 已实现 |
|
||||
| LogoCloudBlock | 客户 Logo 墙 | logos: [{url, title}](暂为图片列表,未接入 Partner Snippet) | ✅ 已实现(简化版) |
|
||||
| CaseStudyBlock | 案例卡片 | case_page: PageChooser(cases.CaseStudyPage), summary | ✅ 已实现 |
|
||||
| PricingBlock | 定价 | plans: [{name, price, features}] | ⬜ 待开发 |
|
||||
| PricingBlock | 定价 | plans: [{name, price, features, highlighted, button}] | ✅ 已实现 |
|
||||
| FAQBlock | 常见问题 | items: [{question, answer(richtext)}] | ✅ 已实现 |
|
||||
| TimelineBlock | 发展历程 | items: [{year, event}] | ⬜ 待开发 |
|
||||
| TeamBlock | 团队展示 | members: [SnippetChooser(TeamMember)] | ⬜ 待开发(依赖 TeamMember Snippet) |
|
||||
| TechStackBlock | 技术架构图 | items: [{icon, label}] | ⬜ 待开发 |
|
||||
| VideoBlock | 视频 | video_file/embed_url, poster | ⬜ 待开发 |
|
||||
| TimelineBlock | 发展历程 | items: [{year, event}] | ✅ 已实现 |
|
||||
| TeamBlock | 团队展示 | members: [SnippetChooser(TeamMember)],`get_api_representation` 展开完整字段(name/role/bio/photo)供前端渲染 | ✅ 已实现(依赖 TeamMember Snippet) |
|
||||
| TechStackBlock | 技术架构图 | items: [{icon, label}] | ✅ 已实现 |
|
||||
| VideoBlock | 视频 | video_url(建议自建 OSS/COS 直链或腾讯视频/哔哩哔哩等国内平台嵌入地址),poster | ✅ 已实现 |
|
||||
| FormBlock | 线索表单 | form: SnippetChooser(forms.FormDefinition),`get_api_representation` 展开完整字段定义供前端渲染 | ✅ 已实现 |
|
||||
| CTABlock | 行动号召 | heading, button_text, button_link | ✅ 已实现 |
|
||||
|
||||
@@ -246,9 +246,10 @@ Snippet(非页面树内容,用于跨页面复用):`TeamMember`、`Testim
|
||||
- 限流:DRF `ScopedRateThrottle`,公开只读接口 `100/min`,表单提交接口 `5/min`(防刷)
|
||||
- 版本弃用策略:新版本上线后旧版本保留至少 2 个发布周期,响应头带 `Deprecation` / `Sunset`
|
||||
|
||||
**预览模式(原文档缺失,重要补充)**
|
||||
- Wagtail 后台"预览"生成一次性 `preview_token`(存 Redis,TTL 5 分钟)
|
||||
- Next.js 提供 `/api/draft` 路由,校验 token 后开启 Next.js Draft Mode,直接向 Wagtail 请求草稿版本 API(`?revision=latest`)
|
||||
**预览模式(已实现,详见 §2.16)**
|
||||
- 后端引入 `wagtail-headless-preview`(0.9.0),`SEOablePage` 混入 `HeadlessPreviewMixin`,编辑器点击"预览"时自动将草稿序列化存入 `PagePreview` 表,生成带签名 token 的预览链接并重定向到 `{FRONTEND_BASE_URL}/preview`
|
||||
- 自定义 `PagePreviewAPIViewSet`(`apps/api/urls.py`)挂载于 `/api/v2/page_preview/`,根据 `content_type`+`token` 从 `PagePreview` 取回草稿快照并序列化返回
|
||||
- Next.js `/preview` Route Handler 校验 token 有效后开启内置 Draft Mode(cookie 方式,非 Redis),并重定向到对应内容类型的前端路由;页面组件通过 `getPreviewOrFallback` 优先拉取草稿,否则回退到正常已发布内容
|
||||
|
||||
**发布后缓存失效(原文档只有框图,这里补齐实现)**
|
||||
```python
|
||||
@@ -279,16 +280,47 @@ Next.js 侧对应 `revalidateTag` API Route,收到 webhook 后失效对应 ISR
|
||||
|
||||
落地方式:使用 Wagtail 原生 `Group` + `GroupPagePermission` + `collections` 权限,不需要额外引入 RBAC 框架。
|
||||
|
||||
**已落地(`python manage.py setup_rbac_groups`,幂等命令,`apps/core/management/commands/setup_rbac_groups.py`):**
|
||||
|
||||
- Super Admin 直接对应 Django `is_superuser=True`,不建单独 Group;其余 5 个角色对应 5 个 Wagtail `Group`:站点管理员/编辑/审核员/作者/查看者。
|
||||
- 页面权限(`GroupPagePermission`,挂载于页面树根节点 `depth=1`,适用于当前单站点部署):站点管理员拥有全部类型(add/change/publish/delete/bulk_delete/lock/unlock/view);编辑拥有 add/change/view(无 publish,创建的页面停留在草稿态等待发布);审核员拥有 change/publish/view(用于打开页面并执行发布,属于对“页面编辑=查看”矩阵条目的必要放宽);**作者仅拥有 add/view(不含 change)**——利用 Wagtail 内置 `OwnershipPermissionPolicy`:只有 `add` 权限时,用户可创建新页面,且只能编辑/删除自己拥有(`owner`)的页面,天然满足“仅自己”要求,无需自定义 `wagtail_hooks`;查看者仅有 view。
|
||||
- Snippet 权限(`TeamMember`/`Testimonial`/`Partner`/`NavigationMenu`/`FormDefinition`):站点管理员与编辑均为 add/change/delete/view,审核员仅 view,作者/查看者无权限。`Lead`(线索,含 PIPL 个人信息)出于数据保护考虑做了比矩阵更严格的收紧:仅站点管理员可管理,审核员仅 view,编辑/作者均不授予。`SiteSettings` 作为站点级单例配置仅站点管理员可 change/view。
|
||||
- Collection(图片/文档)权限(`GroupCollectionPermission`,挂载于根 Collection):站点管理员 add/change/delete/choose/view;编辑与作者 add/change/choose/view(便于上传 StreamField 图片);审核员 choose/view;查看者无。
|
||||
- 用户管理权限(`auth.add_user`/`change_user`/`delete_user`/`view_user`)仅授予站点管理员,对应矩阵“用户管理”列;“Django Admin”列不通过 Group 授予(Django Admin 访问由账号的 `is_staff` 属性决定,仅超级用户具备)。
|
||||
- 简化说明:设计方案 §2.9 的 Workflow 审核模块尚未实现,因此“提交审核/审批发布”当前通过“编辑者无 publish、审核员/站点管理员有 publish”的权限差异形成审核闸门,而非正式的 Wagtail `Workflow`/`TaskState`;待 Workflow 模块落地后可在现有 Group 基础上叠加 `GroupApprovalTask`,无需重新设计权限组。
|
||||
- 测试:`apps/core/tests.py` 新增 7 个用例覆盖 Group 创建、`access_admin` 授予、页面权限矩阵、Author 仅编辑自己页面、Lead 权限收紧、用户管理权限范围、命令幂等性。
|
||||
|
||||
### 2.9 工作流与内容审核
|
||||
|
||||
采用 Wagtail 5+ 内置 `Workflow` 模块,定义 `草稿 → 编辑自检 → 审核 → 通过 → 定时发布`,并在 `TaskState` 变更时通过企业微信机器人 Webhook 通知责任人(替代原文档 Slack 方案)。
|
||||
|
||||
### 2.10 搜索、SEO、多语言
|
||||
|
||||
- 搜索:中文分词是关键缺失点,OpenSearch 需配置 **IK 分词插件**(原文档完全没提中文分词,这是国内落地必须解决的问题)。
|
||||
- 搜索:中文分词是关键缺失点,OpenSearch 需配置 **IK 分词插件**(原文档完全没提中文分词,这是国内落地必须解决的问题)。调研结论见下方小节。
|
||||
- SEO:站点级 `sitemap.xml`(`wagtail.contrib.sitemaps`)+ 页面级 `schema_json`(JSON-LD)+ `robots.txt` 动态生成。
|
||||
- 多语言:`WAGTAIL_I18N_ENABLED` + `wagtail-localize`,URL 采用 `/zh/`、`/en/` 前缀,默认 `/zh/` 对国内用户免前缀(根路径直接是中文)。
|
||||
|
||||
#### OpenSearch + 中文分词调研(Phase 2 提前调研,新增)
|
||||
|
||||
- **Wagtail 原生已支持 OpenSearch**:本仓库安装的 Wagtail 7.4.2 底层依赖 `django-modelsearch`,已内置 `wagtail.search.backends.opensearch2`(OpenSearch 2.x)与 `opensearch3`(OpenSearch 3.x)两个后端模块,配置方式与 Elasticsearch 后端一致,**不需要任何第三方 Wagtail 插件**:
|
||||
```python
|
||||
WAGTAILSEARCH_BACKENDS = {
|
||||
"default": {
|
||||
"BACKEND": "wagtail.search.backends.opensearch2",
|
||||
"URLS": [env("OPENSEARCH_URL", default="http://localhost:9200")],
|
||||
"INDEX_PREFIX": "wagtailcms_",
|
||||
}
|
||||
}
|
||||
```
|
||||
需额外安装客户端包 `pip install "opensearch-py>=2,<3"`(版本需与 OpenSearch 服务端大版本一致)。**此配置尚未写入 settings**,留待真正接入、且本地/测试环境有可用 OpenSearch 服务时再启用,避免开发环境无 OpenSearch 服务导致搜索功能报错。
|
||||
- **中文分词方案**:OpenSearch 官方未内置中文分词器,采用社区插件 [infinilabs/analysis-ik](https://github.com/infinilabs/analysis-ik)(同一套插件同时支持 Elasticsearch 和 OpenSearch,Apache-2.0 协议,17k+ star,长期维护),提供 `ik_smart`(粗粒度,适合短语查询)与 `ik_max_word`(细粒度,适合词项查询)两个分析器,典型字段 mapping 为 `"analyzer": "ik_max_word", "search_analyzer": "ik_smart"`。Wagtail 侧可通过 `WAGTAILSEARCH_BACKENDS['default']['INDEX_SETTINGS']` 覆盖默认 `analysis.analyzer.default` 让 Wagtail 自动生成的索引改用 IK 分词,无需改 Wagtail 源码。
|
||||
- ⚠️ **插件版本必须与 OpenSearch 核心版本完全一致**才能加载成功,接入前需在 https://release.infinilabs.com/ 或用 `bin/opensearch-plugin install https://get.infini.cloud/opensearch/analysis-ik/<version>` 确认存在对应构建。
|
||||
- **本地 PoC 脚手架(已创建,未在本环境验证)**:新增 [docker-compose.opensearch.yml](../docker-compose.opensearch.yml) + [docker/opensearch/Dockerfile](../docker/opensearch/Dockerfile),基于官方 `opensearchproject/opensearch:2.19.6` 镜像叠加 IK 插件安装,`plugins.security.disabled=true` 仅用于本地免证书调试。**当前开发环境未安装 Docker CLI,本次调研未能实际拉起容器验证**,需在具备 Docker 的机器上执行 `docker-compose -f docker-compose.opensearch.yml up -d`,并完成以下验证步骤后再正式接入:
|
||||
1. `curl http://localhost:9200/_cat/plugins` 确认 `analysis-ik` 已加载;
|
||||
2. 安装 `opensearch-py`,在 settings 中启用上面的 `opensearch2` 后端配置;
|
||||
3. `python manage.py update_index` 重建索引,用中文短语(如“产品案例”)人工验证检索召回是否符合预期的分词粒度。
|
||||
- **生产替代方案**:若不想自建/自运维 OpenSearch 集群(插件升级、扩缩容等),可评估阿里云“开放搜索 OpenSearch 版”或腾讯云 ES Serverless 的托管中文分词能力,二者原生支持中文分词、无需自装 IK 插件,代价是绑定云厂商,需按实际预算/团队运维能力二选一(已同步写入 §2.17 风险表)。
|
||||
|
||||
### 2.11 安全设计(落地到具体配置)
|
||||
|
||||
```python
|
||||
@@ -300,6 +332,8 @@ SECURE_HSTS_SECONDS = 31536000
|
||||
SECURE_HSTS_INCLUDE_SUBDOMAINS = True
|
||||
SECURE_CONTENT_TYPE_NOSNIFF = True
|
||||
X_FRAME_OPTIONS = "DENY"
|
||||
# 阿里云 SLB / 腾讯云 CLB 等负载均衡在边缘终止 TLS 时,需要此项避免 SECURE_SSL_REDIRECT 死循环重定向
|
||||
SECURE_PROXY_SSL_HEADER = ("HTTP_X_FORWARDED_PROTO", "https")
|
||||
CORS_ALLOWED_ORIGINS = env.list("CORS_ALLOWED_ORIGINS", default=[])
|
||||
CSRF_TRUSTED_ORIGINS = env.list("CSRF_TRUSTED_ORIGINS", default=[])
|
||||
|
||||
@@ -310,6 +344,12 @@ REST_FRAMEWORK = {
|
||||
```
|
||||
富文本安全:`WAGTAILADMIN_RICH_TEXT_EDITORS` 限定可用 feature(禁用危险 HTML embed),后端渲染前统一走 `bleach` 白名单清洗。
|
||||
|
||||
**白名单配置说明**(三者含义不同,勿混淆):
|
||||
- `ALLOWED_HOSTS`:填**后端自身**域名(如 `cms.example.com`),校验 HTTP `Host` 头,防 Host 头注入。
|
||||
- `CORS_ALLOWED_ORIGINS`:填 **Next.js 前端**域名(如 `https://www.example.com`),允许浏览器跨域调用 Wagtail REST API;`/api/v1/custom/leads/` 为匿名公开接口,仅受此白名单 + 限流保护,不受 CSRF 校验。
|
||||
- `CSRF_TRUSTED_ORIGINS`:仅在"浏览器带 session Cookie 发起跨站不安全请求"时生效(如已登录编辑者调用需认证的预览/管理接口),填涉及认证态跨域请求的域名。
|
||||
- 三者均通过 `.env` 中的 `ALLOWED_HOSTS`/`CORS_ALLOWED_ORIGINS`/`CSRF_TRUSTED_ORIGINS` 环境变量注入,部署时按实际域名填写即可,无需改代码(参见 `.env.example`)。
|
||||
|
||||
### 2.12 可观测性与运维
|
||||
|
||||
- 健康检查:`/healthz`(存活)、`/readyz`(含数据库/Redis 连通性检查),供 K8s liveness/readiness 探针使用。
|
||||
@@ -340,8 +380,8 @@ CI 中要求单元测试覆盖率不低于 70%,核心 `apps/forms`(涉及线
|
||||
- [x] 页脚展示备案号并链接工信部备案查询(`frontend/components/layout/Footer.tsx`,实际备案号 冀ICP备2025130506号-1)
|
||||
- [x] 隐私政策内容已上线(`apps.core.SimpleContentPage` 通用富文本内容页模型 + `seed_privacy_policy` 管理命令 + 前端 `/privacy-policy` 页面 + 页脚链接,内容覆盖 PIPL 要求的收集目的/用途/用户权利/Cookie 说明等)
|
||||
- [x] Cookie 同意条控件(`frontend/components/layout/CookieConsent.tsx`,`"use client"` 组件,首次访问展示"接受/拒绝"横幅,选择结果存入 `localStorage`,已挂载到根 layout)
|
||||
- [ ] 字体/地图/验证码/CDN 均替换为可在国内正常访问的服务
|
||||
- [ ] 表单收集的个人信息需明确告知用途并支持删除请求(PIPL 数据主体权利)
|
||||
- [x] 字体/地图/验证码/CDN 排查:字体已用系统字体栈(`frontend/app/layout.tsx`,未引入 `next/font/google` 等国外字体服务);地图/视频等嵌入类 Block 尚未开发,落地时需选用腾讯地图/高德地图、腾讯视频/B站等国内可访问服务替代 Google Maps/YouTube(见 §2.6 待办);验证码此前仅靠限流(`ScopedRateThrottle` forms scope 5/min)防护,未接入任何第三方验证码,现补充蜜罐反垃圾字段 `website`(`apps/forms/serializers.py`/`views.py` + `frontend/blocks/form/LeadForm.tsx`,正常用户不可见,机器人误填后静默返回"成功"但不落库不发邮件),无需接入极验/腾讯云验证码 API 即可降低自动化垃圾提交,未来如需更强防护可再接入国内验证码厂商;CDN 方面为 `production.py` 新增 `AWS_S3_CUSTOM_DOMAIN`(`OSS_CDN_DOMAIN` 环境变量)用于绑定阿里云/腾讯云 CDN 加速域名到 OSS/COS 源站,避免直接暴露源站地址;另确认 `sentry-sdk` 虽在 `requirements/production.txt` 中但代码里从未调用 `sentry_sdk.init()`,为休眠依赖,未来若启用需评估国内可访问性(自建 GlitchTip 或国内 APM 替代 sentry.io)
|
||||
- [x] 表单收集的个人信息支持删除请求(PIPL 数据主体权利):新增 `LeadDeletionRequest` 模型(`apps/forms/models.py`),匿名用户凭提交表单时留下的邮箱在 `/privacy-policy/delete-request` 页面发起申请(`POST /api/v1/custom/leads/deletion-requests/`,限流 forms scope 5/min,无论邮箱是否存在关联数据均返回相同提示以避免探测),系统发送含 24 小时有效 token 的确认邮件,用户在 `/privacy-policy/delete-confirm?token=` 页面点击确认后(`POST /api/v1/custom/leads/deletion-requests/confirm/`)按邮箱在 `Lead.data` 中做匹配并删除对应记录;`LeadDeletionRequest` 处理记录通过 Django Admin(只读)供合规审计;隐私政策页面已加入申请入口链接
|
||||
|
||||
### 2.16 开发路线图(对齐当前仓库状态)
|
||||
|
||||
@@ -356,24 +396,25 @@ CI 中要求单元测试覆盖率不低于 70%,核心 `apps/forms`(涉及线
|
||||
|
||||
已完成:
|
||||
- [x] 后端 Django + Wagtail 项目骨架,分层 settings(base/dev/production),独立 git 仓库并已提交
|
||||
- [x] `apps/core`:`SEOablePage` 抽象基类、`COMMON_BLOCKS`(9/14 个 Block,见 §2.6)、健康检查 `/healthz` `/readyz`、发布后 `page_published` signal → 前端 revalidate webhook
|
||||
- [x] `apps/core`:`SEOablePage` 抽象基类、`COMMON_BLOCKS`(14/14 个 Block 已全部实现,见 §2.6)、健康检查 `/healthz` `/readyz`、发布后 `page_published` signal → 前端 revalidate webhook
|
||||
- [x] `apps/home`、`apps/blog`(含标签 `ClusterTaggableManager`)
|
||||
- [x] `apps/products`(ProductIndexPage/ProductPage)、`apps/cases`(CaseStudyIndexPage/CaseStudyPage)、`apps/solutions`(SolutionIndexPage/SolutionPage)
|
||||
- [x] `apps/forms`:`FormDefinition`/`FormDefinitionField`/`Lead`(Snippet 方式)+ 提交 API(`/api/v1/custom/leads/`,限流 + 邮件通知)
|
||||
- [x] Wagtail API v2 挂载、分页/过滤/排序(Wagtail 内置)
|
||||
- [x] Next.js 前端脚手架(独立仓库):Header/Footer/layout、首页、博客/产品/案例/解决方案列表与详情页、`BlockRenderer`、ISR + revalidate route,`lint`/`build` 已验证通过
|
||||
- [x] pytest 基础测试框架(pytest-django + factory_boy):`pytest.ini` + `conftest.py`(root_page/home_page fixtures),为 `apps/core`(SEOablePage 字段、FormBlock API 表示、SimpleContentPage 页面树)、`apps/products`、`apps/cases`、`apps/forms`(线索提交 API)编写了基础单元测试,共 11 个用例均通过
|
||||
- [x] pytest 基础测试框架(pytest-django + factory_boy):`pytest.ini` + `conftest.py`(root_page/home_page fixtures),为 `apps/core`(SEOablePage 字段、FormBlock API 表示、SimpleContentPage 页面树、Snippet 模型与只读 API)、`apps/products`、`apps/cases`、`apps/forms`(线索提交 API、蜜罐反垃圾字段、PIPL 数据删除权申请/确认流程)编写了基础单元测试,共 31 个用例均通过
|
||||
- [x] `apps/core.SimpleContentPage`:通用富文本法务/说明类页面模型,用于隐私政策等内容,配套 `seed_privacy_policy` management command 用于幂等创建/更新隐私政策页面(需在 HomePage 实例存在后手动运行)
|
||||
- [x] 部署引导 management command `bootstrap_deployment`:读取 `DJANGO_SUPERUSER_USERNAME`/`_EMAIL`/`_PASSWORD` 三个环境变量,若均已设置且用户名不存在则创建首个超级管理员账号(已存在则跳过,不覆盖密码),随后自动调用 `setup_rbac_groups` 初始化 RBAC 权限组;幂等,可放入部署脚本/容器启动流程的 `migrate` 之后执行,解决“首次部署后如何登录 Wagtail 后台”的鸡生蛋问题;`.env.example` 已补充对应三项环境变量说明;新增 4 个单元测试覆盖建号/跳过/不覆盖密码/联动 RBAC 四种场景
|
||||
- [x] Snippet:`TeamMember`/`Testimonial`/`Partner`(简单 `@register_snippet`,均含 `order` 排序字段)、`NavigationMenu`+`NavigationMenuItem`(`ClusterableModel`+`Orderable`+`InlinePanel`,内部页面/外部链接二选一)、`SiteSettings`(`wagtail.contrib.settings` + `BaseSiteSetting`,公司信息/ICP备案/社交账号全局配置)均已创建;只读 API 挂载于 `/api/v1/custom/core/`(`team/`、`testimonials/`、`partners/`、`navigation/?name=`、`site-settings/`,限流 public 100/min);前端 `Header`/`Footer` 已接入 `NavigationMenu`/`SiteSettings`(接口不可用时回退静态内容),`TeamMember`/`Testimonial`/`Partner` 已提供 service 层,页面级展示留待对应 StreamField Block(见下)落地时接入
|
||||
- [x] 预览模式:后端引入 `wagtail-headless-preview`(0.9.0),`SEOablePage` 混入 `HeadlessPreviewMixin`(自动覆盖全部 6 个页面模型),新增自定义 `PagePreviewAPIViewSet` 挂载于 `/api/v2/page_preview/`;前端新增 `services/preview.service.ts`(`getPreviewOrFallback`/`fetchPreviewPage`/`resolvePreviewPath`)、`/preview`与 `/preview/disable` 两个 Route Handler(基于 Next.js Draft Mode + httpOnly cookie),并在首页/博客/产品/案例/解决方案详情页接入草稿优先逻辑,`layout.tsx` 新增预览模式提示条(含退出链接);`lint`/`build` 均验证通过,后端 31 个现有用例均通过(本次未新增自动化测试,仅依靠人工验证)
|
||||
- [x] 剩余 6 个 StreamField Block(`ProductCardBlock`/`PricingBlock`/`TimelineBlock`/`TeamBlock`/`TechStackBlock`/`VideoBlock`)已全部开发完成:`apps/core/blocks.py` 新增对应 Block 类并扩充 `COMMON_BLOCKS`(现共 14 个内容 Block + 1 个 `richtext`),因 Wagtail StreamField 会将 Block 结构写入迁移(`use_json_field=True` 不代表可跳过迁移),已为 blog/cases/home/products/solutions 五个 Page 模型生成并应用 `alter field body` 迁移;`TeamBlock` 复用 `FormBlock` 的 `get_api_representation` 覆写模式,新增 `TeamMemberChooserBlock(SnippetChooserBlock)` 展开 `TeamMember` 完整字段(含 `serialize_image` 复用),并补充对应的空值/展开单元测试(新增 2 个用例,后端共 33 个测试全部通过);前端 `types/wagtail.ts` 新增 6 个 Block 值类型与 `StreamFieldBlock` 联合类型分支,新建 6 个渲染组件(`ProductCardGrid`/`Pricing`/`Timeline`/`Team`/`TechStack`/`Video`)并接入 `BlockRenderer.tsx` 映射表,`lint`/`build` 均验证通过
|
||||
- [x] RBAC 落地:`python manage.py setup_rbac_groups` 幂等命令创建 5 个 Wagtail Group(站点管理员/编辑/审核员/作者/查看者)并按 §2.8 矩阵配置 `GroupPagePermission`/`GroupCollectionPermission`/Snippet 与用户管理 `Permission`;Author“仅自己”利用 Wagtail 内置 `OwnershipPermissionPolicy`(仅 add 权限)原生实现,无需自定义 hook;新增 7 个后端单元测试,全仓库 pytest 从 33 增至 40,全部通过
|
||||
|
||||
未完成(待规划排期):
|
||||
- [ ] Snippet:`TeamMember`/`Testimonial`/`Partner`/`NavigationMenu`/`SiteSettings`(`BaseSiteSetting`)均未创建
|
||||
- [ ] 预览模式(Wagtail `preview_token` + Next.js Draft Mode)未实现
|
||||
- [ ] 剩余 5 个 StreamField Block(Pricing/Timeline/Team/TechStack/Video/ProductCard)未开发
|
||||
- [ ] RBAC 落地(Wagtail `Group` + `GroupPagePermission` 实际配置)未开始
|
||||
- [ ] 测试覆盖率仍不完整(已有基础单元测试,但集成测试、前端组件测试、E2E 均未编写,当前无 CI 流水线)
|
||||
- [ ] 生产安全 settings(§2.11 中 `SECURE_*`/CORS/CSRF 白名单等)未在 `production.py` 中逐项落实确认
|
||||
- [ ] ICP 备案已完成,页脚已展示实际备案号(冀ICP备2025130506号-1);隐私政策内容已上线,Cookie 同意条控件(`frontend/components/layout/CookieConsent.tsx`,localStorage 记忆用户选择)已实现并挂载到全局 layout
|
||||
- [ ] 中文分词、OpenSearch 集成(Phase 2 提前项)未开始
|
||||
- [ ] 测试覆盖率持续提升中:后端 pytest 用例从 40 个增至 65 个,新增覆盖 `apps/blog`(标签、列表排序/草稿过滤)、`apps/solutions`(页面树、草稿过滤)、`apps/home`(`HomePage.subpage_types` 回归测试,覆盖此前"后台无法添加子页面"的 bug 场景)、`apps/core` 健康检查端点 `/healthz`/`/readyz`、发布后 revalidate webhook 信号(含正常/未配置/网络异常三种路径)、`apps/api` 预览接口 `page_preview`(含无效 token/content_type 场景,过程中顺带修复了一个真实 bug:无效 token 之前会 500 而非文档承诺的 404)、`bootstrap_deployment` 部署引导命令(建号/跳过/不覆盖密码/联动 RBAC);但集成测试、前端组件测试、E2E 均未编写,当前无 CI 流水线
|
||||
- [x] 生产安全 settings(§2.11 中 `SECURE_*`/CORS/CSRF 白名单等)已在 `production.py` 中逐项落实,并补充 `SECURE_PROXY_SSL_HEADER`(适配国内云厂商 SLB/CLB 边缘终止 TLS 场景);实际域名需在部署时通过 `.env` 填入
|
||||
- [x] ICP 备案、页脚备案号展示、隐私政策内容、Cookie 同意条控件、表单 PIPL 数据删除权支持、字体/地图/验证码/CDN 国内可访问排查均已完成,§2.15 合规与本地化清单已全部完成
|
||||
- [ ] 中文分词、OpenSearch 集成(Phase 2 提前项):**调研已完成**(结论见 §2.10 “OpenSearch + 中文分词调研”小节:确认 Wagtail 7.4.2 原生内置 `opensearch2`/`opensearch3` 后端 + infinilabs `analysis-ik` 插件的可行方案,并新增本地 Docker Compose PoC 脚手架 `docker-compose.opensearch.yml`),**实际接入(拉起容器验证插件加载、切换 `WAGTAILSEARCH_BACKENDS`、建站内容重建索引)仍未开始**,待具备 Docker 环境后再验证落地
|
||||
|
||||
### 2.17 风险与备选方案
|
||||
|
||||
|
||||
@@ -5,3 +5,4 @@ django-cors-headers>=4.9,<5.0
|
||||
django-environ>=0.14,<1.0
|
||||
django-filter>=26.0,<27.0
|
||||
requests>=2.34,<3.0
|
||||
wagtail-headless-preview>=0.9,<1.0
|
||||
|
||||
@@ -30,6 +30,7 @@ INSTALLED_APPS = [
|
||||
# Wagtail
|
||||
"wagtail.contrib.forms",
|
||||
"wagtail.contrib.redirects",
|
||||
"wagtail.contrib.settings",
|
||||
"wagtail.embeds",
|
||||
"wagtail.sites",
|
||||
"wagtail.users",
|
||||
@@ -41,6 +42,8 @@ INSTALLED_APPS = [
|
||||
"wagtail",
|
||||
"modelcluster",
|
||||
"taggit",
|
||||
# 预览模式(Wagtail preview_token + Next.js Draft Mode,详见 §2.7)
|
||||
"wagtail_headless_preview",
|
||||
# Django
|
||||
"django.contrib.admin",
|
||||
"django.contrib.auth",
|
||||
@@ -78,6 +81,7 @@ TEMPLATES = [
|
||||
"django.template.context_processors.request",
|
||||
"django.contrib.auth.context_processors.auth",
|
||||
"django.contrib.messages.context_processors.messages",
|
||||
"wagtail.contrib.settings.context_processors.settings",
|
||||
],
|
||||
},
|
||||
},
|
||||
@@ -131,3 +135,15 @@ CORS_ALLOWED_ORIGINS = env.list("CORS_ALLOWED_ORIGINS", default=[])
|
||||
# 发布后通知前端做 ISR 缓存失效(详见设计文档 §2.7)
|
||||
FRONTEND_REVALIDATE_URL = env("FRONTEND_REVALIDATE_URL", default="")
|
||||
REVALIDATE_SECRET = env("REVALIDATE_SECRET", default="")
|
||||
|
||||
# 前端站点基础 URL,用于拼接邮件中的链接,如 PIPL 数据删除确认链接(详见 §2.15)
|
||||
FRONTEND_BASE_URL = env("FRONTEND_BASE_URL", default="http://localhost:3000")
|
||||
|
||||
# Headless 预览模式(wagtail_headless_preview):编辑器点击“预览”时,
|
||||
# 后端会将草稿内容序列化后存入 PagePreview 并重定向到 Next.js 的 /preview 路由,
|
||||
# 预览页面再通过自定义 page_preview API(apps/api/urls.py)拉取草稿数据。
|
||||
WAGTAIL_HEADLESS_PREVIEW = {
|
||||
"CLIENT_URLS": {"default": f"{FRONTEND_BASE_URL}/preview"},
|
||||
# Wagtail 7.1+ 推荐直接重定向到前端而非嵌套 iframe,便于滚动位置保持等体验与本项目跨域部署场景对齐
|
||||
"REDIRECT_ON_PREVIEW": True,
|
||||
}
|
||||
|
||||
@@ -25,6 +25,10 @@ SECURE_HSTS_PRELOAD = True
|
||||
SECURE_CONTENT_TYPE_NOSNIFF = True
|
||||
X_FRAME_OPTIONS = "DENY"
|
||||
|
||||
# 阿里云 SLB / 腾讯云 CLB 等负载均衡在边缘终止 TLS,转发到后端的是明文 HTTP,
|
||||
# 需要通过约定的代理头识别原始请求是否为 HTTPS,否则 SECURE_SSL_REDIRECT 会导致死循环重定向。
|
||||
SECURE_PROXY_SSL_HEADER = ("HTTP_X_FORWARDED_PROTO", "https")
|
||||
|
||||
CSRF_TRUSTED_ORIGINS = env.list("CSRF_TRUSTED_ORIGINS", default=[])
|
||||
|
||||
# 国内对象存储(阿里云 OSS / 腾讯云 COS 均兼容 S3 协议)
|
||||
@@ -36,6 +40,9 @@ AWS_ACCESS_KEY_ID = env("OSS_ACCESS_KEY_ID", default="")
|
||||
AWS_SECRET_ACCESS_KEY = env("OSS_SECRET_ACCESS_KEY", default="")
|
||||
AWS_STORAGE_BUCKET_NAME = env("OSS_BUCKET_NAME", default="")
|
||||
AWS_S3_ENDPOINT_URL = env("OSS_ENDPOINT_URL", default="")
|
||||
# 国内 CDN 加速域名(阿里云 CDN / 腾讯云 CDN / 又拍云,绑定到 OSS/COS 源站后填入),
|
||||
# 配置后媒体文件 URL 会替换成该加速域名,避免直接暴露 OSS 源站地址(详见 §2.15)。
|
||||
AWS_S3_CUSTOM_DOMAIN = env("OSS_CDN_DOMAIN", default="")
|
||||
|
||||
LOGGING = {
|
||||
"version": 1,
|
||||
|
||||
@@ -31,6 +31,8 @@ urlpatterns = [
|
||||
path('api/v2/', api_router.urls),
|
||||
# 自定义业务接口(非 Wagtail 页面路由),如线索表单提交
|
||||
path('api/v1/custom/', include('apps.forms.urls')),
|
||||
# 全局 Snippet 只读接口:团队/证言/合作伙伴/导航菜单/站点设置
|
||||
path('api/v1/custom/core/', include('apps.core.urls')),
|
||||
path('healthz', healthz, name='healthz'),
|
||||
path('readyz', readyz, name='readyz'),
|
||||
]
|
||||
|
||||
Reference in New Issue
Block a user