"""Logging Module. This module is largely borrowed from Ansible Molecule and that is why it is largely uncommented since we didn't write it and luckily, didn't have to! This code just works TM. The same upstream license applies. """ import logging import os import sys import colorama def should_do_markup(): return sys.stdout.isatty() and os.environ.get("TERM") != "dumb" SUCCESS = 100 OUT = 101 class LogFilter(object): def __init__(self, level): self.__level = level def filter(self, logRecord): return logRecord.levelno <= self.__level class CustomLogger(logging.getLoggerClass()): # type: ignore def __init__(self, name, level=logging.NOTSET): super(logging.getLoggerClass(), self).__init__(name, level) logging.addLevelName(SUCCESS, "SUCCESS") logging.addLevelName(OUT, "OUT") def success(self, msg, *args, **kwargs): if self.isEnabledFor(SUCCESS): self._log(SUCCESS, msg, args, **kwargs) def out(self, msg, *args, **kwargs): if self.isEnabledFor(OUT): self._log(OUT, msg, args, **kwargs) class TrailingNewlineFormatter(logging.Formatter): def format(self, record): if record.msg: record.msg = record.msg.rstrip() return super(TrailingNewlineFormatter, self).format(record) def get_logger(name=None): logging.setLoggerClass(CustomLogger) logger = logging.getLogger(name) logger.setLevel(logging.DEBUG) logger.addHandler(_get_info_handler()) logger.addHandler(_get_out_handler()) logger.addHandler(_get_warn_handler()) logger.addHandler(_get_error_handler()) logger.addHandler(_get_critical_handler()) logger.addHandler(_get_success_handler()) logger.propagate = False return logger def _get_info_handler(): handler = logging.StreamHandler(sys.stdout) handler.setLevel(logging.INFO) handler.addFilter(LogFilter(logging.INFO)) handler.setFormatter( TrailingNewlineFormatter("--> {}".format(cyan_text("%(message)s"))) ) return handler def _get_out_handler(): handler = logging.StreamHandler(sys.stdout) handler.setLevel(OUT) handler.addFilter(LogFilter(OUT)) handler.setFormatter(TrailingNewlineFormatter(" %(message)s")) return handler def _get_warn_handler(): handler = logging.StreamHandler(sys.stdout) handler.setLevel(logging.WARN) handler.addFilter(LogFilter(logging.WARN)) handler.setFormatter(TrailingNewlineFormatter(yellow_text("%(message)s"))) return handler def _get_error_handler(): handler = logging.StreamHandler(sys.stderr) handler.setLevel(logging.ERROR) handler.addFilter(LogFilter(logging.ERROR)) handler.setFormatter(TrailingNewlineFormatter(red_text("%(message)s"))) return handler def _get_critical_handler(): handler = logging.StreamHandler(sys.stderr) handler.setLevel(logging.CRITICAL) handler.addFilter(LogFilter(logging.CRITICAL)) handler.setFormatter( TrailingNewlineFormatter(red_text("ERROR: %(message)s")) ) return handler def _get_success_handler(): handler = logging.StreamHandler(sys.stdout) handler.setLevel(SUCCESS) handler.addFilter(LogFilter(SUCCESS)) handler.setFormatter(TrailingNewlineFormatter(green_text("%(message)s"))) return handler def red_text(msg): return color_text(colorama.Fore.RED, msg) def yellow_text(msg): return color_text(colorama.Fore.YELLOW, msg) def green_text(msg): return color_text(colorama.Fore.GREEN, msg) def cyan_text(msg): return color_text(colorama.Fore.CYAN, msg) def color_text(color, msg): if should_do_markup(): return "{}{}{}".format(color, msg, colorama.Style.RESET_ALL) return msg log = get_logger(__name__)