Python爬虫实战：使用Playwright抓取Google Flig

脚本专家

对于需要获取航班价格、航线时刻或碳排放数据的开发者来说，Google Flights 是一个信息量很大的数据源。但它的页面由大量动态脚本驱动，传统 requests 方案很难直接提取内容。本文给出一个基于 Python + Playwright 的抓取实现，覆盖环境搭建、数据类定义、搜索表单填充、航班结果提取、重试机制与 JSON 保存，最后讨论 IP 封锁和 CAPTCHA 的常见应对思路。

环境准备
建议先用虚拟环境隔离依赖，避免污染全局 Python。终端下执行：

2. # 创建虚拟环境
3. python -m venv flights-env
5. # 激活虚拟环境（Windows）
6. flights-env\Scripts\activate
8. # 激活虚拟环境（macOS/Linux）
9. source flights-env/bin/activate

复制代码

接着安装 Playwright、Tenacity 和 asyncio 相关包，并下载 Chromium 浏览器内核：

2. pip install playwright tenacity asyncio
3. playwright install chromium

复制代码

Tenacity 用于实现自动重试，Playwright 负责模拟用户操作 Chromium，asyncio 支撑异步流程。

数据结构设计
为了保持代码清晰，先定义两个数据类：一个描述搜索条件，一个描述单条航班结果。

2. from dataclasses import dataclass
3. from typing import Optional
5. @dataclass
6. class SearchParameters:
7.     departure: str
8.     destination: str
9.     departure_date: str
10.     return_date: Optional[str] = None
11.     ticket_type: str = "One way"
13. @dataclass
14. class FlightData:
15.     airline: str
16.     departure_time: str
17.     arrival_time: str
18.     duration: str
19.     stops: str
20.     price: str
21.     co2_emissions: str
22.     emissions_variation: str

复制代码

SearchParameters 中的 ticket_type 支持 "One way"、"Round trip" 等选项；FlightData 与后续 CSS 选择器提取的字段一一对应。

FlightScraper 类与 CSS 选择器
将抓取逻辑封装在 FlightScraper 中，用类属性集中管理 CSS 选择器。这些选择器会随 Google Flights 页面结构调整而变化，抓取前建议先通过 DevTools 确认。

2. class FlightScraper:
3.     SELECTORS = {
4.         "airline": "div.sSHqwe.tPgKwe.ogfYpf",
5.         "departure_time": 'span[aria-label^="Departure time"]',
6.         "arrival_time": 'span[aria-label^="Arrival time"]',
7.         "duration": 'div[aria-label^="Total duration"]',
8.         "stops": "div.hF6lYb span.rGRiKd",
9.         "price": "div.FpEdX span",
10.         "co2_emissions": "div.O7CXue",
11.         "emissions_variation": "div.N6PNV",
12.     }

复制代码

模拟填写搜索表单
用户手动搜索时，会先选择机票类型、输入出发地、目的地和日期。Playwright 的异步 API 可以模拟这些操作：

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

复制代码

这里使用 aria-label 定位输入框，比依赖 class 更稳定。日期格式需要与页面要求一致，例如 "2024-12-01"。

加载所有可用航班
Google Flights 默认只展示一部分结果，需要点击 "more flights" 按钮展开完整列表。通过循环不断点击，直到按钮消失：

2. async def _load_all_flights(self, page) -> None:
3.     while True:
4.         try:
5.             more_button = await page.wait_for_selector(
6.                 'button[aria-label*="more flights"]', timeout=5000
7.             )
8.             if more_button:
9.                 await more_button.click()
10.                 await page.wait_for_timeout(2000)
11.             else:
12.                 break
13.         except Exception:
14.             break

复制代码

提取航班数据
当所有结果都展开后，使用 li.pIav2d 定位每条航班记录，再根据 SELECTORS 提取对应字段。

2. async def _extract_flight_data(self, page) -> list[FlightData]:
3.     await page.wait_for_selector("li.pIav2d", timeout=30000)
4.     flights = await page.query_selector_all("li.pIav2d")
5.     flights_data = []
7.     for flight in flights:
8.         flight_info = {}
9.         for key, selector in self.SELECTORS.items():
10.             element = await flight.query_selector(selector)
11.             if element:
12.                 flight_info[key] = await element.inner_text()
13.             else:
14.                 flight_info[key] = ""
15.         flights_data.append(FlightData(**flight_info))
17.     return flights_data

复制代码

如果某个元素不存在，这里置为空字符串，避免构造 FlightData 时因缺字段报错。

添加重试逻辑
抓取过程中可能遇到网络抖动或页面暂时拒绝响应，可以用 tenacity 装饰器设置最多重试 3 次、每次等待 5 秒。完整的抓取流程如下：

2. from tenacity import retry, stop_after_attempt, wait_fixed
3. from playwright.async_api import async_playwright
5. @retry(stop=stop_after_attempt(3), wait=wait_fixed(5))
6. async def search_flights(self, params: SearchParameters) -> list[FlightData]:
7.     async with async_playwright() as p:
8.         browser = await p.chromium.launch(headless=True)
9.         context = await browser.new_context()
10.         page = await context.new_page()
12.         await page.goto("https://www.google.com/flights")
13.         await self._fill_search_form(page, params)
14.         await self._load_all_flights(page)
15.         flights = await self._extract_flight_data(page)
17.         await browser.close()
18.         return flights

复制代码

这里将 _load_all_flights 也纳入调用链，确保结果完整。

将结果保存为 JSON
抓取到的数据需要落地保存，便于后续分析。用时间戳生成唯一文件名，并同时记录搜索参数和航班列表：

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

复制代码

运行爬虫
下面是一段可执行的启动示例，搜索从 LAX 到 JFK 的单程航班：

2. import asyncio
4. async def main():
5.     scraper = FlightScraper()
6.     params = SearchParameters(
7.         departure="LAX",
8.         destination="JFK",
9.         departure_date="2024-12-01",
10.         ticket_type="One way"
11.     )
13.     try:
14.         flights = await scraper.search_flights(params)
15.         filename = scraper.save_results(flights, params)
16.         print("Flights scraped successfully.")
17.         print(f"Saved to {filename}")
18.     except Exception as e:
19.         print(f"Error during flight search: {str(e)}")
21. if __name__ == "__main__":
22.     asyncio.run(main())

复制代码

注意：FlightScraper 需要完整定义类的继承关系，实际运行时建议将上述方法都放在同一个类作用域中。

应对 IP 封锁和 CAPTCHA
Google Flights 对自动化流量检测很严格，常见的问题有两个：

一是 IP 被封禁。高频请求会被限制访问，解决办法是使用轮换代理，让每次请求的出口 IP 不同。

二是 CAPTCHA 验证。当被判断为机器人时，页面会弹出人机验证。这类问题很难只靠 Playwright 绕过，通常需要接入第三方打码服务或网络解锁器，例如 Bright Data 的 Web Unlocker 就能自动处理 CAPTCHA 挑战。

从工程角度，抓取频率应尽可能放缓，并实时监控页面是否出现异常元素，以便及时切换 IP 或验证码策略。

本教程给出的是一个可扩展的抓取框架，你可以根据实际业务需求调整选择器、增加代理池或改造数据存储层。Google Flights 页面版本更新频繁，选择器可能需要同步修改，但整体流程保持不变。