Source code for sevaht_utility.log_utility

"""Opinionated logging setup and helpers.

Provides a console-plus-optional-rotating-file logging configuration
(:func:`configure_logging`, :func:`configure_logging_custom`), argparse
integration (:func:`add_log_arguments`), a context manager to silence the
console temporarily (:func:`suppress_console_logging`), and a decorator that
logs and re-raises uncaught exceptions (:func:`log_exceptions`).
"""

from __future__ import annotations

import logging
import sys
from contextlib import contextmanager
from dataclasses import KW_ONLY, dataclass
from functools import wraps
from logging import Handler
from logging.handlers import RotatingFileHandler
from pathlib import Path
from typing import TYPE_CHECKING, ParamSpec, TypeVar

if TYPE_CHECKING:
    import argparse
    from collections.abc import Callable, Iterator


P = ParamSpec("P")
R = TypeVar("R")


logger = logging.getLogger(__name__)




@dataclass
class LogFileOptions:
    path: Path
    _: KW_ONLY
    max_kb: int
    backup_count: int
    level: int = logging.DEBUG
    encoding: str = "utf-8"
    append: bool = True

    def create_handler(self) -> Handler:
        handler = RotatingFileHandler(
            self.path,
            mode="a" if self.append else "w",
            encoding=self.encoding,
            maxBytes=self.max_kb * 1024,
            backupCount=self.backup_count,
        )
        handler.setLevel(self.level)
        return handler





def configure_logging_custom(
    console_level: int, log_file_options: LogFileOptions | None = None
) -> None:
    """Install a console handler and an optional rotating file handler.

    Replaces the root logger's handlers. The console handler honors
    ``console_level`` and drops records flagged ``file_only``; when
    ``log_file_options`` is given, a file handler captures everything at its
    own level with timestamps.

    Args:
        console_level: Minimum level shown on the console.
        log_file_options: File logging configuration, or ``None`` for
            console-only.
    """

    class SuppressFileOnly(logging.Filter):
        def filter(self, record: logging.LogRecord) -> bool:
            return not getattr(record, "file_only", False)

    logging.getLogger().handlers = []
    console_handler = logging.StreamHandler()
    console_handler.setLevel(console_level)
    console_handler.setFormatter(
        logging.Formatter(fmt="{levelname:s}: {message:s}", style="{")
    )
    console_handler.addFilter(SuppressFileOnly())
    logging.getLogger().addHandler(console_handler)
    global_level = console_level
    if log_file_options:
        global_level = min(global_level, log_file_options.level)
        file_handler = log_file_options.create_handler()
        file_handler.setFormatter(
            logging.Formatter(
                fmt=(
                    "[{asctime:s}.{msecs:03.0f}]"
                    " [{levelname:s}] {module:s}: {message:s}"
                ),
                datefmt="%Y-%m-%d %H:%M:%S",
                style="{",
            )
        )
        logging.getLogger().addHandler(file_handler)
    logging.getLogger().setLevel(global_level)
    logger.info("logging configured")





def add_log_arguments(parser: argparse.ArgumentParser) -> None:
    """Add standard logging options to an argument parser.

    Adds ``--log-file`` and a mutually exclusive verbosity group
    (``-v``/``--verbose``, ``-q``/``--quiet``, ``--debug``) writing to the
    ``console_level`` and ``log_file`` destinations consumed by
    :func:`configure_logging`.

    Args:
        parser: The parser (or subparser) to extend.
    """
    log_group = parser.add_argument_group("logging")
    log_group.add_argument(
        "--log-file",
        metavar="FILE",
        help="Path to a file where logs will be written, if specified.",
    )
    log_verbosity_group = log_group.add_mutually_exclusive_group(
        required=False
    )
    log_verbosity_group.add_argument(
        "-v",
        "--verbose",
        action="store_const",
        dest="console_level",
        const=logging.INFO,
        help="Increase console log level to INFO.",
    )
    log_verbosity_group.add_argument(
        "-q",
        "--quiet",
        action="store_const",
        dest="console_level",
        const=logging.ERROR,
        help="Decrease console log level to ERROR.  Overrides -v.",
    )
    log_verbosity_group.add_argument(
        "--debug",
        action="store_const",
        dest="console_level",
        const=logging.DEBUG,
        help="Maximizes console log verbosity to DEBUG.  Overrides -v and -q.",
    )





def configure_logging(
    args: argparse.Namespace,
    *,
    max_kb: int = 512,
    backup_count: int = 1,
    append: bool = True,
) -> None:
    """Configure logging from parsed :func:`add_log_arguments` options.

    The console level comes from ``args.console_level`` (default ``WARNING``);
    when ``args.log_file`` is set, a rotating file handler is added.

    Args:
        args: Parsed arguments containing ``console_level`` and ``log_file``.
        max_kb: Max size per log file in KiB before rotation; ``0`` disables
            rotation.
        backup_count: Number of rotated backups to keep; ``0`` keeps none.
        append: Whether to append to an existing log file rather than truncate.
    """
    configure_logging_custom(
        console_level=args.console_level or logging.WARNING,
        log_file_options=(
            None
            if not args.log_file
            else LogFileOptions(
                path=Path(args.log_file),
                max_kb=max_kb,  # 0 for unbounded size and no rotation
                backup_count=backup_count,  # 0 for no rolling backups
                append=append,
            )
        ),
    )



def is_console_handler(handler: logging.Handler) -> bool:
    if not isinstance(handler, logging.StreamHandler):
        return False

    return handler.stream in (sys.stdout, sys.stderr)




@contextmanager
def suppress_console_logging() -> Iterator[None]:
    """
    Temporarily remove logging handlers that write to the terminal.

    This affects only StreamHandlers whose stream is sys.stdout or sys.stderr.
    Other handlers (file, syslog, HTTP, custom streams) remain intact.

    All removed handlers are restored after the context exits.
    """
    root_logger = logging.getLogger()
    removed_handlers: list[logging.Handler] = []
    try:
        for handler in list(root_logger.handlers):
            if is_console_handler(handler):
                root_logger.removeHandler(handler)
                removed_handlers.append(handler)
        yield
    finally:
        for handler in removed_handlers:
            root_logger.addHandler(handler)





def log_exceptions(
    *,
    logger: logging.Logger | None = None,
    message: str = "uncaught exception",
    file_only: bool = True,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """Decorate a callable to log any exception it raises, then re-raise.

    The exception is logged with a traceback; the call still propagates it.

    Args:
        logger: Logger to use; defaults to one named for the wrapped function's
            module.
        message: Message logged alongside the traceback.
        file_only: Tag the record so :func:`configure_logging_custom`'s console
            handler suppresses it (file handlers still receive it).

    Returns:
        A decorator that wraps a function while preserving its signature.
    """

    def decorator(function: Callable[P, R]) -> Callable[P, R]:
        target_logger = logger or logging.getLogger(function.__module__)

        @wraps(function)
        def wrapped(*args: P.args, **kwargs: P.kwargs) -> R:
            try:
                return function(*args, **kwargs)
            except Exception:
                target_logger.exception(
                    message, extra={"file_only": file_only}
                )
                raise

        return wrapped

    return decorator