Python 爬虫实战：实时抓取飞猪旅行机票折扣信息

前言

飞猪旅行作为阿里旗下的核心在线旅游平台，其机票折扣数据涵盖航班信息、实时票价、优惠活动、舱位等级等关键维度，是出行价格分析、机票比价、行程规划的重要数据源。相较于携程酒店的静态接口，飞猪机票数据具备实时性强、加密参数多、反爬机制复杂等特点，抓取难度更高。本文将从接口分析、加密参数处理、反反爬策略等维度，系统讲解如何使用 Python 实现飞猪机票折扣信息的精准抓取，帮助开发者突破平台限制，获取结构化的机票折扣数据。

摘要

本文聚焦飞猪旅行机票折扣爬虫的全流程实现，核心涵盖加密请求参数解析、实时接口数据抓取、多航线多日期折扣对比三大核心技术点，通过requests库发送 HTTP 请求、pycryptodome处理加密参数、pandas实现数据结构化，结合实战案例完成航班号、起降时间、原价、折扣价、折扣率、舱位等核心字段的抓取。实战目标链接：飞猪机票查询页，最终实现支持多航线、多日期的机票折扣爬虫脚本，并提供折扣数据分析与可视化方案。

一、技术原理与环境准备

1.1 核心技术原理

飞猪机票折扣抓取的核心难点与解决思路：

• 机票价格数据通过加密接口返回，请求参数包含sign（签名）、token等加密字段，需分析签名规则或通过抓包获取有效参数；
• 价格数据实时更新（每分钟可能变动），接口需携带时间戳、设备 ID 等动态参数；
• 高频请求会触发滑块验证、IP 封禁，需结合代理 IP、请求间隔、设备指纹伪装等反反爬策略；
• 折扣信息分散在不同层级的 JSON 数据中，需精准解析嵌套结构。

1.2 环境配置

环境安装命令

bash

运行

pip install requests pycryptodome fake-useragent python-dotenv pandas matplotlib

二、实战开发：飞猪机票折扣爬虫

2.1 核心思路拆解

2.2 完整代码实现

python

运行

import requests
import json
import time
import os
import hashlib
from fake_useragent import UserAgent
import pandas as pd
from dotenv import load_dotenv

# 加载环境变量（存储Token/Sign等敏感参数）
load_dotenv()

class FliggyFlightSpider:
def __init__(self):
"""初始化爬虫配置"""
# 基础配置
self.ua = UserAgent()
self.headers = {
"User-Agent": self.ua.random,
"Referer": "https://www.fliggy.com/",
"Origin": "https://www.fliggy.com",
"Accept": "application/json, text/plain, */*",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
"Connection": "keep-alive",
"Token": os.getenv("FLIGGY_TOKEN", ""), # 从抓包获取
"Content-Type": "application/json;charset=UTF-8"
}
self.timeout = 15 # 请求超时时间
self.delay = 5 # 飞猪反爬严格，建议5-8秒/次
self.all_flight_data = [] # 存储所有机票数据

# 飞猪机票查询接口（简化版，实际需抓包更新）
self.api_url = "https://www.fliggy.com/flights/api/search/flightList"

def get_city_code(self, city_name):
"""
城市名称转机场编码（核心，飞猪使用机场三字码）
:param city_name: 城市名称/机场名称
:return: 机场三字码
"""
city_code_map = {
"北京": "PEK", "上海": "SHA", "广州": "CAN", "深圳": "SZX",
"杭州": "HGH", "成都": "CTU", "重庆": "CKG", "西安": "SIA",
"南京": "NKG", "武汉": "WUH"
}
return city_code_map.get(city_name, "PEK") # 默认北京首都机场

def generate_sign(self, params):
"""
生成接口签名（简化版，实际需分析飞猪签名规则）
:param params: 请求参数
:return: 签名字符串
"""
# 拼接参数并加盐值（盐值从抓包获取）
sign_str = f"{params['departure']}{params['arrival']}{params['date']}{os.getenv('FLIGGY_SALT', 'fliggy2026')}"
# MD5加密生成签名
sign = hashlib.md5(sign_str.encode()).hexdigest().upper()
return sign

def build_request_params(self, departure="北京", arrival="上海", date="2026-01-25"):
"""
构造接口请求参数
:param departure: 出发城市
:param arrival: 目的城市
:param date: 出发日期
:return: 请求参数字典
"""
# 转换为机场三字码
dep_code = self.get_city_code(departure)
arr_code = self.get_city_code(arrival)

