228 lines
7.0 KiB
Python
228 lines
7.0 KiB
Python
import structlog
|
||
import logging
|
||
import sys
|
||
import json
|
||
from datetime import datetime
|
||
from typing import Any, Dict, Optional
|
||
from pathlib import Path
|
||
import colorama
|
||
from colorama import Fore, Back, Style
|
||
|
||
# 尝试导入pythonjsonlogger，如果没有安装则使用备用方案
|
||
try:
|
||
from pythonjsonlogger import jsonlogger
|
||
HAS_JSON_LOGGER = True
|
||
except ImportError:
|
||
HAS_JSON_LOGGER = False
|
||
jsonlogger = None
|
||
|
||
# 初始化colorama
|
||
colorama.init()
|
||
|
||
|
||
class ColoredConsoleRenderer:
|
||
"""带颜色的控制台日志渲染器"""
|
||
|
||
def __call__(self, logger, method_name: str, event_dict: Dict[str, Any]) -> str:
|
||
"""渲染日志事件"""
|
||
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
level = event_dict.get("level", "INFO").upper()
|
||
message = event_dict.get("event", "")
|
||
|
||
# 根据日志级别选择颜色
|
||
level_colors = {
|
||
"DEBUG": Fore.CYAN,
|
||
"INFO": Fore.GREEN,
|
||
"WARNING": Fore.YELLOW,
|
||
"ERROR": Fore.RED,
|
||
"CRITICAL": Fore.RED + Back.WHITE + Style.BRIGHT,
|
||
}
|
||
|
||
color = level_colors.get(level, "")
|
||
reset = Style.RESET_ALL
|
||
|
||
# 基础信息
|
||
log_line = f"{color}[{timestamp}] {level}{reset} {message}"
|
||
|
||
# 添加额外的上下文信息
|
||
if "request_id" in event_dict:
|
||
log_line += f" {Fore.BLUE}[req:{event_dict['request_id']}]{reset}"
|
||
|
||
if "function" in event_dict:
|
||
log_line += f" {Fore.MAGIC}{event_dict['function']}(){reset}"
|
||
|
||
# 添加其他字段
|
||
for key, value in event_dict.items():
|
||
if key not in ["level", "event", "timestamp", "request_id", "function"]:
|
||
log_line += f" {Fore.CYAN}{key}={value}{reset}"
|
||
|
||
return log_line
|
||
|
||
|
||
class JSONRenderer:
|
||
"""JSON格式的日志渲染器"""
|
||
|
||
def __call__(self, logger, method_name: str, event_dict: Dict[str, Any]) -> str:
|
||
"""渲染日志事件为JSON格式"""
|
||
log_data = {
|
||
"timestamp": datetime.now().isoformat(),
|
||
"level": event_dict.get("level", "INFO"),
|
||
"message": event_dict.get("event", ""),
|
||
}
|
||
|
||
# 添加其他字段
|
||
for key, value in event_dict.items():
|
||
if key not in ["level", "event"]:
|
||
log_data[key] = value
|
||
|
||
return json.dumps(log_data, ensure_ascii=False, default=str)
|
||
|
||
|
||
class LoggerManager:
|
||
"""日志管理器"""
|
||
|
||
def __init__(self):
|
||
self._processors = []
|
||
self._configured = False
|
||
|
||
def configure(self,
|
||
log_level: str = "INFO",
|
||
log_format: str = "console",
|
||
log_file: Optional[str] = None,
|
||
log_to_console: bool = True):
|
||
"""配置日志系统"""
|
||
|
||
# 配置structlog处理器
|
||
processors = [
|
||
structlog.stdlib.filter_by_level,
|
||
structlog.stdlib.add_logger_name,
|
||
structlog.stdlib.add_log_level,
|
||
structlog.stdlib.PositionalArgumentsFormatter(),
|
||
structlog.processors.TimeStamper(fmt="iso"),
|
||
structlog.processors.StackInfoRenderer(),
|
||
structlog.processors.format_exc_info,
|
||
]
|
||
|
||
# 添加自定义处理器
|
||
processors.extend(self._processors)
|
||
|
||
# 选择渲染器
|
||
if log_format == "json":
|
||
renderer = JSONRenderer()
|
||
else:
|
||
renderer = ColoredConsoleRenderer()
|
||
|
||
processors.append(renderer)
|
||
|
||
# 配置structlog
|
||
structlog.configure(
|
||
processors=processors,
|
||
wrapper_class=structlog.stdlib.BoundLogger,
|
||
logger_factory=structlog.stdlib.LoggerFactory(),
|
||
cache_logger_on_first_use=True,
|
||
)
|
||
|
||
# 配置标准库logging
|
||
# 防止重复配置
|
||
root_logger = logging.getLogger()
|
||
if not root_logger.handlers:
|
||
level = getattr(logging, log_level.upper())
|
||
|
||
# 如果指定了日志文件，配置文件日志
|
||
if log_file:
|
||
self._setup_file_handler(log_file, log_level)
|
||
|
||
# 如果允许控制台输出，配置控制台日志
|
||
if log_to_console:
|
||
console_handler = logging.StreamHandler(sys.stdout)
|
||
console_handler.setLevel(level)
|
||
console_handler.setFormatter(logging.Formatter("%(message)s"))
|
||
root_logger.addHandler(console_handler)
|
||
else:
|
||
# 如果不输出到控制台，至少设置日志级别
|
||
root_logger.setLevel(level)
|
||
|
||
self._configured = True
|
||
|
||
def _setup_file_handler(self, log_file: str, log_level: str = "INFO"):
|
||
"""设置文件日志处理器"""
|
||
log_path = Path(log_file)
|
||
log_path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
file_handler = logging.FileHandler(log_file, encoding='utf-8')
|
||
level = getattr(logging, log_level.upper())
|
||
file_handler.setLevel(level)
|
||
|
||
if HAS_JSON_LOGGER:
|
||
file_handler.setFormatter(jsonlogger.JsonFormatter(
|
||
'%(asctime)s %(name)s %(levelname)s %(message)s'
|
||
))
|
||
else:
|
||
# 使用标准库的JSON格式化器
|
||
file_handler.setFormatter(logging.Formatter(
|
||
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||
))
|
||
|
||
# 获取根logger并添加文件处理器
|
||
root_logger = logging.getLogger()
|
||
root_logger.addHandler(file_handler)
|
||
root_logger.setLevel(level)
|
||
|
||
def add_processor(self, processor):
|
||
"""添加自定义处理器"""
|
||
self._processors.append(processor)
|
||
|
||
def get_logger(self, name: Optional[str] = None) -> structlog.stdlib.BoundLogger:
|
||
"""获取logger实例"""
|
||
if not self._configured:
|
||
self.configure()
|
||
return structlog.get_logger(name)
|
||
|
||
|
||
# 全局日志管理器实例
|
||
logger_manager = LoggerManager()
|
||
|
||
# 获取logger的便捷函数
|
||
def get_logger(name: Optional[str] = None) -> structlog.stdlib.BoundLogger:
|
||
"""获取logger实例"""
|
||
return logger_manager.get_logger(name)
|
||
|
||
|
||
# 简单的日志打印函数
|
||
def log(message: str, level: str = "info", **kwargs):
|
||
"""简单的日志打印函数
|
||
|
||
Args:
|
||
message: 日志消息
|
||
level: 日志级别 (debug, info, warning, error, critical)
|
||
**kwargs: 额外的上下文信息
|
||
"""
|
||
logger = get_logger()
|
||
log_method = getattr(logger, level.lower(), logger.info)
|
||
log_method(message, **kwargs)
|
||
|
||
|
||
# 带上下文的日志函数
|
||
def log_debug(message: str, **kwargs):
|
||
"""打印debug级别日志"""
|
||
log(message, "debug", **kwargs)
|
||
|
||
|
||
def log_info(message: str, **kwargs):
|
||
"""打印info级别日志"""
|
||
log(message, "info", **kwargs)
|
||
|
||
|
||
def log_warning(message: str, **kwargs):
|
||
"""打印warning级别日志"""
|
||
log(message, "warning", **kwargs)
|
||
|
||
|
||
def log_error(message: str, **kwargs):
|
||
"""打印error级别日志"""
|
||
log(message, "error", **kwargs)
|
||
|
||
|
||
def log_critical(message: str, **kwargs):
|
||
"""打印critical级别日志"""
|
||
log(message, "critical", **kwargs) |