企业官网建设流程全解析

简介：这份面向大麦网抢票场景的 Selenium 自动化脚本，适合具备一定 Python 基础、希望研究移动端抢票流程与支付闭环的开发者参考。压缩包共 3 个文件，约 6KB，由 1 个 Python 主脚本、1 个 Markdown 说明文档和 1 个 ini 配置文件组成，分别承载核心执行逻辑、使用说明与参数配置，结构精简，便于快速定位关键实现。已有 981 人学习，说明其在抢票自动化细分场景中具有一定关注度。读者可从中了解 Android 与 iOS 设备上的脚本运行思路、大麦网抢票与支付宝支付流程的衔接方式，以及日志记录、滑块验证处理等常见环节的组织方法；配置项与说明文档配合，也便于按自身环境调整参数、理解模块职责，并在此基础上做二次开发或排错分析。

1. 抢一张回流票有多难：这套「脚本-某票务-回流-支付」组合到底在解决什么

某场演出开票三分钟售罄，真正的机会反而不在开票那一刻——有人退票、有人付款超时，系统会把票零星释放回池子，这就是「回流票」。人工刷新页面根本抓不住这种几秒钟的时间窗口，而这套「脚本-某票务-回流-支付」组合干的就是把「盯着回流、拼好下单参数（脚本里简称 bp）、拉起支付」这条链路自动化。它不是一个开箱即用的 App，而是一份能跑起来的 Python 工程样本:登录态维持、票档状态轮询、下单参数组装、订单状态流转、支付回调幂等,每一块都能单独拆出来复用到其他有状态接口的自动化场景里。适合已经写过 requests、想学轮询调度和订单幂等的开发者，不适合只想点两下就出票的人。

2. 工程骨架：目录结构、配置加载与启动链路

拿到压缩包第一件事不是python main.py，而是先把目录判一遍。这类脚本最容易出问题的地方从来不是算法，而是「凭据写死在哪、状态存在哪、失败了能不能从中间接着跑」。先看懂骨架，后面调参、加日志、换通知渠道都省事。

2.1 目录结构：五个包各管一件事

拆开后的目录大致是下面这个形态，包名做过脱敏，但职责划分是这类项目的通用套路：

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( level=logging.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()看着可有可无，实际上如果脚本被 Ctrl+C 打断，未关闭的连接池偶尔会让进程挂住不退，加上这一行能省掉不少「按了 Ctrl+C 还得 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, extra=None): """把业务目标翻译成下单接口参数。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_size=10): s = requests.Session() adapter = HTTPAdapter(pool_connections=pool_size, pool_maxsize=pool_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, timeout=5, **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设成并发数的 1～2 倍就够，设太大会让平台侧看到一堆并行连接，反而更显眼。第二层是重试策略，只对 429（限流）和 503（服务不可用）做退避重试，其他 4xx 直接抛，因为参数错了重试一万次还是错，重试只会浪费轮询预算。第三层是指数退避加抖动，第一次等 1 秒、第二次 2 秒、第三次 4 秒，叠上 0～0.3 秒随机，避免多个任务在同一时刻集体重试。

timeout=5这个参数别省。默认不设超时的话，一个卡住的连接能把整个轮询循环拖死，日志上看着像「脚本没反应」，实际是阻塞在网络读上。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"]}, timeout=5, ) 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 表达：

状态机存在的意义是把「重试」和「重复下单」区分开。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(ok=True, order_no=order_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(ok=True, order_no=order_no) if info["status"] == "CLOSED": break return Result(ok=False, order_no=order_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"]}, timeout=5) 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"]}, timeout=self.pay_cfg["pay_timeout"]) def _query(self, order_no): resp = self.session.s.get("https://example.com/api/order/query", params={"orderNo": order_no}, timeout=5) 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_workers=len(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 和订单号，否则同时跑三个目标，收到三条消息根本分不清哪条对应哪场。

从那以后我每接一个这类脚本，都先把订单号落盘和查单优先这两件事补上，再谈提速。速度是玄学，幂等是纪律，纪律没立住，跑得越快亏得越多。希望帮到你。

本文还有配套的精品资源，点击获取