import logging import sys from typing import Dict, Union from loguru import logger from ctp_slack_bot.core.config import settings class InterceptHandler(logging.Handler): """ Intercept standard logging messages toward Loguru. This handler intercepts all standard logging messages and redirects them to Loguru, allowing unified logging across the application. """ def emit(self, record: logging.LogRecord) -> None: # Get corresponding Loguru level if it exists try: level = logger.level(record.levelname).name except ValueError: level = record.levelno # Find caller from where the logged message originated frame, depth = logging.currentframe(), 2 while frame and frame.f_code.co_filename == logging.__file__: frame = frame.f_back depth += 1 logger.opt(depth=depth, exception=record.exc_info).log( level, record.getMessage() ) def setup_logging() -> None: """ Configure logging with Loguru. This function sets up Loguru as the main logging provider, configures the log format based on settings, and intercepts standard logging messages. """ # Remove default loguru handler logger.remove() # Determine log format if settings.LOG_FORMAT == "json": log_format = { "time": "{time:YYYY-MM-DD HH:mm:ss.SSS}", "level": "{level}", "message": "{message}", "module": "{module}", "function": "{function}", "line": "{line}", } format_string = lambda record: record["message"] else: format_string = ( "{time:YYYY-MM-DD HH:mm:ss.SSS} | " "{level: <8} | " "{name}:{function}:{line} - " "{message}" ) # Add console handler logger.add( sys.stderr, format=format_string, level=settings.LOG_LEVEL, serialize=(settings.LOG_FORMAT == "json"), backtrace=True, diagnose=True, ) # Add file handler for non-DEBUG environments if settings.LOG_LEVEL != "DEBUG": logger.add( "logs/app.log", rotation="10 MB", retention="1 week", compression="zip", format=format_string, level=settings.LOG_LEVEL, serialize=(settings.LOG_FORMAT == "json"), ) # Intercept standard logging messages logging.basicConfig(handlers=[InterceptHandler()], level=0, force=True) # Update logging levels for some noisy libraries for logger_name in [ "uvicorn", "uvicorn.error", "fastapi", "httpx", "apscheduler", "pymongo", ]: logging.getLogger(logger_name).setLevel(logging.INFO) logger.info(f"Logging configured with level {settings.LOG_LEVEL}")