Several improvements with logging

Fixes #3

Signed-off-by: Pedro Algarvio <[email protected]>

Author: s0undt3ch

changelog/3.improvement.rst

@@ -0,0 +1 @@
+Several improvements with logging

setup.cfg

@@ -85,6 +85,9 @@ per-file-ignores =
   # D100 Missing docstring in public module
   # D103 Missing docstring in public function
   noxfile.py: D100,D102,D103,D107,D212,E501
+  # D101 Missing docstring in public class
+  # D102 Missing docstring in public method
+  src/ptscripts/logs.py: D101,D102
   # D100 Missing docstring in public module
   # D101 Missing docstring in public class
   # D102 Missing docstring in public method

src/ptscripts/__init__.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import ptscripts.logs
 from ptscripts.parser import command_group
 from ptscripts.parser import Context
 

src/ptscripts/__main__.py

@@ -3,24 +3,9 @@
 import logging
 import sys
 
-from rich.console import Console
-from rich.logging import RichHandler
-
 from ptscripts.parser import Parser
 
-FORMAT = "%(message)s"
-logging.basicConfig(
-    level=logging.INFO,
-    format=FORMAT,
-    datefmt="[%X]",
-    handlers=[
-        RichHandler(
-            console=Console(stderr=True),
-            markup=True,
-            rich_tracebacks=True,
-        ),
-    ],
-)
+log = logging.getLogger(__name__)
 
 
 def main():
@@ -29,6 +14,7 @@ def main():
     """
     parser = Parser()
     cwd = str(parser.repo_root)
+    log.debug(f"Searching for tools in {cwd}")
     if cwd in sys.path:
         sys.path.remove(cwd)
     sys.path.insert(0, cwd)

src/ptscripts/logs.py

@@ -0,0 +1,112 @@
+from __future__ import annotations
+
+import logging
+import os
+import sys
+
+STDOUT = sys.maxsize
+logging.STDOUT = STDOUT  # type: ignore[attr-defined]
+STDERR = sys.maxsize - 1
+logging.STDERR = STDERR  # type: ignore[attr-defined]
+logging.addLevelName(STDOUT, "STDOUT")
+logging.addLevelName(STDERR, "STDERR")
+
+
+class LevelFilter(logging.Filter):
+    def __init__(
+        self,
+        level: int | None = None,
+        not_levels: list[int] | tuple[int, ...] | None = None,
+    ):
+        self.level = level
+        self.not_levels = not_levels or []
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        if self.not_levels and record.levelno in self.not_levels:
+            return False
+        if self.level and record.levelno != self.level:
+            return False
+        return True
+
+
+class DuplicateTimesFormatter(logging.Formatter):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._last_timestamp: str | None = None
+
+    def formatTime(self, record, datefmt=None):
+        formatted_time = super().formatTime(record, datefmt=datefmt)
+        if self._last_timestamp and formatted_time == self._last_timestamp:
+            formatted_time = " " * len(formatted_time)
+        else:
+            self._last_timestamp = formatted_time
+        return formatted_time
+
+    def format(self, record):
+        if "\r\n" in record.msg:
+            line_split = "\r\n"
+        else:
+            line_split = "\n"
+        lines = record.msg.replace("\r\n", "\n").splitlines()
+        outlines = [lines.pop(0)]
+        if self._last_timestamp:
+            prefix = " " * len(self._last_timestamp)
+        else:
+            prefix = " " * len(self.formatTime(record, self.datefmt))
+            self._last_timestamp = None
+        outlines.extend([f"{prefix}{line.rstrip()}" for line in lines])
+        record.msg = line_split.join(outlines).rstrip()
+        if line_split.endswith("\r\n"):
+            record.msg += "\r"
+        return super().format(record)
+
+
+class LoggingClass(logging.Logger):
+    def stderr(self, msg, *args, **kwargs):
+        return self.log(STDERR, msg, *args, **kwargs)
+
+    def stdout(self, msg, *args, **kwargs):
+        return self.log(STDOUT, msg, *args, **kwargs)
+
+
+# Override the python's logging logger class as soon as this module is imported
+if logging.getLoggerClass() is not LoggingClass:
+    logging.setLoggerClass(LoggingClass)
+
+# Reset logging handlers
+logging.root.handlers.clear()
+logging.root.setLevel(logging.INFO)
+
+NO_TIMESTAMP_FORMATTER = logging.Formatter(fmt="%(message)s")
+TIMESTAMP_FORMATTER = DuplicateTimesFormatter(fmt="%(asctime)s%(message)s", datefmt="[%H:%M:%S] ")
+
+DEFAULT_FORMATTER: logging.Formatter | DuplicateTimesFormatter
+if "CI" in os.environ:
+    DEFAULT_FORMATTER = TIMESTAMP_FORMATTER
+else:
+    DEFAULT_FORMATTER = NO_TIMESTAMP_FORMATTER
+STDERR_HANDLER = logging.StreamHandler(stream=sys.stderr)
+STDERR_HANDLER.setLevel(STDERR)
+STDERR_HANDLER.addFilter(LevelFilter(level=STDERR))
+
+STDOUT_HANDLER = logging.StreamHandler(stream=sys.stdout)
+STDOUT_HANDLER.setLevel(STDOUT)
+STDOUT_HANDLER.addFilter(LevelFilter(level=STDOUT))
+
+ROOT_HANDLER = logging.StreamHandler(stream=sys.stderr)
+ROOT_HANDLER.setLevel(logging.DEBUG)
+ROOT_HANDLER.addFilter(LevelFilter(not_levels=(STDERR, STDOUT)))
+
+for handler in (ROOT_HANDLER, STDERR_HANDLER, STDOUT_HANDLER):
+    handler.setFormatter(DEFAULT_FORMATTER)
+    logging.root.addHandler(handler)
+
+
+def include_timestamps() -> bool:
+    """
+    Return True if any of the configured logging handlers includes timestamps.
+    """
+    for handler in logging.root.handlers:
+        if handler.formatter is TIMESTAMP_FORMATTER:
+            return True
+    return False

src/ptscripts/parser.py

@@ -18,9 +18,12 @@
 from typing import TYPE_CHECKING
 from typing import TypedDict
 
+import rich
 from rich.console import Console
 from rich.theme import Theme
 
+from ptscripts import logs
+
 if TYPE_CHECKING:
     from argparse import ArgumentParser
     from argparse import _SubParsersAction
@@ -69,6 +72,8 @@ def __init__(self, parser: Parser):
                 "log-error": "bold red",
                 "exit-ok": "green",
                 "exit-failure": "bold red",
+                "logging.level.stdout": "dim blue",
+                "logging.level.stderr": "dim red",
             }
         )
         console_kwargs = {
@@ -79,6 +84,7 @@ def __init__(self, parser: Parser):
             console_kwargs["force_interactive"] = False
         self.console = Console(stderr=True, **console_kwargs)
         self.console_stdout = Console(**console_kwargs)
+        rich.reconfigure(stderr=True, **console_kwargs)
 
     def print(self, *args, **kwargs):
         """
