1.购买服务器阿里云:服务器购买地址https://t.aliyun.com/U/PfsP97若失效,可用地址
阿里云:
服务器购买地址
https://t.aliyun.com/U/PfsP97若失效,可用地址
https://www.aliyun.com/activity/wuying/dj?source=5176.29345612&userCode=49hts92d腾讯云:
https://curl.qcloud.com/wJpWmSfU若失效,可用地址
https://cloud.tencent.com/act/cps/redirect?redirect=2446&cps_key=ad201ee2ef3b771157f72ee5464b1fea&from=console华为云
https://activity.huaweicloud.com/cps.html?fromacct=64b5cf7cc11b4840bb4ed2ea0b2f4468&utm_source=V1g3MDY4NTY=&utm_medium=cps&utm_campaign=2019052.部署教程
3.代码如下
from copy import deepcopyfrom Cryptodome.Cipher import AESimport timeimport ddddocrimport threadingimport requestsimport base64import jsonimport sysimport urllib3from urllib3.exceptions import InsecureRequestWarningif len(sys.argv) < 4:print('Usage: %s username password batchId <loop>' % sys.argv[0])exit(-1)DEBUG_REQUEST_COUNT = 0urllib3.disable_warnings(InsecureRequestWarning)WorkThreadCount = 8ocr = ddddocr.DdddOcr()def pkcs7padding(data, block_size=16):if type(data) != bytearray and type(data) != bytes:raise TypeError("仅支持 bytearray/bytes 类型!")pl = block_size - (len(data) % block_size)return data + bytearray([pl for i in range(pl)])class iCourses:mutex = threading.Lock()def __init__(self):self.aeskey = ''self.loginname = ''self.password = ''self.captcha = ''self.uuid = ''self.token = ''self.batchId = ''self.s = requests.session()self.is_login = Falseself.favorite = Noneself.select = Noneself.batchlist = Noneself.current = Noneself.error_code = 0self.try_if_capacity_full = Truedef safe_request(self, method, url, **kwargs):"""永不放弃的请求包装器,持续重试直到成功"""while True:try:if method.lower() == 'get':response = self.s.get(url, timeout=10, **kwargs)else:response = self.s.post(url, timeout=10, **kwargs)return responseexcept Exception as e:print(f"请求错误: {str(e)}, 正在重试...")time.sleep(0.5)continuedef login(self, username, password):"""无限重试的登录函数"""while True:try:index = 'https://icourses.jlu.edu.cn/'headers = {'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9','Host': 'icourses.jlu.edu.cn','User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/103.0.5060.114 Safari/537.36 Edg/103.0.1264.62',}html = self.safe_request('get', index, headers=headers, verify=False).textstart = html.find('"', html.find('loginVue.loginForm.aesKey')) + 1end = html.find('"', start)self.aeskey = html[start:end].encode('utf-8')self.loginname = usernameself.password = base64.b64encode(AES.new(self.aeskey, AES.MODE_ECB).encrypt(pkcs7padding(password.encode('utf-8'))))get_url = 'https://icourses.jlu.edu.cn/xsxk/auth/captcha'headers = {'Host': 'icourses.jlu.edu.cn','Origin': 'https://icourses.jlu.edu.cn','Referer': 'https://icourses.jlu.edu.cn/xsxk/profile/index.html','Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9','User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/103.0.5060.114 Safari/537.36 Edg/103.0.1264.62',}try:data = json.loads(self.safe_request('post', get_url, headers=headers, verify=False).text)except:print("验证码获取失败,重试中...")time.sleep(0.5)continueself.uuid = data['data']['uuid']captcha = data['data']['captcha']self.captcha = ocr.classification(base64.b64decode(captcha[captcha.find(',') + 1:]))login_url = 'https://icourses.jlu.edu.cn/xsxk/auth/login'payload = {'loginname': self.loginname,'password': self.password.decode('utf-8'),'captcha': self.captcha,'uuid': self.uuid}response = self.safe_request('post', login_url, headers=headers, data=payload, verify=False)response = json.loads(response.text)if response['code'] == 200 and response['msg'] == '登录成功':self.token = response['data']['token']s = ''s += 'login success!\n's += '=' * 0x40 + '\n's += '\t\t@XH: %s' % response['data']['student']['XH'] + '\n's += '\t\t@XM: %s' % response['data']['student']['XM'] + '\n's += '\t\t@ZYMC: %s' % response['data']['student']['ZYMC'] + '\n's += '=' * 0x40 + '\n'for c in response['data']['student']['electiveBatchList']:s += '\t\t@name: %s' % c['name'] + '\n's += '\t\t@BeginTime: %s' % c['beginTime'] + '\n's += '\t\t@EndTime: %s' % c['endTime'] + '\n's += '=' * 0x40 + '\n'print(s)self.batchlist = response['data']['student']['electiveBatchList']self.is_login = Truereturn Trueelse:print('login failed: %s' % response['msg'])time.sleep(0.5)continueexcept Exception as e:print(f"登录过程出错: {str(e)}, 重试中...")time.sleep(0.5)continuedef setbatchId(self, idx):while True:try:url = 'https://icourses.jlu.edu.cn/xsxk/elective/user'headers = {'Host': 'icourses.jlu.edu.cn','Origin': 'https://icourses.jlu.edu.cn','Referer': 'https://icourses.jlu.edu.cn/xsxk/profile/index.html','Authorization': self.token,'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/103.0.5060.114 Safari/537.36 Edg/103.0.1264.62',}try:self.batchId = self.batchlist[idx]['code']except:print('No such batch Id')returnpayload = {'batchId': self.batchId}response = json.loads(self.safe_request('post', url, headers=headers, data=payload, verify=False).text)if response['code'] != 200:print("set batchid failed")time.sleep(0.5)continuec = self.batchlist[idx]print('Selected BatchId:')s = ''s += '=' * 0x40 + '\n's += '\t\t@name: %s' % c['name'] + '\n's += '\t\t@BeginTime: %s' % c['beginTime'] + '\n's += '\t\t@EndTime: %s' % c['endTime'] + '\n's += '=' * 0x40 + '\n'print(s)url = 'https://icourses.jlu.edu.cn/xsxk/elective/grablessons?batchId=' + self.batchIdheaders = {'Host': 'icourses.jlu.edu.cn','Origin': 'https://icourses.jlu.edu.cn','Referer': 'https://icourses.jlu.edu.cn/xsxk/profile/index.html','Authorization': self.token,'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/103.0.5060.114 Safari/537.36 Edg/103.0.1264.62',}self.safe_request('get', url, headers=headers, verify=False)returnexcept Exception as e:print(f"设置批次ID时出错: {str(e)}, 重试中...")time.sleep(0.5)continuedef get_select(self):while True:try:post_url = 'https://icourses.jlu.edu.cn/xsxk/elective/select'headers = {'Host': 'icourses.jlu.edu.cn','Origin': 'https://icourses.jlu.edu.cn','Referer': 'https://icourses.jlu.edu.cn/xsxk/elective/grablessons?batchId=' + self.batchId,'Authorization': self.token,'batchId': self.batchId,'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/103.0.5060.114 Safari/537.36 Edg/103.0.1264.62',}response = json.loads(self.safe_request('post', post_url, headers=headers, verify=False).text)if response['code'] == 200:self.select = response['data']returnelse:print('get_select failed: %s' % response['msg'])time.sleep(0.5)continueexcept Exception as e:print(f"获取已选课程时出错: {str(e)}, 重试中...")time.sleep(0.5)continuedef get_favorite(self):while True:try:post_url = 'https://icourses.jlu.edu.cn/xsxk/sc/clazz/list'headers = {'Host': 'icourses.jlu.edu.cn','Origin': 'https://icourses.jlu.edu.cn','Referer': 'https://icourses.jlu.edu.cn/xsxk/elective/grablessons?batchId=' + self.batchId,'Authorization': self.token,'batchId': self.batchId,'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/103.0.5060.114 Safari/537.36 Edg/103.0.1264.62',}response = json.loads(self.safe_request('post', post_url, headers=headers, verify=False).text)if response['code'] == 200:self.favorite = response['data']returnelse:print('get_favorite failed: %s' % response['msg'])time.sleep(0.5)continueexcept Exception as e:print(f"获取收藏课程时出错: {str(e)}, 重试中...")time.sleep(0.5)continuedef select_favorite(self, ClassType, ClassId, SecretVal):while True:try:post_url = 'https://icourses.jlu.edu.cn/xsxk/sc/clazz/addxk'headers = {'Host': 'icourses.jlu.edu.cn','Origin': 'https://icourses.jlu.edu.cn','Referer': 'https://icourses.jlu.edu.cn/xsxk/elective/grablessons?batchId=' + self.batchId,'Authorization': self.token,'batchId': self.batchId,'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/103.0.5060.114 Safari/537.36 Edg/103.0.1264.62',}payload = {'clazzType': ClassType,'clazzId': ClassId,'secretVal': SecretVal}response = json.loads(self.safe_request('post', post_url, headers=headers, data=payload, verify=False).text)return responseexcept Exception as e:print(f"选课过程出错: {str(e)}, 重试中...")time.sleep(0.5)continuedef workThread(self, clazzType, clazzId, SecretVal, Name):tmp = deepcopy(self)global DEBUG_REQUEST_COUNTwhile True:try:response = tmp.select_favorite(clazzType, clazzId, SecretVal)code = response['code']msg = response['msg']self.mutex.acquire()try:DEBUG_REQUEST_COUNT += 1if self.current.get(clazzId) == 'doing':if code == 200:print('select [%s] success' % Name)self.current[clazzId] = 'done'self.mutex.release()breakelif code == 500:if msg == '该课程已在选课结果中':print('[%s] %s' % (Name, msg))self.current[clazzId] = 'done'self.mutex.release()breakif msg == '本轮次选课暂未开始':print('[%s]本轮次选课暂未开始' % (Name))self.mutex.release()continueif msg == '课容量已满':print(Name + "课容量已满")self.mutex.release()if self.try_if_capacity_full:continuebreakprint('[%s] %s' % (Name, msg))self.mutex.release()continueelif code == 401:print(msg)self.error_code = 401self.mutex.release()continueelse:print('[%d]: failed, try again' % code)self.mutex.release()continueelse:self.mutex.release()breakexcept:if self.mutex.locked():self.mutex.release()raiseexcept Exception as e:print(f"工作线程出错: {str(e)}, 重试中...")if self.mutex.locked():self.mutex.release()time.sleep(0.5)continuedef PrintSelect(self):print("=" * 20 + 'My Select' + '=' * 20)if self.select != None:for item in self.select:print('Teacher: %-10sName: %-20s Id: %-30s' % (item['SKJS'], item['KCM'], item['JXBID']))def PrintFavorite(self):print("=" * 20 + 'Favorite' + '=' * 20)if self.favorite != None:for item in self.favorite:print('Teacher: %-10sName: %-20s Id: %-30sclazzType: %-10s' % (item['SKJS'], item['KCM'], item['JXBID'], item['teachingClassType']))print('=' * 48)def SelectMyFavorite(self):if self.favorite != None:for item in self.favorite:self.select_favorite(item['teachingClassType'], item['JXBID'], item['secretVal'])def FuckMyFavorite(self):while True:try:self.get_favorite()thread = {}if None != self.favorite:self.current = {}for item in self.favorite:key = item['JXBID']thread[key] = []self.mutex.acquire()self.current[key] = 'doing'self.mutex.release()args = (item['teachingClassType'], item['JXBID'], item['secretVal'], item['KCM'])for i in range(WorkThreadCount):thread[key].append(threading.Thread(target=self.workThread, args=args))thread[key][-1].start()for key in thread:for t in thread[key]:t.join()print('本轮抢课结束,继续检查...')returnexcept Exception as e:print(f"抢课过程出错: {str(e)}, 重试中...")time.sleep(0.5)continueif __name__ == '__main__':while True:try:a = iCourses()# 无限重试登录while not a.is_login:if a.login(sys.argv[1], sys.argv[2]):breakprint('登录失败,重试中...')time.sleep(0.5)a.setbatchId(int(sys.argv[3]))a.get_favorite()a.PrintFavorite()a.FuckMyFavorite()if a.error_code == 401:print('会话过期,准备重新登录...')time.sleep(0.5)continuea.get_select()a.PrintSelect()print('DEBUG_REQUEST_COUNT: %d\n' % DEBUG_REQUEST_COUNT)if len(sys.argv) == 4:breaktime.sleep(0.5)except Exception as e:print(f"主循环出错: {str(e)}, 重试中...")time.sleep(0.5)continue
解析
该脚本为吉大自动抢课选课脚本,主要作用包括:
自动登录 icourses.jlu.edu.cn(抓取页面 AES key,对密码做 AES-ECB + PKCS#7 再 Base64,验证码用 ddddocr 识别)。
设置批次并进入抢课页(按命令行指定的
batchId)。读取"收藏课程"列表,对每门收藏的课并发抢课(多线程轮询提交接口,直到成功/满额/结束)。
可循环运行,会话 401 过期自动重登;网络/请求异常无限重试以提高成功率。
命令行用法:
python script.py username password batchId <loop>(传第4参则循环抢课)。
全局/关键点
WorkThreadCount = 8:每门课开 8 个线程并发提交。try_if_capacity_full = True:满额也继续尝试(有名额释放时可抢到)。统一
requests.session();safe_request封装永不放弃重试。DEBUG_REQUEST_COUNT统计请求次数;mutex保护共享状态。
类与方法职责
类 iCourses
封装登录、数据获取与并发抢课的全部流程。
初始化与工具
__init__: 初始化会话、鉴权/批次/状态字段。safe_request(method, url, **kwargs): 请求重试包装器,异常时 0.5s 间隔一直重试直到成功。pkcs7padding(data, block_size=16)(模块级): AES 填充。
登录与批次
login(username, password):访问首页,解析页面中的
loginVue.loginForm.aesKey作为 AES 密钥;密码 AES-ECB 加密 + Base64;
获取验证码
/auth/captcha,用 ddddocr 识别;POST
/auth/login,成功后保存token与electiveBatchList;失败/异常无限重试。setbatchId(idx):以索引选中
electiveBatchList[idx]的code为batchId;POST
/elective/user绑定批次;访问
/elective/grablessons?batchId=...进入抢课页;失败重试。
查询数据
get_select(): POST/elective/select获取当前已选课程列表到self.select(失败重试)。get_favorite(): POST/sc/clazz/list获取收藏课程列表到self.favorite(失败重试)。
抢课动作
select_favorite(ClassType, ClassId, SecretVal):POST
/sc/clazz/addxk发起选课请求,返回接口 JSON。workThread(clazzType, clazzId, SecretVal, Name):200:成功→标记done、停止该课所有线程;500:常见消息处理401:会话过期,记错误码等待主循环重登;"该课程已在选课结果中"→标记
done;"本轮次选课暂未开始"→继续循环;
"课容量已满"→根据
try_if_capacity_full决定继续或放弃;其他→继续尝试;
单门课的一个工作线程:循环调用
select_favorite;根据返回:
线程间通过
mutex协调self.current[clazzId]状态(doing/done)。FuckMyFavorite()(并发调度器,命名较"放飞"):将该课状态置为
doing;创建
WorkThreadCount个线程并发执行workThread;取收藏列表;对每门课:
等待该课所有线程结束;打印"本轮抢课结束,继续检查…"。
打印/辅助
PrintSelect(): 打印已选课程(教师/课程名/教学班 ID)。PrintFavorite(): 打印收藏课程(教师/课程名/教学班 ID/课程类型)。SelectMyFavorite(): 逐门收藏直接调用select_favorite(无并发)。
主循环 if __name__ == '__main__':
实例化
iCourses,无限重试登录;setbatchId(sys.argv[3])→get_favorite()→PrintFavorite()→FuckMyFavorite()(并发抢收藏);若
error_code == 401,打印并重登;get_select()→PrintSelect()→ 打印DEBUG_REQUEST_COUNT;如只传 3 参(无
<loop>),执行一次后退出;否则 0.5s 休眠继续循环。
注意:
本文部分变量已做脱敏处理,仅用于测试和学习研究,禁止用于商业用途,不能保证其合法性,准确性,完整性和有效性,请根据情况自行判断。技术层面需要提供帮助,可以通过打赏的方式进行探讨。
没有评论:
发表评论