flt/urp-auto
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add interval and fix bugs

Browse Source
This commit is contained in:
flt6
2025-07-06 14:44:06 +08:00
parent 38a9240149
commit 21f2f2dbec
4 changed files with 106 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
.env_demo
View File

@ -3,4 +3,6 @@ password=123456
recap_username=this_is_recap_unam recap_username=this_is_recap_unam
recap_password=123456 recap_password=123456
SC_KEY=this_is_sc_key SC_KEY=this_is_sc_key
FILE = "lessons.xlsx" FILE = "lessons.xlsx"
INTERVAL_1=2
INTERVAL_2=10

6
README.md
View File

@ -50,6 +50,12 @@ SC_KEY=你的ServerChan密钥
# 教务系统地址（可选，默认为辽宁大学） # 教务系统地址（可选，默认为辽宁大学）
base=http://jwstudent.lnu.edu.cn base=http://jwstudent.lnu.edu.cn
# 查询间隔
# 每次选课请求间隔
INTERVAL_1=2
# 完成一轮选课后的间隔
INTERVAL_2=10
``` ```
你可以查看`.env_demo`查看样例。 你可以查看`.env_demo`查看样例。

5
demo.csv
View File

@ -1,4 +1,3 @@
课程号, 课序号,课程名 课程号, 课序号,课程名
2310031,01,体育（三） 2310031,01,体育（二）
2310031,02,体育（三） 2310031,02,体育（二）
1221143,01,有机化学A(二)

1 课程号 课序号 课程名
2 2310031 01 体育（三） 体育（二）
3 2310031 02 体育（三） 体育（二）
1221143 01 有机化学A(二)

143
main.py
View File

