# Logbook

# Source code for logbook.compat

    """
        logbook.compat
        ~~~~~~~~~~~~~~

        Backwards compatibility with stdlib's logging package and the
        warnings module.

        :copyright: (c) 2010 by Armin Ronacher, Georg Brandl.
        :license: BSD, see LICENSE for more details.
    """
    import logging
    import sys
    import warnings
    from collections.abc import Mapping
    from datetime import date, datetime, timezone

    import logbook

    def redirect_logging(set_root_logger_level=True):
        """Permanently redirects logging to the stdlib.  This also
        removes all otherwise registered handlers on root logger of
        the logging system but leaves the other loggers untouched.

        :param set_root_logger_level: controls of the default level of the legacy
           root logger is changed so that all legacy log messages get redirected
           to Logbook
        """
        del logging.root.handlers[:]
        logging.root.addHandler(RedirectLoggingHandler())
        if set_root_logger_level:
            logging.root.setLevel(logging.DEBUG)

    class redirected_logging:
        """Temporarily redirects logging for all threads and reverts
        it later to the old handlers.  Mainly used by the internal
        unittests::

            from logbook.compat import redirected_logging
            with redirected_logging():
                ...
        """

        def __init__(self, set_root_logger_level=True):
            self.old_handlers = logging.root.handlers[:]
            self.old_level = logging.root.level
            self.set_root_logger_level = set_root_logger_level

        def start(self):
            redirect_logging(self.set_root_logger_level)

        def end(self, etype=None, evalue=None, tb=None):
            logging.root.handlers[:] = self.old_handlers
            logging.root.setLevel(self.old_level)

        __enter__ = start
        __exit__ = end

    class LoggingCompatRecord(logbook.LogRecord):
        def _format_message(self, msg, *args, **kwargs):
            if kwargs:
                assert not args
                return msg % kwargs
            else:
                assert not kwargs
                return msg % tuple(args)

    class RedirectLoggingHandler(logging.Handler):
        """A handler for the stdlib's logging system that redirects
        transparently to logbook.  This is used by the
        :func:`redirect_logging` and :func:`redirected_logging`
        functions.

        If you want to customize the redirecting you can subclass it.
        """

        def __init__(self):
            logging.Handler.__init__(self)

        def convert_level(self, level):
            """Converts a logging level into a logbook level."""
            if level >= logging.CRITICAL:
                return logbook.CRITICAL
            if level >= logging.ERROR:
                return logbook.ERROR
            if level >= logging.WARNING:
                return logbook.WARNING
            if level >= logging.INFO:
                return logbook.INFO
            return logbook.DEBUG

        def find_extra(self, old_record):
            """Tries to find custom data from the old logging record.  The
            return value is a dictionary that is merged with the log record
            extra dictionaries.
            """
            rv = vars(old_record).copy()
            for key in (
                "name",
                "msg",
                "args",
                "levelname",
                "levelno",
                "pathname",
                "filename",
                "module",
                "exc_info",
                "exc_text",
                "lineno",
                "funcName",
                "created",
                "msecs",
                "relativeCreated",
                "thread",
                "threadName",
                "greenlet",
                "processName",
                "process",
            ):
                rv.pop(key, None)
            return rv

        def find_caller(self, old_record):
            """Tries to find the caller that issued the call."""
            frm = sys._getframe(2)
            while frm is not None:
                if (
                    frm.f_globals is globals()
                    or frm.f_globals is logbook.base.__dict__
                    or frm.f_globals is logging.__dict__
                ):
                    frm = frm.f_back
                else:
                    return frm

        def convert_time(self, timestamp):
            """Converts the UNIX timestamp of the old record into a
            datetime object as used by logbook.
            """
            return datetime.utcfromtimestamp(timestamp)

        def convert_record(self, old_record):
            """Converts an old logging record into a logbook log record."""
            args = old_record.args
            kwargs = None

            # Logging allows passing a mapping object, in which case args will be a mapping.
            if isinstance(args, Mapping):
                kwargs = args
                args = None
            record = LoggingCompatRecord(
                old_record.name,
                self.convert_level(old_record.levelno),
                old_record.msg,
                args,
                kwargs,
                old_record.exc_info,
                self.find_extra(old_record),
                self.find_caller(old_record),
            )
            record.time = self.convert_time(old_record.created)
            return record

        def emit(self, record):
            logbook.dispatch_record(self.convert_record(record))

    class LoggingHandler(logbook.Handler):
        """Does the opposite of the :class:`RedirectLoggingHandler`, it sends
        messages from logbook to logging.  Because of that, it's a very bad
        idea to configure both.

        This handler is for logbook and will pass stuff over to a logger
        from the standard library.

        Example usage::

            from logbook.compat import LoggingHandler, warn
            with LoggingHandler():
                warn('This goes to logging')
        """

        def __init__(self, logger=None, level=logbook.NOTSET, filter=None, bubble=False):
            logbook.Handler.__init__(self, level, filter, bubble)
            if logger is None:
                logger = logging.getLogger()
            elif isinstance(logger, str):
                logger = logging.getLogger(logger)
            self.logger = logger

        def get_logger(self, record):
            """Returns the logger to use for this record.  This implementation
            always return :attr:`logger`.
            """
            return self.logger

        def convert_level(self, level):
            """Converts a logbook level into a logging level."""
            if level >= logbook.CRITICAL:
                return logging.CRITICAL
            if level >= logbook.ERROR:
                return logging.ERROR
            if level >= logbook.WARNING:
                return logging.WARNING
            if level >= logbook.INFO:
                return logging.INFO
            return logging.DEBUG

        def convert_time(self, dt):
            """Converts a datetime object into a timestamp."""
            if dt.tzinfo is None:
                # Logbook uses naive datetimes to represent UTC (utcnow)
                return dt.replace(tzinfo=timezone.utc).timestamp()
            return dt.timestamp()

        def convert_record(self, old_record):
            """Converts a record from logbook to logging."""
            record = logging.LogRecord(
                old_record.channel,
                self.convert_level(old_record.level),
                old_record.filename,
                old_record.lineno,
                old_record.message,
                (),
                old_record.exc_info,
                func=old_record.func_name,
            )
            for key, value in old_record.extra.items():
                record.__dict__.setdefault(key, value)
            record.created = self.convert_time(old_record.time)
            return record

        def emit(self, record):
            self.get_logger(record).handle(self.convert_record(record))

    def redirect_warnings():
        """Like :func:`redirected_warnings` but will redirect all warnings
        to the shutdown of the interpreter:

        .. code-block:: python

            from logbook.compat import redirect_warnings
            redirect_warnings()
        """
        redirected_warnings().__enter__()

    class redirected_warnings:
        """A context manager that copies and restores the warnings filter upon
        exiting the context, and logs warnings using the logbook system.

        The :attr:`~logbook.LogRecord.channel` attribute of the log record will be
        the import name of the warning.

        Example usage:

        .. code-block:: python

            from logbook.compat import redirected_warnings
            from warnings import warn

            with redirected_warnings():
                warn(DeprecationWarning('logging should be deprecated'))
        """

        def __init__(self):
            self._entered = False

        def message_to_unicode(self, message):
            return str(message)

        def make_record(self, message, exception, filename, lineno):
            category = exception.__name__
            if exception.__module__ not in ("exceptions", "builtins"):
                category = exception.__module__ + "." + category
            rv = logbook.LogRecord(category, logbook.WARNING, message)
            # we don't know the caller, but we get that information from the
            # warning system.  Just attach them.
            rv.filename = filename
            rv.lineno = lineno
            return rv

        def start(self):
            if self._entered:  # pragma: no cover
                raise RuntimeError("Cannot enter %r twice" % self)
            self._entered = True
            self._filters = warnings.filters
            warnings.filters = self._filters[:]
            self._showwarning = warnings.showwarning

            def showwarning(message, category, filename, lineno, file=None, line=None):
                message = self.message_to_unicode(message)
                record = self.make_record(message, category, filename, lineno)
                logbook.dispatch_record(record)

            warnings.showwarning = showwarning

        def end(self, etype=None, evalue=None, tb=None):
            if not self._entered:  # pragma: no cover
                raise RuntimeError("Cannot exit %r without entering first" % self)
            warnings.filters = self._filters
            warnings.showwarning = self._showwarning

        __enter__ = start
        __exit__ = end