Update Colors to have static colors that we can reset to

enables Queues if multiprocessing.Queue() is set in the "log_queue" setting

Now a logger "Log.init_worker_logging" can be attached to the ProcessPoolExecutor,
the init args must be the log_queue set

example:

```py
with concurrent.futures.ProcessPoolExecutor(
    max_workers=max_forks,
    initializer=Log.init_worker_logging,
    initargs=(log_queue,)
) as executor:
````

Move all settings into a settings argument, the structure is defined in LogSettings.
Default settings are in Log.DEFAULT_LOG_SETTINGS

Only log path and log name are parameters

Color output for console is on default enabled, disable via "console_color_output_enabled"
The complete console output can be stopped with "console_enabled"

README.md

@@ -39,8 +39,8 @@ explicit = true
 ```
 
 ```sh
-uv build --native-tls
+uv build
-uv publish --index egra-gitea --token <gitea token> --native-tls
+uv publish --index egra-gitea --token <gitea token>
 ```
 
 ## Test package
@@ -48,7 +48,7 @@ uv publish --index egra-gitea --token <gitea token> --native-tls
 We must set the full index URL here because we run with "--no-project"
 
 ```sh
-uv run --with corelibs --index egra-gitea=https://git.egplusww.jp/api/packages/PyPI/pypi/simple/ --no-project --native-tls -- python -c "import corelibs"
+uv run --with corelibs --index egra-gitea=https://git.egplusww.jp/api/packages/PyPI/pypi/simple/ --no-project -- python -c "import corelibs"
 ```
 
 ### Python tests
@@ -71,16 +71,32 @@ uv run pytest --cov=corelibs
 
 In the test-run folder usage and run tests are located
 
+#### Progress
+
 ```sh
-uv run --native-tls test-run/progress/progress_test.py
+uv run test-run/progress/progress_test.py
+```
+
+#### Double byte string format
+
+```sh
+uv run test-run/double_byte_string_format/double_byte_string_format.py
+```
+
+#### Strings helpers
+
+```sh
+uv run test-run/timestamp_strings/timestamp_strings.py
 ```
 
 ```sh
-uv run --native-tls test-run/double_byte_string_format/double_byte_string_format.py
+uv run test-run/string_handling/string_helpers.py
 ```
 
+#### Log
+
 ```sh
-uv run --native-tls test-run/timestamp_strings/timestamp_strings.py
+uv run test-run/logging_handling/log.py
 ```
 
 ## How to install in another project
@@ -88,19 +104,23 @@ uv run --native-tls test-run/timestamp_strings/timestamp_strings.py
 This will also add the index entry
 
 ```sh
-uv add corelibs --index egra-gitea=https://git.egplusww.jp/api/packages/PyPI/pypi/simple/ --native-tls
+uv add corelibs --index egra-gitea=https://git.egplusww.jp/api/packages/PyPI/pypi/simple/
 ```
 
 ## Python venv setup
 
-In the folder where the script will be located
+After clone, run the command below to install all dependenciss
-
-```sh
-uv venv --python 3.13
-```
-
-Install all neded dependencies
 
 ```sh
 uv sync
 ```
+
+## NOTE on TLS problems
+
+> [!warning] TLS problems with Netskope
+
+If the Netskope service is running all uv runs will fail unless either --native-tls is set or the enviroment variable SSL_CERT_FILE is set, see blow
+
+```sh
+export SSL_CERT_FILE='/Library/Application Support/Netskope/STAgent/data/nscacert_combined.pem'
+```

pyproject.toml

@@ -1,9 +1,9 @@
 # MARK: Project info
 [project]
 name = "corelibs"
-version = "0.7.0"
+version = "0.9.0"
 description = "Collection of utils for Python scripts"
