368 lines
12 KiB
Python
368 lines
12 KiB
Python
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__
|