movie-create/taskflow/logger.py

"""
日志配置模块
提供统一的日志配置和管理
"""

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)