简介:这份资源面向具备一定Python基础、希望深入理解网络编程与网页自动化的开发者,以京东618抢券场景为切入点,演示如何用脚本完成请求发送、页面元素定位、模拟登录与定时触发等操作。压缩包共8个文件,以3个py脚本和3张png截图为主,辅以txt与md说明文档,整体约485KB,体量轻便,便于快速阅读与本地运行。目前已有3894人学习下载,说明该案例在爬虫与自动化练习中具有一定参考价值。读者可从中获得一套可运行的抢券工具骨架,理解requests与Selenium在真实页面中的配合方式,并借鉴多线程并发、异常捕获与日志记录的组织思路;配套截图与说明文档也有助于对照代码梳理执行流程,适合作为网络编程练手项目或自动化脚本的入门参考。
1. 抢券这件事,脚本比手速靠谱
618 大促那几天,我盯着手机屏幕刷新,券在眼前一闪就没了,手指点下去永远是「已抢光」。后来我干脆不拼手速了,用 Python 写了个抢券脚本,把「盯时间、发请求、判结果、重试」这套动作交给程序跑。这个项目就是围绕京东 618 抢券场景做的一个网络编程实战 Demo:用 requests 维持会话、用多线程卡点并发、用定时器对齐开抢时刻,核心不在「抢到多少」,而在把 HTTP 会话、并发控制、异常重试这几件事讲透。适合已经会写 Python 基础语法、想拿一个真实场景练网络编程的开发者,也适合想拆解抢购类工具架构的人。下面按「能干什么 → 怎么跑 → 怎么改」的顺序拆。
2. 项目定位与技术选型:为什么是 requests + 多线程
2.1 这个工具到底解决什么问题
抢券的本质是一次带登录态的 HTTP 请求,在极短的时间窗口内重复发起,直到服务端返回成功。人工操作的问题有三个:一是反应延迟,从看到按钮到点下去至少几百毫秒;二是无法精确卡点,服务器时间和你手机时间有偏差;三是失败后不会自动重试,一次没中就结束了。脚本要解决的就是这三点——提前把请求准备好,在开抢前几十毫秒开始循环发送,拿到成功响应就停。
项目本身不涉及复杂的业务逻辑,价值在于把「会话保持 + 并发 + 定时 + 重试」这条链路写清楚。很多新手写爬虫只会requests.get(url),一到需要登录态、需要并发、需要处理风控就卡住,这个项目正好是那块练手石。
2.2 技术栈与依赖清单
整个项目是纯 Python 实现,不依赖浏览器自动化,走的是接口层请求的路子。核心依赖如下:
| 依赖 | 作用 | 说明 |
|---|---|---|
| requests | 发起 HTTP 请求、维持 Session | 会话对象自动携带 Cookie |
| threading | 多线程并发抢券 | 标准库,无需安装 |
| time / datetime | 定时卡点、请求间隔控制 | 标准库 |
| json | 解析接口返回、构造请求体 | 标准库 |
| logging | 记录每次请求结果,便于排查 | 标准库,比 print 好维护 |
安装只需要一条命令:
pip install requests其余全是标准库,不用额外装。这一点对新手友好——不用配环境、不用装浏览器驱动,clone 下来改个配置就能跑。选 requests 而不是 Selenium 的原因也在这:接口请求比模拟点击快一个数量级,且不依赖页面结构,页面改版不影响脚本(接口改版才影响)。代价是需要自己抓包分析请求,这部分后面会讲。
2.3 目录结构长什么样
项目按职责拆成几个文件,结构不复杂但边界清晰:
jd_coupon_demo/ ├── config.py # 配置:目标接口、Cookie、并发数、开抢时间 ├── session_builder.py # 构建带登录态的 Session,设置请求头 ├── coupon_grabber.py # 核心抢券逻辑:单次请求 + 结果判定 ├── scheduler.py # 定时卡点 + 多线程调度 ├── logger.py # 日志封装 └── main.py # 入口,串起上面几个模块这种拆法的好处是改一处不影响全局:换目标券只改config.py,调并发只动scheduler.py,排查问题看logger.py输出的日志。我见过不少人把所有逻辑塞进一个main.py,两百行之后自己都找不到哪行是发请求的,二次开发成本极高。
2.4 一次抢券请求的完整链路
在写代码之前先把链路想清楚,后面每一段代码都对应链路里的一环:
- 用 Cookie 构建 Session,设置 User-Agent、Referer 等请求头,让请求看起来像正常浏览器发的;
- 从配置里读出目标券的接口地址和请求参数(券 ID、活动 ID 等);
- 主线程计算距离开抢时刻还有多少秒,睡到开抢前 N 毫秒;
- 唤醒多个工作线程,每个线程循环发请求,直到拿到成功标志或达到最大重试次数;
- 每次请求的结果写日志,成功的线程置一个全局标志,其他线程看到标志就退出;
- 主线程汇总结果,打印成功/失败。
这条链路里最容易翻车的是第 1 步和第 4 步——请求头不全被风控拦,或者线程之间没协调好导致重复请求把账号打限流。后面避坑章节会细说。
3. 核心模块拆解:从 Session 构建到并发调度
3.1 构建带登录态的 Session
抢券接口几乎都要求登录态,而登录态体现在 Cookie 里。最直接的做法是从浏览器开发者工具里复制 Cookie 字符串,解析成字典塞进 Session。这里有个细节:Cookie 里往往有分号分隔的多个键值对,直接整个字符串塞进 headers 也能用,但用requests.utils.cookiejar_from_dict转成 CookieJar 更规范,后续想单独改某个 Cookie 也方便。
import requests from requests.utils import cookiejar_from_dict def build_session(cookie_str: str, user_agent: str) -> requests.Session: """根据 Cookie 字符串构建带登录态的 Session""" session = requests.Session() # 把 "k1=v1; k2=v2" 解析成字典 cookie_dict = {} for item in cookie_str.split(";"): item = item.strip() if "=" in item: k, v = item.split("=", 1) cookie_dict[k] = v session.cookies = cookiejar_from_dict(cookie_dict) # 请求头尽量贴近真实浏览器,缺 Referer 容易被判异常 session.headers.update({ "User-Agent": user_agent, "Referer": "https://example.com/activity", # 换成实际活动页 "Accept": "application/json, text/plain, */*", "Accept-Language": "zh-CN,zh;q=0.9", }) return sessioncookie_str是从浏览器复制的原始字符串,user_agent建议直接用你登录时那个浏览器的 UA,别随便编一个。cookiejar_from_dict把普通字典转成 requests 认识的 CookieJar,Session 发请求时会自动带上。请求头里Referer是最容易被忽略的一项——很多接口会校验来源页,缺了直接返回「请求非法」,这个坑我踩过不止一次。
3.2 单次抢券请求与结果判定
抢券接口的返回通常是 JSON,里面有个字段标识成功与否,比如code为 0 表示成功、其他值表示失败原因(已抢光、未开始、频繁等)。判定逻辑要区分「可重试」和「不可重试」:已抢光可以继续试(可能有人退单释放库存),未开始说明时间没到要等,频繁说明被限流要退避。
import time import logging def grab_once(session: requests.Session, url: str, payload: dict) -> dict: """发一次抢券请求,返回结构化结果""" try: resp = session.post(url, data=payload, timeout=3) data = resp.json() except Exception as e: # 网络异常、超时、非 JSON 返回都归到这里 logging.warning("request failed: %s", e) return {"ok": False, "retry": True, "msg": str(e)} code = data.get("code") if code == 0: return {"ok": True, "retry": False, "msg": "success", "raw": data} if code in (1001, 1002): # 已抢光 / 库存不足,可继续试 return {"ok": False, "retry": True, "msg": data.get("msg")} if code == 2001: # 请求频繁,需要退避 time.sleep(0.5) return {"ok": False, "retry": True, "msg": "rate limited"} # 其余视为不可重试(如未登录、活动结束) return {"ok": False, "retry": False, "msg": data.get("msg")}timeout=3是必须的,不设超时的话一个卡住的请求会把线程挂死。返回结构里retry字段决定上层要不要继续循环,这个设计让重试策略和请求逻辑解耦——以后想加新的错误码,只改这个函数就行。code的具体数值是示例,实际要按你抓包看到的返回改,别照抄。
3.3 定时卡点与多线程调度
卡点是抢券的灵魂。思路是主线程先算出距离目标时刻的秒数,睡到目标前 200 毫秒,然后启动 N 个线程同时开抢。线程之间用一个threading.Event协调,谁先成功谁 set 这个事件,其他线程检测到就退出,避免无谓请求把账号打限流。
import threading import time from datetime import datetime def run_grab(session, url, payload, start_at: datetime, threads: int = 5, max_retry: int = 50): success_flag = threading.Event() lock = threading.Lock() results = [] def worker(idx): for _ in range(max_retry): if success_flag.is_set(): return r = grab_once(session, url, payload) with lock: results.append(r) if r["ok"]: success_flag.set() logging.info("thread %d got it", idx) return if not r["retry"]: return time.sleep(0.05) # 小间隔,避免瞬间打爆接口 # 睡到开抢前 200ms delta = (start_at - datetime.now()).total_seconds() - 0.2 if delta > 0: time.sleep(delta) ts = [threading.Thread(target=worker, args=(i,)) for i in range(threads)] for t in ts: t.start() for t in ts: t.join() return success_flag.is_set(), resultsstart_at是开抢时刻,用服务器时间校准过的datetime。threads=5是并发数,不是越大越好——太大容易触发风控,太小又抢不过别人,一般 3 到 8 之间试。max_retry=50是单线程最大尝试次数,配合time.sleep(0.05)大约覆盖几秒的窗口。success_flag是全局停止信号,lock保护results列表的并发写入。这里用Event而不是布尔变量,是因为Event的is_set和set是线程安全的,省得自己加锁。
3.4 入口与配置分离
main.py只做三件事:读配置、建 Session、调调度。配置单独放config.py,改参数不用碰逻辑代码。
# config.py COOKIE = "pt_key=xxx; pt_pin=yyy; ..." # 从浏览器复制 USER_AGENT = "Mozilla/5.0 (...) Chrome/120.0" TARGET_URL = "https://example.com/api/coupon/receive" PAYLOAD = {"couponId": "123456", "activityId": "789"} START_AT = "2024-06-18 00:00:00" THREADS = 5 MAX_RETRY = 50# main.py from datetime import datetime from config import COOKIE, USER_AGENT, TARGET_URL, PAYLOAD, START_AT, THREADS, MAX_RETRY from session_builder import build_session from scheduler import run_grab if __name__ == "__main__": session = build_session(COOKIE, USER_AGENT) start_at = datetime.strptime(START_AT, "%Y-%m-%d %H:%M:%S") ok, results = run_grab(session, TARGET_URL, PAYLOAD, start_at, THREADS, MAX_RETRY) print("抢券结果:", "成功" if ok else "失败", "请求次数:", len(results))配置和逻辑分离之后,换一张券只需要改PAYLOAD里的couponId,换账号只改COOKIE。START_AT用字符串存、入口处再转datetime,是为了让配置文件保持纯文本、方便版本管理。这套结构跑通之后,你会发现大部分抢购类需求都能套进去,只是接口和参数不同。
4. 跑起来:环境准备与首次运行
4.1 环境与依赖安装
Python 版本建议 3.8 以上,requests用最新稳定版即可。虚拟环境不是必须,但强烈建议建一个,避免和系统里其他项目的依赖打架:
python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install requests装完用python -c "import requests; print(requests.__version__)"验证一下,能打印版本号就说明环境没问题。这一步看着简单,但新手最容易在「pip 装到了另一个 Python」上翻车,所以验证一下不吃亏。
4.2 抓包拿到真实的接口和参数
项目里的TARGET_URL和PAYLOAD是占位示例,必须换成你自己抓到的真实值。做法是:浏览器打开活动页,按 F12 进 Network 面板,手动点一次领券按钮,找到那条领券请求,看它的 Request URL、请求方法(GET 还是 POST)、请求体(Form Data 或 JSON)、以及请求头里的 Cookie 和 Referer。把这些抄进config.py对应字段。
注意两点:一是请求体格式要和代码里一致,代码用的是data=payload(表单),如果实际是 JSON 就要改成json=payload;二是 Cookie 有时效,过期了要重新复制,脚本里没法自动续期,这是这类工具的天然限制。
4.3 首次运行与结果观察
配置改好后直接跑:
python main.py如果开抢时间还没到,程序会先睡到开抢前 200 毫秒再启动线程,期间终端没有输出是正常的。跑完会打印「抢券结果」和「请求次数」。第一次建议把START_AT设成当前时间往后一分钟,THREADS设成 1,先验证请求能不能通、返回码是不是预期的,别一上来就 5 线程对着正式活动打——万一参数错了,等于白白给账号刷了一堆异常请求。
观察日志时重点看三样:返回的code是不是你预期的成功码、有没有出现「请求频繁」、请求耗时是否正常。如果全是「请求非法」,八成是请求头或 Cookie 的问题,回到 4.2 重新核对。
5. 避坑与排查:那些让我返工的细节
5.1 现象:所有请求都返回「请求非法」
原因通常是请求头不全或 Cookie 失效。接口会校验Referer、User-Agent,有的还校验Origin和自定义头(比如某些x-开头的签名头)。解决方法是把浏览器里那条真实请求的请求头完整复制过来,逐个对比,缺哪个补哪个。Cookie 过期也会表现为类似错误,重新登录复制一份即可。
5.2 现象:跑着跑着账号被限流,后续请求全失败
原因是并发太高或请求间隔太短,触发了服务端的频率限制。解决方法是把THREADS降到 3 以内,time.sleep(0.05)调到 0.1 到 0.2,并在遇到「频繁」返回码时做指数退避(第一次等 0.5 秒,第二次 1 秒,以此类推)。抢券不是请求越多越好,被限流之后连正常请求都发不出去,反而更亏。
5.3 现象:时间到了但请求发早了,返回「活动未开始」
原因是本地时间和服务器时间有偏差,或者你算的「提前 200 毫秒」太保守。解决办法是先发一个轻量请求(比如活动详情接口),从响应头Date字段读出服务器时间,算出和本地的差值,卡点时用校准后的时间。另外提前量别设太大,200 毫秒是经验值,网络差的可以调到 300 到 500 毫秒。
5.4 现象:多线程下日志错乱、结果列表长度对不上
原因是多个线程同时写results或同时 print,没有加锁。解决方法是所有共享数据的读写都包在with lock:里,日志用logging模块而不是print,logging内部有锁,多线程下输出不会串行错乱。这个坑不致命但很烦,排查问题时日志乱了等于没有日志。
5.5 现象:脚本在本地跑得好,换台机器就抢不到
原因是两台机器的系统时间、网络延迟不同,卡点时刻实际偏差了几百毫秒。解决办法是每台机器跑之前都做一次时间校准(同 5.3),并且把提前量按实测网络 RTT 调整。这类工具对时间敏感,玄学成分有,但大部分「抢不到」都能归到时间没对齐上。
6. 二次开发:把 Demo 改成你自己的抢购工具
这个项目的骨架是通用的,换场景只需要动三处。第一处是config.py里的接口和参数,任何抢购类需求都是「找到接口 → 抄参数」;第二处是grab_once里的结果判定,不同平台的返回码不一样,把成功码和可重试码对上就行;第三处是调度策略,如果目标有多个券要抢,可以把run_grab改成接收一个任务列表,每个任务独立线程组,用优先级排序。
改造时我建议加两个东西。一个是请求耗时统计,在grab_once里记录time.time()差值,跑完打印平均耗时和 P95,能直观看出网络状况,也方便调并发数。另一个是结果落盘,把每次请求的返回写进 CSV,事后分析哪些时间点成功率高。代码不复杂:
import csv, time def log_result(path, idx, r, cost): with open(path, "a", newline="", encoding="utf-8") as f: writer = csv.writer(f) writer.writerow([time.strftime("%H:%M:%S"), idx, r["ok"], r["msg"], f"{cost:.3f}"])验证改造是否有效,别拿正式活动试,自己搭个本地 mock 接口(Flask 十行就能写一个),模拟「前 2 秒返回已抢光、之后返回成功」,看脚本能不能在预期时间内命中。这套验证方法我后来每次改并发逻辑都强制走一遍,比对着线上活动瞎试靠谱得多。
最后说个血泪教训:这类工具最大的风险不是代码写错,而是把账号打限流甚至触发风控,所以并发和间隔一定要保守,先小流量验证再放大。我现在的习惯是任何抢购脚本上线前,先用 1 线程跑通全链路、确认返回码正确,再逐步加并发,从不一上来就拉满。希望帮到你。
本文还有配套的精品资源,点击获取