feat(swagger): generate localized static API schema

apps/bi/serializers.py

@@ -18,10 +18,35 @@ class DatasetCreateUpdateSerializer(CustomModelSerializer):
 
 
 class DatasetSerializer(CustomModelSerializer):
+    description = serializers.CharField(
+        label="适用场景与统计口径",
+        help_text="说明该数据集适合回答的问题、指标口径、参数格式和返回字段含义",
+        required=False,
+        allow_blank=True,
+    )
+    default_param = serializers.JSONField(
+        label="默认查询参数",
+        help_text="执行时可覆盖的参数及默认值；内部 SQL 片段参数应保留默认值",
+        required=False,
+    )
+    test_param = serializers.JSONField(
+        label="测试查询参数",
+        help_text="数据集维护时使用的示例参数，普通查询优先参考 description",
+        required=False,
+    )
+
     class Meta:
         model = Dataset
         fields = '__all__'
 
+
+class DatasetListResponseSerializer(serializers.Serializer):
+    count = serializers.IntegerField(label="数据集总数")
+    next = serializers.URLField(required=False, allow_null=True)
+    previous = serializers.URLField(required=False, allow_null=True)
+    results = DatasetSerializer(many=True)
+
+
 class DatasetRecordSerializer(CustomModelSerializer):
     class Meta:
         model = DatasetRecord
@@ -36,6 +61,14 @@ class DatasetRecordSerializer(CustomModelSerializer):
 
 class DataExecSerializer(serializers.Serializer):
     query = serializers.JSONField(
-        label="查询字典参数", required=False, allow_null=True)
+        label="查询字典参数",
-    is_test = serializers.BooleanField(label='是否测试', default=False)
+        help_text="按所选数据集 description/default_param 声明的业务参数填写",
-    raise_exception = serializers.BooleanField(label='是否直接报错', default=False)
+        required=False,
+        allow_null=True,
+    )
+    is_test = serializers.BooleanField(
+        label='是否测试', help_text="普通业务查询固定为 false", default=False
+    )
+    raise_exception = serializers.BooleanField(
+        label='是否直接报错', help_text="建议为 true，便于修正缺失或非法参数", default=True
+    )

apps/bi/tests.py

@@ -1,3 +1,25 @@
-from django.test import TestCase
+from django.test import SimpleTestCase
 
-# Create your tests here.
+from apps.bi.serializers import (
+    DataExecSerializer,
+    DatasetListResponseSerializer,
+    DatasetSerializer,
+)
+from apps.bi.views import DatasetViewSet
+
+
+class DatasetAgentDiscoveryTests(SimpleTestCase):
+    def test_dataset_catalog_searches_description(self):
+        self.assertEqual(
+            DatasetViewSet.search_fields,
+            ["name", "code", "description"],
+        )
+
+    def test_dataset_schema_explains_discovery_and_exec_parameters(self):
+        dataset = DatasetSerializer()
+        execute = DataExecSerializer()
+
+        self.assertIn("适合回答的问题", dataset.fields["description"].help_text)
+        self.assertIn("业务参数", execute.fields["query"].help_text)
+        self.assertTrue(execute.fields["raise_exception"].default)
+        self.assertIn("results", DatasetListResponseSerializer().fields)

apps/bi/views.py

@@ -3,7 +3,13 @@ from apps.utils.viewsets import CustomModelViewSet, CustomGenericViewSet
 from rest_framework.decorators import action
 from rest_framework.response import Response
 from apps.bi.models import Dataset, DatasetRecord
-from apps.bi.serializers import DatasetSerializer, DatasetCreateUpdateSerializer, DataExecSerializer, DatasetRecordSerializer
+from apps.bi.serializers import (
+    DataExecSerializer,
+    DatasetCreateUpdateSerializer,
+    DatasetListResponseSerializer,
+    DatasetRecordSerializer,
+    DatasetSerializer,
+)
 from django.apps import apps
 import concurrent.futures
 from django.core.cache import cache
@@ -13,6 +19,8 @@ from rest_framework.exceptions import ParseError
 from rest_framework.generics import get_object_or_404
 from apps.utils.mixins import ListModelMixin
 import logging
+from drf_yasg import openapi
+from drf_yasg.utils import swagger_auto_schema
 myLogger = logging.getLogger('log')
 # Create your views here.
 
@@ -22,9 +30,54 @@ class DatasetViewSet(CustomModelViewSet):
     serializer_class = DatasetSerializer
     create_serializer_class = DatasetCreateUpdateSerializer
     update_serializer_class = DatasetCreateUpdateSerializer
