100 lines
3.0 KiB
Python
100 lines
3.0 KiB
Python
|
# =============================================================================
|
||
|
|
# IT智能服务台 — 日志配置
|
||
|
|
# =============================================================================
|
||
|
|
# 说明：统一日志格式，支持 JSON 输出便于日志收集
|
||
|
|
# =============================================================================
|
||
|
|
|
||
|
|
import json
|
||
|
|
import logging
|
||
|
|
import sys
|
||
|
|
from datetime import datetime
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
|
||
|
|
class JSONFormatter(logging.Formatter):
|
||
|
|
"""JSON 格式日志 formatter"""
|
||
|
|
|
||
|
|
def format(self, record: logging.LogRecord) -> str:
|
||
|
|
"""将日志记录格式化为 JSON"""
|
||
|
|
log_data: dict[str, Any] = {
|
||
|
|
"timestamp": datetime.utcnow().isoformat() + "Z",
|
||
|
|
"level": record.levelname,
|
||
|
|
"logger": record.name,
|
||
|
|
"message": record.getMessage(),
|
||
|
|
"module": record.module,
|
||
|
|
"function": record.funcName,
|
||
|
|
"line": record.lineno,
|
||
|
|
}
|
||
|
|
|
||
|
|
# 添加异常信息
|
||
|
|
if record.exc_info:
|
||
|
|
log_data["exception"] = self.formatException(record.exc_info)
|
||
|
|
|
||
|
|
# 添加额外字段
|
||
|
|
if hasattr(record, "request_id"):
|
||
|
|
log_data["request_id"] = record.request_id
|
||
|
|
if hasattr(record, "user_id"):
|
||
|
|
log_data["user_id"] = record.user_id
|
||
|
|
if hasattr(record, "extra"):
|
||
|
|
log_data.update(record.extra)
|
||
|
|
|
||
|
|
return json.dumps(log_data, ensure_ascii=False)
|
||
|
|
|
||
|
|
|
||
|
|
class PlainFormatter(logging.Formatter):
|
||
|
|
"""普通格式日志 formatter（开发环境使用）"""
|
||
|
|
|
||
|
|
def __init__(self):
|
||
|
|
super().__init__(
|
||
|
|
fmt="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
||
|
|
datefmt="%Y-%m-%d %H:%M:%S",
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def setup_logging(level: str = "INFO", json_format: bool = False) -> None:
|
||
|
|
"""配置日志系统
|
||
|
|
|
||
|
|
Args:
|
||
|
|
level: 日志级别 (DEBUG, INFO, WARNING, ERROR, CRITICAL)
|
||
|
|
json_format: 是否使用 JSON 格式输出
|
||
|
|
"""
|
||
|
|
log_level = getattr(logging, level.upper(), logging.INFO)
|
||
|
|
|
||
|
|
# 获取 root logger
|
||
|
|
root_logger = logging.getLogger()
|
||
|
|
root_logger.setLevel(log_level)
|
||
|
|
|
||
|
|
# 清除现有 handlers
|
||
|
|
for handler in root_logger.handlers[:]:
|
||
|
|
root_logger.removeHandler(handler)
|
||
|
|
|
||
|
|
# 创建 console handler
|
||
|
|
console_handler = logging.StreamHandler(sys.stdout)
|
||
|
|
console_handler.setLevel(log_level)
|
||
|
|
|
||
|
|
# 设置 formatter
|
||
|
|
if json_format:
|
||
|
|
formatter = JSONFormatter()
|
||
|
|
else:
|
||
|
|
formatter = PlainFormatter()
|
||
|
|
|
||
|
|
console_handler.setFormatter(formatter)
|
||
|
|
root_logger.addHandler(console_handler)
|
||
|
|
|
||
|
|
# 设置第三方库日志级别
|
||
|
|
logging.getLogger("uvicorn").setLevel(logging.WARNING)
|
||
|
|
logging.getLogger("fastapi").setLevel(logging.WARNING)
|
||
|
|
logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
|
||
|
|
|
||
|
|
|
||
|
|
def get_logger(name: str) -> logging.Logger:
|
||
|
|
"""获取 logger 实例
|
||
|
|
|
||
|
|
Args:
|
||
|
|
name: logger 名称，通常使用 __name__
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
Logger 实例
|
||
|
|
"""
|
||
|
|
return logging.getLogger(name)
|