使用Scrapeless爬虫API构建航班价格跟踪器
Specialist in Anti-Bot Strategies
TL;DR:
- 航班监控是一个航线和一个日期，而不是一个URL。 航空票价没有像零售SKU那样的单一规范页面——同一行程由
departure_id、arrival_id、outbound_date和（对于往返票）return_date组合而成。下面的管道通过Scrapeless Scraping API的scraper.google.flights演员直接查询该组合，而不是渲染和抓取Google Flights自己的界面。 - 行程类型是一个一流参数，而不是URL变体。
data_type在单程、往返和多城市之间切换同一请求，而多城市需要自己的multi_city_json航段数组。零售价格监控从不需要这个维度；航班监控从第一天开始就需要这个维度。 - Google Flights提供了自己的波动信号。 响应的
price_insights对象携带lowest_price和price_level分类（低、典型、高），以及在best_flights和other_flights中的个别报价。这使管道能够提醒“现在这是一个好票价”，而不仅仅是“这比上次便宜”。 - 历史记录以航线和日期为键，每次检查追加一次。 一个JSONL日志，每个
(departure_id, arrival_id, outbound_date, return_date)组合一个记录，就足以计算当前最低价和检测价格下降——直到监控列表超过几条航线之前不需要数据库。 - 价格下降，或
price_level转为低，会触发Webhook。 警报决策结合了数值比较和Google自己的定性信号，然后发送一次到任何接收端点。 - 调度程序运行循环；没有什么需要持续运行。 Cron、任务调度器或无服务器定时器按节奏执行检查并退出——管道没有长时间运行的进程。
- 免费开始。 新的Scrapeless账户包括免费的Scraping API积分——在app.scrapeless.com注册。
介绍：票价是一个具有两个维度而非一个维度的动态目标
零售价格是附加在单一页面上的单一数字。航班票价是航线和日期的函数，同一路线在同一搜索中的每个出发日期价格可能不同。这就是Google自己的航班票价追踪功能存在的全部原因：航空公司会根据需求不断调整座位库存的价格，而一次检查并预订的购物者是在猜测时机。Google的追踪器在跟踪路线发生变化时发送摘要；它不会提供原始数据、Webhook或将该信号与读者已在运行的其他内容结合的方式。
本篇文章构建了一个小型管道，程序化地执行此任务，基于特定的航线和日期窗口，而不是Google Flights的界面。Scrapeless的Scraping API提供了一个专为此表面构建的演员——scraper.google.flights——它接受航线和日期组合并返回结构化的票价数据：每个报价的个别要素、航空公司和每个报价的时长，以及Google对该票价的价格级别判断。管道查询该演员，提取票价信号，将其附加到由航线和日期键入的历史日志中，决定读数是否值得警报，并在值得时触发Webhook。调度程序按节奏运行检查。
该演员为研究本篇文章所用的账户返回了一个实时的disabled actor响应——这是计划级别的差距，而不是管道故障——这个差距在重要点上得到了记录，而不是被忽略。以下的每个阶段都是基于真实执行的Python进行的。
您可以用它做什么
- 个人票价监控。 跟踪特定航线和日期窗口，仅在价格实际向有利方向移动时获得通知。
- 灵活日期购物。 在一系列候选日期中运行相同的航线，并比较
price_insights.price_level以找到最低的时间窗口，而不仅仅是最低的单一日期。 - 企业旅行监控。 旅行部门可以监控重复的通勤或客户航线，并在预定行程的航线价格下降时提醒重新预定。
- 多城市行程跟踪。
multi_city_json允许同一请求表示一个多段行程，因此三或四个航段的行程可以作为一次调用，而不是将三个或四个单程请求拼接在一起。 - 票价历史数据集。 这种仅追加日志也充当每条航线的时间序列，便于后续的趋势分析或用于提供“是否现在预订或等一下”的启发式判断。
为什么选择Scrapeless Scraping API
Scrapeless Scraping API公开了特定网站的演员——scraper.<site>——返回给定目标的结构化JSON，而不是必须解析的渲染页面。对于航班数据而言，这意味着：
- 一个路线和日期请求，而不是一个浏览器会话。 此调用是一个经过身份验证的 POST 请求，带有 JSON 格式的主体；没有页面需要渲染，没有选择器需要锚定，也没有会话需要保持活跃。
- 旅行类型嵌入请求结构中。
data_type包含单程（2）、往返（1）和多城市（3，带有multi_city_json），作为文档参数，而不是针对每种情况的单独抓取逻辑。 - 目标网站自行计算的定价分类。
price_insights.price_level是谷歌对该航线票价低、典型或高的判断——该管道可以依赖此判断，而不是从头猜测一个阈值。 - 一个请求而不是分布堆栈。 否则，航空公司票价数据会通过 GDS 和航空公司直接的拼凑方式到达调用者——这种碎片化的分布 IATA的新分销能力标准 旨在在航空业一侧进行标准化。单个结构化的演员调用完全避开该堆栈，以进行只读的价格监控。
在 app.scrapeless.com 获取一个免费计划的 API 密钥。 抓取 API 产品页面 涵盖了该管道调用的演员家族，针对该特定演员的请求结构在 谷歌航班 API 请求指南 中进行了详细说明。
前提条件
- Python 3.10 或更新版本，此外还需要
requests包（pip install requests）。 - 一个 Scrapeless 帐户和 API 密钥，从
SCRAPELESS_API_KEY环境变量中读取。在将生产流量指向该帐户的计划之前，请确认scraper.google.flights演员已启用——请参见下面的 Fetch 部分中的说明。 - 一个接收警报的 webhook 端点（Slack、Discord 或任何接受 JSON POST 的中继）。
- 一个支持计划任务的主机、Windows 任务计划程序或无服务器调度器，以定期运行检查。
管道概述
该管道分为六个阶段，串联连接：获取 → 发现 → 提取 → 转换 → 存储 → 决定 → 警报，由调度程序驱动整个流程的运行。
- 获取 — 向
scraper.google.flightsPOST 一个路线、一个日期（或日期窗口）和一个旅行类型。 - 发现 — 读取旅行类型响应：响应实际填充的
best_flights、other_flights和price_insights中的哪一个。 - 提取 — 从响应中提取最便宜的报价的价格和谷歌自己的
price_level分类。 - 转换 — 将读取结果规范化为一个以路线和日期为键的权威记录。
- 存储 — 将记录附加到历史日志中。
- 决定和警报 — 与历史数据进行比较，并在读取值得突出显示时触发 webhook。
获取：查询路线和日期窗口
请求是一个单独的经过身份验证的 POST。演员名称放在 actor 中；路线和日期参数放在 input 中。data_type 为往返时为 1，单程时为 2；往返旅行还携带 return_date。
python
import os
import requests
API_URL = "https://api.scrapeless.com/api/v1/scraper/request"
def fetch_route(departure_id: str, arrival_id: str, outbound_date: str,
return_date: str | None = None, gl: str = "us", hl: str = "en") -> dict:
payload = {
"actor": "scraper.google.flights",
"input": {
"departure_id": departure_id,
"arrival_id": arrival_id,
"data_type": 1 if return_date else 2, # 1 = 往返，2 = 单程
"outbound_date": outbound_date,
"gl": gl,
"hl": hl,
},
}
if return_date:
payload["input"]["return_date"] = return_date
headers = {"Content-Type": "application/json", "x-api-token": os.environ["SCRAPELESS_API_KEY"]}
response = requests.post(API_URL, headers=headers, json=payload, timeout=60)
response.raise_for_status()
return response.json()outbound_date 和 return_date 遵循普通的 YYYY-MM-DD 日历日期格式 RFC 3339 定义的互联网日期时间交换——没有时间组件，因为票价搜索是基于出发日期而非出发时刻。多城市行程将单腿字段替换为 data_type: 3 和一个 multi_city_json 数组，每个对象代表一段旅程，各自携带其 departure_id、arrival_id 和 date。
前提差距： 对上述真实路由的调用返回了一个真实的 HTTP 400，针对用于验证该管道的账户：
{"code": 14002, "message": "禁用的操作员：scraper.google.flights"}。该消息与未识别的操作员名称返回的错误（"无效操作员：<名称>"— 通过故意请求一些虚构变体确认）是不同的，这意味着该操作员本身是真实且已编目；它被限制在此处使用的免费计划之外，而不是缺失。账户的基本请求机制没有问题 — 兄弟scraper.google.search操作员在本次研究中针对相同的键和相同的端点返回了真实的 HTTP 200 和有机结果。在构建其上的生产流量之前，请确认目标计划已启用该操作员（GET /api/v1/me报告计划和信用余额），并将从 Discover 之后的所有内容视为针对操作员的文档化响应形状运行，而不是在此会话中实时捕获的页面。
Discover：读取行程类型封装
往返或单程搜索返回两个报价数组 — best_flights 和 other_flights — 加上一个 price_insights 对象。多城市搜索按行程返回等效结构，而不是按单个航段。在提取任何内容之前，管道只需要知道响应实际填充了哪些键，因为狭窄的路线或奇怪的日期可能会导致 other_flights 为空，只有 best_flights 承载报价。
python
def envelope_shape(payload: dict) -> dict:
return {
"has_best": bool(payload.get("best_flights")),
"has_other": bool(payload.get("other_flights")),
"has_insights": bool(payload.get("price_insights")),
}
if __name__ == "__main__":
sample = {"best_flights": [{"price": 312}], "other_flights": [], "price_insights": {}}
print(envelope_shape(sample))针对仅填充了 best_flights 的狭窄响应运行时，envelope_shape 返回 {'has_best': True, 'has_other': False, 'has_insights': False} — 这是提取阶段下面必须容忍的确切情况。
这是一个单行的合理性检查，但很重要：从一个从未填充 best_flights 的封装中提取“最低报价”会产生 None，而不是错误的数字，前提是提取阶段同时检查两个数组，而不是假设其中一个总是存在。
提取：从响应中提取票价信号
提取阶段从两个数组和旁边的 price_insights 字段中提取最低报价。下面的示例固定了为该操作员系列记录的响应形状 — best_flights 和 other_flights 作为 {flights: [...], price, layovers} 对象的数组，而 price_insights 包含 lowest_price、price_level 和 typical_price_range — 因为实时调用是上述提到的差距，而不是本会话捕获的页面。
python
# 与记录的响应形状匹配的示例固定（见
# Fetch 下的前提差距注释） — 不是实时捕获。
FIXTURE_RESPONSE = {
"best_flights": [
{
"flights": [{
"departure_airport": {"id": "JFK", "time": "2026-08-10 08:00"},
"arrival_airport": {"id": "LAX", "time": "2026-08-10 11:24"},
"airline": "示例航空",
"flight_number": "EX 101",
"duration": 384,
"travel_class": "经济舱",
}],
"price": 312,
"layovers": 0,
}
],
"other_flights": [
{"flights": [{"airline": "示例航空公司", "flight_number": "SA 220"}], "price": 349, "layovers": 1}
],
"price_insights": {
"lowest_price": 289,
"price_level": "低",
"typical_price_range": [295, 430],
},
}
def extract_fares(payload: dict) -> dict:
best = payload.get("best_flights", [])
other = payload.get("other_flights", [])
insights = payload.get("price_insights", {})
cheapest_offer = min([*best, *other], key=lambda offer: offer["price"], default=None)
return {
"cheapest_price": cheapest_offer["price"] if cheapest_offer else None,
"cheapest_layovers": cheapest_offer["layovers"] if cheapest_offer else None,
"offer_count": len(best) + len(other),
"price_insights_lowest": insights.get("lowest_price"),
"price_level": insights.get("price_level"),
"typical_price_range": insights.get("typical_price_range"),
}
if __name__ == "__main__":
print(extract_fares(FIXTURE_RESPONSE))针对示例固定运行，extract_fares 返回：
text
{'cheapest_price': 312, 'cheapest_layovers': 0, 'offer_count': 2,
'price_insights_lowest': 289, 'price_level': '低', 'typical_price_range': [295, 430]}注意到 cheapest_price（实际返回的最低单一报价）与 price_insights_lowest（谷歌自身为该航线设定的最低价）的差距。这两者很少完全匹配——price_insights 反映了该航线的广泛情况，而不仅是这个响应中的具体报价——因此保留这两个字段，而不是将它们合并为一个数字。
转换：规范化为航线-日期记录
航班监控的标识是 (departure_id, arrival_id, outbound_date, return_date) 的元组，而不是一个 URL。转换阶段构建这个关键字，同时构建规范记录的形状，以便存储和比较不必重新推导它。
python
from datetime import datetime, timezone
def to_record(route: dict, fares: dict) -> dict:
return {
"departure_id": route["departure_id"],
"arrival_id": route["arrival_id"],
"outbound_date": route["outbound_date"],
"return_date": route.get("return_date"),
"trip_type": "round_trip" if route.get("return_date") else "one_way",
"price": fares["cheapest_price"],
"price_level": fares["price_level"],
"currency": "USD",
"checked_at": datetime.now(timezone.utc).strftime("%d-%b-%Y %H:%M UTC"),
}
def route_key(record: dict) -> str:
return f"{record['departure_id']}-{record['arrival_id']}:{record['outbound_date']}:{record.get('return_date')}"
if __name__ == "__main__":
sample_route = {"departure_id": "JFK", "arrival_id": "LAX",
"outbound_date": "2026-08-10", "return_date": "2026-08-17"}
sample_fares = {"cheapest_price": 312, "price_level": "low"}
record = to_record(sample_route, sample_fares)
print(record)
print(route_key(record))针对固定航线（JFK 到 LAX，8 月 10 日出发，8 月 17 日返回）和上述提取的票价，to_record 返回一个记录，其中 price: 312，price_level: 'low'，trip_type: 'round_trip'，而 route_key 返回 JFK-LAX:2026-08-10:2026-08-17。对同一路线和出发日期的单程监控在缺少 return_date 时会产生一个独特的键，正是这一点：同一路线的两种旅行类型是两个不同的监控，而不是一个。
存储：按航线和日期键入的附加日志
存储是相同的附加 JSONL 模式，零售价格监控使用，唯一的区别是查找在航线-日期键上进行，而不是 URL。
python
import json
HISTORY_FILE = "flight_history.jsonl"
def append_history(record: dict) -> dict:
with open(HISTORY_FILE, "a", encoding="utf-8") as f:
f.write(json.dumps(record) + "\n")
return record
def load_history(key: str) -> list[dict]:
rows = []
try:
with open(HISTORY_FILE, encoding="utf-8") as f:
for line in f:
row = json.loads(line)
if route_key(row) == key:
rows.append(row)
except FileNotFoundError:
pass
return rows
if __name__ == "__main__":
open(HISTORY_FILE, "w").close() # 从空日志开始此演示
sample_record = {"departure_id": "JFK", "arrival_id": "LAX", "outbound_date": "2026-08-10",
"return_date": "2026-08-17", "trip_type": "round_trip",
"price": 312, "price_level": "low", "currency": "USD",
"checked_at": "13-Jul-2026 14:28 UTC"}
append_history(sample_record)
append_history(sample_record)
with open(HISTORY_FILE, encoding="utf-8") as f:
print(f"{len(f.readlines())} line(s) in {HISTORY_FILE} after two checks")两次附加相同记录并通过键重新加载返回了插入顺序中的两个行，确认日志是累积而不是覆盖——这与附加文件所提供的保证相同，只是过滤在不同的键上。在处理少数航线后，将文件替换为一个小的 SQLite 表，使用 (departure_id, arrival_id, outbound_date, return_date) 作为复合键；读取和写入形状保持不变。
决定：与历史和谷歌的价格信号进行比较
零售价格监控有一个决策规则：新价格是否低于迄今为止看到的最低价格。票价监控还有第二个信号可用——price_level——因此决策阶段会检查两者：价格是否真的低于当前的最低价，或者首次读取已经落在 price_level: "low" 上。
python
def is_price_drop(history: list[dict], current_price: float, current_level: str) -> dict:
prior_prices = [r["price"] for r in history if r.get("price") is not None]
previous_low = min(prior_prices) if prior_prices else None
numeric_drop = (
current_price is not None
and previous_low is not None
and current_price < previous_low
)
level_signal = current_level == "low"
return {
"dropped": numeric_drop,
"worth_alerting": numeric_drop or (level_signal and previous_low is None),
"current": current_price,
"previous_low": previous_low,
json
"价格等级": 当前等级,
}
如果 __name__ == "__main__":
历史记录 = [{"价格": 349}, {"价格": 340}]
打印(是否价格下跌(历史记录, 当前价格=312, 当前等级="低"))在先前的两次读数349和340之后，新的读数312，带有价格等级: "低"，返回{'下跌': True, '值得警报': True, '当前': 312, '先前低': 340, '价格等级': '低'}。值得警报字段故意比下跌更广泛：如果谷歌自己的分类已经将其称为低值，它也会在某条路线的第一次读数上触发，因为低运行比较时，还没有可以比较的参考，但定性的信号是存在的。这个区分反映了航空票价市场中离线动态定价的研究视为该类别的结构特征：航空座位库存是根据需求和剩余容量进行重新定价的，而不是根据固定的销售日历，因此单一快照在没有价格历史可供比较的情况下已经可以提供信息。
警报：在下跌时触发Webhook
警报机制与零售监视所使用的单个requests.post相同——一个HTTP调用，没有队列，没有代理——消息是根据路线和日期标识构建的，而不是产品名称。
python
def 发送警报(route_label: str, decision: dict, webhook_url: str) -> int:
消息 = (
f"票价监视: {route_label}\n"
f"现在${decision['当前']} (曾经是${decision['先前低']}), "
f"谷歌价格等级={decision['价格等级']}"
)
响应 = requests.post(webhook_url, json={"text": 消息}, timeout=15)
响应.raise_for_status()
return 响应.status_coderaise_for_status()立即显示一个损坏的Webhook端点，而不是让一个配置错误的中继静默吞噬每一个警报。{"text": 消息}主体与常见的Slack/Discord入站Webhook格式相匹配；根据接收端点的期望调整JSON。
调度：按旅行边界轮询节奏
通过警报将获取与警报纳入一个函数，使得单个check_route()调用可以被调度器运行：
python
def check_route(出发地_id, 到达地_id, 出发日期, 返回日期=None, webhook_url=None):
有效载荷 = fetch_route(出发地_id, 到达地_id, 出发日期, 返回日期)
票价 = extract_fares(有效载荷)
路线 = {"出发地_id": 出发地_id, "到达地_id": 到达地_id,
"出发日期": 出发日期, "返回日期": 返回日期}
记录 = append_history(to_record(路线, 票价))
决策 = is_price_drop(load_history(route_key(记录)), 记录["价格"], 记录["价格等级"])
如果 决策["值得警报"] 和 webhook_url:
send_alert(f"{出发地_id}-{到达地_id} {出发日期}", 决策, webhook_url)
返回 记录, 决策对于大多数个人监视器，每日运行足够；跟踪多个定期路线的企业旅行服务台可以更频繁地运行。与零售SKU不同，航班监视还有一个自然结束：一旦出发日期过去，该路线就不再是实时搜索，历史记录也随之结束。在Linux或macOS上使用的最简单驱动程序是cron条目，使用平台上任何计划任务所使用的相同五字段POSIX crontab时间规范：
bash
# crontab -e — 每天检查两次，07:00和19:00
0 7,19 * * * cd /opt/fare-watch && /usr/bin/python3 check.py >> watch.log 2>&1Windows任务调度程序在等效触发器上运行相同的命令，无服务器计时器也可以工作——这个脚本除了历史文件不持有任何状态，因此无状态调用适合它。对于多个路线的监视列表，循环check_route()遍历该列表，并保持并发适度，以便API密钥的速率限制保持舒适。
您得到的回报
每次检查都会将一条记录附加到历史日志中，键值按路线和日期而非URL：
json
// 架构精确反映了to_record()/append_history()的写入内容。
// 字段值是示例性的，与上面使用的固定值相匹配——不是真实读数。
{
"出发地_id": "JFK",
"到达地_id": "LAX",
"出发日期": "2026-08-10",
"返回日期": "2026-08-17",
"旅行类型": "往返",
"价格": 312,
"价格等级": "低",
"货币": "美元",
"检查于": "2026年7月13日 15:01 UTC"
}在大规模运行之前，有几点值得注意：
价格和价格洞察最低值可能会出现分歧。 前者是此特定响应返回的最低报价；后者是谷歌对该路线的更广泛阅读。如果下游用例关注差异，则应同时存储这两者。- 存储的数字应该是全部票价，而不是基础费率。 美国航空公司在广告中宣传的票价必须根据联邦全面票价广告规则 报价乘客所支付的全部价格，而不是削减后的基础费率——将管道存储的任何字段视为
price，这样便于跨检查的比较，永远不会在某一天出现基础票价而在另一日出现全额总价。 - 多城市更改形状，而不是管道。
data_type: 3和multi_city_json生成的是行程级响应，而不是单腿响应；扩展to_record将每段价格合并为总数，而不是将行程视为统一费率。 gl和hl影响货币和语言，而不仅仅是措辞。 像零售监视器那样将两者固定于每次观察，类似于零售监视器固定proxy_country，这样跨检查的比较可以保持在同一市场。- 可为空的价格不是零。 将缺失的
cheapest_price视为None，正如提取阶段所做的那样——偶尔的奇异响应绝不应将假零注入当前的最低值。
结论：一个路线，一天，一个决定
管道简化为与任何其他价格监视器相同的形状——获取、提取、存储、决定、警报、调度——“监视”的身份为该类别重新定义：路线和日期对，而不是 URL，一个旅行类型参数，而不是单一请求形状，以及一个附加的、目标提供的信号（price_level），与简单的数字比较一起可用。将观察列表扩展为更多路线、更多日期或多城市行程，可以不改变每个阶段；只有传递到获取的参数发生改变。
准备好构建您的 AI 驱动数据管道了吗？
加入社区，与在 Scrapeless 上构建数据管道的开发人员交流笔记：Discord · Telegram。
在app.scrapeless.com注册，获取免费的抓取 API 额度，确认账户的计划上已启用 scraper.google.flights 角色，并将上述阶段调整为观察列表需要的路线。请参阅定价了解计划详情，以及抓取 API 文档以获取当前角色参考。
常见问题
问：scraper.google.flights 是真实的、有文档的角色吗？
是的。故意捏造的角色名称返回 "invalid actor: <name>"；请求 scraper.google.flights 返回 "disabled actor: scraper.google.flights"，这是一个明显的错误，仅适用于已识别的、已目录化的角色。它对某些计划是禁用的，而不是不存在；在将生产流量集中于其之前，确认它已启用针对目标账户的计划。
问：为什么要按路线和日期而不是 URL 来关键历史，就像零售监视器那样？
因为航班票价没有像零售 SKU 那样的单一规范页面。相同的路线在不同的出发日期、返回日期和旅行类型下价格不同，因此，“监视”的身份必须是这种组合，而不是单一 URL。
问：price_insights.price_level 较于简单的“比上次便宜”的比较添加了什么？
这是谷歌对当前票价相对于该路线的低、中、高的分类，基于超出管道刚刚收到的单一响应的数据计算得出。将其与当前最低比较结合，允许该路线的第一次读取具有可操作性，而不仅仅是其第二次及以后的读取。
问：这可以处理多城市行程吗？
可以，在请求级别——data_type: 3 与 multi_city_json 数组的行程是针对该角色系列记录的参数形状。扩展 to_record 将多腿行程折叠为一个可比较的总数是多城市监视器所需的额外逻辑。
问：检查频率应该多久一次？
每天或每天两次的监视适合大多数个人票价监视器。航班监视器也有一个自然的停止点：一旦出发日期过去，路线就不再是实时搜索，这与可以无限期监视的零售 SKU 不同。
问：这可以在没有人工智能代理的情况下运行吗？
可以。Fetch 通过 Schedule 中的 Python 独立端到端运行——调用、提取、存储、决定、警报，让调度程序驱动节奏。代理是一种方便的方式，用自然语言驱动路线和日期选择，但管道本身只需要 Python 和调度程序。
在Scrapeless，我们仅访问公开可用的数据，并严格遵循适用的法律、法规和网站隐私政策。本博客中的内容仅供演示之用，不涉及任何非法或侵权活动。我们对使用本博客或第三方链接中的信息不做任何保证，并免除所有责任。在进行任何抓取活动之前，请咨询您的法律顾问，并审查目标网站的服务条款或获取必要的许可。