@ -31,7 +31,9 @@ logger.addHandler(file) # 日志输出到文件
class LessonsException(Exception): class LessonsException(Exception):
"""自定义异常类""" """自定义异常类"""
pass
class ReloginException(LessonsException):
"""用于处理需要重新登录的异常"""
pass pass
@ -69,6 +71,8 @@ class Lessons:
raise LessonsException(f"请在环境变量中设置{key}") raise LessonsException(f"请在环境变量中设置{key}")
self.base = environ.get("base", "http://jwstudent.lnu.edu.cn") self.base = environ.get("base", "http://jwstudent.lnu.edu.cn")
self.interval_1 = int(environ.get("INTERVAL_1", 2)) # 请求间隔，默认为2秒
self.interval_2 = int(environ.get("INTERVAL_2", 10)) # 请求间隔，默认为10秒
def _retry_request( def _retry_request(
self, func, max_retries: int = 10, error_msg: str = "请求失败" self, func, max_retries: int = 10, error_msg: str = "请求失败"
@ -270,20 +274,42 @@ class Lessons:
"jc": 0, "jc": 0,
} }
response = self._retry_request(lambda: self.session.post(url, data=params)) response = self._retry_request(lambda: self.session.post(url, data=params))
data = response.json()["kylMap"] with open("response.json", "w", encoding="utf-8") as f:
if len(data) == 0: f.write(response.text)
logger.error(f"课程 {cl[2]} 的余量信息为空: {data}") data:dict = response.json()
return
cls:list[dict] = data.get("rwfalist", [])
if not cls:
logger.error(f"课程 {cl[2]} 的课程信息为空: {data}")
return None
for item in cls:
if item["classNum"] in cl[1]:
print(item["classNum"],type(item["classNum"]))
if item["kcm"] != cl[2]:
logger.critical(
f"课程 {cl[2]} 的课程名与查询信息不匹配: {item['kcm']} != {cl[2]}"
)
sc_send(
"选课异常",
desp=f"课程 {cl[2]} 的课程名与查询信息不匹配: {item['kcm']} != {cl[2]}",
)
return None
kyl:dict[str,str] = data["kylMap"]
if len(kyl) == 0:
logger.error(f"课程 {cl[2]} 的余量信息为空: {kyl}")
return
ret = [] ret = []
for kxh in cl[1]: for kxh in cl[1]:
key = f"{self.term}_{cl[0]}_{kxh}" key = f"{self.term}_{cl[0]}_{kxh}"
left = data.get(key, None) left = kyl.get(key, None)
if left is None: if left is None:
logger.error( logger.error(
f"课程 {cl[2]} 的余量信息不存在: {key} not in {data.keys()}" f"课程 {cl[2]} 的余量信息不存在: {key} not in {kyl.keys()}"
) )
ret.append(-1) ret.append(-1)
continue
ret.append(int(left)) ret.append(int(left))
return ret return ret
@ -389,6 +415,10 @@ class Lessons:
if classes.get(id) is None: if classes.get(id) is None:
classes[id] = (id, [kxh], name) classes[id] = (id, [kxh], name)
else: else:
if classes[id][2] != name:
raise LessonsException(
f"课程 {name}_{kxh} 的名称不一致: {classes[id][2]} != {name}"
)
classes[id][1].append(kxh) classes[id][1].append(kxh)
logger.info(f"读取课程信息，共有 {len(classes)} 门课程") logger.info(f"读取课程信息，共有 {len(classes)} 门课程")
@ -406,53 +436,70 @@ class Lessons:
sc_send("选课异常", desp="最近5次获取课程余量异常，尝试重新登录") sc_send("选课异常", desp="最近5次获取课程余量异常，尝试重新登录")
self.login() self.login()
master_err += 1 master_err += 1
errs.clear()
if master_err >= 3: if master_err >= 3:
logger.error("反复发生重要异常，退出程序") logger.error("反复发生重要异常，退出程序")
sc_send("选课异常", desp="反复发生重要异常，退出程序") sc_send("选课异常", desp="反复发生重要异常，退出程序")
return return
for lcl in classes.copy().values(): try:
logger.info(f"检查课程 {lcl[2]} 余量") for lcl in classes.copy().values():
try: logger.info(f"检查课程 {lcl[2]} 余量")
lefts = self.get_left(lcl) try:
if lefts is None: lefts = self.get_left(lcl)
errs.appendleft(time()) if lefts is None:
logger.error(f"获取课程 {lcl[2]} 余量时返回异常")
continue
except Exception as e:
errs.appendleft(time())
logger.error(f"获取课程 {lcl[2]} 余量时发生错误: {e}")
continue
for i, left in enumerate(lefts):
cl = (lcl[0], lcl[1][i], lcl[2])
if left > 0:
logger.info(
f"课程 {cl[2]}_{cl[1]} 有余量: {left}，开始选课"
)
try:
ret = self.select(cl)
if ret:
suc.append(cl)
classes.pop(cl[0])
break
except Exception as e:
errs.appendleft(time()) errs.appendleft(time())
logger.error(f"选课 {cl[2]}_{cl[1]} 时发生错误: {e}") logger.error(f"获取课程 {lcl[2]} 余量时返回异常")
finally: continue
sleep(2) # 避免请求过快导致服务器拒绝 except Exception as e:
elif left == -1: errs.appendleft(time())
logger.error(f"课程 {cl[2]}_{cl[1]} 余量信息异常") logger.error(f"获取课程 {lcl[2]} 余量时发生错误: {e}")
else: continue
logger.info(f"课程 {cl[2]}_{cl[1]} 无余量") for i, left in enumerate(lefts):
sleep(2) cl = (lcl[0], lcl[1][i], lcl[2])
logger.info( if left > 0:
f"当前还有{len(classes)}门课程未选上，分别为{','.join(mp[i] for i in classes.keys())}。等待10秒后继续检查" logger.info(
) f"课程 {cl[2]}_{cl[1]} 有余量: {left}，开始选课"
if suc: )
sc_send( try:
"选课成功", ret = self.select(cl)
desp=f"已成功选上课程: {', '.join(f'{i[2]}_{i[1]}' for i in suc)}", if ret:
suc.append(cl)
classes.pop(cl[0])
break
except Exception as e:
errs.appendleft(time())
logger.error(f"选课 {cl[2]}_{cl[1]} 时发生错误: {e}")
finally:
sleep(self.interval_1) # 避免请求过快导致服务器拒绝
elif left == -1:
logger.error(f"课程 {cl[2]}_{cl[1]} 余量信息异常")
else:
logger.info(f"课程 {cl[2]}_{cl[1]} 无余量")
sleep(self.interval_1)
logger.info(
f"当前还有{len(classes)}门课程未选上，分别为{','.join(mp[i] for i in classes.keys())}。等待{self.interval_2}秒后继续检查"
) )
sleep(10) # 等待10秒后继续检查 if suc:
sc_send(
"选课成功",
desp=f"已成功选上课程: {', '.join(f'{i[2]}_{i[1]}' for i in suc)}",
)
except ReloginException as e:
logger.error(f"需要重新登录: {e}")
sc_send("选课异常", desp=f"需要重新登录: {e}")
self.login()
continue
except LessonsException as e:
logger.error(f"选课过程中发生错误: {e}")
sc_send("选课异常", desp=f"选课过程中发生错误: {e}")
errs.appendleft(time())
continue
except Exception as e:
logger.error(f"意外错误: {e}")
sc_send("选课异常", desp=f"选课过程中发生意外错误: {e}")
errs.appendleft(time())
continue
sleep(self.interval_2) # 等待10秒后继续检查
logger.info("自动选课完成") logger.info("自动选课完成")