""" 日志 django中间件 """ import json import logging from django.conf import settings from django.contrib.auth.models import AnonymousUser from django.core.exceptions import MultipleObjectsReturned from django.http import HttpResponse, HttpResponseServerError from django.utils import translation from django.utils.translation.trans_real import parse_accept_lang_header from django.utils.deprecation import MiddlewareMixin from dvadmin.system.models import OperationLog from dvadmin.utils.request_util import get_request_user, get_request_ip, get_request_data, get_request_path, get_os, \ get_browser, get_verbose_name class ApiLoggingMiddleware(MiddlewareMixin): """ 用于记录API访问日志中间件 """ def __init__(self, get_response=None): super().__init__(get_response) self.enable = getattr(settings, 'API_LOG_ENABLE', None) or False self.methods = getattr(settings, 'API_LOG_METHODS', None) or set() self.operation_log_id = None @classmethod def __handle_request(cls, request): request.request_ip = get_request_ip(request) request.request_data = get_request_data(request) request.request_path = get_request_path(request) def __handle_response(self, request, response): # 判断有无log_id属性，使用All记录时，会出现此情况 if request.request_data.get('log_id', None) is None: return # 移除log_id，不记录此ID log_id = request.request_data.pop('log_id') # request_data,request_ip由PermissionInterfaceMiddleware中间件中添加的属性 body = getattr(request, 'request_data', {}) # 请求含有password则用*替换掉(暂时先用于所有接口的password请求参数) if isinstance(body, dict) and body.get('password', ''): body['password'] = '*' * len(body['password']) if not hasattr(response, 'data') or not isinstance(response.data, dict): response.data = {} try: if not response.data and response.content: content = json.loads(response.content.decode()) response.data = content if isinstance(content, dict) else {} except Exception: return user = get_request_user(request) info = { 'request_ip': getattr(request, 'request_ip', 'unknown'), 'creator': user if not isinstance(user, AnonymousUser) else None, 'dept_belong_id': getattr(request.user, 'dept_id', None), 'request_method': request.method, 'request_path': request.request_path, 'request_body': body, 'response_code': response.data.get('code'), 'request_os': get_os(request), 'request_browser': get_browser(request), 'request_msg': request.session.get('request_msg'), 'status': True if response.data.get('code') in [2000, ] else False, 'json_result': {"code": response.data.get('code'), "msg": response.data.get('msg')}, } # 记录日志不应影响业务接口返回；同时兼容历史脏数据导致的重复id问题 try: operation_log, _created = OperationLog.objects.update_or_create(defaults=info, id=log_id) except MultipleObjectsReturned: OperationLog.objects.filter(id=log_id).update(**info) operation_log = ( OperationLog.objects.filter(id=log_id) .order_by('-create_datetime') .first() ) except Exception: return if operation_log and (not operation_log.request_modular) and settings.API_MODEL_MAP.get(request.request_path, None): operation_log.request_modular = settings.API_MODEL_MAP[request.request_path] operation_log.save(update_fields=['request_modular', 'update_datetime']) def process_view(self, request, view_func, view_args, view_kwargs): if hasattr(view_func, 'cls') and hasattr(view_func.cls, 'queryset'): if self.enable: if self.methods == 'ALL' or request.method in self.methods: log = OperationLog(request_modular=get_verbose_name(view_func.cls.queryset)) log.save() # self.operation_log_id = log.id request.request_data['log_id'] = log.id return def process_request(self, request): self.__handle_request(request) def process_response(self, request, response): """ 主要请求处理完之后记录 :param request: :param response: :return: """ if self.enable: if self.methods == 'ALL' or request.method in self.methods: self.__handle_response(request, response) return response logger = logging.getLogger("healthz") class HealthCheckMiddleware(object): """ 存活检查中间件 """ def __init__(self, get_response): self.get_response = get_response # One-time configuration and initialization. def __call__(self, request): if request.method == "GET": if request.path == "/readiness": return self.readiness(request) elif request.path == "/healthz": return self.healthz(request) return self.get_response(request) def healthz(self, request): """ Returns that the server is alive. """ return HttpResponse("OK") def readiness(self, request): # Connect to each database and do a generic standard SQL query # that doesn't write any data and doesn't depend on any tables # being present. try: from django.db import connections for name in connections: cursor = connections[name].cursor() cursor.execute("SELECT 1;") row = cursor.fetchone() if row is None: return HttpResponseServerError("db: invalid response") except Exception as e: logger.exception(e) return HttpResponseServerError("db: cannot connect to database.") # Call get_stats() to connect to each memcached instance and get it's stats. # This can effectively check if each is online. try: from django.core.cache import caches from django.core.cache.backends.memcached import BaseMemcachedCache for cache in caches.all(): if isinstance(cache, BaseMemcachedCache): stats = cache._cache.get_stats() if len(stats) != len(cache._servers): return HttpResponseServerError("cache: cannot connect to cache.") except Exception as e: logger.exception(e) return HttpResponseServerError("cache: cannot connect to cache.") return HttpResponse("OK") class LocaleMiddleware: """Set request locale from Accept-Language header (D-03: header always wins).""" # Map frontend locale codes to Django locale codes (D-07 fallback order) LOCALE_MAP = { 'en': 'en', 'zh-cn': 'zh-hans', 'zh-tw': 'zh-hant', } def __init__(self, get_response): self.get_response = get_response def __call__(self, request): lang_code = None # 1. Accept-Language header always wins (D-03) accept_lang = request.META.get('HTTP_ACCEPT_LANGUAGE', '') if accept_lang: try: for code, _priority in parse_accept_lang_header(accept_lang): if code == '*': continue # Normalize: 'zh-CN' -> 'zh-cn' normalized = code.lower().replace('_', '-') if normalized in self.LOCALE_MAP: lang_code = self.LOCALE_MAP[normalized] break elif code in dict(settings.LANGUAGES): lang_code = code break except Exception: pass # 2. Authenticated user DB preference as fallback (D-01: header wins) if not lang_code and hasattr(request, 'user') and request.user.is_authenticated: user_lang = getattr(request.user, 'language', None) or 'zh-cn' lang_code = self.LOCALE_MAP.get(user_lang, user_lang) # 3. Default for unauthenticated/unknown (D-06: zh-CN default) if not lang_code: lang_code = 'zh-hans' translation.activate(lang_code) request.LANGUAGE_CODE = lang_code response = self.get_response(request) translation.deactivate() return response