1. 项目缘起为什么从12306开始学爬虫很多朋友想学Python爬虫但一上来就找各种高难度网站结果被复杂的反爬机制劝退最后不了了之。我建议不如从一个“既熟悉又有点挑战”的目标开始——12306。说它熟悉是因为几乎每个人都用过说它有挑战是因为它确实有基础的防护但又不像某些大型电商平台那样铜墙铁壁非常适合新手在实战中理解爬虫的核心逻辑和常见障碍。这个项目的目标很明确写一个Python脚本自动查询指定日期、车次、区间的余票信息并把结果保存下来。更进一步我们还要对这些爬取到的“原始”数据进行清洗、整理和分析比如计算热门车次的余票变化趋势或者筛选出符合我们出行时间的最优车次。这整个过程恰好覆盖了数据工作的完整链路数据采集爬虫 - 数据清洗与处理 - 数据分析/应用。掌握了这一套你不仅学会了爬虫更入门了数据处理价值远超一个简单的查询工具。我见过不少教程只讲到“把数据打印出来”就结束了留下的是一堆杂乱无章的文本毫无用处。这次我会带你走完全程重点分享我在处理12306这类动态渲染、带有参数验证的网站时如何一步步分析请求、构造参数、应对小坑以及如何用pandas把爬下来的数据变成整洁的表格甚至做点简单的可视化。你会发现爬虫不是魔法数据处理也不神秘它们都是一步一步的逻辑操作。2. 战前准备理解12306的“游戏规则”与工具选择在动手写代码之前我们必须花点时间搞清楚目标网站是怎么工作的。盲目地请求只会得到一堆“请求异常”的提示。2.1 12306查询接口的核心机制打开12306官网kyfw.12306.cn手动进行一次车票查询同时打开浏览器的“开发者工具”F12切换到“网络”(Network)选项卡。你会发现点击“查询”按钮后浏览器并非直接跳转一个新页面而是向后端发送了一个HTTPS POST请求。这个请求的URL通常是类似于https://kyfw.12306.cn/otn/leftTicket/queryX这样的地址。关键在于它的请求参数和请求头。关键参数分析请求会携带一串经过编码的查询参数通常包括leftTicketDTO.train_date: 出发日期格式为YYYY-MM-DD。leftTicketDTO.from_station: 出发站的电报码如“北京”是BJP。leftTicketDTO.to_station: 到达站的电报码如“上海”是SHH。purpose_codes: 票种成人票通常是ADULT。这里最大的“坑”就是车站电报码。12306内部不使用中文站名通信而是用一套特定的缩写代码。我们必须先获取到“北京-BJP”这样的映射关系。请求头(Headers)的重要性12306会校验一些基本的请求头以判断请求是否来自“像样”的浏览器。其中User-Agent是最基本的我们需要把它设置成一个常见的浏览器标识。Referer来源页有时也需要正确设置表明请求是从官网页面发起的。响应数据格式查询接口返回的是JSON格式的数据结构清晰易于解析。数据中包含了车次列表每个车次的信息如列车号、出发到达时间、历时、各席别余票、价格等被压缩在一个长长的字符串里不同字段间以|竖线分隔。我们需要按照固定的位置索引去“拆解”这个字符串。2.2 开发环境与库的选择工欲善其事必先利其器。我们选择最主流、最稳定的组合。Python环境建议使用Python 3.8及以上版本。新手推荐通过Anaconda安装它集成了很多科学计算库包括我们后面要用的pandas。集成开发环境(IDE)VSCode或PyCharm均可。VSCode轻量灵活安装Python插件和Pylance插件后体验很好。PyCharm是专业IDE对项目管理、调试支持更完善。核心Python库requests: 用于发送HTTP请求获取网页数据。这是爬虫的基石库务必熟练掌握。pandas: 数据处理和分析的核心库。我们将用它来清洗、转换和保存爬取的数据。json: Python标准库用于解析接口返回的JSON数据。安装这些库非常简单在命令行终端中执行以下命令即可pip install requests pandas注意请务必使用国内镜像源来加速安装例如清华源pip install requests pandas -i https://pypi.tuna.tsinghua.edu.cn/simple。这能避免因网络问题导致的安装失败。3. 实战爬取从分析到获取原始数据现在我们开始动手编写爬虫的核心部分。整个过程可以分解为三个关键步骤。3.1 第一步获取车站电报码映射表这是整个爬虫能正确运行的“地基”。我们需要先从一个接口获取全国车站名与电报码的对应关系。12306提供了一个静态JSON文件通常地址是https://kyfw.12306.cn/otn/resources/js/framework/station_name.js?station_version1.xxxx。这个文件内容是一个巨大的字符串格式如bjb|北京北|VAP|...。我们需要编写一个函数来下载并解析这个文件import requests import re def get_station_code(): 获取12306车站名与电报码的映射字典。 返回格式{北京北: VAP, 北京: BJP, ...} url https://kyfw.12306.cn/otn/resources/js/framework/station_name.js # 添加一个User-Agent模拟浏览器 headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36 } try: response requests.get(url, headersheaders, timeout10) response.encoding utf-8 # 使用正则表达式提取车站数据部分 # 文件内容类似var station_names bjb|北京北|VAP|beijingbei|...; station_str re.findall(rvar station_names \(.*?)\;, response.text)[0] station_list station_str.split()[1:] # 以分割第一个元素是空的 station_dict {} for station in station_list: items station.split(|) if len(items) 3: # items[1]是中文站名 items[2]是电报码 station_dict[items[1]] items[2] print(f成功获取 {len(station_dict)} 个车站信息) return station_dict except Exception as e: print(f获取车站信息失败: {e}) return {} # 测试一下 stations get_station_code() print(f北京站的电报码是{stations.get(北京)}) print(f上海站的电报码是{stations.get(上海)})实操心得这个映射表最好在程序开始时获取一次并保存到本地文件如stations.json或内存中避免每次查询都去请求提高效率且避免给服务器造成不必要的压力。3.2 第二步构造查询请求并获取响应有了电报码我们就可以构造真实的查询请求了。这里需要仔细模仿浏览器发送的请求。import requests import json from datetime import datetime def query_tickets(from_station_name, to_station_name, train_date): 查询指定日期、区间的车票信息。 :param from_station_name: 出发站中文名如北京 :param to_station_name: 到达站中文名如上海 :param train_date: 出发日期格式YYYY-MM-DD如2023-10-01 :return: 包含车票信息的列表如果失败返回None # 1. 获取车站电报码 stations get_station_code() # 这里可以优化为从内存或文件读取 from_station_code stations.get(from_station_name) to_station_code stations.get(to_station_name) if not from_station_code or not to_station_code: print(f错误未找到车站电报码。请检查车站名‘{from_station_name}’或‘{to_station_name}’是否正确。) return None # 2. 构造请求URL和参数 url https://kyfw.12306.cn/otn/leftTicket/query # 关键参数 params { leftTicketDTO.train_date: train_date, leftTicketDTO.from_station: from_station_code, leftTicketDTO.to_station: to_station_code, purpose_codes: ADULT } # 3. 构造请求头模拟浏览器 headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36, Referer: https://kyfw.12306.cn/otn/leftTicket/init, # 表明请求来自官网查询页面 X-Requested-With: XMLHttpRequest # 表明是Ajax请求 } # 4. 发送请求 try: # 注意这里是GET请求参数放在URL查询字符串中 response requests.get(url, headersheaders, paramsparams, timeout15) response.encoding utf-8 # 5. 检查响应状态和内容 if response.status_code 200: result_json response.json() # 12306返回的数据有一个状态码字段 if result_json.get(httpstatus) 200 and result_json.get(data): print(f查询成功共找到 {len(result_json[data][result])} 个车次。) return result_json[data][result] # 返回原始结果列表 else: print(f接口返回数据异常: {result_json.get(messages, 未知错误)}) return None else: print(f网络请求失败状态码: {response.status_code}) return None except requests.exceptions.Timeout: print(请求超时请检查网络或稍后重试。) return None except requests.exceptions.RequestException as e: print(f网络请求发生异常: {e}) return None except json.JSONDecodeError: print(响应内容不是有效的JSON格式可能页面结构已变化。) return None # 测试查询 raw_trains query_tickets(北京, 上海, 2023-10-01) if raw_trains: # 先打印第一条数据看看结构 print(第一条原始数据样例) print(raw_trains[0])踩坑记录这里最容易出错的有两点。一是车站名不对比如输入“北京市”而不是“北京”导致查不到电报码。二是日期格式必须是YYYY-MM-DD并且不能查询过去很久的日期或太远的未来日期12306预售期有限制。务必在代码中加入严格的校验和友好的错误提示。3.3 第三步解析复杂的原始数据字符串上一步返回的raw_trains列表里每个元素都是一个字符串车次的所有信息都用|拼接在一起。这个字符串的字段顺序是固定的但不同车次类型高铁、动车、普速的字段数可能略有差异。我们需要找到目标字段的索引。def parse_train_info(raw_train_str): 解析单个车次的原始字符串数据。 :param raw_train_str: 包含车次信息的原始字符串 :return: 解析后的字典包含关键信息 # 将字符串按竖线分割成列表 fields raw_train_str.split(|) # 关键字段的索引位置这个索引相对稳定但若12306更新接口可能需要调整 # 可以通过对比官网显示的数据和这个列表来确认索引 train_info { 车次: fields[3], # 索引3是车次号如G1 出发站: fields[6], # 索引6是出发站电报码 到达站: fields[7], # 索引7是到达站电报码 出发时间: fields[8], # 索引8是出发时间 到达时间: fields[9], # 索引9是到达时间 历时: fields[10], # 索引10是运行时间 商务座/特等座: fields[32] or fields[25], # 索引32是商务座某些车次是25 一等座: fields[31], # 索引31是一等座 二等座/二等包座: fields[30], # 索引30是二等座 高级软卧: fields[21], # 索引21是高级软卧 软卧/一等卧: fields[23], # 索引23是软卧 动卧: fields[33], # 索引33是动卧 硬卧/二等卧: fields[28], # 索引28是硬卧 软座: fields[24], # 索引24是软座 硬座: fields[29], # 索引29是硬座 无座: fields[26], # 索引26是无座 其他: fields[22], # 索引22是其他 备注: fields[1] if len(fields) 1 else # 索引1有时是备注信息如“列车停运” } # 处理余票显示有数字表示票数有表示有票但数量未知或None表示无票 for seat_type in [商务座/特等座, 一等座, 二等座/二等包座, 高级软卧, 软卧/一等座, 动卧, 硬卧/二等卧, 软座, 硬座, 无座]: ticket_info train_info[seat_type] if ticket_info or ticket_info 无 or ticket_info is None: train_info[seat_type] 无票 elif ticket_info.isdigit(): train_info[seat_type] f{ticket_info}张 # 其他情况如‘有’保持不变 return train_info # 测试解析函数 if raw_trains: parsed_train parse_train_info(raw_trains[0]) print(\n解析后的第一条车次信息) for key, value in parsed_train.items(): print(f{key}: {value})重要提示字段索引不是一成不变的虽然目前基于广泛使用的公开索引相对稳定但12306有可能调整接口返回的数据结构。最可靠的方法是在浏览器开发者工具中找到查询接口返回的JSON数据复制一条结果字符串数一数你关心的信息如“二等座”前面有多少个|。这是爬虫工程师的必备调试技能。4. 数据处理用Pandas将杂乱数据变废为宝爬取到的数据还是原始的、杂乱的列表。我们需要将其转化为结构清晰、便于分析的表格DataFrame并进行清洗和整理。4.1 数据清洗与结构化我们将所有解析后的车次信息存入一个列表然后用pandas的DataFrame来承接。import pandas as pd def raw_data_to_dataframe(raw_trains_list, station_dict): 将原始车次数据列表转换为pandas DataFrame并进行初步清洗。 :param raw_trains_list: query_tickets函数返回的原始列表 :param station_dict: 车站电报码映射字典 :return: 清洗后的DataFrame if not raw_trains_list: print(没有车次数据可处理。) return pd.DataFrame() parsed_list [] for raw_train in raw_trains_list: try: train_dict parse_train_info(raw_train) parsed_list.append(train_dict) except IndexError as e: # 如果解析出错可能是字段索引对不上跳过该条数据并记录 print(f解析车次数据时出错可能数据结构有变跳过: {e}) continue if not parsed_list: print(所有数据解析均失败。) return pd.DataFrame() # 创建DataFrame df pd.DataFrame(parsed_list) # 数据清洗步骤 # 1. 将电报码转换回中文站名 # 反转车站字典从{‘北京’‘BJP’} 变成 {‘BJP’‘北京’} code_to_name {v: k for k, v in station_dict.items()} df[出发站] df[出发站].map(code_to_name).fillna(df[出发站]) # 如果映射失败保留原值 df[到达站] df[到达站].map(code_to_name).fillna(df[到达站]) # 2. 处理时间列方便后续按时间排序或计算 # 将‘出发时间’和‘到达时间’转换为datetime.time类型如果只关心时间 # 注意这里缺少日期信息所以用time类型 df[出发时间] pd.to_datetime(df[出发时间], format%H:%M, errorscoerce).dt.time df[到达时间] pd.to_datetime(df[到达时间], format%H:%M, errorscoerce).dt.time # 3. 将‘历时’如‘04:18’转换为分钟数便于比较 def duration_to_minutes(dur_str): try: if : in dur_str: h, m map(int, dur_str.split(:)) return h * 60 m else: return None except: return None df[历时_分钟] df[历时].apply(duration_to_minutes) # 4. 重命名列使其更直观 df.rename(columns{ 车次: train_no, 出发站: from_station, 到达站: to_station, 出发时间: departure_time, 到达时间: arrival_time, 历时: duration, 历时_分钟: duration_min, 商务座/特等座: business_seat, 一等座: first_class, 二等座/二等包座: second_class, 高级软卧: advanced_soft_sleeper, 软卧/一等卧: soft_sleeper, 动卧: motor_sleeper, 硬卧/二等卧: hard_sleeper, 软座: soft_seat, 硬座: hard_seat, 无座: no_seat, 备注: notes }, inplaceTrue) # 5. 按出发时间排序 df.sort_values(bydeparture_time, inplaceTrue) df.reset_index(dropTrue, inplaceTrue) # 重置索引 print(f数据处理完成共 {len(df)} 条有效车次记录。) return df # 整合测试 stations get_station_code() raw_data query_tickets(北京, 上海, 2023-10-01) if raw_data: df_trains raw_data_to_dataframe(raw_data, stations) # 查看前几行数据 print(\n处理后的数据前5行) print(df_trains.head().to_string()) # 使用to_string避免显示省略号 # 查看数据基本信息 print(\n数据概览) print(df_trains.info())经验之谈在数据清洗时一定要做好异常处理try...except和缺失值填充fillna。网络爬取的数据质量不可控可能某些字段为空或格式异常。我们的目标是让程序健壮地运行下去而不是因为一条脏数据而崩溃。将电报码转回中文站名这一步极大地提升了数据的可读性。4.2 数据筛选与简单分析有了整洁的DataFrame我们就可以像操作Excel表格一样进行各种查询和分析。def analyze_ticket_data(df): 对车票DataFrame进行简单的分析和筛选。 if df.empty: print(没有数据可供分析。) return print( 基础统计 ) print(f总车次数量: {len(df)}) # 1. 筛选有二等座余票的车次 # 注意我们的‘second_class’列现在是‘有’、‘无票’或‘X张’这样的字符串 # 我们需要筛选出不是‘无票’的 df_with_ticket df[df[second_class] ! 无票] print(f\n拥有二等座余票的车次数量: {len(df_with_ticket)}) # 2. 找出最早和最晚出发的车次 earliest_train df.loc[df[departure_time].idxmin()] latest_train df.loc[df[departure_time].idxmax()] print(f\n最早出发车次: {earliest_train[train_no]} 于 {earliest_train[departure_time]}) print(f最晚出发车次: {latest_train[train_no]} 于 {latest_train[departure_time]}) # 3. 找出运行时间最短最快的车次 # 先过滤掉历时为空的数据 df_valid_duration df.dropna(subset[duration_min]) if not df_valid_duration.empty: fastest_train df_valid_duration.loc[df_valid_duration[duration_min].idxmin()] print(f\n运行时间最短的车次: {fastest_train[train_no]}, 历时 {fastest_train[duration]} ({fastest_train[duration_min]}分钟)) # 4. 按车次类型统计例如G字头为高铁D字头为动车 df[train_type] df[train_no].str[0] # 提取车次号首字母 type_counts df[train_type].value_counts() print(f\n各类型车次统计:) for train_type, count in type_counts.items(): print(f {train_type}字头: {count} 趟) # 5. 创建一个更直观的余票情况视图 # 我们只关心几个主要席别 seat_columns [business_seat, first_class, second_class, soft_sleeper, hard_sleeper] print(f\n部分车次主要席别余票情况预览:) # 选择前10趟车展示关键信息 preview_df df[[train_no, departure_time, arrival_time, duration] seat_columns].head(10) print(preview_df.to_string(indexFalse)) # 不显示行索引 return df_with_ticket # 执行分析 if not df_trains.empty: available_trains_df analyze_ticket_data(df_trains)这个分析函数展示了pandas的基本能力布尔索引筛选、按列计算极值、字符串操作、分组统计。你可以根据自己的需求轻松地修改和扩展这些分析逻辑比如筛选“上午出发的车次”、“有卧铺的车次”等等。4.3 数据持久化保存到文件处理好的数据当然要保存下来方便日后查看或进行更深入的分析比如对比不同日期的余票情况。def save_data_to_file(df, filename_prefixtrain_tickets): 将DataFrame保存到CSV和Excel文件。 :param df: 要保存的DataFrame :param filename_prefix: 文件名前缀会自动加上日期时间戳 if df.empty: print(没有数据可保存。) return # 生成带时间戳的文件名避免覆盖 from datetime import datetime timestamp datetime.now().strftime(%Y%m%d_%H%M%S) csv_filename f{filename_prefix}_{timestamp}.csv excel_filename f{filename_prefix}_{timestamp}.xlsx try: # 保存为CSV通用性好体积小 df.to_csv(csv_filename, indexFalse, encodingutf-8-sig) # utf-8-sig解决Excel打开中文乱码 print(f数据已保存为CSV文件: {csv_filename}) # 保存为Excel格式美观支持多工作表 # 需要安装 openpyxl 库: pip install openpyxl df.to_excel(excel_filename, indexFalse, sheet_name车票信息) print(f数据已保存为Excel文件: {excel_filename}) except Exception as e: print(f保存文件时出错: {e}) # 保存我们处理好的数据 if not df_trains.empty: save_data_to_file(df_trains, beijing_shanghai_tickets)小技巧保存为CSV是通用性最好的选择任何文本编辑器都能打开。保存为Excel则更适合需要手动查看、标注或分享给非技术同事的场景。使用utf-8-sig编码可以确保用Excel打开CSV文件时中文字符正常显示。5. 进阶与避坑让爬虫更稳健、更高效一个只能运行一次的脚本是玩具一个能持续稳定运行的脚本才是工具。下面分享几个让脚本更实用的进阶技巧和常见坑的解决方法。5.1 应对请求频率限制与超时12306对频繁访问会有一定的限制。在脚本中直接使用requests.get()而不加控制短时间内大量请求可能导致IP被暂时屏蔽。添加延时在连续请求之间使用time.sleep()添加随机延时模拟人工操作。import time import random # 在关键请求如循环查询多个日期后添加延时 time.sleep(random.uniform(1, 3)) # 随机休眠1到3秒使用Session对象requests.Session()可以复用TCP连接并在会话级别保持一些头信息和cookies效率更高也更接近浏览器行为。session requests.Session() session.headers.update({ User-Agent: 你的User-Agent, Accept-Language: zh-CN,zh;q0.9, }) # 后续所有请求都用 session.get() 或 session.post() response session.get(url, paramsparams, timeout10)设置合理超时网络环境复杂必须设置timeout参数如timeout(5, 15)分别代表连接超时和读取超时避免程序因网络问题无限挂起。5.2 处理动态变化的查询参数有时查询接口需要一些动态参数如_json_att或一个加密的RAIL_DEVICEID。这些参数通常可以在首页的HTML源码或某个JS文件里找到或者通过首次访问其他接口获得。应对策略先访问一次首页用Session对象先请求一下https://kyfw.12306.cn/otn/leftTicket/init让Session自动管理cookies。提取动态参数如果发现查询请求需要额外的参数仔细分析浏览器首次加载页面时发起的其他请求看是否有获取这些参数的接口。然后先请求那个接口提取参数再拼接到主查询请求中。保持会话坚持使用同一个Session对象进行所有关联请求是处理这类依赖cookies或token的网站最有效的方法。5.3 错误处理与日志记录一个健壮的程序必须能妥善处理各种异常并记录下发生了什么方便排查问题。import logging # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(ticket_crawler.log, encodingutf-8), logging.StreamHandler() # 同时输出到控制台 ] ) logger logging.getLogger(__name__) def robust_query_tickets(from_station, to_station, train_date, retry_times2): 带重试和错误处理的查询函数。 for attempt in range(retry_times 1): try: logger.info(f尝试第 {attempt 1} 次查询 {from_station} - {to_station} 在 {train_date}) result query_tickets(from_station, to_station, train_date) # 调用我们之前写的函数 if result is not None: return result else: logger.warning(f第 {attempt 1} 次查询返回空结果。) except requests.exceptions.Timeout: logger.error(f第 {attempt 1} 次查询超时。) except Exception as e: logger.exception(f第 {attempt 1} 次查询发生未知错误: {e}) # 会记录完整的异常堆栈 if attempt retry_times: wait_time (attempt 1) * 3 # 重试等待时间递增 logger.info(f等待 {wait_time} 秒后重试...) time.sleep(wait_time) logger.error(f查询 {from_station} - {to_station} 在 {train_date} 失败已达最大重试次数。) return None将核心函数用try...except包裹并记录日志这样当脚本在后台运行时我们也能通过查看日志文件ticket_crawler.log来了解运行状态和错误原因。5.4 将脚本模块化与函数化随着功能增多把所有代码堆在一个文件里会难以维护。最佳实践是将代码模块化config.py: 存放常量如请求头模板、URL、字段索引映射。station_manager.py: 负责车站信息的获取、缓存和查询。crawler.py: 包含核心的查询和解析函数。data_processor.py: 包含数据清洗、分析和保存的函数。main.py: 主程序入口组织调用逻辑。这样不仅代码清晰也方便复用。例如station_manager模块可以设计为首次运行时下载车站信息并保存到本地文件下次启动时直接读取本地文件大大提升速度。6. 项目总结与扩展思路走到这里你已经完成了一个完整的、具有实用价值的12306车票查询与数据处理脚本。它不仅能查票还能把结果规规矩矩地整理好、分析好、保存好。这个过程里最重要的不是代码本身而是分析问题、拆解步骤、调试和解决问题的思路。我个人在多次编写类似爬虫后最深的体会是耐心和细致比技术更重要。90%的时间花在分析网络请求、理解数据格式、处理边界情况上。写代码可能只要半小时但让代码稳定可靠地运行需要数倍的调试和优化时间。这个项目还有很大的扩展空间你可以尝试多日期/多车次批量查询写一个循环爬取未来一周内每天的车票信息分析余票释放规律。价格信息获取查询接口返回的数据中通常包含票价信息在另一个字段串中可以尝试解析并计算均价、最贵/最便宜车次。简单可视化用matplotlib或seaborn将热门车次的出发时间分布、余票数量趋势画成图表。封装成GUI工具或Web服务使用tkinter、PyQt做一个桌面小工具或者用Flask/FastAPI做一个简单的查询网页分享给不会编程的朋友用。接入通知当出现符合你条件的车票如下午出发、有二等座时自动发送邮件或微信消息提醒。记住爬虫技术应当用在学习和提升效率上务必遵守网站的robots.txt协议控制请求频率不要对目标服务器造成压力。通过这个项目打下的基础你可以去探索更广阔的数据世界。 Python爬虫实战：从12306车票查询到数据处理全流程解析
1. 项目缘起为什么从12306开始学爬虫很多朋友想学Python爬虫但一上来就找各种高难度网站结果被复杂的反爬机制劝退最后不了了之。我建议不如从一个“既熟悉又有点挑战”的目标开始——12306。说它熟悉是因为几乎每个人都用过说它有挑战是因为它确实有基础的防护但又不像某些大型电商平台那样铜墙铁壁非常适合新手在实战中理解爬虫的核心逻辑和常见障碍。这个项目的目标很明确写一个Python脚本自动查询指定日期、车次、区间的余票信息并把结果保存下来。更进一步我们还要对这些爬取到的“原始”数据进行清洗、整理和分析比如计算热门车次的余票变化趋势或者筛选出符合我们出行时间的最优车次。这整个过程恰好覆盖了数据工作的完整链路数据采集爬虫 - 数据清洗与处理 - 数据分析/应用。掌握了这一套你不仅学会了爬虫更入门了数据处理价值远超一个简单的查询工具。我见过不少教程只讲到“把数据打印出来”就结束了留下的是一堆杂乱无章的文本毫无用处。这次我会带你走完全程重点分享我在处理12306这类动态渲染、带有参数验证的网站时如何一步步分析请求、构造参数、应对小坑以及如何用pandas把爬下来的数据变成整洁的表格甚至做点简单的可视化。你会发现爬虫不是魔法数据处理也不神秘它们都是一步一步的逻辑操作。2. 战前准备理解12306的“游戏规则”与工具选择在动手写代码之前我们必须花点时间搞清楚目标网站是怎么工作的。盲目地请求只会得到一堆“请求异常”的提示。2.1 12306查询接口的核心机制打开12306官网kyfw.12306.cn手动进行一次车票查询同时打开浏览器的“开发者工具”F12切换到“网络”(Network)选项卡。你会发现点击“查询”按钮后浏览器并非直接跳转一个新页面而是向后端发送了一个HTTPS POST请求。这个请求的URL通常是类似于https://kyfw.12306.cn/otn/leftTicket/queryX这样的地址。关键在于它的请求参数和请求头。关键参数分析请求会携带一串经过编码的查询参数通常包括leftTicketDTO.train_date: 出发日期格式为YYYY-MM-DD。leftTicketDTO.from_station: 出发站的电报码如“北京”是BJP。leftTicketDTO.to_station: 到达站的电报码如“上海”是SHH。purpose_codes: 票种成人票通常是ADULT。这里最大的“坑”就是车站电报码。12306内部不使用中文站名通信而是用一套特定的缩写代码。我们必须先获取到“北京-BJP”这样的映射关系。请求头(Headers)的重要性12306会校验一些基本的请求头以判断请求是否来自“像样”的浏览器。其中User-Agent是最基本的我们需要把它设置成一个常见的浏览器标识。Referer来源页有时也需要正确设置表明请求是从官网页面发起的。响应数据格式查询接口返回的是JSON格式的数据结构清晰易于解析。数据中包含了车次列表每个车次的信息如列车号、出发到达时间、历时、各席别余票、价格等被压缩在一个长长的字符串里不同字段间以|竖线分隔。我们需要按照固定的位置索引去“拆解”这个字符串。2.2 开发环境与库的选择工欲善其事必先利其器。我们选择最主流、最稳定的组合。Python环境建议使用Python 3.8及以上版本。新手推荐通过Anaconda安装它集成了很多科学计算库包括我们后面要用的pandas。集成开发环境(IDE)VSCode或PyCharm均可。VSCode轻量灵活安装Python插件和Pylance插件后体验很好。PyCharm是专业IDE对项目管理、调试支持更完善。核心Python库requests: 用于发送HTTP请求获取网页数据。这是爬虫的基石库务必熟练掌握。pandas: 数据处理和分析的核心库。我们将用它来清洗、转换和保存爬取的数据。json: Python标准库用于解析接口返回的JSON数据。安装这些库非常简单在命令行终端中执行以下命令即可pip install requests pandas注意请务必使用国内镜像源来加速安装例如清华源pip install requests pandas -i https://pypi.tuna.tsinghua.edu.cn/simple。这能避免因网络问题导致的安装失败。3. 实战爬取从分析到获取原始数据现在我们开始动手编写爬虫的核心部分。整个过程可以分解为三个关键步骤。3.1 第一步获取车站电报码映射表这是整个爬虫能正确运行的“地基”。我们需要先从一个接口获取全国车站名与电报码的对应关系。12306提供了一个静态JSON文件通常地址是https://kyfw.12306.cn/otn/resources/js/framework/station_name.js?station_version1.xxxx。这个文件内容是一个巨大的字符串格式如bjb|北京北|VAP|...。我们需要编写一个函数来下载并解析这个文件import requests import re def get_station_code(): 获取12306车站名与电报码的映射字典。 返回格式{北京北: VAP, 北京: BJP, ...} url https://kyfw.12306.cn/otn/resources/js/framework/station_name.js # 添加一个User-Agent模拟浏览器 headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36 } try: response requests.get(url, headersheaders, timeout10) response.encoding utf-8 # 使用正则表达式提取车站数据部分 # 文件内容类似var station_names bjb|北京北|VAP|beijingbei|...; station_str re.findall(rvar station_names \(.*?)\;, response.text)[0] station_list station_str.split()[1:] # 以分割第一个元素是空的 station_dict {} for station in station_list: items station.split(|) if len(items) 3: # items[1]是中文站名 items[2]是电报码 station_dict[items[1]] items[2] print(f成功获取 {len(station_dict)} 个车站信息) return station_dict except Exception as e: print(f获取车站信息失败: {e}) return {} # 测试一下 stations get_station_code() print(f北京站的电报码是{stations.get(北京)}) print(f上海站的电报码是{stations.get(上海)})实操心得这个映射表最好在程序开始时获取一次并保存到本地文件如stations.json或内存中避免每次查询都去请求提高效率且避免给服务器造成不必要的压力。3.2 第二步构造查询请求并获取响应有了电报码我们就可以构造真实的查询请求了。这里需要仔细模仿浏览器发送的请求。import requests import json from datetime import datetime def query_tickets(from_station_name, to_station_name, train_date): 查询指定日期、区间的车票信息。 :param from_station_name: 出发站中文名如北京 :param to_station_name: 到达站中文名如上海 :param train_date: 出发日期格式YYYY-MM-DD如2023-10-01 :return: 包含车票信息的列表如果失败返回None # 1. 获取车站电报码 stations get_station_code() # 这里可以优化为从内存或文件读取 from_station_code stations.get(from_station_name) to_station_code stations.get(to_station_name) if not from_station_code or not to_station_code: print(f错误未找到车站电报码。请检查车站名‘{from_station_name}’或‘{to_station_name}’是否正确。) return None # 2. 构造请求URL和参数 url https://kyfw.12306.cn/otn/leftTicket/query # 关键参数 params { leftTicketDTO.train_date: train_date, leftTicketDTO.from_station: from_station_code, leftTicketDTO.to_station: to_station_code, purpose_codes: ADULT } # 3. 构造请求头模拟浏览器 headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36, Referer: https://kyfw.12306.cn/otn/leftTicket/init, # 表明请求来自官网查询页面 X-Requested-With: XMLHttpRequest # 表明是Ajax请求 } # 4. 发送请求 try: # 注意这里是GET请求参数放在URL查询字符串中 response requests.get(url, headersheaders, paramsparams, timeout15) response.encoding utf-8 # 5. 检查响应状态和内容 if response.status_code 200: result_json response.json() # 12306返回的数据有一个状态码字段 if result_json.get(httpstatus) 200 and result_json.get(data): print(f查询成功共找到 {len(result_json[data][result])} 个车次。) return result_json[data][result] # 返回原始结果列表 else: print(f接口返回数据异常: {result_json.get(messages, 未知错误)}) return None else: print(f网络请求失败状态码: {response.status_code}) return None except requests.exceptions.Timeout: print(请求超时请检查网络或稍后重试。) return None except requests.exceptions.RequestException as e: print(f网络请求发生异常: {e}) return None except json.JSONDecodeError: print(响应内容不是有效的JSON格式可能页面结构已变化。) return None # 测试查询 raw_trains query_tickets(北京, 上海, 2023-10-01) if raw_trains: # 先打印第一条数据看看结构 print(第一条原始数据样例) print(raw_trains[0])踩坑记录这里最容易出错的有两点。一是车站名不对比如输入“北京市”而不是“北京”导致查不到电报码。二是日期格式必须是YYYY-MM-DD并且不能查询过去很久的日期或太远的未来日期12306预售期有限制。务必在代码中加入严格的校验和友好的错误提示。3.3 第三步解析复杂的原始数据字符串上一步返回的raw_trains列表里每个元素都是一个字符串车次的所有信息都用|拼接在一起。这个字符串的字段顺序是固定的但不同车次类型高铁、动车、普速的字段数可能略有差异。我们需要找到目标字段的索引。def parse_train_info(raw_train_str): 解析单个车次的原始字符串数据。 :param raw_train_str: 包含车次信息的原始字符串 :return: 解析后的字典包含关键信息 # 将字符串按竖线分割成列表 fields raw_train_str.split(|) # 关键字段的索引位置这个索引相对稳定但若12306更新接口可能需要调整 # 可以通过对比官网显示的数据和这个列表来确认索引 train_info { 车次: fields[3], # 索引3是车次号如G1 出发站: fields[6], # 索引6是出发站电报码 到达站: fields[7], # 索引7是到达站电报码 出发时间: fields[8], # 索引8是出发时间 到达时间: fields[9], # 索引9是到达时间 历时: fields[10], # 索引10是运行时间 商务座/特等座: fields[32] or fields[25], # 索引32是商务座某些车次是25 一等座: fields[31], # 索引31是一等座 二等座/二等包座: fields[30], # 索引30是二等座 高级软卧: fields[21], # 索引21是高级软卧 软卧/一等卧: fields[23], # 索引23是软卧 动卧: fields[33], # 索引33是动卧 硬卧/二等卧: fields[28], # 索引28是硬卧 软座: fields[24], # 索引24是软座 硬座: fields[29], # 索引29是硬座 无座: fields[26], # 索引26是无座 其他: fields[22], # 索引22是其他 备注: fields[1] if len(fields) 1 else # 索引1有时是备注信息如“列车停运” } # 处理余票显示有数字表示票数有表示有票但数量未知或None表示无票 for seat_type in [商务座/特等座, 一等座, 二等座/二等包座, 高级软卧, 软卧/一等座, 动卧, 硬卧/二等卧, 软座, 硬座, 无座]: ticket_info train_info[seat_type] if ticket_info or ticket_info 无 or ticket_info is None: train_info[seat_type] 无票 elif ticket_info.isdigit(): train_info[seat_type] f{ticket_info}张 # 其他情况如‘有’保持不变 return train_info # 测试解析函数 if raw_trains: parsed_train parse_train_info(raw_trains[0]) print(\n解析后的第一条车次信息) for key, value in parsed_train.items(): print(f{key}: {value})重要提示字段索引不是一成不变的虽然目前基于广泛使用的公开索引相对稳定但12306有可能调整接口返回的数据结构。最可靠的方法是在浏览器开发者工具中找到查询接口返回的JSON数据复制一条结果字符串数一数你关心的信息如“二等座”前面有多少个|。这是爬虫工程师的必备调试技能。4. 数据处理用Pandas将杂乱数据变废为宝爬取到的数据还是原始的、杂乱的列表。我们需要将其转化为结构清晰、便于分析的表格DataFrame并进行清洗和整理。4.1 数据清洗与结构化我们将所有解析后的车次信息存入一个列表然后用pandas的DataFrame来承接。import pandas as pd def raw_data_to_dataframe(raw_trains_list, station_dict): 将原始车次数据列表转换为pandas DataFrame并进行初步清洗。 :param raw_trains_list: query_tickets函数返回的原始列表 :param station_dict: 车站电报码映射字典 :return: 清洗后的DataFrame if not raw_trains_list: print(没有车次数据可处理。) return pd.DataFrame() parsed_list [] for raw_train in raw_trains_list: try: train_dict parse_train_info(raw_train) parsed_list.append(train_dict) except IndexError as e: # 如果解析出错可能是字段索引对不上跳过该条数据并记录 print(f解析车次数据时出错可能数据结构有变跳过: {e}) continue if not parsed_list: print(所有数据解析均失败。) return pd.DataFrame() # 创建DataFrame df pd.DataFrame(parsed_list) # 数据清洗步骤 # 1. 将电报码转换回中文站名 # 反转车站字典从{‘北京’‘BJP’} 变成 {‘BJP’‘北京’} code_to_name {v: k for k, v in station_dict.items()} df[出发站] df[出发站].map(code_to_name).fillna(df[出发站]) # 如果映射失败保留原值 df[到达站] df[到达站].map(code_to_name).fillna(df[到达站]) # 2. 处理时间列方便后续按时间排序或计算 # 将‘出发时间’和‘到达时间’转换为datetime.time类型如果只关心时间 # 注意这里缺少日期信息所以用time类型 df[出发时间] pd.to_datetime(df[出发时间], format%H:%M, errorscoerce).dt.time df[到达时间] pd.to_datetime(df[到达时间], format%H:%M, errorscoerce).dt.time # 3. 将‘历时’如‘04:18’转换为分钟数便于比较 def duration_to_minutes(dur_str): try: if : in dur_str: h, m map(int, dur_str.split(:)) return h * 60 m else: return None except: return None df[历时_分钟] df[历时].apply(duration_to_minutes) # 4. 重命名列使其更直观 df.rename(columns{ 车次: train_no, 出发站: from_station, 到达站: to_station, 出发时间: departure_time, 到达时间: arrival_time, 历时: duration, 历时_分钟: duration_min, 商务座/特等座: business_seat, 一等座: first_class, 二等座/二等包座: second_class, 高级软卧: advanced_soft_sleeper, 软卧/一等卧: soft_sleeper, 动卧: motor_sleeper, 硬卧/二等卧: hard_sleeper, 软座: soft_seat, 硬座: hard_seat, 无座: no_seat, 备注: notes }, inplaceTrue) # 5. 按出发时间排序 df.sort_values(bydeparture_time, inplaceTrue) df.reset_index(dropTrue, inplaceTrue) # 重置索引 print(f数据处理完成共 {len(df)} 条有效车次记录。) return df # 整合测试 stations get_station_code() raw_data query_tickets(北京, 上海, 2023-10-01) if raw_data: df_trains raw_data_to_dataframe(raw_data, stations) # 查看前几行数据 print(\n处理后的数据前5行) print(df_trains.head().to_string()) # 使用to_string避免显示省略号 # 查看数据基本信息 print(\n数据概览) print(df_trains.info())经验之谈在数据清洗时一定要做好异常处理try...except和缺失值填充fillna。网络爬取的数据质量不可控可能某些字段为空或格式异常。我们的目标是让程序健壮地运行下去而不是因为一条脏数据而崩溃。将电报码转回中文站名这一步极大地提升了数据的可读性。4.2 数据筛选与简单分析有了整洁的DataFrame我们就可以像操作Excel表格一样进行各种查询和分析。def analyze_ticket_data(df): 对车票DataFrame进行简单的分析和筛选。 if df.empty: print(没有数据可供分析。) return print( 基础统计 ) print(f总车次数量: {len(df)}) # 1. 筛选有二等座余票的车次 # 注意我们的‘second_class’列现在是‘有’、‘无票’或‘X张’这样的字符串 # 我们需要筛选出不是‘无票’的 df_with_ticket df[df[second_class] ! 无票] print(f\n拥有二等座余票的车次数量: {len(df_with_ticket)}) # 2. 找出最早和最晚出发的车次 earliest_train df.loc[df[departure_time].idxmin()] latest_train df.loc[df[departure_time].idxmax()] print(f\n最早出发车次: {earliest_train[train_no]} 于 {earliest_train[departure_time]}) print(f最晚出发车次: {latest_train[train_no]} 于 {latest_train[departure_time]}) # 3. 找出运行时间最短最快的车次 # 先过滤掉历时为空的数据 df_valid_duration df.dropna(subset[duration_min]) if not df_valid_duration.empty: fastest_train df_valid_duration.loc[df_valid_duration[duration_min].idxmin()] print(f\n运行时间最短的车次: {fastest_train[train_no]}, 历时 {fastest_train[duration]} ({fastest_train[duration_min]}分钟)) # 4. 按车次类型统计例如G字头为高铁D字头为动车 df[train_type] df[train_no].str[0] # 提取车次号首字母 type_counts df[train_type].value_counts() print(f\n各类型车次统计:) for train_type, count in type_counts.items(): print(f {train_type}字头: {count} 趟) # 5. 创建一个更直观的余票情况视图 # 我们只关心几个主要席别 seat_columns [business_seat, first_class, second_class, soft_sleeper, hard_sleeper] print(f\n部分车次主要席别余票情况预览:) # 选择前10趟车展示关键信息 preview_df df[[train_no, departure_time, arrival_time, duration] seat_columns].head(10) print(preview_df.to_string(indexFalse)) # 不显示行索引 return df_with_ticket # 执行分析 if not df_trains.empty: available_trains_df analyze_ticket_data(df_trains)这个分析函数展示了pandas的基本能力布尔索引筛选、按列计算极值、字符串操作、分组统计。你可以根据自己的需求轻松地修改和扩展这些分析逻辑比如筛选“上午出发的车次”、“有卧铺的车次”等等。4.3 数据持久化保存到文件处理好的数据当然要保存下来方便日后查看或进行更深入的分析比如对比不同日期的余票情况。def save_data_to_file(df, filename_prefixtrain_tickets): 将DataFrame保存到CSV和Excel文件。 :param df: 要保存的DataFrame :param filename_prefix: 文件名前缀会自动加上日期时间戳 if df.empty: print(没有数据可保存。) return # 生成带时间戳的文件名避免覆盖 from datetime import datetime timestamp datetime.now().strftime(%Y%m%d_%H%M%S) csv_filename f{filename_prefix}_{timestamp}.csv excel_filename f{filename_prefix}_{timestamp}.xlsx try: # 保存为CSV通用性好体积小 df.to_csv(csv_filename, indexFalse, encodingutf-8-sig) # utf-8-sig解决Excel打开中文乱码 print(f数据已保存为CSV文件: {csv_filename}) # 保存为Excel格式美观支持多工作表 # 需要安装 openpyxl 库: pip install openpyxl df.to_excel(excel_filename, indexFalse, sheet_name车票信息) print(f数据已保存为Excel文件: {excel_filename}) except Exception as e: print(f保存文件时出错: {e}) # 保存我们处理好的数据 if not df_trains.empty: save_data_to_file(df_trains, beijing_shanghai_tickets)小技巧保存为CSV是通用性最好的选择任何文本编辑器都能打开。保存为Excel则更适合需要手动查看、标注或分享给非技术同事的场景。使用utf-8-sig编码可以确保用Excel打开CSV文件时中文字符正常显示。5. 进阶与避坑让爬虫更稳健、更高效一个只能运行一次的脚本是玩具一个能持续稳定运行的脚本才是工具。下面分享几个让脚本更实用的进阶技巧和常见坑的解决方法。5.1 应对请求频率限制与超时12306对频繁访问会有一定的限制。在脚本中直接使用requests.get()而不加控制短时间内大量请求可能导致IP被暂时屏蔽。添加延时在连续请求之间使用time.sleep()添加随机延时模拟人工操作。import time import random # 在关键请求如循环查询多个日期后添加延时 time.sleep(random.uniform(1, 3)) # 随机休眠1到3秒使用Session对象requests.Session()可以复用TCP连接并在会话级别保持一些头信息和cookies效率更高也更接近浏览器行为。session requests.Session() session.headers.update({ User-Agent: 你的User-Agent, Accept-Language: zh-CN,zh;q0.9, }) # 后续所有请求都用 session.get() 或 session.post() response session.get(url, paramsparams, timeout10)设置合理超时网络环境复杂必须设置timeout参数如timeout(5, 15)分别代表连接超时和读取超时避免程序因网络问题无限挂起。5.2 处理动态变化的查询参数有时查询接口需要一些动态参数如_json_att或一个加密的RAIL_DEVICEID。这些参数通常可以在首页的HTML源码或某个JS文件里找到或者通过首次访问其他接口获得。应对策略先访问一次首页用Session对象先请求一下https://kyfw.12306.cn/otn/leftTicket/init让Session自动管理cookies。提取动态参数如果发现查询请求需要额外的参数仔细分析浏览器首次加载页面时发起的其他请求看是否有获取这些参数的接口。然后先请求那个接口提取参数再拼接到主查询请求中。保持会话坚持使用同一个Session对象进行所有关联请求是处理这类依赖cookies或token的网站最有效的方法。5.3 错误处理与日志记录一个健壮的程序必须能妥善处理各种异常并记录下发生了什么方便排查问题。import logging # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(ticket_crawler.log, encodingutf-8), logging.StreamHandler() # 同时输出到控制台 ] ) logger logging.getLogger(__name__) def robust_query_tickets(from_station, to_station, train_date, retry_times2): 带重试和错误处理的查询函数。 for attempt in range(retry_times 1): try: logger.info(f尝试第 {attempt 1} 次查询 {from_station} - {to_station} 在 {train_date}) result query_tickets(from_station, to_station, train_date) # 调用我们之前写的函数 if result is not None: return result else: logger.warning(f第 {attempt 1} 次查询返回空结果。) except requests.exceptions.Timeout: logger.error(f第 {attempt 1} 次查询超时。) except Exception as e: logger.exception(f第 {attempt 1} 次查询发生未知错误: {e}) # 会记录完整的异常堆栈 if attempt retry_times: wait_time (attempt 1) * 3 # 重试等待时间递增 logger.info(f等待 {wait_time} 秒后重试...) time.sleep(wait_time) logger.error(f查询 {from_station} - {to_station} 在 {train_date} 失败已达最大重试次数。) return None将核心函数用try...except包裹并记录日志这样当脚本在后台运行时我们也能通过查看日志文件ticket_crawler.log来了解运行状态和错误原因。5.4 将脚本模块化与函数化随着功能增多把所有代码堆在一个文件里会难以维护。最佳实践是将代码模块化config.py: 存放常量如请求头模板、URL、字段索引映射。station_manager.py: 负责车站信息的获取、缓存和查询。crawler.py: 包含核心的查询和解析函数。data_processor.py: 包含数据清洗、分析和保存的函数。main.py: 主程序入口组织调用逻辑。这样不仅代码清晰也方便复用。例如station_manager模块可以设计为首次运行时下载车站信息并保存到本地文件下次启动时直接读取本地文件大大提升速度。6. 项目总结与扩展思路走到这里你已经完成了一个完整的、具有实用价值的12306车票查询与数据处理脚本。它不仅能查票还能把结果规规矩矩地整理好、分析好、保存好。这个过程里最重要的不是代码本身而是分析问题、拆解步骤、调试和解决问题的思路。我个人在多次编写类似爬虫后最深的体会是耐心和细致比技术更重要。90%的时间花在分析网络请求、理解数据格式、处理边界情况上。写代码可能只要半小时但让代码稳定可靠地运行需要数倍的调试和优化时间。这个项目还有很大的扩展空间你可以尝试多日期/多车次批量查询写一个循环爬取未来一周内每天的车票信息分析余票释放规律。价格信息获取查询接口返回的数据中通常包含票价信息在另一个字段串中可以尝试解析并计算均价、最贵/最便宜车次。简单可视化用matplotlib或seaborn将热门车次的出发时间分布、余票数量趋势画成图表。封装成GUI工具或Web服务使用tkinter、PyQt做一个桌面小工具或者用Flask/FastAPI做一个简单的查询网页分享给不会编程的朋友用。接入通知当出现符合你条件的车票如下午出发、有二等座时自动发送邮件或微信消息提醒。记住爬虫技术应当用在学习和提升效率上务必遵守网站的robots.txt协议控制请求频率不要对目标服务器造成压力。通过这个项目打下的基础你可以去探索更广阔的数据世界。