从零搭建”一物一码”合格证查询系统——Django + DRF 后端架构详解

系列文章（2/5）：本篇将深入拆解后端核心架构，涵盖环境配置、数据模型、视图层、权限体系、中间件审计、JWT 认证与服务层设计。如果你还没阅读第一篇系统总览，建议先回顾后再继续。

一、Django 配置与环境管理

在实际项目中，开发环境和生产环境的配置往往差异巨大。我们使用 django-environ 配合 APP_ENV 环境变量，实现了一套零硬编码的配置方案。

1.1 基于 APP_ENV 的多环境加载

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# settings.py
import os
import environ
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent.parent

env = environ.Env(
    DEBUG=(bool, False),  # 默认关闭 DEBUG，防止生产环境泄露
)

# 核心：根据 APP_ENV 加载对应的 .env 文件
# 开发时 APP_ENV=development → 加载 .env.development
# 部署时 APP_ENV=production  → 加载 .env.production
APP_ENV = os.environ.get("APP_ENV", "development")
env_file = BASE_DIR / f".env.{APP_ENV}"
if env_file.is_file():
    environ.Env.read_env(str(env_file))

这样做的好处是：.env.development 和 .env.production 可以同时存在于项目中（开发用前者，CI/CD 注入后者），而 settings.py 本身不需要任何 if-else 分支。

1.2 关键配置项

1
2
3
4
5
# 安全相关
SECRET_KEY = env("SECRET_KEY")           # 绝不硬编码！
DEBUG = env("DEBUG")                     # 生产环境必须为 False
ALLOWED_HOSTS = env.list("ALLOWED_HOSTS", default=["localhost", "127.0.0.1"])
BASE_URL = env("BASE_URL", default="http://localhost:8000")

1.3 生产环境安全加固

当 APP_ENV=production 时，我们额外启用一系列安全中间件：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
if APP_ENV == "production":
    # HSTS：强制浏览器只用 HTTPS 访问，有效期一年
    SECURE_HSTS_SECONDS = 31536000
    SECURE_HSTS_INCLUDE_SUBDOMAINS = True
    SECURE_HSTS_PRELOAD = True

    # 自动将 HTTP 请求 301 重定向到 HTTPS
    SECURE_SSL_REDIRECT = True

    # Cookie 安全：仅在 HTTPS 下发送，防中间人窃取
    SESSION_COOKIE_SECURE = True
    CSRF_COOKIE_SECURE = True

    # 防止浏览器嗅探 Content-Type
    SECURE_CONTENT_TYPE_NOSNIFF = True

小贴士：SECURE_HSTS_SECONDS 一旦设置，浏览器会在指定时间内强制 HTTPS。首次部署建议先设为较短时间（如 3600），验证无误后再改为一年。

二、数据模型设计

模型层是整个系统的基石。在”一物一码”场景中，一张合格证（Certificate）绑定一批货物（Cargo），货物又关联一个产品（Product）。我们在模型层做了大量数据完整性保护。

2.1 FROZEN_FIELDS 不可变守卫

已发布的合格证具有法律效力，核心字段绝不能被篡改。我们在 save() 方法中实现了字段冻结机制：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
class Certificate(models.Model):
    # 一旦合格证发布，以下字段不可修改
    FROZEN_FIELDS = {
        "cert_number",       # 证书编号
        "inspection_org",    # 检验机构
        "inspection_date",   # 检验日期
        "standard_code",     # 执行标准
        "product",           # 关联产品
    }

    cert_number = models.CharField("证书编号", max_length=64, unique=True)
    publish_status = models.CharField(
        "发布状态", max_length=16,
        choices=[("draft", "草稿"), ("published", "已发布"), ("revoked", "已吊销")],
        default="draft",
    )
    # ... 其他字段省略

    def save(self, *args, **kwargs):
        if self.pk:
            # 从数据库重新读取旧值，逐字段比对
            old = type(self).objects.filter(pk=self.pk).first()
            if old and old.publish_status == "published":
                for f in self.FROZEN_FIELDS:
                    if getattr(old, f) != getattr(self, f):
                        raise PermissionError(
                            f"已发布证书的「{f}」字段不可修改"
                        )
        super().save(*args, **kwargs)

