DRF项目为各操作增加操作日志

定义一个装饰器decorators.py：

from functools import wrapsfrom django.utils.decorators import method_decoratorfrom django.contrib.auth.models import AnonymousUserfrom . import func_parse_tokenfrom .models import Record_Logfrom django.urls import resolvefrom wrapper.Params import OPERATION_TYPEimport jsondef get_operation_details(request, response): \"\"\" 从请求和响应中提取操作详细信息 \"\"\" method = request.method details = { \'query_params\': dict(request.GET), \'request_data\': dict(request.POST), \'files\': [f.name for f in request.FILES.values()] if request.FILES else [] } # 根据请求方法确定操作类型 if method == \'GET\': op_type = \'查询\' if request.path_info.endswith(\'export/\'): op_type = \'导出\' elif request.path_info.endswith(\'template/\'): op_type = \'下载\' elif method == \'POST\': op_type = \'创建\' if request.path_info.endswith(\'import/\'): op_type = \'导入\' elif request.path_info.endswith(\'upload/\'): op_type = \'上传\' elif method == \'PUT\': op_type = \'更新\' elif method == \'DELETE\': op_type = \'删除\' else: op_type = method # 尝试从响应中获取更多信息 if hasattr(response, \'data\'): try: response_data = response.data if isinstance(response_data, dict): if \'count\' in response_data:  details[\'total_count\'] = response_data[\'count\'] if \'data\' in response_data:  if isinstance(response_data[\'data\'], list): details[\'result_count\'] = len(response_data[\'data\'])  elif isinstance(response_data[\'data\'], dict): details[\'result_data\'] = response_data[\'data\'] except: pass return op_type, detailsdef operation_logger(operation_name=None): \"\"\" 装饰器：记录操作日志 :param operation_name: 操作名称，如果不提供则自动从URL名称获取 \"\"\" def decorator(func): @wraps(func) def _wrapped_view(request, *args, **kwargs): # 执行原始视图函数 response = func(request, *args, **kwargs) # 获取操作名称 if operation_name: action_name = operation_name else: url_name = resolve(request.path_info).url_name action_name = f\"{OPERATION_TYPE.get(request.method, \'\')} {url_name}\" # 记录日志 if 200 <= response.status_code < 300: # 获取操作详细信息 op_type, details = get_operation_details(request, response) log = Record_Log() log.action = request.path_info # 检查业务逻辑是否成功 is_success = True if hasattr(response, \'data\') and isinstance(response.data, dict):  # 检查是否有错误信息  if \'code\' in response.data and response.data[\'code\'] != 200: is_success = False  # 对于导入/上传操作，检查是否有文件  elif op_type in [\'导入\', \'上传\'] and not request.FILES: is_success = False log.action_result = \'操作成功\' if is_success else \'操作失败\' # 处理匿名用户的情况 if hasattr(request, \'user\') and not isinstance(request.user, AnonymousUser):  log.user_name = request.user.name else:  log.user_name = \"anonymous\" ip_addr = func_parse_token.get_username_and_ip_from_cookie(request)[1] log.ip_addr = ip_addr if ip_addr else \"unknown ip\" log.detail_info = json.dumps(details, ensure_ascii=False) log.action_method = request.method log.action_name = f\"{op_type}_{action_name}\" log.operation_level = \'work\' if is_success else \'warning\' log.save() return response return _wrapped_view return decoratordef class_operation_logger(operation_name=None): \"\"\" 类视图的操作日志装饰器 \"\"\" def decorator(cls): dispatch = cls.dispatch @method_decorator(operation_logger(operation_name)) def _wrapped_dispatch(self, request, *args, **kwargs): return dispatch(self, request, *args, **kwargs)  cls.dispatch = _wrapped_dispatch return cls return decorator 

在view.py的视图类上方添加该装饰器：

# 要报规则列表@class_operation_logger(operation_name=\"要报规则\")class SysmanageRuleListView(CreateWithTimeViewSet): queryset = SysmanageRule.objects.all() serializer_class = SysmanageRuleSerializer ordering = [\'-id\'] # view_field = [\'rule_name\', \'规则名称\'] # view_model = SysmanageRule filterset_fields = { \'id\': [\'exact\'], \'rule_name\': [\'icontains\'], \'rule_status\': [\'exact\'], } no_update_columns = [\'create_time\', \'create_user\'] def create(self, request, *args, **kwargs): data = request.data.copy() rule_name = data.get(\'rule_name\') # 去除前后空格，避免空格导致重复绕过 if rule_name: rule_name = rule_name.strip() data[\'rule_name\'] = rule_name # 检查是否存在重复的规则名称 if SysmanageRule.objects.filter(rule_name=rule_name).exists(): return ErrorResponse(data=False, msg=\"该规则名称已存在\") data[\'create_user\'] = get_user_name(request) data[\'create_time\'] = datetime.now() serializer = self.get_serializer(data=data) serializer.is_valid(raise_exception=True) self.perform_create(serializer) return self.check_repeat(request, serializer, msg=\"新增成功\")