本站资源均为网友、站长推荐收集整理而来，请勿商业运营，仅供学习和研究，请在下载后24小时内删除！！
快手世界杯好礼自动抢兑脚本完整教程 | Bidwarp
⚠️ 学术研究！请勿自用或非法用途！
快手世界杯好礼 自动抢兑 脚本完整教程
Bidwarp2026-07-31自动化 / 技术教程
目录
01 前言：活动背景与脚本目标
快手极速版在 2026 年世界杯期间推出了「夏日足球盛宴」好礼兑换活动（活动代号 wc26）。用户通过日常任务积攒金豆，在每天 12:00 / 18:00 / 21:00 三个时段抢兑实物奖品——从维达湿巾到 iPhone 17 Pro Max，库存极度稀缺（热门商品仅 1 件）。
手动点击根本抢不到。本教程记录了从零开始逆向 API、搭建自动抢兑脚本、经过多轮迭代到 v8 版本的完整技术过程，最终实现毫秒级精准提交，首响延迟从 100ms+ 优化到 55-78ms。
02 活动机制分析
活动基本规则
- 活动代号：wc26（2026 世界杯主题）
- 活动域名：
s1-hb1azN-idc-ppg.viviv.com（N 为机房编号，不同账号分配不同） - 活动页面：H5 WebView，路径
/doodle/{activityToken}.html - 每日场次：12:00 / 18:00 / 21:00 三个时段，每场 2 小时
- 货币：金豆（通过签到、答题、看视频等任务获取）
- 兑换规则：每个商品有固定金豆价格和库存，先到先得
商品层级示例
| 商品名称 | 金豆价格 | 库存 | 竞争激烈度 |
|---|---|---|---|
| iPhone 17 Pro Max 1TB | 1,180,000 | 1 | 🔥🔥🔥🔥🔥 |
| 大疆 Osmo Action 6 | 299,800 | 1 | 🔥🔥🔥🔥 |
| 海信电视 E5Q 75英寸 | 600,000 | 1-2 | 🔥🔥🔥🔥 |
| 小米高速吹风机 | 18,600 | 2 | 🔥🔥🔥 |
| 快手定制球衣 | 6,000 | 3 | 🔥🔥 |
| 维达湿巾 272 片 | 5,320 | 18 | 🔥 |
| 迪卡侬运动毛巾 | 2,990 | 35 | 🔥 |
💡 关键发现 库存 1 件的热门商品，从开抢到售空通常只需 200-400ms。这意味着脚本必须在开抢瞬间的前几百毫秒内完成提交，才有机会命中。
03 API 逆向分析
抓包工具与环境
- 工具：Reqable（安卓端 HTTPS 抓包）
- 目标：快手极速版 APP 内的 H5 活动页面
- 关键：快手极速版未做 SSL Pinning（H5 页面走系统 WebView），可直接抓包
三个核心 API
| 接口路径 | 方法 | 作用 | 请求体 |
|---|---|---|---|
/rest/doodle/wc26/exchange/query-multi-rule | POST | 查询所有场次商品列表 | {} |
/rest/doodle/wc26/coin/balance | POST | 查询金豆余额 | {} |
/rest/doodle/wc26/exchange/do-exchange | POST | 提交兑换（核心） | {"activityToken":"...","exchangeKey":"...","exchangeRuleId":...} |
兑换提交接口详解
HTTPdo-exchange 请求结构
POST /rest/doodle/wc26/exchange/do-exchange HTTP/1.1
Host: s1-hb1azN-idc-ppg.viviv.com
Content-Type: application/json
jimu-token: {activityToken}
Cookie: kuaishou.api_st={...}; kuaishou.h5_st={...}; userId={...}; did={...}; ...
{
"activityToken": "sQbgvQvb",
"exchangeKey": "E_1782279244206_883648",
"exchangeRuleId": 1519
}请求参数说明
| 参数 | 说明 | 获取方式 |
|---|---|---|
activityToken | 活动标识，整个活动期间固定 | HAR 抓包 / URL 路径 |
exchangeKey | 场次标识，每场不同 | query-multi-rule 接口返回 |
exchangeRuleId | 商品 ID（整数） | query-multi-rule 接口返回 |
认证方式
Cookie 中的以下字段是核心鉴权参数，缺一不可：
| Cookie 字段 | 说明 | 是否设备相关 |
|---|---|---|
kuaishou.api_st | API 认证令牌（账号级） | ❌ 账号绑定 |
kuaishou.h5_st | H5 页面令牌 | ❌ 账号绑定 |
userId | 用户 ID | ❌ 账号绑定 |
did | 设备 ID | ✅ 设备绑定 |
egid | 设备全局 ID | ✅ 设备绑定 |
odid / rdid | 设备原始 ID | ✅ 设备绑定 |
kpn | 渠道标识 | ✅ KUAISHOU_SPEED=极速版 / NEBULA=苹果端 |
⚠️ 设备指纹校验 v4 版本发现快手升级了 do-exchange 接口风控，新增设备指纹校验。仅传认证 Cookie 会返回 6001（配置错误）。必须从 HAR 抓包中提取完整设备 Cookie（did、egid、odid、rdid、sig、client_key 等约 40 个字段），每个账号绑定独立设备指纹。
兑换结果码（exchangeRes）
| 值 | 含义 | 脚本处理 |
|---|---|---|
1 | 兑换成功 | 🎉 停止射击，推送通知 |
4 | 活动太火爆（限流） | 继续重试，限流率 >90% 则停止 |
6000 | 未处理 | 继续重试 |
6001 | 配置错误 | ⚠️ 设备 Cookie 可能失效 |
6002 | 未开始 | 正常（开抢前的盲射会收到这个） |
6003 | 已结束 | 停止 |
6101 | 库存不足（秒没） | 连续 3 次则停止 |
6201 | 超出兑换限制 | 停止 |
6203 | 金豆不足 | 停止 |
6301 | 兑换发放中（也算成功） | 🎉 停止射击 |
6302 | 反作弊审核中（也算成功） | 🎉 停止射击 |
6303 | 反作弊拒绝 | 停止 |
商品状态码（status）
| 值 | 含义 |
|---|---|
1 | 可兑换 |
2 | 已抢完 |
3 | 已兑换 |
4 | 未开始 |
5 | 审核中 |
6 | 金豆不足 |
04 脚本架构与版本迭代
脚本经历了 8 个主要版本迭代，每个版本解决上一版暴露的核心问题：
v1
同步 requests + 3 并发
最简版，同步 HTTP 请求，3 线程并发。首响 100ms+ 偏慢。
v2
命令行传参 + 多账号
支持命令行传入场次和商品 ID，无需改代码即可换目标。加入多账号支持。
v3
异步 asyncio + 盲射策略
切换到 asyncio 异步并发，引入「盲射」概念——不提前查询库存，直接提交 do-exchange。云端延迟 50ms。
v4
设备指纹修复
快手升级风控，do-exchange 返回 6001。从 HAR 提取完整设备 Cookie，每账号绑定独立设备指纹。修复后恢复正常。
v5
时间校准 + 分阶段射击
引入 NTP 式时间校准（用 balance 接口的时间戳计算服务器偏移），分阶段：保温→预热→脉冲→追加。
v6
HTTP/2 多路复用 + 极速版通道
切换 httpx HTTP/2，单连接多路复用 100 并发。kpn 切 KUAISHOU_SPEED 走极速版流量池，限流远松于主站。
v7
反限流优化
轻量预热（10 发建立连接）+ balance 校准（避免 do-exchange 触发限流）+ 限流率 90% 自动停止。
v8
最终版：极速版通道 + 全优化
综合所有优化：HTTP/2 + 极速版通道 + 时间校准 + 分阶段盲射 + 限流对抗 + 自动依赖安装 + 开抢前刷新场次。最终稳定版本。
射击时序图
T-180s 进入预热阶段
同步查询余额 + 场次信息↓T-5s 盲射保温开始
balance 接口 10-20发/秒
保持 HTTP/2 连接热↓T-1s 切换 do-exchange 预热
少量探路请求（2-3发）
预期返回 6002（未开始）↓T=0 正点脉冲！
100 发 HTTP/2 齐射↓T+0~5s 随机节奏追加
每发间隔 20-60ms 随机
直到成功或库存空
05 关键技术点详解
1. HTTP/2 单连接多路复用
传统 HTTP/1.1 每个并发请求需要独立 TCP 连接，建连开销大。HTTP/2 在单个 TCP 连接上多路复用，100 个并发请求共享一个连接，大幅降低延迟和资源消耗。
Python
async with httpx.AsyncClient(
http2=True, # 启用 HTTP/2
verify=False,
timeout=httpx.Timeout(connect=5, read=5, write=5, pool=5),
limits=httpx.Limits(
max_connections=140,
max_keepalive_connections=70,
keepalive_expiry=30,
),
) as client:
# 100 个请求共享单个 HTTP/2 连接
await asyncio.gather(*[client.post(url, ...) for _ in range(100)])2. 时间校准（NTP 式）
本地时间与快手服务器可能有几十毫秒偏差。通过 balance 接口返回的 timestamp 字段计算偏移量，校准开抢时机。
Python
# 5 轮校准，去极值取中位数
for i in range(5):
t0 = time.time()
resp = await client.post(BALANCE_URL, ...)
t1 = time.time()
server_ts = resp.json()["timestamp"] / 1000
local_mid = (t0 + t1) / 2
offset = server_ts - local_mid # 服务器时间 - 本地时间
offsets.append(offset)
# 去掉最高最低，取中位数
offsets.sort()
trimmed = offsets[1:-1]
final_offset = trimmed[len(trimmed) // 2]3. 盲射策略
不提前查询库存（query-multi-rule），直接在开抢前几秒开始提交 do-exchange。提前提交会返回 6002（未开始），但请求已经到达服务器，连接已建立，开抢瞬间无需额外建连延迟。
🔑 核心思路 盲射的本质是用请求换时间——提前发射的请求虽然会收到 6002，但 HTTP/2 连接已经热好，开抢瞬间的请求无需 TCP/TLS 握手，直接复用已有连接发送。
4. 极速版通道（kpn=KUAISHOU_SPEED）
快手主站 DAU 约 3.9 亿，极速版仅 1.3 亿。通过设置 Cookie 中 kpn=KUAISHOU_SPEED，请求走极速版流量池，竞争更少，限流更松。
5. 限流对抗
- 分阶段降频：保温阶段用 balance 接口（低风险），最后 1 秒才切 do-exchange
- 随机节奏：追加阶段每发间隔 20-60ms 随机，避免规律性请求被识别
- 自动熔断：限流率超过 90% 且已打 50+ 发时自动停止，避免无效轰炸
- accept-encoding: identity：不接受压缩响应，避免 gzip/br 解码开销
6. 预序列化优化
JSON 请求体在循环前序列化一次，每发请求复用同一份 bytes，避免重复序列化开销。Cookie 字符串同理预构建。
Python
# 循环前序列化，每发复用
exchange_payload = json.dumps({
"activityToken": ACTIVITY_TOKEN,
"exchangeKey": exchange_key,
"exchangeRuleId": rule_id,
}).encode("utf-8")
# 预构建 Cookie header
fire_headers["Cookie"] = build_cookie_string(account)
# 射击时直接传 bytes，零序列化开销
await client.post(EXCHANGE_URL, content=exchange_payload, headers=fire_headers)06 完整脚本代码（公开版）
以下代码已清除所有个人 token、userId、设备指纹、推送密钥等敏感信息，替换为占位注释。填入自己的抓包数据后即可运行。
⚠️ 使用前必读 运行前必须通过抓包获取自己的 Cookie 数据，替换所有 # 替换为你的... 标记处。直接运行不替换会报 6001 配置错误。
Pythonks_seckill.py — 完整公开版
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
快手极速版 世界杯好礼自动抢兑脚本（公开版）
功能：定时精准抢兑快手世界杯活动限量商品
依赖：Python 3.8+, httpx[http2], requests
用法：python3 ks_seckill.py [场次名] [商品ID]
"""
import os
for _k in list(os.environ):
if 'proxy' in _k.lower():
del os.environ[_k]
import asyncio
import time
import json
import sys
from datetime import datetime, timezone, timedelta
# ===== 自动安装缺失依赖 =====
def _ensure_deps():
missing = []
try:
import httpx
except ImportError:
missing.append("httpx[http2]")
try:
import requests
except ImportError:
missing.append("requests")
if missing:
import subprocess
for pkg in missing:
print(f"[DEP] 安装缺失依赖: {pkg}")
subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", pkg],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
_ensure_deps()
import httpx
import requests
# ============ 配置区 ============
ACTIVITY_TOKEN = "sQbgvQvb" # 活动标识（从 HAR 抓包获取，整个活动期间固定）
TARGET_SESSION = "0719-1" # 目标场次名（从 query-multi-rule 获取）
TARGET_RULE_ID = 1951 # 目标商品 ID（从 query-multi-rule 获取）
# 射击参数
CONCURRENT = 100 # HTTP/2 并发数
PRE_START_SECONDS = 180 # 开抢前多少秒进入预热阶段
PRE_FIRE_MS = 5000 # 提前多少毫秒开始盲射保温
BLIND_WARMUP_PER_SEC = 20 # 保温阶段：balance 接口 每秒发数
BLIND_PREHEAT_SEC = 1.0 # 最后多少秒切换 do-exchange 预热
BLIND_PREHEAT_PER_SEC = 30 # 预热阶段：do-exchange 每秒发数
WAVE_SIZE = 100 # 正点脉冲一波多少发
SUSTAINED_SEC = 5 # 持续追加多少秒
RANDOM_DELAY_MIN_MS = 20 # 随机间隔最小值
RANDOM_DELAY_MAX_MS = 60 # 随机间隔最大值
# 活动域名（从 HAR 抓包获取，不同账号可能分配不同机房）
# 替换为你的 HAR 中实际看到的域名
BASE_URL = "https://s1-hb1azN-idc-ppg.viviv.com"
BALANCE_URL = f"{BASE_URL}/rest/doodle/wc26/coin/balance"
EXCHANGE_URL = f"{BASE_URL}/rest/doodle/wc26/exchange/do-exchange"
RULES_URL = f"{BASE_URL}/rest/doodle/wc26/exchange/query-multi-rule"
# ============ 设备指纹 Cookie ============
# ⚠️ 替换为你的 HAR 抓包数据！以下为占位示例。
# 抓包方法：Reqable 抓快手极速版，打开世界杯活动页面，找到任意请求的 Cookie 头
DEVICE_COOKIES = {
"odid": "YOUR_ODID", # 替换为你的设备 ODID
"kpf": "ANDROID_PHONE", # 安卓手机=ANDROID_PHONE，iPhone=IPHONE
"appver": "14.5.50.48908", # 替换为你的 APP 版本号
"os": "android", # 或 "26.5"（iOS 版本号）
"c": "XIAOMI", # 安卓厂商，iPhone 填 "a"
"kpn": "KUAISHOU_SPEED", # 极速版=KUAISHOU_SPEED，苹果端=NEBULA
"did": "YOUR_DID", # 替换为你的设备 DID
"egid": "YOUR_EGID", # 替换为你的设备 EGID
"odid": "YOUR_ODID", # 替换为你的设备 ODID
"rdid": "YOUR_RDID", # 替换为你的设备 RDID
"client_key": "YOUR_CLIENT_KEY", # 替换为你的 client_key
"mod": "Xiaomi%28M2104K10AC%29", # 替换为你的手机型号
"sys": "ANDROID_13", # 替换为你的系统版本
"sh": "2400", # 屏幕高度
"sw": "1080", # 屏幕宽度
"language": "zh-cn",
"country_code": "CN",
"net": "WIFI",
# ... 其余设备字段按 HAR 原样填写
}
# ============ 请求头 ============
EXTRA_HEADERS = {
# 替换为你的 HAR 中实际 User-Agent
"User-Agent": "kwai-android aegon/4.50.1",
"X-Requested-With": "XMLHttpRequest",
"Origin": BASE_URL,
"Referer": f"{BASE_URL}/doodle/{ACTIVITY_TOKEN}.html",
"Accept": "application/json, text/plain, */*",
"yohost": "ppg.viviv.com",
"zyck": "jimu_next.worldcup",
"accept-encoding": "identity", # 不接受压缩，避免解码开销
"accept-language": "zh-CN,zh-Hans;q=0.9",
}
# ============ 账号配置 ============
# ⚠️ 替换为你的抓包数据！
ACCOUNTS = [
{
"name": "我的账号",
"api_st": "YOUR_API_ST", # 替换：kuaishou.api_st 的值
"target_rule_id": TARGET_RULE_ID,
"extra_cookies": {
"kuaishou.h5_st": "YOUR_H5_ST", # 替换：kuaishou.h5_st 的值
"userId": "YOUR_USER_ID", # 替换：你的 userId
"token": "YOUR_TOKEN", # 替换：token 字段
"sid": "YOUR_SID", # 替换：sid 字段
"client_key": "YOUR_CLIENT_KEY",
},
},
]
# ============ 推送通知（可选）============
# 替换为你的 PushPlus token，或注释掉推送功能
PUSHPLUS_TOKEN = "YOUR_PUSHPLUS_TOKEN"
# 命令行传参覆盖
if len(sys.argv) > 1:
TARGET_SESSION = sys.argv[1]
if len(sys.argv) > 2:
TARGET_RULE_ID = int(sys.argv[2])
ACCOUNTS[0]["target_rule_id"] = TARGET_RULE_ID
HEADERS = {
"Content-Type": "application/json",
"jimu-token": ACTIVITY_TOKEN,
}
# ============ 结果码对照 ============
EXCHANGE_RES = {
1: "兑换成功", 4: "活动太火爆(限流)",
6000: "未处理", 6001: "配置错误", 6002: "未开始", 6003: "已结束",
6004: "已下线", 6101: "库存不足(秒没)", 6102: "超出库存周期限制",
6104: "目标已下线", 6201: "超出兑换限制", 6202: "已有审核记录",
6203: "金豆不足", 6204: "超出来源限制", 6205: "超出目标限制",
6301: "兑换发放中(成功)", 6302: "反作弊审核中(成功)", 6303: "反作弊拒绝",
}
def push_notify(title, content):
"""PushPlus 微信推送通知"""
try:
resp = requests.post("http://www.pushplus.plus/send", json={
"token": PUSHPLUS_TOKEN,
"title": title,
"content": content,
}, timeout=10)
result = resp.json()
print(f"[PUSH] 推送结果: code={result.get('code')} msg={result.get('msg')}")
except Exception as e:
print(f"[PUSH] 推送失败: {e}")
def build_cookie_string(account):
"""构建完整 Cookie header 字符串"""
dev_cookies = account.get("device_cookies") or DEVICE_COOKIES
parts = [f"kuaishou.api_st={account['api_st']}"]
for k, v in dev_cookies.items():
parts.append(f"{k}={v}")
if account.get("extra_cookies"):
for k, v in account["extra_cookies"].items():
parts.append(f"{k}={v}")
return "; ".join(parts)
def build_sync_session(api_st, account=None):
"""同步 requests Session（用于余额查询、场次查询等低延迟场景）"""
s = requests.Session()
s.cookies.set("kuaishou.api_st", api_st)
dev_cookies = account.get("device_cookies") if account and account.get("device_cookies") else DEVICE_COOKIES
for k, v in dev_cookies.items():
s.cookies.set(k, v)
if account and account.get("extra_cookies"):
for k, v in account["extra_cookies"].items():
s.cookies.set(k, v)
s.headers.update(HEADERS)
s.headers.update(EXTRA_HEADERS)
adapter = requests.adapters.HTTPAdapter(pool_connections=50, pool_maxsize=50, max_retries=0)
s.mount("https://", adapter)
s.mount("http://", adapter)
return s
def query_rules(session):
"""查询所有场次商品列表"""
try:
resp = session.post(RULES_URL, json={}, timeout=10)
data = resp.json()
if data.get("result") == 1:
return data["data"]
except Exception as e:
print(f"[ERROR] query_rules: {e}")
return None
def find_target_session(session_data):
"""从场次列表中找到目标场次"""
for s in session_data["sessions"]:
if s["sessionName"] == TARGET_SESSION:
start_time = s["startTime"] / 1000
exchange_key = s["exchangeKey"]
target_item = None
for rule in s["rules"]:
if rule["id"] == TARGET_RULE_ID or rule["id"] == ACCOUNTS[0].get("target_rule_id", 0):
target_item = rule
break
return {"exchangeKey": exchange_key, "startTime": start_time, "item": target_item}
return None
def check_balance(session):
"""查询金豆余额"""
try:
resp = session.post(BALANCE_URL, json={}, timeout=10)
data = resp.json()
if data.get("result") == 1:
return data["data"]
except:
pass
return None
async def calibrate_time_h2(client, fire_headers, rounds=5):
"""时间校准：用 balance 接口的时间戳计算服务器偏移"""
balance_payload = json.dumps({}).encode("utf-8")
offsets = []
for i in range(rounds):
try:
t0 = time.time()
resp = await client.post(BALANCE_URL, content=balance_payload, headers=fire_headers)
t1 = time.time()
data = resp.json()
if "timestamp" in data:
server_ts = data["timestamp"] / 1000
local_mid = (t0 + t1) / 2
rtt = (t1 - t0) * 1000
offset = server_ts - local_mid
offsets.append(offset)
print(f" 校准第{i+1}轮: 偏移={offset*1000:.0f}ms RTT={rtt:.0f}ms")
except Exception as e:
print(f" 校准第{i+1}轮异常: {e}")
await asyncio.sleep(0.3)
if offsets:
offsets.sort()
trimmed = offsets[1:-1] if len(offsets) >= 3 else offsets
median_offset = trimmed[len(trimmed) // 2]
print(f"[校准] 去极值中位数={median_offset*1000:.0f}ms")
return median_offset
return 0
# ============ HTTP/2 异步射击核心 ============
class FireStats:
"""射击统计与状态管理"""
def __init__(self):
self.lock = asyncio.Lock()
self.total = 0
self.code_6002 = 0 # 未开始
self.code_4 = 0 # 限流
self.code_6101 = 0 # 库存不足
self.code_success = 0 # 成功
self.code_other = 0 # 其他
self.success_data = None
self.stop = False
self.first_response_ms = None
self.error_count = 0
async def record(self, res_code, data, latency, name, shot_id):
async with self.lock:
self.total += 1
if self.first_response_ms is None and res_code not in (6002,):
self.first_response_ms = latency
if res_code in (1, 6301, 6302):
self.code_success += 1
msg = EXCHANGE_RES.get(res_code, "")
print(f" ★★★ 第{shot_id}发命中！{msg} ({latency:.0f}ms) ★★★")
self.success_data = {"account": name, "data": data}
self.stop = True
elif res_code == 6002:
self.code_6002 += 1
if self.code_6002 <= 3:
print(f" [{name}] 未开始 ({latency:.0f}ms)")
elif res_code == 4:
self.code_4 += 1
if self.code_4 <= 5:
print(f" [{name}] 限流 ({latency:.0f}ms)")
# 限流率 > 90% 且已打 50+ 发，提前终止
if self.total >= 50 and self.code_4 / self.total > 0.9:
print(f" ⚠️ 限流率{self.code_4}/{self.total}={self.code_4/self.total*100:.0f}% > 90%，提前终止")
self.stop = True
elif res_code == 6101:
self.code_6101 += 1
if self.code_6101 <= 3:
print(f" [{name}] 库存不足 ({latency:.0f}ms)")
if self.code_6101 >= 3:
self.stop = True
elif res_code == 6001:
self.code_other += 1
if self.code_other <= 3:
print(f" [{name}] 配置错误 ({latency:.0f}ms)")
if self.code_other >= 5:
self.stop = True
else:
self.code_other += 1
msg = EXCHANGE_RES.get(res_code, "未知")
print(f" [{name}] {res_code}-{msg} ({latency:.0f}ms)")
if res_code in (6003, 6201, 6202, 6203, 6303):
self.stop = True
async def async_fire(client, url, content_bytes, headers, stats, name, shot_id):
"""单次异步射击"""
if stats.stop:
return
try:
t0 = time.time()
resp = await client.post(url, content=content_bytes, headers=headers)
latency = (time.time() - t0) * 1000
data = resp.json()
result = data.get("result")
if result == 1 and isinstance(data.get("data"), dict):
res_code = data["data"].get("exchangeRes")
else:
res_code = result
await stats.record(res_code, data, latency, name, shot_id)
except asyncio.CancelledError:
return
except Exception as e:
async with stats.lock:
stats.error_count += 1
if stats.error_count <= 3:
print(f" [{name}] 网络异常: {type(e).__name__}")
async def async_fire_wave(client, url, content_bytes, headers, stats, name, wave_id, count):
"""一波齐射"""
if stats.stop:
return
tasks = []
for i in range(count):
shot_id = f"{wave_id}-{i+1}"
tasks.append(async_fire(client, url, content_bytes, headers, stats, name, shot_id))
await asyncio.gather(*tasks, return_exceptions=True)
async def run_http2_burst(account, exchange_key, rule_id, start_time):
"""HTTP/2 异步爆破主函数"""
name = account["name"]
# 预序列化 JSON body（只编一次，每发复用）
exchange_payload = json.dumps({
"activityToken": ACTIVITY_TOKEN,
"exchangeKey": exchange_key,
"exchangeRuleId": rule_id,
}).encode("utf-8")
balance_payload = json.dumps({}).encode("utf-8")
# 预构建完整 headers（含 Cookie，跳过 cookie jar 开销）
fire_headers = dict(EXTRA_HEADERS)
fire_headers.update(HEADERS)
fire_headers["Content-Type"] = "application/json"
fire_headers["Cookie"] = build_cookie_string(account)
stats = FireStats()
async with httpx.AsyncClient(
http2=True,
verify=False,
timeout=httpx.Timeout(connect=5, read=5, write=5, pool=5),
limits=httpx.Limits(max_connections=140, max_keepalive_connections=70, keepalive_expiry=30),
) as client:
# ===== 预热 HTTP/2 连接 =====
print(f"[{name}] 预热 HTTP/2 连接（10发）...")
warmup_tasks = [client.post(BALANCE_URL, content=balance_payload, headers=fire_headers) for _ in range(10)]
warmup_results = await asyncio.gather(*warmup_tasks, return_exceptions=True)
warmup_ok = sum(1 for r in warmup_results if isinstance(r, httpx.Response) and r.status_code == 200)
print(f"[{name}] 连接预热完成: {warmup_ok}/10 成功")
# ===== 时间校准 =====
print(f"\n[{name}] 时间校准...")
time_offset = await calibrate_time_h2(client, fire_headers, rounds=5)
print(f"[校准] 最终偏移: {time_offset*1000:.0f}ms")
# ===== 阶段1a: 盲射保温（balance 接口）=====
warmup_start = start_time - (PRE_FIRE_MS / 1000)
calibrated_now = time.time() + time_offset
remaining = warmup_start - calibrated_now
if remaining > 0:
print(f"[{name}] 等待盲射时间: {remaining:.1f}秒")
await asyncio.sleep(max(0, remaining - 0.5))
while time.time() + time_offset < warmup_start:
await asyncio.sleep(0.001)
print(f"[{name}] ★ 盲射保温阶段开始！(balance, {BLIND_WARMUP_PER_SEC}发/秒)")
blind_warmup = 0
interval = 1.0 / BLIND_WARMUP_PER_SEC
preheat_time = start_time - BLIND_PREHEAT_SEC
while time.time() + time_offset < preheat_time:
if stats.stop:
break
blind_warmup += 1
asyncio.create_task(client.post(BALANCE_URL, content=balance_payload, headers=fire_headers))
await asyncio.sleep(interval)
print(f"[{name}] 保温完成: {blind_warmup}发 balance")
# ===== 阶段1b: do-exchange 预热 =====
print(f"[{name}] ★ 切换 do-exchange 预热！({BLIND_PREHEAT_PER_SEC}发/秒)")
blind_preheat = 0
interval = 1.0 / BLIND_PREHEAT_PER_SEC
while time.time() + time_offset < start_time - 0.1:
if stats.stop:
break
blind_preheat += 1
asyncio.create_task(async_fire(client, EXCHANGE_URL, exchange_payload, fire_headers, stats, name, f"预热-{blind_preheat*2-1}"))
asyncio.create_task(async_fire(client, EXCHANGE_URL, exchange_payload, fire_headers, stats, name, f"预热-{blind_preheat*2}"))
await asyncio.sleep(interval)
print(f"[{name}] 预热完成: {blind_preheat*2}发 do-exchange")
# ===== 阶段2: 正点脉冲 =====
calibrated_now = time.time() + time_offset
wait = start_time - calibrated_now
if wait > 0:
await asyncio.sleep(max(0, wait - 0.005))
while time.time() + time_offset < start_time:
await asyncio.sleep(0.0005)
if not stats.stop:
ts = time.strftime("%H:%M:%S", time.localtime())
ms = int(time.time() * 1000) % 1000
print(f"[{ts}.{ms:03d}][{name}] ★★★ 正点脉冲！({WAVE_SIZE}发 HTTP/2)")
await async_fire_wave(client, EXCHANGE_URL, exchange_payload, fire_headers, stats, name, "脉冲", WAVE_SIZE)
# ===== 阶段3: 随机节奏追加 =====
import random
sustained_start = time.time()
sustained_shot = 0
print(f"[{name}] ★ 随机节奏追加（间隔{RANDOM_DELAY_MIN_MS}-{RANDOM_DELAY_MAX_MS}ms）")
while not stats.stop and (time.time() - sustained_start) < SUSTAINED_SEC:
sustained_shot += 1
delay = random.uniform(RANDOM_DELAY_MIN_MS / 1000, RANDOM_DELAY_MAX_MS / 1000)
await async_fire(client, EXCHANGE_URL, exchange_payload, fire_headers, stats, name, f"续-{sustained_shot}")
await asyncio.sleep(delay)
return stats
def main():
bj_tz = timezone(timedelta(hours=8))
print("=" * 50)
print(f"快手极速版 自动抢兑脚本（公开版）")
print(f"目标场次: {TARGET_SESSION} | 商品ID: {TARGET_RULE_ID}")
print("=" * 50)
# 1. 检查余额
for acc in ACCOUNTS:
s = build_sync_session(acc["api_st"], acc)
balance = check_balance(s)
if balance is not None:
print(f"[{acc['name']}] 金豆余额: {balance}")
else:
print(f"[{acc['name']}] ⚠️ Cookie 可能过期！请重新抓包")
push_notify(f"Cookie过期", f"账号{acc['name']}的Cookie可能过期，请重新抓包")
# 2. 获取场次信息
print(f"\n[INFO] 获取 {TARGET_SESSION} 场次信息...")
session_data = query_rules(s)
if not session_data:
print("[ERROR] 无法获取商品列表，检查 Cookie 是否有效")
return
for sess in session_data["sessions"]:
start_dt = datetime.fromtimestamp(sess["startTime"]/1000, tz=bj_tz)
print(f"\n场次: {sess['sessionName']} ({start_dt.strftime('%m/%d %H:%M')})")
for r in sess.get("rules", []):
prize = r.get('exchangePrizeList', [{}])[0]
print(f" ID={r['id']} {prize.get('rewardName','')} 金豆={r.get('costGoldBean','')} 库存={prize.get('stockTotal','')}")
target = find_target_session(session_data)
if not target:
print(f"[ERROR] 找不到场次 {TARGET_SESSION}")
return
exchange_key = target["exchangeKey"]
start_time = target["startTime"]
print(f"\n[INFO] exchangeKey: {exchange_key}")
print(f"[INFO] 开抢时间: {datetime.fromtimestamp(start_time, tz=bj_tz).strftime('%Y-%m-%d %H:%M:%S')}")
# 3. 验证接口
print(f"\n[验证] 测试 do-exchange 接口...")
payload = {
"activityToken": ACTIVITY_TOKEN,
"exchangeKey": exchange_key,
"exchangeRuleId": ACCOUNTS[0].get("target_rule_id", TARGET_RULE_ID),
}
try:
resp = s.post(EXCHANGE_URL, json=payload, timeout=5)
data = resp.json()
result = data.get("result")
res_code = data["data"].get("exchangeRes") if result == 1 and isinstance(data.get("data"), dict) else result
if res_code == 6001:
print(f"[ERROR] ❌ 返回6001配置错误！设备 Cookie 可能失效！")
return
elif res_code == 6002:
print(f"[验证] ✅ 接口正常返回6002(未开始)")
else:
print(f"[验证] 返回: {res_code} - {EXCHANGE_RES.get(res_code, '未知')}")
except Exception as e:
print(f"[验证] 异常: {e}")
# 4. 等待开抢前 3 分钟
wait = start_time - time.time() - PRE_START_SECONDS
if wait > 0:
print(f"\n[INFO] 距开抢 {wait:.0f}秒，等待中...")
time.sleep(wait)
# 5. 开抢前重新确认 exchangeKey
print(f"\n[PREP] 重新确认场次...")
session_data = query_rules(s)
if not session_data:
print(f"[ERROR] 刷新场次失败！中止运行")
return
target = find_target_session(session_data)
if not target:
print(f"[ERROR] 刷新后找不到场次，中止运行")
return
new_key = target["exchangeKey"]
if new_key != exchange_key:
print(f"[WARN] exchangeKey 变了! -> {new_key}")
exchange_key = new_key
# 6. HTTP/2 爆破
print(f"\n{'='*50}")
print(f"★ HTTP/2 抢兑启动！并发={CONCURRENT} 脉冲{WAVE_SIZE}发+随机追加")
print(f"{'='*50}")
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
stats = loop.run_until_complete(
run_http2_burst(ACCOUNTS[0], exchange_key, ACCOUNTS[0]["target_rule_id"], start_time)
)
loop.close()
# 汇总
print(f"\n--- 结果 ---")
print(f"总提交: {stats.total}次 | 未开始: {stats.code_6002} | 限流: {stats.code_4} | 库存不足: {stats.code_6101} | 其他: {stats.code_other}")
if stats.first_response_ms:
print(f"首响延迟: {stats.first_response_ms:.0f}ms")
if stats.success_data:
print(f"★ 兑换成功！命中账号: {stats.success_data['account']}")
push_notify("快手抢到了！", f"兑换成功！详情: {json.dumps(stats.success_data['data'], ensure_ascii=False)}")
else:
print("未能抢到")
push_notify("快手没抢到", f"总提交{stats.total}次，未开始{stats.code_6002}/限流{stats.code_4}/库存不足{stats.code_6101}")
if __name__ == "__main__":
main()07 配置与部署
环境要求
| 项目 | 要求 |
|---|---|
| Python | 3.8 及以上 |
| 依赖 | httpx[http2]、requests（脚本自动安装） |
| 运行环境 | 任意能访问快手服务器的 Linux/macOS/Windows |
| 网络要求 | 能访问 *.viviv.com，延迟越低越好 |
抓包获取 Cookie
1安装抓包工具
安卓手机安装 Reqable，配置 HTTPS 证书（系统证书，非用户证书）。
2打开快手极速版
进入世界杯好礼兑换页面，随便操作（刷新、查看规则等）。
3导出 HAR
在 Reqable 中找到 viviv.com 域名的请求，导出 HAR 文件。
4提取关键参数
从 HAR 中提取以下内容，填入脚本配置区：
BASE_URL：请求的 HostACTIVITY_TOKEN：URL 路径中的 doodle/{token}api_st：Cookie 中的kuaishou.api_sth5_st：Cookie 中的kuaishou.h5_stuserId：Cookie 中的userId- 全部设备 Cookie：did、egid、odid、rdid、sig、client_key 等约 40 个字段
User-Agent：请求头中的 UA
⚠️ Cookie 时效性api_st 和 h5_st 有时效，通常 1-3 天过期。每次抢购前务必重新抓包确认 Cookie 仍然有效。脚本启动时会自动检查余额验证 Cookie。
运行方式
bash
# 基本运行（使用脚本内默认场次和商品）
python3 ks_seckill.py
# 指定场次和商品
python3 ks_seckill.py 0719-1 1951
# 后台运行（适合配合定时任务）
nohup python3 ks_seckill.py 0719-1 1951 > grab.log 2>&1 &
# 配合 crontab 定时启动（开抢前3分钟）
# 例：12:00 开抢，11:57 启动
57 11 * * * cd /path/to/script && python3 ks_seckill.py 0719-1 1951 >> grab.log 2>&1参数调优建议
| 参数 | 保守 | 标准 | 激进 |
|---|---|---|---|
| CONCURRENT | 30 | 100 | 150 |
| PRE_FIRE_MS | 3000 | 5000 | 8000 |
| WAVE_SIZE | 50 | 100 | 150 |
| SUSTAINED_SEC | 3 | 5 | 8 |
| 触发限流概率 | 低 | 中 | 高 |
08 注意事项与免责声明
技术注意事项
- Cookie 时效：
api_st通常 1-3 天过期，每次抢购前重新抓包 - 活动时效：活动结束后所有接口失效，脚本无法使用
- 参数变化：快手可能随时调整接口参数或风控策略，需重新抓包分析
- 单脚本运行：严禁两个脚本实例同时运行，会导致请求翻倍触发限流
- 依赖安装：脚本首次运行会自动安装 httpx 和 requests，需联网
- 代理干扰：脚本开头会清除所有代理环境变量，避免代理增加延迟
法律风险评估
基于公开信息检索（2026 年 7 月），结论如下：
- 全网无快手起诉普通个人抢购脚本用户的案例
- 被法律追究的均为卖脚本牟利的开发者和商业刷量公司
- 普通用户单账号自用仅违反用户协议，最高后果为封号，不构成违法
- 脚本调用的是公开 H5 接口，请求行为与手动点击一致
- 单账号 300-400 次请求不会对服务器造成实质影响
⚠️ 免责声明 本教程所有内容仅供学术研究、技术学习与交流使用。快手及相关商标版权归北京快手科技有限公司所有。使用者需对自身行为承担全部法律责任，本站不承担任何连带责任。请遵守相关平台的服务条款。
📞 交流合作 如需项目投稿、技术合作等，请联系站长邮箱：3996980822@qq.com
Bidwarp · 资源整合站 · 01.21r.cn
本站所有内容仅供学术研究与技术交流，严禁用于任何非法用途
NOTICE
3996980822@qq.com | Copyright © 2023~2026
01
DISCLAIMER
本站资源大多来自网络，如有侵犯权益请联系管理员，我们会第一时间审核删除。站内资源仅供学习测试，未经许可禁止商用，请在24小时内删除。
02
VIP PRIVILEGE
遇到付费内容？升级终身VIP即可全站免费畅享所有资源，可以联系我的微信进行开通。
© 版权声明 All Rights Reserved
© 版权声明
文章版权归作者所有，未经允许请勿转载。
THE END
暂无评论内容