# 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): 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): def inner_function(*args, **kwargs): 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