src/ctp_slack_bot/core/logging.py

from dependency_injector.wiring import Provide
from logging import __file__ as logging_file, basicConfig, currentframe, getLogger, Handler, INFO, LogRecord
from loguru import logger
from sys import stderr
from typing import Self

from ctp_slack_bot.containers import Container

class InterceptHandler(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: Self, record: 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 = 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(container: "Container") -> None: # TODO: Perhaps get rid of the container dependence since we only need two settings.
    """
    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.
    """
    from ctp_slack_bot.containers import Container
    settings = container.settings() if container else Provide[Container.settings]

    # 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 = (
            "<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | "
            "<level>{level: <8}</level> | "
            "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
            "<level>{message}</level>"
        )

    # Add console handler
    logger.add(
        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
    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"):
        getLogger(logger_name).setLevel(INFO)

    logger.info(f"Logging configured with level {settings.LOG_LEVEL}")