logger/custom_logger.py

import logging
import os
import yaml
from datetime import datetime
import sys
from typing import List

class CustomLoggerTracker:
    _instance = None
    _initialized = False
    
    def __new__(cls, config_path='logging_config.yaml'):
        """Singleton pattern to ensure only one instance exists."""
        if cls._instance is None:
            cls._instance = super(CustomLoggerTracker, cls).__new__(cls)
        return cls._instance
    
    def __init__(self, config_path='logging_config.yaml'):
        """Initialize the custom logger with configuration."""
        if self._initialized:
            return
            
        self.config = self._load_config(config_path)
        self.loggers = {}
        self.base_log_dir = self.config.get('base_log_dir', 'logs')
        self._setup_base_directory()
        self._initialized = True
        
    def _load_config(self, config_path):
        """Load configuration from YAML file."""
        try:
            with open(config_path, 'r') as file:
                return yaml.safe_load(file)
        except FileNotFoundError:
            # Default configuration if file not found
            return {
                'base_log_dir': 'logs',
                'default_level': 'INFO',
                'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s',
                'console_output': True,
                'modules': {
                    'main': {'level': 'INFO'},
                    'utils': {'level': 'INFO'},
                    'old_docs': {'level': 'INFO'},
                    'rag': {'level': 'INFO'},
                    'query_utils': {'level': 'INFO'},
                    'prompt_temp': {'level': 'INFO'}
                }
            }

    def _setup_base_directory(self):
        """Setup the base directory structure for logs."""
        if not os.path.exists(self.base_log_dir):
            os.makedirs(self.base_log_dir)

    def _get_log_path(self, module_name):
        """Generate the hierarchical path for log files."""
        now = datetime.now()
        year_dir = os.path.join(self.base_log_dir, str(now.year))
        month_dir = os.path.join(year_dir, f"{now.month:02d}")
        day_dir = os.path.join(month_dir, f"{now.day:02d}")
        os.makedirs(day_dir, exist_ok=True)
        return os.path.join(day_dir, f"{module_name}.log")

    def get_logger(self, module_name):
        """Get or create a logger for a specific module."""
        if module_name in self.loggers:
            return self.loggers[module_name]

        # Create new logger & Models Specific Config
        logger = logging.getLogger(module_name)
        module_config = self.config['modules'].get(module_name, {})
        level = getattr(logging, module_config.get('level', self.config['default_level']))
        logger.setLevel(level)

        # Clear existing handlers to avoid duplicates
        logger.handlers.clear()

        # Create formatter
        formatter = logging.Formatter(self.config.get('format'))

        # Create file handler with the hierarchical path
        log_path = self._get_log_path(module_name)
        file_handler = logging.FileHandler(log_path)
        file_handler.setFormatter(formatter)
        logger.addHandler(file_handler)

        # Optionally add console handler
        if self.config.get('console_output', True):
            console_handler = logging.StreamHandler()
            console_handler.setFormatter(formatter)
            logger.addHandler(console_handler)

        # Prevent propagation to avoid duplicate messages
        logger.propagate = False
        
        self.loggers[module_name] = logger
        return logger

    def update_config(self, new_config):
        """Update logger configuration."""
        self.config.update(new_config)
        # Reset all loggers to apply new configuration
        for module_name in self.loggers:
            logger = self.loggers[module_name]
            for handler in logger.handlers[:]:
                logger.removeHandler(handler)
        self.loggers = {}

    def log_message(self, process_log: List[str], message: str, level: str = "info", module: str = "default") -> None:
        """
        Append to process_log AND send to the central logger.
        
        Args:
            process_log: List to append the message to
            message: The message to log
            level: Log level ('info', 'warning', 'error')
            module: Module name for the logger (optional, defaults to 'default')
        """
        process_log.append(message)
        
        # Get the logger for the specified module
        logger = self.get_logger(module)
        
        # Log the message at the appropriate level
        if level.lower() == "error":
            logger.error(message)
        elif level.lower() == "warning":
            logger.warning(message)
        else:
            logger.info(message)

    def log_info(self, message: str, module: str = "default") -> None:
        """Log an info message."""
        logger = self.get_logger(module)
        logger.info(message)

    def log_warning(self, message: str, module: str = "default") -> None:
        """Log a warning message."""
        logger = self.get_logger(module)
        logger.warning(message)

    def log_error(self, message: str, module: str = "default") -> None:
        """Log an error message."""
        logger = self.get_logger(module)
        logger.error(message)

    # Alternative method names that match your original _log function pattern
    def _log(self, process_log: List[str], message: str, level: str = "info", module: str = "default") -> None:
        """Alias for log_message to match your original function name."""
        self.log_message(process_log, message, level, module)


# Create a default instance for easy importing
default_logger = CustomLoggerTracker()

# Expose the methods at module level for easy importing
log_message = default_logger.log_message
log_info = default_logger.log_info
log_warning = default_logger.log_warning
log_error = default_logger.log_error
_log = default_logger._log


# Example usage
if __name__ == "__main__":
    # Method 1: Create your own instance
    logger_tracker = CustomLoggerTracker()
    process_log = []
    
    logger_tracker.log_message(process_log, "This is a test info message", "info", "registration")
    logger_tracker.log_message(process_log, "This is a warning message", "warning", "registration")
    logger_tracker.log_message(process_log, "This is an error message", "error", "registration")
    
    # Method 2: Use the default instance functions
    process_log2 = []
    log_message(process_log2, "Using default logger", "info", "detection")
    _log(process_log2, "Using _log alias", "warning", "detection")
    
    # Method 3: Direct logging without process_log
    log_info("Direct info message", "main")
    log_warning("Direct warning message", "main")
    log_error("Direct error message", "main")
    
    print("Process log 1 contents:")
    for log_entry in process_log:
        print(f"  {log_entry}")
    
    print("Process log 2 contents:")
    for log_entry in process_log2:
        print(f"  {log_entry}")