项目介绍 这是一个面向学校选课系统的桌面端辅助工具,主要用于课程信息查询、待抢课程监控、自动选课尝试、冲突换课与失败回滚。项目采用纯 API 交互方式,不依赖浏览器自动化,核心目标是把选课过程从“手工反复刷新”改成“可持续、可恢复、可追踪”的监控流程。本文记录的是这个工具从可用到可长期运行的开发过程和关键技术取舍。
项目地址:https://github.com/YHalo-wyh/YNU-xk_spider-Pro
这个项目一开始并不复杂,目标就是把“手工刷课”变成“可持续监控”。真正做起来后,难点很快从接口调用转成了稳定性:监控能不能跑一晚上,Session 失效能不能自愈,冲突换课失败后能不能把旧课抢回来。后来几个版本几乎都在围绕这个方向做收敛。
如果把版本线拉开看,v1.2 是纯 API 架构成型,v1.4-v1.5 是稳定性修补期,重点处理长时间运行卡顿、线程退出和健康检查,v1.6-v1.8 是守护能力逐步独立,v1.9.0 才算把守护状态协议彻底改干净。很多看起来“只是优化体验”的改动,背后其实是为了减少不可控状态。
项目目前是典型的桌面端分层:ui.py 管交互和展示,workers.py 管核心业务链路,run_watchdog.py 管进程级守护,logger.py 管日志追踪,config.py 管课程类型和接口映射。这个分层不是一开始就有的,是边踩坑边拆出来的。早期把逻辑全堆在一个流程里时,问题是出了故障很难判断是 UI 卡了、线程挂了,还是请求超时了。
核心线程是 MultiGrabWorker。它不是简单循环请求,而是带了连接池、课程状态缓存、心跳、登录状态检测、自动重登互斥、健康检查和自动恢复。换句话说,抢课只是它的一部分职责,更多代码其实在处理异常分支。
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 class MultiGrabWorker (QThread ): """ 高并发非阻塞抢课 Worker 每门课程独立监控线程,互不阻塞 """ success = pyqtSignal(str , dict ) failed = pyqtSignal(str ) status = pyqtSignal(str ) need_relogin = pyqtSignal() heartbeat = pyqtSignal(int ) def __init__ (self, courses, student_code, batch_code, token, cookies, campus='02' , username='' , password='' , max_workers=5 , serverchan_key='' ): super ().__init__() self .student_code = student_code self .batch_code = batch_code self .token = token self .cookies = cookies self .campus = campus self .username = username self .password = password self .max_workers = max_workers self ._courses_mutex = QMutex() self ._courses = list (courses) self ._relogin_in_progress = False self ._relogin_mutex = QMutex() self .http_session = requests.Session() adapter = HTTPAdapter(pool_connections=20 , pool_maxsize=20 , max_retries=Retry(total=2 , backoff_factor=0.1 , status_forcelist=[500 , 502 , 503 , 504 ])) self .http_session.mount('https://' , adapter) self .http_session.mount('http://' , adapter)
监控主循环里有一条我后面一直坚持的原则:查询失败时不盲抢。以前有过“查询异常但仍发起选课”的做法,短期看像提高命中率,长期看风险太高,尤其是换课链路里容易引发误退。现在逻辑是查询失败直接跳过,状态日志写清楚,下一轮再试。
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 def _monitor_course_loop (self, course ): """ 核心安全策略: 1) 查询失败不盲抢 2) isFull 优先于余量计算 3) 仅在 isFull=False 且 remain>0 时行动 """ remain, capacity, course_info = self ._api_query_course_capacity(course) if remain is None : self .status.emit(f"[SKIP] {course_name} 查询失败,跳过本次循环" ) time.sleep(1.5 ) continue is_full_flag = course_info.get('isFull' , False ) if course_info else False if is_full_flag: if remain > 0 : self .status.emit(f"[GHOST] {course_name} 余量显示异常,跳过" ) time.sleep(1.0 ) continue if remain > 0 : success, msg, need_rollback = self ._api_select_course_fast(course)
选课和退课接口本身不复杂,复杂的是返回值分叉。一个“失败”到底是人数满、冲突、Session 过期、验证码链路问题,还是短暂网络抖动,处理策略完全不同。现在 volunteer.do 和 deleteVolunteer.do 都挂了同一套会话过期判断,检测到 302 或结果码异常会触发自动重登,再做一次重试,避免把一次偶发失效当成永久失败。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 def _api_select_course_fast (self, course, retry_on_expired=True ): resp = self .http_session.post(url, headers=self ._get_headers(), cookies=self ._parse_cookies(self .cookies), data=payload, timeout=(3 , 5 ), verify=False , allow_redirects=False ) if resp.status_code == 302 or self ._is_session_expired(response=resp): if retry_on_expired and self ._handle_session_expired(): return self ._api_select_course_fast(course, retry_on_expired=False ) return False , "session_expired" , False result = resp.json() code = result.get('code' , '' ) msg = result.get('msg' , '' ) if code == '1' : return True , "选课成功" , False elif '冲突' in msg: return False , f"时间冲突: {msg} " , True elif '容量' in msg or '已满' in msg: return False , "课程已满" , False return False , msg or "选课失败" , False
自动重登这块真正解决的问题不是“能不能重新登录”,而是“并发监控时怎么避免重登风暴”。如果十几个监控线程同时发现 token 失效,全部去打验证码接口,结果只会更糟。现在做法是互斥锁串行化重登,其他线程等待结果,成功后继续跑。
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 def _handle_session_expired (self ): if not self ._relogin_mutex.tryLock(): self .status.emit("[自动重登] 等待其他线程完成重登..." ) max_wait = 30 waited = 0 while waited < max_wait: time.sleep(0.5 ) waited += 0.5 if self ._relogin_mutex.tryLock(): self ._relogin_mutex.unlock() break return bool (self .token) try : self ._relogin_in_progress = True for _ in range (3 ): success, new_token, new_cookies = self ._do_relogin() if success: self .token = new_token self .cookies = new_cookies self .session_updated.emit(new_token, new_cookies) return True time.sleep(0.5 ) return False finally : self ._relogin_in_progress = False self ._relogin_mutex.unlock()
冲突换课是最敏感的一段。正常流程是定位冲突课、退旧、选新、核验,但线上最危险的情况是退旧成功而选新失败。这个项目后面加了“亡命回滚”:只要没确认抢回旧课就一直重试,直到成功或用户停止监控。这个策略看起来激进,但比“超时后放弃”更符合实际场景。
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 def _handle_conflict_rollback (self, course ): success, msg = self ._api_delete_course(conflict_tc_id, conflict_type) if not success: return False , conflict_course success, msg, _ = self ._api_select_course_fast(course) if success and self ._check_course_selected(tc_id): return True , conflict_course while self ._running: rollback_success, rollback_msg, _ = self ._api_select_course_fast({ 'JXBID' : conflict_tc_id, 'type' : conflict_type }) if rollback_success: is_selected = self ._check_course_selected(conflict_tc_id) if is_selected or is_selected is None : return False , conflict_course if rollback_msg and ('已选' in rollback_msg or '重复' in rollback_msg): return False , conflict_course time.sleep(0.7 )
UI 层在 v1.9.0 前后改动也不少。课程卡片的状态标签以前只覆盖单状态,现在支持“已满 + 冲突”同时展示,且顺序固定,避免用户误读。搜索也收敛成只匹配课程名和教师名,空结果会在列表和中间区域同时给出提示,避免“看起来没反应”的错觉。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 is_conflict = self .course_data.get('isConflict' , False ) is_full = self .course_data.get('isFull' , False ) if is_full: status_layout.addWidget(full_label) if is_conflict: status_layout.addWidget(conflict_label) if (search_keyword in grouped_course_name_lower or search_keyword in tc_course_name or search_keyword in teacher_name): matched_tc_list.append(tc) if not courses_grouped and self ._is_searching: self .course_count_label.setText("未找到结果" ) self .show_search_empty_state(self ._current_search_keyword)
守护机制的改造是 v1.9.0 最关键的一次结构调整。早期守护是否继续运行依赖 monitor_state.json 的 is_monitoring + timestamp,很容易遇到状态过期判断和竞态边界。现在的方案是单独引入 watchdog_signal.json,只用 action=start/stop 表达用户意图。开始监控就写 start 并拉起守护,手动停止就写 stop,守护进程按动作执行,不再猜测。
1 2 3 4 5 6 7 8 self .write_watchdog_signal('start' , pid=os.getpid())self .start_watchdog_process()if clear_state: self .write_watchdog_signal('stop' ) self .clear_monitor_state()
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 def should_continue_guarding (): signal = load_signal() if not signal: return False return str (signal.get('action' , '' )).lower() == 'start' def main_loop (main_pid ): while True : time.sleep(CHECK_INTERVAL) if not should_continue_guarding(): log("检测到 stop/无效信号,守护进程退出" ) break if not psutil.pid_exists(main_pid): new_pid = start_main() if new_pid: main_pid = new_pid save_signal({'action' : 'start' , 'pid' : new_pid, 'updated_at' : time.time()})
这套机制配合 monitor_state.json 的课程快照,就形成了“进程守护 + 业务恢复”两层兜底。守护负责拉起进程,UI 负责恢复待抢列表和监控参数,职责边界比以前清楚很多。
日志系统是后面才补齐的,但它对线上排障非常关键。程序和守护都按日期写日志并自动清理旧文件,出问题时可以从时间轴看到“何时掉线、何时重登、何时重启、何时回滚”。这比单纯在控制台 print 有效得多。
1 2 3 4 5 6 7 8 9 10 11 class AppLogger : LOG_DIR = 'logs' LOG_FILE_PREFIX = 'run' RETENTION_DAYS = 7 def _cleanup_old_logs (self ): cutoff_date = datetime.now() - timedelta(days=self .RETENTION_DAYS) pattern = os.path.join(self .LOG_DIR, f'{self.LOG_FILE_PREFIX} _*.log' ) for log_file in glob.glob(pattern): ...
打包链路上,这个项目走的是 PyInstaller 双产物:主程序 YNU选课助手Pro.exe + 守护 Watchdog.exe,再由 NSIS 打安装包。这个结构的好处是守护进程可以独立演进,不会被 GUI 依赖拖胖。后面专门处理了打包环境下的 OCR 提示和路径识别问题,避免用户第一次运行就被无关弹窗打断。
从开发过程回看,这个项目最有价值的经验并不是“抢课 API 怎么调”,而是状态管理要尽量显式,失败路径要比成功路径更完整。只要是长时间运行的工具,最先写完的通常不是最终稳定版本,真正的稳定是靠一次次失败样本喂出来的。下一步我会继续做两件事:把 ui.py 里偏业务的分支继续下沉到 worker/service 层,另外把关键路径日志再结构化,减少靠字符串搜索定位问题的成本。
免责声明 本工具仅用于学习交流与技术研究,旨在帮助用户更高效地查看与监控课程信息。使用者应自行确认并严格遵守学校关于选课系统使用的各项规定。因使用本工具导致的任何后果,包括但不限于账号异常、选课结果偏差、网络请求受限等,均由使用者自行承担。
免费声明 本工具及其相关功能、文档和发布版本均为免费提供,严禁任何单位或个人以售卖、代购、付费代运行、捆绑收费等形式进行商业化牟利。如发现收费行为,请以仓库公开信息为准并及时反馈。