-readme = "ReadMe.md"
+readme = "README.md"
 requires-python = ">=3.13"
 dependencies = [
     "jmespath>=1.0.1",

src/corelibs/logging_handling/log.py

@@ -1,11 +1,70 @@
 """
 A log handler wrapper
+if log_settings['log_queue'] is set to multiprocessing.Queue it will launch with listeners
+attach "init_worker_logging" with the set log_queue
 """
 
+import re
 import logging.handlers
 import logging
 from pathlib import Path
-from typing import Mapping
+from typing import Mapping, TextIO, TypedDict, Any, TYPE_CHECKING, cast
+from corelibs.string_handling.text_colors import Colors
+
+if TYPE_CHECKING:
+    from multiprocessing import Queue
+
+
+class LogSettings(TypedDict):
+    """
+    log settings
+
+    Arguments:
+        TypedDict {_type_} -- _description_
+    """
+    log_level_console: str
+    log_level_file: str
+    console_enabled: bool
+    console_color_output_enabled: bool
+    add_start_info: bool
+    add_end_info: bool
+    log_queue: 'Queue[str] | None'
+
+
+class CustomConsoleFormatter(logging.Formatter):
+    """
+    Custom formatter with colors for console output
+    """
+
+    COLORS = {
+        "DEBUG": Colors.cyan,
+        "INFO": Colors.green,
+        "WARNING": Colors.yellow,
+        "ERROR": Colors.red,
+        "CRITICAL": Colors.magenta,
+        "EXCEPTION": Colors.magenta_bold,
+    }
+
+    def format(self, record: logging.LogRecord) -> str:
+        """
+        set the color highlight
+
+        Arguments:
+            record {logging.LogRecord} -- _description_
+
+        Returns:
+            str -- _description_
+        """
+        # Add color to levelname for console output
+        reset = Colors.reset
+        color = self.COLORS.get(record.levelname, reset)
+        # only highlight level for basic
+        if record.levelname in ['DEBUG', 'INFO']:
+            record.levelname = f"{color}{record.levelname}{reset}"
+            return super().format(record)
+        # highlight whole line
+        message = super().format(record)
+        return f"{color}{message}{reset}"
 
 
 class Log:
@@ -13,57 +72,148 @@ class Log:
     logger setup
     """
 
+    # exception level
     EXCEPTION: int = 60
+    # spacer lenght characters and the character
+    SPACER_CHAR: str = '='
+    SPACER_LENGTH: int = 32
+    # default logging level
+    DEFAULT_LOG_LEVEL: str = 'WARNING'
+    # default settings
+    DEFAULT_LOG_SETTINGS: LogSettings = {
+            "log_level_console": "WARNING",
+            "log_level_file": "DEBUG",
+            "console_enabled": True,
+            "console_color_output_enabled": True,
+            "add_start_info": True,
+            "add_end_info": False,
+            "log_queue": None,
+        }
 
     def __init__(
         self,
         log_path: Path,
         log_name: str,
-        log_level_console: str = 'WARNING',
+        log_settings: dict[str, 'str | bool | None | Queue[str]'] | LogSettings | None = None,
-        log_level_file: str = 'DEBUG',
-        add_start_info: bool = True
     ):
+        # add new level for EXCEPTION
         logging.addLevelName(Log.EXCEPTION, 'EXCEPTION')
-        if not log_name.endswith('.log'):
+        # parse the logging settings
+        self.log_settings = self.__parse_log_settings(log_settings)
+        # if path, set log name with .log
+        # if log name with .log, strip .log for naming
+        if log_path.is_dir():
+            __log_file_name = re.sub(r'[^a-zA-Z0-9]', '', log_name)
+            if not log_name.endswith('.log'):
+                log_path = log_path.joinpath(Path(__log_file_name).with_suffix('.log'))
+            else:
+                log_path = log_path.joinpath(__log_file_name)
+        elif not log_path.suffix == '.log':
+            # add .log if the path is a file but without .log
             log_path = log_path.with_suffix('.log')
-        # overall logger settings
+        # stip .log from the log name if set
-        self.logger = logging.getLogger(log_name)
+        if not log_name.endswith('.log'):
-        # set maximum logging level for all logging output
+            log_name = Path(log_name).stem
-        self.logger.setLevel(logging.DEBUG)
+        # general log name
+        self.log_name = log_name
 
-        # self.handlers = []
+        self.log_queue: 'Queue[str] | None' = None
-        # console logger
+        self.listener: logging.handlers.QueueListener | None = None
-        self.__console_handler(log_level_console)
+
-        # file logger
+        # setup handlers
-        self.__file_handler(log_level_file, log_path)
+        # NOTE if console with color is set first, some of the color formatting is set
+        # in the file writer too, for the ones where color is set BEFORE the format
+        self.handlers: list[logging.StreamHandler[TextIO] | logging.handlers.TimedRotatingFileHandler] = [
+            # file handler, always
+            self.__file_handler(self.log_settings['log_level_file'], log_path)
+        ]
+        if self.log_settings['console_enabled']:
+            # console
+            self.handlers.append(self.__console_handler(self.log_settings['log_level_console']))
+        # init listener if we have a log_queue set
+        self.__init_listener(self.log_settings['log_queue'])
+
+        # overall logger start
+        self.__init_log(log_name)
         # if requests set a start log
-        if add_start_info is True:
+        if self.log_settings['add_start_info'] is True:
             self.break_line('START')
 
+    def __del__(self):
+        """
+        Call when class is destroyed, make sure the listender is closed or else we throw a thread error
+        """
+        if self.log_settings['add_end_info']:
+            self.break_line('END')
+        self.stop_listener()
+
+    def __parse_log_settings(
+        self,
+        log_settings: dict[str, 'str | bool | None | Queue[str]'] | LogSettings | None
+    ) -> LogSettings:
+        # skip with defaul it not set
+        if log_settings is None:
+            return self.DEFAULT_LOG_SETTINGS
+        # check entries
+        default_log_settings = self.DEFAULT_LOG_SETTINGS
+        # check log levels
+        for __log_entry in ['log_level_console', 'log_level_file']:
+            if log_settings.get(__log_entry) is None:
+                continue
+            # if not valid reset to default, if not in default set to WARNING
+            if not self.validate_log_level(_log_level := log_settings.get(__log_entry, '')):
+                _log_level = self.DEFAULT_LOG_SETTINGS.get(
+                    __log_entry, self.DEFAULT_LOG_LEVEL
+                )
+            default_log_settings[__log_entry] = str(_log_level)
+        # check bool
+        for __log_entry in [
+            "console_enabled",
+            "console_color_output_enabled",
+            "add_start_info",
+            "add_end_info",
+        ]:
+            if log_settings.get(__log_entry) is None:
+                continue
+            if not isinstance(__setting := log_settings.get(__log_entry, ''), bool):
+                __setting = self.DEFAULT_LOG_SETTINGS.get(__log_entry, True)
+            default_log_settings[__log_entry] = __setting
+        # check log queue
+        __setting = log_settings.get('log_queue', self.DEFAULT_LOG_SETTINGS['log_queue'])
+        if __setting is not None:
+            __setting = cast('Queue[str]', __setting)
+        default_log_settings['log_queue'] = __setting
+        return default_log_settings
+
     def __filter_exceptions(self, record: logging.LogRecord) -> bool:
         return record.levelname != "EXCEPTION"
 
-    def __console_handler(self, log_level_console: str = 'WARNING'):
+    def __console_handler(self, log_level_console: str = 'WARNING') -> logging.StreamHandler[TextIO]:
         # console logger
         if not isinstance(getattr(logging, log_level_console.upper(), None), int):
             log_level_console = 'WARNING'
         console_handler = logging.StreamHandler()
-        formatter_console = logging.Formatter(
+        # format layouts
-            (
+        format_string = (
-                '[%(asctime)s.%(msecs)03d] '
+            '[%(asctime)s.%(msecs)03d] '
-                '[%(filename)s:%(funcName)s:%(lineno)d] '
+            '[%(name)s] '
-                '<%(levelname)s> '
+            '[%(filename)s:%(funcName)s:%(lineno)d] '
-                '%(message)s'
+            '<%(levelname)s> '
-            ),
+            '%(message)s'
-            datefmt="%Y-%m-%d %H:%M:%S",
         )
+        format_date = "%Y-%m-%d %H:%M:%S"
+        # color or not
+        if self.log_settings['console_color_output_enabled']:
+            formatter_console = CustomConsoleFormatter(format_string, datefmt=format_date)
+        else:
+            formatter_console = logging.Formatter(format_string, datefmt=format_date)
         console_handler.setLevel(log_level_console)
         # do not show exceptions logs on console
         console_handler.addFilter(self.__filter_exceptions)
         console_handler.setFormatter(formatter_console)
-        self.logger.addHandler(console_handler)
+        return console_handler
 
-    def __file_handler(self, log_level_file: str, log_path: Path) -> None:
+    def __file_handler(self, log_level_file: str, log_path: Path) -> logging.handlers.TimedRotatingFileHandler:
         # file logger
         if not isinstance(getattr(logging, log_level_file.upper(), None), int):
             log_level_file = 'DEBUG'
@@ -75,17 +225,80 @@ class Log:
         )
         formatter_file_handler = logging.Formatter(
             (
+                # time stamp
                 '[%(asctime)s.%(msecs)03d] '
-                '[%(name)s:%(process)d] '
+                # log name
+                '[%(name)s] '
+                # filename + pid
+                '[%(filename)s:%(process)d] '
+                # path + func + line number
                 '[%(pathname)s:%(funcName)s:%(lineno)d] '
+                # error level
                 '<%(levelname)s> '
+                # message
                 '%(message)s'
             ),
             datefmt="%Y-%m-%dT%H:%M:%S",
         )
         file_handler.setLevel(log_level_file)
         file_handler.setFormatter(formatter_file_handler)
-        self.logger.addHandler(file_handler)
+        return file_handler
+
+    def __init_listener(self, log_queue: 'Queue[str] | None' = None):
+        """
+        If we have a Queue option start the logging queue
+
+        Keyword Arguments:
+            log_queue {Queue[str] | None} -- _description_ (default: {None})
+        """
+        if log_queue is None:
+            return
+        self.log_queue = log_queue
+        self.listener = logging.handlers.QueueListener(self.log_queue, *self.handlers)
+        self.listener.start()
+
+    def __init_log(self, log_name: str) -> None:
+        """
+        Initialize the main loggger
+        """
+        queue_handler: logging.handlers.QueueHandler | None = None
+        if self.log_queue is not None:
+            queue_handler = logging.handlers.QueueHandler(self.log_queue)
+        # overall logger settings
+        self.logger = logging.getLogger(log_name)
+        # add all the handlers
+        if queue_handler is None:
+            for handler in self.handlers:
+                self.logger.addHandler(handler)
+        else:
+            self.logger.addHandler(queue_handler)
+        # set maximum logging level for all logging output
+        # log level filtering is done per handler
+        self.logger.setLevel(logging.DEBUG)
+
+    @staticmethod
+    def init_worker_logging(log_queue: 'Queue[str]') -> logging.Logger:
+        """
+        This initalizes a logger that can be used in pool/thread queue calls
+        """
+        queue_handler = logging.handlers.QueueHandler(log_queue)
+        # getLogger call MUST be WITHOUT and logger name
+        root_logger = logging.getLogger()
+        root_logger.setLevel(logging.DEBUG)
+        root_logger.handlers.clear()
+        root_logger.addHandler(queue_handler)
+
+        # for debug only
+        root_logger.debug('[LOGGER] Init log: %s - %s', log_queue, root_logger.handlers)
+
+        return root_logger
+
+    def stop_listener(self):
+        """
+        stop the listener
+        """
+        if self.listener is not None:
+            self.listener.stop()
 
     def break_line(self, info: str = "BREAK"):
         """
@@ -94,7 +307,7 @@ class Log:
         Keyword Arguments:
             info {str} -- _description_ (default: {"BREAK"})
         """
-        self.logger.info("[%s] ================================>", info)
+        self.logger.info("[%s] %s>", info, self.SPACER_CHAR * self.SPACER_LENGTH)
 
     def exception(self, msg: object, *args: object, extra: Mapping[str, object] | None = None) -> None:
         """
@@ -107,16 +320,21 @@ class Log:
         """
         self.logger.log(Log.EXCEPTION, msg, *args, exc_info=True, extra=extra)
 
-    def validate_log_level(self, log_level: str) -> bool:
+    def validate_log_level(self, log_level: Any) -> bool:
         """
         if the log level is invalid, will erturn false
 
         Args:
-            log_level (str): _description_
+            log_level (Any): _description_
 
         Returns:
             bool: _description_
         """
-        return isinstance(getattr(logging, log_level.upper(), None), int)
+        if isinstance(log_level, int):
+            return any(getattr(logging, attr) == log_level for attr in dir(logging))
+        elif isinstance(log_level, str):
+            return isinstance(getattr(logging, log_level.upper(), None), int)
+        else:
+            return False
 
 # __END__

src/corelibs/string_handling/text_colors.py

@@ -0,0 +1,156 @@
+"""
+Basic ANSI colors
+
+Set colors with print(f"something {Colors.yellow}colorful{Colors.end})
+bold + underline + color combinations are possible.
+"""
+
+
+class Colors:
+    """
+    ANSI colors defined
+    """
+    # General sets, these should not be accessd
+    __BOLD = '\033[1m'
+    __UNDERLINE = '\033[4m'
+    __END = '\033[0m'
+    __RESET = '\033[0m'
+    # Define ANSI color codes as class attributes
+    __BLACK = "\033[30m"
+    __RED = "\033[31m"
+    __GREEN = "\033[32m"
+    __YELLOW = "\033[33m"
+    __BLUE = "\033[34m"
+    __MAGENTA = "\033[35m"
+    __CYAN = "\033[36m"
+    __WHITE = "\033[37m"
+
+    # Define bold/bright versions of the colors
+    __BLACK_BOLD = "\033[1;30m"
+    __RED_BOLD = "\033[1;31m"
+    __GREEN_BOLD = "\033[1;32m"
+    __YELLOW_BOLD = "\033[1;33m"
+    __BLUE_BOLD = "\033[1;34m"
+    __MAGENTA_BOLD = "\033[1;35m"
+    __CYAN_BOLD = "\033[1;36m"
+    __WHITE_BOLD = "\033[1;37m"
+
+    # BRIGHT, alternative
+    __BLACK_BRIGHT = '\033[90m'
+    __RED_BRIGHT = '\033[91m'
+    __GREEN_BRIGHT = '\033[92m'
+    __YELLOW_BRIGHT = '\033[93m'
+    __BLUE_BRIGHT = '\033[94m'
+    __MAGENTA_BRIGHT = '\033[95m'
+    __CYAN_BRIGHT = '\033[96m'
+    __WHITE_BRIGHT = '\033[97m'
+
+    # set access vars
+    bold = __BOLD
+    underline = __UNDERLINE
+    end = __END
+    reset = __RESET
+    # normal
+    black = __BLACK
+    red = __RED
+    green = __GREEN
+    yellow = __YELLOW
+    blue = __BLUE
+    magenta = __MAGENTA
+    cyan = __CYAN
+    white = __WHITE
+    # bold
+    black_bold = __BLACK_BOLD
+    red_bold = __RED_BOLD
+    green_bold = __GREEN_BOLD
+    yellow_bold = __YELLOW_BOLD
+    blue_bold = __BLUE_BOLD
+    magenta_bold = __MAGENTA_BOLD
+    cyan_bold = __CYAN_BOLD
+    white_bold = __WHITE_BOLD
+    # bright
+    black_bright = __BLACK_BRIGHT
+    red_bright = __RED_BRIGHT
+    green_bright = __GREEN_BRIGHT
+    yellow_bright = __YELLOW_BRIGHT
+    blue_bright = __BLUE_BRIGHT
+    magenta_bright = __MAGENTA_BRIGHT
+    cyan_bright = __CYAN_BRIGHT
+    white_bright = __WHITE_BRIGHT
+
+    @staticmethod
+    def disable():
+        """
+        No colors
+        """
+        Colors.bold = ''
+        Colors.underline = ''
+        Colors.end = ''
+        Colors.reset = ''
+        # normal
+        Colors.black = ''
+        Colors.red = ''
+        Colors.green = ''
+        Colors.yellow = ''
+        Colors.blue = ''
+        Colors.magenta = ''
+        Colors.cyan = ''
+        Colors.white = ''
+        # bold/bright
+        Colors.black_bold = ''
+        Colors.red_bold = ''
+        Colors.green_bold = ''
+        Colors.yellow_bold = ''
+        Colors.blue_bold = ''
+        Colors.magenta_bold = ''
+        Colors.cyan_bold = ''
+        Colors.white_bold = ''
+        # bold/bright alt
+        Colors.black_bright = ''
+        Colors.red_bright = ''
+        Colors.green_bright = ''
+        Colors.yellow_bright = ''
+        Colors.blue_bright = ''
+        Colors.magenta_bright = ''
+        Colors.cyan_bright = ''
+        Colors.white_bright = ''
+
+    @staticmethod
+    def reset_colors():
+        """
+        reset colors to the original ones
+        """
+        # set access vars
+        Colors.bold = Colors.__BOLD
+        Colors.underline = Colors.__UNDERLINE
+        Colors.end = Colors.__END
+        Colors.reset = Colors.__RESET
+        # normal
+        Colors.black = Colors.__BLACK
+        Colors.red = Colors.__RED
+        Colors.green = Colors.__GREEN
+        Colors.yellow = Colors.__YELLOW
+        Colors.blue = Colors.__BLUE
+        Colors.magenta = Colors.__MAGENTA
+        Colors.cyan = Colors.__CYAN
+        Colors.white = Colors.__WHITE
+        # bold
+        Colors.black_bold = Colors.__BLACK_BOLD
+        Colors.red_bold = Colors.__RED_BOLD
+        Colors.green_bold = Colors.__GREEN_BOLD
+        Colors.yellow_bold = Colors.__YELLOW_BOLD
+        Colors.blue_bold = Colors.__BLUE_BOLD
+        Colors.magenta_bold = Colors.__MAGENTA_BOLD
+        Colors.cyan_bold = Colors.__CYAN_BOLD
+        Colors.white_bold = Colors.__WHITE_BOLD
+        # bright
+        Colors.black_bright = Colors.__BLACK_BRIGHT
+        Colors.red_bright = Colors.__RED_BRIGHT
+        Colors.green_bright = Colors.__GREEN_BRIGHT
+        Colors.yellow_bright = Colors.__YELLOW_BRIGHT
+        Colors.blue_bright = Colors.__BLUE_BRIGHT
+        Colors.magenta_bright = Colors.__MAGENTA_BRIGHT
+        Colors.cyan_bright = Colors.__CYAN_BRIGHT
+        Colors.white_bright = Colors.__WHITE_BRIGHT
+
+# __END__

src/corelibs/string_handling/timestamp_strings.py

@@ -11,11 +11,11 @@ class TimestampStrings:
     set default time stamps
     """
 
-    time_zone: str = 'Asia/Tokyo'
+    TIME_ZONE: str = 'Asia/Tokyo'
 
     def __init__(self, time_zone: str | None = None):
         self.timestamp_now = datetime.now()
-        self.time_zone = time_zone if time_zone is not None else __class__.time_zone
+        self.time_zone = time_zone if time_zone is not None else self.TIME_ZONE
         try:
             self.timestamp_now_tz = datetime.now(ZoneInfo(self.time_zone))
         except ZoneInfoNotFoundError as e:

test-run/logging_handling/log.py

@@ -0,0 +1,48 @@
+"""
+Log logging_handling.log testing
+"""
+
+# import atexit
+from pathlib import Path
+from multiprocessing import Queue
+# this is for testing only
+from queue_logger.log_queue import QueueLogger
+from corelibs.logging_handling.log import Log
+
+
+def main():
+    """
+    Log testing
+    """
+    script_path: Path = Path(__file__).resolve().parent
+    log = Log(
+        log_path=script_path.joinpath('log', 'test.log'),
+        log_name="Test Log",
+        log_settings={
+            "log_level_console": 'DEBUG',
+            "log_level_file": 'DEBUG',
+            # "console_color_output_enabled": False,
+        }
+    )
+
+    log.logger.debug('Debug test: %s', log.logger.name)
+    log.logger.info('Info test: %s', log.logger.name)
+    log.logger.warning('Warning test: %s', log.logger.name)
+    log.logger.error('Error test: %s', log.logger.name)
+    log.logger.critical('Critical test: %s', log.logger.name)
+    log.exception('Exception test: %s', log.logger.name)
+
+    log_queue: 'Queue[str]' = Queue()
+    log_q = QueueLogger(
+        log_file=script_path.joinpath('log', 'test_queue.log'),
+        log_name="Test Log Queue",
+        log_queue=log_queue
+    )
+    log_q.mlog.info('Log test: %s', log.logger.name)
+    # log_q.stop_listener()
+
+
+if __name__ == "__main__":
+    main()
+
+# __END__

test-run/logging_handling/log/.gitignore

@@ -0,0 +1,2 @@
+*
+!.gitignore

test-run/logging_handling/log_queue.py

@@ -0,0 +1,75 @@
+"""
+Pool Queue log handling
+Thread Queue log handling
+"""
+
+import random
+import time
+from multiprocessing import Queue
+import concurrent.futures
+import logging
+from pathlib import Path
+from corelibs.logging_handling.log import Log
+
+
+def work_function(log_name: str, worker_id: int, data: list[int]) -> int:
+    """
+    simulate worker
+
+    Arguments:
+        worker_id {int} -- _description_
+        data {list[int]} -- _description_
+
+    Returns:
+        int -- _description_
+    """
+    log = logging.getLogger(f'{log_name}-WorkerFn-{worker_id}')
+    log.info('Starting worker: %s', worker_id)
+    time.sleep(random.uniform(1, 3))
+    result = sum(data) * worker_id
+    return result
+
+
+def main():
+    """
+    Queue log tester
+    """
+    print("[START] Queue logger test")
+    log_queue: 'Queue[str]' = Queue()
+    script_path: Path = Path(__file__).resolve().parent
+    log = Log(
+        log_path=script_path.joinpath('log', 'test.log'),
+        log_name="Test Log",
+        log_settings={
+            "log_level_console": 'INFO',
+            "log_level_file": 'INFO',
+            "log_queue": log_queue,
+        }
+    )
+    log.logger.info('Pool Fork logging test')
+    max_forks = 2
+    data_sets = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]
+    with concurrent.futures.ProcessPoolExecutor(
+        max_workers=max_forks,
+        initializer=Log.init_worker_logging,
+        initargs=(log_queue,)
+    ) as executor:
+        log.logger.info('Start workers')
+        futures = [
+            executor.submit(work_function, log.log_name, worker_id, data)
+            for worker_id, data in enumerate(data_sets, 1)
+        ]
+        log.logger.info('Workders started')
+
+        for future in concurrent.futures.as_completed(futures):
+            log.logger.info('Processing result: %s', future.result())
+            print(f"Processing result: {future.result()}")
+
+    log.logger.info('[END] Queue logger test')
+    log.stop_listener()
+
+
+if __name__ == "__main__":
+    main()
+
+# __END__

test-run/logging_handling/queue_logger/log_queue.py

@@ -0,0 +1,96 @@
+"""
+test queue logger interface
+NOTE: this has all moved to the default log interface
+"""
+
+import logging
+import logging.handlers
+from pathlib import Path
+from multiprocessing import Queue
+
+
+class QueueLogger:
+    """
+    Queue logger
+    """
+
+    def __init__(self, log_file: Path, log_name: str, log_queue: 'Queue[str] | None' = None):
+        self.log_file = log_file
+        self.log_name = log_name
+        self.handlers = self.setup_logging()
+        self.log_queue: 'Queue[str]' = log_queue if log_queue is not None else Queue()
+        self.listener = logging.handlers.QueueListener(self.log_queue, *self.handlers)
+        self.listener.start()
+
+        self.mlog: logging.Logger = self.main_log(log_name)
+
+    def __del__(self):
+        self.mlog.info("[%s] ================================>", "END")
+        self.listener.stop()
+
+    def setup_logging(self):
+        """
+        setup basic logging
+        """
+
+        # Create formatters
+        file_formatter = logging.Formatter(
+            '%(asctime)s - %(name)s - %(levelname)s - [PID:%(process)d] [%(filename)s:%(lineno)d] - %(message)s'
+        )
+
+        console_formatter = logging.Formatter(
+            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+        )
+
+        # Create handlers
+        file_handler = logging.FileHandler(self.log_file)
+        file_handler.setFormatter(file_formatter)
+        file_handler.setLevel(logging.DEBUG)
+
+        console_handler = logging.StreamHandler()
+        console_handler.setFormatter(console_formatter)
+        console_handler.setLevel(logging.DEBUG)
+
+        return [file_handler, console_handler]
+
+    def main_log(self, log_name: str) -> logging.Logger:
+        """
+        main logger
+
+        Arguments:
+            log_name {str} -- _description_
+
+        Returns:
+            logging.Logger -- _description_
+        """
+        mlog_handler = logging.handlers.QueueHandler(self.log_queue)
+        mlog = logging.getLogger(f'{log_name}-MainProcess')
+        mlog.addHandler(mlog_handler)
+        mlog.setLevel(logging.DEBUG)
+        return mlog
+
+    @staticmethod
+    def init_worker_logging(log_queue: 'Queue[str]', log_name: str, ):
+        """
+        Initialize logging for worker processes
+        """
+
+        # Create QueueHandler
+        queue_handler = logging.handlers.QueueHandler(log_queue)
+
+        # Setup root logger for this process
+        # NOTE: This must be EMPTY or new SINGLE NEW logger is created, we need one for EACH fork
+        root_logger = logging.getLogger()
+        root_logger.setLevel(logging.DEBUG)
+        root_logger.handlers.clear()
+        root_logger.addHandler(queue_handler)
+
+        root_logger.info('[LOGGER] Init log: %s - %s', log_queue, log_name)
+
+        return root_logger
+
+    def stop_listener(self):
+        """
+        stop the listener
+        """
+        self.listener.stop()

test-run/string_handling/string_helpers.py

@@ -6,6 +6,7 @@ import sys
 from decimal import Decimal, getcontext
 from textwrap import shorten
 from corelibs.string_handling.string_helpers import shorten_string, format_number
+from corelibs.string_handling.text_colors import Colors
 
 
 def __sh_shorten_string():
@@ -16,7 +17,7 @@ def __sh_shorten_string():
         result = shorten_string(string, length, placeholder=placeholder)
         print(f"IN: {string} -> {result}")
     except ValueError as e:
-        print(f"Failed: {e}")
+        print(f"{Colors.red}Failed: {Colors.bold}{e}{Colors.end}")
     try:
         result = shorten(string, width=length, placeholder=placeholder)
         print(f"IN: {string} -> {result}")
@@ -51,12 +52,34 @@ def __sh_format_number():
     print(f"Format {number} ({precision}) -> {result}")
 
 
+def _sh_colors():
+    for color in [
+        "black",
+        "red",
+        "green",
+        "yellow",
+        "blue",
+        "magenta",
+        "cyan",
+        "white",
+    ]:
+        for change in ['', '_bold', '_bright']:
+            _color = f"{color}{change}"
+            print(f"Color: {getattr(Colors, _color)}{_color}{Colors.end}")
+
+    print(f"Underline: {Colors.underline}UNDERLINE{Colors.reset}")
+    print(f"Bold: {Colors.bold}BOLD{Colors.reset}")
+    print(f"Underline/Yellow: {Colors.underline}{Colors.yellow}UNDERLINE YELLOW{Colors.reset}")
+    print(f"Underline/Yellow/Bold: {Colors.underline}{Colors.bold}{Colors.yellow}UNDERLINE YELLOW BOLD{Colors.reset}")
+
+
 def main():
     """
     Test: corelibs.string_handling.string_helpers
     """
     __sh_shorten_string()
     __sh_format_number()
+    _sh_colors()
 
 
 if __name__ == "__main__":

tests/unit/string_handling/test_timestamp_strings.py

@@ -82,7 +82,7 @@ class TestTimestampStrings:
 
     def test_class_default_timezone(self):
         """Test that class default timezone is correctly set"""
-        assert TimestampStrings.time_zone == 'Asia/Tokyo'
+        assert TimestampStrings.TIME_ZONE == 'Asia/Tokyo'
 
     def test_none_timezone_uses_default(self):
         """Test that passing None for timezone uses class default"""

uv.lock

@@ -44,7 +44,7 @@ wheels = [
 
 [[package]]
 name = "corelibs"
-version = "0.7.0"
+version = "0.9.0"
 source = { editable = "." }
 dependencies = [
     { name = "jmespath" },