import typing as _typing import enum as _enum import sys as _sys import json as _json import datetime as _datetime from helpers import * class enum_log_level(_enum.Enum): DEBUG = 0 NOTICE = 1 INFO = 2 WARNING = 3 ERROR = 4 def __lt__(self, other): return (self.value < other.value) _log_outputs = [] def log_add_output( output ): _log_outputs.append(output) def log_generic( level, report ): def encode_level(level): return { enum_log_level.ERROR: "err", enum_log_level.WARNING: "wrn", enum_log_level.NOTICE: "ntc", enum_log_level.INFO: "inf", enum_log_level.DEBUG: "dbg", }[level] for output in _log_outputs: if (level < output["min_level"]): pass else: if (output["format"] == "jsonl"): _sys.stderr.write( _json.dumps( { "level": encode_level(level), "timestamp": int(_datetime.datetime.now().timestamp()), "incident": report["incident"], "details": report["details"], } ) + "\n" ) elif (output["format"] == "human_readable"): _sys.stderr.write( string_coin( "<{{time}}> [{{level}}] {{incident}} | {{details}}", { "time": _datetime.datetime.now().isoformat()[:19], "level": encode_level(level), "incident": report["incident"], "details": _json.dumps(report["details"])[:127], } ) + "\n" ) else: raise ValueError("invalid log format: %s" % output["format"]) def log_error( incident, details = None ): log_generic( enum_log_level.ERROR, { "incident": incident, "details": (details or {}) } ) def log_warning( incident, details = None ): log_generic( enum_log_level.WARNING, { "incident": incident, "details": (details or {}) } ) def log_notice( incident, details = None ): log_generic( enum_log_level.NOTICE, { "incident": incident, "details": (details or {}) } ) def log_info( incident, details = None ): log_generic( enum_log_level.INFO, { "incident": incident, "details": (details or {}) } ) def log_debug( incident, details = None ): log_generic( enum_log_level.DEBUG, { "incident": incident, "details": (details or {}) } )