#!/usr/bin/env python
# coding: utf-8
"""
增强的日志中间件
提供请求追踪、结构化日志、性能监控等功能
"""
import logging
import os
import time
import uuid
from datetime import datetime
from typing import Optional, Dict, Any, List
from functools import wraps
from .._json import dumps_str
from .base import Middleware
[文档]
class EnhancedLoggingMiddleware(Middleware):
"""
增强的日志中间件
功能特性：
- 请求追踪：为每个请求生成唯一 ID
- 结构化日志：支持 JSON 格式输出
- 性能监控：记录详细的性能指标
- 敏感信息过滤：自动过滤敏感参数
- 智能日志级别：根据响应状态码自动调整
"""
# 敏感参数名称列表
SENSITIVE_PARAMS = [
'password', 'passwd', 'pwd',
'token', 'access_token', 'refresh_token',
'secret', 'api_key', 'apikey',
'credit_card', 'card_number',
'ssn', 'social_security_number'
]
def __init__(
self,
app,
logger: Optional[logging.Logger] = None,
structured: bool = False,
sensitive_params: Optional[List[str]] = None,
exclude_paths: Optional[List[str]] = None,
log_request_body: bool = False,
log_response_body: bool = False,
max_body_length: int = 1000
):
"""
初始化增强的日志中间件
Args:
app: Litefs 应用实例
logger: 日志记录器，如果为 None 则使用应用的日志记录器
structured: 是否使用结构化日志（JSON 格式）
sensitive_params: 自定义敏感参数列表
exclude_paths: 不记录日志的路径列表（支持通配符）
log_request_body: 是否记录请求体
log_response_body: 是否记录响应体
max_body_length: 记录的最大请求/响应体长度
"""
super(EnhancedLoggingMiddleware, self).__init__(app)
if logger is None:
self.logger = getattr(app, "logger", logging.getLogger(__name__))
else:
self.logger = logger
self.structured = structured
self.sensitive_params = self.SENSITIVE_PARAMS + (sensitive_params or [])
self.exclude_paths = exclude_paths or []
self.log_request_body = log_request_body
self.log_response_body = log_response_body
self.max_body_length = max_body_length
def _should_skip_logging(self, path: str) -> bool:
"""
检查是否应该跳过日志记录
Args:
path: 请求路径
Returns:
是否跳过
"""
import fnmatch
for pattern in self.exclude_paths:
if fnmatch.fnmatch(path, pattern):
return True
return False
def _generate_request_id(self) -> str:
"""
生成请求 ID
Returns:
请求 ID 字符串
"""
return str(uuid.uuid4())
def _filter_sensitive_data(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""
过滤敏感数据
Args:
data: 原始数据字典
Returns:
过滤后的数据字典
"""
if not isinstance(data, dict):
return data
filtered = {}
for key, value in data.items():
if key.lower() in [param.lower() for param in self.sensitive_params]:
filtered[key] = '***FILTERED***'
elif isinstance(value, dict):
filtered[key] = self._filter_sensitive_data(value)
else:
filtered[key] = value
return filtered
def _truncate_body(self, body: Any, max_length: int = None) -> str:
"""
截断请求/响应体
Args:
body: 请求/响应体
max_length: 最大长度
Returns:
截断后的字符串
"""
if max_length is None:
max_length = self.max_body_length
if body is None:
return ''
if isinstance(body, bytes):
try:
body = body.decode('utf-8')
except UnicodeDecodeError:
return f'<binary data: {len(body)} bytes>'
body_str = str(body)
if len(body_str) > max_length:
return body_str[:max_length] + '... (truncated)'
return body_str
def _get_log_level(self, status_code: int) -> int:
"""
根据状态码获取日志级别
Args:
status_code: HTTP 状态码
Returns:
日志级别
"""
if status_code >= 500:
return logging.ERROR
elif status_code >= 400:
return logging.WARNING
else:
return logging.INFO
def _format_log_message(self, data: Dict[str, Any], level: int) -> str:
"""
格式化日志消息
Args:
data: 日志数据
level: 日志级别
Returns:
格式化后的日志消息
"""
if self.structured:
return dumps_str(data, default=str)
else:
# 传统格式
request_id = data.get('request_id', '-')
method = data.get('method', 'GET')
path = data.get('path', '/')
status_code = data.get('status_code', 200)
duration = data.get('duration', 0)
remote_addr = data.get('remote_addr', '-')
if 'error' in data:
return (
f"[{request_id}] {remote_addr} - {method} {path} - {status_code} - "
f"{duration:.3f}s - ERROR: {data['error']}"
)
else:
return (
f"[{request_id}] {remote_addr} - {method} {path} - "
f"{status_code} - {duration:.3f}s"
)
[文档]
def process_request(self, request_handler):
"""
处理请求，记录请求开始信息
Args:
request_handler: 请求处理器实例
"""
# 检查是否跳过日志记录
path = request_handler.environ.get("PATH_INFO", "/")
if self._should_skip_logging(path):
return None
# 生成请求 ID
request_id = self._generate_request_id()
request_handler.request_id = request_id
request_handler._start_time = time.time()
# 收集请求信息
request_data = {
'request_id': request_id,
'timestamp': datetime.now().isoformat(),
'method': request_handler.environ.get("REQUEST_METHOD", "GET"),
'path': path,
'query_string': request_handler.environ.get("QUERY_STRING", ""),
'remote_addr': request_handler.environ.get("REMOTE_ADDR", "-"),
'user_agent': request_handler.environ.get("HTTP_USER_AGENT", "-"),
'referer': request_handler.environ.get("HTTP_REFERER", "-"),
'content_type': request_handler.environ.get("CONTENT_TYPE", ""),
'content_length': request_handler.environ.get("CONTENT_LENGTH", "0"),
'server_protocol': request_handler.environ.get("SERVER_PROTOCOL", "HTTP/1.1"),
'pid': os.getpid()
}
# 记录请求参数（GET 和 POST）
if hasattr(request_handler, 'get') and request_handler.get:
request_data['get_params'] = self._filter_sensitive_data(dict(request_handler.get))
if hasattr(request_handler, 'post') and request_handler.post:
request_data['post_params'] = self._filter_sensitive_data(dict(request_handler.post))
# 记录请求体
if self.log_request_body and hasattr(request_handler, 'body') and request_handler.body:
request_data['request_body'] = self._truncate_body(request_handler.body)
# 记录请求开始
log_message = self._format_log_message(request_data, logging.INFO)
self.logger.info(f"Request started: {log_message}")
# 保存请求数据，供后续使用
request_handler._log_data = request_data
return None
[文档]
def process_response(self, request_handler, response):
"""
处理响应，记录请求处理结果
Args:
request_handler: 请求处理器实例
response: 响应数据
Returns:
响应数据
"""
# 检查是否跳过日志记录
path = request_handler.environ.get("PATH_INFO", "/")
if self._should_skip_logging(path):
return response
# 计算请求处理时间
duration = 0
if hasattr(request_handler, '_start_time'):
duration = time.time() - request_handler._start_time
# 解析响应状态码
status_code = 200
if isinstance(response, tuple) and len(response) == 3:
status, headers, content = response
status_code = int(status.split()[0]) if isinstance(status, str) else status
elif isinstance(response, dict):
status_code = response.get('status_code', 200)
# 构建响应数据
response_data = {
'request_id': getattr(request_handler, 'request_id', '-'),
'timestamp': datetime.now().isoformat(),
'method': request_handler.environ.get("REQUEST_METHOD", "GET"),
'path': path,
'status_code': status_code,
'duration': round(duration, 3),
'remote_addr': request_handler.environ.get("REMOTE_ADDR", "-"),
'pid': os.getpid()
}
# 记录响应体
if self.log_response_body and isinstance(response, tuple) and len(response) == 3:
_, _, content = response
response_data['response_body'] = self._truncate_body(content)
# 根据状态码确定日志级别
log_level = self._get_log_level(status_code)
# 记录响应
log_message = self._format_log_message(response_data, log_level)
self.logger.log(log_level, f"Request completed: {log_message}")
# 添加性能响应头
if isinstance(response, tuple) and len(response) == 3:
status, headers, content = response
headers = list(headers)
headers.append(('X-Request-ID', str(response_data['request_id'])))
headers.append(('X-Response-Time', f'{duration:.3f}s'))
return status, headers, content
return response
[文档]
def process_exception(self, request_handler, exception):
"""
处理异常，记录异常信息
Args:
request_handler: 请求处理器实例
exception: 异常对象
Returns:
None: 继续抛出异常
"""
# 计算请求处理时间
duration = 0
if hasattr(request_handler, '_start_time'):
duration = time.time() - request_handler._start_time
# 构建错误数据
error_data = {
'request_id': getattr(request_handler, 'request_id', '-'),
'timestamp': datetime.now().isoformat(),
'method': request_handler.environ.get("REQUEST_METHOD", "GET"),
'path': request_handler.environ.get("PATH_INFO", "/"),
'status_code': 500,
'duration': round(duration, 3),
'remote_addr': request_handler.environ.get("REMOTE_ADDR", "-"),
'pid': os.getpid(),
'error': str(exception),
'error_type': type(exception).__name__
}
# 记录异常
log_message = self._format_log_message(error_data, logging.ERROR)
self.logger.error(f"Request failed: {log_message}", exc_info=True)
return None
[文档]
class RequestContextLogger:
"""
请求上下文日志记录器
提供在请求上下文中记录日志的便捷方法
"""
def __init__(self, request_handler):
"""
初始化请求上下文日志记录器
Args:
request_handler: 请求处理器实例
"""
self.request_handler = request_handler
self.request_id = getattr(request_handler, 'request_id', '-')
[文档]
def info(self, message: str, **kwargs):
"""
记录 INFO 级别日志
Args:
message: 日志消息
**kwargs: 额外的上下文数据
"""
extra_data = {'request_id': self.request_id, **kwargs}
logging.info(f"[{self.request_id}] {message}", extra=extra_data)
[文档]
def warning(self, message: str, **kwargs):
"""
记录 WARNING 级别日志
Args:
message: 日志消息
**kwargs: 额外的上下文数据
"""
extra_data = {'request_id': self.request_id, **kwargs}
logging.warning(f"[{self.request_id}] {message}", extra=extra_data)
[文档]
def error(self, message: str, **kwargs):
"""
记录 ERROR 级别日志
Args:
message: 日志消息
**kwargs: 额外的上下文数据
"""
extra_data = {'request_id': self.request_id, **kwargs}
logging.error(f"[{self.request_id}] {message}", extra=extra_data, exc_info=True)
[文档]
def debug(self, message: str, **kwargs):
"""
记录 DEBUG 级别日志
Args:
message: 日志消息
**kwargs: 额外的上下文数据
"""
extra_data = {'request_id': self.request_id, **kwargs}
logging.debug(f"[{self.request_id}] {message}", extra=extra_data)