from __future__ import print_function import syslog import sys import inspect import re from os.path import basename from datetime import datetime from typing import List, Tuple, Optional, Any from . import misc # --- python 2 / 3 compatibility stuff try: basestring # type: ignore except NameError: basestring = str _special_chars = { '\a' : '\\a', '\b' : '\\b', '\t' : '\\t', '\n' : '\\n', '\v' : '\\v', '\f' : '\\f', '\r' : '\\r', } _special_char_regex = re.compile("(%s)" % "|".join(map(re.escape, _special_chars.keys()))) EMERG = int(syslog.LOG_EMERG) ALERT = int(syslog.LOG_ALERT) CRIT = int(syslog.LOG_CRIT) ERR = int(syslog.LOG_ERR) WARNING = int(syslog.LOG_WARNING) NOTICE = int(syslog.LOG_NOTICE) INFO = int(syslog.LOG_INFO) DEBUG = int(syslog.LOG_DEBUG) DEVEL = int(syslog.LOG_DEBUG + 1) OFF = DEVEL + 1 _level = NOTICE CONSOLE_FONT_BOLD = '\033[1m' CONSOLE_FONT_RED = '\033[31m' CONSOLE_FONT_GREEN = '\033[32m' CONSOLE_FONT_YELLOW = '\033[33m' CONSOLE_FONT_BLUE = '\033[34m' CONSOLE_FONT_MAGENTA = '\033[35m' CONSOLE_FONT_CYAN = '\033[36m' CONSOLE_FONT_WHITE = '\033[37m' CONSOLE_FONT_BLINK = '\033[5m' CONSOLE_FONT_OFF = '\033[m' f_position = 'position' f_date = 'date' f_stderr = 'stderr' f_stdout = 'stdout' f_prio = 'prio' f_color = 'color' f_default = [ f_position, f_stderr, f_prio, f_color ] _flags = set(f_default) _log_prefix = '' _file_name_len = 20 _short_prio_str = { EMERG : '', ALERT : '', CRIT : '', ERR : '', WARNING : '', NOTICE : '', INFO : '', DEBUG : '', DEVEL : '', } _prio_colors = { DEVEL : [ "", "" ], DEBUG : [ "", "" ], INFO : [ CONSOLE_FONT_BLUE, CONSOLE_FONT_OFF ], NOTICE : [ CONSOLE_FONT_GREEN, CONSOLE_FONT_OFF ], WARNING : [ CONSOLE_FONT_YELLOW, CONSOLE_FONT_OFF ], ERR : [ CONSOLE_FONT_BOLD + CONSOLE_FONT_RED, CONSOLE_FONT_OFF ], CRIT : [ CONSOLE_FONT_BOLD + CONSOLE_FONT_MAGENTA, CONSOLE_FONT_OFF ], ALERT : [ CONSOLE_FONT_BOLD + CONSOLE_FONT_MAGENTA, CONSOLE_FONT_OFF ], EMERG : [ CONSOLE_FONT_BOLD + CONSOLE_FONT_MAGENTA, CONSOLE_FONT_OFF ], } def prio_gets_logged(prio: int) -> bool: # export if prio > _level: return False return True def log_level(s: str=None) -> int: # export if s is None: return _level def get_caller_pos(up: int = 1, kwargs: Optional[dict[str, Any]] = None) -> Tuple[str, int]: if kwargs and 'caller' in kwargs: r = kwargs['caller'] del kwargs['caller'] return r caller = inspect.stack()[up+1] return (basename(caller.filename), caller.lineno) def slog_m(prio: int, *args, **kwargs) -> None: # export if prio > _level: return if len(args): margs = '' for a in args: if isinstance(a, list): margs += '\n'.join([str(elem) for elem in a]) continue margs += ' ' + str(a) if 'caller' not in kwargs: caller = get_caller_pos(1) else: caller = kwargs['caller'] del kwargs['caller'] for line in margs[1:].split('\n'): slog(prio, line, **kwargs, caller=caller) def slog(prio: int, *args, only_printable: bool=False, **kwargs) -> None: # export if prio > _level: return msg = '' color_on = '' color_off = '' if f_date in _flags: msg += datetime.now().strftime("%b %d %H:%M:%S.%f ") if f_prio in _flags: msg += _short_prio_str[prio] + ' ' if f_position in _flags: if 'caller' in kwargs: name, line = kwargs['caller'] else: name, line = get_caller_pos(1) msg += misc.pad(name, _file_name_len) + '[' + misc.pad(str(line), 4, True) + ']' if f_color in _flags: color_on, color_off = console_color_chars(prio) msg += _log_prefix if len(args): margs = '' for a in args: margs += ' ' + str(a) if only_printable: margs = _special_char_regex.sub(lambda mo: _special_chars[mo.string[mo.start():mo.end()]], margs) margs = re.sub('[\x01-\x1f]', '.', margs) msg += color_on + margs + color_off if not len(msg): return files = [] if f_stdout in _flags: files.append(sys.stdout) if f_stderr in _flags: files.append(sys.stderr) if not len(files): files = [ sys.stdout ] for file in files: print(msg, file=file) def throw(*args, prio=ERR, caller=None, **kwargs) -> None: if caller is None: caller = get_caller_pos(1) msg = ' '.join([str(arg) for arg in args]) slog(prio, msg, caller=caller) raise Exception(msg) def parse_log_prio_str(prio: str) -> int: # export try: r = int(prio) if r < 0 or r > DEVEL: raise Exception("Invalid log priority ", prio) except ValueError: map_prio_str_to_val = { "EMERG" : EMERG, "emerg" : EMERG, "ALERT" : ALERT, "alert" : ALERT, "CRIT" : CRIT, "crit" : CRIT, "ERR" : ERR, "err" : ERR, "WARNING" : WARNING, "warning" : WARNING, "NOTICE" : NOTICE, "notice" : NOTICE, "INFO" : INFO, "info" : INFO, "DEBUG" : DEBUG, "debug" : DEBUG, "DEVEL" : DEVEL, "devel" : DEVEL, "OFF" : OFF, "off" : OFF, } if prio in map_prio_str_to_val: return map_prio_str_to_val[prio] raise Exception("Unknown priority string \"", prio, "\"") def console_color_chars(prio: int) -> List[str]: # export if not sys.stdout.isatty(): return [ '', '' ] return _prio_colors[prio] def set_level(level_: str) -> None: # export global _level if isinstance(level_, basestring): _level = parse_log_prio_str(level_) return _level = level_ def set_flags(flags_: str) -> None: # export global _flags _flags = set(flags_.split(',')) #syslog #console #color #prio #position #ide #trace_rename_thread_to_shorter #trace_rename_thread_to_longer #trace_inout #skip_openlog #id #date #pid #highlight_first_error def append_to_prefix(prefix: str) -> str: # export global _log_prefix r = _log_prefix if prefix: _log_prefix += prefix return r def remove_from_prefix(count) -> str: # export if isinstance(count, str): count = len(count) global _log_prefix r = _log_prefix _log_prefix = _log_prefix[:-count] return r def set_filename_length(l: int) -> int: # export global _file_name_len r = _file_name_len if l: _file_name_len = l return r