/
elective_bot.py
337 lines (294 loc) · 11.5 KB
/
elective_bot.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
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
import requests
import random
import io
import time
import threading
import tkinter
from tkinter import ttk, messagebox
from bs4 import BeautifulSoup
from PIL import Image, ImageTk
import hashlib
import captcha
from logger import Logger,log_q
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
TIMEOUT_S=30
auth={
'username': None,
'password': None,
'channel': None,
}
class ElectiveBot:
def __init__(self,name):
adapter=requests.adapters.HTTPAdapter(pool_connections=1,pool_maxsize=1,pool_block=True)
self.s=requests.Session()
self.s.mount('http://elective.pku.edu.cn',adapter)
self.s.mount('https://elective.pku.edu.cn',adapter)
self.s.verify=False
self.s.trust_env=True
self.s.headers={
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/98.0.4758.102 Safari/537.36',
'Referer': 'https://elective.pku.edu.cn/elective2008/edu/pku/stu/elective/controller/help/HelpController.jpf',
'Cache-Control': 'max-age=0',
}
self._status='init'
self.name=name
self.log=Logger(name)
self.last_loop_time=time.time()
@property
def status(self):
return self._status
@status.setter
def status(self,s):
self._status=s
log_q.put(None)
def auth(self):
assert auth['username'] and auth['password'], 'credential not set'
self.log('debug','before auth')
res=self.s.post(
'https://iaaa.pku.edu.cn/iaaa/oauthlogin.do',
data={
'appid': 'syllabus',
'userName': auth['username'],
'password': auth['password'],
'randCode': '',
'smsCode': '',
'otpCode': '',
'redirUrl': 'http://elective.pku.edu.cn:80/elective2008/agent4Iaaa.jsp/../ssoLogin.do'
},
cookies={
'userName': auth['username'],
},
)
res.raise_for_status()
json=res.json()
assert json['success'], json
token=json['token']
self.log('debug',f'get auth token {token}')
res=self.s.get(
'https://elective.pku.edu.cn/elective2008/ssoLogin.do',
params={
'rand': '%.10f'%random.random(),
'token': token,
},
)
res.raise_for_status()
if '<title>帮助-总体流程</title>' not in res.text:
if '/scnStAthVef.jsp/' in res.text:
if not auth['channel']:
raise RuntimeError('需要身份信息，参见 README')
else:
sida=res.text.partition('/ssoLogin.do?sida=')[2].partition('&')[0]
assert sida.isalnum(), 'invalid sida'
self.log('debug','choose channel %s for sida %s'%(auth['channel'],sida))
res=self.s.get(
'https://elective.pku.edu.cn/elective2008/ssoLogin.do',
params={
'sida': sida,
'sttp': auth['channel'],
},
)
res.raise_for_status()
if '<title>帮助-总体流程</title>' not in res.text:
raise RuntimeError('after login check failed')
else:
raise RuntimeError('after login check failed')
def proc_course_elem(self,rows):
for row in rows:
name=row.select('td:nth-of-type(1) span')[0].get_text()
classid=row.select('td:nth-of-type(6) span')[0].get_text()
teacher=row.select('td:nth-of-type(5) span')[0].get_text()
selectbtn=(row.select('a[href^="/elective2008/edu/pku/stu/elective/controller/supplement/electSupplement.do"]') or [None])[0]
volume_cnt,_,elected_cnt=row.select('td span[id^="electedNum"]')[0].get_text(strip=True).partition(' / ')
if '/' in elected_cnt:
elected_cnt=elected_cnt.partition('/')[0]
volume_cnt=int(volume_cnt)
elected_cnt=int(elected_cnt)
if elected_cnt==0: # buggy
elected_cnt=volume_cnt
self.log('warning',f'loop got buggy elected count for {name}')
if not selectbtn: # already chosen
continue
yield {
'name': name,
'classid': classid,
'teacher': teacher,
'selecturl': f'https://elective.pku.edu.cn{selectbtn.attrs["href"]}',
'volume_cnt': volume_cnt,
'elected_cnt': elected_cnt,
}
@staticmethod
def detect_err(soup):
fatal_err_elems=soup.select('td[background="/elective2008/resources/images/11-1.png"] td.black')
if fatal_err_elems:
return fatal_err_elems[0].get_text(strip=True)
@staticmethod
def detect_tips(soup):
tips_elems=soup.select('#msgTips td[width="100%"]')
if tips_elems:
return tips_elems[0].get_text(strip=True)
def loop_(self,url=None):
if url is None:
url = f'https://elective.pku.edu.cn/elective2008/edu/pku/stu/elective/controller/supplement/SupplyCancel.do?xh={auth["username"]}'
self.log('debug',f'loop get {url}')
res=self.s.get(
url,
params={'xh': auth['username']},
timeout=TIMEOUT_S,
)
res.raise_for_status()
soup=BeautifulSoup(res.text,'lxml')
fatal_err=self.detect_err(soup)
if fatal_err:
if '目前不是补退选时间' not in fatal_err and '目前是跨院系选课数据准备时间' not in fatal_err and '目前时段不能选课' not in fatal_err:
self.status='dead'
self.log('critical', f'loop fatal err {fatal_err}')
return
if soup.title.get_text()!='补选退选':
self.status='dead'
self.log('critical',f'loop title is {soup.title}')
return
courses=list(self.proc_course_elem(soup.select('tr.datagrid-all, tr.datagrid-odd, tr.datagrid-even')))
# fixme
next_link=soup.find('a',text='Next')
if not next_link:
return courses
else:
try:
next=self.loop_(f'https://elective.pku.edu.cn{next_link.attrs["href"]}')
return courses+next
except Exception as e:
self.log('warning', f'loop error {type(e)} {str(e)}')
return courses
def loop(self,callback_fatal):
self.status='loop'
self.last_loop_time=time.time()
try:
res=self.loop_() or []
except Exception as e:
self.log('warning',f'loop error {type(e)} {str(e)}')
return []
else:
if res:
self.log('info','loop ok')
return res
finally:
if self.status=='dead':
callback_fatal()
else:
self.status='idle'
def get_captcha(self):
try:
self.log('debug','get captcha')
res=self.s.get(
'https://elective.pku.edu.cn/elective2008/DrawServlet',
params={
'Rand': '%.10f'%(10000*random.random()),
},
timeout=TIMEOUT_S,
)
res.raise_for_status()
return Image.open(io.BytesIO(res.content))
except Exception as e:
self.log('warning',f'get captcha error {type(e)} {str(e)}')
return Image.open('error.gif')
def verify_captcha(self,captcha):
try:
self.log('debug',f'check captcha {captcha}')
res=self.s.post(
'https://elective.pku.edu.cn/elective2008/edu/pku/stu/elective/controller/supplement/validate.do',
data={
'validCode': captcha,
'xh': auth['username'],
},
timeout=TIMEOUT_S,
)
res.raise_for_status()
assert res.json()['valid']=='2', res.json()
return True
except Exception as e:
self.log('warning',f'bad captcha {type(e)} {str(e)}')
return False
def enter_captcha(self,tk,should_autoinput=False):
if should_autoinput:
self.log('info','doing auto captcha')
try:
img=self.get_captcha()
self.log('debug',f'downloaded captcha')
result=captcha.recognize(img)
self.log('debug',f'recognized captcha {result}')
assert self.verify_captcha(result), 'incorrect captcha'
except Exception as e:
raise RuntimeError(f'captcha recognize error {type(e)} {str(e)}')
else:
self.log('info','auto captcha verify ok')
self.status='idle'
return
else:
self.log('info','doing manual captcha')
tl=tkinter.Toplevel(tk)
try:
tl.wm_attributes('-toolwindow',True)
except Exception: # toolwindow only works on Windows
pass
tl.title(f'Captcha for {self.name}')
captcha_var=tkinter.StringVar(tl)
label=ttk.Label(tl)
label.pack()
def submit_captcha():
if self.verify_captcha(captcha_var.get()):
self.log('info','captcha verify ok')
tl.destroy()
self.status='idle'
else:
messagebox.showerror('Error', 'Incorrect captcha')
entry.focus_set()
def skip_captcha():
img=self.get_captcha()
#img.seek(15)
label._image=ImageTk.PhotoImage(img)
label['image']=label._image
tl.update_idletasks()
entry=ttk.Entry(tl,textvariable=captcha_var)
entry.bind('<Return>',lambda _:submit_captcha())
entry.pack()
ttk.Button(tl,text='Next Captcha',command=skip_captcha).pack()
entry.focus_set()
tl.after_idle(skip_captcha)
def select_(self,url):
self.log('debug',f'select {url}')
res=self.s.post(
url,
timeout=TIMEOUT_S,
)
res.raise_for_status()
soup=BeautifulSoup(res.text,'lxml')
fatal_err=self.detect_err(soup)
if fatal_err:
self.log('warning', f'select err {fatal_err}')
return False, fatal_err
tips=self.detect_tips(soup)
if tips:
if '成功，请查看已选上列表确认' in res.text:
self.log('info',f'select ok {tips}')
return True, tips
else:
self.log('warning',f'select fail {tips}')
return False, tips
else:
self.log('warning','select no tips')
return False, ''
def select(self,url,callback=lambda *_: None):
self.status='select'
def do_select():
try:
ok,reason=self.select_(url)
except Exception as e:
reason=f'select err {type(e)} {str(e)}'
self.log('warning',reason)
callback(False,reason)
else:
callback(ok,reason)
finally:
self.status='idle'
threading.Thread(target=do_select,daemon=True).start()