"""日志配置工具模块
基于Loguru库提供统一的日志配置，支持控制台输出和文件轮转输出。
使用单例模式确保全局只有一个日志实例。
支持通过 config.yaml 的 logging.debug_modules 配置模块级 DEBUG 屏蔽（黑名单模式）。
"""
from __future__ import annotations
import sys
from collections.abc import Callable
from pathlib import Path
from loguru import logger
class LoggerManager:
"""日志管理器(单例模式)
管理全局日志实例的配置和生命周期，确保系统中只有一个日志实例。
支持控制台输出和文件输出，文件输出支持自动轮转和压缩。
Attributes:
_instance: 单例实例
_initialized: 是否已初始化
_setup_called: 是否已调用setup方法
_console_level: 控制台日志级别
_file_level: 文件日志级别
_log_path: 日志文件路径
Examples:
>>> manager = LoggerManager()
>>> manager.setup(
... log_path="logs/app.log",
... console_level="INFO",
... file_level="DEBUG"
... )
>>> logger = manager.get_logger()
>>> logger.info("应用启动")
"""
_instance: LoggerManager | None = None
_initialized: bool = False
_setup_called: bool = False
def __new__(cls) -> LoggerManager:
"""创建单例实例
Returns:
LoggerManager: 唯一的日志管理器实例
"""
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self) -> None:
"""初始化日志管理器
设置默认的日志配置参数，但不立即配置日志系统。
"""
if self._initialized:
return
self._initialized = True
self._console_level = "INFO"
self._file_level = "DEBUG"
self._log_path = "logs/certflow.log"
self._debug_modules: set[str] = set()
def setup(
self,
log_path: str = "logs/certflow.log",
console_level: str = "INFO",
file_level: str = "DEBUG",
debug_modules: list[str] | None = None,
force_reinit: bool = False,
) -> None:
"""配置日志系统
配置日志输出目标、格式、级别和轮转策略。
如果已经配置过且未强制重新初始化，则跳过重复配置。
Args:
log_path: 日志文件路径，默认为"logs/certflow.log"
console_level: 控制台输出级别，可选: DEBUG/INFO/WARNING/ERROR/CRITICAL
file_level: 文件输出级别，可选: DEBUG/INFO/WARNING/ERROR/CRITICAL
debug_modules: 屏蔽 DEBUG 输出的模块黑名单列表。
- ["*"] 表示屏蔽所有模块的 DEBUG
- [] 或 None 表示所有模块都输出 DEBUG
- ["module"] 表示屏蔽整个模块的 DEBUG
- ["module:function"] 表示仅屏蔽该模块指定函数的 DEBUG
force_reinit: 是否强制重新初始化，默认为False
Examples:
>>> manager = LoggerManager()
>>> # 基本配置
>>> manager.setup()
>>>
>>> # 自定义配置
>>> manager.setup(
... log_path="logs/myapp.log",
... console_level="DEBUG",
... file_level="INFO"
... )
"""
if self._setup_called and not force_reinit:
logger.info("日志系统已初始化，跳过重复配置")
return
self._log_path = log_path
self._console_level = console_level
self._file_level = file_level
self._debug_modules = set(debug_modules or [])
# 确保日志目录存在
Path(log_path).parent.mkdir(parents=True, exist_ok=True)
# 移除所有已有的handler
logger.remove()
# 添加控制台输出（打包后的 GUI 应用无控制台，此时 sys.stdout 为 None）
# 注意：不能直接把 sys.stdout 对象作为 loguru sink 绑定——pytest 在采集阶段会
# 替换并关闭 sys.stdout，导致 loguru 仍持有已关闭的 stdout 引用，在 pytest
# 收尾（stop_global_capturing）时报 'ValueError: I/O operation on closed file.'。
# 改用动态 callable sink，每次写入时解析当前 sys.stdout，兼容 pytest 的 capture 替换。
if sys.stdout is not None:
logger.add(
lambda msg: sys.stdout.write(msg),
format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
level=console_level,
)
# 构建文件 handler 的 DEBUG 过滤器
debug_filter = self._build_debug_filter()
# 添加文件输出
logger.add(
log_path,
rotation="1 day", # 每天轮转一次
retention="30 days", # 保留30天日志
compression="zip", # 轮转后压缩为zip格式
format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | {name}:{function}:{line} - {message}",
level=file_level,
filter=debug_filter,
)
self._setup_called = True
if debug_filter is None:
logger.info("DEBUG 过滤器: 未启用（全部 DEBUG 写入文件）")
else:
logger.info(f"DEBUG 过滤器: 已启用，屏蔽 {len(self._debug_modules)} 个规则")
logger.info(f"日志系统初始化完成 - 控制台: {console_level}, 文件: {file_level}")
logger.info(f"日志文件路径: {log_path}")
if self._debug_modules:
logger.info(f"DEBUG 模块黑名单（已屏蔽）: {sorted(self._debug_modules)}")
def _parse_block_rules(self) -> tuple[set[str], list[tuple[str, str]]]:
"""解析黑名单规则为模块级和函数级两个集合
Returns:
tuple[set[str], list[tuple[str, str]]]: (模块屏蔽集合, 函数屏蔽列表)
"""
module_blocks: set[str] = set()
func_blocks: list[tuple[str, str]] = []
for entry in self._debug_modules:
if ":" in entry:
mod, func = entry.split(":", 1)
func_blocks.append((mod, func))
else:
module_blocks.add(entry)
return module_blocks, func_blocks
def _is_blocked(
self,
record_module: str,
record_function: str,
module_blocks: set[str],
func_blocks: list[tuple[str, str]],
) -> bool:
"""判断记录是否命中黑名单"""
for m in module_blocks:
if record_module.startswith(m):
return True
for mod, func in func_blocks:
if record_module.startswith(mod) and record_function == func:
return True
return False
def _build_debug_filter(self) -> Callable[[dict], bool] | None:
"""构建 DEBUG 日志过滤器（黑名单模式）
屏蔽黑名单中的模块/函数的 DEBUG 日志，其余模块正常输出。
INFO 及以上级别的日志始终通过。
配置格式：
- "certflow.handlers.sorter" → 屏蔽整个模块
- "certflow.handlers.sorter:_process" → 仅屏蔽该函数
- "*" → 屏蔽所有模块
Returns:
callable: filter 函数，返回 True 表示通过
"""
blocked_modules = self._debug_modules
if "*" in blocked_modules:
def _filter(record: dict) -> bool:
return record["level"].name != "DEBUG"
return _filter
if not blocked_modules:
return None
module_blocks, func_blocks = self._parse_block_rules()
def _filter(record: dict) -> bool:
if record["level"].name != "DEBUG":
return True
return not self._is_blocked(
record["name"], record["function"], module_blocks, func_blocks
)
return _filter
def get_logger(self) -> logger.__class__:
"""获取日志实例
Returns:
loguru.Logger: 配置好的日志实例
"""
return logger
def set_console_level(self, level: str) -> None:
"""动态设置控制台日志级别
运行时修改控制台输出级别，不影响文件输出。
保留当前 debug_modules 配置。
Args:
level: 新的日志级别，可选: DEBUG/INFO/WARNING/ERROR/CRITICAL
Examples:
>>> manager = LoggerManager()
>>> manager.set_console_level("DEBUG") # 临时启用调试输出
"""
self._console_level = level
self.setup(
log_path=self._log_path,
console_level=level,
file_level=self._file_level,
debug_modules=list(self._debug_modules),
force_reinit=True,
)
def set_file_level(self, level: str) -> None:
"""动态设置文件日志级别
运行时修改文件输出级别，不影响控制台输出。
保留当前 debug_modules 配置。
Args:
level: 新的日志级别，可选: DEBUG/INFO/WARNING/ERROR/CRITICAL
Examples:
>>> manager = LoggerManager()
>>> manager.set_file_level("WARNING") # 只记录警告及以上级别
"""
self._file_level = level
self.setup(
log_path=self._log_path,
console_level=self._console_level,
file_level=level,
debug_modules=list(self._debug_modules),
force_reinit=True,
)
# 全局单例实例
_logger_manager = LoggerManager()
[文档]
def setup_logger(
log_path: str = "logs/certflow.log",
console_level: str = "INFO",
file_level: str = "DEBUG",
debug_modules: list[str] | None = None,
) -> logger.__class__:
"""配置日志系统(兼容旧接口)
提供简化的函数式接口，用于快速配置日志系统。
Args:
log_path: 日志文件路径，默认为"logs/certflow.log"
console_level: 控制台输出级别，默认为"INFO"
file_level: 文件输出级别，默认为"DEBUG"
debug_modules: DEBUG 模块黑名单，["*"] 屏蔽全部，[] 全部输出
Returns:
loguru.Logger: 配置好的日志实例
Examples:
>>> from certflow.utils.logger import setup_logger
>>>
>>> # 使用默认配置
>>> logger = setup_logger()
>>>
>>> # 自定义配置
>>> logger = setup_logger(
... log_path="logs/app.log",
... console_level="DEBUG",
... file_level="INFO"
... )
>>> logger.info("应用启动成功")
"""
_logger_manager.setup(log_path, console_level, file_level, debug_modules)
return _logger_manager.get_logger()
[文档]
def get_logger():
"""获取全局日志实例
获取已配置的全局日志实例，如果尚未配置则使用默认配置。
Returns:
loguru.Logger: 全局日志实例
Examples:
>>> from certflow.utils.logger import get_logger
>>>
>>> logger = get_logger()
>>> logger.info("这是一条信息日志")
>>> logger.error("这是一条错误日志")
"""
return _logger_manager.get_logger()
# 为了向后兼容，直接导出 logger
__all__ = ["setup_logger", "get_logger", "logger"]