Compare commits

...
10 Commits
Author SHA1 Message Date
Zhengen 306068bc22 完善 .gitignore:补充 IDE/OS/日志/覆盖率等常见忽略规则 2026-08-13 14:39:33 +08:00
Zhengen 147107501b 补充测试覆盖率 + 新增部署引导命令 bootstrap_deployment
- 新增 apps/blog、apps/solutions、apps/home、apps/api 测试文件,覆盖标签/排序/
  草稿过滤、HomePage.subpage_types 回归测试、page_preview 预览接口
- apps/core/tests.py 新增健康检查端点、发布 webhook 信号、bootstrap_deployment
  命令的测试用例,pytest 用例数从 40 增至 65,全部通过
- 修复 apps/api/urls.py 中 PagePreviewAPIViewSet.get_object() 未捕获
  DoesNotExist 导致 500 而非文档承诺 404 的 bug
- 新增 apps/core/management/commands/bootstrap_deployment.py:部署时按环境变量
  幂等创建首个超级管理员账号并联动初始化 RBAC 权限组,.env.example 补充对应
  环境变量说明
- 新增 OpenSearch + 中文分词本地 PoC 脚手架(docker-compose.opensearch.yml +
  docker/opensearch/Dockerfile),调研结论详见设计文档 §2.10
- 更新设计文档 §2.16 Phase 1 进度清单
2026-08-11 14:47:10 +08:00
Zhengen f37571f395 feat(rbac): 落地 Wagtail Group + GroupPagePermission/GroupCollectionPermission/Snippet 权限体系
新增 setup_rbac_groups 幂等管理命令,按设计方案 §2.8 矩阵创建 5 个 Group(站点管理员/编辑/审核员/作者/查看者),配置页面/图片文档/Snippet/用户管理权限;Author 仅编辑自己页面利用 Wagtail 内置 OwnershipPermissionPolicy 实现,无需自定义 hook;新增 7 个单元测试,pytest 33→40 全部通过。
2026-08-07 17:07:39 +08:00
Zhengen 3944e11330 feat(core): 补全剩余 6 个 StreamField Block (ProductCard/Pricing/Timeline/Team/TechStack/Video)
- apps/core/blocks.py 新增 ProductCardBlock/PricingBlock/TimelineBlock/TeamBlock/TechStackBlock/VideoBlock,COMMON_BLOCKS 扩充至 14 个内容 Block
- TeamBlock 新增 TeamMemberChooserBlock(SnippetChooserBlock),复用 FormBlock 已有的 get_api_representation 展开模式,展开 TeamMember 完整字段(含 serialize_image 复用)
- 为 blog/cases/home/products/solutions 五个 Page 模型生成并应用 alter field body 迁移(StreamField block 定义变更会写入迁移 state)
- apps/core/tests.py 新增 TeamMemberChooserBlock 空值/展开单元测试,pytest 33 个用例全部通过
- 设计文档 §2.6/§2.16 同步更新为已实现
2026-08-07 15:58:44 +08:00
Zhengen 8d440b9e4a feat: 实现预览模式(wagtail-headless-preview + Next.js Draft Mode) 2026-08-07 15:34:46 +08:00
Zhengen 16e4f97e74 国内可访问服务替换排查:新增蜜罐反垃圾字段 + CDN 加速域名配置 2026-08-07 14:49:00 +08:00
Zhengen d6503bb46f feat: 实现PIPL数据主体删除权支持功能
- 新增 LeadDeletionRequest 模型,支持邮箱验证的删除申请/确认流程
- POST /api/v1/custom/leads/deletion-requests/ 发起申请(通用响应避免探测邮箱是否存在)
- POST /api/v1/custom/leads/deletion-requests/confirm/ 确认删除(24小时token有效期)
- 按邮箱在 Lead.data 中匹配并删除对应记录,跨SQLite/Postgres使用Python级匹配
- LeadDeletionRequest 通过 Django Admin 只读展示供合规审计
- 新增 FRONTEND_BASE_URL 设置用于拼接邮件确认链接
- apps/forms/tests.py 新增6个测试用例(含节流缓存隔离修复),全仓库30个用例通过
- 更新设计文档 §2.15/§2.16 勾选项
2026-08-07 13:29:20 +08:00
Zhengen 8fbbd2e6b0 feat: 新增Snippet全局内容模型(TeamMember/Testimonial/Partner/NavigationMenu/SiteSettings)
- 新增 apps.core.TeamMember/Testimonial/Partner:简单 register_snippet,含 order 排序字段
- 新增 apps.core.NavigationMenu+NavigationMenuItem:ClusterableModel+Orderable+InlinePanel,
  支持内部页面或外部链接二选一
