-
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathasynchronous.py
172 lines (141 loc) · 8.06 KB
/
asynchronous.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
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
import asyncio
from typing import Optional, List
from aiohttp import ClientSession
import UCube
import UCube.models
from UCube import UCubeClientAsync
from dotenv import load_dotenv
from os import getenv
"""
asynchronous.py
Asynchronous Example for UCube.
This file should be able to run if a token or a username and password is supplied.
If you would like to view more specific information,
then it would be suggested to look at the API Docs at https://ucube.readthedocs.io/en/latest/
or run this file in a debugger.
In order to get a token: https://ucube.readthedocs.io/en/latest/api.html#get-account-token
In order to set a username and password (RECOMMENDED), Rename .env.example to .env and set UCUBE_USERNAME to
the username and UCUBE_PASSWORD to the password.
In order to set a token (NOT RECOMMENDED - will expire quickly),
Rename .env.example to .env and set UCUBE_AUTH to the token.
Before running this file, confirm that you used `pip install -r requirements.txt` in the examples directory.
If you are running in a Synchronous environment, go take a look at the Synchronous examples file.
"""
def get_formatted_time(seconds):
"""Turn seconds into days, hours, minutes, and seconds.
Not related to UCube.
:param seconds: Amount of seconds to convert.
"""
seconds = round(seconds)
minute, sec = divmod(seconds, 60)
hour, minute = divmod(minute, 60)
day, hour = divmod(hour, 24)
return f"{f'{day}d ' if day else ''}" \
f"{f'{hour}h ' if hour else ''}" \
f"{f'{minute}m ' if minute else ''}" \
f"{f'{sec}s' if sec else ''}" \
f"{f'0s' if seconds < 1 else ''}"
class Example:
def __init__(self):
self.kwargs: dict = {
# Only pass in the authorization method you plan to use
# you can choose between passing in a username and password or by putting a token.
# If you put both, it will prioritize username & password login and create tokens from that.
'username': getenv("UCUBE_USERNAME"), # ucube username
'password': getenv("UCUBE_PASSWORD"), # ucube password
'token': getenv("UCUBE_AUTH"), # not suggested to pass in a token. This token will expire very quickly.
# verbose will not go to a logger, but just print messages for more info. Should usually set to False.
'verbose': True,
'web_session': None,
'hook': self.on_new_notifications # SET THIS TO YOUR OWN METHOD TO RECEIVE NEW NOTIFICATIONS
# It is not necessary to use a hook as you can manually create the loop if you want. However, as it's not
# recommended, an example won't be shown for that. If you're curious about doing so,
# you can look into the loop code for the hook and
# not pass in a hook. (UCube.ucubeasync.UCubeClientAsync._start_loop_for_hook)
}
self.ucube_client: Optional[UCubeClientAsync] = None
async def start(self):
if not self.ucube_client:
my_web_session = ClientSession()
self.kwargs["web_session"] = my_web_session
self.ucube_client = UCubeClientAsync(**self.kwargs)
# Create settings for how the history is created.
# This will create cache for the following:
# This should be customized to YOUR Client.
# These settings are just in preference to load up the cache quickly since this example
# does not need past media/posts/... stored.
start_kwargs = {
"load_boards": True, # loads up all club boards.
"load_posts": False, # wouldn't work without load_boards set to True
"load_notices": False, # wouldn't work without load_posts set to True
"load_media": False, # wouldn't work without load_posts set to True
"load_from_artist": True, # wouldn't work without load_posts set to True
"load_to_artist": False, # wouldn't work without load_posts set to True
"load_talk": False, # wouldn't work without load_posts set to True
"load_comments": False, # wouldn't work without load_posts set to True
"follow_all_clubs": False # will follow all clubs.
}
# login, start loading previous cache (if specified) and set the hook.
try:
await self.ucube_client.start(**start_kwargs)
except UCube.InvalidToken:
print("An Invalid Token was supplied to UCube.")
... # Should consider using credentials instead, tokens expire very quickly.
except UCube.InvalidCredentials:
print("An Invalid Username and Password were supplied to UCube.")
... # update the username and password
except UCube.BeingRateLimited:
print("UCube was rate-limited at some point. This should usually not occur.")
... # this would be considered an issue with UCube, make an issue on the github repository if this occurs.
except UCube.LoginFailed:
print("Failed to log into UCube Client.")
# You could attempt to reboot, perhaps their server was down or another status error occurred.
# This issue may occur in a method past the start method. If that is the case, and the cache is already
# loaded, you can attempt to login and see if that works. This would be considered a manual boot up without
# resetting the cache of the current client. (Calling the start method again will reset the current cache).
if self.ucube_client.cache_loaded:
# will occur in a task and will not raise an exception here. Exception would be
# raised in _wait_for_login
await self.ucube_client._try_login()
# wait for login or an exception to occur.
# you may want to try-except _wait_for_login(), but this example will not contain it for readability
await self.ucube_client._wait_for_login()
# if we get to this line of code, that means the login was successful.
# we can now start the hook loop again
await self.ucube_client._start_loop_for_hook()
# the code should now remain here unless another exception occurs
...
except asyncio.exceptions.TimeoutError:
print("It took too much time to log in.")
...
except Exception as e:
print(f"{e} - UCube Exception")
# if a hook is passed in,
# the start method will not reach this point (unless an exception occurs) and will check for new notifications.
async def on_new_notifications(self, notifications: List[UCube.models.Notification]):
"""
Decides what to do with a list of Notifications.
This is a hook method. Whenever there is a notification, this method will be called.
"""
for notification in notifications:
post = self.ucube_client.get_post(notification.post_slug)
if not post:
# you can attempt to fetch the post with:
# await self.ucube_client.fetch_post(notification.post_slug)
...
continue # skipping the notification - not recommended since you're keeping track of new notifications.
club = self.ucube_client.get_club(notification.club_slug) # the club the notification belongs to
# now that you have the post, you can go ahead and do whatever you want with the object.
# make sure to READ THE DOCS for specific objects. Search up the Post object if you're interested
# in things you may be able to do.
# There is a UCube discord bot (below) if you want to see a real usage.
# https://github.com/MujyKun/united-cube-bot
breakpoint()
if __name__ == '__main__':
print("=================================")
print("||Starting Asynchronous Example||")
print("=================================")
load_dotenv() # load the .env vars -- Important.
example_object = Example()
loop = asyncio.get_event_loop()
loop.run_until_complete(example_object.start())