import logging import os import sys from loguru import logger as loguru_logger from yuxi.utils.datetime_utils import shanghai_now SAVE_DIR = "saves" DATETIME = shanghai_now().strftime("%Y-%m-%d") LOG_FILE = f"{SAVE_DIR}/logs/yuxi-{DATETIME}.log" class LoguruHandler(logging.Handler): """将 Python logging 桥接到 loguru 的 handler""" def emit(self, record: logging.LogRecord): level_map = { logging.DEBUG: "DEBUG", logging.INFO: "INFO", logging.WARNING: "WARNING", logging.ERROR: "ERROR", logging.CRITICAL: "CRITICAL", } level = level_map.get(record.levelno, "DEBUG") try: msg = self.format(record) except Exception: msg = record.getMessage() loguru_logger.opt(depth=1, exception=record.exc_info).log(level, msg) def _setup_logging_bridge(): """配置 logging 到 loguru 的桥接,捕获第三方库日志""" loguru_handler = LoguruHandler() loguru_handler.setLevel(logging.DEBUG) formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s") loguru_handler.setFormatter(formatter) for lib in ["httpx", "openai", "neo4j", "urllib3"]: lib_logger = logging.getLogger(lib) lib_logger.addHandler(loguru_handler) lib_logger.setLevel(logging.WARNING) lib_logger.propagate = False def setup_logger(name, level="DEBUG", console=True): """使用 loguru 设置日志记录器""" os.makedirs(f"{SAVE_DIR}/logs", exist_ok=True) # 移除默认的 handler loguru_logger.remove() # 添加文件日志(无颜色) loguru_logger.add( LOG_FILE, level=level, format="{time:YYYY-MM-DD HH:mm:ss} - {level} - {file}:{line} - {message}", encoding="utf-8", rotation="10 MB", retention="30 days", compression="zip", enqueue=True, ) # 添加控制台日志(有颜色) if console: loguru_logger.add( sys.stderr, level=level, format=( "{time:MM-DD HH:mm:ss} " "{level} " "{file}:{line}: " "{message}" ), colorize=True, enqueue=True, ) return loguru_logger # 设置根日志记录器 logger = setup_logger("Yuxi") # 初始化 logging 桥接 _setup_logging_bridge() __all__ = ["logger"]