pluto/handler/util.py

# Other library imports
import logging
from datetime import datetime, timezone

from httpx import ReadError, ReadTimeout, RemoteProtocolError

# Project imports
from settings import settings

log = logging.getLogger("util")

debug = False

# Color definitions
BLACK, RED, GREEN, YELLOW, BLUE, MAGENTA, CYAN, WHITE = range(8)
COLORS = {
    "WARNING": YELLOW,
    "INFO": WHITE,
    "DEBUG": BLUE,
    "CRITICAL": YELLOW,
    "ERROR": RED,
}
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


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)18s $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)
    if debug:
        log.setLevel(logging.DEBUG)

    # Add the handler and stop it being silly and printing everything twice
    log.addHandler(ch)
    log.propagate = False
    return log


class Base(object):
    def __init__(self):
        name = self.__class__.__name__

        # Set up all the logging stuff
        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
    """
    if "+" in date:
        # for LBTC
        # 2022-04-16T08:53:58+00:00
        date_split = date.split("+")
        date_split[1].replace(".", "")
        date_split[1].replace(":", "")
        date = "+".join(date_split)
        date_string = "%Y-%m-%dT%H:%M:%S%z"
        now = datetime.now(timezone.utc)
    else:
        date_string = "%Y-%m-%dT%H:%M:%S.%fZ"
        now = datetime.now()
    date_parsed = datetime.strptime(date, date_string)
    sec_ago_date = (now - date_parsed).total_seconds()
    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


def get_settings(platform):
    if platform == "agora":
        return settings.Agora
    elif platform == "lbtc":
        return settings.LocalBitcoins