feat: added ApiAuditLog

api_v1/admin.py

@@ -1,6 +1,6 @@
 from django.contrib import admin
 from django.contrib.admin import action
-from api_v1.models import UploadedFile, DataSync, MDYPlateOrderStaging
+from api_v1.models import UploadedFile, DataSync, MDYPlateOrderStaging, ApiAuditLog
 from .tasks import backup_database
 
 
@@ -41,3 +41,21 @@ class MDYPlateOrderStagingAdmin(admin.ModelAdmin):
     readonly_fields = ['created_at', 'updated_at']
     date_hierarchy = 'created_at'
     ordering = ['-created_at']
+
+
+@admin.register(ApiAuditLog)
+class ApiAuditLogAdmin(admin.ModelAdmin):
+    """API审计日志管理"""
+    list_display = ['id', 'method', 'url', 'username', 'user_id', 'response_status', 'created_at']
+    list_filter = ['method', 'response_status', 'created_at']
+    search_fields = ['url', 'username']
+    readonly_fields = ['url', 'method', 'request_data', 'query_params', 'user_id', 'username', 'response_status', 'created_at']
+    date_hierarchy = 'created_at'
+    ordering = ['-created_at']
+    
+    # 禁止添加和修改，只允许查看和删除
+    def has_add_permission(self, request):
+        return False
+    
+    def has_change_permission(self, request, obj=None):
+        return False

api_v1/migrations/0009_apiauditlog.py

@@ -0,0 +1,34 @@
+# Generated by Django 5.2.8 on 2026-01-12 03:16
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('api_v1', '0008_alter_datasync_table_name'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='ApiAuditLog',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('url', models.CharField(help_text='完整的请求路径', max_length=500, verbose_name='请求URL')),
+                ('method', models.CharField(help_text='HTTP方法，如POST', max_length=10, verbose_name='请求方法')),
+                ('request_data', models.JSONField(default=dict, help_text='POST请求的body数据（JSON格式）', verbose_name='请求体数据')),
+                ('query_params', models.JSONField(default=dict, help_text='URL查询参数', verbose_name='查询参数')),
+                ('user_id', models.IntegerField(blank=True, db_index=True, help_text='操作用户的ID', null=True, verbose_name='用户ID')),
+                ('username', models.CharField(blank=True, default='', help_text='操作用户的用户名快照', max_length=150, verbose_name='用户名')),
+                ('response_status', models.IntegerField(blank=True, help_text='HTTP响应状态码', null=True, verbose_name='响应状态码')),
+                ('created_at', models.DateTimeField(auto_now_add=True, db_index=True, verbose_name='创建时间')),
+            ],
+            options={
+                'verbose_name': 'API审计日志',
+                'verbose_name_plural': 'API审计日志',
+                'db_table': 'api_audit_log',
+                'ordering': ['-created_at'],
+                'indexes': [models.Index(fields=['user_id', 'created_at'], name='api_audit_l_user_id_b15ad2_idx'), models.Index(fields=['url', 'created_at'], name='api_audit_l_url_338a46_idx')],
+            },
+        ),
+    ]

api_v1/models.py

@@ -113,12 +113,12 @@ class DataSync(ModelBase):
 
 
 class MDYPlateOrderStaging(ModelBase):
-    """明道云“开版数据表”同步暂存表（不影响 printing.PlateOrder）。
+    """明道云"开版数据表"同步暂存表（不影响 printing.PlateOrder）。
 
     设计目标：
     - 与现有业务模型完全隔离，避免影响业务数据与流程
     - 仅保留 rowid 便于检索排查，其余信息全部落在 raw(JSON) 中
