Source code for laborchestrator.logging_manager

import logging
from datetime import datetime
import os

_log_format = "%(asctime)s - [%(levelname)s] - %(name)s - (%(filename)s).%(funcName)s(%(lineno)d) - %(message)s"
_console_format = "%(levelname)s:%(message)s"

default_session_name = f"session_{datetime.today().ctime()}"
default_session_name = default_session_name.replace(" ", "_").replace(":", "_")

lvl_name = {
    logging.DEBUG: "DEBUG",
    logging.INFO: "INFO",
    logging.WARNING: "WARNING",
    logging.ERROR: "ERROR",
    logging.CRITICAL: "CRITICAL",
}

def get_file_handler(handler_name: str, lvl: int):
    file_handler = logging.FileHandler(handler_name)
    file_handler.setLevel(lvl)
    file_handler.setFormatter(logging.Formatter(_log_format))
    return file_handler

def get_stream_handler(lvl: int):
    stream_handler = logging.StreamHandler()
    stream_handler.setLevel(lvl)
    stream_handler.setFormatter(logging.Formatter(_console_format))
    return stream_handler



def get_standard_logger(session_name: str = default_session_name, stream_lvl: int = logging.INFO):
    """
    Returns a custom logger named 'Orchestrator logger' with separate file and console handlers.
    This logger is independent of the root logger so that calls like logging.warning() do not use these settings.
    """
    # Create a dedicated logger
    logger = logging.getLogger("Orchestrator logger")
    logger.setLevel(logging.DEBUG)
    logger.propagate = False  # prevent log messages from being passed to the root logger

    # Clear any existing handlers (if reconfiguring)
    if logger.hasHandlers():
        logger.handlers.clear()

    # Create a session-specific directory for log files
    base_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "logs")
    session_path = os.path.join(base_path, session_name)
    try:
        if not os.path.exists(session_path):
            os.makedirs(session_path)
    except OSError as err:
        logger.error(f"Could not create log directory ({err}). Logs may not be saved.")

    # Set up file handlers for each logging level
    for lvl, name in lvl_name.items():
        file_name = os.path.join(session_path, f"{session_name}_{name}.log")
        logger.addHandler(get_file_handler(file_name, lvl))

    # Add stream (console) handler with the desired level and format
    logger.addHandler(get_stream_handler(stream_lvl))
    return logger


# Create a module-level logger that can be imported as StandardLogger
StandardLogger = get_standard_logger()

# Optionally, define __all__ to restrict module exports
__all__ = ['StandardLogger', 'get_standard_logger']


if __name__ == "__main__":
    test_logger=StandardLogger
    test_logger.debug("Hello World")
    test_logger.info("the end is near")
    test_logger.warning("please prepare for the end!")
    test_logger.error("Goodbye cruel world :-(")