这种”模型层拦截”比在视图层做校验更可靠——无论是 API 调用、管理后台操作还是 manage.py shell 手动修改，都会被拦截。

2.2 部分唯一约束（Partial Unique Constraint）

同一批货物只能有一张有效的已发布证书，但草稿和已吊销的不受限制：

1
2
3
4
5
6
7
8
9
class Meta:
    constraints = [
        models.UniqueConstraint(
            fields=["cargo"],
            # 仅当 publish_status="published" 时生效
            condition=models.Q(publish_status="published"),
            name="uniq_active_published_cert_per_cargo",
        ),
    ]

这利用了 PostgreSQL 的 Partial Index 特性。当你尝试为同一批货物发布第二张证书时，数据库层面就会直接拒绝——连 ORM 都绕不过去。

2.3 自动生成 ID

系统中有三类需要自动生成的标识符，各有不同的生成策略：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import uuid
import secrets
from datetime import date

def _new_token():
    """扫码查询用的随机令牌，URL 安全、32 字符"""
    return secrets.token_urlsafe(24)  # 24 字节 → 32 字符

def _new_cargo_code():
    """货物批次号：日期前缀 + 8 位随机串，便于人工识读"""
    prefix = date.today().strftime("%Y%m%d")
    suffix = uuid.uuid4().hex[:8].upper()
    return f"CG-{prefix}-{suffix}"

def _new_cert_number():
    """证书编号：年份 + 12 位随机串，全局唯一"""
    year = date.today().strftime("%Y")
    rand = uuid.uuid4().hex[:12].upper()
    return f"CERT-{year}-{rand}"

使用 secrets.token_urlsafe 生成扫码令牌是有意为之——相比 uuid4，它提供了密码学安全的随机性，防止令牌被暴力枚举。

2.4 仅追加模型（Append-Only）

操作日志和审计记录是系统的”黑匣子”，一旦写入就不能修改或删除：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
class EditLog(models.Model):
    """编辑日志：只能创建，不能修改，不能删除"""
    certificate = models.ForeignKey(Certificate, on_delete=models.CASCADE)
    changed_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True)
    changed_at = models.DateTimeField(auto_now_add=True)
    field_name = models.CharField(max_length=64)
    old_value = models.TextField(blank=True)
    new_value = models.TextField(blank=True)

    def save(self, *args, **kwargs):
        if self.pk:
            raise PermissionError("编辑日志不可修改")
        super().save(*args, **kwargs)

    def delete(self, *args, **kwargs):
        raise PermissionError("编辑日志不可删除")


class AuditEntry(models.Model):
    """审计追踪：所有关键操作的不可篡改记录"""
    action = models.CharField("操作类型", max_length=32)
    actor = models.ForeignKey(User, on_delete=models.SET_NULL, null=True)
    ip_address = models.GenericIPAddressField("IP 地址", null=True)
    timestamp = models.DateTimeField(auto_now_add=True)
    detail = models.JSONField("详情", default=dict)

    def save(self, *args, **kwargs):
        if self.pk:
            raise PermissionError("审计记录不可修改")
        super().save(*args, **kwargs)

    def delete(self, *args, **kwargs):
        raise PermissionError("审计记录不可删除")

同样的手法——在 ORM 层强制执行业务规则，让上层代码无法”绕道”。

三、DRF 视图层架构

视图层采用 Django REST Framework 的 ViewSet 模式，将 CRUD 和自定义操作统一归集到一个类中。

3.1 ViewSet 基本结构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from rest_framework import viewsets, status
from rest_framework.decorators import action
from rest_framework.response import Response

class CertificateViewSet(viewsets.ModelViewSet):
    """合格证 CRUD + 业务操作"""
    queryset = Certificate.objects.select_related("product", "cargo")
    serializer_class = CertificateSerializer
    permission_classes = [IsAuthenticated, IsAdminOrReadOnly]

    def get_queryset(self):
        """普通操作员只能看到自己创建的证书"""
        qs = super().get_queryset()
        if not is_admin(self.request.user):
            qs = qs.filter(created_by=self.request.user)
        return qs

