简介这是一份面向Python爬虫零基础学习者的项目实战资源以12306网站车次信息为目标场景演示从分析接口、构造请求到解析返回数据并提取车次的完整过程。压缩包共4个文件内含2个Python源码文件、1个课堂笔记文档和1份PPT课件源码划分演示步骤便于对照运行笔记与课件补充了关键知识点与调试思路整体仅165KB下载后即可快速浏览。已有77人参与学习适合配合教程循序练习也可作为课程设计或毕业设计的参考资料。通过该资源读者能掌握真实站点爬取中的请求头处理、数据清洗、结果格式化等实用技巧并为后续扩展余票、票价等爬取功能打下基础。1. 12306车次查询的难点接口远比页面易拿选“12306爬取车次”当Python爬虫项目的人多半是先被自动抢票吸引过来的。但把题目落到“爬取车次”这四个字上之后真正值得做的事其实是把12306的查询接口吃透参数怎么编码、车站代码怎么映射、返回数据怎么解析、并发查询怎么绕过频率限制。这个项目里票据侧的逻辑很少反而把HTTP请求的格式、会话状态依赖、数据结构化这三件事占全了适合作为requests库、异步并发和网络调试的综合练兵场。常见做法是先通过浏览器开发者工具抓出otn/leftTicket/query接口用Python模拟发送查询请求再对返回的字符串数组按字段切分得到车次、出发到达时间、余票和票价信息。整个过程不涉及JS渲染和验证码识别只要把接口字段理清请求头保持和浏览器一致就能稳定拿数据。2. 先拆接口leftTicket/query 的请求结构与返回字段做爬虫的第一步永远不是写代码而是打开浏览器开发者工具用Network面板抓一次真实的查询请求。12306的Web端查询页不会把车次直接渲染在HTML里而是通过XHR异步加载JSON后再拼表。这个XHR就是你唯一需要的接口。2.1 用浏览器抓包确认请求格式在12306余票查询页选择任意日期和区间点击查询后Network面板会多出一条leftTicket/query请求它的完整请求URL大致是https://kyfw.12306.cn/otn/leftTicket/query?leftTicketDTO.train_date2025-06-01leftTicketDTO.from_stationSHHleftTicketDTO.to_stationBJPpurpose_codesADULT这里是GET请求查询条件全部放在Query String里。注意from_station和to_station不是中文站名而是SHH和BJP这样的电报码这属于12306自己定义的车站编码体系后续必须通过车站表转换无法跳过。请求还需要带上Cookie、User-Agent、Referer等请求头其中Cookie要从首页访问后获取否则接口会返回登录跳转或直接拒绝数据。2.2 请求头与Cookie依赖关系12306的车次查询接口本身不需要登录但要求请求方先访问过站点主页拿到会话Cookie后才能正常响应。我用requests库实现时通常会先创建一个requests.Session()用Session保持Cookie状态在真正查询前先GET一次https://kyfw.12306.cn/otn/leftTicket/init让服务端把JSESSIONID下发到本地再携带这个Cookie去请求查询接口。import requests session requests.Session() session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36, Referer: https://kyfw.12306.cn/otn/leftTicket/init, }) def init_session(): # 先访问首页服务端会写入 JSESSIONID 等会话 Cookie resp session.get(https://kyfw.12306.cn/otn/leftTicket/init, timeout10) resp.raise_for_status() def query_train(date_str, from_code, to_code): url https://kyfw.12306.cn/otn/leftTicket/query params { leftTicketDTO.train_date: date_str, leftTicketDTO.from_station: from_code, leftTicketDTO.to_station: to_code, purpose_codes: ADULT, } resp session.get(url, paramsparams, timeout10) resp.raise_for_status() return resp.json()代码里session.get返回的响应对象调用resp.json()是因为12306这个接口返回的是JSON格式外层结构是{httpstatus:200,data:{result:[...]}}。params参数不需要手动做URL编码requests库会处理train_date里的横杠等字符。2.3 返回数据格式与字段位置data.result是一个字符串数组每个元素对应一趟车次但元素不是JSON对象而是一条用竖线分隔的长字符串。你直接用split(|)切分就能按位置取到出发站、到达站、出发时间、到达时间、历时、一等座余票、二等座余票、软卧、硬卧、硬座、无座等字段。这是12306查询接口最容易迷惑新手的点按字段位置依赖字符串结构来取数属于典型的历史遗留格式。def parse_train(raw_line: str): parts raw_line.split(|) train_data { 车次号: parts[2], 出发站码: parts[6], 到达站码: parts[7], 发车时间: parts[8], 到达时间: parts[9], 历时: parts[10], 商务座: parts[32], 一等座: parts[31], 二等座: parts[30], 无座: parts[26], } return train_data这里的parts[2]对应车次号parts[6]和parts[7]是出发、到达站的编码parts[8]和parts[9]是发车和到达时间parts[30]到parts[32]是坐席余票。注意parts里某些字段可能为空串比如无座没有余票时返回空字符串解析时要做好默认值处理否则后续写入数据库会报错。拿到字符串位置规律后这一步实际上就完成了整个爬虫项目里最核心的数据解析工作。提示有时候resp.json()里的data是None多半是Cookie失效或请求频率太高被临时拦了。先检查Session是否正常、是否重开过进程。2.4 purpose_codes 参数的作用purpose_codes控制的是成人和学生两种购票类型取ADULT即可。有些教程里写成0X00那是老版本接口的参数值当前版本用ADULT更稳。这个参数看起来只是多一个字段但在实际爬取中会影响返程余票的字段位置个别情况下成人票和学生票的返回结构会出现细微差异所以别直接拿学生票请求的结果套成人的解析逻辑。3. 车站表与参数构造用 station_name.js 完成三码映射查询接口要求参数值必须是车站电报码但用户习惯输入中文站名。从中文站名到电报码的映射需要从12306的静态资源里获取车站数据这也是整个项目里最容易被遗漏的一步。好在12306把全国车站代码放在一个JS文件里直接用正则提取就能拿到映射表不需要另找数据库导入。3.1 获取并解析 station_name.js这个文件固定在https://kyfw.12306.cn/otn/resources/js/framework/station_name.js内容是一段JavaScript赋值语句变量名为station_names里面用符号分隔的脚本结构封装了多个字段。每个车站的格式是拼音首字母缩写|站名|电报码|拼音|简拼|序号。解析思路是通过正则把所有开头的记录切出来再用|分割字段。import re import requests def load_station_map(session: requests.Session): url https://kyfw.12306.cn/otn/resources/js/framework/station_name.js resp session.get(url, timeout10) resp.encoding utf-8 text resp.text # 取出 分隔的每条记录 pattern r(\w)\|([\u4e00-\u9fa5])\|([A-Z])\|(\w)\|(\w)\|(\d) stations re.findall(pattern, text) name_to_code {} code_to_name {} for item in stations: # item (首字母, 站名, 电报码, 拼音, 简拼, 序号) station_name item[1] station_code item[2] name_to_code[station_name] station_code code_to_name[station_code] station_name return name_to_code, code_to_name正则里[\u4e00-\u9fa5]匹配中文站名[A-Z]匹配电报码必须保证顺序和JS文件里一致否则会提取出错误映射。station_name.js体积大约几百KB每次爬取时动态获取一次即可也可以把解析结果存成本地JSON便于离线查询。3.2 中文站名到电报码的查表转换拿到映射表之后构造查询参数就很简单了。把用户输入的中文站名换成电报码字符串再传给第2章的query_train函数。注意模糊匹配有些站名带“站”字后缀例如“北京南站”在车站表里是“北京南”建议在查表前先做一次去“站”处理避免映射不到。反过来返回结果里的出发站和到达站也是电报码要展示给用户看时再转回中文站名。name_to_code, code_to_name load_station_map(session) from_station 北京南 to_station 上海虹桥 # 去除尾部站字避免匹配失败 from_code name_to_code.get(from_station.rstrip(站)) to_code name_to_code.get(to_station.rstrip(站)) if not from_code or not to_code: raise ValueError(f无法识别车站: {from_station} 或 {to_station}) data query_train(2025-06-01, from_code, to_code) for line in data[data][result]: train parse_train(line) print(train[车次号], train[发车时间], train[到达时间])这里把车站名称当作字典的key查询复杂度是O(1)对后续做并发查询很重要。每次调用前都先判断是否None能直接暴露输入站名不在车站表里的问题。12306的车站表覆盖了全国几千个站但偶尔有新开通的线路车站没更新遇到这种情况自查没有错等官方更新后再抓。3.3 把车站表固化到本地复用每次都去请求station_name.js虽然没问题但项目做大了以后会拖慢初始化速度而且没必要。最常见做法是在首次加载后存成stations.json文件后续启动时直接读取本地文件只有文件缺失时才重新拉取。同时把获取时间也存下来方便定期刷新。import json import os STATION_FILE stations.json def get_station_map(session: requests.Session): if os.path.exists(STATION_FILE): with open(STATION_FILE, r, encodingutf-8) as f: data json.load(f) return data[name_to_code], data[code_to_name] name_to_code, code_to_name load_station_map(session) with open(STATION_FILE, w, encodingutf-8) as f: json.dump({name_to_code: name_to_code, code_to_name: code_to_name}, f, ensure_asciiFalse, indent2) return name_to_code, code_to_name这种本地缓存策略在爬虫工程中很常见不只是12306的项目里适用。把不常变化的静态资源缓存下来为后续并发请求节省时间也让整个脚本的启动流程更干净。station_name.js里的数据一般一个月内不会天天变不用追求每次启动强制更新。提示如果返回的车次列表里出发站和到达站字段拆出来和预期不一致先查code_to_name里是否映射到了正确中文名关注parts[6]和parts[7]到底是当前车次的始发站还是你查询的乘车站12306返回的是区间票池信息不一定等于用户查询条件。4. 并发与反爬把单次查询放大到日期维度单次查询只能拿到某一天某一对车站的车次但实际项目中往往要查多天、多区间。比如帮人盯票时要同时查未来5天从北京到上海的所有车次简单用for循环逐次请求会太慢而且容易被服务端识别为异常流量。这个阶段要把并发设计提上来常见方案是线程池加延迟控制必要时接上代理池。4.1 先用线程池实现日期维度并行Python里做并发请求最简单可控的是concurrent.futures.ThreadPoolExecutor。requests库本身是同步的但在等待网络响应时可以把CPU让给其他线程这是IO密集型任务的正确加速手段。from concurrent.futures import ThreadPoolExecutor, as_completed def query_multiple_dates(date_list, from_code, to_code, max_workers3): results [] def safe_query(date_str): try: resp_data query_train(date_str, from_code, to_code) trains resp_data[data][result] return date_str, trains except Exception as exc: return date_str, ferror: {exc} with ThreadPoolExecutor(max_workersmax_workers) as executor: future_map { executor.submit(safe_query, date_str): date_str for date_str in date_list } for future in as_completed(future_map): date_str, trains future.result() results.append((date_str, trains)) return results这里max_workers设为3既保证并发能提速又不至于触发12306的拦截。如果你拿网络上的“多线程快进”教程比着做把workers调到20以上大概率会遇到请求状态码为403或返回数据为空。12306的频控逻辑不透明稳妥做法是先从3到5个线程起步观察返回正常再慢慢增加。4.2 控制请求频率延时、UA池与IP代理只靠线程池不足以长时间稳定运行更稳妥的工程化方案是组合使用三件套请求间随机延时、轮换User-Agent、维护可用代理池。特别是做多日期批量爬取时如果所有请求都来自同一个IP且无间隔很容易被服务端识别为脚本行为导致接口短暂失效。import random import time USER_AGENTS [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36, ] def build_session_with_ua(): s requests.Session() s.headers.update({ User-Agent: random.choice(USER_AGENTS), Referer: https://kyfw.12306.cn/otn/leftTicket/init, }) return s def safe_request_with_retry(session, url, params, retries3): for attempt in range(retries): try: resp session.get(url, paramsparams, timeout10) if resp.status_code 200: return resp.json() except requests.RequestException: pass time.sleep(1 attempt * 2) return NoneUSER_AGENTS至少准备3个主流浏览器版本轮换使用能降低被指纹识别锁定的概率。safe_request_with_retry里的退避重试策略是应对偶发网络抖动的兜底方案。如果你用云服务器跑爬虫单IP请求量较大时最有效的手段是接入付费代理IP池按请求维度轮换出口IP。代理配置在requests里只需要给session.proxies赋值即可但代理质量参差不齐失效IP会导致连接超时增加需要配代理检测逻辑。4.3 验证并发改造后的效果做完并发以后至少要从三个维度验证返回车次数量是否正确、多天结果是否有缺失、请求成功率有没有明显下降。查询北京南到上海虹桥、未来5天的车次正常情况下每天都有几十趟高铁如果某天返回的result数组长度明显偏小大概率是请求被限流或解析错位。日期车次数目耗时(秒)状态2025-06-01460.8正常2025-06-02440.9正常2025-06-03402.1接口偶发超时2025-06-04450.7正常上表是一次典型的并发测试结果。max_workers3时总耗时约5秒左右如果换成单线程顺序请求同样的数据量至少要15到20秒。2025-06-03那次耗时偏高是重试机制在起作用。对整个项目而言能接受这种零散波动只要没有连续大面积失败就行。5. 数据落库与结果验证从临时打印到可复现的查询工具最后一层要做的事情是让爬取结果真正能留痕而不仅是打印在控制台里。车次数据的特点是高度文本化、字段位置固定且存在大量空值把它写进SQLite后就能做条件筛选和趋势分析。这块工作不复杂但直接影响整个项目能不能从“能跑”变成“能用”。5.1 用SQLite保存车次明细并去重12306车次数据具有时效性同一日期同一车次的余票数量会变化所以落库时不能简单插入。常见设计是加一个query_time字段记录查询时间同时以车次号日期出发站到达站作为唯一索引来避免重复行累积。import sqlite3 from datetime import datetime def init_db(): conn sqlite3.connect(trains.db) conn.execute( CREATE TABLE IF NOT EXISTS train_schedule ( id INTEGER PRIMARY KEY AUTOINCREMENT, query_date TEXT, train_no TEXT, from_station TEXT, to_station TEXT, dep_time TEXT, arr_time TEXT, duration TEXT, biz_seat TEXT, first_seat TEXT, second_seat TEXT, no_seat TEXT, query_time TEXT, UNIQUE(query_date, train_no, from_station, to_station) ) ) conn.commit() return conn def save_trains(conn, query_date, from_name, to_name, train_list): now datetime.now().strftime(%Y-%m-%d %H:%M:%S) for t in train_list: try: conn.execute( INSERT OR REPLACE INTO train_schedule (query_date, train_no, from_station, to_station, dep_time, arr_time, duration, biz_seat, first_seat, second_seat, no_seat, query_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( query_date, t[车次号], from_name, to_name, t[发车时间], t[到达时间], t[历时], t[商务座], t[一等座], t[二等座], t[无座], now )) except Exception as exc: print(f写入失败: {exc}) conn.commit()这里用INSERT OR REPLACE处理重复数据同一次查询条件如果再次写入会以最新余票覆盖旧记录。字段里biz_seat和first_seat是字符串类型空值存成空字符串未来按条件筛选时再统一转换。train_no字段在一个查询日期内是唯一的但不同日期会出现相同车次号所以唯一索引必须带上query_date。5.2 写一个最小可独立运行的验证脚本爬虫写完以后要快速验证它还能不能工作不重新跑整个项目最直接的方法是封装一个main()入口。脚本接收日期和站点参数执行完整流程并打印结果数量方便定时任务或手动调用。def main(): session requests.Session() session.headers.update({User-Agent: Mozilla/5.0}) init_session(session) name_to_code, code_to_name get_station_map(session) date_str 2025-06-01 from_station 北京南 to_station 上海虹桥 from_code name_to_code.get(from_station.rstrip(站)) to_code name_to_code.get(to_station.rstrip(站)) result query_train(date_str, from_code, to_code) trains result[data][result] parsed [parse_train(line) for line in trains] print(f查询到 {len(parsed)} 趟车次) conn init_db() save_trains(conn, date_str, from_station, to_station, parsed) conn.close() if __name__ __main__: main()这个脚本把会话初始化、车站表加载、接口查询、解析、落库五件事串成一条线。跑通之后再做定时任务调用就有把握了。利用运维侧的crontab或Windows任务计划程序每天固定时间写一次数据就能积累一份可追溯的车次表。5.3 进阶把线程改成异步IO并接入定时监控线程池能解决并发但换到几十个日期的大批量查询线程切换成本就不划算了。这时候把requests换成httpx.AsyncClient用asyncio协程管理所有请求才是把爬虫性能真正提上去的做法。切换成本不高核心逻辑和query_train类似只是把requests.get换成await client.get。另外在抓取余票的场景里可以对重点车次做秒级轮询监控但务必把轮询间隔控制在15秒以上避免单IP高频请求被封。配合beep提示或Webhook推送你就能从“手动查一遍”升级成“自动盯票”的小工具这也正是本项目的可复用价值。本文还有配套的精品资源点击获取
Python爬虫实战：12306车次查询接口解析与并发实现
简介这是一份面向Python爬虫零基础学习者的项目实战资源以12306网站车次信息为目标场景演示从分析接口、构造请求到解析返回数据并提取车次的完整过程。压缩包共4个文件内含2个Python源码文件、1个课堂笔记文档和1份PPT课件源码划分演示步骤便于对照运行笔记与课件补充了关键知识点与调试思路整体仅165KB下载后即可快速浏览。已有77人参与学习适合配合教程循序练习也可作为课程设计或毕业设计的参考资料。通过该资源读者能掌握真实站点爬取中的请求头处理、数据清洗、结果格式化等实用技巧并为后续扩展余票、票价等爬取功能打下基础。1. 12306车次查询的难点接口远比页面易拿选“12306爬取车次”当Python爬虫项目的人多半是先被自动抢票吸引过来的。但把题目落到“爬取车次”这四个字上之后真正值得做的事其实是把12306的查询接口吃透参数怎么编码、车站代码怎么映射、返回数据怎么解析、并发查询怎么绕过频率限制。这个项目里票据侧的逻辑很少反而把HTTP请求的格式、会话状态依赖、数据结构化这三件事占全了适合作为requests库、异步并发和网络调试的综合练兵场。常见做法是先通过浏览器开发者工具抓出otn/leftTicket/query接口用Python模拟发送查询请求再对返回的字符串数组按字段切分得到车次、出发到达时间、余票和票价信息。整个过程不涉及JS渲染和验证码识别只要把接口字段理清请求头保持和浏览器一致就能稳定拿数据。2. 先拆接口leftTicket/query 的请求结构与返回字段做爬虫的第一步永远不是写代码而是打开浏览器开发者工具用Network面板抓一次真实的查询请求。12306的Web端查询页不会把车次直接渲染在HTML里而是通过XHR异步加载JSON后再拼表。这个XHR就是你唯一需要的接口。2.1 用浏览器抓包确认请求格式在12306余票查询页选择任意日期和区间点击查询后Network面板会多出一条leftTicket/query请求它的完整请求URL大致是https://kyfw.12306.cn/otn/leftTicket/query?leftTicketDTO.train_date2025-06-01leftTicketDTO.from_stationSHHleftTicketDTO.to_stationBJPpurpose_codesADULT这里是GET请求查询条件全部放在Query String里。注意from_station和to_station不是中文站名而是SHH和BJP这样的电报码这属于12306自己定义的车站编码体系后续必须通过车站表转换无法跳过。请求还需要带上Cookie、User-Agent、Referer等请求头其中Cookie要从首页访问后获取否则接口会返回登录跳转或直接拒绝数据。2.2 请求头与Cookie依赖关系12306的车次查询接口本身不需要登录但要求请求方先访问过站点主页拿到会话Cookie后才能正常响应。我用requests库实现时通常会先创建一个requests.Session()用Session保持Cookie状态在真正查询前先GET一次https://kyfw.12306.cn/otn/leftTicket/init让服务端把JSESSIONID下发到本地再携带这个Cookie去请求查询接口。import requests session requests.Session() session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36, Referer: https://kyfw.12306.cn/otn/leftTicket/init, }) def init_session(): # 先访问首页服务端会写入 JSESSIONID 等会话 Cookie resp session.get(https://kyfw.12306.cn/otn/leftTicket/init, timeout10) resp.raise_for_status() def query_train(date_str, from_code, to_code): url https://kyfw.12306.cn/otn/leftTicket/query params { leftTicketDTO.train_date: date_str, leftTicketDTO.from_station: from_code, leftTicketDTO.to_station: to_code, purpose_codes: ADULT, } resp session.get(url, paramsparams, timeout10) resp.raise_for_status() return resp.json()代码里session.get返回的响应对象调用resp.json()是因为12306这个接口返回的是JSON格式外层结构是{httpstatus:200,data:{result:[...]}}。params参数不需要手动做URL编码requests库会处理train_date里的横杠等字符。2.3 返回数据格式与字段位置data.result是一个字符串数组每个元素对应一趟车次但元素不是JSON对象而是一条用竖线分隔的长字符串。你直接用split(|)切分就能按位置取到出发站、到达站、出发时间、到达时间、历时、一等座余票、二等座余票、软卧、硬卧、硬座、无座等字段。这是12306查询接口最容易迷惑新手的点按字段位置依赖字符串结构来取数属于典型的历史遗留格式。def parse_train(raw_line: str): parts raw_line.split(|) train_data { 车次号: parts[2], 出发站码: parts[6], 到达站码: parts[7], 发车时间: parts[8], 到达时间: parts[9], 历时: parts[10], 商务座: parts[32], 一等座: parts[31], 二等座: parts[30], 无座: parts[26], } return train_data这里的parts[2]对应车次号parts[6]和parts[7]是出发、到达站的编码parts[8]和parts[9]是发车和到达时间parts[30]到parts[32]是坐席余票。注意parts里某些字段可能为空串比如无座没有余票时返回空字符串解析时要做好默认值处理否则后续写入数据库会报错。拿到字符串位置规律后这一步实际上就完成了整个爬虫项目里最核心的数据解析工作。提示有时候resp.json()里的data是None多半是Cookie失效或请求频率太高被临时拦了。先检查Session是否正常、是否重开过进程。2.4 purpose_codes 参数的作用purpose_codes控制的是成人和学生两种购票类型取ADULT即可。有些教程里写成0X00那是老版本接口的参数值当前版本用ADULT更稳。这个参数看起来只是多一个字段但在实际爬取中会影响返程余票的字段位置个别情况下成人票和学生票的返回结构会出现细微差异所以别直接拿学生票请求的结果套成人的解析逻辑。3. 车站表与参数构造用 station_name.js 完成三码映射查询接口要求参数值必须是车站电报码但用户习惯输入中文站名。从中文站名到电报码的映射需要从12306的静态资源里获取车站数据这也是整个项目里最容易被遗漏的一步。好在12306把全国车站代码放在一个JS文件里直接用正则提取就能拿到映射表不需要另找数据库导入。3.1 获取并解析 station_name.js这个文件固定在https://kyfw.12306.cn/otn/resources/js/framework/station_name.js内容是一段JavaScript赋值语句变量名为station_names里面用符号分隔的脚本结构封装了多个字段。每个车站的格式是拼音首字母缩写|站名|电报码|拼音|简拼|序号。解析思路是通过正则把所有开头的记录切出来再用|分割字段。import re import requests def load_station_map(session: requests.Session): url https://kyfw.12306.cn/otn/resources/js/framework/station_name.js resp session.get(url, timeout10) resp.encoding utf-8 text resp.text # 取出 分隔的每条记录 pattern r(\w)\|([\u4e00-\u9fa5])\|([A-Z])\|(\w)\|(\w)\|(\d) stations re.findall(pattern, text) name_to_code {} code_to_name {} for item in stations: # item (首字母, 站名, 电报码, 拼音, 简拼, 序号) station_name item[1] station_code item[2] name_to_code[station_name] station_code code_to_name[station_code] station_name return name_to_code, code_to_name正则里[\u4e00-\u9fa5]匹配中文站名[A-Z]匹配电报码必须保证顺序和JS文件里一致否则会提取出错误映射。station_name.js体积大约几百KB每次爬取时动态获取一次即可也可以把解析结果存成本地JSON便于离线查询。3.2 中文站名到电报码的查表转换拿到映射表之后构造查询参数就很简单了。把用户输入的中文站名换成电报码字符串再传给第2章的query_train函数。注意模糊匹配有些站名带“站”字后缀例如“北京南站”在车站表里是“北京南”建议在查表前先做一次去“站”处理避免映射不到。反过来返回结果里的出发站和到达站也是电报码要展示给用户看时再转回中文站名。name_to_code, code_to_name load_station_map(session) from_station 北京南 to_station 上海虹桥 # 去除尾部站字避免匹配失败 from_code name_to_code.get(from_station.rstrip(站)) to_code name_to_code.get(to_station.rstrip(站)) if not from_code or not to_code: raise ValueError(f无法识别车站: {from_station} 或 {to_station}) data query_train(2025-06-01, from_code, to_code) for line in data[data][result]: train parse_train(line) print(train[车次号], train[发车时间], train[到达时间])这里把车站名称当作字典的key查询复杂度是O(1)对后续做并发查询很重要。每次调用前都先判断是否None能直接暴露输入站名不在车站表里的问题。12306的车站表覆盖了全国几千个站但偶尔有新开通的线路车站没更新遇到这种情况自查没有错等官方更新后再抓。3.3 把车站表固化到本地复用每次都去请求station_name.js虽然没问题但项目做大了以后会拖慢初始化速度而且没必要。最常见做法是在首次加载后存成stations.json文件后续启动时直接读取本地文件只有文件缺失时才重新拉取。同时把获取时间也存下来方便定期刷新。import json import os STATION_FILE stations.json def get_station_map(session: requests.Session): if os.path.exists(STATION_FILE): with open(STATION_FILE, r, encodingutf-8) as f: data json.load(f) return data[name_to_code], data[code_to_name] name_to_code, code_to_name load_station_map(session) with open(STATION_FILE, w, encodingutf-8) as f: json.dump({name_to_code: name_to_code, code_to_name: code_to_name}, f, ensure_asciiFalse, indent2) return name_to_code, code_to_name这种本地缓存策略在爬虫工程中很常见不只是12306的项目里适用。把不常变化的静态资源缓存下来为后续并发请求节省时间也让整个脚本的启动流程更干净。station_name.js里的数据一般一个月内不会天天变不用追求每次启动强制更新。提示如果返回的车次列表里出发站和到达站字段拆出来和预期不一致先查code_to_name里是否映射到了正确中文名关注parts[6]和parts[7]到底是当前车次的始发站还是你查询的乘车站12306返回的是区间票池信息不一定等于用户查询条件。4. 并发与反爬把单次查询放大到日期维度单次查询只能拿到某一天某一对车站的车次但实际项目中往往要查多天、多区间。比如帮人盯票时要同时查未来5天从北京到上海的所有车次简单用for循环逐次请求会太慢而且容易被服务端识别为异常流量。这个阶段要把并发设计提上来常见方案是线程池加延迟控制必要时接上代理池。4.1 先用线程池实现日期维度并行Python里做并发请求最简单可控的是concurrent.futures.ThreadPoolExecutor。requests库本身是同步的但在等待网络响应时可以把CPU让给其他线程这是IO密集型任务的正确加速手段。from concurrent.futures import ThreadPoolExecutor, as_completed def query_multiple_dates(date_list, from_code, to_code, max_workers3): results [] def safe_query(date_str): try: resp_data query_train(date_str, from_code, to_code) trains resp_data[data][result] return date_str, trains except Exception as exc: return date_str, ferror: {exc} with ThreadPoolExecutor(max_workersmax_workers) as executor: future_map { executor.submit(safe_query, date_str): date_str for date_str in date_list } for future in as_completed(future_map): date_str, trains future.result() results.append((date_str, trains)) return results这里max_workers设为3既保证并发能提速又不至于触发12306的拦截。如果你拿网络上的“多线程快进”教程比着做把workers调到20以上大概率会遇到请求状态码为403或返回数据为空。12306的频控逻辑不透明稳妥做法是先从3到5个线程起步观察返回正常再慢慢增加。4.2 控制请求频率延时、UA池与IP代理只靠线程池不足以长时间稳定运行更稳妥的工程化方案是组合使用三件套请求间随机延时、轮换User-Agent、维护可用代理池。特别是做多日期批量爬取时如果所有请求都来自同一个IP且无间隔很容易被服务端识别为脚本行为导致接口短暂失效。import random import time USER_AGENTS [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36, ] def build_session_with_ua(): s requests.Session() s.headers.update({ User-Agent: random.choice(USER_AGENTS), Referer: https://kyfw.12306.cn/otn/leftTicket/init, }) return s def safe_request_with_retry(session, url, params, retries3): for attempt in range(retries): try: resp session.get(url, paramsparams, timeout10) if resp.status_code 200: return resp.json() except requests.RequestException: pass time.sleep(1 attempt * 2) return NoneUSER_AGENTS至少准备3个主流浏览器版本轮换使用能降低被指纹识别锁定的概率。safe_request_with_retry里的退避重试策略是应对偶发网络抖动的兜底方案。如果你用云服务器跑爬虫单IP请求量较大时最有效的手段是接入付费代理IP池按请求维度轮换出口IP。代理配置在requests里只需要给session.proxies赋值即可但代理质量参差不齐失效IP会导致连接超时增加需要配代理检测逻辑。4.3 验证并发改造后的效果做完并发以后至少要从三个维度验证返回车次数量是否正确、多天结果是否有缺失、请求成功率有没有明显下降。查询北京南到上海虹桥、未来5天的车次正常情况下每天都有几十趟高铁如果某天返回的result数组长度明显偏小大概率是请求被限流或解析错位。日期车次数目耗时(秒)状态2025-06-01460.8正常2025-06-02440.9正常2025-06-03402.1接口偶发超时2025-06-04450.7正常上表是一次典型的并发测试结果。max_workers3时总耗时约5秒左右如果换成单线程顺序请求同样的数据量至少要15到20秒。2025-06-03那次耗时偏高是重试机制在起作用。对整个项目而言能接受这种零散波动只要没有连续大面积失败就行。5. 数据落库与结果验证从临时打印到可复现的查询工具最后一层要做的事情是让爬取结果真正能留痕而不仅是打印在控制台里。车次数据的特点是高度文本化、字段位置固定且存在大量空值把它写进SQLite后就能做条件筛选和趋势分析。这块工作不复杂但直接影响整个项目能不能从“能跑”变成“能用”。5.1 用SQLite保存车次明细并去重12306车次数据具有时效性同一日期同一车次的余票数量会变化所以落库时不能简单插入。常见设计是加一个query_time字段记录查询时间同时以车次号日期出发站到达站作为唯一索引来避免重复行累积。import sqlite3 from datetime import datetime def init_db(): conn sqlite3.connect(trains.db) conn.execute( CREATE TABLE IF NOT EXISTS train_schedule ( id INTEGER PRIMARY KEY AUTOINCREMENT, query_date TEXT, train_no TEXT, from_station TEXT, to_station TEXT, dep_time TEXT, arr_time TEXT, duration TEXT, biz_seat TEXT, first_seat TEXT, second_seat TEXT, no_seat TEXT, query_time TEXT, UNIQUE(query_date, train_no, from_station, to_station) ) ) conn.commit() return conn def save_trains(conn, query_date, from_name, to_name, train_list): now datetime.now().strftime(%Y-%m-%d %H:%M:%S) for t in train_list: try: conn.execute( INSERT OR REPLACE INTO train_schedule (query_date, train_no, from_station, to_station, dep_time, arr_time, duration, biz_seat, first_seat, second_seat, no_seat, query_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( query_date, t[车次号], from_name, to_name, t[发车时间], t[到达时间], t[历时], t[商务座], t[一等座], t[二等座], t[无座], now )) except Exception as exc: print(f写入失败: {exc}) conn.commit()这里用INSERT OR REPLACE处理重复数据同一次查询条件如果再次写入会以最新余票覆盖旧记录。字段里biz_seat和first_seat是字符串类型空值存成空字符串未来按条件筛选时再统一转换。train_no字段在一个查询日期内是唯一的但不同日期会出现相同车次号所以唯一索引必须带上query_date。5.2 写一个最小可独立运行的验证脚本爬虫写完以后要快速验证它还能不能工作不重新跑整个项目最直接的方法是封装一个main()入口。脚本接收日期和站点参数执行完整流程并打印结果数量方便定时任务或手动调用。def main(): session requests.Session() session.headers.update({User-Agent: Mozilla/5.0}) init_session(session) name_to_code, code_to_name get_station_map(session) date_str 2025-06-01 from_station 北京南 to_station 上海虹桥 from_code name_to_code.get(from_station.rstrip(站)) to_code name_to_code.get(to_station.rstrip(站)) result query_train(date_str, from_code, to_code) trains result[data][result] parsed [parse_train(line) for line in trains] print(f查询到 {len(parsed)} 趟车次) conn init_db() save_trains(conn, date_str, from_station, to_station, parsed) conn.close() if __name__ __main__: main()这个脚本把会话初始化、车站表加载、接口查询、解析、落库五件事串成一条线。跑通之后再做定时任务调用就有把握了。利用运维侧的crontab或Windows任务计划程序每天固定时间写一次数据就能积累一份可追溯的车次表。5.3 进阶把线程改成异步IO并接入定时监控线程池能解决并发但换到几十个日期的大批量查询线程切换成本就不划算了。这时候把requests换成httpx.AsyncClient用asyncio协程管理所有请求才是把爬虫性能真正提上去的做法。切换成本不高核心逻辑和query_train类似只是把requests.get换成await client.get。另外在抓取余票的场景里可以对重点车次做秒级轮询监控但务必把轮询间隔控制在15秒以上避免单IP高频请求被封。配合beep提示或Webhook推送你就能从“手动查一遍”升级成“自动盯票”的小工具这也正是本项目的可复用价值。本文还有配套的精品资源点击获取
