简介这份面向大麦网抢票场景的 Selenium 自动化脚本适合具备一定 Python 基础、希望研究移动端抢票流程与支付闭环的开发者参考。压缩包共 3 个文件约 6KB由 1 个 Python 主脚本、1 个 Markdown 说明文档和 1 个 ini 配置文件组成分别承载核心执行逻辑、使用说明与参数配置结构精简便于快速定位关键实现。已有 981 人学习说明其在抢票自动化细分场景中具有一定关注度。读者可从中了解 Android 与 iOS 设备上的脚本运行思路、大麦网抢票与支付宝支付流程的衔接方式以及日志记录、滑块验证处理等常见环节的组织方法配置项与说明文档配合也便于按自身环境调整参数、理解模块职责并在此基础上做二次开发或排错分析。1. 抢一张回流票有多难这套「脚本-某票务-回流-支付」组合到底在解决什么某场演出开票三分钟售罄真正的机会反而不在开票那一刻——有人退票、有人付款超时系统会把票零星释放回池子这就是「回流票」。人工刷新页面根本抓不住这种几秒钟的时间窗口而这套「脚本-某票务-回流-支付」组合干的就是把「盯着回流、拼好下单参数脚本里简称 bp、拉起支付」这条链路自动化。它不是一个开箱即用的 App而是一份能跑起来的 Python 工程样本:登录态维持、票档状态轮询、下单参数组装、订单状态流转、支付回调幂等,每一块都能单独拆出来复用到其他有状态接口的自动化场景里。适合已经写过 requests、想学轮询调度和订单幂等的开发者不适合只想点两下就出票的人。2. 工程骨架目录结构、配置加载与启动链路拿到压缩包第一件事不是python main.py而是先把目录判一遍。这类脚本最容易出问题的地方从来不是算法而是「凭据写死在哪、状态存在哪、失败了能不能从中间接着跑」。先看懂骨架后面调参、加日志、换通知渠道都省事。2.1 目录结构五个包各管一件事拆开后的目录大致是下面这个形态包名做过脱敏但职责划分是这类项目的通用套路路径职责二次开发时最常动的地方config.py账号、目标商品、轮询与支付参数换票档、改轮询间隔core/session.py会话创建、Cookie/Token 装载与刷新登录态失效处理core/bp.py下单请求参数bp 层组装观演人、票档映射core/poller.py回流轮询与去重轮询节奏、触发条件core/order.py订单状态机与支付调用支付渠道、超时策略utils/日志、重试装饰器、通知加埋点、换 webhookmain.py组装依赖、启动主循环多目标并发core下的四个文件基本对应一条订单的生命周期拿到会话 → 拼参数 → 轮询等到有票 → 下单 → 拉起支付 → 等回调。utils是横切关注点日志和重试几乎每个模块都会 import。我一般会先看utils/logger.py因为脚本跑起来的第一手信息全靠它日志格式乱的项目后面排查基本靠猜。值得提醒的是这类项目通常不带数据库状态全部在内存里一个 dict 或者 dataclass。好处是部署简单坏处是进程一挂跑到一半的订单信息就没了。如果你打算长期挂着跑第一件改造就是把订单状态落盘这件事在第 6 章会展开。2.2 配置与环境一份 config 决定跑不跑得起来环境上没什么玄学Python 3.8 以上、requests、loguru或者标准库 logging就够了有的版本会带schedule或apscheduler做定时。依赖一般在一个requirements.txt里三五行装完就能 import。真正的门槛在配置。集中式配置写成 py 文件的写法# config.py CONFIG { # 账号侧登录后从浏览器开发者工具里取的会话凭据 account: { cookie: , token: , refresh_gap: 600, # 秒提前刷新登录态的间隔 }, # 目标侧想买的到底是什么 target: { item_id: , # 演出/商品 ID show_id: , # 场次 ID sku_id: , # 票档 ID buy_num: 1, # 购买张数 audience_ids: [], # 观演人 ID 列表 }, # 轮询侧 poll: { interval: 0.8, # 基础间隔秒 jitter: 0.3, # 随机抖动上限秒 max_round: 3600, # 最大轮询轮数防止无限跑 }, # 支付侧 pay: { channel: balance, # 支付渠道标识 pay_timeout: 15, # 单次支付请求超时秒 retry: 2, # 支付失败重试次数 }, notify: {webhook: }, # 结果通知地址留空则不推 }这份配置里item_id / show_id / sku_id三个 ID 决定「买什么」。它们不是随便填的必须和平台前端的请求参数一一对应抓错一个层级最常见的结果是请求返回成功但业务码报「参数错误」。audience_ids是实名观演人数量必须和buy_num对齐否则下单接口会直接拒掉。interval jitter这一对是给轮询节奏用的固定间隔容易让请求呈现明显的周期性加抖动是为了让节奏看起来更自然同时也避免多条任务在同一毫秒撞在一起。这里有个血泪经验不要把凭据写进config.py然后提交到任何仓库。我一般改成os.getenv读取本地用.env配置文件里只留os.getenv(TICKET_TOKEN, )这样的默认值。省下的不是安全审计的功夫是某天发现自己 token 被同步到公共仓库时的那一身冷汗。环境跑起来的顺序也简单装依赖 → 填 config → 先跑一次python -m core.session --check这类自检入口有的版本没写就自己在 main 里加一段确认登录态有效 → 再开主循环。跳过自检直接开跑最容易出现的现象是轮询了几百轮全是 401白等半小时。2.3 启动链路从 main 到第一个请求发出去main.py通常很短短到很多人以为它没用其实它决定了整个脚本的组装方式和退出时机。典型写法# main.py import logging from config import CONFIG from core.session import Session from core.poller import Poller from core.order import OrderFlow from utils.notify import push def main(): logging.basicConfig( levellogging.INFO, format%(asctime)s %(levelname)s %(name)s %(message)s, ) # 1. 建会话内部会装载 cookie/token并做一次可用性探测 session Session(CONFIG[account]) # 2. 轮询器只负责「发现可买的 sku」不负责下单 poller Poller(session, CONFIG[target], CONFIG[poll]) # 3. 订单流负责下单 支付依赖同一个会话 flow OrderFlow(session, CONFIG[target], CONFIG[pay]) try: for hit in poller.watch(): # 生成器yield 出可买信号 result flow.run(hit) # 下单并尝试支付 push(CONFIG[notify], result) # 无论成败都推一条 if result.ok: break # 买到即停避免重复下单 except KeyboardInterrupt: logging.info(手动中断正在退出) finally: session.close() if __name__ __main__: main()这段代码的关键在于职责边界Poller只判断「有没有票」OrderFlow只处理「有票之后怎么办」。两者通过一个生成器watch()解耦好处是轮询逻辑可以单独测——把watch()换成返回假数据的迭代器就能在不碰真实接口的情况下把下单和支付链路跑通。这是我在拆这类项目时最看重的一点能离线测的部分越多线上调试的轮次越少。session是唯一被两个模块共享的对象目的是复用连接池和登录态。很多翻车案例都源于每个模块自己requests.get()结果连接不复用、Cookie 不同步、限速也做不了。finally里调session.close()看着可有可无实际上如果脚本被 CtrlC 打断未关闭的连接池偶尔会让进程挂住不退加上这一行能省掉不少「按了 CtrlC 还得 kill -9」的烦躁。参数上要注意poller.watch()是生成器意味着轮询是惰性的for循环拿到第一个信号就开始走下单下单失败后继续回到watch()等待下一个回流。这个设计天然支持「抢到就停」和「失败重试」但你得自己保证break条件写对——少写一个if result.ok: break脚本会在买到之后继续下单超买的部分只能自己消化。3. bp 参数组装一次下单请求里到底塞了什么bp 这一层是整套脚本里最枯燥、也最容易出错的部分。它做的事情就是把「人想买的东西」翻译成「接口认识的字段」。翻译错一个字段前面所有轮询都白费所以值得单独拎出来讲。3.1 会话与登录态cookie、token 怎么在类之间流转Session类通常是薄薄一层但承担了三个职责装载凭据、探测有效性、定时刷新。参考实现# core/session.py import time import requests class Session: def __init__(self, account_cfg): self.cfg account_cfg self.s requests.Session() self.s.headers.update({ User-Agent: Mozilla/5.0 ..., # 换成你自己的浏览器 UA Referer: https://example.com/, # 平台对应页面 }) self._born time.time() self._load_auth() def _load_auth(self): # 凭据只在创建时装一次后续靠 refresh 续期 self.s.cookies.set(cookie, self.cfg[cookie]) self.token self.cfg[token] def need_refresh(self): # 距上次装载超过阈值就认为需要续期 return time.time() - self._born self.cfg[refresh_gap] def refresh(self): # 实际项目里这里走平台的续期接口拿到新 token 回填 new_token self._renew_token() if new_token: self.token new_token self._born time.time() def close(self): self.s.close()三个参数值得说refresh_gap是最容易被忽略的一个设得比平台实际过期时间短一截才安全比如会话实际 30 分钟过期阈值就设 600 秒留出重试余量。_load_auth里的 cookie 用set而不是直接塞 headers是为了让requests在后续请求里自动带上并处理更新。_born记录装载时间配合need_refresh实现惰性刷新——不是开个线程定时刷而是在每次发起请求前判断一次结构简单也不会在空闲时白白刷。3.2 bp 层参数拼装商品、场次、票档、观演人真正给下单接口用的参数在core/bp.py里组装。它的输入是 config 里的target和当前会话的 token输出是一个扁平 dict# core/bp.py import time def build_bp_params(target, token, extraNone): 把业务目标翻译成下单接口参数。extra 用于透传平台侧必填的额外字段 params { itemId: target[item_id], # 演出 ID showId: target[show_id], # 场次 ID skuId: target[sku_id], # 票档 ID buyNum: target[buy_num], # 张数 audienceIds: ,.join(target[audience_ids]), # 观演人逗号拼接 timestamp: int(time.time() * 1000), # 毫秒时间戳防重放 token: token, } if extra: params.update(extra) # 渠道、活动等可选字段 return {k: v for k, v in params.items() if v not in (None, )} # 清掉空值audienceIds用逗号拼接还是数组取决于平台接口约定脚本里常见的是字符串拼接因为很多平台的表单接口不接受嵌套结构。timestamp是毫秒级和平台侧允许的时间偏差一般只有几十秒机器时间跑偏超过阈值就会稳定报「请求过期」这一点在第 5 章会当成坑单独讲。最后一行清空值的处理看着不起眼但它能避免把extra里没填的字段塞进请求某些接口对空字符串是严格校验的。参数组装完还有一个隐含步骤签名。签名算法由平台侧决定脚本侧一般是在合法会话内复用已有的登录态和前端约定的字段顺序不自己去逆向算。真要改这块先确认你手上的账号和操作符合平台规则再谈实现。3.3 请求发送与限速连接复用、退避与失败判定参数有了发请求这一段反而是最需要工程感的。一个能长期跑的脚本和一个跑十分钟就被限流的脚本差别基本都在这里# core/http.py import time, random, requests from requests.adapters import HTTPAdapter def make_session(pool_size10): s requests.Session() adapter HTTPAdapter(pool_connectionspool_size, pool_maxsizepool_size) s.mount(https://, adapter) # 复用连接减少握手开销 return s def request_with_backoff(session, method, url, **kwargs): base 1.0 for attempt in range(4): resp session.request(method, url, timeout5, **kwargs) if resp.status_code 200: return resp if resp.status_code in (429, 503): # 明确的可重试状态 sleep base * (2 ** attempt) random.uniform(0, 0.3) time.sleep(sleep) # 指数退避 抖动 continue resp.raise_for_status() # 其余错误直接抛 raise RuntimeError(重试次数用尽)逻辑上分三层。第一层是连接池pool_maxsize设成并发数的 12 倍就够设太大会让平台侧看到一堆并行连接反而更显眼。第二层是重试策略只对 429限流和 503服务不可用做退避重试其他 4xx 直接抛因为参数错了重试一万次还是错重试只会浪费轮询预算。第三层是指数退避加抖动第一次等 1 秒、第二次 2 秒、第三次 4 秒叠上 00.3 秒随机避免多个任务在同一时刻集体重试。timeout5这个参数别省。默认不设超时的话一个卡住的连接能把整个轮询循环拖死日志上看着像「脚本没反应」实际是阻塞在网络读上。5 秒对这类接口绰绰有余真需要慢接口再单独放宽。4. 回流监控与订单状态机从「有票了」到「支付完成」轮询和支付是这条链路上时间跨度最大的两段也是最容易出「跑通一次、跑不通第二次」的地方。前者考的是节奏控制后者考的是幂等。4.1 回流轮询间隔、去重、失效判定轮询的核心不是「一直请求」而是「用最少的请求换最高的命中率」。参考实现# core/poller.py import time, random, logging log logging.getLogger(__name__) class Poller: def __init__(self, session, target, poll_cfg): self.session session self.target target self.cfg poll_cfg self.seen set() # 已处理过的 sku防重复下单 def watch(self): for rnd in range(self.cfg[max_round]): if self.session.need_refresh(): self.session.refresh() # 轮询间隙顺手续期 state self._fetch_state() if state is None: time.sleep(self._sleep()) # 请求失败也按节奏走不空转 continue if state[remain] 0 and state[sku_id] not in self.seen: log.info(命中回流 sku%s remain%s, state[sku_id], state[remain]) self.seen.add(state[sku_id]) # 先占位再交给下游 yield state time.sleep(self._sleep()) def _sleep(self): return self.cfg[interval] random.uniform(0, self.cfg[jitter]) def _fetch_state(self): try: resp self.session.s.get( https://example.com/api/sku/state, params{skuId: self.target[sku_id]}, timeout5, ) data resp.json() return data.get(data) if data.get(code) 0 else None except Exception as e: log.warning(查询票档状态失败: %s, e) return None三个设计点。seen集合在yield之前就写入这是防重入的关键——如果等到下游下单成功再标记轮询在等待期间可能又命中同一个 sku导致重复提交。_sleep()每次重新计算抖动是逐轮变化的不会形成固定节拍。_fetch_state把异常吞掉返回 None是因为轮询周期里偶发超时太正常了一次异常就退出循环等于白白浪费前面的等待但要打 warning 日志别让问题静默。max_round是兜底配合interval大致能算出最长运行时长。3600 轮 × 1 秒左右大约一小时够覆盖大部分回流高峰。真需要挂通宵把max_round调大别改成while True——没有上限的循环在出错时很难停下来。4.2 订单状态机待支付、已支付、超时关闭订单不是「下单成功就完事」它有一组明确的状态迁移。脚本里一般用一个小枚举加一个内存 dict 表达状态触发条件下一步动作CREATED下单接口返回订单号立即发起支付PAYING支付请求已发出轮询查单结果PAID查单返回已支付记录、通知、退出CLOSED超时未支付 / 平台关闭清理本地状态重新进入轮询FAILED下单或支付明确失败记录原因视情况重试状态机存在的意义是把「重试」和「重复下单」区分开。PAYING状态下重试是安全的因为订单号已经确定重试的只是支付动作CREATED状态如果因为超时重发下单请求可能生成两笔订单这时候必须靠本地订单号或者幂等键来挡。4.3 支付链路对接下单、查单、回调幂等支付这段最容易踩的是「请求超时但实际成功」。网络抖一下支付请求没拿到响应本地以为失败去重试平台侧却已经扣款。解决思路是支付前先查、支付后必查# core/order.py import time, logging log logging.getLogger(__name__) class OrderFlow: def __init__(self, session, target, pay_cfg): self.session session self.target target self.pay_cfg pay_cfg self.local_orders {} # order_no - status落盘更稳 def run(self, hit): order_no self.local_orders.get(hit[sku_id]) if order_no and self._query(order_no)[status] PAID: return Result(okTrue, order_noorder_no) # 幂等已付直接返回 order_no order_no or self._create_order(hit) # 有单号就不再重复下单 self.local_orders[hit[sku_id]] order_no for i in range(self.pay_cfg[retry] 1): self._pay(order_no) time.sleep(1) # 给平台侧落账留点时间 info self._query(order_no) if info[status] PAID: return Result(okTrue, order_noorder_no) if info[status] CLOSED: break return Result(okFalse, order_noorder_no) def _create_order(self, hit): resp self.session.s.post(https://example.com/api/order/create, json{skuId: hit[sku_id], num: self.target[buy_num]}, timeout5) return resp.json()[data][orderNo] def _pay(self, order_no): self.session.s.post(https://example.com/api/pay, json{orderNo: order_no, channel: self.pay_cfg[channel]}, timeoutself.pay_cfg[pay_timeout]) def _query(self, order_no): resp self.session.s.get(https://example.com/api/order/query, params{orderNo: order_no}, timeout5) return resp.json()[data]这里有三处幂等设计。开头先查一次如果本地记着单号且平台已显示已支付直接返回成功不再走下单。order_no order_no or self._create_order(hit)保证同一个 sku 只创建一次订单重试只重试支付。循环里每次_pay之后都_query以平台查询结果为准而不是以支付请求的返回值当结论——支付接口返回成功不等于订单已支付中间还有渠道处理。pay_timeout单独给到 15 秒比普通请求宽因为支付链路涉及的环节更多。time.sleep(1)是给平台侧落账留的缓冲不加的话查单常常抢在状态更新之前表现为「明明付了却显示待支付」然后白白多跑一轮重试。5. 避坑与排查那些跑通一次、跑不通第二次的地方下面这几条都是这类脚本里高频复现的问题。写出来不是为了吓人是因为每一条我都真实遇到过且每一条的现象都很容易被误判成「平台封了」。5.1 请求层的坑现象轮询几百轮全是 401/403重启后正常几分钟又不行。原因通常是会话续期逻辑没触发或者refresh_gap设得比平台实际过期时间长。脚本装载凭据后一直用同一个 token看起来请求都发出去了实际全被拒。解决是打开 debug 日志确认 token 装载时间把refresh_gap压到过期时间的 1/2 以下同时在请求前加一次need_refresh()判断让续期发生在使用之前而不是出错之后。现象本机跑一切正常换台机器就报「请求过期」或「签名不合法」。八成是系统时间偏移。毫秒级时间戳参与参数拼装时机器时间和标准时间差几十秒就会稳定失败。解决是先对一下时间同步状态容器环境尤其注意宿主机时间确认无误后再看参数里时间戳的单位是秒还是毫秒这一处写错在本地往往也能「偶尔成功」最难排查。现象一开始命中率不错跑久了几乎抓不到票。常见原因是轮询节奏太规律或者太密。固定间隔的请求会呈现明显的周期性间隔过短又会触发限流表现为返回 429 或者返回正常但数据是空。解决是把固定sleep换成「基础间隔 逐轮变化抖动」并把轮询轮数限制住不要开着while True挂几天。密度这件事没有通用数值只能在自己的账号权限范围内小步调调一次观察日志十分钟再继续。5.2 状态与支付的坑现象日志显示下单成功但平台上有两笔待支付订单。原因是没有在yield之前占位轮询在等待下游结果期间又命中了同一个 sku重复走了_create_order。解决是把「已处理 sku」集合的写入提前到生成器内部同时给_create_order加一层本地单号缓存有单号就不再调下单接口。现象支付请求超时重试后提示「订单已支付」本地状态却是失败。这是典型的「请求失败但业务成功」。解决是永远以查单结果为准支付前查一次支付后必查一次查单返回已支付就立刻终止重试。同时把本地订单状态落盘进程重启后还能读到PAYING的订单继续查而不是从头再来一遍。我见过最惨的情况是脚本崩了重启把同一批票重复下了三次最后只能一张张手动关单。现象明明看到票档有余量下单却返回「库存不足」。回流票的余量是瞬时值从轮询看到到真正下单中间隔了几百毫秒被别人抢走很正常。解决是不要因为一次下单失败就认为脚本坏了把「下单失败」当作正常分支处理记录原因后回到轮询继续等下一个回流。真正需要报警的是连续多次参数类错误那才是配置写错了。6. 二次开发把它改成你自己的调度与通知中台原始脚本的形态是「一个进程盯一个目标」实际用起来很快就会想扩成多目标、多账号、可观测。改造的抓手有三处按投入产出比排序。6.1 把轮询从主循环里拆出来做成任务现在main.py是一个for循环加第二个目标就得复制一份。更省事的做法是把Poller和OrderFlow组合成一个 task交给调度器统一管理# scheduler.py from concurrent.futures import ThreadPoolExecutor from config import CONFIG from core.session import Session from core.poller import Poller from core.order import OrderFlow def run_task(target_cfg, session_pool): session session_pool.get(target_cfg[account]) poller Poller(session, target_cfg, CONFIG[poll]) flow OrderFlow(session, target_cfg, CONFIG[pay]) for hit in poller.watch(): result flow.run(hit) if result.ok: return result return None def start(tasks): # 每个目标一个线程线程数等于目标数别开太大 with ThreadPoolExecutor(max_workerslen(tasks)) as pool: futures [pool.submit(run_task, t, SESSIONS) for t in tasks] for f in futures: print(f.result()) # 实际项目里走通知不打 print要点是会话复用同一个账号下的多个目标共用一个Session不同账号之间隔离。线程数不要盲目放大目标数量多于 5 个时考虑改成队列 固定 worker避免瞬间并发把账号推到限流边缘。6.2 加一层落盘和一条通知内存状态最大的问题是重启即失忆。最小成本的改造是用 sqlite 建一张表把订单号和状态写进去-- schema.sql CREATE TABLE IF NOT EXISTS orders ( order_no TEXT PRIMARY KEY, -- 平台订单号天然幂等键 sku_id TEXT NOT NULL, status TEXT NOT NULL, -- CREATED / PAYING / PAID / CLOSED updated_at INTEGER NOT NULL );order_no做主键插入用INSERT OR REPLACE重启后先按status PAYING查一遍把未完成的订单接着查完。这一层加上之后脚本崩了重启不再是灾难只是多跑几秒查询。通知那侧同理别在成功分支里直接print统一走utils/notify把成功、失败、超时三种结果都推出去——抢票这件事失败的通知比成功更有价值因为失败意味着你得赶紧人工接手。一个具体的坑多目标并发时不要在通知里只发「抢到了」要带上 sku 和订单号否则同时跑三个目标收到三条消息根本分不清哪条对应哪场。从那以后我每接一个这类脚本都先把订单号落盘和查单优先这两件事补上再谈提速。速度是玄学幂等是纪律纪律没立住跑得越快亏得越多。希望帮到你。本文还有配套的精品资源点击获取 
长河君 · 2026/10/23 13:07:35 · 阅读 1.2k 抢回流票脚本工程化：Python轮询、订单状态机与支付幂等
简介这份面向大麦网抢票场景的 Selenium 自动化脚本适合具备一定 Python 基础、希望研究移动端抢票流程与支付闭环的开发者参考。压缩包共 3 个文件约 6KB由 1 个 Python 主脚本、1 个 Markdown 说明文档和 1 个 ini 配置文件组成分别承载核心执行逻辑、使用说明与参数配置结构精简便于快速定位关键实现。已有 981 人学习说明其在抢票自动化细分场景中具有一定关注度。读者可从中了解 Android 与 iOS 设备上的脚本运行思路、大麦网抢票与支付宝支付流程的衔接方式以及日志记录、滑块验证处理等常见环节的组织方法配置项与说明文档配合也便于按自身环境调整参数、理解模块职责并在此基础上做二次开发或排错分析。1. 抢一张回流票有多难这套「脚本-某票务-回流-支付」组合到底在解决什么某场演出开票三分钟售罄真正的机会反而不在开票那一刻——有人退票、有人付款超时系统会把票零星释放回池子这就是「回流票」。人工刷新页面根本抓不住这种几秒钟的时间窗口而这套「脚本-某票务-回流-支付」组合干的就是把「盯着回流、拼好下单参数脚本里简称 bp、拉起支付」这条链路自动化。它不是一个开箱即用的 App而是一份能跑起来的 Python 工程样本:登录态维持、票档状态轮询、下单参数组装、订单状态流转、支付回调幂等,每一块都能单独拆出来复用到其他有状态接口的自动化场景里。适合已经写过 requests、想学轮询调度和订单幂等的开发者不适合只想点两下就出票的人。2. 工程骨架目录结构、配置加载与启动链路拿到压缩包第一件事不是python main.py而是先把目录判一遍。这类脚本最容易出问题的地方从来不是算法而是「凭据写死在哪、状态存在哪、失败了能不能从中间接着跑」。先看懂骨架后面调参、加日志、换通知渠道都省事。2.1 目录结构五个包各管一件事拆开后的目录大致是下面这个形态包名做过脱敏但职责划分是这类项目的通用套路路径职责二次开发时最常动的地方config.py账号、目标商品、轮询与支付参数换票档、改轮询间隔core/session.py会话创建、Cookie/Token 装载与刷新登录态失效处理core/bp.py下单请求参数bp 层组装观演人、票档映射core/poller.py回流轮询与去重轮询节奏、触发条件core/order.py订单状态机与支付调用支付渠道、超时策略utils/日志、重试装饰器、通知加埋点、换 webhookmain.py组装依赖、启动主循环多目标并发core下的四个文件基本对应一条订单的生命周期拿到会话 → 拼参数 → 轮询等到有票 → 下单 → 拉起支付 → 等回调。utils是横切关注点日志和重试几乎每个模块都会 import。我一般会先看utils/logger.py因为脚本跑起来的第一手信息全靠它日志格式乱的项目后面排查基本靠猜。值得提醒的是这类项目通常不带数据库状态全部在内存里一个 dict 或者 dataclass。好处是部署简单坏处是进程一挂跑到一半的订单信息就没了。如果你打算长期挂着跑第一件改造就是把订单状态落盘这件事在第 6 章会展开。2.2 配置与环境一份 config 决定跑不跑得起来环境上没什么玄学Python 3.8 以上、requests、loguru或者标准库 logging就够了有的版本会带schedule或apscheduler做定时。依赖一般在一个requirements.txt里三五行装完就能 import。真正的门槛在配置。集中式配置写成 py 文件的写法# config.py CONFIG { # 账号侧登录后从浏览器开发者工具里取的会话凭据 account: { cookie: , token: , refresh_gap: 600, # 秒提前刷新登录态的间隔 }, # 目标侧想买的到底是什么 target: { item_id: , # 演出/商品 ID show_id: , # 场次 ID sku_id: , # 票档 ID buy_num: 1, # 购买张数 audience_ids: [], # 观演人 ID 列表 }, # 轮询侧 poll: { interval: 0.8, # 基础间隔秒 jitter: 0.3, # 随机抖动上限秒 max_round: 3600, # 最大轮询轮数防止无限跑 }, # 支付侧 pay: { channel: balance, # 支付渠道标识 pay_timeout: 15, # 单次支付请求超时秒 retry: 2, # 支付失败重试次数 }, notify: {webhook: }, # 结果通知地址留空则不推 }这份配置里item_id / show_id / sku_id三个 ID 决定「买什么」。它们不是随便填的必须和平台前端的请求参数一一对应抓错一个层级最常见的结果是请求返回成功但业务码报「参数错误」。audience_ids是实名观演人数量必须和buy_num对齐否则下单接口会直接拒掉。interval jitter这一对是给轮询节奏用的固定间隔容易让请求呈现明显的周期性加抖动是为了让节奏看起来更自然同时也避免多条任务在同一毫秒撞在一起。这里有个血泪经验不要把凭据写进config.py然后提交到任何仓库。我一般改成os.getenv读取本地用.env配置文件里只留os.getenv(TICKET_TOKEN, )这样的默认值。省下的不是安全审计的功夫是某天发现自己 token 被同步到公共仓库时的那一身冷汗。环境跑起来的顺序也简单装依赖 → 填 config → 先跑一次python -m core.session --check这类自检入口有的版本没写就自己在 main 里加一段确认登录态有效 → 再开主循环。跳过自检直接开跑最容易出现的现象是轮询了几百轮全是 401白等半小时。2.3 启动链路从 main 到第一个请求发出去main.py通常很短短到很多人以为它没用其实它决定了整个脚本的组装方式和退出时机。典型写法# main.py import logging from config import CONFIG from core.session import Session from core.poller import Poller from core.order import OrderFlow from utils.notify import push def main(): logging.basicConfig( levellogging.INFO, format%(asctime)s %(levelname)s %(name)s %(message)s, ) # 1. 建会话内部会装载 cookie/token并做一次可用性探测 session Session(CONFIG[account]) # 2. 轮询器只负责「发现可买的 sku」不负责下单 poller Poller(session, CONFIG[target], CONFIG[poll]) # 3. 订单流负责下单 支付依赖同一个会话 flow OrderFlow(session, CONFIG[target], CONFIG[pay]) try: for hit in poller.watch(): # 生成器yield 出可买信号 result flow.run(hit) # 下单并尝试支付 push(CONFIG[notify], result) # 无论成败都推一条 if result.ok: break # 买到即停避免重复下单 except KeyboardInterrupt: logging.info(手动中断正在退出) finally: session.close() if __name__ __main__: main()这段代码的关键在于职责边界Poller只判断「有没有票」OrderFlow只处理「有票之后怎么办」。两者通过一个生成器watch()解耦好处是轮询逻辑可以单独测——把watch()换成返回假数据的迭代器就能在不碰真实接口的情况下把下单和支付链路跑通。这是我在拆这类项目时最看重的一点能离线测的部分越多线上调试的轮次越少。session是唯一被两个模块共享的对象目的是复用连接池和登录态。很多翻车案例都源于每个模块自己requests.get()结果连接不复用、Cookie 不同步、限速也做不了。finally里调session.close()看着可有可无实际上如果脚本被 CtrlC 打断未关闭的连接池偶尔会让进程挂住不退加上这一行能省掉不少「按了 CtrlC 还得 kill -9」的烦躁。参数上要注意poller.watch()是生成器意味着轮询是惰性的for循环拿到第一个信号就开始走下单下单失败后继续回到watch()等待下一个回流。这个设计天然支持「抢到就停」和「失败重试」但你得自己保证break条件写对——少写一个if result.ok: break脚本会在买到之后继续下单超买的部分只能自己消化。3. bp 参数组装一次下单请求里到底塞了什么bp 这一层是整套脚本里最枯燥、也最容易出错的部分。它做的事情就是把「人想买的东西」翻译成「接口认识的字段」。翻译错一个字段前面所有轮询都白费所以值得单独拎出来讲。3.1 会话与登录态cookie、token 怎么在类之间流转Session类通常是薄薄一层但承担了三个职责装载凭据、探测有效性、定时刷新。参考实现# core/session.py import time import requests class Session: def __init__(self, account_cfg): self.cfg account_cfg self.s requests.Session() self.s.headers.update({ User-Agent: Mozilla/5.0 ..., # 换成你自己的浏览器 UA Referer: https://example.com/, # 平台对应页面 }) self._born time.time() self._load_auth() def _load_auth(self): # 凭据只在创建时装一次后续靠 refresh 续期 self.s.cookies.set(cookie, self.cfg[cookie]) self.token self.cfg[token] def need_refresh(self): # 距上次装载超过阈值就认为需要续期 return time.time() - self._born self.cfg[refresh_gap] def refresh(self): # 实际项目里这里走平台的续期接口拿到新 token 回填 new_token self._renew_token() if new_token: self.token new_token self._born time.time() def close(self): self.s.close()三个参数值得说refresh_gap是最容易被忽略的一个设得比平台实际过期时间短一截才安全比如会话实际 30 分钟过期阈值就设 600 秒留出重试余量。_load_auth里的 cookie 用set而不是直接塞 headers是为了让requests在后续请求里自动带上并处理更新。_born记录装载时间配合need_refresh实现惰性刷新——不是开个线程定时刷而是在每次发起请求前判断一次结构简单也不会在空闲时白白刷。3.2 bp 层参数拼装商品、场次、票档、观演人真正给下单接口用的参数在core/bp.py里组装。它的输入是 config 里的target和当前会话的 token输出是一个扁平 dict# core/bp.py import time def build_bp_params(target, token, extraNone): 把业务目标翻译成下单接口参数。extra 用于透传平台侧必填的额外字段 params { itemId: target[item_id], # 演出 ID showId: target[show_id], # 场次 ID skuId: target[sku_id], # 票档 ID buyNum: target[buy_num], # 张数 audienceIds: ,.join(target[audience_ids]), # 观演人逗号拼接 timestamp: int(time.time() * 1000), # 毫秒时间戳防重放 token: token, } if extra: params.update(extra) # 渠道、活动等可选字段 return {k: v for k, v in params.items() if v not in (None, )} # 清掉空值audienceIds用逗号拼接还是数组取决于平台接口约定脚本里常见的是字符串拼接因为很多平台的表单接口不接受嵌套结构。timestamp是毫秒级和平台侧允许的时间偏差一般只有几十秒机器时间跑偏超过阈值就会稳定报「请求过期」这一点在第 5 章会当成坑单独讲。最后一行清空值的处理看着不起眼但它能避免把extra里没填的字段塞进请求某些接口对空字符串是严格校验的。参数组装完还有一个隐含步骤签名。签名算法由平台侧决定脚本侧一般是在合法会话内复用已有的登录态和前端约定的字段顺序不自己去逆向算。真要改这块先确认你手上的账号和操作符合平台规则再谈实现。3.3 请求发送与限速连接复用、退避与失败判定参数有了发请求这一段反而是最需要工程感的。一个能长期跑的脚本和一个跑十分钟就被限流的脚本差别基本都在这里# core/http.py import time, random, requests from requests.adapters import HTTPAdapter def make_session(pool_size10): s requests.Session() adapter HTTPAdapter(pool_connectionspool_size, pool_maxsizepool_size) s.mount(https://, adapter) # 复用连接减少握手开销 return s def request_with_backoff(session, method, url, **kwargs): base 1.0 for attempt in range(4): resp session.request(method, url, timeout5, **kwargs) if resp.status_code 200: return resp if resp.status_code in (429, 503): # 明确的可重试状态 sleep base * (2 ** attempt) random.uniform(0, 0.3) time.sleep(sleep) # 指数退避 抖动 continue resp.raise_for_status() # 其余错误直接抛 raise RuntimeError(重试次数用尽)逻辑上分三层。第一层是连接池pool_maxsize设成并发数的 12 倍就够设太大会让平台侧看到一堆并行连接反而更显眼。第二层是重试策略只对 429限流和 503服务不可用做退避重试其他 4xx 直接抛因为参数错了重试一万次还是错重试只会浪费轮询预算。第三层是指数退避加抖动第一次等 1 秒、第二次 2 秒、第三次 4 秒叠上 00.3 秒随机避免多个任务在同一时刻集体重试。timeout5这个参数别省。默认不设超时的话一个卡住的连接能把整个轮询循环拖死日志上看着像「脚本没反应」实际是阻塞在网络读上。5 秒对这类接口绰绰有余真需要慢接口再单独放宽。4. 回流监控与订单状态机从「有票了」到「支付完成」轮询和支付是这条链路上时间跨度最大的两段也是最容易出「跑通一次、跑不通第二次」的地方。前者考的是节奏控制后者考的是幂等。4.1 回流轮询间隔、去重、失效判定轮询的核心不是「一直请求」而是「用最少的请求换最高的命中率」。参考实现# core/poller.py import time, random, logging log logging.getLogger(__name__) class Poller: def __init__(self, session, target, poll_cfg): self.session session self.target target self.cfg poll_cfg self.seen set() # 已处理过的 sku防重复下单 def watch(self): for rnd in range(self.cfg[max_round]): if self.session.need_refresh(): self.session.refresh() # 轮询间隙顺手续期 state self._fetch_state() if state is None: time.sleep(self._sleep()) # 请求失败也按节奏走不空转 continue if state[remain] 0 and state[sku_id] not in self.seen: log.info(命中回流 sku%s remain%s, state[sku_id], state[remain]) self.seen.add(state[sku_id]) # 先占位再交给下游 yield state time.sleep(self._sleep()) def _sleep(self): return self.cfg[interval] random.uniform(0, self.cfg[jitter]) def _fetch_state(self): try: resp self.session.s.get( https://example.com/api/sku/state, params{skuId: self.target[sku_id]}, timeout5, ) data resp.json() return data.get(data) if data.get(code) 0 else None except Exception as e: log.warning(查询票档状态失败: %s, e) return None三个设计点。seen集合在yield之前就写入这是防重入的关键——如果等到下游下单成功再标记轮询在等待期间可能又命中同一个 sku导致重复提交。_sleep()每次重新计算抖动是逐轮变化的不会形成固定节拍。_fetch_state把异常吞掉返回 None是因为轮询周期里偶发超时太正常了一次异常就退出循环等于白白浪费前面的等待但要打 warning 日志别让问题静默。max_round是兜底配合interval大致能算出最长运行时长。3600 轮 × 1 秒左右大约一小时够覆盖大部分回流高峰。真需要挂通宵把max_round调大别改成while True——没有上限的循环在出错时很难停下来。4.2 订单状态机待支付、已支付、超时关闭订单不是「下单成功就完事」它有一组明确的状态迁移。脚本里一般用一个小枚举加一个内存 dict 表达状态触发条件下一步动作CREATED下单接口返回订单号立即发起支付PAYING支付请求已发出轮询查单结果PAID查单返回已支付记录、通知、退出CLOSED超时未支付 / 平台关闭清理本地状态重新进入轮询FAILED下单或支付明确失败记录原因视情况重试状态机存在的意义是把「重试」和「重复下单」区分开。PAYING状态下重试是安全的因为订单号已经确定重试的只是支付动作CREATED状态如果因为超时重发下单请求可能生成两笔订单这时候必须靠本地订单号或者幂等键来挡。4.3 支付链路对接下单、查单、回调幂等支付这段最容易踩的是「请求超时但实际成功」。网络抖一下支付请求没拿到响应本地以为失败去重试平台侧却已经扣款。解决思路是支付前先查、支付后必查# core/order.py import time, logging log logging.getLogger(__name__) class OrderFlow: def __init__(self, session, target, pay_cfg): self.session session self.target target self.pay_cfg pay_cfg self.local_orders {} # order_no - status落盘更稳 def run(self, hit): order_no self.local_orders.get(hit[sku_id]) if order_no and self._query(order_no)[status] PAID: return Result(okTrue, order_noorder_no) # 幂等已付直接返回 order_no order_no or self._create_order(hit) # 有单号就不再重复下单 self.local_orders[hit[sku_id]] order_no for i in range(self.pay_cfg[retry] 1): self._pay(order_no) time.sleep(1) # 给平台侧落账留点时间 info self._query(order_no) if info[status] PAID: return Result(okTrue, order_noorder_no) if info[status] CLOSED: break return Result(okFalse, order_noorder_no) def _create_order(self, hit): resp self.session.s.post(https://example.com/api/order/create, json{skuId: hit[sku_id], num: self.target[buy_num]}, timeout5) return resp.json()[data][orderNo] def _pay(self, order_no): self.session.s.post(https://example.com/api/pay, json{orderNo: order_no, channel: self.pay_cfg[channel]}, timeoutself.pay_cfg[pay_timeout]) def _query(self, order_no): resp self.session.s.get(https://example.com/api/order/query, params{orderNo: order_no}, timeout5) return resp.json()[data]这里有三处幂等设计。开头先查一次如果本地记着单号且平台已显示已支付直接返回成功不再走下单。order_no order_no or self._create_order(hit)保证同一个 sku 只创建一次订单重试只重试支付。循环里每次_pay之后都_query以平台查询结果为准而不是以支付请求的返回值当结论——支付接口返回成功不等于订单已支付中间还有渠道处理。pay_timeout单独给到 15 秒比普通请求宽因为支付链路涉及的环节更多。time.sleep(1)是给平台侧落账留的缓冲不加的话查单常常抢在状态更新之前表现为「明明付了却显示待支付」然后白白多跑一轮重试。5. 避坑与排查那些跑通一次、跑不通第二次的地方下面这几条都是这类脚本里高频复现的问题。写出来不是为了吓人是因为每一条我都真实遇到过且每一条的现象都很容易被误判成「平台封了」。5.1 请求层的坑现象轮询几百轮全是 401/403重启后正常几分钟又不行。原因通常是会话续期逻辑没触发或者refresh_gap设得比平台实际过期时间长。脚本装载凭据后一直用同一个 token看起来请求都发出去了实际全被拒。解决是打开 debug 日志确认 token 装载时间把refresh_gap压到过期时间的 1/2 以下同时在请求前加一次need_refresh()判断让续期发生在使用之前而不是出错之后。现象本机跑一切正常换台机器就报「请求过期」或「签名不合法」。八成是系统时间偏移。毫秒级时间戳参与参数拼装时机器时间和标准时间差几十秒就会稳定失败。解决是先对一下时间同步状态容器环境尤其注意宿主机时间确认无误后再看参数里时间戳的单位是秒还是毫秒这一处写错在本地往往也能「偶尔成功」最难排查。现象一开始命中率不错跑久了几乎抓不到票。常见原因是轮询节奏太规律或者太密。固定间隔的请求会呈现明显的周期性间隔过短又会触发限流表现为返回 429 或者返回正常但数据是空。解决是把固定sleep换成「基础间隔 逐轮变化抖动」并把轮询轮数限制住不要开着while True挂几天。密度这件事没有通用数值只能在自己的账号权限范围内小步调调一次观察日志十分钟再继续。5.2 状态与支付的坑现象日志显示下单成功但平台上有两笔待支付订单。原因是没有在yield之前占位轮询在等待下游结果期间又命中了同一个 sku重复走了_create_order。解决是把「已处理 sku」集合的写入提前到生成器内部同时给_create_order加一层本地单号缓存有单号就不再调下单接口。现象支付请求超时重试后提示「订单已支付」本地状态却是失败。这是典型的「请求失败但业务成功」。解决是永远以查单结果为准支付前查一次支付后必查一次查单返回已支付就立刻终止重试。同时把本地订单状态落盘进程重启后还能读到PAYING的订单继续查而不是从头再来一遍。我见过最惨的情况是脚本崩了重启把同一批票重复下了三次最后只能一张张手动关单。现象明明看到票档有余量下单却返回「库存不足」。回流票的余量是瞬时值从轮询看到到真正下单中间隔了几百毫秒被别人抢走很正常。解决是不要因为一次下单失败就认为脚本坏了把「下单失败」当作正常分支处理记录原因后回到轮询继续等下一个回流。真正需要报警的是连续多次参数类错误那才是配置写错了。6. 二次开发把它改成你自己的调度与通知中台原始脚本的形态是「一个进程盯一个目标」实际用起来很快就会想扩成多目标、多账号、可观测。改造的抓手有三处按投入产出比排序。6.1 把轮询从主循环里拆出来做成任务现在main.py是一个for循环加第二个目标就得复制一份。更省事的做法是把Poller和OrderFlow组合成一个 task交给调度器统一管理# scheduler.py from concurrent.futures import ThreadPoolExecutor from config import CONFIG from core.session import Session from core.poller import Poller from core.order import OrderFlow def run_task(target_cfg, session_pool): session session_pool.get(target_cfg[account]) poller Poller(session, target_cfg, CONFIG[poll]) flow OrderFlow(session, target_cfg, CONFIG[pay]) for hit in poller.watch(): result flow.run(hit) if result.ok: return result return None def start(tasks): # 每个目标一个线程线程数等于目标数别开太大 with ThreadPoolExecutor(max_workerslen(tasks)) as pool: futures [pool.submit(run_task, t, SESSIONS) for t in tasks] for f in futures: print(f.result()) # 实际项目里走通知不打 print要点是会话复用同一个账号下的多个目标共用一个Session不同账号之间隔离。线程数不要盲目放大目标数量多于 5 个时考虑改成队列 固定 worker避免瞬间并发把账号推到限流边缘。6.2 加一层落盘和一条通知内存状态最大的问题是重启即失忆。最小成本的改造是用 sqlite 建一张表把订单号和状态写进去-- schema.sql CREATE TABLE IF NOT EXISTS orders ( order_no TEXT PRIMARY KEY, -- 平台订单号天然幂等键 sku_id TEXT NOT NULL, status TEXT NOT NULL, -- CREATED / PAYING / PAID / CLOSED updated_at INTEGER NOT NULL );order_no做主键插入用INSERT OR REPLACE重启后先按status PAYING查一遍把未完成的订单接着查完。这一层加上之后脚本崩了重启不再是灾难只是多跑几秒查询。通知那侧同理别在成功分支里直接print统一走utils/notify把成功、失败、超时三种结果都推出去——抢票这件事失败的通知比成功更有价值因为失败意味着你得赶紧人工接手。一个具体的坑多目标并发时不要在通知里只发「抢到了」要带上 sku 和订单号否则同时跑三个目标收到三条消息根本分不清哪条对应哪场。从那以后我每接一个这类脚本都先把订单号落盘和查单优先这两件事补上再谈提速。速度是玄学幂等是纪律纪律没立住跑得越快亏得越多。希望帮到你。本文还有配套的精品资源点击获取