140 lines
2.2 KiB
Python
140 lines
2.2 KiB
Python
|
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 {})
|
||
|
}
|
||
|
)
|
||
|
|