34 lines
1.0 KiB
Python
34 lines
1.0 KiB
Python
"""请求日志上下文和敏感摘要清理。"""
|
|
|
|
import contextvars
|
|
import logging
|
|
from collections.abc import Mapping
|
|
from typing import Any
|
|
|
|
request_id_context: contextvars.ContextVar[str] = contextvars.ContextVar(
|
|
"request_id", default="-"
|
|
)
|
|
|
|
SENSITIVE_KEYS = {"api_key", "authorization", "cookie", "password", "secret", "token"}
|
|
|
|
|
|
class RequestContextFilter(logging.Filter):
|
|
"""向每条日志补充请求关联标识，后台任务没有请求时使用短横线。"""
|
|
|
|
def filter(self, record: logging.LogRecord) -> bool:
|
|
record.request_id = request_id_context.get()
|
|
return True
|
|
|
|
|
|
def sanitize_summary(value: Any) -> Any:
|
|
"""递归遮蔽常见敏感键；调用方仍需限制摘要的业务范围和长度。"""
|
|
|
|
if isinstance(value, Mapping):
|
|
return {
|
|
str(key): "***" if str(key).lower() in SENSITIVE_KEYS else sanitize_summary(item)
|
|
for key, item in value.items()
|
|
}
|
|
if isinstance(value, list):
|
|
return [sanitize_summary(item) for item in value]
|
|
return value
|