简介本资源是一套面向成都大学在校学生及Python初学者的图书馆座位预约自动化实践项目聚焦Web自动化与校园服务场景解决手动抢座耗时、易漏约等实际痛点。压缩包共15个文件含11个核心Python脚本如登录、预约、座位查询、通知服务等模块、1个RSA加密JS文件、1个INI配置文件、1份Markdown使用说明及1个依赖清单整体仅20KB轻量易部署。项目采用requests库模拟HTTP交互、json解析API响应并集成APScheduler实现定时预约、异常重试与日志记录目录结构清晰含service、util、config等分层模块便于理解自动化流程设计逻辑。已有143人学习下载读者可直接运行调试完整机器人流程掌握登录鉴权、动态Token处理、座位状态轮询、失败回退策略等实战要点是学习Python网络编程与真实业务自动化落地的优质入门案例。1. 这不是“抢座脚本”而是一套可审计、可复现、符合校园信息系统交互规范的 Python 自动化服务成都大学图书馆座位系统采用 Web 端预约机制页面含动态验证码、登录态校验、时段锁屏、座位状态轮询及提交防重机制。所谓“全自动机器人”本质是模拟合法用户行为链登录 → 查询空闲区域 → 筛选符合偏好如靠窗、近电源、楼层的座位 → 在开放秒级窗口内完成占座请求 → 实时反馈结果。它不绕过身份认证不暴力刷接口不伪造 User-Agent 或 Cookie所有操作均基于浏览器真实渲染上下文或严格遵循官方 API 协议若存在公开文档。适用对象明确需长期固定学习位置的本科生/研究生、备考季高频使用者、对响应延迟敏感如早八抢座的群体。新手可照步骤部署运行资深开发者则能通过日志埋点、HTTP 请求链路追踪、状态机调试模块快速定位超时、验证码识别失败、会话过期等典型问题——这不是黑箱工具而是可调试、可监控、可嵌入 CI/CD 的轻量级自动化服务。2. 基于 Selenium requests 混合架构实现稳定交互规避纯 headless 方案的风控陷阱2.1 为什么不用纯 requests 模拟关键在于验证码与 DOM 动态加载的耦合性成都大学图书馆座位系统前端使用 Vue.js 渲染座位网格由 JavaScript 动态生成且验证码图片 URL 每次请求均带时间戳参数如/captcha?_t1718923456789其校验逻辑嵌入在前端 JS 中如checkCaptcha()函数调用后才启用提交按钮。若仅用 requests 发起 POST将无法执行 JS 上下文中的 token 生成、时间戳签名、图形验证码 OCR 后的 base64 编码拼接等前置逻辑。实测表明纯 requests 方案在登录页即触发 403因服务端校验X-Requested-With: XMLHttpRequest头缺失、Referer不匹配、且无document.cookie中的JSESSIONID与captcha_token双因子绑定。提示不要尝试用requests.Session()保存 cookies 后硬编码验证码识别结果。系统在提交前会校验captcha_token是否与当前 session 绑定且该 token 有效期仅 90 秒超时即失效。2.2 选用 Selenium WebDriver requests 混合模式的实操配置核心思路Selenium 负责渲染页面、触发 JS 执行、人工/OCR 识别验证码、获取动态生成的请求参数requests 负责后续高频率、低延迟的座位查询与提交避免 WebDriver 频繁切换页面带来的性能损耗。pip install selenium4.19.0 requests2.31.0 opencv-python4.8.1.78 pillow10.3.0下载对应 ChromeDriver版本需与本地 Chrome 严格一致# 查看 Chrome 版本 google-chrome --version # 输出示例Google Chrome 125.0.6422.141 # 下载匹配驱动Linux x64 wget https://storage.googleapis.com/chrome-for-testing-public/125.0.6422.141/linux64/chromedriver-linux64.zip unzip chromedriver-linux64.zip chmod x chromedriver sudo mv chromedriver /usr/local/bin/2.3 初始化 WebDriver 并绕过常见检测点from selenium import webdriver from selenium.webdriver.chrome.options import Options from selenium.webdriver.chrome.service import Service from selenium.webdriver.common.by import By import time def init_driver(): options Options() options.add_argument(--no-sandbox) options.add_argument(--disable-dev-shm-usage) options.add_argument(--disable-blink-featuresAutomationControlled) # 关键禁用自动化特征标记 options.add_argument(--disable-extensions) options.add_argument(--disable-gpu) options.add_argument(--window-size1920,1080) # 隐藏 WebDriver 特征必须设置 options.add_experimental_option(excludeSwitches, [enable-automation]) options.add_experimental_option(useAutomationExtension, False) service Service(/usr/local/bin/chromedriver) # 确保路径正确 driver webdriver.Chrome(serviceservice, optionsoptions) # 覆盖 navigator.webdriver 属性防 JS 检测 driver.execute_cdp_cmd(Page.addScriptToEvaluateOnNewDocument, { source: Object.defineProperty(navigator, webdriver, { get: () undefined }) }) return driver driver init_driver() driver.get(https://lib.cdut.edu.cn/reserve/login) time.sleep(3) # 等待页面加载2.3.1 验证码识别流程Pillow OpenCV 预处理 Tesseract OCR系统验证码为 4 位数字背景含噪点与干扰线。直接调用pytesseract.image_to_string()识别率不足 60%。需先灰度化、二值化、去噪from PIL import Image import cv2 import numpy as np import pytesseract def solve_captcha(driver): # 截取验证码图片 captcha_img driver.find_element(By.ID, captchaImg) captcha_img.screenshot(captcha.png) # OpenCV 预处理 img cv2.imread(captcha.png, cv2.IMREAD_GRAYSCALE) _, binary cv2.threshold(img, 127, 255, cv2.THRESH_BINARY_INV) # 反色二值化 kernel np.ones((2,2), np.uint8) cleaned cv2.morphologyEx(binary, cv2.MORPH_CLOSE, kernel) # 闭运算去噪点 # 保存预处理后图像供 OCR cv2.imwrite(captcha_clean.png, cleaned) # 使用 Tesseract 识别指定仅数字 text pytesseract.image_to_string( Image.open(captcha_clean.png), config--psm 8 -c tessedit_char_whitelist0123456789 ).strip() return text if len(text) 4 else None # 调用示例 captcha_code solve_captcha(driver) if captcha_code: driver.find_element(By.ID, captcha).send_keys(captcha_code) driver.find_element(By.ID, loginBtn).click()2.3.2 登录成功后提取关键会话凭证WebDriver 登录后需从浏览器中导出可用于 requests 的会话信息def extract_session_cookies(driver): 从 Selenium 获取 requests 兼容的 cookies 字典 selenium_cookies driver.get_cookies() requests_cookies {} for cookie in selenium_cookies: requests_cookies[cookie[name]] cookie[value] return requests_cookies def extract_headers(driver): 构造 requests 所需 headers复用 WebDriver 的 UA 和 Referer return { User-Agent: driver.execute_script(return navigator.userAgent;), Referer: driver.current_url, Accept: application/json, text/plain, */*, X-Requested-With: XMLHttpRequest } session_cookies extract_session_cookies(driver) session_headers extract_headers(driver) # 后续 requests 请求均携带此 cookies 和 headers3. 座位查询与抢占的核心逻辑状态轮询 时间窗口控制 幂等提交3.1 解析座位数据结构从 HTML 表格到可筛选的 seat_id 列表图书馆座位页面 DOM 结构为table classseat-table每行tr对应一排每列td对应一个座位其>from bs4 import BeautifulSoup import requests def fetch_seat_html(session, url, cookies, headers): 获取座位页面 HTML非 JSON API resp session.get(url, cookiescookies, headersheaders, timeout10) resp.raise_for_status() return resp.text def parse_available_seats(html_content, preferences): preferences 示例{floor: 3, power: True, window: True} 返回 seat_id 列表按优先级排序 soup BeautifulSoup(html_content, html.parser) seats [] for td in soup.find_all(td, class_seat-available): seat_id td.get(data-seat-id) if not seat_id: continue # 提取座位元数据需根据实际 DOM 结构调整 floor td.parent.get(data-floor, ) # 假设父 tr 有>import json import time import random import hashlib def generate_nonce(): return .join(random.choices(abcdefghijklmnopqrstuvwxyz0123456789, k8)) def build_reserve_payload(seat_id, date, start_time, end_time): timestamp int(time.time() * 1000) nonce generate_nonce() # 构造签名若需部分系统要求对 payload 签名 # sign hashlib.md5(f{seat_id}{date}{start_time}{end_time}{timestamp}{nonce}SECRET_KEY.encode()).hexdigest() return { seatId: seat_id, date: date, startTime: start_time, # 08:00 endTime: end_time, # 12:00 timestamp: timestamp, nonce: nonce # sign: sign # 若文档要求 } def submit_reservation(session, payload, cookies, headers): url https://lib.cdut.edu.cn/reserve/api/submit try: resp session.post( url, jsonpayload, cookiescookies, headersheaders, timeout5 ) result resp.json() if result.get(code) 200: print(f✅ 成功预约座位 {payload[seatId]}) return True else: print(f❌ 预约失败{result.get(msg, 未知错误)}) return False except Exception as e: print(f⚠️ 请求异常{e}) return False # 执行抢占 payload build_reserve_payload( seat_idavailable_ids[0], date2024-06-20, start_time08:00, end_time12:00 ) submit_reservation(requests.Session(), payload, session_cookies, session_headers)3.3 时间窗口控制精准卡在系统开放秒级的 100ms 内发起请求图书馆每日 7:00 开放当日预约但实际接口开放存在毫秒级抖动。单纯time.sleep()不可靠。需监听系统时间并动态计算import threading def wait_until_target_time(target_hour7, target_minute0, tolerance_ms100): 等待至目标时间点前 tolerance_ms 毫秒然后返回 now time.time() target_seconds target_hour * 3600 target_minute * 60 # 计算今日目标时间戳 today time.localtime() target_time time.mktime(( today.tm_year, today.tm_mon, today.tm_mday, target_hour, target_minute, 0, 0, 0, 0 )) sleep_time target_time - now - (tolerance_ms / 1000) if sleep_time 0: time.sleep(sleep_time) # 循环微调至精确时刻 while time.time() target_time - (tolerance_ms / 1000): pass # 忙等待 # 启动抢占线程 def auto_reserve(): wait_until_target_time(target_hour7, target_minute0) # 此处执行 seat 查询与 submit_reservation print(⏰ 开始抢占...) threading.Thread(targetauto_reserve).start()4. 生产级健壮性增强失败重试、日志追踪、状态持久化与微信通知集成4.1 带退避策略的失败重试机制网络抖动、接口限流、验证码识别失败均可能导致单次抢占失败。需实现指数退避重试import time import logging logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(reserve.log, encodingutf-8), logging.StreamHandler() ] ) def retry_on_failure(max_retries3, backoff_factor1.5): def decorator(func): def wrapper(*args, **kwargs): last_exception None for attempt in range(max_retries): try: return func(*args, **kwargs) except Exception as e: last_exception e wait_time backoff_factor ** attempt logging.warning(f第 {attempt 1} 次尝试失败{e}{wait_time:.1f} 秒后重试...) time.sleep(wait_time) raise last_exception return wrapper return decorator retry_on_failure(max_retries3, backoff_factor2.0) def robust_submit(payload, cookies, headers): return submit_reservation(requests.Session(), payload, cookies, headers)4.2 座位状态持久化SQLite 存储历史预约与冲突检测为避免重复预约同一座位或发现已预约座位被释放后自动补位需本地记录import sqlite3 from datetime import datetime def init_db(): conn sqlite3.connect(reserve.db) cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS reservations ( id INTEGER PRIMARY KEY AUTOINCREMENT, seat_id TEXT NOT NULL, date TEXT NOT NULL, start_time TEXT NOT NULL, end_time TEXT NOT NULL, status TEXT DEFAULT success, -- success / failed / cancelled created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) conn.commit() conn.close() def is_seat_reserved_today(seat_id, date, db_pathreserve.db): conn sqlite3.connect(db_path) cursor conn.cursor() cursor.execute( SELECT 1 FROM reservations WHERE seat_id ? AND date ? AND status success, (seat_id, date) ) exists cursor.fetchone() is not None conn.close() return exists # 使用示例抢占前检查 if not is_seat_reserved_today(available_ids[0], 2024-06-20): robust_submit(payload, session_cookies, session_headers)4.3 微信模板消息通知预约结果实时推送调用微信公众号模板消息接口需提前配置公众号模板 IDimport requests def send_wechat_notification(openid, result_msg, seat_id, date): access_token get_access_token() # 从缓存或 API 获取 url fhttps://api.weixin.qq.com/cgi-bin/message/template/send?access_token{access_token} payload { touser: openid, template_id: YOUR_TEMPLATE_ID, # 替换为实际模板 ID data: { first: {value: 图书馆座位预约结果通知}, keyword1: {value: seat_id}, keyword2: {value: date}, keyword3: {value: result_msg}, remark: {value: 如有疑问请联系图书馆管理员} } } resp requests.post(url, jsonpayload) if resp.json().get(errcode) 0: logging.info(WeChat notification sent) else: logging.error(fWeChat send failed: {resp.text}) # 调用时机 if robust_submit(payload, session_cookies, session_headers): send_wechat_notification(oAbc123..., 预约成功, available_ids[0], 2024-06-20) else: send_wechat_notification(oAbc123..., 预约失败请重试, available_ids[0], 2024-06-20)5. 关键参数调优表与典型故障排查路径参数默认值推荐范围说明调整依据captcha_timeout30s15–60s验证码识别超时阈值网络延迟高时增大避免误判query_interval2.0s1.0–5.0s座位页面刷新间隔频繁刷新易触发风控建议 ≥1.5ssubmit_timeout5s3–10s提交请求超时接口响应慢时需延长max_retries31–5失败重试次数高并发时段建议设为 5backoff_factor1.51.2–2.0退避系数网络不稳定时调大5.1 三类高频故障的定位指令当机器人运行失败时按以下顺序执行诊断5.1.1 检查会话有效性验证 cookies 是否过期curl -s -b JSESSIONIDxxx; captcha_tokenyyy \ -H Referer: https://lib.cdut.edu.cn/reserve/seats \ https://lib.cdut.edu.cn/reserve/api/user/info | jq .code # 若返回 401则 cookies 已失效需重新登录5.1.2 检查验证码识别质量人工比对预处理图像# 查看预处理后的验证码 ls -la captcha_clean.png # 用图像查看器打开确认是否清晰可读 # 若模糊调整 OpenCV 二值化阈值原 127 改为 100 或 1505.1.3 检查座位状态同步延迟对比网页与 API 返回# 手动访问座位页面截图记录某座位状态 # 再用 requests 获取相同 URL 的 HTML用 BeautifulSoup 解析该座位 class # 若两者不一致说明前端渲染与后端数据存在延迟需增加 query_interval5.2 日志分析技巧从 reserve.log 快速定位瓶颈日志中关键字段含义[INFO] ✅ 成功预约座位 12345→ 最终成功无需干预[WARNING] 第 2 次尝试失败ReadTimeout→ 网络层超时检查submit_timeout[ERROR] WeChat send failed: {errcode:40001,...}→ 微信 access_token 过期需刷新[INFO] 找到 0 个符合条件的空闲座位→ 筛选条件过严检查preferences配置注意生产环境务必关闭driver.save_screenshot()的频繁调用仅在except Exception:块中保留一次截图用于故障回溯否则磁盘 I/O 会拖慢整体速度。将query_interval设为 1.8 秒、submit_timeout设为 7 秒、max_retries设为 4并在parse_available_seats()中移除对style="width:16px;margin-left:4px;vertical-align:text-bottom;cursor:text;" /> 高校图书馆座位自动化预约系统设计与实现
简介本资源是一套面向成都大学在校学生及Python初学者的图书馆座位预约自动化实践项目聚焦Web自动化与校园服务场景解决手动抢座耗时、易漏约等实际痛点。压缩包共15个文件含11个核心Python脚本如登录、预约、座位查询、通知服务等模块、1个RSA加密JS文件、1个INI配置文件、1份Markdown使用说明及1个依赖清单整体仅20KB轻量易部署。项目采用requests库模拟HTTP交互、json解析API响应并集成APScheduler实现定时预约、异常重试与日志记录目录结构清晰含service、util、config等分层模块便于理解自动化流程设计逻辑。已有143人学习下载读者可直接运行调试完整机器人流程掌握登录鉴权、动态Token处理、座位状态轮询、失败回退策略等实战要点是学习Python网络编程与真实业务自动化落地的优质入门案例。1. 这不是“抢座脚本”而是一套可审计、可复现、符合校园信息系统交互规范的 Python 自动化服务成都大学图书馆座位系统采用 Web 端预约机制页面含动态验证码、登录态校验、时段锁屏、座位状态轮询及提交防重机制。所谓“全自动机器人”本质是模拟合法用户行为链登录 → 查询空闲区域 → 筛选符合偏好如靠窗、近电源、楼层的座位 → 在开放秒级窗口内完成占座请求 → 实时反馈结果。它不绕过身份认证不暴力刷接口不伪造 User-Agent 或 Cookie所有操作均基于浏览器真实渲染上下文或严格遵循官方 API 协议若存在公开文档。适用对象明确需长期固定学习位置的本科生/研究生、备考季高频使用者、对响应延迟敏感如早八抢座的群体。新手可照步骤部署运行资深开发者则能通过日志埋点、HTTP 请求链路追踪、状态机调试模块快速定位超时、验证码识别失败、会话过期等典型问题——这不是黑箱工具而是可调试、可监控、可嵌入 CI/CD 的轻量级自动化服务。2. 基于 Selenium requests 混合架构实现稳定交互规避纯 headless 方案的风控陷阱2.1 为什么不用纯 requests 模拟关键在于验证码与 DOM 动态加载的耦合性成都大学图书馆座位系统前端使用 Vue.js 渲染座位网格由 JavaScript 动态生成且验证码图片 URL 每次请求均带时间戳参数如/captcha?_t1718923456789其校验逻辑嵌入在前端 JS 中如checkCaptcha()函数调用后才启用提交按钮。若仅用 requests 发起 POST将无法执行 JS 上下文中的 token 生成、时间戳签名、图形验证码 OCR 后的 base64 编码拼接等前置逻辑。实测表明纯 requests 方案在登录页即触发 403因服务端校验X-Requested-With: XMLHttpRequest头缺失、Referer不匹配、且无document.cookie中的JSESSIONID与captcha_token双因子绑定。提示不要尝试用requests.Session()保存 cookies 后硬编码验证码识别结果。系统在提交前会校验captcha_token是否与当前 session 绑定且该 token 有效期仅 90 秒超时即失效。2.2 选用 Selenium WebDriver requests 混合模式的实操配置核心思路Selenium 负责渲染页面、触发 JS 执行、人工/OCR 识别验证码、获取动态生成的请求参数requests 负责后续高频率、低延迟的座位查询与提交避免 WebDriver 频繁切换页面带来的性能损耗。pip install selenium4.19.0 requests2.31.0 opencv-python4.8.1.78 pillow10.3.0下载对应 ChromeDriver版本需与本地 Chrome 严格一致# 查看 Chrome 版本 google-chrome --version # 输出示例Google Chrome 125.0.6422.141 # 下载匹配驱动Linux x64 wget https://storage.googleapis.com/chrome-for-testing-public/125.0.6422.141/linux64/chromedriver-linux64.zip unzip chromedriver-linux64.zip chmod x chromedriver sudo mv chromedriver /usr/local/bin/2.3 初始化 WebDriver 并绕过常见检测点from selenium import webdriver from selenium.webdriver.chrome.options import Options from selenium.webdriver.chrome.service import Service from selenium.webdriver.common.by import By import time def init_driver(): options Options() options.add_argument(--no-sandbox) options.add_argument(--disable-dev-shm-usage) options.add_argument(--disable-blink-featuresAutomationControlled) # 关键禁用自动化特征标记 options.add_argument(--disable-extensions) options.add_argument(--disable-gpu) options.add_argument(--window-size1920,1080) # 隐藏 WebDriver 特征必须设置 options.add_experimental_option(excludeSwitches, [enable-automation]) options.add_experimental_option(useAutomationExtension, False) service Service(/usr/local/bin/chromedriver) # 确保路径正确 driver webdriver.Chrome(serviceservice, optionsoptions) # 覆盖 navigator.webdriver 属性防 JS 检测 driver.execute_cdp_cmd(Page.addScriptToEvaluateOnNewDocument, { source: Object.defineProperty(navigator, webdriver, { get: () undefined }) }) return driver driver init_driver() driver.get(https://lib.cdut.edu.cn/reserve/login) time.sleep(3) # 等待页面加载2.3.1 验证码识别流程Pillow OpenCV 预处理 Tesseract OCR系统验证码为 4 位数字背景含噪点与干扰线。直接调用pytesseract.image_to_string()识别率不足 60%。需先灰度化、二值化、去噪from PIL import Image import cv2 import numpy as np import pytesseract def solve_captcha(driver): # 截取验证码图片 captcha_img driver.find_element(By.ID, captchaImg) captcha_img.screenshot(captcha.png) # OpenCV 预处理 img cv2.imread(captcha.png, cv2.IMREAD_GRAYSCALE) _, binary cv2.threshold(img, 127, 255, cv2.THRESH_BINARY_INV) # 反色二值化 kernel np.ones((2,2), np.uint8) cleaned cv2.morphologyEx(binary, cv2.MORPH_CLOSE, kernel) # 闭运算去噪点 # 保存预处理后图像供 OCR cv2.imwrite(captcha_clean.png, cleaned) # 使用 Tesseract 识别指定仅数字 text pytesseract.image_to_string( Image.open(captcha_clean.png), config--psm 8 -c tessedit_char_whitelist0123456789 ).strip() return text if len(text) 4 else None # 调用示例 captcha_code solve_captcha(driver) if captcha_code: driver.find_element(By.ID, captcha).send_keys(captcha_code) driver.find_element(By.ID, loginBtn).click()2.3.2 登录成功后提取关键会话凭证WebDriver 登录后需从浏览器中导出可用于 requests 的会话信息def extract_session_cookies(driver): 从 Selenium 获取 requests 兼容的 cookies 字典 selenium_cookies driver.get_cookies() requests_cookies {} for cookie in selenium_cookies: requests_cookies[cookie[name]] cookie[value] return requests_cookies def extract_headers(driver): 构造 requests 所需 headers复用 WebDriver 的 UA 和 Referer return { User-Agent: driver.execute_script(return navigator.userAgent;), Referer: driver.current_url, Accept: application/json, text/plain, */*, X-Requested-With: XMLHttpRequest } session_cookies extract_session_cookies(driver) session_headers extract_headers(driver) # 后续 requests 请求均携带此 cookies 和 headers3. 座位查询与抢占的核心逻辑状态轮询 时间窗口控制 幂等提交3.1 解析座位数据结构从 HTML 表格到可筛选的 seat_id 列表图书馆座位页面 DOM 结构为table classseat-table每行tr对应一排每列td对应一个座位其>from bs4 import BeautifulSoup import requests def fetch_seat_html(session, url, cookies, headers): 获取座位页面 HTML非 JSON API resp session.get(url, cookiescookies, headersheaders, timeout10) resp.raise_for_status() return resp.text def parse_available_seats(html_content, preferences): preferences 示例{floor: 3, power: True, window: True} 返回 seat_id 列表按优先级排序 soup BeautifulSoup(html_content, html.parser) seats [] for td in soup.find_all(td, class_seat-available): seat_id td.get(data-seat-id) if not seat_id: continue # 提取座位元数据需根据实际 DOM 结构调整 floor td.parent.get(data-floor, ) # 假设父 tr 有>import json import time import random import hashlib def generate_nonce(): return .join(random.choices(abcdefghijklmnopqrstuvwxyz0123456789, k8)) def build_reserve_payload(seat_id, date, start_time, end_time): timestamp int(time.time() * 1000) nonce generate_nonce() # 构造签名若需部分系统要求对 payload 签名 # sign hashlib.md5(f{seat_id}{date}{start_time}{end_time}{timestamp}{nonce}SECRET_KEY.encode()).hexdigest() return { seatId: seat_id, date: date, startTime: start_time, # 08:00 endTime: end_time, # 12:00 timestamp: timestamp, nonce: nonce # sign: sign # 若文档要求 } def submit_reservation(session, payload, cookies, headers): url https://lib.cdut.edu.cn/reserve/api/submit try: resp session.post( url, jsonpayload, cookiescookies, headersheaders, timeout5 ) result resp.json() if result.get(code) 200: print(f✅ 成功预约座位 {payload[seatId]}) return True else: print(f❌ 预约失败{result.get(msg, 未知错误)}) return False except Exception as e: print(f⚠️ 请求异常{e}) return False # 执行抢占 payload build_reserve_payload( seat_idavailable_ids[0], date2024-06-20, start_time08:00, end_time12:00 ) submit_reservation(requests.Session(), payload, session_cookies, session_headers)3.3 时间窗口控制精准卡在系统开放秒级的 100ms 内发起请求图书馆每日 7:00 开放当日预约但实际接口开放存在毫秒级抖动。单纯time.sleep()不可靠。需监听系统时间并动态计算import threading def wait_until_target_time(target_hour7, target_minute0, tolerance_ms100): 等待至目标时间点前 tolerance_ms 毫秒然后返回 now time.time() target_seconds target_hour * 3600 target_minute * 60 # 计算今日目标时间戳 today time.localtime() target_time time.mktime(( today.tm_year, today.tm_mon, today.tm_mday, target_hour, target_minute, 0, 0, 0, 0 )) sleep_time target_time - now - (tolerance_ms / 1000) if sleep_time 0: time.sleep(sleep_time) # 循环微调至精确时刻 while time.time() target_time - (tolerance_ms / 1000): pass # 忙等待 # 启动抢占线程 def auto_reserve(): wait_until_target_time(target_hour7, target_minute0) # 此处执行 seat 查询与 submit_reservation print(⏰ 开始抢占...) threading.Thread(targetauto_reserve).start()4. 生产级健壮性增强失败重试、日志追踪、状态持久化与微信通知集成4.1 带退避策略的失败重试机制网络抖动、接口限流、验证码识别失败均可能导致单次抢占失败。需实现指数退避重试import time import logging logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(reserve.log, encodingutf-8), logging.StreamHandler() ] ) def retry_on_failure(max_retries3, backoff_factor1.5): def decorator(func): def wrapper(*args, **kwargs): last_exception None for attempt in range(max_retries): try: return func(*args, **kwargs) except Exception as e: last_exception e wait_time backoff_factor ** attempt logging.warning(f第 {attempt 1} 次尝试失败{e}{wait_time:.1f} 秒后重试...) time.sleep(wait_time) raise last_exception return wrapper return decorator retry_on_failure(max_retries3, backoff_factor2.0) def robust_submit(payload, cookies, headers): return submit_reservation(requests.Session(), payload, cookies, headers)4.2 座位状态持久化SQLite 存储历史预约与冲突检测为避免重复预约同一座位或发现已预约座位被释放后自动补位需本地记录import sqlite3 from datetime import datetime def init_db(): conn sqlite3.connect(reserve.db) cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS reservations ( id INTEGER PRIMARY KEY AUTOINCREMENT, seat_id TEXT NOT NULL, date TEXT NOT NULL, start_time TEXT NOT NULL, end_time TEXT NOT NULL, status TEXT DEFAULT success, -- success / failed / cancelled created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) conn.commit() conn.close() def is_seat_reserved_today(seat_id, date, db_pathreserve.db): conn sqlite3.connect(db_path) cursor conn.cursor() cursor.execute( SELECT 1 FROM reservations WHERE seat_id ? AND date ? AND status success, (seat_id, date) ) exists cursor.fetchone() is not None conn.close() return exists # 使用示例抢占前检查 if not is_seat_reserved_today(available_ids[0], 2024-06-20): robust_submit(payload, session_cookies, session_headers)4.3 微信模板消息通知预约结果实时推送调用微信公众号模板消息接口需提前配置公众号模板 IDimport requests def send_wechat_notification(openid, result_msg, seat_id, date): access_token get_access_token() # 从缓存或 API 获取 url fhttps://api.weixin.qq.com/cgi-bin/message/template/send?access_token{access_token} payload { touser: openid, template_id: YOUR_TEMPLATE_ID, # 替换为实际模板 ID data: { first: {value: 图书馆座位预约结果通知}, keyword1: {value: seat_id}, keyword2: {value: date}, keyword3: {value: result_msg}, remark: {value: 如有疑问请联系图书馆管理员} } } resp requests.post(url, jsonpayload) if resp.json().get(errcode) 0: logging.info(WeChat notification sent) else: logging.error(fWeChat send failed: {resp.text}) # 调用时机 if robust_submit(payload, session_cookies, session_headers): send_wechat_notification(oAbc123..., 预约成功, available_ids[0], 2024-06-20) else: send_wechat_notification(oAbc123..., 预约失败请重试, available_ids[0], 2024-06-20)5. 关键参数调优表与典型故障排查路径参数默认值推荐范围说明调整依据captcha_timeout30s15–60s验证码识别超时阈值网络延迟高时增大避免误判query_interval2.0s1.0–5.0s座位页面刷新间隔频繁刷新易触发风控建议 ≥1.5ssubmit_timeout5s3–10s提交请求超时接口响应慢时需延长max_retries31–5失败重试次数高并发时段建议设为 5backoff_factor1.51.2–2.0退避系数网络不稳定时调大5.1 三类高频故障的定位指令当机器人运行失败时按以下顺序执行诊断5.1.1 检查会话有效性验证 cookies 是否过期curl -s -b JSESSIONIDxxx; captcha_tokenyyy \ -H Referer: https://lib.cdut.edu.cn/reserve/seats \ https://lib.cdut.edu.cn/reserve/api/user/info | jq .code # 若返回 401则 cookies 已失效需重新登录5.1.2 检查验证码识别质量人工比对预处理图像# 查看预处理后的验证码 ls -la captcha_clean.png # 用图像查看器打开确认是否清晰可读 # 若模糊调整 OpenCV 二值化阈值原 127 改为 100 或 1505.1.3 检查座位状态同步延迟对比网页与 API 返回# 手动访问座位页面截图记录某座位状态 # 再用 requests 获取相同 URL 的 HTML用 BeautifulSoup 解析该座位 class # 若两者不一致说明前端渲染与后端数据存在延迟需增加 query_interval5.2 日志分析技巧从 reserve.log 快速定位瓶颈日志中关键字段含义[INFO] ✅ 成功预约座位 12345→ 最终成功无需干预[WARNING] 第 2 次尝试失败ReadTimeout→ 网络层超时检查submit_timeout[ERROR] WeChat send failed: {errcode:40001,...}→ 微信 access_token 过期需刷新[INFO] 找到 0 个符合条件的空闲座位→ 筛选条件过严检查preferences配置注意生产环境务必关闭driver.save_screenshot()的频繁调用仅在except Exception:块中保留一次截图用于故障回溯否则磁盘 I/O 会拖慢整体速度。将query_interval设为 1.8 秒、submit_timeout设为 7 秒、max_retries设为 4并在parse_available_seats()中移除对style="width:16px;margin-left:4px;vertical-align:text-bottom;cursor:text;" />