-    search_fields = ['name', 'code']
+    search_fields = ['name', 'code', 'description']
     ordering = ['name', 'code', 'id']
 
+    @swagger_auto_schema(
+        operation_id="bi_dataset_list",
+        operation_summary="查询复杂统计报表的数据集目录",
+        operation_description=(
+            "产量、良率、缺陷、库存、绩效、趋势和按日/月汇总等统计聚合查询的统一入口。"
+            "先调用本接口，根据 name、description、default_param 和 test_param 选择数据集，"
+            "再调用 bi_dataset_exec。建议使用 query={id,name,code,description,default_param,"
+            "test_param,enabled} 裁剪字段，并设置 page_size=100 查看完整目录；"
+            "search 可按名称、code 或 description 检索。"
+        ),
+        manual_parameters=[
+            openapi.Parameter(
+                "search",
+                openapi.IN_QUERY,
+                description="按数据集名称、code 或适用场景关键词检索",
+                type=openapi.TYPE_STRING,
+            ),
+            openapi.Parameter(
+                "page",
+                openapi.IN_QUERY,
+                description="页码，从 1 开始",
+                type=openapi.TYPE_INTEGER,
+            ),
+            openapi.Parameter(
+                "page_size",
+                openapi.IN_QUERY,
+                description="每页数量；当前目录建议传 100",
+                type=openapi.TYPE_INTEGER,
+            ),
+            openapi.Parameter(
+                "query",
+                openapi.IN_QUERY,
+                description=(
+                    "django-restql 字段裁剪表达式，例如 "
+                    "{id,name,code,description,default_param,test_param,enabled}"
+                ),
+                type=openapi.TYPE_STRING,
+            ),
+        ],
+        responses={200: DatasetListResponseSerializer},
+        tags=["BI 数据集与报表"],
+    )
+    def list(self, request, *args, **kwargs):
+        return super().list(request, *args, **kwargs)
+
     def get_object(self):
         """
         Returns the object the view is displaying.
@@ -57,6 +110,18 @@ class DatasetViewSet(CustomModelViewSet):
 
         return obj
 
+    @swagger_auto_schema(
+        operation_id="bi_dataset_exec",
+        operation_summary="执行已配置的只读统计数据集",
+        operation_description=(
+            "使用 dataset list 返回的 id 或 code 执行数据集。body.query 只填写该数据集"
+            "description/default_param 声明的业务参数；正常查询设置 is_test=false。"
+            "统计聚合使用本接口，日志和业务明细列表用于逐条追溯。"
+        ),
+        request_body=DataExecSerializer,
+        responses={200: DatasetSerializer},
+        tags=["BI 数据集与报表"],
+    )
     @action(methods=['post'], detail=True, perms_map={'post': 'dataset.exec'}, serializer_class=DataExecSerializer, cache_seconds=0, logging_methods=[])
     def exec(self, request, pk=None):
         """执行sql查询

apps/edu/views.py

@@ -71,6 +71,8 @@ class ExamViewSet(CustomModelViewSet):
 
     def get_queryset(self):
         qs = super().get_queryset()
+        if getattr(self, 'swagger_fake_view', False):
+            return qs
         if has_perm(self.request.user, ["exam.view"]):
             return qs
         user:User = self.request.user
