简介这是一份面向Python开发者与自动化脚本学习者的火车票抢票系统完整源码围绕登录验证、车次查询、席位选择、订单提交等购票环节设计自动化策略既能帮助理解接口封装、任务调度也能作为Python后端与JavaScript/HTML/CSS前端联动实现的参考。压缩包共83个文件、6.17MB主体为42个Python脚本负责车站数据、请求封装、查询、订单、日志、验证码识别等后端功能6个JavaScript文件结合HTML、CSS以及字体图片资源组成可交互的前端购票界面另含Dockerfile、依赖清单、.gitignore、示例配置和说明文档便于部署与版本管理。项目按cluster、query、order、user等目录模块划分结构清晰能直观看到一个多语言协作项目的组织方式。已有370人学习下载对希望掌握py12306抢票思路、自动化脚本落地和前端联动的开发者有较高参考价值。1. py12306 是什么从抢票工具到自动化框架火车票抢票这件事大多数人第一反应是打开 12306 网页或 App 手动刷新。但当你把抢票需求交给 Python 时问题就变成了如何把“查票、选车次、提交订单”变成可重复执行的自动化流程。py12306 就是这类项目中结构比较完整的一个它不仅包含了登录、余票查询、下单等后端逻辑还用 Flask 搭建了一个 Web 前端让用户可以在浏览器里看到任务状态、手动确认订单。对于后端工程师来说它是研究多线程任务调度和 API 封装的现成案例对于前端开发者它可以拆出来当 Flask JavaScript 的交互模板对于普通用户配好环境后也能直接拿来用。整个工程有 74 个文件其中 33 个 Python 脚本说明它的逻辑拆分比一般脚本要细致得多。这也是我把它当作研究对象而不是简单工具的原因。2. 项目结构与核心模块拆解从目录看懂 py123062.1 顶层目录与职责划分打开压缩包后首先看到的是upload.zip里的整个工程根目录。虽然命名为 upload但实际是一个完整的 Git 仓库包含.gitignore、README或readme.txt、LICENSE、Dockerfile、requirements.txt以及runtime、data、images等资源目录。如果你习惯从零搭建 Python 项目会发现这种布局非常标准代码、配置、资源分离便于部署和二次开发。py12306/ ├── config.py # 全局配置入口 ├── exceptions/ # 自定义异常 ├── helpers/ # 工具函数API请求、验证码、站点查询 ├── order/ # 订单相关下单、订单状态 ├── query/ # 余票查询车次、票价、坐席 ├── user/ # 用户模块登录、用户任务 ├── cluster/ # 集群支持Redis 分布式锁 ├── web/ # Flask Web 控制台 ├── templates/ # Jinja2 HTML 模板 ├── static/ # 前端静态资源CSS/JS/图片 ├── log/ # 统一日志模块 ├── main.py # 程序入口 ├── env.py.example # 环境变量示例 └── requirements.txt # Python 依赖这个目录结构最大的价值在于它把抢票流程拆成了user、query、order三个独立模块而不是把所有逻辑堆在一个main.py里。这样做的直接好处是后续如果你想单独测试查询接口或者替换登录验证码识别方案不需要动其他功能模块。2.2 核心模块的调用关系从函数调用的角度整个系统的启动流程从main.py开始先读config.py然后初始化user模块再启动query和order的定时任务。web.py作为 Flask 进程并行运行提供浏览器界面。具体调用关系可以用下面这个表来理解模块核心职责关键文件依赖helpers封装 12306 接口、验证码、站点数据func.py, api.py, station.pyrequestsuser用户登录、会话维护、多账号任务user.py, job.pyhelpersquery余票查询、车次筛选、坐席缓存query.py, job.pyhelpersorder提交订单、乘车人选择、订单确认order.pyuser, queryclusterRedis 锁、分布式任务协调cluster.py, redis.pyredis-pywebFlask 路由、API、前端渲染web.py, templates/, static/Flasklog分模块日志磁盘写入common_log.py, user_log.py, query_log.pylogging这里面容易忽略的是exceptions/目录它把“用户未登录”“车次不存在”“订单提交失败”等异常单独定义成类。这样上层代码不需要用try...except去匹配字符串直接用except UserNotLogin就可以捕获。如果你写过调用第三方 HTTP 接口的代码会发现这是一种很有效的防御式编程方式。2.3 配置体系与多环境支持项目提供了env.py.example、env.slave.py.example、env.docker.py.example三个配置样例分别对应本地运行、从节点运行、Docker 容器运行。实际使用时需要复制env.py.example为env.py。配置项里最重要的两个部分是账号信息和任务策略。# config.py 中读取环境变量的常见写法 import os class Config: # 12306 账号 USER_ACCOUNT os.getenv(PY12306_USER, ) USER_PASSWORD os.getenv(PY12306_PASSWORD, ) # 抢票任务配置 QUERY_INTERVAL float(os.getenv(PY12306_QUERY_INTERVAL, 5)) ORDER_WAIT_SECONDS int(os.getenv(PY12306_ORDER_WAIT, 3)) CLUSTER_ENABLED os.getenv(PY12306_CLUSTER, false).lower() true这里把间隔时间设置为5秒实际使用中不建议低于 3 秒否则会给 12306 服务器造成压力也可能触发封锁。CLUSTER_ENABLED是控制是否启用 Redis 集群协作的关键开关后面第 5 章会专门讲。3. 抢票核心流程实现登录、查询与下单3.1 登录与验证码识别流程登录是抢票的前提也是所有功能里最容易因为接口变动而出错的环节。py12306 的做法是把登录拆成两步第一步用账号密码获取登录会话第二步处理验证码。验证码识别在helpers/OCR.py里实现同时也支持第三方打码平台比如vender/ruokuai是接入了若快平台的接口。# helpers/auth_code.py 中调用 OCR 识别验证码的简化流程 def get_auth_code(image_bytes): # 优先使用本地 OCR失败时回退到第三方打码 try: from helpers.OCR import local_recognize return local_recognize(image_bytes) except Exception as e: # 这里可以替换成你自己的打码平台调用 from vender.ruokuai import recognize_by_ruokuai return recognize_by_ruokuai(image_bytes, username, password) # user/user.py 中登录方法 def login(session, account, password): # 获取登录前必需的 cookie session.get(https://dynamic.12306.cn/.../init) # 提交账号密码 resp session.post(https://.../login, data{ username: account, password: password, answer: get_auth_code(captcha_image_bytes) }) if resp.json().get(result_code) 0: return True else: raise LoginError(resp.json().get(result_message))这个代码里最关键的参数是answer它对应验证码的识别结果。12306 的验证码有时是文字点选有时是图片分类所以get_auth_code返回的格式会根据验证码类型变化。常见做法是让 OCR 模块返回一组坐标或字符串然后登录接口再按照约定格式拼接。如果你是接到自己的项目里建议先把验证码图片保存到本地人工看一次再决定用哪种识别方案。3.2 车票余票查询与筛选策略查询模块是抢票系统的“眼睛”它决定了你能多快看到余票。py12306 的query/query.py里维护了一个定时任务每隔几秒请求一次余票查询接口。为了提高效率它会把同一个发到终到站的车次请求合并并缓存结果到内存或 Redis。# query/query.py 余票查询中的关键逻辑 LEFT_TICKET_URL https://kyfw.12306.cn/otn/leftTicket/query def query_left_ticket(session, from_station, to_station, date): payload { leftTicketDTO.train_date: date, leftTicketDTO.from_station: from_station, leftTicketDTO.to_station: to_station, purpose_codes: ADULT } resp session.get(LEFT_TICKET_URL, paramspayload) data resp.json()[data][result] result [] for item in data: parts item.split(|) # 常见的字段顺序: 车次、始发站、终到站、发时、到时、商务座、一等座... train_no parts[3] from_code parts[6] to_code parts[7] seat_status parse_seat(parts[30:40]) # 不同版本的字段有偏移 result.append({ train_no: train_no, from: from_code, to: to_code, seats: seat_status }) return result这里要特别注意的是leftTicketDTO.from_station和to_station的值。12306 接口里用的不是“北京”或“上海”这样的中文而是电报码比如北京是BJP。项目里的helpers/station.py就负责把中文站名转换成电报码并生成一份stations.txt字典。你如果直接复制代码去用记得先调用update_stations()获取最新站点列表因为新开通的车站会不定期增加。purpose_codes参数决定查询的是成人票还是学生票ADULT就是成人票。如果你的目标是抢高铁票需要同时关注G字头车次的二等座字段。这里很容易踩的坑是不同日期、不同车次类型返回字段中的座席顺序不一致所以解析时必须先根据车次类型做二次判断不能硬编码索引。3.3 订单提交与状态轮询查到余票后下一步就是提交订单。这个阶段比查询更复杂因为涉及乘车人选择、席别提交、订单确认三步操作。py12306 的order/order.py里把这三步封装成三个方法并且用状态机来管理流程。# order/order.py 提交订单的简化流程 def submit_order(session, train_no, seat_type, passengers, date): # 1. 检查是否已经提交过订单 if order_exists(session, train_no, date): return ORDER_EXISTS # 2. 预提交获取订单 token pre_resp session.post(https://.../order/preSubmit, data{ train_no: train_no, train_date: date, station_train_code: code, seat_detail_type: seat_type }) token pre_resp.json()[data][global_db_query] # 3. 检查乘车人并提交 check_resp session.post(https://.../order/checkUser, data{ passenger_id: passengers[0][id] }) if check_resp.json()[data][flag]: submit_data build_order_data(token, train_no, passengers, seat_type) final_resp session.post(https://.../order/submit, datasubmit_data) return final_resp.json() # build_order_data 里重点设置乘客信息和席别 def build_order_data(token, train_no, passengers, seat_type): return { secret_str: token, train_no: train_no, passenger_passengers: passengers[0][name], passenger_id_type_code: 1, passenger_id_no: passengers[0][id], seat_detail_type: seat_type, whats_up: N }submit_order返回后抢票流程并没有结束。12306 在下单后通常需要等待几秒钟让服务器确认是否有票。所以 py12306 在提交后启动一个轮询任务每隔 2~3 秒请求一次订单状态如果返回待支付就调用通知接口提醒用户。这个“提交后轮询”的机制很值得借鉴很多自动化工具有时候不是被卡在提交而是卡在提交后没有及时获取结果导致重复提交。4. 前端实现Web 控制台与实时交互4.1 Flask Web 后端与 API 设计py12306 的前端不是独立前端工程而是 Flask 直接渲染模板加少量原生 JavaScript。web/web.py里定义了简单的路由主要负责两类工作一类是渲染页面另一类是提供 JSON 接口给前端轮询。这种设计非常适合单人或者小团队使用省去了 Nginx 反向代理和对前端构建工具链的维护成本。# web/web.py 中典型的路由定义 from flask import Flask, render_template, jsonify app Flask(__name__) app.route(/) def index(): # 渲染主控制台页面 return render_template(index.html, titlepy12306 控制台) app.route(/api/tasks) def api_tasks(): # 返回当前所有抢票任务的状态供前端轮询 tasks get_all_tasks() return jsonify({ code: 0, data: [{ id: t.id, train_no: t.train_no, status: t.status, updated_at: t.updated_at.strftime(%H:%M:%S) } for t in tasks] })render_template会从templates/目录下加载 HTML 文件同时把静态资源放到static/下。这里你可能会注意到它返回给前端的updated_at是格式化后的字符串而不是时间戳这样前端就不用再new Date(timeStamp).toLocaleString()格式化一次能少写几行代码。4.2 前端页面与 JavaScript 交互主控制台页面是用index.html加static/js/下的几个 JavaScript 文件实现的。JavaScript 部分使用原生的fetch函数定时请求/api/tasks然后动态更新表格。因为没有引入 Vue 或 React所以逻辑非常直观适合前端入门者学习 DOM 操作。!-- templates/index.html 中轮询任务状态的逻辑 -- script const POLL_INTERVAL 3000; let isPolling true; function formatStatus(status) { const map { waiting: 等待中, running: 查询中, submitted: 已提交, paid: 已支付 }; return map[status] || status; } async function pollTasks() { if (!isPolling) return; try { const resp await fetch(/api/tasks); const data await resp.json(); const tbody document.querySelector(#task-table tbody); tbody.innerHTML ; data.data.forEach(task { const tr document.createElement(tr); tr.innerHTML td${task.train_no}/td td classstatus${formatStatus(task.status)}/td td${task.updated_at}/td; tbody.appendChild(tr); }); } catch (e) { console.error(轮询失败, e); } } setInterval(pollTasks, POLL_INTERVAL); window.addEventListener(beforeunload, () { isPolling false; });这个脚本里有几个细节值得学习。POLL_INTERVAL设置为 3000 毫秒和后端查询任务的间隔保持倍数关系避免前端请求太密集。formatStatus把英文状态映射成中文显示是前后端约定好的枚举值。beforeunload事件里把isPolling置为 false是为了防止用户关闭页面时浏览器还在发不必要的网络请求。4.3 前端可视化与用户体验除了任务列表py12306 的页面里还放了一张抢票成功后的截图images/order_success.png这个不是装饰而是把它作为默认的“订单成功”占位图。当后台收到成功通知时前端会弹出一个模态框显示这张图并提示用户尽快去 12306 支付。这种设计把后端事件和前端提示很好地连接起来了。在static/css/里样式使用的是自定义类名而不是 UI 框架。好处是页面加载速度快不依赖 CDN坏处是如果你想调整整体风格需要自己手写很多细节。这里我的建议是如果你计划把这个项目改造成给非技术人员使用可以保留 Flask 路由但把前端换成 Vue 或者 React用 webpack 构建出来的静态文件放在static/dist下这样改动成本最小。5. 部署、调参与抢票效率优化5.1 环境准备与快速启动拿到源码后第一步不是直接跑而是先看requirements.txt里的依赖。常见依赖包括requests、Flask、redis、APScheduler等。建议用虚拟环境隔离python -m venv venv source venv/bin/activate # Windows 下执行 venv\Scripts\activate pip install -r requirements.txt cp env.py.example env.py # 编辑 env.py填入 12306 账号密码以及查询区间 python main.py启动后main.py会加载配置文件并启动两个进程一个是任务调度进程一个是 Flask Web 服务。浏览器访问http://127.0.0.1:8000即可看到控制台。如果 Flask 端口被占用可以修改web.py里的app.run(port8000)。5.2 Redis 集群模式与并发控制项目中的cluster/模块支持通过 Redis 实现多台机器协同抢票。在多实例场景下如果不加控制两台机器会重复提交同一个订单导致 12306 提示“订单重复”。解决方法是使用 Redis 的SETNX命令做分布式锁抢到锁的实例才允许提交订单。# cluster/redis.py 中获取锁的示意逻辑 import redis import time r redis.Redis(host127.0.0.1, port6379, db0) def acquire_lock(train_no, timeout3): lock_key flock:order:{train_no} # SET key value NX EX只有不存在时才设置并自动过期 acquired r.set(lock_key, 1, nxTrue, extimeout) return acquired def release_lock(train_no): lock_key flock:order:{train_no} r.delete(lock_key)这里最容易忽略的是锁的过期时间。如果设置太短订单提交还没完成锁就过期了别的实例会重复提交如果设置太长抢票失败后锁一直不释放影响后续重试。我一般会把超时时间设置成 10~15 秒并配合第 3 章说的订单轮询逻辑等轮询确认失败后再释放锁。5.3 抢票参数调优与日志分析最后聊一个实际调优技巧。很多人拿到源码后只改账号和日期忽略了QUERY_INTERVAL和ORDER_WAIT_SECONDS这两个参数。QUERY_INTERVAL控制查询余票的频率它并不一定是越小越好。如果频率太高可能会被 12306 临时限制访问表现为查询接口返回空数据或者提示“操作频繁”。根据我自己的测试5 秒到 8 秒是一个相对安全的区间对于热门车次可以降低到 3 秒但需要配合代理 IP。另一个参数ORDER_WAIT_SECONDS控制提交订单后等待服务端确认的时间建议设置 2 到 3 秒因为 12306 的秒杀场景中等待太久车票会被释放给其他人。日志模块也值得花时间研究。log/目录下按照user_log、query_log、order_log分文件输出这意味着你可以在运行几小时后用下面的命令快速定位问题tail -f log/order_log.log | grep ERROR从日志里能看出是登录失败、查询超时还是订单提交被拒。如果发现大量timeout优先检查本地网络到 12306 的连通性如果发现secret_str相关异常多半是接口字段更新需要对照抓包数据修正order/order.py中的参数名。调优没有统一模板唯一切实有效的路径就是多跑几轮把每次失败时的日志和上下文对照起来看。本文还有配套的精品资源点击获取