3.2 自定义批量操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
class CargoViewSet(viewsets.ModelViewSet):

    @action(detail=False, methods=["post"], url_path="batch-create")
    def batch_create(self, request):
        """批量创建货物批次，一次最多 500 条"""
        items = request.data.get("items", [])
        if len(items) > 500:
            return Response(
                {"error": "单次批量不超过 500 条"},
                status=status.HTTP_400_BAD_REQUEST,
            )
        created = cert_service.batch_create_cargo(items, actor=request.user)
        return Response({"created": len(created)}, status=status.HTTP_201_CREATED)

    @action(detail=False, methods=["post"], url_path="batch-bind-cert")
    def batch_bind_cert(self, request):
        """批量绑定证书到货物"""
        bindings = request.data.get("bindings", [])
        results = cert_service.batch_bind_certificate(bindings, actor=request.user)
        return Response(results)

    @action(detail=False, methods=["post"], url_path="batch-delete")
    def batch_delete(self, request):
        """批量删除草稿状态的货物（已发布的不可删）"""
        ids = request.data.get("ids", [])
        deleted = cert_service.batch_delete_cargo(ids, actor=request.user)
        return Response({"deleted": deleted})

    @action(detail=False, methods=["get"], url_path="export-xlsx")
    def export_xlsx(self, request):
        """导出 Excel 报表"""
        buffer = cert_service.export_cargo_xlsx(request.user)
        response = HttpResponse(
            buffer.getvalue(),
            content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        )
        response["Content-Disposition"] = 'attachment; filename="cargo_export.xlsx"'
        return response

3.3 审核流操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class CertificateViewSet(viewsets.ModelViewSet):

    @action(detail=True, methods=["post"], url_path="submit-review")
    def submit_review(self, request, pk=None):
        """提交审核：草稿 → 待审核"""
        cert = self.get_object()
        cert_service.submit_for_review(cert, actor=request.user)
        return Response({"status": "pending_review"})

    @action(detail=True, methods=["post"])
    def approve(self, request, pk=None):
        """审核通过：待审核 → 已发布（含双控检查）"""
        cert = self.get_object()
        self._check_dual_control(cert, request.user)
        cert_service.approve(cert, actor=request.user)
        return Response({"status": "published"})

    @action(detail=True, methods=["post"])
    def reject(self, request, pk=None):
        """审核驳回：待审核 → 草稿"""
        cert = self.get_object()
        reason = request.data.get("reason", "")
        cert_service.reject(cert, actor=request.user, reason=reason)
        return Response({"status": "draft"})

    @action(detail=True, methods=["post"])
    def revoke(self, request, pk=None):
        """吊销证书：已发布 → 已吊销"""
        cert = self.get_object()
        reason = request.data.get("reason", "")
        cert_service.revoke(cert, actor=request.user, reason=reason)
        return Response({"status": "revoked"})

    @action(detail=True, methods=["post"])
    def reissue(self, request, pk=None):
        """补发证书：基于已吊销的证书创建新草稿"""
        cert = self.get_object()
        new_cert = cert_service.reissue(cert, actor=request.user)
        return Response(CertificateSerializer(new_cert).data)

3.4 公开扫码查询端点

这是面向消费者的核心接口——扫描二维码后调用此端点验证合格证真伪：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
from rest_framework.throttling import AnonRateThrottle

class ScanRateThrottle(AnonRateThrottle):
    rate = "30/minute"  # 每分钟最多 30 次，防爬虫刷接口