# 基础参数
params = {
"departure": dep_code,
"arrival": arr_code,
"date": date,
"type": "ONE_WAY", # 单程：ONE_WAY，往返：ROUND_TRIP
"adultNum": 1, # 成人数量
"childNum": 0, # 儿童数量
"infantNum": 0, # 婴儿数量
"timestamp": int(time.time() * 1000), # 时间戳
"deviceId": f"DEVICE_{int(time.time() * 1000) % 1000000}" # 随机设备ID
}

# 生成签名
params["sign"] = self.generate_sign(params)

return params

def get_flight_data(self, params):
"""
发送请求获取机票数据
:param params: 请求参数
:return: 解析后的JSON数据/None
"""
try:
# 延迟请求，规避反爬
time.sleep(self.delay)

# 发送POST请求（飞猪机票接口多为POST）
response = requests.post(
url=self.api_url,
headers=self.headers,
json=params, # POST请求使用json参数传递数据
timeout=self.timeout
)

# 检查请求状态
response.raise_for_status()

# 解析JSON数据
json_data = response.json()
return json_data

except requests.exceptions.RequestException as e:
print(f"请求接口失败：{str(e)}")
return None
except json.JSONDecodeError as e:
print(f"JSON解析失败：{str(e)}")
return None

def parse_flight_data(self, json_data, departure, arrival, date):
"""
解析机票数据，提取核心折扣信息
:param json_data: 接口返回的JSON数据
:param departure: 出发城市
:param arrival: 目的城市
:param date: 出发日期
:return: 结构化机票数据列表
"""
if not json_data or json_data.get("code") != 0:
print("接口返回异常或无数据")
return []

flight_list = []
# 定位航班列表数据（适配飞猪JSON结构）
flights = json_data.get("data", {}).get("flightList", [])

for flight in flights:
try:
# 核心字段提取
airline = flight.get("airlineName", "未知航空公司") # 航空公司
flight_no = flight.get("flightNo", "未知航班号") # 航班号
dep_time = flight.get("departureTime", "未知起飞时间") # 起飞时间
arr_time = flight.get("arrivalTime", "未知降落时间") # 降落时间
seat_class = flight.get("cabinClassName", "未知舱位") # 舱位等级
original_price = flight.get("originalPrice", 0) # 原价
discount_price = flight.get("salePrice", 0) # 折扣价

# 计算折扣率（保留2位小数）
discount_rate = round((discount_price / original_price) * 10, 2) if original_price > 0 else 0.0

# 封装机票数据
flight_info = {
"出发城市": departure,
"目的城市": arrival,
"出发日期": date,
"航空公司": airline,
"航班号": flight_no,
"起飞时间": dep_time,
"降落时间": arr_time,
"舱位等级": seat_class,
"原价(元)": original_price,
"折扣价(元)": discount_price,
"折扣率(折)": discount_rate
}
flight_list.append(flight_info)

except Exception as e:
print(f"解析单条航班数据失败：{str(e)}")
continue

return flight_list

def crawl_flights(self, route_list):
"""
批量抓取多航线/多日期的机票折扣数据
:param route_list: 航线列表，格式：[(出发城市, 目的城市, 日期), …]
:return: 所有机票数据
"""
print(f"开始抓取{len(route_list)}条航线的机票折扣数据…")

for idx, (departure, arrival, date) in enumerate(route_list, 1):
print(f"\\n正在抓取第{idx}条航线：{departure} → {arrival}（{date}）")

# 构造请求参数
params = self.build_request_params(departure, arrival, date)

# 获取接口数据
json_data = self.get_flight_data(params)

# 解析机票数据
flight_data = self.parse_flight_data(json_data, departure, arrival, date)

if not flight_data:
print(f"该航线无数据，跳过")
continue

# 添加到总列表
self.all_flight_data.extend(flight_data)
print(f"该航线抓取完成，共{len(flight_data)}个航班")

print(f"\\n抓取完成！总计获取{len(self.all_flight_data)}个航班的折扣数据")
return self.all_flight_data

def save_to_excel(self, file_path="fliggy_flight_discounts.xlsx"):
"""
将机票折扣数据保存为Excel文件
:param file_path: 保存路径
"""
if not self.all_flight_data:
print("无数据可保存")
return

try:
# 转换为DataFrame
df = pd.DataFrame(self.all_flight_data)
# 按折扣率升序排序（折扣力度越大越靠前）
df = df.sort_values(by="折扣率(折)", ascending=True)
# 保存为Excel
df.to_excel(file_path, index=False, engine="openpyxl")
print(f"数据已成功保存到：{file_path}")
except Exception as e:
print(f"保存Excel失败：{str(e)}")

def print_discount_ranking(self, top_n=10):
"""
打印折扣力度TOP N的航班
:param top_n: 展示数量
"""
if not self.all_flight_data:
print("无抓取结果")
return

