src/utils/logging.py

"""
Centralized logging configuration for the Executive Education RAG Chatbot.
"""
from threading import Lock
import logging, os, sys, warnings, copy
from pathlib import Path
from typing import Optional

import colorama
from colorama import Fore, Style

# Initialize colorama for cross-platform color support
colorama.init()

class CodeBlockFormatter(logging.Formatter):
    ALIASES = {
        'DEBUG':    'DEBUG',
        'INFO':     'INFO ',
        'WARNING':  'WARN ',
        'ERROR':    'ERROR',
        'CRITICAL': 'CRITC'
    }

    def format(self, record):
        if hasattr(record, 'levelname') and record.levelname in self.ALIASES:
            record.levelname = self.ALIASES[record.levelname]

        if hasattr(record, 'name'):
            rname = record.name 
            if len(rname) <= 17: rname += ' '*(17-len(rname))
            else: rname = rname[14:] + '...'
            record.name = rname
        
        return super().format(record)


class ColoredFormatter(logging.Formatter):
    """Custom formatter with color support for console output."""
    
    COLORS = {
        'DEBUG':    Fore.CYAN,
        'INFO':     Fore.GREEN,
        'WARNING':  Fore.YELLOW,
        'ERROR':    Fore.RED,
        'CRITICAL': Fore.MAGENTA + Style.BRIGHT,
    }
    ALIASES = {
        'DEBUG':    'DEBUG',
        'INFO':     'INFO ',
        'WARNING':  'WARN ',
        'ERROR':    'ERROR',
        'CRITICAL': 'CRITC'
    }
    
    def format(self, record):
        # Add color to the level name
        if hasattr(record, 'levelname') and record.levelname in self.COLORS:
            lname = record.levelname
            if hasattr(record, 'message') and lname == 'ERROR':
                record.message = f"{self.COLORS[lname]}{record.message}{Style.RESET_ALL}"

            record.levelname = f"{self.COLORS[lname]}{self.ALIASES[lname]}{Style.RESET_ALL}"
            

        if hasattr(record, 'name'):
            rname = record.name if len(record.name) <= 17 else record.name[:14] + '...'
            record.name = f"{Fore.CYAN}{rname}{Style.RESET_ALL}"

        return super().format(record)


class CachedLogHandler(logging.Handler):
    def __init__(self, level = 0) -> None:
        super().__init__(level)
        self._cache = []
        self._max_lines = 50
        self._lock = Lock()


    def emit(self, record):
        record_copy = copy.copy(record)
        msg = self.format(record_copy)
        with self._lock:
            self._cache.append(msg)

    def get_logs(self):
        with self._lock:
            return '\n'.join(self._cache)


cached_log_handler: CachedLogHandler = CachedLogHandler()

def setup_logging(
    level: str = "INFO",
    log_file: Optional[str] = None,
    interactive_mode: bool = False,
    module_name: Optional[str] = None
) -> logging.Logger:
    """
    Set up centralized logging configuration.
    
    Args:
        level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
        log_file: Optional path to log file. If None, uses default location.
        interactive_mode: If True, logs only to file in interactive mode
        module_name: Name of the module requesting the logger
    
    Returns:
        Configured logger instance
    """
    global cached_log_handler

    # Convert string level to logging constant
    numeric_level = getattr(logging, level.upper(), logging.INFO)
    
    # Create logger
    logger = logging.getLogger()
    
    # Avoid duplicate handlers if logger already configured
    if logger.handlers:
        logger.handlers.clear()
    
    logger.setLevel(numeric_level)
    
    # Create formatters
    detailed_formatter = logging.Formatter(
        "(%(asctime)s) %(name)s\t %(levelname)s: %(message)s",
        datefmt="%Y.%m.%d %H:%M:%S"
    )
    
    colored_formatter = ColoredFormatter(
        "(%(asctime)s) %(name)s\t %(levelname)s: %(message)s",
        datefmt="%Y.%m.%d %H:%M:%S"
    )

    codeblock_formatter = CodeBlockFormatter(
       "%(levelname)s %(name)s\t : %(message)s", 
    )
    
    # Set up file logging
    if log_file or interactive_mode:
        if not log_file:
            # Default log file location
            log_dir = Path("logs")
            log_dir.mkdir(exist_ok=True)
            log_file = log_dir / "rag_chatbot.log"
        
        file_handler = logging.FileHandler(log_file, encoding='utf-8')
        file_handler.setLevel(numeric_level)
        file_handler.setFormatter(detailed_formatter)
        logger.addHandler(file_handler)
    
    # Set up the cached log handler for gradio Code block output
    cached_log_handler.setLevel(numeric_level)
    cached_log_handler.setFormatter(codeblock_formatter)

    # Set up console logging (unless in interactive mode)
    if not interactive_mode:
        console_handler = logging.StreamHandler(sys.stdout)
        console_handler.setLevel(numeric_level)
        
        # Use colored formatter if terminal supports it
        if _supports_color():
            console_handler.setFormatter(colored_formatter)
        else:
            console_handler.setFormatter(detailed_formatter)
        
        logger.addHandler(cached_log_handler)
        logger.addHandler(console_handler)
    
    return logger


def get_logger(module_name: str) -> logging.Logger:
    """
    Get a logger for a specific module.
    
    Args:
        module_name: Name of the module requesting the logger
    
    Returns:
        Logger instance
    """
    logger = logging.getLogger(module_name)
    logger.propagate = True       
    return logger


def _supports_color() -> bool:
    """
    Check if the terminal supports color output.
    
    Returns:
        True if color is supported, False otherwise
    """
    # Check if we're in a terminal
    if not hasattr(sys.stdout, 'isatty') or not sys.stdout.isatty():
        return False
    
    # Check environment variables
    if os.getenv('NO_COLOR'):
        return False
    
    if os.getenv('FORCE_COLOR'):
        return True
    
    # Check terminal type
    term = os.getenv('TERM', '').lower()
    if 'color' in term or term in ['xterm', 'xterm-256color', 'screen']:
        return True
    
    return False


def detect_interactive_mode() -> bool:
    """
    Detect if the application is running in interactive mode.
    
    Returns:
        True if in interactive mode, False otherwise
    """
    # Check if no command line arguments were provided (except script name)
    if len(sys.argv) == 1:
        return True
    
    # Check if only the script name and no other arguments
    # This indicates the chatbot is running in default interactive mode
    return False


def configure_external_loggers(level: str = "WARNING") -> None:
    """
    Configure logging for external libraries to reduce noise.
    
    Args:
        level: Logging level for external libraries
    """
    external_loggers = [
        'selenium',
        'urllib3',
        'requests',
        'chromadb',
        'docling',
        'weaviate',
        'langchain',
        'langgraph',
        'openai',
        'httpx'
    ]
    
    numeric_level = getattr(logging, level.upper(), logging.WARNING)
    
    for logger_name in external_loggers:
        logging.getLogger(logger_name).setLevel(numeric_level)


# Global configuration function
def init_logging(
    level: str = "INFO",
    log_file: Optional[str] = None,
    interactive_mode: Optional[bool] = None
) -> None:
    """
    Initialize the global logging configuration.
    
    Args:
        level: Logging level
        log_file: Optional log file path
        interactive_mode: If None, auto-detect interactive mode
    """
    if interactive_mode is None:
        interactive_mode = detect_interactive_mode()
    
    warnings.filterwarnings("ignore")

    # Set up root logger
    setup_logging(
        level=level,
        log_file=log_file,
        interactive_mode=interactive_mode
    )
    
    # Configure external library loggers
    configure_external_loggers()