- 新增 apps.core.SiteSettings:接入 wagtail.contrib.settings + BaseSiteSetting,
  存放公司信息/ICP备案号/社交账号等全局配置
- 新增只读 API:/api/v1/custom/core/{team,testimonials,partners,navigation,site-settings}/
- 补充 apps/core/tests.py 单元测试,pytest 用例由 11 个增至 24 个,全部通过
- 更新设计文档 §2.16 Phase 1 进度清单
2026-08-07 13:06:22 +08:00
Zhengen b04de9f5c1 docs: 修正Phase1进度清单中过期的复选框状态 2026-08-07 11:05:41 +08:00
Zhengen ec2e629db2 fix: 生产环境补充SECURE_PROXY_SSL_HEADER,完善白名单配置文档说明 2026-08-07 09:42:29 +08:00
35 changed files with 2318 additions and 32 deletions
+11
View File
@@ -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
View File
@@ -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/
+47
View File
@@ -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
+48
View File
@@ -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
+85
View File
@@ -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
+67
View File
@@ -0,0 +1,67 @@
"""
apps.core 全局 Snippet 只读接口。挂载在 /api/v1/custom/core/
(详见 apps/core/urls.py 与 documents/设计方案分析与完善版.md §2.7)。
公开只读数据,限流走 DRF ScopedRateThrottle 的 "public" scope100/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)
+115
View File
@@ -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
View File
@@ -1,14 +1,23 @@
"""
公共抽象模型。所有业务页面应继承 SEOablePage 而非直接继承 wagtail.models.Page
以统一获得 SEO 字段(详见 documents/设计方案分析与完善版.md §2.5)。
本文件同时存放跨页面复用的全局 SnippetTeamMember/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 = "站点设置"
+74
View File
@@ -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
View File
@@ -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 webhookapps/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))
+17
View File
@@ -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"),
]
+28
View File
@@ -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'],
},
),
]
+53
View File
@@ -2,8 +2,14 @@
线索表单:FormDefinition 是可复用的表单定义 Snippet,通过 core.blocks.FormBlock
嵌入任意 StreamField 页面;提交结果保存为 Lead。
详见 documents/设计方案分析与完善版.md §2.6FormBlock)与 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)
+15
View File
@@ -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
View File
@@ -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
View File
@@ -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
View File
@@ -2,17 +2,27 @@
线索提交接口挂载在 /api/v1/custom/leads/详见 apps/forms/urls.py
documents/设计方案分析与完善版.md §2.7 自定义业务接口设计
公开表单提交接口限流走 DRF ScopedRateThrottle "forms" scope5/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
+53
View File
@@ -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
+39
View File
@@ -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
+34
View File
@@ -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:
+11
View File
@@ -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
View File
@@ -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`(存 RedisTTL 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 Modecookie 方式,非 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 和 OpenSearchApache-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 项目骨架,分层 settingsbase/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 BlockPricing/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 风险与备选方案
+1
View File
@@ -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
+16
View File
@@ -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 APIapps/api/urls.py)拉取草稿数据。
WAGTAIL_HEADLESS_PREVIEW = {
"CLIENT_URLS": {"default": f"{FRONTEND_BASE_URL}/preview"},
# Wagtail 7.1+ 推荐直接重定向到前端而非嵌套 iframe,便于滚动位置保持等体验与本项目跨域部署场景对齐
"REDIRECT_ON_PREVIEW": True,
}
+7
View File
@@ -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,
+2
View File
@@ -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'),
]