@@ -142,6 +144,8 @@ class ExamRecordViewSet(ListModelMixin, DestroyModelMixin, RetrieveModelMixin, C
 
     def get_queryset(self):
         qs = super().get_queryset()
+        if getattr(self, 'swagger_fake_view', False):
+            return qs
         if has_perm(self.request.user, ["examrecord.view"]):
             return qs
         return qs.filter(create_by=self.request.user)
@@ -207,6 +211,8 @@ class TrainRecordViewSet(CustomModelViewSet):
 
     def get_queryset(self):
         qs = super().get_queryset()
+        if getattr(self, 'swagger_fake_view', False):
+            return qs
         if has_perm(self.request.user, ["train.view"]):
             return qs
         return qs.filter(create_by=self.request.user)

apps/system/views.py

@@ -651,6 +651,7 @@ class FileViewSet(BulkCreateModelMixin, RetrieveModelMixin, CustomListModelMixin
 class ApkViewSet(MyLoggingMixin, CustomListModelMixin, BulkCreateModelMixin, GenericViewSet):
     perms_map = {'get': '*', 'post': 'apk.upload'}
     serializer_class = ApkSerializer
+    filter_backends = []
 
     def get_authenticators(self):
         if self.request.method == 'GET':

apps/third/views.py

@@ -69,6 +69,7 @@ class SpeakerViewSet(CustomGenericViewSet):
     """
     perms_map = {}
     serializer_class = serializers.Serializer
+    filter_backends = []
 
     @action(methods=['get'], detail=False,
             permission_classes=[IsAuthenticated])
@@ -125,6 +126,7 @@ class XxTestView(APIView):
 class XxCommonViewSet(CreateModelMixin, CustomGenericViewSet):
     perms_map = {'post': '*'}
     serializer_class = RequestCommonSerializer
+    filter_backends = []
 
     def create(self, request, *args, **kwargs):
         """
@@ -258,6 +260,7 @@ class KingCommonViewSet(CreateModelMixin, CustomGenericViewSet):
 class DhCommonViewSet(CreateModelMixin, CustomGenericViewSet):
     perms_map = {'post': '*'}
     serializer_class = RequestCommonSerializer
+    filter_backends = []
 
     def create(self, request, *args, **kwargs):
         """

apps/utils/management/commands/build_swagger.py

@@ -0,0 +1,71 @@
+import json
+import os
+from io import StringIO
+from pathlib import Path
+
+from django.conf import settings
+from django.core.management import BaseCommand, CommandError, call_command
+
+
+class Command(BaseCommand):
+    help = "生成供 Swagger UI 和 ReDoc 使用的静态 Swagger JSON"
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--output",
+            help="输出路径，默认使用 settings.SWAGGER_SCHEMA_PATH",
+        )
+        parser.add_argument(
+            "--url",
+            help="文档中的 API 根地址，默认使用 settings.BASE_URL",
+        )
+
+    def handle(self, *args, **options):
+        target = Path(options["output"] or settings.SWAGGER_SCHEMA_PATH)
+        if not target.is_absolute():
+            target = Path(settings.BASE_DIR) / target
+        target = target.resolve()
+        target.parent.mkdir(parents=True, exist_ok=True)
+
+        temporary = target.with_name(f".{target.name}.{os.getpid()}.tmp")
+        try:
+            output = StringIO()
+            call_command(
+                "generate_swagger",
+                "-",
+                format="json",
+                api_url=options["url"] or settings.BASE_URL,
+                mock=True,
+                verbosity=0,
+                stdout=output,
+            )
+            content = output.getvalue()
+            schema = json.loads(content)
+            if schema.get("swagger") != "2.0" or not schema.get("paths"):
+                raise CommandError("生成的 Swagger 文档缺少版本或接口路径")
+
+            content = json.dumps(
+                schema,
+                ensure_ascii=False,
+                separators=(",", ":"),
+            )
+            temporary.write_text(content, encoding="utf-8")
+            os.replace(temporary, target)
+        except Exception as exc:
+            if isinstance(exc, CommandError):
+                raise
+            raise CommandError(f"生成 Swagger 文档失败：{exc}") from exc
+        finally:
+            temporary.unlink(missing_ok=True)
+
+        operation_count = sum(
+            method.lower() in {"get", "post", "put", "patch", "delete"}
+            for path in schema["paths"].values()
+            for method in path
+        )
+        self.stdout.write(
+            self.style.SUCCESS(
+                f"Swagger文档已生成：{target} "
+                f"（{len(schema['paths'])}个路径，{operation_count}个操作）"
+            )
+        )

apps/utils/mixins.py

@@ -207,12 +207,28 @@ class CustomRetrieveModelMixin(RetrieveModelMixin):
 
 class CustomListModelMixin(ListModelMixin):
     
-    @swagger_auto_schema(manual_parameters=[
+    @swagger_auto_schema(
-        openapi.Parameter(name="query", in_=openapi.IN_QUERY, description="定制返回数据",
+        operation_description=(
-                          type=openapi.TYPE_STRING, required=False),
+            "通用列表接口用于记录或目录浏览以及逐条追溯。跨时间范围的产量、良率、缺陷、"
-        openapi.Parameter(name="with_children", in_=openapi.IN_QUERY, description="带有children(yes/no/count)",
+            "库存、绩效和趋势等统计聚合，优先查询 BI dataset 目录并执行匹配的数据集。"
-                          type=openapi.TYPE_STRING, required=False),
+        ),
-    ])
+        manual_parameters=[
+            openapi.Parameter(
+                name="query",
+                in_=openapi.IN_QUERY,
+                description="django-restql 返回字段裁剪表达式",
+                type=openapi.TYPE_STRING,
+                required=False,
+            ),
+            openapi.Parameter(
+                name="with_children",
+                in_=openapi.IN_QUERY,
+                description="带有children(yes/no/count)",
+                type=openapi.TYPE_STRING,
+                required=False,
+            ),
+        ],
+    )
     def list(self, request, *args, **kwargs):
 
         queryset = self.filter_queryset(self.get_queryset())

apps/utils/swagger.py

@@ -0,0 +1,367 @@
+import re
+from pathlib import Path
+
+from django.apps import apps
+from django.conf import settings
+from django.http import FileResponse, JsonResponse
+from drf_yasg import openapi
+from drf_yasg.inspectors import FieldInspector, SwaggerAutoSchema
+from drf_yasg.inspectors.base import NotHandled
+
+
+CRUD_SUMMARIES = {
+    "list": "查询{resource}列表",
+    "retrieve": "查询{resource}详情",
+    "create": "新增{resource}",
+    "update": "更新{resource}",
+    "partial_update": "部分更新{resource}",
+    "destroy": "删除{resource}",
+}
+
+TAG_NAMES = {
+    "am": "区域与准入管理",
+    "asm": "资产管理",
+    "cm": "标签管理",
+    "cms": "内容管理",
+    "develop": "开发工具",
+    "ecm": "事件管理",
+    "edu": "培训考试",
+    "em": "设备管理",
+    "enm": "能源管理",
+    "inm": "库存管理",
+    "mpr": "物资申购与领用",
+    "mtm": "物料与工艺管理",
+    "ofm": "办公管理",
+    "opm": "作业许可",
+    "pm": "生产任务管理",
+    "pum": "采购管理",
+    "qm": "质量管理",
+    "rem": "研发项目管理",
+    "rpm": "相关方管理",
+    "sam": "销售管理",
+    "third": "第三方集成",
+    "utils": "通用工具",
+    "wpm": "生产管理",
+    "wpmw": "动态产品管理",
+    "file": "文件管理",
+}
+
+FIELD_NAMES = {
+    "id": "主键ID",
+    "ids": "主键ID列表",
+    "access": "访问令牌",
+    "refresh": "刷新令牌",
+    "password_check": "密码确认",
+    "base64": "Base64数据",
+    "server_time": "服务器时间",
+    "timezone": "时区",
+    "next": "下一页",
+    "previous": "上一页",
+    "results": "结果列表",
+    "detail": "详情",
+    "items": "明细列表",
+    "files": "附件列表",
+    "echart_options": "图表配置",
+    "tdata_list": "数据列表",
+    "page": "页码",
+    "page_size": "每页数量",
+    "ordering": "排序字段",
+    "querys": "查询条件列表",
+    "annotate_field_list": "聚合字段列表",
+}
+
+FIELD_TOKENS = {
+    "name": "名称",
+    "code": "编码",
+    "number": "编号",
+    "description": "说明",
+    "note": "备注",
+    "employee": "人员",
+    "user": "用户",
+    "leader": "负责人",
+    "manager": "负责人",
+    "keeper": "保管人",
+    "participant": "参与人",
+    "post": "岗位",
+    "dept": "部门",
+    "belong": "所属",
+    "create": "创建",
+    "update": "更新",
+    "submit": "提交",
+    "handle": "处理",
+    "test": "检验",
+    "material": "物料",
+    "supplier": "供应商",
+    "defect": "缺陷",
+    "equipment": "设备",
+    "warehouse": "仓库",
+    "process": "工序",
+    "operation": "操作",
+    "state": "状态",
+    "cate": "分类",
+    "type": "类型",
+    "area": "区域",
+    "team": "班组",
+    "shift": "班次",
+    "ticket": "工单",
+    "file": "文件",
+    "photo": "照片",
+    "image": "图片",
+    "origin": "来源",
+    "in": "入库",
+    "out": "出库",
+    "list": "列表",
+    "count": "数量",
+    "total": "总计",
+    "enabled": "是否启用",
+}
+
+QUERY_PARAMETERS = {
+    "page": "页码",
+    "page_size": "每页数量",
+    "search": "搜索关键字",
+    "ordering": "排序字段，字段名前加“-”表示倒序",
+    "format": "响应格式",
+}
+
+LOOKUP_NAMES = {
+    "in": "属于列表",
+    "contains": "包含",
+    "icontains": "包含（忽略大小写）",
+    "gte": "大于或等于",
+    "gt": "大于",
+    "lte": "小于或等于",
+    "lt": "小于",
+    "isnull": "是否为空",
+    "exact": "等于",
+}
+
+
+def _contains_chinese(value):
+    return bool(re.search(r"[\u4e00-\u9fff]", str(value or "")))
+
+
+def swagger_schema_file(request):
+    schema_path = Path(settings.SWAGGER_SCHEMA_PATH)
+    if not schema_path.is_file():
+        return JsonResponse(
+            {"detail": "Swagger文档尚未生成，请先运行 manage.py build_swagger"},
+            status=503,
+        )
+
+    response = FileResponse(
+        schema_path.open("rb"),
+        content_type="application/json; charset=utf-8",
+        filename="swagger.json",
+    )
+    response["Content-Disposition"] = 'inline; filename="swagger.json"'
+    response["Cache-Control"] = "no-cache"
+    return response
+
+
+def _serializer_model(field):
+    parent = getattr(field, "parent", None)
+    while parent is not None:
+        meta = getattr(parent, "Meta", None)
+        model = getattr(meta, "model", None)
+        if model is not None:
+            return model
+        parent = getattr(parent, "parent", None)
+    return None
+
+
+def _model_path_label(model, parts):
+    labels = []
+    for part in parts:
+        if model is None:
+            break
+        try:
+            model_field = model._meta.get_field(part)
+        except Exception:
+            break
+        verbose_name = getattr(model_field, "verbose_name", "")
+        if _contains_chinese(verbose_name):
+            labels.append(str(verbose_name))
+        model = getattr(model_field, "related_model", None)
+    return " / ".join(labels)
+
+
+def _field_name_label(field_name):
+    field_name = str(field_name or "").strip("_")
+    if field_name in FIELD_NAMES:
+        return FIELD_NAMES[field_name]
+    tokens = field_name.split("_")
+    if tokens and all(token in FIELD_TOKENS for token in tokens):
+        return "".join(FIELD_TOKENS[token] for token in tokens)
+    return ""
+
+
+class ChineseFieldInspector(FieldInspector):
+    """优先使用模型字段中文名称补全 serializer 字段标题。"""
+
+    def field_to_swagger_object(self, field, **kwargs):
+        return NotHandled
+
+    def process_result(self, result, method_name, obj, **kwargs):
+        if (
+            method_name != "field_to_swagger_object"
+            or not isinstance(result, openapi.SwaggerDict)
+            or "$ref" in result
+            or _contains_chinese(result.get("title"))
+        ):
+            return result
+
+        source_attrs = getattr(obj, "source_attrs", None) or []
+        model_label = _model_path_label(_serializer_model(obj), source_attrs)
+        label = model_label or _field_name_label(getattr(obj, "field_name", ""))
+        field_name = getattr(obj, "field_name", "")
+        if label:
+            result["title"] = label
+        elif field_name:
+            result["title"] = f"{field_name}（字段）"
+        return result
+
+
+class ChineseSwaggerAutoSchema(SwaggerAutoSchema):
+    """为未显式编写文档的接口补充稳定、可读的中文展示信息。"""
+
+    field_inspectors = [ChineseFieldInspector] + SwaggerAutoSchema.field_inspectors
+
+    def get_operation(self, operation_keys=None):
+        operation = super().get_operation(operation_keys)
+        model = getattr(getattr(self.view, "queryset", None), "model", None)
+        for parameter in operation.get("parameters", []):
+            current = parameter.get("description", "")
+            if _contains_chinese(current):
+                continue
+            location = parameter.get("in")
+            if location == openapi.IN_BODY:
+                description = "请求数据"
+            elif location == openapi.IN_PATH:
+                description = f"路径参数：{parameter.get('name', '')}"
+            else:
+                description = self._get_parameter_description(
+                    parameter.get("name", ""), model
+                )
+            if current:
+                description = f"{description}；{current}"
+            parameter["description"] = description
+        return operation
+
+    def get_summary_and_description(self):
+        summary, description = super().get_summary_and_description()
+        if summary:
+            if description and not _contains_chinese(description):
+                description = f"{summary}\n\n{description}"
+            return summary, description or summary
+
+        resource = self._get_resource_name()
+        action = getattr(self.view, "action", None)
+        template = CRUD_SUMMARIES.get(action)
+        if template:
+            summary = template.format(resource=resource)
+        elif resource:
+            action_name = str(action or self.method).replace("_", " ")
+            display_resource = resource
+            if not _contains_chinese(display_resource):
+                display_resource = f"{display_resource}接口"
+            summary = f"{display_resource}：{action_name}"
+
+        if description and not _contains_chinese(description):
+            description = f"{summary}\n\n{description}"
+        return summary, description or summary
+
+    def get_request_body_parameters(self, consumes):
+        parameters = super().get_request_body_parameters(consumes)
+        for parameter in parameters:
+            if parameter.get("in") == openapi.IN_BODY and not _contains_chinese(
+                parameter.get("description")
+            ):
+                parameter["description"] = "请求数据"
+        return parameters
+
+    def get_query_parameters(self):
+        parameters = super().get_query_parameters()
+        model = getattr(getattr(self.view, "queryset", None), "model", None)
+        for parameter in parameters:
+            current = parameter.get("description", "")
+            if _contains_chinese(current):
+                continue
+            description = self._get_parameter_description(
+                parameter.get("name", ""), model
+            )
+            if current:
+                description = f"{description}；{current}"
+            parameter["description"] = description
+        return parameters
+
+    def get_responses(self):
+        responses = super().get_responses()
+        descriptions = {
+            "200": "请求成功",
+            "201": "创建成功",
+            "202": "请求已接受",
+            "204": "操作成功，无响应内容",
+            "400": "请求参数错误",
+            "401": "未认证或认证已失效",
+            "403": "无权访问",
+            "404": "资源不存在",
+        }
+        for status, response in responses.items():
+            if not response.get("description"):
+                response["description"] = descriptions.get(str(status), "接口响应")
+        return responses
+
+    def _get_parameter_description(self, name, model):
+        if name in QUERY_PARAMETERS:
+            return QUERY_PARAMETERS[name]
+
+        parts = str(name).split("__")
+        lookup = LOOKUP_NAMES.get(parts[-1])
+        field_parts = parts[:-1] if lookup else parts
+        label = _model_path_label(model, field_parts)
+        if not label:
+            label = _field_name_label(field_parts[-1] if field_parts else name)
+        if not label:
+            label = f"查询参数：{name}"
+        if lookup:
+            label = f"{label}（{lookup}）"
+        return label
+
+    def get_tags(self, operation_keys=None):
+        tags = super().get_tags(operation_keys)
+        if self.overrides.get("tags") or not tags:
+            return tags
+
+        if tags[0] in TAG_NAMES:
+            return [TAG_NAMES[tags[0]]]
+
+        try:
+            app_config = apps.get_app_config(tags[0])
+        except LookupError:
+            return tags
+
+        if _contains_chinese(app_config.verbose_name):
+            return [str(app_config.verbose_name)]
+        return tags
+
+    def _get_resource_name(self):
+        queryset = getattr(self.view, "queryset", None)
+        model = getattr(queryset, "model", None)
+
+        if model is None:
+            serializer_class = getattr(self.view, "serializer_class", None)
+            meta = getattr(serializer_class, "Meta", None)
+            model = getattr(meta, "model", None)
+
+        if model is None:
+            return "接口"
+
+        match = re.search(r"TN\s*[:：]\s*([^\n\r]+)", model.__doc__ or "")
+        if match:
+            return match.group(1).strip()
+
+        verbose_name = str(model._meta.verbose_name)
+        if _contains_chinese(verbose_name):
+            return verbose_name
+        return model.__name__

apps/utils/test_swagger.py

@@ -0,0 +1,142 @@
+import json
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from django.conf import settings
+from django.core.management import call_command
+from django.test import SimpleTestCase, override_settings
+
+from apps.am.models import Area
+from apps.am.views import AreaViewSet
+from apps.utils.swagger import ChineseSwaggerAutoSchema, swagger_schema_file
+
+
+class ChineseSwaggerAutoSchemaTests(SimpleTestCase):
+    def make_schema(self, view, method="GET"):
+        schema = ChineseSwaggerAutoSchema.__new__(ChineseSwaggerAutoSchema)
+        schema.view = view
+        schema.method = method
+        schema.path = "/am/area/"
+        schema.overrides = {}
+        schema.operation_keys = ("am", "area", "list")
+        schema._sch = SimpleNamespace(get_description=lambda path, method: "")
+        return schema
+
+    def test_crud_summary_uses_model_chinese_name(self):
+        view = SimpleNamespace(queryset=Area.objects.all(), action="list")
+        schema = self.make_schema(view)
+
+        summary, description = schema.get_summary_and_description()
+
+        self.assertEqual(summary, "查询地图区域列表")
+        self.assertEqual(description, "查询地图区域列表")
+
+    def test_explicit_summary_takes_priority(self):
+        view = SimpleNamespace(queryset=Area.objects.all(), action="list")
+        schema = self.make_schema(view)
+        schema.overrides = {
+            "operation_summary": "区域自定义查询",
+            "operation_description": "自定义说明",
+        }
+
+        summary, description = schema.get_summary_and_description()
+
+        self.assertEqual(summary, "区域自定义查询")
+        self.assertEqual(description, "自定义说明")
+
+    def test_custom_action_with_english_model_name_has_chinese_hint(self):
+        model = SimpleNamespace(
+            __doc__="",
+            __name__="Dataset",
+            _meta=SimpleNamespace(verbose_name="dataset"),
+        )
+        queryset = SimpleNamespace(model=model)
+        view = SimpleNamespace(queryset=queryset, action="base")
+        schema = self.make_schema(view)
+
+        summary, _ = schema.get_summary_and_description()
+
+        self.assertEqual(summary, "Dataset接口：base")
+
+    def test_tag_uses_chinese_business_module_name(self):
+        view = SimpleNamespace(queryset=Area.objects.all(), action="list")
+        schema = self.make_schema(view)
+
+        self.assertEqual(schema.get_tags(("am", "area", "list")), ["区域与准入管理"])
+
+    def test_filter_parameter_uses_model_field_labels(self):
+        view = SimpleNamespace(queryset=Area.objects.all(), action="list")
+        schema = self.make_schema(view)
+
+        description = schema._get_parameter_description(
+            "manager__name__contains",
+            Area,
+        )
+
+        self.assertIn("区域负责人", description)
+        self.assertIn("包含", description)
+
+    def test_swagger_queryset_skips_permission_data_lookup(self):
+        view = AreaViewSet(basename="area")
+        view.action = "list"
+        view.swagger_fake_view = True
+
+        with patch("apps.utils.viewsets.get_user_perms_map") as permission_lookup:
+            queryset = view.get_queryset()
+
+        self.assertIs(queryset.model, Area)
+        permission_lookup.assert_not_called()
+
+
+class SwaggerSettingsTests(SimpleTestCase):
+    def test_swagger_supports_jwt_authorization_header(self):
+        from django.conf import settings
+
+        bearer = settings.SWAGGER_SETTINGS["SECURITY_DEFINITIONS"]["Bearer"]
+
+        self.assertEqual(bearer["type"], "apiKey")
+        self.assertEqual(bearer["name"], "Authorization")
+        self.assertEqual(bearer["in"], "header")
+
+    def test_swagger_ui_uses_static_schema(self):
+        from django.conf import settings
+
+        self.assertEqual(settings.SWAGGER_SETTINGS["SPEC_URL"], "schema-swagger-json")
+        self.assertEqual(settings.REDOC_SETTINGS["SPEC_URL"], "schema-swagger-json")
+
+
+class BuildSwaggerCommandTests(SimpleTestCase):
+    def test_command_writes_valid_utf8_schema(self):
+        schema = {
+            "swagger": "2.0",
+            "info": {"title": "中文文档"},
+            "paths": {"/demo/": {"get": {}}},
+        }
+
+        def generate_schema(command_name, output_file, **options):
+            self.assertEqual(command_name, "generate_swagger")
+            self.assertEqual(output_file, "-")
+            options["stdout"].write(json.dumps(schema, ensure_ascii=False))
+
+        with TemporaryDirectory(dir=settings.BASE_DIR) as directory:
+            target = Path(directory) / "swagger.json"
+            with override_settings(SWAGGER_SCHEMA_PATH=str(target)):
+                with patch(
+                    "apps.utils.management.commands.build_swagger.call_command",
+                    side_effect=generate_schema,
+                ):
+                    call_command("build_swagger", verbosity=0)
+
+                content = target.read_text(encoding="utf-8")
+                self.assertIn("中文文档", content)
+                self.assertEqual(json.loads(content), schema)
+
+                with override_settings(SWAGGER_SCHEMA_PATH=str(target)):
+                    response = swagger_schema_file(SimpleNamespace())
+                    body = b"".join(response.streaming_content)
+                    response.close()
+
+                self.assertEqual(response.status_code, 200)
+                self.assertEqual(json.loads(body), schema)

apps/utils/viewsets.py

@@ -154,6 +154,9 @@ class CustomGenericViewSet(MyLoggingMixin, GenericViewSet):
     def get_queryset(self):
         queryset = super().get_queryset()
         queryset = self.get_queryset_custom(queryset)
+        # drf-yasg 生成文档时不应读取权限或业务数据。
+        if getattr(self, 'swagger_fake_view', False):
+            return queryset
         if self.data_filter:
             user = self.request.user
             if user.is_superuser:
@@ -232,4 +235,4 @@ class EuModelViewSet(BulkCreateModelMixin, CustomListModelMixin,
                                CustomRetrieveModelMixin, BulkDestroyModelMixin, ComplexQueryMixin, CustomGenericViewSet):
     """
     不支持更新的增强ModelViewSet
-    """
+    """

apps/wpm/views.py

@@ -354,6 +354,7 @@ class MlogViewSet(CustomModelViewSet):
         return super().get_serializer_class()
 
     @swagger_auto_schema(
+        operation_summary="查询生产日志明细（逐条追溯）",
         manual_parameters=[
             openapi.Parameter(name="query", in_=openapi.IN_QUERY, description="定制返回数据", type=openapi.TYPE_STRING, required=False),
             openapi.Parameter(name="with_children", in_=openapi.IN_QUERY, description="带有children(yes/no/count)", type=openapi.TYPE_STRING, required=False),

server/settings.py

@@ -178,6 +178,7 @@ USE_TZ = True
 
 STATIC_URL = '/static/'
 STATIC_ROOT = os.path.join(BASE_DIR, 'dist/static')
+SWAGGER_SCHEMA_PATH = os.path.join(STATIC_ROOT, 'openapi/swagger.json')
 # STATICFILES_DIRS = (
 #     os.path.join(BASE_DIR, 'dist/static'),
 # )
@@ -267,8 +268,27 @@ CELERYD_SOFT_TIME_LIMIT = 60*10
 
 # swagger配置
 SWAGGER_SETTINGS = {
+    'DEFAULT_INFO': 'server.swagger.api_info',
+    'DEFAULT_API_URL': BASE_URL,
+    'SPEC_URL': 'schema-swagger-json',
     'LOGIN_URL': '/django/admin/login/',
     'LOGOUT_URL': '/django/admin/logout/',
+    'DEFAULT_AUTO_SCHEMA_CLASS': 'apps.utils.swagger.ChineseSwaggerAutoSchema',
+    'SECURITY_DEFINITIONS': {
+        'Bearer': {
+            'type': 'apiKey',
+            'name': 'Authorization',
+            'in': 'header',
+            'description': 'JWT认证，请输入：Bearer <access token>',
+        },
+        'Basic': {
+            'type': 'basic',
+        },
+    },
+}
+
+REDOC_SETTINGS = {
+    'SPEC_URL': 'schema-swagger-json',
 }
 
 # 日志配置

server/swagger.py

@@ -0,0 +1,12 @@
+from django.conf import settings
+from drf_yasg import openapi
+
+from server.settings import get_sysconfig
+
+
+api_info = openapi.Info(
+    title=f'{settings.SYS_NAME}--{get_sysconfig("base.base_name", "demo")}',
+    default_version=settings.SYS_VERSION,
+    contact=openapi.Contact(email="caoqianming@foxmail.com"),
+    license=openapi.License(name="MIT License"),
+)

server/urls.py

@@ -17,19 +17,14 @@ from django.conf import settings
 from django.conf.urls.static import static
 from django.contrib import admin
 from django.urls import include, path
-from drf_yasg import openapi
 from drf_yasg.views import get_schema_view
 from rest_framework.documentation import include_docs_urls
 from django.views.generic import TemplateView
-from server.settings import get_sysconfig
+from apps.utils.swagger import swagger_schema_file
+from server.swagger import api_info
 
 schema_view = get_schema_view(
-    openapi.Info(
+    api_info,
-        title=f'{settings.SYS_NAME}--{get_sysconfig("base.base_name", "demo")}',
-        default_version=settings.SYS_VERSION,
-        contact=openapi.Contact(email="caoqianming@foxmail.com"),
-        license=openapi.License(name="MIT License"),
-    ),
     public=True,
     permission_classes=[],
     url=settings.BASE_URL
@@ -89,8 +84,10 @@ urlpatterns = [
 if getattr(settings, 'ENABLE_SWAGGER', True):
     urlpatterns += [
         # api文档
+        path('api/swagger.json', swagger_schema_file,
+             name='schema-swagger-json'),
         path('api/swagger/', schema_view.with_ui('swagger',
             cache_timeout=0), name='schema-swagger-ui'),
         path('api/redoc/', schema_view.with_ui('redoc',
             cache_timeout=0), name='schema-redoc'),
-    ]
+    ]