py_common_logger/common_logger.py

"""Common logging for Python scripts and CLI programs."""

import argparse
import datetime
import inspect
import logging
import logging.config
import logging.handlers
import sys
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path
from typing import ClassVar

DEBUG = logging.DEBUG
INFO = logging.INFO
WARNING = logging.WARNING
ERROR = logging.ERROR
CRITICAL = logging.CRITICAL
FATAL = logging.FATAL


@dataclass(frozen=True)
class PlatformDef:
    """Define some platform specific values."""

    socket: str
    logdir: str

    _DEFS: ClassVar[dict[str, "PlatformDef"]]

    @classmethod
    def for_platform(cls, platform: str = sys.platform) -> "PlatformDef":
        """Return the correct values for the current platform."""
        return cls._DEFS.get(platform, cls._DEFS["linux"])


PlatformDef._DEFS = {  # noqa: SLF001  # pylint: disable=protected-access
    "darwin": PlatformDef(
        socket="/var/run/syslog",
        logdir=Path("~/Library/Logs").expanduser().as_posix(),
    ),
    "linux": PlatformDef(
        socket="/dev/log",
        logdir=Path("/srv/log").as_posix(),
    ),
}


class UcFormatter(logging.Formatter):
    """Class for formatting the date & time correctly when logging."""

    def converter(self, timestamp: float = 0.0) -> datetime.datetime:
        """Add the local timezone to the the provided timestamp."""
        tzinfo = datetime.datetime.now(datetime.UTC).astimezone().tzinfo or datetime.UTC
        return datetime.datetime.fromtimestamp(timestamp, tz=tzinfo)

    def formatTime(  # noqa: N802
        self,
        record: logging.LogRecord,
        datefmt: str = "",
    ) -> str:
        """Format the timestamp for the log record."""
        recdt = self.converter(record.created)
        if datefmt:
            return recdt.strftime(datefmt)

        try:
            return recdt.isoformat(timespec="seconds")
        except TypeError:
            return recdt.isoformat()


class Logger:
    """Class for implementing a consistent logging format and location."""

    def __init__(  # TODO @scott: #000 pylint: disable=too-many-arguments
        self,
        logname: str = "",
        *,
        level: int = INFO,
        enable_logfile: bool = True,
        syslog_facility: int = logging.handlers.SysLogHandler.LOG_USER,
        logpath: str = PlatformDef.for_platform().logdir,
    ) -> None:
        """Initialize the logger."""
        self.enable_logfile = enable_logfile

        # Configure logging basics
        logger = logging.getLogger()
        logging.config.dictConfig(
            {
                "version": 1,
                "disable_existing_loggers": True,
            },
        )
        logger.setLevel(level)

        # Add stream handler
        stream_handler = logging.StreamHandler()
        logger.addHandler(stream_handler)

        if logname:
            self.source = logname
        else:
            self.source = Path(
                inspect.getframeinfo(inspect.stack()[-1][0]).filename,
            ).name

        # Set log formatting
        basefmt = f"%(levelname)s {self.source}: %(message)s"
        local_fmt = UcFormatter(
            fmt=f"%(asctime)s {basefmt}",
            datefmt="%Y-%m-%d %H:%M:%S %z",
        )
        syslog_fmt = logging.Formatter(fmt=basefmt)

        # Set formatter for StreamHandler
        stream_handler.setFormatter(local_fmt)

        if self.enable_logfile:
            # Set file handler
            try:
                file_handler = logging.handlers.TimedRotatingFileHandler(
                    Path(logpath) / f"{Path(self.source).stem}.log",
                    when="midnight",
                    backupCount=90,
                )
            except OSError as error:
                logging.warning("FileHandler: %s", error)  # noqa: LOG015
            else:
                # Configure main logger with file handler
                file_handler.setFormatter(local_fmt)
                logger.addHandler(file_handler)

        syslog_socket = PlatformDef.for_platform().socket
        try:
            syslog_handler = logging.handlers.SysLogHandler(
                address=syslog_socket,
                facility=syslog_facility,
            )
        except OSError as error:
            logging.warning(  # noqa: LOG015
                "SyslogHandler: %s: %s",
                error,
                syslog_socket,
            )
        else:
            syslog_handler.setFormatter(syslog_fmt)
            logger.addHandler(syslog_handler)

        self.logger = logger

    def __getattr__(self, name: str) -> Callable:
        """Ensure calls to info(), warning(), etc. are passed through to self.logger."""
        return getattr(self.logger, name)

    def __dir__(self) -> list[str]:
        """Ensure the underlying methods in self.logger are discoverable."""
        return list(super().__dir__()) + dir(self.logger)


if __name__ == "__main__":

    def parse_args() -> argparse.Namespace:
        """Parse the CLI arguments."""
        parser = argparse.ArgumentParser()

        group = parser.add_mutually_exclusive_group()
        for level in ["debug", "info", "warning", "error"]:
            group.add_argument(
                f"-{level[0]}",
                f"--{level}",
                action="store_true",
                help=f"log message at level {level.upper()}",
            )

        parser.add_argument(
            "-n",
            "--name",
            nargs=1,
            default=[None],
            help="basename of the log file to write to",
        )

        parser.add_argument(
            "message",
            nargs=argparse.REMAINDER,
            help="message to log.  Reads STDIN if not provided.",
        )

        return parser.parse_args()

    def main() -> None:
        """Act as the main entrypoint for the CLI."""
        args = parse_args()

        logger = Logger(args.name[0])

        if args.debug:
            log = logger.debug
        elif args.info:
            log = logger.info
        elif args.warning:
            log = logger.warning
        elif args.error:
            log = logger.error
        else:
            # Default to INFO
            log = logger.info

        if args.message:
            log(" ".join(args.message))

        # Check if we have data from stdin
        if not sys.stdin.isatty():
            data = sys.stdin.read().strip()
            if data:
                for line in data.split("\n"):
                    log(line)

    main()