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__