""" 日志配置模块 提供统一的日志配置和管理 """ import logging import sys from pathlib import Path from typing import Optional, Union def _get_log_level(level: Union[str, int]) -> int: """ 将字符串形式的日志级别转换为logging常量 Args: level: 日志级别(字符串或int) Returns: logging级别常量 """ if isinstance(level, int): return level level_map = { "DEBUG": logging.DEBUG, "INFO": logging.INFO, "WARNING": logging.WARNING, "ERROR": logging.ERROR, "CRITICAL": logging.CRITICAL, } return level_map.get(level.upper(), logging.INFO) def _load_logging_config_from_config(): """ 从配置中加载日志设置 Returns: 包含日志配置的字典,如果配置不可用则返回None """ try: from .config import get_config config = get_config() logging_config = config.get_section("logging") if logging_config: return logging_config except Exception: # 如果配置模块不可用或加载失败,返回None pass return None def setup_logger( name: str = "taskflow", level: Union[str, int] = logging.INFO, log_file: Optional[str] = None, format_string: Optional[str] = None, console_output: Optional[bool] = None, use_config: bool = True ) -> logging.Logger: """ 设置并返回配置好的日志记录器 Args: name: 日志记录器名称 level: 日志级别(logging.DEBUG, INFO, WARNING, ERROR, CRITICAL 或字符串形式) log_file: 日志文件路径(如果为None,则尝试从配置中读取) format_string: 日志格式字符串(如果为None,则尝试从配置中读取或使用默认格式) console_output: 是否输出到控制台(如果为None,则尝试从配置中读取) use_config: 是否从配置中读取日志设置(默认True) Returns: 配置好的日志记录器 使用示例: >>> logger = setup_logger(name="my_project", level=logging.DEBUG) >>> logger.info("这是一条信息") >>> logger.error("这是一条错误") """ logger = logging.getLogger(name) # 避免重复添加处理器 if logger.handlers: return logger # 从配置中加载日志设置 config_logging = None if use_config: config_logging = _load_logging_config_from_config() # 确定日志级别 if isinstance(level, str): level = _get_log_level(level) elif config_logging and level == logging.INFO: # 如果使用默认值且配置可用,尝试从配置读取 config_level = config_logging.get("level") if config_level: level = _get_log_level(config_level) logger.setLevel(level) # 确定格式字符串 if format_string is None: if config_logging: format_string = config_logging.get("format") if format_string is None: format_string = '%(asctime)s - %(name)s - %(levelname)s - %(message)s' # 确定日期格式 if config_logging: date_format = config_logging.get("date_format", "%Y-%m-%d %H:%M:%S") else: date_format = '%Y-%m-%d %H:%M:%S' formatter = logging.Formatter(format_string, datefmt=date_format) # 确定是否输出到控制台 if console_output is None: if config_logging: console_output = config_logging.get("console_output", True) else: console_output = True # 控制台处理器 if console_output: console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(level) console_handler.setFormatter(formatter) logger.addHandler(console_handler) # 确定日志文件路径 if log_file is None and config_logging: file_output = config_logging.get("file_output", False) if file_output: log_file = config_logging.get("log_file") # 文件处理器 if log_file: log_path = Path(log_file) log_path.parent.mkdir(parents=True, exist_ok=True) file_handler = logging.FileHandler(log_file, encoding='utf-8') file_handler.setLevel(level) file_handler.setFormatter(formatter) logger.addHandler(file_handler) return logger def get_logger(name: Optional[str] = None) -> logging.Logger: """ 获取日志记录器 如果指定名称的记录器不存在,会创建一个默认配置的记录器。 会自动从配置中读取日志设置。 Args: name: 日志记录器名称(如果为None,使用 "taskflow") Returns: 日志记录器实例 """ if name is None: name = "taskflow" logger = logging.getLogger(name) # 如果记录器还没有配置,从配置中读取设置并创建 if not logger.handlers: logger = setup_logger(name=name, use_config=True) return logger # 创建默认的 taskflow 日志记录器 default_logger = setup_logger(name="taskflow", level=logging.INFO)