from loguru import logger
from typing import Dict, Optional, Union, List
import sys
import os
from types import ModuleType
from pathlib import Path
from dotenv import load_dotenv
# from ..plugins.chat.config import global_config
load_dotenv()
# 保存原生处理器ID
default_handler_id = None
for handler_id in logger._core.handlers:
default_handler_id = handler_id
break
# 移除默认处理器
if default_handler_id is not None:
logger.remove(default_handler_id)
# 类型别名
LoguruLogger = logger.__class__
# 全局注册表:记录模块与处理器ID的映射
_handler_registry: Dict[str, List[int]] = {}
# 获取日志存储根地址
current_file_path = Path(__file__).resolve()
LOG_ROOT = "logs"
# 从环境变量获取是否启用高级输出
# ENABLE_ADVANCE_OUTPUT = True
ENABLE_ADVANCE_OUTPUT = False
if ENABLE_ADVANCE_OUTPUT:
# 默认全局配置
DEFAULT_CONFIG = {
# 日志级别配置
"console_level": "INFO",
"file_level": "DEBUG",
# 格式配置
"console_format": (
"{time:YYYY-MM-DD HH:mm:ss} | "
"{level: <8} | "
"{extra[module]: <12} | "
"{message}"
),
"file_format": (
"{time:YYYY-MM-DD HH:mm:ss} | "
"{level: <8} | "
"{extra[module]: <15} | "
"{message}"
),
"log_dir": LOG_ROOT,
"rotation": "00:00",
"retention": "3 days",
"compression": "zip",
}
else:
DEFAULT_CONFIG = {
# 日志级别配置
"console_level": "INFO",
"file_level": "DEBUG",
# 格式配置
"console_format": (
"{time:MM-DD HH:mm} | "
"{extra[module]} | "
"{message}"
),
"file_format": (
"{time:YYYY-MM-DD HH:mm:ss} | "
"{level: <8} | "
"{extra[module]: <15} | "
"{message}"
),
"log_dir": LOG_ROOT,
"rotation": "00:00",
"retention": "3 days",
"compression": "zip",
}
# 控制nonebot日志输出的环境变量
NONEBOT_LOG_ENABLED = False
# 海马体日志样式配置
MEMORY_STYLE_CONFIG = {
"advanced": {
"console_format": (
"{time:YYYY-MM-DD HH:mm:ss} | "
"{level: <8} | "
"{extra[module]: <12} | "
"海马体 | "
"{message}"
),
"file_format": (
"{time:YYYY-MM-DD HH:mm:ss} | "
"{level: <8} | "
"{extra[module]: <15} | "
"海马体 | "
"{message}"
)
},
"simple": {
"console_format": (
"{time:MM-DD HH:mm} | "
"海马体 | "
"{message}"
),
"file_format": (
"{time:YYYY-MM-DD HH:mm:ss} | "
"{level: <8} | "
"{extra[module]: <15} | "
"海马体 | "
"{message}"
)
}
}
# Topic日志样式配置
TOPIC_STYLE_CONFIG = {
"advanced": {
"console_format": (
"{time:YYYY-MM-DD HH:mm:ss} | "
"{level: <8} | "
"{extra[module]: <12} | "
"话题 | "
"{message}"
),
"file_format": (
"{time:YYYY-MM-DD HH:mm:ss} | "
"{level: <8} | "
"{extra[module]: <15} | "
"话题 | "
"{message}"
)
},
"simple": {
"console_format": (
"{time:MM-DD HH:mm} | "
"主题 | "
"{message}"
),
"file_format": (
"{time:YYYY-MM-DD HH:mm:ss} | "
"{level: <8} | "
"{extra[module]: <15} | "
"话题 | "
"{message}"
)
}
}
# 根据ENABLE_ADVANCE_OUTPUT选择配置
MEMORY_STYLE_CONFIG = MEMORY_STYLE_CONFIG["advanced"] if ENABLE_ADVANCE_OUTPUT else MEMORY_STYLE_CONFIG["simple"]
TOPIC_STYLE_CONFIG = TOPIC_STYLE_CONFIG["advanced"] if ENABLE_ADVANCE_OUTPUT else TOPIC_STYLE_CONFIG["simple"]
def filter_nonebot(record: dict) -> bool:
"""过滤nonebot的日志"""
return record["extra"].get("module") != "nonebot"
def is_registered_module(record: dict) -> bool:
"""检查是否为已注册的模块"""
return record["extra"].get("module") in _handler_registry
def is_unregistered_module(record: dict) -> bool:
"""检查是否为未注册的模块"""
return not is_registered_module(record)
def log_patcher(record: dict) -> None:
"""自动填充未设置模块名的日志记录,保留原生模块名称"""
if "module" not in record["extra"]:
# 尝试从name中提取模块名
module_name = record.get("name", "")
if module_name == "":
module_name = "root"
record["extra"]["module"] = module_name
# 应用全局修补器
logger.configure(patcher=log_patcher)
class LogConfig:
"""日志配置类"""
def __init__(self, **kwargs):
self.config = DEFAULT_CONFIG.copy()
self.config.update(kwargs)
def to_dict(self) -> dict:
return self.config.copy()
def update(self, **kwargs):
self.config.update(kwargs)
def get_module_logger(
module: Union[str, ModuleType],
*,
console_level: Optional[str] = None,
file_level: Optional[str] = None,
extra_handlers: Optional[List[dict]] = None,
config: Optional[LogConfig] = None
) -> LoguruLogger:
module_name = module if isinstance(module, str) else module.__name__
current_config = config.config if config else DEFAULT_CONFIG
# 清理旧处理器
if module_name in _handler_registry:
for handler_id in _handler_registry[module_name]:
logger.remove(handler_id)
del _handler_registry[module_name]
handler_ids = []
# 控制台处理器
console_id = logger.add(
sink=sys.stderr,
level=os.getenv("CONSOLE_LOG_LEVEL", console_level or current_config["console_level"]),
format=current_config["console_format"],
filter=lambda record: record["extra"].get("module") == module_name,
enqueue=True,
)
handler_ids.append(console_id)
# 文件处理器
log_dir = Path(current_config["log_dir"])
log_dir.mkdir(parents=True, exist_ok=True)
log_file = log_dir / module_name / f"{{time:YYYY-MM-DD}}.log"
log_file.parent.mkdir(parents=True, exist_ok=True)
file_id = logger.add(
sink=str(log_file),
level=os.getenv("FILE_LOG_LEVEL", file_level or current_config["file_level"]),
format=current_config["file_format"],
rotation=current_config["rotation"],
retention=current_config["retention"],
compression=current_config["compression"],
encoding="utf-8",
filter=lambda record: record["extra"].get("module") == module_name,
enqueue=True,
)
handler_ids.append(file_id)
# 额外处理器
if extra_handlers:
for handler in extra_handlers:
handler_id = logger.add(**handler)
handler_ids.append(handler_id)
# 更新注册表
_handler_registry[module_name] = handler_ids
return logger.bind(module=module_name)
def remove_module_logger(module_name: str) -> None:
"""清理指定模块的日志处理器"""
if module_name in _handler_registry:
for handler_id in _handler_registry[module_name]:
logger.remove(handler_id)
del _handler_registry[module_name]
# 添加全局默认处理器(只处理未注册模块的日志--->控制台)
DEFAULT_GLOBAL_HANDLER = logger.add(
sink=sys.stderr,
level=os.getenv("DEFAULT_CONSOLE_LOG_LEVEL", "SUCCESS"),
format=(
"{time:YYYY-MM-DD HH:mm:ss} | "
"{level: <8} | "
"{name: <12} | "
"{message}"
),
filter=lambda record: is_unregistered_module(record) and filter_nonebot(record), # 只处理未注册模块的日志,并过滤nonebot
enqueue=True,
)
# 添加全局默认文件处理器(只处理未注册模块的日志--->logs文件夹)
log_dir = Path(DEFAULT_CONFIG["log_dir"])
log_dir.mkdir(parents=True, exist_ok=True)
other_log_dir = log_dir / "other"
other_log_dir.mkdir(parents=True, exist_ok=True)
DEFAULT_FILE_HANDLER = logger.add(
sink=str(other_log_dir / f"{{time:YYYY-MM-DD}}.log"),
level=os.getenv("DEFAULT_FILE_LOG_LEVEL", "DEBUG"),
format=(
"{time:YYYY-MM-DD HH:mm:ss} | "
"{level: <8} | "
"{name: <15} | "
"{message}"
),
rotation=DEFAULT_CONFIG["rotation"],
retention=DEFAULT_CONFIG["retention"],
compression=DEFAULT_CONFIG["compression"],
encoding="utf-8",
filter=lambda record: is_unregistered_module(record) and filter_nonebot(record), # 只处理未注册模块的日志,并过滤nonebot
enqueue=True,
)