★ FEATURED ARTICLE 简介这是一份基于Python与Selenium实现的大麦网自动抢票脚本面向具备Python 3.6基础、希望学习浏览器自动化与页面操作流程的开发者。脚本通过ChromeDriver驱动Chrome将场次、票价、实名者序号、昵称、购票数量和大麦网址等参数集中到config.json中便于按需调整优先级并校验登录状态适合作为自动化测试、网页元素定位与配置化脚本设计的练习素材。压缩包共29个文件约51KB主要包含py主程序、json参数配置、md说明文档以及sample、head、refs、objects等Git版本管理相关文件结构轻量方便快速查看与二次修改。目前已有5083人学习下载读者可从中了解Selenium脚本的初始化、参数读取、优先级选择与页面交互思路并据此复现和调试自己的自动化流程。1. 开票那一秒的手速焦虑这个 Python 抢票脚本到底解决什么问题演出开票的那一秒页面还在转圈票已经变成“缺货登记”。这套 Python 抢票脚本干的事很具体把“刷新页面—选场次—选票档—提交订单”这串手动动作换成会话复用、库存轮询、定时提交和结果通知四段程序化流程。它适合想搞清楚 HTTP 会话保持、并发调度、时间同步这几件事的开发者拿去当练手项目比看十篇教程都实在不适合想批量囤票倒卖的人脚本里的限流和单账号锁就是为了挡住这种用法。读完你能拿到一份能跑的骨架、一份能改的配置以及一堆我踩过的坑。2. 项目骨架与技术选型目录里每个文件负责哪一段链路拿到一份抢票脚本最容易犯的错是从main.py第一行开始读。这类项目真正的信息量在目录切分上——它决定了你后面改一个参数要动几个文件也决定了出问题时日志能不能定位到具体环节。我一般先看文件清单再看依赖最后才进函数体。2.1 目录结构一份能跑起来的脚本该长什么样下面这套结构是我在多个类似项目里收敛出来的版本核心原则是“配置与代码分离、链路与调度分离、通知与业务分离”。抢票这件事本质是一条短链路查库存 → 拼参数 → 提交 → 判定结果所以按链路切模块比按“工具类/帮助类”切要清晰得多。文件或目录职责二次开发时的改动频率config.yaml场次 ID、票档 ID、开售时间、并发数、限流阈值极高每次换场次都要改main.py入口装配 session、调度器、通知器低core/session.py构建 requests.Session注入 Cookie 与请求头中换账号或换设备标识时改core/probe.py探测场次与票档状态解析库存字段高接口结构一变就得改core/order.py拼装下单参数、提交、判定返回码高core/scheduler.py定时触发、线程池、限流与重试中utils/time_sync.py校准本地与服务端的时间偏差低但强烈建议保留utils/logger.py结构化日志落盘带毫秒时间戳低data/cookies.json登录态快照从浏览器导出高会过期这张表最值得说的是probe.py和order.py的拆分。很多新手版本把“查库存”和“下单”写在一个 while 循环里结果库存接口返回结构一变下单逻辑也跟着报错排查时完全没有抓手。拆开之后探测失败只影响探测下单失败只影响下单日志里一眼就能看出断在哪一段。config.yaml单独拎出来是因为换一场演出要改的东西实在太多场次 ID、票档 ID、开售时间戳、期望票数、是否自动选座。把这些硬编码在 Python 里改一次要翻五个文件。# config.yaml —— 换场次只改这个文件 item: item_id: 700000000000 # 演出/商品 ID从详情页 URL 里取 sku_id: 5000000000001 # 场次 票档组合出来的 SKU ticket_num: 1 # 购票张数多数平台限 4 张以内 seat_plan: random # random 随机 / fixed 指定座位 timing: sale_at: 2025-01-01 20:00:00 # 开售时间本地时区 advance_ms: 300 # 提前 300ms 开始自旋等待 offset_ms: 0 # 校时后写入的补偿值 concurrency: probe_interval_ms: 120 # 库存轮询间隔 submit_workers: 4 # 提交线程数别超过 8 qps_limit: 5 # 全局每秒请求上限 notify: webhook: # 自建推送服务的回调地址留空则不推送这份配置里有两个参数值得单独解释。advance_ms不是“提前多少毫秒发请求”而是“提前多少毫秒从 sleep 切换成自旋”因为time.sleep在 Windows 上的实际误差能到十几毫秒关键时刻靠不住。qps_limit是给整条链路兜底的总闸探测线程和提交线程共享这一个令牌桶避免开售瞬间把自己打成异常流量。2.2 技术选型requests 加线程池够不够什么时候才上异步抢票脚本的技术栈没有想象中复杂真正用到的就三样HTTP 客户端、并发模型、配置解析。我见过不少项目一上来就 asyncio httpx结果调试成本翻倍收益却看不出来——因为瓶颈根本不在客户端而在服务端的响应时间和你的时间精度。先把依赖摆出来这里故意不锁死版本号本地跑通后自己pip freeze固化一份更稳# requirements.txt requests # 会话保持、Cookie 复用抢票场景里比 aiohttp 更好调试 PyYAML # 配置解析 loguru # 结构化日志带毫秒时间戳排查时序问题离不开三种并发模型的取舍我用一张表说清楚选型时对着看就行模型单机理论并发调试难度适用场景纯同步 while 循环1最低单账号、单场次、只求跑通requests 线程池几十中多票档同时探测、多账号轮询httpx asyncio几百高需要同时盯几十个 SKU 的监控型任务绝大多数人卡在第二档就够了。抢票的并发不是“越高越好”你开 50 个线程去打同一场次服务端大概率直接给你返回频繁访问反而把正常请求挤掉了。线程池的意义在于“同时盯几个票档”而不是“用数量堆成功率”。我一般把submit_workers控制在 4 以内探测线程单独一个两者共享限流器。真要用异步判断标准很朴素需要长期挂机盯十几个场次的回流票用 asyncio 划算只抢一场开售票用线程池写完半小时就能跑没必要给自己找麻烦。2.3 配置驱动把场次、票档、时间抽到 YAML 之后配置读进来要立刻转成强类型对象别一路dict[item][sku_id]传下去——这种写法在参数名改一次之后会炸得到处都是。用dataclass包一层编辑器还能补全。# core/config.py from dataclasses import dataclass from pathlib import Path import yaml dataclass class ItemCfg: item_id: str sku_id: str ticket_num: int 1 seat_plan: str random dataclass class TimingCfg: sale_at: str advance_ms: int 300 offset_ms: int 0 dataclass class ConcurrencyCfg: probe_interval_ms: int 120 submit_workers: int 4 qps_limit: int 5 def load_config(path: str config.yaml) - dict: raw yaml.safe_load(Path(path).read_text(encodingutf-8)) # 只保留已知字段配置里写错 key 时能尽早暴露 return { item: ItemCfg(**raw[item]), timing: TimingCfg(**raw[timing]), concurrency: ConcurrencyCfg(**raw[concurrency]), notify: raw.get(notify, {}), }ItemCfg(**raw[item])这行的好处是YAML 里多写一个字段、少写一个必填字段启动阶段就会抛TypeError而不是等到开售前 10 秒才在某个函数深处报KeyError。抢票这种一次性任务最怕的就是“跑到一半才发现配置错”提前失败反而是最友好的设计。offset_ms单独放在timing下面是因为它由校时模块动态写回不进 YAML 也能跑但写进去之后你能从配置文件里直接看到上次校准的结果排查时间问题时省事。3. 核心链路拆解从登录态校验到下单请求的完整走向链路这一段是整个项目最容易翻车的地方。参数少一个字段、请求头差一个来源标识、返回码分类判错都会表现为“代码没报错但就是抢不到”。下面按请求实际发生的顺序拆。3.1 会话与登录态Cookie 复用比重新登录靠谱抢票脚本不处理账号密码登录正确做法是人工在浏览器里登录好导出 Cookie 存成 JSON脚本启动时灌进requests.Session。这样既不碰验证码也不至于把账号密码写进代码里。# core/session.py import json import pathlib import requests def build_session(cookie_file: str data/cookies.json) - requests.Session: s requests.Session() # 请求头尽量贴近真实浏览器缺 Referer 时部分接口会直接 403 s.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36, Referer: https://example.com/, Accept: application/json, text/plain, */*, }) cookies json.loads(pathlib.Path(cookie_file).read_text(encodingutf-8)) for c in cookies: # domain 不写时默认作用于当前域跨子域接口需要显式指定 s.cookies.set(c[name], c[value], domainc.get(domain)) return s参数说明cookie_file指向浏览器导出的 JSON 数组每条至少包含name、value两个键domain建议保留导出的原值删掉之后跨子域的接口会丢 Cookie表现为“明明登录了却返回未登录”。User-Agent和Referer不是摆设很多接口会校验来源缺了直接 403而且返回体是 HTML 而不是 JSON解析时报JSONDecodeError看起来像代码 bug其实是请求头问题。灌完 Cookie 别急着下单先打一个轻量的用户信息接口验证登录态返回未登录就立刻退出别等到开售那一刻才发现 Cookie 早就过期了。def check_login(session, base: str) - bool: # 用一个只读的用户信息接口探测成本低、无副作用 r session.get(f{base}/api/user/profile, timeout5) if r.status_code ! 200: return False return r.json().get(code) 0这个校验放在main.py启动阶段用一行if not check_login(s, BASE): raise SystemExit(Cookie 已失效请重新导出)拦住。省下来的不是代码行数是开售前 30 秒发现登录态没了的那种绝望。3.2 场次与票档探测先把 sku_id 拿到手下单接口需要的不是“演出 ID”而是“演出 场次 票档”三者组合出来的sku_id。这个值一般能从商品详情接口里取到也可以直接在页面 URL 或下单页的源码里搜。# core/probe.py import time def fetch_sku_status(session, base: str, item_id: str, timeout: float 3.0) - dict: url f{base}/api/item/{item_id}/sku r session.get(url, timeouttimeout) r.raise_for_status() data r.json() result {} for sku in data.get(data, {}).get(skuList, []): # 库存字段各平台命名不同常见是 stock / remain / inventory stock sku.get(stock) or sku.get(remain) or 0 result[sku[skuId]] { stock: int(stock), price: sku.get(price), status: sku.get(status), # 0 可售 / 1 售罄 / 2 未开售 } return result def wait_for_stock(session, base, cfg, deadline: float): # 轮询到目标 SKU 有库存或超时为止 while time.time() deadline: status fetch_sku_status(session, base, cfg.item.item_id) target status.get(cfg.item.sku_id) if target and target[stock] cfg.item.ticket_num: return True time.sleep(cfg.concurrency.probe_interval_ms / 1000) return Falsefetch_sku_status里那个or链看着糙但实用不同接口对库存字段的命名差得很远写死一个换场次就要改代码。用链式取值兜住常见命名字段名对不上时至少返回 0 而不是抛异常。wait_for_stock的deadline参数别省没有超时的轮询就是死循环脚本会一直挂着不放第二天你才发现。probe_interval_ms设成 120 是个经验值太快容易触发限流太慢会错过回流票。回流票别人取消订单释放出来的票出现的时间点很随机间隔超过 500ms 基本就抢不到了。3.3 下单请求参数拼装与返回码判定下单是整条链路唯一有副作用的动作所以这里要做两件事参数必须一次成型失败必须分类处理。参数漏一个字段服务端往往不报“缺参数”而是返回一个笼统的失败码你根本不知道错在哪。# core/order.py import json import time def build_payload(cfg, extra: dict | None None) - dict: payload { itemId: cfg.item.item_id, skuId: cfg.item.sku_id, quantity: cfg.item.ticket_num, seatPlan: cfg.item.seat_plan, # 多数下单接口要求带上幂等键重复提交时服务端据此去重 bizNo: f{cfg.item.sku_id}-{int(time.time() * 1000)}, } if extra: payload.update(extra) return payload def submit_order(session, base: str, payload: dict, timeout: float 3.0) - tuple: r session.post(f{base}/api/order/create, jsonpayload, timeouttimeout) if r.status_code ! 200: return http_error, r.status_code, r.text[:200] body r.json() code body.get(code) # 返回码按语义分类比裸判成功/失败有用得多 mapping { 0: success, # 下单成功 1001: sold_out, # 库存不足可继续轮询 1002: not_login, # 登录态失效立即停止 1003: too_frequent,# 触发限流需要退避 } return mapping.get(code, unknown), code, body.get(msg, )参数说明bizNo是幂等键用skuId 毫秒时间戳拼出来保证同一次提交重复发送时服务端能识别为同一单避免多扣款或生成重复订单。timeout别设太大开售瞬间网络抖动很正常3 秒拿不到响应就该放弃这一枪把线程让给下一次尝试。返回码分类是这段代码的核心价值。sold_out要继续轮询too_frequent要退避等待not_login要立刻终止整条链路。如果只判code 0这三种情况在你眼里都是“失败”然后你会继续用同样的节奏猛打越打越糟。3.4 结果落地日志、通知与失败快照抢票脚本跑完就退出如果不留证据事后完全是个黑匣子——抢到了不知道为什么成功没抢到也不知道断在哪一步。所以每次提交都要落一行结构化日志附上返回体的前 200 字符。# utils/logger.py import sys from loguru import logger def setup_logger(log_path: str logs/grab.log): logger.remove() # 终端输出带颜色文件输出带毫秒时间戳便于对齐服务端时间 logger.add(sys.stderr, levelINFO, formatgreen{time:HH:mm:ss.SSS}/green | {message}) logger.add(log_path, levelDEBUG, rotation10 MB, encodingutf-8, format{time:YYYY-MM-DD HH:mm:ss.SSS} | {level} | {message}) return loggerrotation10 MB防止长时间挂机把磁盘写满。日志格式里的.SSS是必须的排查“为什么晚发了 80 毫秒”这类问题时秒级精度完全不够用。通知那一段做成可插拔的notify.webhook留空就不推送填了就把下单结果 POST 过去。不要在业务代码里直接调某个具体服务换个推送渠道就要改核心逻辑这是最典型的耦合。4. 并发与时间同步把“手速”变成可控参数同一份代码有人能抢到有人抢不到差别往往不在逻辑在时间。本地时钟和服务端差 200 毫秒你的“准点 20:00:00 发请求”其实是 19:59:59.8 发的服务端直接判未开售。这一章讲的就是怎么把这点差距量出来、补回去再配合并发把命中率稳定住。4.1 校时本地时钟偏差是抢票的玄学来源不用装 NTP 客户端也能校时——只要服务端返回的响应头里有Date字段就能算出偏差。原理是记录发包前后两个时间点用往返时延的一半去修正。# utils/time_sync.py import time from email.utils import parsedate_to_datetime def measure_offset(session, base: str, rounds: int 5) - float: # 多次采样取中位数避开单次网络抖动 samples [] for _ in range(rounds): t0 time.time() r session.head(base, timeout3) t1 time.time() server_dt parsedate_to_datetime(r.headers[Date]) server_ts server_dt.timestamp() rtt t1 - t0 # 假设往返对称服务端时间戳对应的本地时刻是 t0 rtt/2 samples.append(server_ts rtt / 2 - (t0 t1) / 2) time.sleep(0.2) samples.sort() return samples[len(samples) // 2]参数说明rounds取 5 次足够多了浪费时间返回值的单位是秒正值表示本地比服务端慢加回去就行。用HEAD而不是GET是为了减少传输量响应头里的Date和请求方法无关。Date字段的精度只到秒所以单次采样误差可能有几百毫秒这也是为什么要多轮取中位数——均值会被抖动拉偏中位数更稳。拿到 offset 之后代码里所有“当前时间”都要换成time.time() offset包括定时等待和日志时间戳。这一步不做前面所有精度优化都是白费。4.2 定时触发提前多少毫秒发第一枪太关键开售时间是固定值但你的等待方式决定了实际触发时刻的精度。time.sleep(3600)睡一个小时醒来时误差可能累积到几十毫秒越接近目标时间越要切换成细粒度等待。# core/scheduler.py import time def wait_until(target_ts: float, offset: float, advance: float 0.3): # target_ts 是服务端时间戳offset 是本地相对服务端的偏差 while True: left target_ts - (time.time() offset) if left 0: return if left advance: # 距离还远粗睡省 CPU time.sleep(min(left - advance, 1.0)) else: # 进入最后 300ms让出时间片自旋把精度拉到毫秒级 time.sleep(0)advance0.3就是配置里的advance_ms。这个值的取法有点讲究太小会把 CPU 占满太大则最后一段仍然依赖 sleep 的精度。Windows 上time.sleep(0.001)实际可能睡 15ms所以最后 300ms 必须自旋。Linux 上sleep精度好一些但自旋仍然更稳。要注意自旋阶段别在里面做 I/O只用来看时间否则每次循环的耗时会变得不可预测。4.3 并发模型轮询线程和提交线程池怎么配合一个常见的错误是“探测到有库存就立刻在当前线程提交”结果探测被提交阻塞下一轮探测晚了 500 毫秒票早没了。正确做法是探测线程只管发现发现之后把任务丢进队列由线程池去提交。# core/scheduler.py续 import queue import threading from concurrent.futures import ThreadPoolExecutor class Grabber: def __init__(self, session, base, cfg, logger): self.session session self.base base self.cfg cfg self.log logger self.task_q queue.Queue(maxsize16) self.stop_flag threading.Event() def _worker(self): while not self.stop_flag.is_set(): try: payload self.task_q.get(timeout0.5) except queue.Empty: continue # 真正提交的地方异常必须吞掉不能拖垮线程 try: status, code, msg submit_order(self.session, self.base, payload) self.log.info(fsubmit status{status} code{code} msg{msg}) if status success: self.stop_flag.set() except Exception as e: self.log.warning(fsubmit exception: {e}) def run(self): with ThreadPoolExecutor(max_workersself.cfg.concurrency.submit_workers) as pool: for _ in range(self.cfg.concurrency.submit_workers): pool.submit(self._worker) wait_for_stock_and_fire(self) # 探测线程逻辑发现库存后 put 任务maxsize16是给队列加个上限防止探测速度远快于提交速度时内存无限增长。stop_flag一旦成功就置位所有 worker 在下一轮循环退出避免抢到票之后还继续猛发请求。这里没有做全局 QPS 限流实际项目里要加一个令牌桶探测和提交共享。不加的后果是探测线程太快服务端返回限流码连带着提交也被拒。4.4 重试与退避哪些错误该重试哪些必须立刻停不是所有失败都值得重试。sold_out可以继续too_frequent必须退避not_login必须停。不加区分地重试只会让你更快被封。def submit_with_backoff(grabber, payload, max_retry: int 5): delay 0.2 for i in range(max_retry): status, code, msg submit_order(grabber.session, grabber.base, payload) if status in (success, not_login): return status if status too_frequent: # 指数退避 抖动避免多个线程同一时刻集体重试 time.sleep(delay random.uniform(0, 0.1)) delay min(delay * 2, 2.0) continue # sold_out / unknown短间隔重试即可 time.sleep(0.1) return give_up加抖动这一步很容易被忽略但四个线程同时退避、同时重试效果等于没退避。random.uniform(0, 0.1)把重试时刻打散服务端看到的请求是平缓的不会一拥而上。5. 避坑与排查跑不通、抢不到、被限流的现场记录这一章是我自己踩坑踩出来的清单五条都是真实发生过的现象。每条按“现象 → 原因 → 解决”写照着对号入座就行。5.1 现象脚本不报错但返回的是登录页原因是 Cookie 过期或者请求头缺少来源标识服务端把你当成未登录用户。检查方法很简单把返回体打印出来看前 200 个字符是 HTML 基本就中了。if html in r.text[:50].lower(): raise RuntimeError(返回 HTML大概率是登录态失效或风控拦截)解决方式是重新从浏览器导出 Cookie同时确认Referer和User-Agent都带上了。如果这两样都正常还是返回登录页那多半是设备指纹变了换个浏览器重新登录导出。5.2 现象时间明明到了请求却晚发了 100 多毫秒这是多线程的锅。自旋等待放在主线程但提交任务是丢给线程池的线程池的 worker 需要从队列取任务、拼参数、建连接这一串下来几十毫秒很正常。血泪经验线程池必须在开售前就预热。我的做法是在等待阶段就跑一轮空的健康检查请求把 TCP 连接建好放回连接池而不是等到开售才第一次握手。# 预热提前建连避免开售瞬间多一次三次握手 for _ in range(cfg.concurrency.submit_workers): session.get(f{base}/api/ping, timeout2)5.3 现象抢到了但生成了两个订单原因是多线程同时提交服务端没认出来是同一单。解决分两层客户端加单次提交锁服务端靠bizNo幂等键去重。submit_lock threading.Lock() def safe_submit(grabber, payload): if not submit_lock.acquire(blockingFalse): return skip # 已有线程在提交本次直接跳过 try: return submit_order(grabber.session, grabber.base, payload) finally: submit_lock.release()注意是blockingFalse抢不到锁就跳过而不是排队等——排队等意味着多打一次无意义的请求。5.4 现象返回“请求过于频繁”之后所有请求都被拒原因就一个QPS 超了。探测间隔 120ms 看着不高但四个提交线程加上重试峰值能到每秒十几发。加一个全局令牌桶探测和提交共享额度。import threading class TokenBucket: def __init__(self, qps: float, capacity: int | None None): self.qps qps self.capacity capacity or int(qps) self.tokens float(self.capacity) self.last time.time() self.lock threading.Lock() def acquire(self): with self.lock: now time.time() self.tokens min(self.capacity, self.tokens (now - self.last) * self.qps) self.last now if self.tokens 1: return False self.tokens - 1 return Trueqps设成配置里的qps_limitcapacity用默认值即可。注意这个桶是“拿不到就返回 False”在抢票场景里正确做法是让调用方等一个很短的时间再试而不是直接放弃。5.5 现象日志时间和服务端时间对不上排查全乱套原因是日志用的是本地时间请求用的是校时后的时间两边混在一起完全没法对齐。解决很直接日志统一用校时后的时间戳或者至少把 offset 写进每条日志。from loguru import logger OFFSET 0.0 # 校时后由 main 写入 logger.add(logs/grab.log, format{time:HH:mm:ss.SSS} | off f{OFFSET:.3f} | {{message}})这条坑看起来是小事但它会毁掉你所有的事后分析——你会以为请求发晚了其实是日志记早了。6. 二次开发把抢票脚本改成选座与推送中间件跑通之后再往下改最值钱的不是“抢得更快”而是把它从一次性脚本变成一个能复用的中间件。我自己改过三版最后一版把探测、提交、通知全部插件化换场次只改 YAML换平台只实现一个适配器类。抽象方式是把“平台差异”收进一个Adapter接口fetch_sku、build_payload、classify三个方法。新平台接进来只需要实现这三个方法调度、限流、重试、日志都不用动。# adapters/base.py from abc import ABC, abstractmethod class Adapter(ABC): abstractmethod def fetch_sku(self, session, cfg) - dict: ... abstractmethod def build_payload(self, cfg, extraNone) - dict: ... abstractmethod def classify(self, body: dict) - tuple: ...classify是最值得抽象的方法因为每个平台的返回码都不一样把映射关系关在适配器内部上层只需要认识success、sold_out、too_frequent、not_login四种语义。改造难易度参考这张表按收益排序改造点难度实际收益通知插件化Webhook / 桌面弹窗低高抢到能第一时间知道配置支持多 SKU 轮流探测中高热门场次的备选票档很有用座位图解析与指定座位中高中取决于平台是否开放选座接口本地 Web 控制台高中调参直观但开发成本大第一个改造我强烈建议先做因为它和抢票成功率无关但和你的使用体验强相关。抢到票的通知晚十分钟才看到和没抢到区别不大。验证改造是否成功的办法是回放把一次真实请求的响应体存成 JSON 文件用适配器的classify去解析断言返回语义正确。这样不用真的开售也能测。从那以后我每次改这类脚本都强制先跑一遍离线回放再上真实场次。抢票这事没有后悔药一次配置写错就是白等一场。希望帮到你。本文还有配套的精品资源点击获取
阅读完成 · 觉得有帮助？