# 转换为DataFrame并排序
df = pd.DataFrame(self.all_flight_data)
df_sorted = df.sort_values(by="折扣率(折)", ascending=True).head(top_n)

print(f"\\n===== 飞猪机票折扣力度TOP {top_n} =====")
print(df_sorted.to_string(index=False))

if __name__ == "__main__":
# 实例化爬虫
spider = FliggyFlightSpider()

# 定义待抓取的航线列表（可扩展）
route_list = [
("北京", "上海", "2026-01-25"),
("北京", "广州", "2026-01-25"),
("上海", "深圳", "2026-01-26")
]

# 抓取机票折扣数据
spider.crawl_flights(route_list)

# 打印折扣TOP10
spider.print_discount_ranking(top_n=10)

# 保存数据到Excel
spider.save_to_excel()

2.3 环境变量配置（.env 文件）

在项目根目录创建.env文件，填入飞猪接口所需的敏感参数（从抓包获取）：

env

FLIGGY_TOKEN="你的飞猪Token"
FLIGGY_SALT="你的飞猪签名盐值"

2.4 代码输出结果示例

plaintext

开始抓取3条航线的机票折扣数据…

正在抓取第1条航线：北京 → 上海（2026-01-25）
该航线抓取完成，共32个航班

正在抓取第2条航线：北京 → 广州（2026-01-25）
该航线抓取完成，共28个航班

正在抓取第3条航线：上海 → 深圳（2026-01-26）
该航线抓取完成，共25个航班

抓取完成！总计获取85个航班的折扣数据

===== 飞猪机票折扣力度TOP 10 =====
出发城市 目的城市 出发日期 航空公司 航班号 起飞时间 降落时间 舱位等级 原价(元) 折扣价(元) 折扣率(折)
北京 上海 2026-01-25 中国东方航空 MU5102 2026-01-25 08:00 2026-01-25 10:15 经济舱 1200 480 4.0
北京 上海 2026-01-25 中国南方航空 CZ8888 2026-01-25 14:30 2026-01-25 16:45 经济舱 1180 488 4.14
北京 广州 2026-01-25 中国南方航空 CZ3101 2026-01-25 09:15 2026-01-25 11:50 经济舱 1350 567 4.2
北京 上海 2026-01-25 中国国际航空 CA1509 2026-01-25 10:00 2026-01-25 12:10 经济舱 1250 538 4.3
上海 深圳 2026-01-26 春秋航空 9C8937 2026-01-26 07:30 2026-01-26 09:55 经济舱 1100 484 4.4
北京 广州 2026-01-25 海南航空 HU7131 2026-01-25 11:00 2026-01-25 13:35 经济舱 1400 627 4.48
上海 深圳 2026-01-26 中国南方航空 CZ3562 2026-01-26 10:15 2026-01-26 12:40 经济舱 1280 582 4.55
北京 上海 2026-01-25 吉祥航空 HO1256 2026-01-25 16:00 2026-01-25 18:10 经济舱 1150 536 4.66
北京 广州 2026-01-25 中国国际航空 CA1321 2026-01-25 13:00 2026-01-25 15:30 经济舱 1380 649 4.7
上海 深圳 2026-01-26 东方航空 MU5391 2026-01-26 14:00 2026-01-26 16:25 经济舱 1220 582 4.77
数据已成功保存到：fliggy_flight_discounts.xlsx

2.5 核心代码原理说明

三、反反爬优化策略

3.1 代理 IP 池与请求指纹优化

飞猪对 IP 和设备指纹敏感，集成代理池并随机化请求特征：

python

运行

def get_proxy(self):
"""从代理池获取高匿代理IP"""
try:
proxy_res = requests.get("http://你的代理池接口/get_proxy")
proxy_info = proxy_res.json()
return {
"http": f"http://{proxy_info['user']}:{proxy_info['pwd']}@{proxy_info['ip']}:{proxy_info['port']}",
"https": f"https://{proxy_info['user']}:{proxy_info['pwd']}@{proxy_info['ip']}:{proxy_info['port']}"
}
except Exception as e:
print(f"获取代理失败：{str(e)}")
return None

def randomize_request_headers(self):
"""随机化请求头特征"""
self.headers["User-Agent"] = self.ua.random
self.headers["Accept-Encoding"] = random.choice(["gzip, deflate, br", "gzip, deflate"])
self.headers["Cache-Control"] = random.choice(["no-cache", "max-age=0"])
return self.headers

# 在get_flight_data方法中调用
# self.headers = self.randomize_request_headers()
# proxies = self.get_proxy()
# response = requests.post(url=self.api_url, headers=self.headers, json=params, proxies=proxies, timeout=self.timeout)

3.2 滑块验证自动处理