class PublicCertView(APIView):
    """扫码查询——无需登录，面向消费者"""
    authentication_classes = []  # 公开接口，无需认证
    permission_classes = [AllowAny]
    throttle_classes = [ScanRateThrottle]

    def get(self, request, token):
        cargo = get_object_or_404(Cargo, token=token)
        cert = cargo.certificates.filter(publish_status="published").first()
        if not cert:
            return Response(
                {"error": "未找到有效合格证"},
                status=status.HTTP_404_NOT_FOUND,
            )

        result = {
            "cert_number": cert.cert_number,
            "product_name": cert.product.name,
            "inspection_org": cert.inspection_org,
            "inspection_date": cert.inspection_date,
            "standard_code": cert.standard_code,
        }

        # 风险标记：查询频次异常高的令牌
        scan_count = ScanLog.objects.filter(
            token=token,
            scanned_at__gte=now() - timedelta(hours=24),
        ).count()
        if scan_count > 50:
            result["risk_flag"] = "该码 24 小时内被扫描超过 50 次，请注意辨别"

        # 过期预警
        if cert.expiry_date and cert.expiry_date < date.today():
            result["expiry_alert"] = "该合格证已过期"

        # 完整性校验（Ed25519 签名验证）
        if cert.signature:
            is_valid = signing.verify_certificate(cert)
            result["integrity"] = "valid" if is_valid else "tampered"

        # 记录扫码日志
        ScanLog.objects.create(
            token=token,
            ip_address=get_client_ip(request),
            user_agent=request.META.get("HTTP_USER_AGENT", ""),
        )

        return Response(result)

3.5 认证级联（Authentication Cascade）

后台管理接口需要支持多种客户端访问场景，我们实现了三级认证回退：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def _try_authenticate(request):
    """
    认证级联：按优先级依次尝试三种认证方式
    1. JWT Header  — 标准 SPA 前端（Authorization: Bearer xxx）
    2. Django Session — 管理后台（admin站点）
    3. Cookie JWT   — 跨端口开发时的 Cookie 回退方案
    """
    # 第一优先：标准 JWT（Authorization header）
    from rest_framework_simplejwt.authentication import JWTAuthentication
    jwt_auth = JWTAuthentication()
    try:
        result = jwt_auth.authenticate(request)
        if result:
            return result[0]  # 返回 user 对象
    except Exception:
        pass

    # 第二优先：Django Session（已登录 admin 后台）
    if hasattr(request, "user") and request.user.is_authenticated:
        return request.user

    # 第三优先：Cookie 中的 JWT（开发环境跨端口场景）
    token = request.COOKIES.get("access_token")
    if token:
        try:
            validated = jwt_auth.get_validated_token(token)
            return jwt_auth.get_user(validated)
        except Exception:
            pass

    return None

3.6 双控校验（Dual Control）

合格证的”创建”和”审批”必须由不同的人完成，防止一人同时扮演两个角色：

1
2
3
4
5
6
7
8
9
def _check_dual_control(self, cert, approver):
    """
    双控检查：审批人 ≠ 创建人
    防止同一人既创建又审批，确保四眼原则（Four-Eyes Principle）
    """
    if cert.created_by == approver:
        raise PermissionDenied(
            "创建人不能审批自己创建的合格证，请交由其他管理员处理"
        )

四、三角色权限体系

系统采用三级角色设计，既简洁又够用：

角色 判定条件 核心权限
superadmin is_superuser=True 所有操作 + 用户管理
admin 属于 "admin" 分组 证书审批、吊销、补发
operator 任何已认证用户 创建草稿、提交审核

4.1 角色判定函数

1
2
3
4
5
6
7
8
9
def is_admin(user) -> bool:
    """判断用户是否为管理员（含超级管理员）"""
    return user.is_authenticated and (
        user.is_superuser or user.groups.filter(name="admin").exists()
    )

def is_superadmin(user) -> bool:
    """判断用户是否为超级管理员"""
    return user.is_authenticated and user.is_superuser

4.2 权限类实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from rest_framework.permissions import BasePermission, SAFE_METHODS

class IsSuperAdmin(BasePermission):
    """仅超级管理员可访问"""
    def has_permission(self, request, view):
        return is_superadmin(request.user)

class IsAdmin(BasePermission):
    """管理员及以上可访问"""
    def has_permission(self, request, view):
        return is_admin(request.user)

