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=201905
2.部署教程
3.代码如下
from copy import deepcopy
from Cryptodome.Cipher import AES
import time
import ddddocr
import threading
import requests
import base64
import json
import sys
import urllib3
from urllib3.exceptions import InsecureRequestWarning
if len(sys.argv) < 4:
print('Usage: %s username password batchId <loop>' % sys.argv[0])
exit(-1)
DEBUG_REQUEST_COUNT = 0
urllib3.disable_warnings(InsecureRequestWarning)
WorkThreadCount = 8
ocr = 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 = False
self.favorite = None
self.select = None
self.batchlist = None
self.current = None
self.error_code = 0
self.try_if_capacity_full = True
def 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 response
except Exception as e:
print(f"请求错误: {str(e)}, 正在重试...")
time.sleep(0.5)
continue
def 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).text
start = html.find('"', html.find('loginVue.loginForm.aesKey')) + 1
end = html.find('"', start)
self.aeskey = html[start:end].encode('utf-8')
self.loginname = username
self.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)
continue
self.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 = True
return True
else:
print('login failed: %s' % response['msg'])
time.sleep(0.5)
continue
except Exception as e:
print(f"登录过程出错: {str(e)}, 重试中...")
time.sleep(0.5)
continue
def 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')
return
payload = {
'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)
continue
c = 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.batchId
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',
}
self.safe_request('get', url, headers=headers, verify=False)
return
except Exception as e:
print(f"设置批次ID时出错: {str(e)}, 重试中...")
time.sleep(0.5)
continue
def 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']
return
else:
print('get_select failed: %s' % response['msg'])
time.sleep(0.5)
continue
except Exception as e:
print(f"获取已选课程时出错: {str(e)}, 重试中...")
time.sleep(0.5)
continue
def 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']
return
else:
print('get_favorite failed: %s' % response['msg'])
time.sleep(0.5)
continue
except Exception as e:
print(f"获取收藏课程时出错: {str(e)}, 重试中...")
time.sleep(0.5)
continue
def 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 response
except Exception as e:
print(f"选课过程出错: {str(e)}, 重试中...")
time.sleep(0.5)
continue
def workThread(self, clazzType, clazzId, SecretVal, Name):
tmp = deepcopy(self)
global DEBUG_REQUEST_COUNT
while True:
try:
response = tmp.select_favorite(clazzType, clazzId, SecretVal)
code = response['code']
msg = response['msg']
self.mutex.acquire()
try:
DEBUG_REQUEST_COUNT += 1
if self.current.get(clazzId) == 'doing':
if code == 200:
print('select [%s] success' % Name)
self.current[clazzId] = 'done'
self.mutex.release()
break
elif code == 500:
if msg == '该课程已在选课结果中':
print('[%s] %s' % (Name, msg))
self.current[clazzId] = 'done'
self.mutex.release()
break
if msg == '本轮次选课暂未开始':
print('[%s]本轮次选课暂未开始' % (Name))
self.mutex.release()
continue
if msg == '课容量已满':
print(Name + "课容量已满")
self.mutex.release()
if self.try_if_capacity_full:
continue
break
print('[%s] %s' % (Name, msg))
self.mutex.release()
continue
elif code == 401:
print(msg)
self.error_code = 401
self.mutex.release()
continue
else:
print('[%d]: failed, try again' % code)
self.mutex.release()
continue
else:
self.mutex.release()
break
except:
if self.mutex.locked():
self.mutex.release()
raise
except Exception as e:
print(f"工作线程出错: {str(e)}, 重试中...")
if self.mutex.locked():
self.mutex.release()
time.sleep(0.5)
continue
def 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('本轮抢课结束,继续检查...')
return
except Exception as e:
print(f"抢课过程出错: {str(e)}, 重试中...")
time.sleep(0.5)
continue
if __name__ == '__main__':
while True:
try:
a = iCourses()
# 无限重试登录
while not a.is_login:
if a.login(sys.argv[1], sys.argv[2]):
break
print('登录失败,重试中...')
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)
continue
a.get_select()
a.PrintSelect()
print('DEBUG_REQUEST_COUNT: %d\n' % DEBUG_REQUEST_COUNT)
if len(sys.argv) == 4:
break
time.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 休眠继续循环。
注意:
本文部分变量已做脱敏处理,仅用于测试和学习研究,禁止用于商业用途,不能保证其合法性,准确性,完整性和有效性,请根据情况自行判断。技术层面需要提供帮助,可以通过打赏的方式进行探讨。
没有评论:
发表评论