SpotifyRecAlg/swingmusic/logger.py

"""
Logger module
"""

import datetime as dt
import json
import logging
import logging.config
import logging.handlers
from pathlib import Path

LOG_RECORD_BUILTIN_ATTRS = {
    "args",
    "asctime",
    "created",
    "exc_info",
    "exc_text",
    "filename",
    "funcName",
    "levelname",
    "levelno",
    "lineno",
    "module",
    "msecs",
    "message",
    "msg",
    "name",
    "pathname",
    "process",
    "processName",
    "relativeCreated",
    "stack_info",
    "thread",
    "threadName",
    "taskName",
}


class JsonFormat(logging.Formatter):
    def __init__(
        self,
        *,
        fmt_keys: dict[str, str] | None = None,
    ):

        super().__init__()
        self.fmt_keys = fmt_keys or {}

    def format(self, record: logging.LogRecord) -> str:
        message = self._prepare_log_dict(record)
        return json.dumps(message, default=str)

    def _prepare_log_dict(self, record: logging.LogRecord):
        always_fields = {
            "args": record.args,
            "name": record.name,
            "line": record.lineno,
            "message": record.getMessage(),
            "timestamp": dt.datetime.fromtimestamp(
                record.created, tz=dt.UTC
            ).isoformat(),
            "who": record.name,
        }

        if record.exc_info is not None:
            always_fields["exc_info"] = self.formatException(record.exc_info)

        if record.stack_info is not None:
            always_fields["stack_info"] = self.formatStack(record.stack_info)

        message = {}

        for key, val in self.fmt_keys.items():
            if (msg_val := always_fields.pop(val, None)) is not None:
                message[key] = msg_val
            else:
                message[key] = getattr(record, val)

        message.update(always_fields)

        for key, val in record.__dict__.items():
            if key not in LOG_RECORD_BUILTIN_ATTRS:
                message[key] = val

        return message


class CustomFormatter(logging.Formatter):
    """
    Custom log formatter
    """

    grey = "\033[92m"
    yellow = "\x1b[33;20m"
    red = "\033[41m"
    bold_red = "\x1b[31;1m"
    reset = "\x1b[0m"
    # format_ = "[%(asctime)s] %(name)s %(levelname)s %(message)s (%(filename)s:%(lineno)d)"
    format_ = "[%(asctime)s] [%(levelname)s] %(message)s (%(filename)s:%(lineno)d)"
    # format_ = "%(message)s"

    FORMATS = {
        logging.DEBUG: grey + format_ + reset,
        logging.INFO: grey + format_ + reset,
        logging.WARNING: yellow + format_ + reset,
        logging.ERROR: red + format_ + reset,
        logging.CRITICAL: bold_red + format_ + reset,
    }

    def __init__(
        self,
        *,
        fmt_keys: dict[str, str] | None = None,
    ):

        super().__init__()
        self.fmt_keys = fmt_keys or {}

    def format(self, record):
        log_fmt = self.FORMATS.get(record.levelno)
        # record.exc_info = None
        # record.exc_text = None
        self._style = logging.PercentStyle(log_fmt)
        self._fmt = self._style._fmt

        self.datefmt = "%H:%M:%S"
        return super().format(record)

    def formatException(self, e):
        # Exceptions are logged to file but not shown on CLI to avoid cluttering output.
        # Non-terminal exceptions are handled gracefully - the user doesn't need to be
        # informed as they don't affect normal operation.
        return ""

    def formatStack(self, stack_info):
        return ""


CONFIG = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "json": {
            "()": JsonFormat,
            "fmt_keys": {
                "level": "levelname",
                "message": "message",
                "timestamp": "timestamp",
                "logger": "name",
                "module": "module",
                "function": "funcName",
                "line": "lineno",
            },
        },
        "custom": {
            "()": CustomFormatter,
            "fmt_keys": {
                "level": "levelname",
                "message": "message",
                "timestamp": "timestamp",
                "logger": "name",
                "module": "module",
                "function": "funcName",
                "line": "lineno",
            },
        },
    },
    "handlers": {
        "stdout": {
            "class": "logging.StreamHandler",
            "level": "INFO",
            "formatter": "custom",
            "stream": "ext://sys.stderr",
        },
        "file": {
            "class": "logging.handlers.RotatingFileHandler",
            "level": "DEBUG",
            "formatter": "json",
            "maxBytes": 5 * 1024 * 1024,  # 5 MB
            "backupCount": 5,
        },
        "remote": {
            "class": "logging.handlers.SocketHandler",
            "level": "DEBUG",
            "formatter": "json",
            "host": "127.0.0.2",
            "port": "19996",
        },
    },
    "loggers": {
        "swingmusic": {
            "level": "DEBUG",
            "propagate": False,
            "handlers": ["stdout", "file"],
        },
        "waitress": {
            "level": "ERROR",
            "propagate": False,
            "handlers": ["stdout", "file"],
        },
    },
}

# Always expose a usable logger object, even before setup_logger runs.
log = logging.getLogger("swingmusic")


def get_logger(name: str | None = None) -> logging.Logger:
    """
    Returns a configured logger instance.
    Falls back to stdlib logger if setup_logger has not run yet.
    """
    if name:
        return logging.getLogger(name)
    return log or logging.getLogger("swingmusic")


def debug(message, *args, **kwargs):
    get_logger().debug(message, *args, **kwargs)


def info(message, *args, **kwargs):
    get_logger().info(message, *args, **kwargs)


def warning(message, *args, **kwargs):
    get_logger().warning(message, *args, **kwargs)


def error(message, *args, **kwargs):
    get_logger().error(message, *args, **kwargs)


def critical(message, *args, **kwargs):
    get_logger().critical(message, *args, **kwargs)


def exception(message, *args, **kwargs):
    get_logger().exception(message, *args, **kwargs)


def setup_logger(app_dir: Path, debug=False):
    """
    setup logger
    needs to be called at the beginning and at least once

    :param app_dir: logging directory
    :param debug: When True Loglevel is set to DEBUG and enable Socket log
    """

    if Path.home().resolve().as_posix() == app_dir.resolve().as_posix():
        app_name = ".swingmusic"
    else:
        app_name = "swingmusic"

    log_dir = Path(app_dir) / app_name / "logs"
    if not log_dir.exists():
        log_dir.mkdir(parents=True)

    CONFIG["handlers"]["file"]["filename"] = log_dir / "log.jsonl"

    # enable socket log
    if debug:
        logging.warning("YOU ARE IN DEBUG MODE.")
        for key in CONFIG["loggers"]:
            CONFIG["loggers"][key]["handlers"].append("remote")
            CONFIG["loggers"][key]["level"] = "DEBUG"

    logging.config.dictConfig(CONFIG)

    global log
    log = logging.getLogger("swingmusic")