src/ctp_slack_bot/core/logging.py

from logging import __file__ as logging_file, basicConfig, currentframe, ERROR, getLogger, Handler, INFO, LogRecord, WARNING
from loguru import logger
from os import access, getenv, W_OK
from sys import stderr
from typing import Self


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() -> None:
    """
    Configure logging with Loguru.
    
    This function sets up Loguru as the main logging provider, configures the log format based on environment variables,
    and intercepts standard logging messages.
    """

    # Get logger configuration from environment variables.
    log_level = getenv("log_level", "INFO")
    log_format = getenv("log_format", "text")

    # Remove default loguru handler.
    logger.remove()

    # Determine log format.
    if 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=log_level,
        serialize=(log_format == "json"),
        backtrace=True,
        diagnose=True,
    )

    # Add file handler for non-DEBUG environments.
    if log_level != "DEBUG":
        if access('/data', W_OK):
            logger.add(
                "/data/app.log",
                rotation="10 MB",
                retention="1 week",
                compression="zip",
                format=format_string,
                level=log_level,
                serialize=(log_format == "json"),
            )
        else:
            logger.warning("The log directory (/data) is not writable!")

    # Intercept standard logging messages.
    basicConfig(handlers=[InterceptHandler()], level=0, force=True)

    # Update logging levels for some noisy libraries.
    for logger_name in ("logging", "uvicorn", "uvicorn.error", "fastapi", "httpx", "pymongo"):
        getLogger(logger_name).setLevel(INFO)
    for logger_name in ("apscheduler"):
        getLogger(logger_name).setLevel(WARNING)
    getLogger().setLevel(WARNING)

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