如何使用 Python 搜刮 Google 航班：分步指南

在这里，我将向您展示如何创建一个谷歌航班使用 Python 和 Playwright 库的 scraper。从设置环境到提取和保存数据，我们将逐步进行。我还将分享处理常见刮擦问题的技巧，例如 IP 禁止和验证码，因为这些挑战经常出现在谷歌航班等流量巨大的网站上。让我们深入了解一下！

为什么要抓取 Google 航班？

搜索 Google Flights 可以获取大量与航班相关的数据，这些数据有多种用途：

跟踪价格变化： 跟踪航班价格随时间的波动情况，这有助于确定最佳预订时间。
比较航班选择： 查找符合特定要求的航班，如直飞航班、停留时间较短的航班或经济实惠的航班。
市场分析： 对于企业来说，跟踪航空公司和航线的数据可以为定价策略和市场趋势提供洞察力。
环境影响： 提取二氧化碳排放数据，评估不同飞行方案对环境的影响。

自动获取谷歌航班数据的最佳工具

下面列出了一些不同的工具，它们可以帮助你实现数据收集过程的自动化，并节省大量时间和潜在资源：

让我们从设置环境开始，构建我们的自定义 Google Flights 搜刮器。

步骤 1：设置 Python 环境

在开始编码之前，请确保您有一个干净的 Python 环境。最好使用虚拟环境来隔离依赖关系。

创建并激活虚拟环境

打开终端，执行以下命令：

# Create a virtual environment
python -m venv flights-scraper-env
# Activate the virtual environment
# On Windows:
.flights-scraper-env\Scripts\activate
# On macOS/Linux:
source flights-scraper-env/bin/activate

安装必要的软件包

我们将使用 Playwright 与 Google Flights 网站进行交互，因为它可以有效地自动执行动态网页上的操作。Tenacity 将提供重试机制，以提高可靠性。

# Install required packages
pip install playwright tenacity asyncio
# Install Playwright's browser dependencies
playwright install chromium

步骤 2：定义数据结构

为了使代码井井有条，我们将定义数据类来存储搜索参数和提取的航班数据。

from dataclasses import dataclass
from typing import Optional
@dataclass
class SearchParameters:
departure: str
destination: str
departure_date: str
return_date: Optional[str] = None
ticket_type： str = "One way"
@dataclass
class FlightData:
airline: str
departure_time: str
arrival_time: str
duration: str
stops: str
price: str
co2_emissions: str
emissions_variation: str

SearchParameters 包含航班搜索所需的详细信息。
FlightData 存储每个航班的信息，包括航空公司、时间、持续时间、停靠站、价格和二氧化碳排放量。

步骤 3：制作飞行爬虫类

我们的刮板核心将位于 FlightScraper 类中，我们将分阶段构建该类。

定义 CSS 选择器

使用 CSS 选择器，我们可以针对页面上的特定元素提取详细信息，如航空公司名称、起飞时间、价格等。

class FlightScraper:
SELECTORS = {
"airline": "div.sSHqwe.tPgKwe.ogfYpf",
"departure_time": 'span[aria-label^="Departure time"]',
"arrival_time": 'span[aria-label^="Arrival time"]',
"duration": div[aria-label^="Total duration"]',
"stops": "div.hF6lYb span.rGRiKd",
"price": "div.FpEdX span",
"co2_emissions": "div.O7CXue",
"emissions_variation": "div.N6PNV",
}

模拟填写搜索表单

We’ll use Playwright’s asynchronous functions to mimic a user filling in the search parameters on Google Flights.

async def _fill_search_form(self, page, params: SearchParameters) -> None:
ticket_type_div = page.locator("div.VfPpkd-TkwUic[jsname='oYxtQd']").first
await ticket_type_div.click()
await page.locator("li").filter(has_text=params.ticket_type).nth(0).click()
from_input = page.locator("input[aria-label='Where from?']")
await from_input.fill(params.departure)
to_input = page.locator("input[aria-label='Where to?']")
await to_input.fill(params.destination)
date_input = page.locator("input[aria-label='Departure date']")
await date_input.fill(params.departure_date)

加载所有可用航班

Google Flights 通常要求用户点击 "显示更多航班 "来显示所有选项。通过循环点击该按钮，直到不再显示航班为止，从而实现自动化。

async def _load_all_flights(self, page) -> None:
while True:
try:
more_button = await page.wait_for_selector(
'button[aria-label*="more flights"]', timeout=5000
)
if more_button:
await more_button.click()
await page.wait_for_timeout(2000)
else:
break
except:
break

提取飞行数据

所有航班都可见后，循环查看每个航班的元素，并使用 CSS 选择器检索详细信息。

async def _extract_flight_data(self, page) -> list[FlightData]:
await page.wait_for_selector("li.pIav2d", timeout=30000)
flights = await page.query_selector_all("li.pIav2d")
flights_data = []
for flight in flights:
flight_info = {}
for key, selector in self.SELECTORS.items()：
element = await flight.query_selector(selector)
flight_info[key] = await self._extract_text(element)
flights_data.append(FlightData(**flight_info))
return flights_data

步骤 4：实施重试逻辑以提高可靠性

为了使我们的抓取器更有弹性，我们将使用 tenacity 库实现重试机制。

from tenacity import retry, stop_after_attempt, wait_fixed
@retry(stop=stop_after_attempt(3), wait=wait_fixed(5))
async def search_flights(self, params: SearchParameters) -> list[FlightData]:
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
context = await browser.new_context()
page = await context.new_page()
await page.goto("https://www.google.com/flights")
await self._fill_search_form(page, params)
flights = await self._extract_flight_data(page)
await browser.close()
return flights

第 5 步：将结果保存到 JSON 文件中

刮擦数据后，将其存储在 JSON 文件中，以便于检索和分析。

import json
from datetime import datetime
def save_results(self, flights: list[FlightData]、 params: SearchParameters) -> str：
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"flight_results_{params.departure}_{params.destination}_{timestamp}.json"
output_data = {
"search_parameters": vars(params),
"flights": [vars(flight) for flight in flights],
}
with open(filename, "w", encoding="utf-8") as f:
json.dump(output_data, f, indent=2, ensure_ascii=False)
return filename

步骤 6：运行爬虫

以下是如何启动附带搜索参数示例的搜索器。

import asyncio
async def main():
scraper = FlightScraper()
params = SearchParameters(
departure="LAX",
destination="JFK",
departure_date="2024-12-01",
ticket_type="One way"
)
try:
flights = await scraper.search_flights(params)
scraper.save_results(flights, params)
print("Flights scraped successfully.")
except Exception as e:
print(f"Error during flight search: {str(e)}")
if __name__ == "__main__":
asyncio.run(main())

克服常见的搜索难题：IP 屏蔽和验证码

搜索 Google Flights 会遇到独特的挑战。下面介绍如何应对这些挑战：

IP 屏蔽： 使用旋转代理，避免被 Google 的反搜索措施发现。
验证码 实施验证码解决方案，如 Bright Data 的 Web Unlocker，它可以自动绕过验证码挑战。

结论

搜索 Google Flights 可以深入了解旅行模式、价格和航空公司选择。通过使用 Playwright 在 Python 中设置刮擦工具，并遵循可靠性最佳实践，您可以提取有价值的航班数据，供个人或企业使用。为了提高效率，在扩展时应充分利用代理和验证码处理工具。本指南为您提供了基础步骤，有了 Python，您的 Google Flights 搜索能力将无限强大！