pluto/handler/util.py

# Twisted/Klein imports
from twisted.logger import Logger

# Other library imports
from httpx import ReadTimeout, ReadError, RemoteProtocolError
from datetime import datetime

log = Logger("util.global")


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("API error: {code}", code=code)
                                return False
                        else:
                            log.error("API error: {code}", code=rtrn["response"]["error"])
                            return False
        return rtrn

    return inner_function