简介这是一份面向演唱会抢票场景的多平台自动化脚本工具覆盖大麦网、淘票票、缤玩岛等主流票务渠道适合具备一定 Python 基础、希望提升抢票效率的用户研究使用。资源包共 19 个文件约 13.57MB以 xml 配置、html 页面、md 说明文档为主另含 py 主脚本、exe 驱动、json 配置及 yml 工作流文件结构上区分了脚本逻辑、页面模板与浏览器驱动等模块。目前已有 1001 人学习下载。通过阅读主脚本与配置文件读者可以了解多平台抢票的请求构造、页面元素定位与驱动调用思路并借助说明文档快速理解各模块职责为二次开发或排错提供参考。1. 多平台抢票脚本的工程化落地从单点脚本到可维护系统抢票脚本这件事很多人第一反应是写个 selenium 循环点击不就完了。但真正跑过演唱会抢票的人都知道单平台单脚本的玩法早就过时了。现在一场热门演出往往同时在多个票务平台上架每个平台的页面结构、接口协议、风控策略都不一样。你需要的不是一段能跑的代码而是一套能同时对接多个平台、可配置、可扩展、能快速适配页面改版的工程化方案。这篇文章面向的是有基本 Python 能力、想自己搭建一套多平台抢票工具链的开发者。我会从架构设计讲到具体实现把大麦、淘票票、缤玩岛这三个典型平台的对接方式拆开揉碎给出可复现的代码骨架和参数配置。不涉及任何绕过平台安全机制的手段纯粹从工程效率角度讨论如何减少手动操作延迟、提高信息获取速度。如果你正在纠结要不要投入时间做这件事我的判断是值得但前提是你把它当成一个需要持续维护的工程项目而不是一劳永逸的神器。2. 多平台抢票脚本的架构选型与核心模块拆解2.1 为什么不用纯 Selenium 方案纯 Selenium 方案在抢票场景下有致命缺陷。浏览器渲染本身有延迟从 DOM 加载到元素可点击通常需要 200-500ms而抢票窗口期可能只有几秒。更麻烦的是大麦和淘票票这类平台对 WebDriver 特征检测越来越严纯 Selenium 很容易被识别。我一般采用的混合架构是接口层用 HTTP 直连获取票务状态和提交订单渲染层用 Playwright 处理登录和复杂验证场景。这样既保留了接口调用的速度优势又能在必要时切换到浏览器模式处理动态内容。核心模块划分如下模块职责技术选型配置中心管理各平台账号、场次、票档、优先级YAML 环境变量平台适配器封装各平台接口差异抽象基类 子类实现调度引擎控制抢票时序、重试策略asyncio 优先级队列状态监控实时检测票务余量变化轮询 WebSocket通知模块抢票结果推送Webhook / 邮件2.2 平台适配器的抽象设计三个平台的接口差异主要体现在请求签名、参数命名和返回结构上。大麦的接口需要携带_m_h5_tk令牌和sign签名淘票票走的是阿里系统一的 MTOP 网关协议缤玩岛相对简单但字段命名自成一套。抽象基类定义统一接口from abc import ABC, abstractmethod from dataclasses import dataclass from typing import Optional dataclass class TicketRequest: 统一的票务请求参数 show_id: str # 演出ID session_id: str # 场次ID price_level: str # 票档 quantity: int 1 # 购买数量 buyer_ids: list None # 观演人ID列表 dataclass class TicketResult: 统一的返回结构 success: bool order_id: Optional[str] None message: str raw_response: dict None class BasePlatformAdapter(ABC): 平台适配器抽象基类 def __init__(self, config: dict): self.config config self.session None abstractmethod async def login(self) - bool: 登录并维持会话 ... abstractmethod async def get_ticket_status(self, req: TicketRequest) - dict: 查询票务余量 ... abstractmethod async def submit_order(self, req: TicketRequest) - TicketResult: 提交订单 ... abstractmethod def _sign_request(self, params: dict) - dict: 平台特定的请求签名 ...这个抽象层的价值在于新增一个平台只需要实现四个方法调度引擎和通知模块完全不用改。我实测下来从零适配一个新平台大约需要 2-3 小时主要时间花在逆向接口参数上。2.3 调度引擎的时序控制抢票的核心竞争力在于时间精度。我的做法是提前 5 分钟启动预热提前 30 秒进入高频轮询开售瞬间并发提交。import asyncio import time from datetime import datetime class TicketScheduler: def __init__(self, adapters: list, config: dict): self.adapters adapters self.target_time config[sale_time] # 开售时间戳 self.preheat_seconds config.get(preheat, 300) self.burst_seconds config.get(burst, 30) async def run(self, req: TicketRequest): now time.time() # 阶段一预热低频检查登录态 while now self.target_time - self.burst_seconds: await self._check_sessions() await asyncio.sleep(10) now time.time() # 阶段二高频轮询多平台并发 tasks [] for adapter in self.adapters: tasks.append(self._burst_attempt(adapter, req)) results await asyncio.gather(*tasks, return_exceptionsTrue) # 阶段三结果汇总任一成功即停止 for r in results: if isinstance(r, TicketResult) and r.success: return r return TicketResult(successFalse, messageall platforms failed) async def _burst_attempt(self, adapter, req): 在开售瞬间以最高频率尝试 while time.time() self.target_time 5: try: status await adapter.get_ticket_status(req) if status.get(available): return await adapter.submit_order(req) except Exception: pass await asyncio.sleep(0.1) # 100ms 间隔避免触发限流 return TicketResult(successFalse)这里有个关键参数asyncio.sleep(0.1)的轮询间隔。设太短会被风控标记设太长会错过票量释放的瞬间。我试过 50ms 到 500ms 的范围100ms 是相对安全的平衡点。另外注意_burst_attempt里的异常吞掉逻辑——抢票过程中网络抖动是常态不能因为一次请求失败就退出循环。2.4 配置中心的设计多平台意味着多套账号和参数。我用 YAML 做配置环境变量存敏感信息# config.yaml platforms: damai: enabled: true base_url: https://mtop.damai.cn app_key: 23781380 # cookie 从环境变量 DAMAI_COOKIE 读取 taopiaopiao: enabled: true base_url: https://mtop.taopiaopiao.com app_key: 4272 binwandao: enabled: false base_url: https://api.binwandao.com target: show_id: your_show_id session_id: your_session_id price_level: 880 quantity: 1 sale_time: 1712000000 # 开售时间戳 scheduler: preheat: 300 burst: 30 retry_interval: 0.1配置和代码分离的好处是换一场演出只需要改 YAML不用动代码。enabled字段让你可以快速切换平台比如某个平台当天维护就直接关掉。3. 三大平台接口对接的实操细节3.1 大麦接口的签名机制与请求构造大麦的 H5 接口走的是 MTOP 网关核心难点在sign参数的计算。签名算法是md5(token t appKey data)其中token来自 cookie 中的_m_h5_tk字段。import hashlib import time import requests class DamaiAdapter(BasePlatformAdapter): def __init__(self, config): super().__init__(config) self.base_url config[base_url] self.app_key config[app_key] self.session requests.Session() def _get_token(self): 从 cookie 中提取 _m_h5_tk 令牌 for cookie in self.session.cookies: if cookie.name _m_h5_tk: return cookie.value.split(_)[0] raise ValueError(token not found, please login first) def _sign_request(self, params: dict) - dict: 构造 MTOP 签名参数 t str(int(time.time() * 1000)) token self._get_token() data params.get(data, {}) # 签名公式md5(tokentappKeydata) raw f{token}{t}{self.app_key}{data} sign hashlib.md5(raw.encode()).hexdigest() params.update({ t: t, sign: sign, appKey: self.app_key, }) return params async def get_ticket_status(self, req: TicketRequest) - dict: 查询余票走 mtopticket 接口 api mtop.damai.wireless.project.getProjectPage params { data: f{{itemId:{req.show_id},platform:8}}, } signed self._sign_request(params) resp self.session.get( f{self.base_url}/h5/{api}/1.0/, paramssigned, timeout5 ) result resp.json() # 解析余票状态 sku_list result.get(data, {}).get(skuList, []) for sku in sku_list: if sku.get(priceName) req.price_level: return {available: sku.get(qty, 0) 0, qty: sku.get(qty)} return {available: False, qty: 0}这里有几个容易翻车的点。第一_m_h5_tk令牌有有效期通常 24 小时过期过期后签名会返回FAIL_SYS_TOKEN_EXOIRED需要重新登录刷新。第二data参数必须是紧凑 JSON不能有空格否则签名对不上。第三请求头需要带Referer: https://m.damai.cn/否则会被网关拦截。3.2 淘票票的 MTOP 协议适配淘票票同样走 MTOP但 appKey 和接口名不同且部分接口需要x-sgext扩展头。它的签名逻辑和大麦一致所以可以复用签名函数只需要替换 appKey 和接口路径。class TaopiaopiaoAdapter(BasePlatformAdapter): def __init__(self, config): super().__init__(config) self.base_url config[base_url] self.app_key config[app_key] self.session requests.Session() def _sign_request(self, params: dict) - dict: 复用 MTOP 签名逻辑appKey 不同 t str(int(time.time() * 1000)) token self._get_token() data params.get(data, {}) raw f{token}{t}{self.app_key}{data} sign hashlib.md5(raw.encode()).hexdigest() params.update({t: t, sign: sign, appKey: self.app_key}) return params async def submit_order(self, req: TicketRequest) - TicketResult: 提交订单注意需要先调用 buildOrder 接口 # 第一步构建订单 build_api mtop.taopiaopiao.order.build build_params { data: f{{itemId:{req.show_id},skuId:{req.session_id}, fquantity:{req.quantity},buyerIds:{req.buyer_ids}}} } signed self._sign_request(build_params) resp self.session.post( f{self.base_url}/h5/{build_api}/1.0/, datasigned, timeout5 ) build_result resp.json() if not build_result.get(data, {}).get(buildSuccess): return TicketResult(successFalse, messagebuild order failed) # 第二步确认提交 submit_api mtop.taopiaopiao.order.submit submit_params { data: f{{buildId:{build_result[data][buildId]}}} } signed self._sign_request(submit_params) resp self.session.post( f{self.base_url}/h5/{submit_api}/1.0/, datasigned, timeout5 ) result resp.json() if result.get(ret, [])[0].startswith(SUCCESS): return TicketResult( successTrue, order_idresult[data].get(orderId), raw_responseresult ) return TicketResult(successFalse, messagestr(result.get(ret)))淘票票的两阶段提交build submit是个坑点。很多人只调了 submit 接口结果一直报参数错误。必须先 build 拿到buildId再提交。另外buyerIds需要提前在账号里添加观演人信息接口不会帮你创建。3.3 缤玩岛的简化对接与通用重试策略缤玩岛的接口相对简单没有复杂的签名机制但字段命名比较独特。它的余票查询返回的是stock字段而非qty票档用ticketTypeName而非priceName。class BinwandaoAdapter(BasePlatformAdapter): def __init__(self, config): super().__init__(config) self.base_url config[base_url] self.session requests.Session() def _sign_request(self, params: dict) - dict: 缤玩岛使用简单的 token 头认证 params[token] self.config.get(token, ) return params async def get_ticket_status(self, req: TicketRequest) - dict: api /api/ticket/stock params { showId: req.show_id, sessionId: req.session_id, } signed self._sign_request(params) resp self.session.get( f{self.base_url}{api}, paramssigned, timeout5 ) result resp.json() for ticket in result.get(data, {}).get(tickets, []): if ticket.get(ticketTypeName) req.price_level: return { available: ticket.get(stock, 0) 0, qty: ticket.get(stock) } return {available: False, qty: 0}通用重试策略我封装了一个装饰器所有平台的网络请求都套上import functools import asyncio def retry_on_failure(max_retries3, delay0.5, backoff2): 指数退避重试装饰器 def decorator(func): functools.wraps(func) async def wrapper(*args, **kwargs): last_exc None for attempt in range(max_retries): try: return await func(*args, **kwargs) except (requests.Timeout, requests.ConnectionError) as e: last_exc e wait delay * (backoff ** attempt) await asyncio.sleep(wait) raise last_exc return wrapper return decorator参数说明max_retries3是经验值再多会拖慢整体时序delay0.5起步配合backoff2形成 0.5s、1s、2s 的退避序列。注意这个装饰器只捕获网络异常业务逻辑错误比如签名失败不应该重试因为重试也不会成功。4. 抢票脚本避坑指南5 个血泪教训4.1 Cookie 过期导致签名全部失效现象脚本前几天跑得好好的突然所有请求返回FAIL_SYS_TOKEN_EXPIRED或FAIL_SYS_ILLEGAL_ACCESS。原因大麦和淘票票的_m_h5_tk令牌有效期通常为 24 小时过期后签名计算用的 token 就是错的网关直接拒绝。解决在调度引擎里加一个登录态检查任务每 6 小时跑一次login()刷新 cookie。更稳妥的做法是每次抢票前 30 分钟强制重新登录一次。代码里可以这样判断async def _check_sessions(self): 预热阶段定期检查登录态 for adapter in self.adapters: try: # 用一个轻量接口探测登录态 await adapter.get_ticket_status( TicketRequest(show_idtest, session_idtest, price_leveltest) ) except Exception as e: if TOKEN in str(e).upper(): await adapter.login() # 重新登录4.2 高频轮询触发风控封 IP现象开售瞬间脚本疯狂发请求前几秒正常然后突然全部超时换 IP 后恢复。原因平台风控系统检测到同一 IP 在短时间内大量请求触发限流或封禁。大麦的风控阈值大约在每秒 5-10 次请求。解决把轮询间隔控制在 100ms 以上并且多个平台不要同时用同一个出口 IP。如果条件允许给每个平台适配器配置独立的请求会话和代理。另外在_burst_attempt里加一个计数器连续失败 10 次就暂停 1 秒再继续。4.3 订单提交成功但未支付被取消现象脚本显示submit_order返回成功但去订单列表一看订单状态是已取消。原因抢票脚本只完成了下单动作没有完成支付。大部分平台的下单和支付是分离的订单有 5-15 分钟的支付窗口期超时自动取消。解决在TicketResult里增加支付链接字段抢到票后立即通过通知模块推送到手机人工完成支付。不要试图在脚本里自动支付——支付环节涉及密码或生物验证自动化风险极高且容易触发账户安全锁定。4.4 场次 ID 和票档 ID 混淆现象脚本能查到余票但提交订单时报参数错误或票档不存在。原因大麦的sessionId和priceLevel是两套 ID 体系。sessionId标识场次比如8月15日 19:30priceLevel标识票档比如880元看台。很多人把票档名称直接当 ID 传进去接口不认。解决提前调一次getProjectPage接口把返回的skuList里的skuId和priceName对应关系存下来。提交订单时用skuId而不是priceName。我一般会在配置里同时写名称和 ID脚本启动时校验一遍。4.5 多平台并发导致本地资源耗尽现象同时开启三个平台的抢票任务后脚本卡死日志停止输出。原因每个平台适配器都开了独立的requests.Session加上 asyncio 的并发任务如果没控制好并发数本地文件描述符或内存会爆。解决用asyncio.Semaphore限制全局并发数建议不超过 10。另外每个适配器的 session 要复用不要每次请求都新建。在调度引擎初始化时创建 session 池class AdapterPool: def __init__(self, max_concurrent10): self.semaphore asyncio.Semaphore(max_concurrent) async def execute(self, adapter, method, *args): async with self.semaphore: return await getattr(adapter, method)(*args)5. 抢票脚本的进阶调优与效果验证5.1 用时间同步校准本地时钟抢票对时间精度要求极高本地时钟和服务器时间差 1 秒就可能错过。我习惯在脚本启动时先做一次 NTP 校准或者直接请求平台服务器的时间戳接口计算本地偏移量。import time import requests def calibrate_time_offset(server_url: str) - float: 计算本地时间与服务器时间的偏移 t1 time.time() resp requests.head(server_url, timeout3) t2 time.time() server_time float(resp.headers.get(Date, 0)) if server_time 0: return 0.0 # 粗略估算请求往返时间的一半作为补偿 rtt t2 - t1 local_estimate t1 rtt / 2 return server_time - local_estimate拿到偏移量后在调度引擎里用time.time() offset替代time.time()这样开售时间的判断能精确到毫秒级。5.2 验证脚本是否真正生效写完脚本不要直接等开售日先用一场不太热门的演出做全流程演练。验证清单如下验证项方法预期结果登录态调用 get_ticket_status返回正常余票数据签名正确检查返回 ret 字段无 TOKEN/SIGN 错误下单流程用测试场次走 buildsubmit生成订单不支付通知推送触发一次成功回调手机收到消息时间精度对比本地和服务器时间偏差 500ms演练时把sale_time设成当前时间加 2 分钟观察脚本是否在正确的时间点进入 burst 阶段。如果提前或延后超过 1 秒检查时间偏移量是否计算正确。5.3 持续维护的习惯抢票脚本不是一次性投入。平台接口平均每 2-3 个月会有一次小改动页面结构可能半年大改一次。我的习惯是每次抢票前一周跑一次全流程测试确认三个平台的接口都还通。如果某个平台返回结构变了优先看raw_response里的错误码大部分问题出在签名参数或接口版本号上。另外把每次抢票的日志完整保留下来包括请求参数、返回结果、时间戳。出问题的时候这些日志就是唯一的后悔药。我一般按日期分目录存保留最近 30 天。最后说一个心态问题抢票脚本能提高成功率但不能保证 100% 抢到。热门演出的票量可能在开售 3 秒内售罄脚本的优势在于比手动快那么几百毫秒。把预期放平把工程做扎实剩下的交给运气。希望帮到你。本文还有配套的精品资源点击获取