触发滑块验证时，使用selenium+ddddocr自动识别并通过验证：

bash

运行

pip install selenium ddddocr webdriver-manager

python

运行

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from webdriver_manager.chrome import ChromeDriverManager
import ddddocr
import time

def handle_slider_verification(self):
"""自动处理滑块验证"""
# 初始化浏览器
driver = webdriver.Chrome(ChromeDriverManager().install())
driver.get("https://www.fliggy.com/flights/")
driver.maximize_window()

try:
# 等待滑块验证出现
slider = WebDriverWait(driver, 10).until(
EC.presence_of_element_located((By.CLASS_NAME, "slider-verify"))
)

# 截取验证图片
slider.screenshot("slider.png")
ocr = ddddocr.DdddOcr()
# 识别滑块缺口位置（简化版，实际需精准定位）
res = ocr.slide_match("slider.png", "slider_bg.png", simple_target=True)

# 模拟滑动
action = webdriver.ActionChains(driver)
action.click_and_hold(slider).move_by_offset(res["target"][0], 0).release().perform()
time.sleep(2)

# 获取验证后的Cookie/Token
cookies = driver.get_cookies()
token = driver.execute_script("return window.localStorage.getItem('fliggy_token')")

# 更新请求头
self.headers["Cookie"] = "; ".join([f"{c['name']}={c['value']}" for c in cookies])
self.headers["Token"] = token

print("滑块验证通过，已更新请求参数")
except Exception as e:
print(f"滑块验证处理失败：{str(e)}")
finally:
driver.quit()

3.3 失败重试机制

添加请求失败自动重试，提升爬虫稳定性：

python

运行

def get_flight_data_with_retry(self, params, retry_times=3):
"""带重试机制的请求方法"""
for i in range(retry_times):
json_data = self.get_flight_data(params)
if json_data:
return json_data
print(f"第{i+1}次请求失败，{retry_times-i-1}次重试机会")
time.sleep(5) # 重试间隔5秒
print("所有重试均失败，放弃该请求")
return None

四、数据扩展与分析

4.1 折扣趋势监控

实现定时抓取，监控不同日期的机票折扣变化：

python

运行

import schedule

def scheduled_discount_monitor():
"""定时监控机票折扣"""
spider = FliggyFlightSpider()
# 监控未来7天的北京→上海航线
monitor_dates = [f"2026-01-{20+i}" for i in range(7)]
route_list = [("北京", "上海", date) for date in monitor_dates]

spider.crawl_flights(route_list)
# 按日期分表保存
with pd.ExcelWriter(f"fliggy_discount_monitor_{time.strftime('%Y%m%d')}.xlsx") as writer:
for date in monitor_dates:
df_date = pd.DataFrame(spider.all_flight_data)[lambda x: x["出发日期"] == date]
df_date.to_excel(writer, sheet_name=date, index=False)

print(f"定时监控完成，数据已保存")

# 每天10点、16点执行监控
schedule.every().day.at("10:00").do(scheduled_discount_monitor)
schedule.every().day.at("16:00").do(scheduled_discount_monitor)

# 启动定时任务
while True:
schedule.run_pending()
time.sleep(60)

4.2 折扣数据可视化

绘制不同航线的折扣率对比柱状图：

python

运行

import matplotlib.pyplot as plt

def visualize_discount_comparison(self):
"""可视化不同航线的折扣率对比"""
if not self.all_flight_data:
print("无数据可可视化")
return

# 设置中文字体
plt.rcParams["font.sans-serif"] = ["SimHei"]
plt.rcParams["axes.unicode_minus"] = False

# 按航线分组计算平均折扣率
df = pd.DataFrame(self.all_flight_data)
route_avg = df.groupby(["出发城市", "目的城市"])["折扣率(折)"].mean().reset_index()
route_avg["航线"] = route_avg["出发城市"] + "→" + route_avg["目的城市"]

# 绘制柱状图
plt.figure(figsize=(12, 6))
bars = plt.bar(route_avg["航线"], route_avg["折扣率(折)"], color=["#4CAF50", "#2196F3", "#FF9800"])

# 添加数值标签
for bar in bars:
height = bar.get_height()
plt.text(bar.get_x() + bar.get_width()/2., height + 0.05,
f"{height:.2f}折", ha="center", va="bottom")

plt.title("飞猪不同航线机票平均折扣率对比")
plt.xlabel("航线")
plt.ylabel("平均折扣率（折）")
plt.ylim(0, 10) # 折扣率范围0-10折
plt.grid(axis="y", alpha=0.3)
plt.savefig("flight_discount_comparison.png", dpi=300, bbox_inches="tight")
plt.show()

五、注意事项与合规声明

总结