You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
111 lines
3.8 KiB
Python
111 lines
3.8 KiB
Python
# Twisted/Klein imports
|
|
from twisted.logger import Logger
|
|
from twisted.internet import reactor
|
|
from twisted.internet.task import LoopingCall, deferLater
|
|
|
|
# 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 setup_call_loops(token_setting, function_init, function_continuous, delay, function_post_start=None):
|
|
"""
|
|
Setup the loops for dealing with access, refresh and auth tokens for various providers.
|
|
:param token_setting: the setting for whether to do the initial authentication
|
|
:param function_init: the initial authentication function
|
|
:param function_continuous: the ongoing authentication function (refresh_token -> access_token)
|
|
:param delay: time in seconds to wait between calls to function_continuous
|
|
:param function_post_start: an optional function to run after the access token is obtained
|
|
"""
|
|
if token_setting == "1":
|
|
deferLater(reactor, 1, function_init)
|
|
else:
|
|
deferLater(reactor, 1, function_continuous, True)
|
|
if function_post_start:
|
|
deferLater(reactor, 4, function_post_start)
|
|
|
|
lc = LoopingCall(function_continuous)
|
|
lc.start(delay)
|
|
|
|
|
|
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
|