import logging import os from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler from typing import Union class Logger: """ 日志工具类封装 功能: 1. 支持控制台和文件两种输出方式 2. 支持按大小或时间轮转日志文件 3. 支持自定义日志格式 4. 支持不同日志级别 5. 线程安全 """ def __init__( self, name: str = "root", level: Union[int, str] = logging.INFO, console: bool = True, file: bool = False, file_path: str = "logs/app.log", max_bytes: int = 10 * 1024 * 1024, # 10MB backup_count: int = 5, when: str = "midnight", interval: int = 1, fmt: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s", datefmt: str = "%Y-%m-%d %H:%M:%S", mode: str = "size" # 'size' 或 'time' ): """ 初始化日志工具 :param name: 日志名称 :param level: 日志级别 :param console: 是否输出到控制台 :param file: 是否输出到文件 :param file_path: 日志文件路径 :param max_bytes: 每个日志文件的最大大小(字节),仅mode='size'时有效 :param backup_count: 保留的备份日志文件数量 :param when: 日志轮转时间单位,如'S'(秒)、'M'(分)、'H'(小时)、'D'(天)、'midnight'(午夜),仅mode='time'时有效 :param interval: 轮转间隔,仅mode='time'时有效 :param fmt: 日志格式 :param datefmt: 日期格式 :param mode: 日志轮转模式,'size'按大小轮转,'time'按时间轮转 """ self.logger = logging.getLogger(name) self.logger.setLevel(level) # 避免重复添加handler if self.logger.handlers: return formatter = logging.Formatter(fmt=fmt, datefmt=datefmt) # 控制台输出 if console: console_handler = logging.StreamHandler() console_handler.setFormatter(formatter) self.logger.addHandler(console_handler) # 文件输出 if file: # 创建日志目录 log_dir = os.path.dirname(file_path) if log_dir and not os.path.exists(log_dir): os.makedirs(log_dir) if mode == "size": # 按大小轮转 file_handler = RotatingFileHandler( filename=file_path, maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8" ) else: # 按时间轮转 file_handler = TimedRotatingFileHandler( filename=file_path, when=when, interval=interval, backupCount=backup_count, encoding="utf-8" ) file_handler.setFormatter(formatter) self.logger.addHandler(file_handler) def debug(self, msg: str, *args, **kwargs): """记录调试信息""" self.logger.debug(msg, *args, **kwargs) def info(self, msg: str, *args, **kwargs): """记录普通信息""" self.logger.info(msg, *args, **kwargs) def warning(self, msg: str, *args, **kwargs): """记录警告信息""" self.logger.warning(msg, *args, **kwargs) def error(self, msg: str, *args, **kwargs): """记录错误信息""" self.logger.error(msg, *args, **kwargs) def critical(self, msg: str, *args, **kwargs): """记录严重错误信息""" self.logger.critical(msg, *args, **kwargs) def exception(self, msg: str, *args, exc_info=True, **kwargs): """记录异常信息""" self.logger.exception(msg, *args, exc_info=exc_info, **kwargs) def log(self, level: int, msg: str, *args, **kwargs): """通用日志记录方法""" self.logger.log(level, msg, *args, **kwargs) def set_level(self, level: Union[int, str]): """设置日志级别""" self.logger.setLevel(level) def add_handler(self, handler: logging.Handler): """添加自定义handler""" self.logger.addHandler(handler) def remove_handler(self, handler: logging.Handler): """移除handler""" self.logger.removeHandler(handler) logger: Logger def app_logger() -> Logger: return logger def init() -> Logger: # 创建日志实例 global logger logger = Logger( name="my_app", level=logging.DEBUG, console=True, file=True, file_path="logs/app.log", max_bytes=1024 * 1024, # 1MB backup_count=3, mode="size" ) return logger