-
Notifications
You must be signed in to change notification settings - Fork 1
/
wrapper.py
97 lines (85 loc) · 4.6 KB
/
wrapper.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
import aiohttp
class ApiWrapper:
def __init__(self, key):
self.key = key
self.base_url = "https://api.tatsumaki.xyz"
self.headers = {"Authorization": key}
self.background_url = "https://www.tatsumaki.xyz/images/backgrounds/profile/{}.png"
self.badge_url = "https://www.tatsumaki.xyz/images/badges/{}.png"
async def request(self, url):
# get request to the tatsu api
async with aiohttp.ClientSession() as session:
async with session.get(url=url, headers=self.headers) as result:
if result.status != 200:
print(result)
return False
try:
return await result.json()
except aiohttp.client_exceptions.ContentTypeError:
return await result.text()
async def put(self, url, data=None):
# put api calls
async with aiohttp.ClientSession() as session:
async with session.put(url=url, headers=self.headers, data=data) as result:
if result.status != 200:
print(result)
return False
return await result.json()
async def user(self, user_id):
"""Returns the available information about a user. Requires a user id"""
result = await self.request(url=self.base_url + "/users/{}".format(user_id))
if not result:
raise Exception("The API is currently having issues, please try again later")
background = result.get("background", None)
if background is not None:
result["background_url"] = self.background_url.format(background)
badges = result.get("badgeSlots", None)
if badges is not None:
result["badgeURLs"] = [self.badge_url.format(i) for i in badges]
return result
async def leaderboard(self, server_id, rank=10):
"""Returns 10 ranks, where the rank argument is the last rank. Requires a server id and rank limit (default
is the first 10 ranks)"""
if rank < 10:
rank = 10
result = await self.request(url=self.base_url + "/guilds/{0}/leaderboard".format(server_id))
if not result:
raise Exception("The API is currently having issues, please try again later")
return result[rank - 10:rank]
def update_key(self, key):
"""Update the API key. Requires a new API key"""
self.key = key
return True
async def update_user_score(self, guild, user, amount, action="set"):
"""Update a users server score. Requires a guild id, user id, amount and an action (add, set, remove)
default is set. You need the manage channels permission in this server to use this. The API endpoint for this
is currently not working."""
if amount > 50000 or amount < 0:
raise Exception("Make sure that the amount is between 0 and 50.000")
if not action in ["set", "add", "remove"]:
raise Exception("Make sure that the action is either 'set', 'add' or 'remove'")
data = {"amount": amount, "action": action}
result = await self.put(url=self.base_url + "/guilds/{0}/members/{1}/score".format(guild, user), data=data)
if not result:
raise Exception("The API is currently having issues, please try again later")
return True
async def update_user_points(self, guild, user, amount, action="set"):
"""Update a users server points. Requires a guild id, user id, amount and an action (add, set, remove)
default is set. You need the manage channels permission in this server to use this. The API endpoint for this
is currently not working."""
if amount > 50000 or amount < 0:
raise Exception("Make sure that the amount is between 0 and 50.000")
if not action in ["set", "add", "remove"]:
raise Exception("Make sure that the action is either 'set', 'add' or 'remove'")
data = {"amount": amount, "action": action}
result = await self.put(url=self.base_url + "/guilds/{0}/members/{1}/points".format(guild, user), data=data)
if not result:
raise Exception("The API is currently having issues, please try again later")
return True
async def get_user_stats(self, guild_id, user_id):
"""Get stats about a user. Requires a guild and user id."""
result = await self.request(url=self.base_url + "/guilds/{0}/members/{1}/stats".format(guild_id, user_id))
if not result:
raise Exception("The API is currently having issues, please try again later")
else:
return result