Implement log_manager module with LoggerPrint and LogManager
This commit is contained in:
74
src/utils/log_manager/__init__.py
Normal file
74
src/utils/log_manager/__init__.py
Normal file
@@ -0,0 +1,74 @@
|
||||
from .manager import (
|
||||
LogManager,
|
||||
get_log_manager,
|
||||
register,
|
||||
register_global_params,
|
||||
get_logger,
|
||||
setup,
|
||||
print,
|
||||
)
|
||||
|
||||
from .constants import (
|
||||
LOG_CATEGORY,
|
||||
LOG_CONSOLE,
|
||||
LOG_STDERR,
|
||||
LOG_FILE,
|
||||
LOG_PATH,
|
||||
LOG_LEVEL,
|
||||
LOG_FILE_LEVEL,
|
||||
LOG_ROTATION,
|
||||
LOG_ROTATION_SIZE,
|
||||
LOG_ROTATION_COUNT,
|
||||
LOG_TIMESTAMP,
|
||||
LOG_BUFFERED,
|
||||
MODULE_LOG_CONSOLE,
|
||||
MODULE_LOG_STDERR,
|
||||
MODULE_LOG_FILE,
|
||||
MODULE_LOG_LEVEL,
|
||||
DEFAULT_TIMESTAMP,
|
||||
DEFAULT_LOG_LEVEL,
|
||||
DEFAULT_FILE_LEVEL,
|
||||
DEFAULT_LOG_PATH,
|
||||
DEFAULT_LOG_FILE,
|
||||
DEFAULT_ROTATION,
|
||||
DEFAULT_ROTATION_SIZE,
|
||||
DEFAULT_ROTATION_COUNT,
|
||||
)
|
||||
|
||||
from .factory import FileHandlerFactory, parse_size
|
||||
from .logger import LoggerPrint
|
||||
|
||||
__all__ = [
|
||||
"LogManager",
|
||||
"get_log_manager",
|
||||
"register",
|
||||
"register_global_params",
|
||||
"get_logger",
|
||||
"setup",
|
||||
"print",
|
||||
"LoggerPrint",
|
||||
"LOG_CATEGORY",
|
||||
"LOG_CONSOLE",
|
||||
"LOG_STDERR",
|
||||
"LOG_FILE",
|
||||
"LOG_PATH",
|
||||
"LOG_LEVEL",
|
||||
"LOG_FILE_LEVEL",
|
||||
"LOG_ROTATION",
|
||||
"LOG_ROTATION_SIZE",
|
||||
"LOG_ROTATION_COUNT",
|
||||
"LOG_TIMESTAMP",
|
||||
"LOG_BUFFERED",
|
||||
"MODULE_LOG_CONSOLE",
|
||||
"MODULE_LOG_STDERR",
|
||||
"MODULE_LOG_FILE",
|
||||
"MODULE_LOG_LEVEL",
|
||||
"DEFAULT_TIMESTAMP",
|
||||
"DEFAULT_LOG_LEVEL",
|
||||
"DEFAULT_FILE_LEVEL",
|
||||
"DEFAULT_LOG_PATH",
|
||||
"DEFAULT_LOG_FILE",
|
||||
"DEFAULT_ROTATION",
|
||||
"DEFAULT_ROTATION_SIZE",
|
||||
"DEFAULT_ROTATION_COUNT",
|
||||
]
|
||||
31
src/utils/log_manager/constants.py
Normal file
31
src/utils/log_manager/constants.py
Normal file
@@ -0,0 +1,31 @@
|
||||
# Category
|
||||
LOG_CATEGORY = "logger"
|
||||
|
||||
# Global config keys
|
||||
LOG_CONSOLE = "log_console"
|
||||
LOG_STDERR = "log_stderr"
|
||||
LOG_FILE = "log_file"
|
||||
LOG_PATH = "log_path"
|
||||
LOG_LEVEL = "log_level"
|
||||
LOG_FILE_LEVEL = "log_file_level"
|
||||
LOG_ROTATION = "log_rotation"
|
||||
LOG_ROTATION_SIZE = "log_rotation_size"
|
||||
LOG_ROTATION_COUNT = "log_rotation_count"
|
||||
LOG_TIMESTAMP = "log_timestamp"
|
||||
LOG_BUFFERED = "log_buffered"
|
||||
|
||||
# Module config keys
|
||||
MODULE_LOG_CONSOLE = "log_console"
|
||||
MODULE_LOG_STDERR = "log_stderr"
|
||||
MODULE_LOG_FILE = "log_file"
|
||||
MODULE_LOG_LEVEL = "log_level"
|
||||
|
||||
# Defaults
|
||||
DEFAULT_TIMESTAMP = "%Y-%m-%d %H:%M:%S"
|
||||
DEFAULT_LOG_LEVEL = "INFO"
|
||||
DEFAULT_FILE_LEVEL = "DEBUG"
|
||||
DEFAULT_LOG_PATH = "./log"
|
||||
DEFAULT_LOG_FILE = "app.log"
|
||||
DEFAULT_ROTATION = "size"
|
||||
DEFAULT_ROTATION_SIZE = "10MB"
|
||||
DEFAULT_ROTATION_COUNT = 5
|
||||
37
src/utils/log_manager/factory.py
Normal file
37
src/utils/log_manager/factory.py
Normal file
@@ -0,0 +1,37 @@
|
||||
import logging
|
||||
import logging.handlers
|
||||
from pathlib import Path
|
||||
from typing import Union
|
||||
|
||||
|
||||
def parse_size(size_str: str) -> int:
|
||||
units = [("GB", 1024 * 1024 * 1024), ("MB", 1024 * 1024), ("KB", 1024), ("B", 1)]
|
||||
size_str = size_str.upper().strip()
|
||||
for unit, multiplier in units:
|
||||
if size_str.endswith(unit):
|
||||
return int(size_str[: -len(unit)]) * multiplier
|
||||
return int(size_str)
|
||||
|
||||
|
||||
class FileHandlerFactory:
|
||||
@staticmethod
|
||||
def create(
|
||||
path: Union[str, Path],
|
||||
filename: str,
|
||||
rotation: str = "size",
|
||||
max_bytes: str = "10MB",
|
||||
backup_count: int = 5,
|
||||
) -> logging.FileHandler:
|
||||
base_path = Path(path)
|
||||
base_path.mkdir(parents=True, exist_ok=True)
|
||||
filepath = base_path / filename
|
||||
|
||||
if rotation == "size":
|
||||
return logging.handlers.RotatingFileHandler(
|
||||
filepath,
|
||||
maxBytes=parse_size(max_bytes),
|
||||
backupCount=backup_count,
|
||||
encoding="utf-8",
|
||||
)
|
||||
else:
|
||||
return logging.FileHandler(filepath, encoding="utf-8")
|
||||
63
src/utils/log_manager/logger.py
Normal file
63
src/utils/log_manager/logger.py
Normal file
@@ -0,0 +1,63 @@
|
||||
import logging
|
||||
import sys
|
||||
import time
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class LoggerPrint(logging.Logger):
|
||||
def __init__(
|
||||
self,
|
||||
name: str,
|
||||
module_name: str,
|
||||
console_enabled: bool = True,
|
||||
stderr_enabled: bool = True,
|
||||
file_enabled: bool = False,
|
||||
file_handler: Optional[logging.FileHandler] = None,
|
||||
timestamp_format: str = "%Y-%m-%d %H:%M:%S",
|
||||
):
|
||||
super().__init__(name)
|
||||
self.module_name = module_name
|
||||
self.console_enabled = console_enabled
|
||||
self.stderr_enabled = stderr_enabled
|
||||
self.file_enabled = file_enabled
|
||||
self.file_handler = file_handler
|
||||
self.timestamp_format = timestamp_format
|
||||
|
||||
if file_handler:
|
||||
file_handler.setFormatter(
|
||||
logging.Formatter(
|
||||
"%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
||||
datefmt=timestamp_format,
|
||||
)
|
||||
)
|
||||
|
||||
def print(self, msg: str, level: str = "info"):
|
||||
effective_level = getattr(logging, level.upper(), logging.INFO)
|
||||
log_record = logging.LogRecord(
|
||||
name=self.name,
|
||||
level=effective_level,
|
||||
pathname="",
|
||||
lineno=0,
|
||||
msg=msg,
|
||||
args=(),
|
||||
exc_info=None,
|
||||
)
|
||||
|
||||
if self.console_enabled and effective_level <= logging.INFO:
|
||||
formatted = self._format(log_record)
|
||||
sys.stdout.write(formatted + "\n")
|
||||
sys.stdout.flush()
|
||||
|
||||
if self.stderr_enabled and effective_level >= logging.WARNING:
|
||||
formatted = self._format(log_record)
|
||||
sys.stderr.write(formatted + "\n")
|
||||
sys.stderr.flush()
|
||||
|
||||
if self.file_enabled and self.file_handler:
|
||||
self.file_handler.emit(log_record)
|
||||
|
||||
def _format(self, record: logging.LogRecord) -> str:
|
||||
timestamp = time.strftime(self.timestamp_format, time.localtime(record.created))
|
||||
return (
|
||||
f"{timestamp} - {record.name} - {record.levelname} - {record.getMessage()}"
|
||||
)
|
||||
217
src/utils/log_manager/manager.py
Normal file
217
src/utils/log_manager/manager.py
Normal file
@@ -0,0 +1,217 @@
|
||||
import logging
|
||||
import src.utils.config_manager as config_manager
|
||||
from pathlib import Path
|
||||
from typing import Dict, Optional, Any
|
||||
|
||||
from .constants import (
|
||||
LOG_CATEGORY,
|
||||
LOG_CONSOLE,
|
||||
LOG_STDERR,
|
||||
LOG_FILE,
|
||||
LOG_PATH,
|
||||
LOG_LEVEL,
|
||||
LOG_FILE_LEVEL,
|
||||
LOG_ROTATION,
|
||||
LOG_ROTATION_SIZE,
|
||||
LOG_ROTATION_COUNT,
|
||||
LOG_TIMESTAMP,
|
||||
LOG_BUFFERED,
|
||||
MODULE_LOG_CONSOLE,
|
||||
MODULE_LOG_STDERR,
|
||||
MODULE_LOG_FILE,
|
||||
MODULE_LOG_LEVEL,
|
||||
DEFAULT_TIMESTAMP,
|
||||
DEFAULT_LOG_LEVEL,
|
||||
DEFAULT_FILE_LEVEL,
|
||||
DEFAULT_LOG_PATH,
|
||||
DEFAULT_LOG_FILE,
|
||||
DEFAULT_ROTATION,
|
||||
DEFAULT_ROTATION_SIZE,
|
||||
DEFAULT_ROTATION_COUNT,
|
||||
)
|
||||
from .factory import FileHandlerFactory
|
||||
from .logger import LoggerPrint
|
||||
|
||||
|
||||
class LogManager:
|
||||
def __init__(self, config=None):
|
||||
self.config = config or config_manager.config
|
||||
self._loggers: Dict[str, LoggerPrint] = {}
|
||||
self._root_logger: Optional[LoggerPrint] = None
|
||||
self._file_handlers: Dict[str, logging.FileHandler] = {}
|
||||
self._setup_done = False
|
||||
|
||||
def register_global_params(self):
|
||||
self.config.register(
|
||||
name=LOG_CONSOLE, val=True, cat=LOG_CATEGORY, desc="Enable console output"
|
||||
)
|
||||
self.config.register(
|
||||
name=LOG_STDERR, val=True, cat=LOG_CATEGORY, desc="Enable stderr for errors"
|
||||
)
|
||||
self.config.register(
|
||||
name=LOG_FILE, val=DEFAULT_LOG_FILE, cat=LOG_CATEGORY, desc="Log file name"
|
||||
)
|
||||
self.config.register(
|
||||
name=LOG_PATH,
|
||||
val=DEFAULT_LOG_PATH,
|
||||
cat=LOG_CATEGORY,
|
||||
desc="Log directory path",
|
||||
)
|
||||
self.config.register(
|
||||
name=LOG_LEVEL,
|
||||
val=DEFAULT_LOG_LEVEL,
|
||||
cat=LOG_CATEGORY,
|
||||
desc="Console log level",
|
||||
)
|
||||
self.config.register(
|
||||
name=LOG_FILE_LEVEL,
|
||||
val=DEFAULT_FILE_LEVEL,
|
||||
cat=LOG_CATEGORY,
|
||||
desc="File log level",
|
||||
)
|
||||
self.config.register(
|
||||
name=LOG_ROTATION,
|
||||
val=DEFAULT_ROTATION,
|
||||
cat=LOG_CATEGORY,
|
||||
desc="Rotation type: size|external",
|
||||
)
|
||||
self.config.register(
|
||||
name=LOG_ROTATION_SIZE,
|
||||
val=DEFAULT_ROTATION_SIZE,
|
||||
cat=LOG_CATEGORY,
|
||||
desc="Max file size for rotation",
|
||||
)
|
||||
self.config.register(
|
||||
name=LOG_ROTATION_COUNT,
|
||||
val=DEFAULT_ROTATION_COUNT,
|
||||
cat=LOG_CATEGORY,
|
||||
desc="Number of rotated files to keep",
|
||||
)
|
||||
self.config.register(
|
||||
name=LOG_TIMESTAMP,
|
||||
val=DEFAULT_TIMESTAMP,
|
||||
cat=LOG_CATEGORY,
|
||||
desc="Log timestamp format",
|
||||
)
|
||||
self.config.register(
|
||||
name=LOG_BUFFERED, val=False, cat=LOG_CATEGORY, desc="Buffer file writes"
|
||||
)
|
||||
|
||||
def register(
|
||||
self,
|
||||
module: str,
|
||||
log_console: bool = None,
|
||||
log_stderr: bool = None,
|
||||
log_file: str = None,
|
||||
log_level: str = None,
|
||||
):
|
||||
cfg = self._get_module_config(module)
|
||||
|
||||
console_enabled = (
|
||||
log_console if log_console is not None else cfg.get(LOG_CONSOLE)
|
||||
)
|
||||
stderr_enabled = log_stderr if log_stderr is not None else cfg.get(LOG_STDERR)
|
||||
file_enabled = log_file if log_file is not None else cfg.get(LOG_FILE)
|
||||
level = log_level if log_level is not None else cfg.get(LOG_LEVEL)
|
||||
|
||||
name = f"komAI.{module}"
|
||||
logger = LoggerPrint(
|
||||
name=name,
|
||||
module_name=module,
|
||||
console_enabled=console_enabled,
|
||||
stderr_enabled=stderr_enabled,
|
||||
file_enabled=bool(file_enabled),
|
||||
timestamp_format=self.config.get(LOG_TIMESTAMP, cat=LOG_CATEGORY)
|
||||
or DEFAULT_TIMESTAMP,
|
||||
)
|
||||
|
||||
if file_enabled:
|
||||
file_handler = self._create_file_handler(module, file_enabled)
|
||||
logger.addHandler(file_handler)
|
||||
logger.file_enabled = True
|
||||
logger.file_handler = file_handler
|
||||
|
||||
self._loggers[module] = logger
|
||||
|
||||
def _get_module_config(self, module: str) -> Dict[str, Any]:
|
||||
return {
|
||||
LOG_CONSOLE: self.config.get(MODULE_LOG_CONSOLE, cat=module),
|
||||
LOG_STDERR: self.config.get(MODULE_LOG_STDERR, cat=module),
|
||||
LOG_FILE: self.config.get(MODULE_LOG_FILE, cat=module),
|
||||
LOG_LEVEL: self.config.get(MODULE_LOG_LEVEL, cat=module),
|
||||
}
|
||||
|
||||
def _get_global_config(self) -> Dict[str, Any]:
|
||||
return {
|
||||
LOG_CONSOLE: self.config.get(LOG_CONSOLE, cat=LOG_CATEGORY),
|
||||
LOG_STDERR: self.config.get(LOG_STDERR, cat=LOG_CATEGORY),
|
||||
LOG_FILE: self.config.get(LOG_FILE, cat=LOG_CATEGORY),
|
||||
LOG_LEVEL: self.config.get(LOG_LEVEL, cat=LOG_CATEGORY),
|
||||
LOG_PATH: self.config.get(LOG_PATH, cat=LOG_CATEGORY) or DEFAULT_LOG_PATH,
|
||||
LOG_FILE_LEVEL: self.config.get(LOG_FILE_LEVEL, cat=LOG_CATEGORY)
|
||||
or DEFAULT_FILE_LEVEL,
|
||||
LOG_ROTATION: self.config.get(LOG_ROTATION, cat=LOG_CATEGORY)
|
||||
or DEFAULT_ROTATION,
|
||||
LOG_ROTATION_SIZE: self.config.get(LOG_ROTATION_SIZE, cat=LOG_CATEGORY)
|
||||
or DEFAULT_ROTATION_SIZE,
|
||||
LOG_ROTATION_COUNT: self.config.get(LOG_ROTATION_COUNT, cat=LOG_CATEGORY)
|
||||
or DEFAULT_ROTATION_COUNT,
|
||||
}
|
||||
|
||||
def _create_file_handler(self, module: str, filename: str) -> logging.FileHandler:
|
||||
global_cfg = self._get_global_config()
|
||||
return FileHandlerFactory.create(
|
||||
path=global_cfg[LOG_PATH],
|
||||
filename=filename,
|
||||
rotation=global_cfg[LOG_ROTATION],
|
||||
max_bytes=global_cfg[LOG_ROTATION_SIZE],
|
||||
backup_count=global_cfg[LOG_ROTATION_COUNT],
|
||||
)
|
||||
|
||||
def setup(self):
|
||||
self._setup_done = True
|
||||
|
||||
def get_logger(self, module: str) -> Optional[LoggerPrint]:
|
||||
return self._loggers.get(module)
|
||||
|
||||
def print(self, msg: str, level: str = "info"):
|
||||
if self._root_logger:
|
||||
self._root_logger.print(msg, level)
|
||||
else:
|
||||
__builtins__["print"](msg)
|
||||
|
||||
|
||||
_log_manager: Optional[LogManager] = None
|
||||
|
||||
|
||||
def get_log_manager() -> LogManager:
|
||||
global _log_manager
|
||||
if _log_manager is None:
|
||||
_log_manager = LogManager()
|
||||
return _log_manager
|
||||
|
||||
|
||||
def register_global_params():
|
||||
get_log_manager().register_global_params()
|
||||
|
||||
|
||||
def register(
|
||||
module: str,
|
||||
log_console: bool = None,
|
||||
log_stderr: bool = None,
|
||||
log_file: str = None,
|
||||
log_level: str = None,
|
||||
):
|
||||
get_log_manager().register(module, log_console, log_stderr, log_file, log_level)
|
||||
|
||||
|
||||
def get_logger(module: str) -> Optional[LoggerPrint]:
|
||||
return get_log_manager().get_logger(module)
|
||||
|
||||
|
||||
def setup():
|
||||
get_log_manager().setup()
|
||||
|
||||
|
||||
def print(msg: str, level: str = "info"):
|
||||
get_log_manager().print(msg, level)
|
||||
Reference in New Issue
Block a user