class IsAdminOrReadOnly(BasePermission):
    """管理员可写，其他人只读"""
    def has_permission(self, request, view):
        if request.method in SAFE_METHODS:
            return request.user.is_authenticated
        return is_admin(request.user)

class IsAdminAction(BasePermission):
    """用于装饰特定 action，非管理员直接拒绝"""
    def has_permission(self, request, view):
        admin_actions = {"approve", "reject", "revoke", "reissue"}
        if view.action in admin_actions:
            return is_admin(request.user)
        return True  # 其他 action 交由其他权限类处理

4.3 权限映射关系

1
2
3
4
5
6
7
8
9
POST   /api/certs/                   → operator+  （创建草稿）
GET    /api/certs/                   → operator+  （查看列表）
POST   /api/certs/{id}/submit-review → operator+  （提交审核）
POST   /api/certs/{id}/approve       → admin+     （审核通过）
POST   /api/certs/{id}/reject        → admin+     （审核驳回）
POST   /api/certs/{id}/revoke        → admin+     （吊销证书）
POST   /api/certs/{id}/reissue       → admin+     （补发证书）
GET    /api/users/                   → superadmin （用户管理）
GET    /api/scan/{token}             → 公开        （扫码查询）

五、中间件与审计

5.1 CurrentActorMiddleware

Django 的 ORM 信号（post_save, post_delete）触发时没有 request 上下文。但审计记录需要知道”谁做的”和”从哪里来的”。解决方案是使用 threading.local() 在中间件中存储当前请求者：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
import threading
from django.utils.deprecation import MiddlewareMixin

# threading.local() 确保每个线程有独立的存储空间
# 在多线程 WSGI 服务器（如 gunicorn --threads）中不会串数据
_local = threading.local()

def get_client_ip(request):
    """从请求头中提取真实客户端 IP（支持反向代理）"""
    x_forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR")
    if x_forwarded_for:
        return x_forwarded_for.split(",")[0].strip()
    return request.META.get("REMOTE_ADDR")

class CurrentActorMiddleware(MiddlewareMixin):
    """
    在请求生命周期内，将当前用户和 IP 存入线程局部变量。
    这样 ORM 信号处理器就能通过 get_current_actor() 获取操作者信息。
    """
    def __call__(self, request):
        # 请求进入时：设置 actor 和 ip
        if hasattr(request, "user") and request.user.is_authenticated:
            _local.actor = request.user
        else:
            _local.actor = None
        _local.ip = get_client_ip(request)

        try:
            response = self.get_response(request)
            return response
        finally:
            # 请求结束后：必须清理，防止线程复用导致数据泄露
            _local.actor = None
            _local.ip = None


def get_current_actor():
    """供 ORM 信号处理器调用，获取当前操作者"""
    return getattr(_local, "actor", None)

def get_current_ip():
    """供 ORM 信号处理器调用，获取当前请求 IP"""
    return getattr(_local, "ip", None)

5.2 信号处理器中使用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from django.db.models.signals import post_save
from django.dispatch import receiver

@receiver(post_save, sender=Certificate)
def log_certificate_change(sender, instance, created, **kwargs):
    """证书变更时自动创建审计记录"""
    actor = get_current_actor()
    ip = get_current_ip()

    AuditEntry.objects.create(
        action="cert_created" if created else "cert_updated",
        actor=actor,
        ip_address=ip,
        detail={
            "cert_id": instance.pk,
            "cert_number": instance.cert_number,
            "status": instance.publish_status,
        },
    )

为什么不直接在视图里写审计？ 因为数据变更可能发生在任何地方——视图、管理命令、Celery 任务、甚至 manage.py shell。信号 + 中间件的组合确保了无死角审计。

6.1 SimpleJWT 配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# settings.py
from datetime import timedelta

SIMPLE_JWT = {
    # Access Token：60 分钟有效，用于 API 调用
    "ACCESS_TOKEN_LIFETIME": timedelta(minutes=60),

    # Refresh Token：7 天有效，用于无感续期
    "REFRESH_TOKEN_LIFETIME": timedelta(days=7),

    # 每次刷新时轮换 Refresh Token（旧的立即失效）
    # 这意味着 Refresh Token 是一次性的，被盗后窗口期很短
    "ROTATE_REFRESH_TOKENS": True,

    # 轮换后将旧 Token 加入黑名单
    "BLACKLIST_AFTER_ROTATION": True,

    # 使用 RS256 或 HS256 签名算法
    "ALGORITHM": "HS256",
    "SIGNING_KEY": SECRET_KEY,
}

