qichi.liang
84e907b93e
- 日志文件按日期分片存储 (logs/YYYY-MM/YYYY-MM-DD.log) - 增强尺寸箱量解析支持多种格式 (*和×分隔符) - 支持带括号和无括号两种尺寸格式
172 lines
5.1 KiB
Python
172 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
统一日志配置模块
|
||
提供统一的日志配置，避免各模块自行配置
|
||
支持按日期分片存储日志
|
||
"""
|
||
import os
|
||
import logging
|
||
import sys
|
||
from datetime import datetime
|
||
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler
|
||
from typing import Optional
|
||
|
||
from src.config import config
|
||
|
||
|
||
def setup_logging(
|
||
log_file: Optional[str] = None,
|
||
console_level: int = logging.INFO,
|
||
file_level: int = logging.DEBUG,
|
||
use_date_split: bool = True,
|
||
date_folder_format: str = "%Y-%m", # 按月份分文件夹
|
||
max_bytes: int = 10 * 1024 * 1024, # 10MB
|
||
backup_count: int = 5
|
||
) -> logging.Logger:
|
||
"""
|
||
配置统一的日志系统
|
||
|
||
参数:
|
||
log_file: 日志文件路径，如果为None则使用默认路径
|
||
console_level: 控制台日志级别
|
||
file_level: 文件日志级别
|
||
use_date_split: 是否使用日期分片
|
||
date_folder_format: 日期文件夹格式（默认按月份，如 logs/2025-12/）
|
||
max_bytes: 单个日志文件最大大小
|
||
backup_count: 备份文件数量
|
||
|
||
返回:
|
||
配置好的根日志器
|
||
"""
|
||
# 获取当前日期用于构建路径
|
||
now = datetime.now()
|
||
|
||
if log_file is None:
|
||
log_dir = 'logs'
|
||
if use_date_split:
|
||
# 按日期分片：logs/2025-12/2025-12-30.log
|
||
date_folder = now.strftime(date_folder_format)
|
||
log_dir = os.path.join('logs', date_folder)
|
||
log_file = os.path.join(log_dir, now.strftime('%Y-%m-%d.log'))
|
||
else:
|
||
log_file = os.path.join(log_dir, 'app.log')
|
||
else:
|
||
log_dir = os.path.dirname(log_file)
|
||
|
||
if log_dir and not os.path.exists(log_dir):
|
||
os.makedirs(log_dir, exist_ok=True)
|
||
|
||
# 获取根日志器
|
||
logger = logging.getLogger()
|
||
logger.setLevel(logging.DEBUG) # 根日志器设置为最低级别
|
||
|
||
# 清除现有handler，避免重复添加
|
||
logger.handlers.clear()
|
||
|
||
# 控制台handler
|
||
console_handler = logging.StreamHandler(sys.stdout)
|
||
console_handler.setLevel(console_level)
|
||
console_formatter = logging.Formatter(
|
||
'%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
||
datefmt='%Y-%m-%d %H:%M:%S'
|
||
)
|
||
console_handler.setFormatter(console_formatter)
|
||
logger.addHandler(console_handler)
|
||
|
||
# 文件handler（日期分片或大小轮转）
|
||
if use_date_split:
|
||
# 使用TimedRotatingFileHandler，每天午夜轮转
|
||
file_handler = TimedRotatingFileHandler(
|
||
log_file,
|
||
when='midnight',
|
||
interval=1,
|
||
backupCount=backup_count,
|
||
encoding='utf-8',
|
||
atTime=datetime.strptime('00:00:00', '%H:%M:%S')
|
||
)
|
||
logger.info(f"日志系统已初始化，使用日期分片: {log_file}")
|
||
else:
|
||
# 使用RotatingFileHandler，按大小轮转
|
||
file_handler = RotatingFileHandler(
|
||
log_file,
|
||
maxBytes=max_bytes,
|
||
backupCount=backup_count,
|
||
encoding='utf-8'
|
||
)
|
||
logger.info(f"日志系统已初始化，使用大小轮转: {log_file}")
|
||
|
||
file_handler.setLevel(file_level)
|
||
file_formatter = logging.Formatter(
|
||
'%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s',
|
||
datefmt='%Y-%m-%d %H:%M:%S'
|
||
)
|
||
file_handler.setFormatter(file_formatter)
|
||
logger.addHandler(file_handler)
|
||
|
||
# 设置第三方库的日志级别
|
||
logging.getLogger('urllib3').setLevel(logging.WARNING)
|
||
logging.getLogger('requests').setLevel(logging.WARNING)
|
||
|
||
logger.info(f"控制台日志级别: {logging.getLevelName(console_level)}")
|
||
logger.info(f"文件日志级别: {logging.getLevelName(file_level)}")
|
||
|
||
return logger
|
||
|
||
|
||
def get_logger(name: str) -> logging.Logger:
|
||
"""
|
||
获取指定名称的日志器
|
||
|
||
参数:
|
||
name: 日志器名称，通常使用 __name__
|
||
|
||
返回:
|
||
配置好的日志器
|
||
"""
|
||
return logging.getLogger(name)
|
||
|
||
|
||
# 自动初始化日志系统
|
||
if not logging.getLogger().handlers:
|
||
# 只有在没有handler时才初始化，避免重复初始化
|
||
setup_logging()
|
||
|
||
|
||
# 便捷函数
|
||
def info(msg: str, *args, **kwargs):
|
||
"""记录INFO级别日志"""
|
||
logging.info(msg, *args, **kwargs)
|
||
|
||
|
||
def warning(msg: str, *args, **kwargs):
|
||
"""记录WARNING级别日志"""
|
||
logging.warning(msg, *args, **kwargs)
|
||
|
||
|
||
def error(msg: str, *args, **kwargs):
|
||
"""记录ERROR级别日志"""
|
||
logging.error(msg, *args, **kwargs)
|
||
|
||
|
||
def debug(msg: str, *args, **kwargs):
|
||
"""记录DEBUG级别日志"""
|
||
logging.debug(msg, *args, **kwargs)
|
||
|
||
|
||
def exception(msg: str, *args, **kwargs):
|
||
"""记录异常日志"""
|
||
logging.exception(msg, *args, **kwargs)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
# 测试日志配置
|
||
logger = get_logger(__name__)
|
||
logger.info("测试INFO日志")
|
||
logger.warning("测试WARNING日志")
|
||
logger.error("测试ERROR日志")
|
||
logger.debug("测试DEBUG日志")
|
||
|
||
try:
|
||
raise ValueError("测试异常")
|
||
except ValueError as e:
|
||
logger.exception("捕获到异常: %s", e) |