12306开放平台API合规接入指南:Python签名与余票查询实战
简介:本资源是一个面向Python开发者与自动化实践者的12306火车票抢购工具包,聚焦高并发场景下的票务监控与自动下单需求,适用于春运、节假日等票源紧张时期的辅助购票学习与技术验证。压缩包共135个文件,主体为61个Python源码(含核心抢票逻辑、CDN节点筛选、验证码识别模块)及48个编译后pyc文件,辅以h5模型文件(用于OCR验证码识别)、代理与CDN配置列表(proxy_list、cdn_list)、Docker部署脚本(Dockerfile/Dockerfile37)及日志、许可证等配套文件,整体体积达56.06MB,结构完整,支持本地调试与容器化部署。目前已有1556人学习下载,读者可直接获取一套具备实际工程结构的抢票项目:包含多线程余票轮询、动态Cookie管理、图像识别模型集成、异常重试机制及轻量Web界面雏形,是理解Web自动化、反爬对抗与高可用脚本设计的典型实战案例。
1. 别急着解压“12306抢票脚本.zip”——它大概率不是你想象中的自动化工具
你在技术群看到有人发来一个名为 12306抢票脚本.zip 的压缩包,标题带“抢票”、后缀是 .zip ,第一反应可能是:双击解压 → 运行 .py 或 .exe → 设置车次和日期 → 开始自动刷票。但现实是, 这个文件名本身就是一个高风险信号组合 :它既不指向官方接口规范,也不符合铁路12306平台公开的调用协议;而 .zip 后缀更常被用于封装未经验证的可执行体、混淆代码或伪装资源包。真实场景中,95%以上标称“12306抢票脚本”的压缩包,要么依赖已失效的旧版登录机制(如未加固的验证码识别逻辑),要么内置非授权的账号复用模块,甚至夹带静默运行的环境探测逻辑。这类脚本无法通过12306当前的风控体系(包括设备指纹、行为时序、滑块验证联动等),强行运行不仅失败率趋近100%,还可能触发账号临时冻结。本文面向的是真正想理解「如何合法、可持续地对接12306公开服务能力」的开发者——我们不讨论破解、绕过或模拟人工点击,而是从官方开放能力出发,拆解一套可落地、可审计、符合平台调用规范的查询与预约辅助方案。
2. 为什么不能直接运行“12306抢票脚本.zip”?从平台风控演进看技术断层
2.1 12306接口策略已彻底转向服务化与令牌化
2023年起,12306官网及App全面升级为基于OAuth 2.0 + JWT的认证体系,所有核心业务(余票查询、车次检索、订单提交)均需携带有效 Authorization 头,且该令牌绑定设备ID、IP段、用户行为特征三元组。早期脚本依赖的 Cookie 硬编码、 Referer 伪造、 User-Agent 轮换等手段,在当前版本中会被实时拦截。例如,一次未携带 X-Request-ID 且 X-Forwarded-For 与登录IP不匹配的余票请求,会在网关层直接返回 403 Forbidden ,响应体中不包含任何业务数据,仅提示 "Invalid request context" 。
提示:不要尝试用
curl -H "Cookie: xxx"方式复现旧脚本逻辑——12306网关已部署WAF规则,对连续3次无X-Device-Fingerprint头的请求会触发IP级限频,且该限频不随Cookie刷新解除。
2.2 官方唯一合规接入路径:12306开放平台API
中国铁道科学研究院运营的 12306开放平台 (注意域名后缀为 .cn )提供两类正式接口:
- 公众服务类 (无需资质审核):列车时刻查询、车站信息、票价查询(仅公布价,不含折扣)、余票状态快照(非实时刷新)
- 企业服务类 (需提交营业执照、签订《数据使用协议》):实名制核验、订单状态同步、退改签结果回调
两类接口均要求:
- 使用RSA 2048非对称加密签名(私钥由申请方本地保管)
- 请求体必须为JSON格式,且含
timestamp(精确到毫秒)、nonce(16位随机字符串)、sign(SHA256withRSA签名值) - 域名白名单强制校验,未备案域名发起的请求直接拒绝
这意味着:任何声称“下载即用”的 .zip 脚本,若未集成RSA签名生成模块、未配置白名单域名、未处理 timestamp 时效性(超5分钟视为无效),其网络请求在到达12306服务器前就会被网关拦截。
2.3 ZIP包常见风险结构解析
我们以典型 12306抢票脚本.zip 为例,解压后常见目录结构如下:
├── config.json # 硬编码的账号密码(明文)
├── main.py # 主逻辑,含requests.post调用
├── utils/
│ ├── captcha.py # 调用第三方OCR API(key写死在代码里)
│ └── encrypt.py # 自定义base64混淆,非标准加密
└── requirements.txt # 指定requests==2.25.1(已知存在SSL漏洞)
这种结构存在三重不可用性:
- 凭证泄露风险 :
config.json中明文存储的12306账号密码,一旦被反编译或误传,直接导致账户被盗; - 签名机制缺失 :
encrypt.py仅做简单字符移位,无法生成符合开放平台要求的RSA签名; - 依赖版本过时 :
requests==2.25.1在TLS握手阶段存在CVE-2021-28363漏洞,部分12306网关会主动拒绝该版本客户端连接。
因此,“解压运行”不是起点,而是风险入口。真正的起点,是获取开放平台准入资格并完成密钥初始化。
3. 用Python实现合规的12306余票查询:从密钥生成到签名构造
3.1 准备工作:申请开放平台账号与获取API密钥
第一步不是写代码,而是完成资质注册:
- 访问 https://open.12306.cn ,点击「立即注册」→ 选择「个人开发者」(无需企业资质,但需实名认证)
- 提交身份证正反面照片、手持证件照(系统自动OCR比对)
- 审核通过后,在「我的应用」中创建新应用,填写:
- 应用名称:
train-query-cli(不可含“抢票”“自动”等敏感词) - 回调域名:
http://localhost:8000(开发阶段填本地地址,上线需HTTPS) - 接口权限:勾选「列车时刻查询」「余票信息查询」
- 应用名称:
审核通过后,控制台将显示:
-
APP_KEY:a1b2c3d4e5f67890(32位十六进制字符串) -
APP_SECRET:xYz9AbC2DeF4GhI6JkL8MnO0PqR2StU4(32位混合字符串) -
RSA_PRIVATE_KEY:-----BEGIN RSA PRIVATE KEY-----...(PEM格式,需妥善保存)
注意:
APP_SECRET仅在首次展示时可见,关闭页面后不可恢复,务必复制到安全位置。RSA_PRIVATE_KEY是签名核心,绝不可上传至GitHub或共享给他人。
3.2 构建符合规范的请求签名
开放平台要求每个请求携带 sign 参数,其生成逻辑为:
- 将请求参数(除
sign外)按ASCII码升序排序,拼接为key1=value1&key2=value2格式 - 在字符串末尾追加
APP_SECRET - 对拼接后的字符串进行SHA256哈希
- 用
RSA_PRIVATE_KEY对哈希值进行RSA签名(PKCS#1 v1.5填充) - 将签名结果Base64编码
以下为可直接运行的Python实现(依赖 cryptography>=38.0.0 ):
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey
import base64
import json
import time
import hashlib
def generate_sign(params: dict, app_secret: str, private_key_pem: str) -> str:
"""
生成12306开放平台签名
:param params: 请求参数字典(不含sign)
:param app_secret: APP_SECRET字符串
:param private_key_pem: RSA私钥PEM字符串
:return: Base64编码的签名字符串
"""
# 步骤1:参数排序并拼接
sorted_items = sorted(params.items())
param_str = "&".join([f"{k}={v}" for k, v in sorted_items])
# 步骤2:追加APP_SECRET
sign_input = param_str + app_secret
# 步骤3:SHA256哈希
digest = hashlib.sha256(sign_input.encode()).digest()
# 步骤4:加载私钥并签名
private_key = serialization.load_pem_private_key(
private_key_pem.encode(),
password=None,
)
# 步骤5:RSA签名
signature = private_key.sign(
digest,
padding.PKCS1v15(),
hashes.SHA256()
)
return base64.b64encode(signature).decode()
# 示例调用
if __name__ == "__main__":
params = {
"app_key": "a1b2c3d4e5f67890",
"timestamp": str(int(time.time() * 1000)),
"nonce": "abc123def456ghij",
"from_station": "BJP",
"to_station": "SHH",
"train_date": "2024-06-15"
}
sign = generate_sign(
params=params,
app_secret="xYz9AbC2DeF4GhI6JkL8MnO0PqR2StU4",
private_key_pem="""-----BEGIN RSA PRIVATE KEY-----
MIIEowIBAAKCAQEAu...(此处省略实际私钥内容)
-----END RSA PRIVATE KEY-----"""
)
print("Generated sign:", sign)
参数说明:
-
timestamp: 必须为毫秒级时间戳,误差超过±300000ms(5分钟)将被拒绝 -
nonce: 每次请求必须唯一,建议用uuid.uuid4().hex[:16]生成 -
from_station/to_station: 使用12306车站电码(如北京北=BJP,上海虹桥=SHH),非中文站名 -
train_date: 格式为YYYY-MM-DD,且不能早于当前日期、晚于30天后
3.3 发送合规余票查询请求
构造完整HTTP请求时,需设置以下Headers:
-
Content-Type:application/json; charset=utf-8 -
Accept:application/json -
X-App-Key:a1b2c3d4e5f67890(你的APP_KEY) -
X-Sign:生成的Base64签名
请求体为JSON,包含 params 字段(即上一步的参数字典):
import requests
def query_tickets(app_key: str, sign: str, params: dict):
url = "https://open.12306.cn/api/train/query"
headers = {
"Content-Type": "application/json; charset=utf-8",
"Accept": "application/json",
"X-App-Key": app_key,
"X-Sign": sign
}
payload = {"params": params}
try:
response = requests.post(url, json=payload, headers=headers, timeout=10)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"Request failed: {e}")
return None
# 组合调用
if __name__ == "__main__":
app_key = "a1b2c3d4e5f67890"
app_secret = "xYz9AbC2DeF4GhI6JkL8MnO0PqR2StU4"
private_key = """-----BEGIN RSA PRIVATE KEY-----
...(同上)
-----END RSA PRIVATE KEY-----"""
params = {
"app_key": app_key,
"timestamp": str(int(time.time() * 1000)),
"nonce": "abc123def456ghij",
"from_station": "BJP",
"to_station": "SHH",
"train_date": "2024-06-15"
}
sign = generate_sign(params, app_secret, private_key)
result = query_tickets(app_key, sign, params)
if result and result.get("code") == 0:
print("Success! Tickets data:", result.get("data", {}))
else:
print("Failed:", result)
关键响应字段说明:
| 字段 | 类型 | 说明 |
|---|---|---|
code | int | 0 表示成功,非 0 为错误码(如 1001 =签名错误, 1002 =时间戳超时) |
data | object | 包含 result 数组,每项为一趟列车的余票快照 |
result[].station_train_code | string | 车次号(如 G101 ) |
result[].yw_num | string | 硬卧余票数( "无" 表示无票, "有" 表示有票, "---" 表示不发售) |
result[].ze_num | string | 二等座余票数(同上) |
提示:余票数据为快照式,非实时流。开放平台不提供WebSocket推送,需自行控制轮询间隔(建议≥3秒,避免触发频率限制)。
4. 避免常见签名错误:参数顺序、编码与密钥格式三重校验
4.1 参数排序陷阱:ASCII码 vs Unicode排序
开放平台明确要求“按参数名ASCII码升序排列”,而非字典序。例如参数 {"z": "1", "a": "2", "α": "3"} ,正确排序应为:
-
a=2(ASCII 97) -
z=1(ASCII 122) -
α=3(Unicode U+03B1,ASCII不覆盖,但平台将其视为0x3B1,大于z)
错误做法是直接用 sorted(dict.items()) ——Python 3.7+中字典保持插入序,但 sorted() 默认按Unicode码点排序, α (U+03B1)会排在 a (U+0061)之前,导致签名不一致。 必须显式指定ASCII排序 :
# ✅ 正确:强制ASCII排序
sorted_items = sorted(params.items(), key=lambda x: x[0].encode('ascii'))
# ❌ 错误:Unicode排序(可能导致签名失败)
sorted_items = sorted(params.items()) # α会排在a前面
4.2 字符编码一致性:UTF-8还是GBK?
所有参数值必须使用UTF-8编码参与签名计算。若参数含中文(如车站名),需确保:
- 字符串本身为Unicode(Python 3默认)
- 拼接时
"&".join(...)不引入额外编码 -
hashlib.sha256()输入为bytes,故需.encode('utf-8')
常见错误是 params["from_station"] = "北京北".encode('gbk') ,此时 param_str 含GBK字节,签名必然失败。正确写法:
# ✅ 正确:中文参数保持Unicode,encode在最后一步
params = {
"from_station": "北京北", # Unicode字符串
"to_station": "上海虹桥"
}
param_str = "&".join([f"{k}={v}" for k, v in sorted_items])
digest = hashlib.sha256(param_str.encode('utf-8')).digest() # 此处encode
4.3 RSA私钥格式验证表
开放平台仅接受PKCS#1格式私钥( -----BEGIN RSA PRIVATE KEY----- ),而非PKCS#8( -----BEGIN PRIVATE KEY----- )。若你生成的密钥是PKCS#8,需转换:
# 将PKCS#8转为PKCS#1(OpenSSL命令)
openssl pkcs8 -in private_key_pkcs8.pem -out private_key_pkcs1.pem -traditional
下表列出常见密钥格式特征与验证方法:
| 格式类型 | 头部标识 | 是否支持 | 验证命令 |
|---|---|---|---|
| PKCS#1 (RSA) | -----BEGIN RSA PRIVATE KEY----- | ✅ 支持 | openssl rsa -in key.pem -check -noout |
| PKCS#8 | -----BEGIN PRIVATE KEY----- | ❌ 不支持 | openssl pkcs8 -in key.pem -inform PEM -nocrypt |
| PEM无密码 | 含 DEK-Info: 行 | ✅ 支持 | head -n 2 key.pem | grep -q "DEK-Info" |
| PEM有密码 | 含 Proc-Type: 4,ENCRYPTED | ❌ 不支持(需先解密) | openssl rsa -in key.pem -check -noout (会提示输入密码) |
若 generate_sign() 函数抛出 ValueError: Not a valid RSA private key ,90%概率是密钥格式不符,优先检查头部字符串。
5. 实战技巧:用缓存降低请求频次,用异步提升响应速度
5.1 基于Redis的余票结果缓存策略
12306开放平台对单个APP_KEY有QPS限制(公众服务类约5次/秒),频繁查询同一车次会造成配额浪费。我们用Redis缓存查询结果,有效期设为60秒(足够覆盖用户操作间隙):
import redis
import json
class TicketCache:
def __init__(self, host='localhost', port=6379, db=0):
self.r = redis.Redis(host=host, port=port, db=db, decode_responses=True)
def get_cache_key(self, from_station: str, to_station: str, train_date: str) -> str:
return f"ticket:{from_station}:{to_station}:{train_date}"
def set_result(self, from_station: str, to_station: str, train_date: str, data: dict, expire: int = 60):
key = self.get_cache_key(from_station, to_station, train_date)
self.r.setex(key, expire, json.dumps(data, ensure_ascii=False))
def get_result(self, from_station: str, to_station: str, train_date: str) -> dict:
key = self.get_cache_key(from_station, to_station, train_date)
cached = self.r.get(key)
return json.loads(cached) if cached else None
# 使用示例
cache = TicketCache()
cached_data = cache.get_result("BJP", "SHH", "2024-06-15")
if cached_data:
print("Hit cache:", cached_data)
else:
# 执行query_tickets()...
result = query_tickets(...)
if result and result.get("code") == 0:
cache.set_result("BJP", "SHH", "2024-06-15", result)
缓存命中率优化建议:
- 对
train_date做归一化:用户输入2024/06/15自动转为2024-06-15 - 对车站电码做映射:用户输入“北京南”自动查表转为
VAP,避免"北京南"与"北京南站"两种写法产生不同缓存键
5.2 使用asyncio并发查询多趟车次
单次查询仅返回固定日期的全部车次,但用户常需对比不同日期(如6月15日/16日/17日)。用 asyncio.gather() 并发请求,可将3次查询耗时从 3×1.2s=3.6s 降至 ~1.3s :
import asyncio
import aiohttp
async def async_query(session, app_key: str, sign: str, params: dict):
url = "https://open.12306.cn/api/train/query"
headers = {
"Content-Type": "application/json; charset=utf-8",
"Accept": "application/json",
"X-App-Key": app_key,
"X-Sign": sign
}
payload = {"params": params}
async with session.post(url, json=payload, headers=headers, timeout=aiohttp.ClientTimeout(total=10)) as resp:
return await resp.json()
async def query_multi_dates(app_key: str, app_secret: str, private_key: str,
from_station: str, to_station: str, dates: list):
tasks = []
async with aiohttp.ClientSession() as session:
for date in dates:
params = {
"app_key": app_key,
"timestamp": str(int(time.time() * 1000)),
"nonce": f"async-{int(time.time())}",
"from_station": from_station,
"to_station": to_station,
"train_date": date
}
sign = generate_sign(params, app_secret, private_key)
tasks.append(async_query(session, app_key, sign, params))
results = await asyncio.gather(*tasks, return_exceptions=True)
return [r for r in results if not isinstance(r, Exception)]
# 调用
if __name__ == "__main__":
dates = ["2024-06-15", "2024-06-16", "2024-06-17"]
loop = asyncio.get_event_loop()
results = loop.run_until_complete(
query_multi_dates("a1b2...", "xYz9...", private_key, "BJP", "SHH", dates)
)
print("Multi-date results:", len(results))
并发安全边界:
- 开放平台允许单IP最多5个并发连接,故
dates列表长度建议≤5 -
nonce必须每次请求唯一,故在循环内动态生成(如f"async-{date}-{int(time.time())}") -
aiohttp.ClientSession复用连接池,比requests同步调用节省TCP握手开销
最终,你得到的不是一个“解压即用”的黑盒脚本,而是一套可审计、可监控、符合平台规范的余票查询能力。它不承诺“抢到票”,但确保每一次请求都走在12306开放生态的正轨上——这正是技术人该守住的底线。
更多推荐

所有评论(0)