企业官网建设流程全解析

简介：这是一份基于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 目录结构：一份能跑起来的脚本该长什么样

下面这套结构是我在多个类似项目里收敛出来的版本，核心原则是“配置与代码分离、链路与调度分离、通知与业务分离”。抢票这件事本质是一条短链路：查库存 → 拼参数 → 提交 → 判定结果，所以按链路切模块比按“工具类/帮助类”切要清晰得多。

这张表最值得说的是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 # 结构化日志，带毫秒时间戳，排查时序问题离不开

三种并发模型的取舍，我用一张表说清楚，选型时对着看就行：

绝大多数人卡在第二档就够了。抢票的并发不是“越高越好”，你开 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(encoding="utf-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(encoding="utf-8")) for c in cookies: # domain 不写时默认作用于当前域，跨子域接口需要显式指定 s.cookies.set(c["name"], c["value"], domain=c.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", timeout=5) 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, timeout=timeout) 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 False

fetch_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", json=payload, timeout=timeout) 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, level="INFO", format="<green>{time:HH:mm:ss.SSS}</green> | {message}") logger.add(log_path, level="DEBUG", rotation="10 MB", encoding="utf-8", format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level} | {message}") return logger

rotation="10 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, timeout=3) 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)

advance=0.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(maxsize=16) self.stop_flag = threading.Event() def _worker(self): while not self.stop_flag.is_set(): try: payload = self.task_q.get(timeout=0.5) except queue.Empty: continue # 真正提交的地方，异常必须吞掉，不能拖垮线程 try: status, code, msg = submit_order(self.session, self.base, payload) self.log.info(f"submit status={status} code={code} msg={msg}") if status == "success": self.stop_flag.set() except Exception as e: self.log.warning(f"submit exception: {e}") def run(self): with ThreadPoolExecutor(max_workers=self.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 任务

maxsize=16是给队列加个上限，防止探测速度远快于提交速度时内存无限增长。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", timeout=2)

5.3 现象：抢到了，但生成了两个订单

原因是多线程同时提交，服务端没认出来是同一单。解决分两层：客户端加单次提交锁，服务端靠bizNo幂等键去重。

submit_lock = threading.Lock() def safe_submit(grabber, payload): if not submit_lock.acquire(blocking=False): return "skip" # 已有线程在提交，本次直接跳过 try: return submit_order(grabber.session, grabber.base, payload) finally: submit_lock.release()

注意是blocking=False，抢不到锁就跳过，而不是排队等——排队等意味着多打一次无意义的请求。

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 True

qps设成配置里的qps_limit，capacity用默认值即可。注意这个桶是“拿不到就返回 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, extra=None) -> dict: ... @abstractmethod def classify(self, body: dict) -> tuple: ...

classify是最值得抽象的方法，因为每个平台的返回码都不一样，把映射关系关在适配器内部，上层只需要认识success、sold_out、too_frequent、not_login四种语义。

改造难易度参考这张表，按收益排序：

第一个改造我强烈建议先做，因为它和抢票成功率无关，但和你的使用体验强相关。抢到票的通知晚十分钟才看到，和没抢到区别不大。

验证改造是否成功的办法是回放：把一次真实请求的响应体存成 JSON 文件，用适配器的classify去解析，断言返回语义正确。这样不用真的开售也能测。

从那以后我每次改这类脚本，都强制先跑一遍离线回放，再上真实场次。抢票这事没有后悔药，一次配置写错就是白等一场。希望帮到你。

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