-    - 随着同步深入再逐步“属性化”字段（从 raw 实时提取/映射）
+    - 随着同步深入再逐步"属性化"字段（从 raw 实时提取/映射）
     """
 
     mdy_rowid = models.CharField(
@@ -161,7 +161,7 @@ class MDYPlateOrderStaging(ModelBase):
         verbose_name_plural = '明道云开版暂存'
         ordering = ['-created_at']
         indexes = [
-            # 明道云字段：62d52f4b8d2972284492dd0e（开版表“订单id/设计编号”）
+            # 明道云字段：62d52f4b8d2972284492dd0e（开版表"订单id/设计编号"）
             # 用于在暂存表中按该字段快速检索
             models.Index(
                 KeyTextTransform('62d52f4b8d2972284492dd0e', 'raw'),
@@ -171,3 +171,69 @@ class MDYPlateOrderStaging(ModelBase):
 
     def __str__(self):
         return f'{self.mdy_rowid}'
+
+
+class ApiAuditLog(models.Model):
+    """API审计日志 - 记录创建操作的历史现场
+    
+    用于保存特定模块的POST请求，记录"创建"操作的完整请求信息。
+    通过Celery异步写入，避免影响API响应性能。
+    """
+    url = models.CharField(
+        max_length=500,
+        verbose_name='请求URL',
+        help_text='完整的请求路径'
+    )
+    method = models.CharField(
+        max_length=10,
+        verbose_name='请求方法',
+        help_text='HTTP方法，如POST'
+    )
+    request_data = models.JSONField(
+        default=dict,
+        verbose_name='请求体数据',
+        help_text='POST请求的body数据（JSON格式）'
+    )
+    query_params = models.JSONField(
+        default=dict,
+        verbose_name='查询参数',
+        help_text='URL查询参数'
+    )
+    user_id = models.IntegerField(
+        null=True,
+        blank=True,
+        db_index=True,
+        verbose_name='用户ID',
+        help_text='操作用户的ID'
+    )
+    username = models.CharField(
+        max_length=150,
+        blank=True,
+        default='',
+        verbose_name='用户名',
+        help_text='操作用户的用户名快照'
+    )
+    response_status = models.IntegerField(
+        null=True,
+        blank=True,
+        verbose_name='响应状态码',
+        help_text='HTTP响应状态码'
+    )
+    created_at = models.DateTimeField(
+        auto_now_add=True,
+        db_index=True,
+        verbose_name='创建时间'
+    )
+    
+    class Meta:
+        db_table = 'api_audit_log'
+        verbose_name = 'API审计日志'
+        verbose_name_plural = 'API审计日志'
+        ordering = ['-created_at']
+        indexes = [
+            models.Index(fields=['user_id', 'created_at']),
+            models.Index(fields=['url', 'created_at']),
+        ]
+    
+    def __str__(self):
+        return f'{self.method} {self.url} ({self.username or "anonymous"}) - {self.response_status}'

api_v1/tasks.py

@@ -466,3 +466,42 @@ def sync_mdy_plate_orders(
     )
     payload["task_id"] = self.request.id
     return payload
+
+
+@shared_task
+def save_api_audit_log(
+    url: str,
+    method: str,
+    request_data: dict,
+    query_params: dict,
+    user_id: int | None,
+    username: str,
+    response_status: int | None,
+):
+    """
+    异步保存API审计日志。
+    
+    通过Celery队列异步执行，避免阻塞API响应。
+    
+    参数:
+        url: 请求URL路径
+        method: HTTP方法（POST等）
+        request_data: 请求体数据（JSON格式）
+        query_params: URL查询参数
+        user_id: 操作用户ID
+        username: 操作用户的用户名
+        response_status: HTTP响应状态码
+    """
+    try:
+        api_models.ApiAuditLog.objects.create(
+            url=url,
+            method=method,
+            request_data=request_data,
+            query_params=query_params,
+            user_id=user_id,
+            username=username,
+            response_status=response_status,
+        )
+        logger.debug('API审计日志已保存: %s %s (user=%s, status=%s)', method, url, username, response_status)
+    except Exception as e:
+        logger.error('保存API审计日志失败: %s', str(e), exc_info=True)

docs/2026-01-11_summary.md

@@ -0,0 +1,70 @@
+# 2026-01-11 工作日志
+
+## 完成事项
+
+### API 审计日志中间件
+
+实现了一个用于记录特定模块POST请求的中间件，用于保存"创建"操作的历史现场。
+
+#### 1. 新增 `ApiAuditLog` 模型 (`api_v1/models.py`)
+
+记录以下信息：
+- `url`: 请求URL路径
+- `method`: HTTP方法
+- `request_data`: 请求体数据（JSON格式）
+- `query_params`: URL查询参数
+- `user_id`: 操作用户ID
+- `username`: 操作用户的用户名快照
+- `response_status`: HTTP响应状态码
+- `created_at`: 创建时间
+
+#### 2. 新增 Celery Task (`api_v1/tasks.py`)
+
+- `save_api_audit_log`: 异步保存审计日志任务
+- 通过Celery队列异步执行，避免阻塞API响应
+
+#### 3. 新增中间件 (`flower/middleware.py`)
+
+- `ApiAuditLogMiddleware`: API审计日志中间件
+- 只记录POST请求
+- 通过URL前缀白名单过滤
+- 支持multipart/form-data请求，文件字段只记录元信息（文件名、大小、类型）
+- 在AuthenticationMiddleware之后运行，可获取request.user
+- 对于JWT认证，手动调用JWTAuthentication获取用户信息
+
+#### 4. 配置更新 (`flower/settings.py`)
+
+新增配置项：
+```python
+# API 审计日志配置
+API_AUDIT_LOG_ENABLED = True
+API_AUDIT_LOG_URL_PREFIXES = [
+    '/api/v1/',
+    '/api/v2/',
+    '/api/backend/',
+]
+```
+
+中间件注册：
+```python
+MIDDLEWARE = [
+    ...
+    'flower.middleware.ApiAuditLogMiddleware',  # 放在最后
+]
+```
+
+#### 5. 数据库迁移
+
+- 迁移文件: `api_v1/migrations/0009_apiauditlog.py`
+
+#### 6. Admin 注册 (`api_v1/admin.py`)
+
+- 新增 `ApiAuditLogAdmin` 管理类
+- 支持按 URL、用户名搜索
+- 支持按方法、响应状态码、时间过滤
+- 所有字段只读，禁止添加和修改（只允许查看和删除）
+
+## 待办事项
+
+- [ ] 执行数据库迁移 (`uv run python manage.py migrate`)
+- [ ] 测试中间件功能

flower/middleware.py

@@ -0,0 +1,191 @@
+"""
+自定义中间件模块
+"""
+import json
+import logging
+
+from django.conf import settings
+from rest_framework_simplejwt.authentication import JWTAuthentication
+
+logger = logging.getLogger(__name__)
+
+
+class ApiAuditLogMiddleware:
+    """
+    API审计日志中间件
+    
+    用于记录特定URL前缀的POST请求，保存"创建"操作的历史现场。
+    
+    特性：
+    - 只记录POST请求
+    - 通过URL前缀白名单过滤
+    - 支持multipart/form-data请求，文件字段只记录元信息
+    - 通过Celery异步写入，不阻塞API响应
+    - 在AuthenticationMiddleware之后运行，可获取request.user
+    
+    配置项（settings.py）：
+    - API_AUDIT_LOG_ENABLED: 是否启用（默认True）
+    - API_AUDIT_LOG_URL_PREFIXES: URL前缀白名单列表
+    """
+    
+    def __init__(self, get_response):
+        self.get_response = get_response
+        self.enabled = getattr(settings, 'API_AUDIT_LOG_ENABLED', True)
+        self.url_prefixes = getattr(settings, 'API_AUDIT_LOG_URL_PREFIXES', [])
+    
+    def __call__(self, request):
+        # 检查是否启用
+        if not self.enabled:
+            return self.get_response(request)
+        
+        # 只处理POST请求
+        if request.method != 'POST':
+            return self.get_response(request)
+        
+        # 检查URL是否在白名单中
+        if not self._is_url_matched(request.path):
+            return self.get_response(request)
+        
+        # 在请求进入视图前，缓存请求体数据
+        # 注意：request.body只能读取一次，需要在这里缓存
+        request_data = self._extract_request_data(request)
+        query_params = dict(request.GET)
+        
+        # 执行视图
+        response = self.get_response(request)
+        
+        # 获取用户信息（此时DRF的JWT认证已完成）
+        user_id, username = self._get_user_info(request)
+        
+        # 异步保存审计日志
+        self._save_audit_log(
+            url=request.path,
+            method=request.method,
+            request_data=request_data,
+            query_params=query_params,
+            user_id=user_id,
+            username=username,
+            response_status=response.status_code,
+        )
+        
+        return response
+    
+    def _is_url_matched(self, path: str) -> bool:
+        """检查URL是否匹配白名单前缀"""
+        for prefix in self.url_prefixes:
+            if path.startswith(prefix):
+                return True
+        return False
+    
+    def _extract_request_data(self, request) -> dict:
+        """
+        提取请求数据
+        
+        对于multipart/form-data请求，分别处理表单字段和文件字段。
+        文件字段只记录元信息（文件名、大小、类型），不记录二进制内容。
+        """
+        content_type = request.content_type or ''
+        
+        if 'multipart/form-data' in content_type:
+            # multipart请求：分别处理表单字段和文件
+            data = {}
+            
+            # 处理普通表单字段
+            for key, values in request.POST.lists():
+                if len(values) == 1:
+                    data[key] = values[0]
+                else:
+                    data[key] = values
+            
+            # 处理文件字段：只记录元信息
+            for key, files in request.FILES.lists():
+                file_infos = []
+                for f in files:
+                    file_infos.append({
+                        '_type': 'file',
+                        'name': f.name,
+                        'size': f.size,
+                        'content_type': f.content_type,
+                    })
+                if len(file_infos) == 1:
+                    data[key] = file_infos[0]
+                else:
+                    data[key] = file_infos
+            
+            return data
+        
+        elif 'application/json' in content_type:
+            # JSON请求：解析body
+            try:
+                return json.loads(request.body.decode('utf-8'))
+            except (json.JSONDecodeError, UnicodeDecodeError):
+                return {'_raw': request.body.decode('utf-8', errors='replace')}
+        
+        elif 'application/x-www-form-urlencoded' in content_type:
+            # 表单请求
+            data = {}
+            for key, values in request.POST.lists():
+                if len(values) == 1:
+                    data[key] = values[0]
+                else:
+                    data[key] = values
+            return data
+        
+        else:
+            # 其他类型：尝试作为文本记录
+            try:
+                body = request.body.decode('utf-8')
+                if body:
+                    return {'_raw': body}
+                return {}
+            except UnicodeDecodeError:
+                return {'_raw': '<binary data>'}
+    
+    def _get_user_info(self, request) -> tuple[int | None, str]:
+        """
+        获取用户信息
+        
+        优先从request.user获取（Session认证或DRF已设置）。
+        如果未认证，尝试手动进行JWT认证。
+        """
+        # 尝试从request.user获取
+        if hasattr(request, 'user') and request.user.is_authenticated:
+            return request.user.id, request.user.username
+        
+        # 尝试手动JWT认证
+        try:
+            auth = JWTAuthentication()
+            user_auth_tuple = auth.authenticate(request)
+            if user_auth_tuple:
+                user = user_auth_tuple[0]
+                return user.id, user.username
+        except Exception:
+            pass
+        
+        return None, ''
+    
+    def _save_audit_log(
+        self,
+        url: str,
+        method: str,
+        request_data: dict,
+        query_params: dict,
+        user_id: int | None,
+        username: str,
+        response_status: int | None,
+    ):
+        """通过Celery异步保存审计日志"""
+        try:
+            from api_v1.tasks import save_api_audit_log
+            save_api_audit_log.delay(
+                url=url,
+                method=method,
+                request_data=request_data,
+                query_params=query_params,
+                user_id=user_id,
+                username=username,
+                response_status=response_status,
+            )
+        except Exception as e:
+            # 日志保存失败不应影响正常请求
+            logger.error('发送API审计日志任务失败: %s', str(e), exc_info=True)

flower/settings.py

@@ -137,6 +137,7 @@ MIDDLEWARE = [
     'django.contrib.auth.middleware.AuthenticationMiddleware',
     'django.contrib.messages.middleware.MessageMiddleware',
     'django.middleware.clickjacking.XFrameOptionsMiddleware',
+    'flower.middleware.ApiAuditLogMiddleware',  # API审计日志中间件，需在AuthenticationMiddleware之后
 ]
 
 REST_FRAMEWORK = {
@@ -401,6 +402,18 @@ STATEFLOW_CURRENT_STATE_MODE = 'NEXT'
 PRINTING_DEFAULT_PROCESS_ID = 1  # 默认印染流程ID
 PLATE_ORDER_DEFAULT_PROCESS_ID = 2  # 默认开版流程ID
 
+# API 审计日志配置
+# ------------------------------------------------------------------------------
+# 用于记录特定模块POST请求的历史现场，便于追踪"创建"操作。
+# 通过Celery异步写入，不影响API响应性能。
+# ------------------------------------------------------------------------------
+API_AUDIT_LOG_ENABLED = True
+API_AUDIT_LOG_URL_PREFIXES = [
+    '/api/v1/printing',
+    '/api/v1/plate',
+    # '/api/v2/',
+]
+
 # Celery 配置
 # RabbitMQ Broker 配置：优先使用 CELERY_BROKER_URL，否则从各组件构建
 # 默认 localhost（适合宿主机运行），Docker Compose 可通过环境变量指定 CELERY_BROKER_HOST=rabbitmq