6.2 Token 生命周期图

1
2
3
4
5
6
7
8
9
10
11
用户登录
  │
  ├─ 颁发 Access Token  ──── 有效 60 分钟 ──── 过期
  │                                                │
  └─ 颁发 Refresh Token ──── 有效 7 天     用 Refresh 换新 Token
                                                    │
                                            ┌───────┴───────┐
                                            │ 新 Access (60m)│
                                            │ 新 Refresh (7d)│
                                            └────────────────┘
                                            旧 Refresh → 黑名单

6.3 CookieJWTAuthentication

在开发环境中，前端（端口 3000）和后端（端口 8000）跨端口部署，Authorization 头在某些场景下不方便传递。我们实现了 Cookie 回退方案：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from rest_framework_simplejwt.authentication import JWTAuthentication

class CookieJWTAuthentication(JWTAuthentication):
    """
    从 Cookie 中读取 JWT，作为 Header 认证的补充。
    主要用于：
    1. 开发环境前后端跨端口
    2. 服务端渲染（SSR）页面
    """
    def authenticate(self, request):
        # 优先检查标准 Authorization header
        header = self.get_header(request)
        if header:
            return super().authenticate(request)

        # 回退到 Cookie
        raw_token = request.COOKIES.get("access_token")
        if raw_token is None:
            return None

        validated_token = self.get_validated_token(raw_token)
        return self.get_user(validated_token), validated_token

登录接口在返回 JSON 的同时，也会设置 HttpOnly Cookie：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
class LoginView(TokenObtainPairView):
    def post(self, request, *args, **kwargs):
        response = super().post(request, *args, **kwargs)
        if response.status_code == 200:
            # 同时以 Cookie 形式设置 Token
            response.set_cookie(
                "access_token",
                response.data["access"],
                httponly=True,       # JS 无法读取，防 XSS
                samesite="Lax",      # 防 CSRF
                secure=not DEBUG,    # 生产环境强制 HTTPS
                max_age=3600,        # 与 Access Token 有效期一致
            )
        return response

七、服务层设计

视图层应该”薄”——只负责参数校验和响应格式化。真正的业务逻辑封装在 services/ 目录下：

1
2
3
4
5
6
services/
├── cert_service.py     # 合格证业务逻辑（核心）
├── signing.py          # Ed25519 数字签名（第四篇详解）
├── pdf_service.py      # PDF 合格证生成
├── qr_service.py       # 二维码图片生成
└── wechat_service.py   # 微信 OAuth 集成

7.1 cert_service.py —— 业务逻辑中枢

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
# services/cert_service.py
from django.db import transaction

def create_cargo(product_id, quantity, actor):
    """创建货物批次，自动生成批次号和扫码令牌"""
    product = Product.objects.get(pk=product_id)
    cargo = Cargo.objects.create(
        product=product,
        quantity=quantity,
        cargo_code=_new_cargo_code(),
        token=_new_token(),
        created_by=actor,
    )
    return cargo

@transaction.atomic
def bind_certificate(cargo_id, cert_data, actor):
    """
    将合格证绑定到货物批次。
    使用数据库事务确保：要么全部成功，要么全部回滚。
    """
    cargo = Cargo.objects.select_for_update().get(pk=cargo_id)
    cert = Certificate.objects.create(
        cargo=cargo,
        cert_number=_new_cert_number(),
        created_by=actor,
        **cert_data,
    )
    return cert

def submit_for_review(cert, actor):
    """提交审核：草稿 → 待审核"""
    if cert.publish_status != "draft":
        raise ValueError("只有草稿状态的证书可以提交审核")
    cert.publish_status = "pending_review"
    cert.save()

