-
Notifications
You must be signed in to change notification settings - Fork 0
/
monitor.py
247 lines (195 loc) · 9.45 KB
/
monitor.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
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
import requests, time, re, random, keyring, logging, json
from tqdm import tqdm
from bs4 import BeautifulSoup
logging.basicConfig(level=logging.INFO)
user_agent = {'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.71 Safari/537.36 Edg/94.0.992.38'}
# initial config
with open('config.json', 'r') as cfg:
tmp = json.load(cfg)
if not tmp['webhook_url']:
webhook = input("Please enter your webhook url: ")
tmp['webhook_url'] = webhook
with open('config.json', 'w') as update:
json.dump(tmp, update)
logging.info("Webhook successfully set")
else:
webhook = tmp['webhook_url']
logging.info("Webhook retrieved")
isl = 'https://is.muni.cz/auth/'
exams_link = 'https://is.muni.cz/auth/student/prihl_na_zkousky'
block_link = 'https://is.muni.cz/auth/student/poznamkove_bloky_nahled'
min_sleep = int(input("Minimum sleep time (ex. 300): ")) # time in seconds, make sure to not get rate limited
max_sleep = int(input("Maximum sleep time (ex. 700): "))
def login(session):
logging.info("Logging in...")
user = keyring.get_password('is-mon', 'uco') # retrieve credentials from the python keyring library
password = keyring.get_password('is-mon', 'password')
while True:
try:
init = session.get(isl, allow_redirects=True)
login_post = session.post(init.url, data={"akce":"login", "credential_0":user, "credential_1":password, "uloz":"uloz"}, allow_redirects=True, timeout = 10)
break
except ConnectionError:
logging.info("Connection error, trying again in 10 seconds")
time.sleep(10)
if login_post.url == isl:
logging.info("Login successful")
else:
logging.error("Login failed, invalid credentials, reset your keyring")
exit(1)
return session
def get_notes(session): # get the notebook page, return the most recent change
init = session.get(block_link, timeout = 10)
soup = BeautifulSoup(init.text,'html.parser')
last_change = soup.find('a',{'id':'odkaz_na_posledni_akci'})
return last_change, session
def monitor_notebook(session):
while True:
try:
last_change, session = get_notes(session) # initial fetch of the notebook
break
except Exception as e:
sl_t = random.randint(min_sleep, max_sleep)
logging.error(f'Exception {e} occurred while setting up. Sleeping for {sl_t} seconds.')
for i in tqdm(range(100)):
time.sleep(sl_t/100)
logging.info("Started monitoring...")
while True:
try:
new_change, session = get_notes(session) # new fetch of the notebook
if new_change.text != last_change.text: # comparison between the two
change_req = session.get(block_link + new_change['href'])
soup = BeautifulSoup(change_req.text,'html.parser')
# get information about the change
row = soup.find('div',{'id':str(re.sub('#','', new_change['href']))})
title = row.find('div',{'class':'column small-12 medium-3 tucne ipb-nazev'}).text
desc = row.find('pre').text
title = title[8:len(title)-7]
new_split = new_change.text.split(',')
logging.info(f"Detected a change... Title: {title} Description: {desc}")
embed = {'embeds':[{'title': new_split[3],'color':7988011,'fields':[{'name':f'**{title}**','value':desc}],'footer':{'text': f'{new_split[0][16:]}, {new_split[1]}'}}]}
requests.post(webhook,json = embed) # post information about the change to the discord webhook
logging.info("Successfully posted to webhook")
last_change = new_change # update the change
sl_t = random.randint(min_sleep, max_sleep) # random sleep interval
logging.info(f'Sleeping for {sl_t} seconds.')
for i in tqdm(range(100)): # progress bar using tqdm
time.sleep(sl_t/100)
except Exception as e:
sl_t = random.randint(min_sleep, max_sleep)
logging.error(f'Exception {e} occurred. Sleeping for {sl_t} seconds.')
for i in tqdm(range(100)):
time.sleep(sl_t/100)
def exam_signup(session):
# fetch available subjects
logging.info('Fetching subject list...')
while True:
try:
exam_master = session.get(exams_link)
break
except Exception as e:
logging.error(f'Exception {e} occurred while setting up. Sleeping for 10 seconds.')
time.sleep(10)
soup = BeautifulSoup(exam_master.text, 'html.parser')
sub_dict = {}
print("\nAvailable subjects")
print("------------------")
i = 0
for subject in soup.find('main',{'id':'app_content'}).find('ul').find_all('li'):
# retrieve info about the subject
sub_code = subject.text.split(' ')[0]
sub_href = f"{exams_link}{subject.find('a')['href'][18:]}"
sub_dict[i] = sub_href
print(f"{i}: {sub_code}")
i += 1
chosen_sub = int(input('Please choose a subject code from the options above (number): '))
# after the subject is chosen, check for available exam dates
logging.info('Fetching exam dates...')
exam_entries_req = session.get(sub_dict[chosen_sub], timeout = 10)
soup = BeautifulSoup(exam_entries_req.text,'html.parser')
# edge case where the subject has no exams or is already completed
notif = soup.find('div',{'class':'zdurazneni info'})
if notif:
notif_text = notif.find('p').text
if notif_text.endswith('není v budoucnosti vypsán již žádný termín, nebo máte předmět již úspěšně ukončen.'):
logging.info('The subject has no exam dates or you have already completed it.')
exit(0)
exam_entries = {}
count = 0
print("\nAvailable dates")
print("------------------")
# go through all available exams
for entry in soup.find_all('tr',{'valign':'top'}):
# retrieve information about the exam
exam_status = entry.find_all('td')[0].text
exam_href = f"{exams_link}{entry.find_all('td')[2].find_all('font')[3].find('a')['href'][18:]}"
exam_date = entry.find_all('td')[2].find('b').text
capacity_status = entry.find_all('td')[2].text
max_cap = re.search(r'max. (\d+)', capacity_status)
if max_cap:
max_cap = max_cap[1]
else:
max_cap = ''
current_cap = re.search(r'přihlášeno (\d+)', capacity_status)[1]
exam_entries[str(count)] = {
'date': exam_date,
'status': exam_status,
'link': exam_href,
'max_capacity': max_cap,
'current_signedup': current_cap
}
print(f'{count}: {exam_date}, CAPACITY: {current_cap}/{max_cap}')
count += 1
# get input from the user about which exam date they want to register for
while True:
chosen_date = input(f'Please choose a date from the options above [0-{count}]: ').lower()
if int(chosen_date) > count or int(chosen_date) < 0:
logging.error('Invalid choice, please try again.')
else:
break
logging.info('Trying to sign up...')
exam_link = exam_entries[chosen_date]['link']
if 'burza' in exam_link:
logging.error("You are already signed up for this exam date")
exit(1)
while True:
# this is the main loop where we try to sign up for the exam repeatedly
signup_req = session.get(exam_link, timeout = 10)
soup = BeautifulSoup(signup_req.text, 'html.parser')
# the user has successfully signed up
success_status = soup.find('div', {'class': 'zdurazneni potvrzeni'})
if success_status:
logging.info('Successfully signed up for the exam.')
embed = {'embeds':[{'title': "Exam signup",'color':7988011,'fields':[{'name':f'**{exam_entries[chosen_date]["date"]}**','value':"Signed up!"}]}]}
requests.post(webhook, json = embed)
break
# the user doesn't meet the requirements to sign up
notification_status = soup.find('div', {'class': 'zdurazneni upozorneni'})
if notification_status:
logging.error('You are already signed up for a different exam, please unsubscribe and try again.')
break
# the exam is full
error_status = soup.find('div',{'class':'zdurazneni chyba'})
if error_status:
error_text = error_status.find('h3').text
if error_text == 'Na tento termín se nelze přihlásit. Kapacitní limit zkušebního termínu je již zaplněn.':
# retrying in case someone drops their registration
sl_t = random.randint(min_sleep, max_sleep)
logging.error(f'Exam capacity is full. Sleeping for {sl_t} seconds.')
for i in tqdm(range(100)):
time.sleep(sl_t/100)
print('1: Notebook monitoring')
print('2: Exam signup')
while True:
mode = int(input('Please enter your desired mode: '))
if mode < 1 or mode > 2:
logging.error('Invalid choice, try again.')
else:
break
session = requests.Session()
session.headers.update(user_agent)
session = login(session)
if mode == 1:
monitor_notebook(session)
elif mode == 2:
exam_signup(session)