Compare commits
No commits in common. "ed952d2d3a1fee5c5b8390a8e1b7b210d8be1ad7" and "5fb179eb9e678877be785388ef8ebc8c05640b19" have entirely different histories.
ed952d2d3a
...
5fb179eb9e
|
|
@ -18,35 +18,10 @@ 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
|
||||
|
|
@ -61,14 +36,6 @@ class DatasetRecordSerializer(CustomModelSerializer):
|
|||
|
||||
class DataExecSerializer(serializers.Serializer):
|
||||
query = serializers.JSONField(
|
||||
label="查询字典参数",
|
||||
help_text="按所选数据集 description/default_param 声明的业务参数填写",
|
||||
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
|
||||
)
|
||||
label="查询字典参数", required=False, allow_null=True)
|
||||
is_test = serializers.BooleanField(label='是否测试', default=False)
|
||||
raise_exception = serializers.BooleanField(label='是否直接报错', default=False)
|
||||
|
|
|
|||
|
|
@ -1,25 +1,3 @@
|
|||
from django.test import SimpleTestCase
|
||||
from django.test import TestCase
|
||||
|
||||
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)
|
||||
# Create your tests here.
|
||||
|
|
|
|||
|
|
@ -3,13 +3,7 @@ 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 (
|
||||
DataExecSerializer,
|
||||
DatasetCreateUpdateSerializer,
|
||||
DatasetListResponseSerializer,
|
||||
DatasetRecordSerializer,
|
||||
DatasetSerializer,
|
||||
)
|
||||
from apps.bi.serializers import DatasetSerializer, DatasetCreateUpdateSerializer, DataExecSerializer, DatasetRecordSerializer
|
||||
from django.apps import apps
|
||||
import concurrent.futures
|
||||
from django.core.cache import cache
|
||||
|
|
@ -19,8 +13,6 @@ 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.
|
||||
|
||||
|
|
@ -30,54 +22,9 @@ class DatasetViewSet(CustomModelViewSet):
|
|||
serializer_class = DatasetSerializer
|
||||
create_serializer_class = DatasetCreateUpdateSerializer
|
||||
update_serializer_class = DatasetCreateUpdateSerializer
|
||||
search_fields = ['name', 'code', 'description']
|
||||
search_fields = ['name', 'code']
|
||||
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.
|
||||
|
|
@ -110,18 +57,6 @@ 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查询
|
||||
|
|
|
|||
|
|
@ -71,8 +71,6 @@ 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
|
||||
|
|
@ -144,8 +142,6 @@ 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)
|
||||
|
|
@ -211,8 +207,6 @@ 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)
|
||||
|
|
|
|||
|
|
@ -651,7 +651,6 @@ 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':
|
||||
|
|
|
|||
|
|
@ -69,7 +69,6 @@ class SpeakerViewSet(CustomGenericViewSet):
|
|||
"""
|
||||
perms_map = {}
|
||||
serializer_class = serializers.Serializer
|
||||
filter_backends = []
|
||||
|
||||
@action(methods=['get'], detail=False,
|
||||
permission_classes=[IsAuthenticated])
|
||||
|
|
@ -126,7 +125,6 @@ class XxTestView(APIView):
|
|||
class XxCommonViewSet(CreateModelMixin, CustomGenericViewSet):
|
||||
perms_map = {'post': '*'}
|
||||
serializer_class = RequestCommonSerializer
|
||||
filter_backends = []
|
||||
|
||||
def create(self, request, *args, **kwargs):
|
||||
"""
|
||||
|
|
@ -260,7 +258,6 @@ class KingCommonViewSet(CreateModelMixin, CustomGenericViewSet):
|
|||
class DhCommonViewSet(CreateModelMixin, CustomGenericViewSet):
|
||||
perms_map = {'post': '*'}
|
||||
serializer_class = RequestCommonSerializer
|
||||
filter_backends = []
|
||||
|
||||
def create(self, request, *args, **kwargs):
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -1,71 +0,0 @@
|
|||
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}个操作）"
|
||||
)
|
||||
)
|
||||
|
|
@ -207,28 +207,12 @@ class CustomRetrieveModelMixin(RetrieveModelMixin):
|
|||
|
||||
class CustomListModelMixin(ListModelMixin):
|
||||
|
||||
@swagger_auto_schema(
|
||||
operation_description=(
|
||||
"通用列表接口用于记录或目录浏览以及逐条追溯。跨时间范围的产量、良率、缺陷、"
|
||||
"库存、绩效和趋势等统计聚合，优先查询 BI dataset 目录并执行匹配的数据集。"
|
||||
),
|
||||
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,
|
||||
),
|
||||
],
|
||||
)
|
||||
@swagger_auto_schema(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),
|
||||
])
|
||||
def list(self, request, *args, **kwargs):
|
||||
|
||||
queryset = self.filter_queryset(self.get_queryset())
|
||||
|
|
|
|||
|
|
@ -1,367 +0,0 @@
|
|||
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__
|
||||
|
|
@ -1,142 +0,0 @@
|
|||
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)
|
||||
|
|
@ -154,9 +154,6 @@ 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:
|
||||
|
|
@ -235,4 +232,4 @@ class EuModelViewSet(BulkCreateModelMixin, CustomListModelMixin,
|
|||
CustomRetrieveModelMixin, BulkDestroyModelMixin, ComplexQueryMixin, CustomGenericViewSet):
|
||||
"""
|
||||
不支持更新的增强ModelViewSet
|
||||
"""
|
||||
"""
|
||||
|
|
@ -354,7 +354,6 @@ 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),
|
||||
|
|
|
|||
|
|
@ -178,7 +178,6 @@ 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'),
|
||||
# )
|
||||
|
|
@ -268,27 +267,8 @@ 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',
|
||||
}
|
||||
|
||||
# 日志配置
|
||||
|
|
|
|||
|
|
@ -1,12 +0,0 @@
|
|||
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"),
|
||||
)
|
||||
|
|
@ -17,14 +17,19 @@ 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 apps.utils.swagger import swagger_schema_file
|
||||
from server.swagger import api_info
|
||||
from server.settings import get_sysconfig
|
||||
|
||||
schema_view = get_schema_view(
|
||||
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"),
|
||||
),
|
||||
public=True,
|
||||
permission_classes=[],
|
||||
url=settings.BASE_URL
|
||||
|
|
@ -84,10 +89,8 @@ 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'),
|
||||
]
|
||||
]
|
||||
Loading…
Reference in New Issue