Spaces:
Sleeping
Sleeping
File size: 7,174 Bytes
7dfe46c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 |
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}") |