@transaction.atomic
def approve(cert, actor):
    """
    审核通过：
    1. 状态变更：待审核 → 已发布
    2. 生成 Ed25519 数字签名
    3. 生成 PDF 合格证文件
    4. 记录审计日志
    """
    if cert.publish_status != "pending_review":
        raise ValueError("只有待审核状态的证书可以审批")

    cert.publish_status = "published"
    cert.approved_by = actor
    cert.approved_at = timezone.now()

    # 生成数字签名（详见第四篇）
    cert.signature = signing.sign_certificate(cert)

    cert.save()

    # 异步生成 PDF（不阻塞响应）
    generate_pdf_task.delay(cert.pk)

def revoke(cert, actor, reason=""):
    """吊销证书：已发布 → 已吊销"""
    if cert.publish_status != "published":
        raise ValueError("只有已发布的证书可以吊销")
    cert.publish_status = "revoked"
    cert.revoke_reason = reason
    cert.save()

def reissue(cert, actor):
    """补发证书：基于已吊销的证书创建新草稿，继承核心字段"""
    if cert.publish_status != "revoked":
        raise ValueError("只有已吊销的证书可以补发")
    new_cert = Certificate.objects.create(
        cargo=cert.cargo,
        cert_number=_new_cert_number(),
        product=cert.product,
        inspection_org=cert.inspection_org,
        standard_code=cert.standard_code,
        created_by=actor,
        reissued_from=cert,   # 追溯关系
        publish_status="draft",
    )
    return new_cert

7.2 其他服务模块概览

pdf_service.py —— 使用 ReportLab 生成 PDF 格式的合格证：

1
2
3
4
5
6
7
8
9
10
from reportlab.lib.pagesizes import A4
from reportlab.pdfgen import canvas

def generate_certificate_pdf(cert):
    """生成 A4 尺寸的合格证 PDF，包含二维码和数字签名信息"""
    buffer = BytesIO()
    c = canvas.Canvas(buffer, pagesize=A4)
    # ... 布局逻辑：表头、产品信息、检验信息、二维码、签名摘要
    c.save()
    return buffer

qr_service.py —— 生成扫码查询用的二维码：

1
2
3
4
5
6
7
8
9
10
11
import qrcode

def generate_qr_image(scan_url):
    """
    生成二维码 PNG 图片
    scan_url 格式：https://example.com/scan/{token}
    """
    qr = qrcode.make(scan_url, error_correction=qrcode.constants.ERROR_CORRECT_H)
    buffer = BytesIO()
    qr.save(buffer, format="PNG")
    return buffer

wechat_service.py —— 微信扫码登录 OAuth 集成：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def get_wechat_auth_url(redirect_uri):
    """生成微信授权跳转 URL"""
    params = {
        "appid": settings.WECHAT_APP_ID,
        "redirect_uri": redirect_uri,
        "response_type": "code",
        "scope": "snsapi_userinfo",
    }
    return f"https://open.weixin.qq.com/connect/oauth2/authorize?{urlencode(params)}"

def exchange_code_for_user(code):
    """用授权码换取用户信息"""
    # 1. code → access_token + openid
    # 2. access_token + openid → 用户昵称、头像等
    ...

总结

本篇覆盖了后端架构的七个核心模块：

模块 核心思想
环境配置 APP_ENV + django-environ，零硬编码
数据模型 FROZEN_FIELDS 冻结 + 部分唯一约束 + 仅追加审计
视图层 ViewSet + 自定义 action + 认证级联
权限体系 三角色（superadmin / admin / operator）+ 双控审批
中间件审计 threading.local() + ORM 信号 = 无死角审计
JWT 认证 双通道（Header + Cookie）+ Token 轮换
服务层 薄视图 + 厚服务，事务保护关键操作

整个后端的设计哲学可以归纳为：在尽可能低的层级（模型层、数据库层）执行业务规则，让上层代码想犯错都难。

下一篇预告：[第三篇] React + Ant Design 前端架构详解——我们将深入前端的路由设计、状态管理、组件拆分，以及如何与后端 API 优雅对接。敬请期待！