@@ -148,16 +154,33 @@ def __new__(cls):
                 epilog="These tools are discovered under `<repo-root>/tools`.",
                 allow_abbrev=False,
             )
-            group = instance.parser.add_mutually_exclusive_group()
-            # group.add_argument(
-            #    "--quiet",
-            #    "-q",
-            #    dest="quiet",
-            #    action="store_true",
-            #    default=False,
-            #    help="Show info messages",
-            # )
-            group.add_argument(
+            log_group = instance.parser.add_argument_group("Logging")
+            timestamp_meg = log_group.add_mutually_exclusive_group()
+            timestamp_meg.add_argument(
+                "--timestamps",
+                "--ts",
+                action="store_true",
+                help="Add time stamps to logs",
+                dest="timestamps",
+            )
+            timestamp_meg.add_argument(
+                "--no-timestamps",
+                "--nts",
+                action="store_false",
+                default=True,
+                help="Remove time stamps from logs",
+                dest="timestamps",
+            )
+            level_group = log_group.add_mutually_exclusive_group()
+            level_group.add_argument(
+                "--quiet",
+                "-q",
+                dest="quiet",
+                action="store_true",
+                default=False,
+                help="Disable logging",
+            )
+            level_group.add_argument(
                 "--debug",
                 "-d",
                 action="store_true",
@@ -176,12 +199,19 @@ def parse_args(self):
         Parse CLI.
         """
         options = self.parser.parse_args()
-        self.options = options
-        logging.root.setLevel(logging.INFO)
-        if options.debug:
+        if options.quiet:
+            logging.root.setLevel(logging.CRITICAL + 1)
+        elif options.debug:
             logging.root.setLevel(logging.DEBUG)
-        # elif options.quiet:
-        #    logging.root.setLevel(logging.FATAL)
+        else:
+            logging.root.setLevel(logging.INFO)
+        if options.timestamps:
+            for handler in logging.root.handlers:
+                handler.setFormatter(logs.TIMESTAMP_FORMATTER)
+        else:
+            for handler in logging.root.handlers:
+                handler.setFormatter(logs.NO_TIMESTAMP_FORMATTER)
+        self.options = options
         if "func" not in options:
             self.context.exit(1, "No command was passed.")
         log.debug(f"CLI parsed options {options}")
@@ -327,6 +357,7 @@ def command(
             flags = kwargs.pop("flags", None)  # type: ignore[misc]
             if flags is None:
                 flags = [f"--{parameter.name.replace('_', '-')}"]
+            log.debug("Adding Command %r. Flags: %s; KwArgs: %s", name, flags, kwargs)
             command.add_argument(*flags, **kwargs)
         command.set_defaults(func=partial(self, func))
         return func