pluto/handler/util.py

170 lines
5.2 KiB
Python
Raw Normal View History

# Other library imports
from httpx import ReadTimeout, ReadError, RemoteProtocolError
from datetime import datetime
import logging
log = logging.getLogger("util")
BLACK, RED, GREEN, YELLOW, BLUE, MAGENTA, CYAN, WHITE = range(8)
# The background is set with 40 plus the number of the color, and the foreground with 30
# These are the sequences need to get colored ouput
RESET_SEQ = "\033[0m"
COLOR_SEQ = "\033[1;%dm"
BOLD_SEQ = "\033[1m"
def formatter_message(message, use_color=True):
if use_color:
message = message.replace("$RESET", RESET_SEQ).replace("$BOLD", BOLD_SEQ)
else:
message = message.replace("$RESET", "").replace("$BOLD", "")
return message
COLORS = {"WARNING": YELLOW, "INFO": WHITE, "DEBUG": BLUE, "CRITICAL": YELLOW, "ERROR": RED}
class ColoredFormatter(logging.Formatter):
def __init__(self, msg, use_color=True):
logging.Formatter.__init__(self, msg)
self.use_color = use_color
def format(self, record):
levelname = record.levelname
if self.use_color and levelname in COLORS:
levelname_color = COLOR_SEQ % (30 + COLORS[levelname]) + levelname + RESET_SEQ
record.levelname = levelname_color
return logging.Formatter.format(self, record)
def get_logger(name):
# Define the logging format
FORMAT = "%(asctime)s %(levelname)s $BOLD%(name)13s$RESET - %(message)s"
COLOR_FORMAT = formatter_message(FORMAT, True)
color_formatter = ColoredFormatter(COLOR_FORMAT)
# formatter = logging.Formatter(
# Why is this so complicated?
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
# ch.setFormatter(formatter)
ch.setFormatter(color_formatter)
# Define the logger on the base class
log = logging.getLogger(name)
# Add the handler and stop it being silly and printing everything twice
log.addHandler(ch)
log.propagate = False
return log
class ColoredLogger(logging.Logger):
FORMAT = "[$BOLD%(name)-20s$RESET][%(levelname)-18s] %(message)s ($BOLD%(filename)s$RESET:%(lineno)d)"
COLOR_FORMAT = formatter_message(FORMAT, True)
def __init__(self, name):
logging.Logger.__init__(self, name, logging.DEBUG)
color_formatter = ColoredFormatter(self.COLOR_FORMAT)
console = logging.StreamHandler()
console.setFormatter(color_formatter)
self.addHandler(console)
return
class Base(object):
def __init__(self):
name = self.__class__.__name__
# Set up all the logging stuff
2022-03-06 10:03:52 +00:00
self.log = get_logger(name)
self.log.info("Class initialised")
def xmerge_attrs(init_map):
"""
Given a dictionary of strings and classes, set all corresponding class.<string> attributes
on each class, to every other class.
"a": A(), "b": B() -> A.b = B_instance, B.a = A_instance
:param init_map: dict of class names to classes
"""
for classname, object_instance in init_map.items():
# notify, Notify
for classname_inside, object_instance_inside in init_map.items():
if not classname == classname_inside:
# irc, bot
setattr(object_instance, classname_inside, object_instance_inside)
def convert(data):
"""
Recursively convert a dictionary.
"""
if isinstance(data, bytes):
return data.decode("ascii")
if isinstance(data, dict):
return dict(map(convert, data.items()))
if isinstance(data, tuple):
return map(convert, data)
if isinstance(data, list):
return list(map(convert, data))
return data
def last_online_recent(date):
"""
Check if the last online date was recent.
:param date: date last online
:type date: string
:return: bool indicating whether the date was recent enough
:rtype: bool
"""
date_parsed = datetime.strptime(date, "%Y-%m-%dT%H:%M:%S.%fZ")
now = datetime.now()
sec_ago_date = (now - date_parsed).total_seconds()
# self.log.debug("Seconds ago date for {date} ^ {now}: {x}", date=date, now=str(now), x=sec_ago_date)
return sec_ago_date < 172800
def handle_exceptions(func):
"""
Wrapper helper to handle Agora API errors.
:param func: function to wrap
:rtype: func
:return: the wrapped function
"""
def inner_function(*args, **kwargs):
"""
Inner wrapper helper.
:rtype: any or bool
:return: False or the normal return
"""
try:
rtrn = func(*args, **kwargs)
except (ReadTimeout, ReadError, RemoteProtocolError):
return False
if isinstance(rtrn, dict):
if "success" in rtrn:
if "message" in rtrn:
if not rtrn["success"] and rtrn["message"] == "API ERROR":
if "error_code" in rtrn["response"]["error"]:
code = rtrn["response"]["error"]["error_code"]
if not code == 136:
log.error(f"API error: {code}")
return False
else:
log.error(f"API error: {rtrn['response']['error']}")
return False
return rtrn
return inner_function