Python爬虫实战:使用Playwright抓取Google Flig
对于需要获取航班价格、航线时刻或碳排放数据的开发者来说,Google Flights 是一个信息量很大的数据源。但它的页面由大量动态脚本驱动,传统 requests 方案很难直接提取内容。本文给出一个基于 Python + Playwright 的抓取实现,覆盖环境搭建、数据类定义、搜索表单填充、航班结果提取、重试机制与 JSON 保存,最后讨论 IP 封锁和 CAPTCHA 的常见应对思路。环境准备
建议先用虚拟环境隔离依赖,避免污染全局 Python。终端下执行:
# 创建虚拟环境
python -m venv flights-env
# 激活虚拟环境(Windows)
flights-env\Scripts\activate
# 激活虚拟环境(macOS/Linux)
source flights-env/bin/activate
接着安装 Playwright、Tenacity 和 asyncio 相关包,并下载 Chromium 浏览器内核:
pip install playwright tenacity asyncio
playwright install chromium
Tenacity 用于实现自动重试,Playwright 负责模拟用户操作 Chromium,asyncio 支撑异步流程。
数据结构设计
为了保持代码清晰,先定义两个数据类:一个描述搜索条件,一个描述单条航班结果。
from dataclasses import dataclass
from typing import Optional
@dataclass
class SearchParameters:
departure: str
destination: str
departure_date: str
return_date: Optional = 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 中的 ticket_type 支持 "One way"、"Round trip" 等选项;FlightData 与后续 CSS 选择器提取的字段一一对应。
FlightScraper 类与 CSS 选择器
将抓取逻辑封装在 FlightScraper 中,用类属性集中管理 CSS 选择器。这些选择器会随 Google Flights 页面结构调整而变化,抓取前建议先通过 DevTools 确认。
class FlightScraper:
SELECTORS = {
"airline": "div.sSHqwe.tPgKwe.ogfYpf",
"departure_time": 'span',
"arrival_time": 'span',
"duration": 'div',
"stops": "div.hF6lYb span.rGRiKd",
"price": "div.FpEdX span",
"co2_emissions": "div.O7CXue",
"emissions_variation": "div.N6PNV",
}
模拟填写搜索表单
用户手动搜索时,会先选择机票类型、输入出发地、目的地和日期。Playwright 的异步 API 可以模拟这些操作:
async def _fill_search_form(self, page, params: SearchParameters) -> None:
ticket_type_div = page.locator("div.VfPpkd-TkwUic").first
await ticket_type_div.click()
await page.locator("li").filter(has_text=params.ticket_type).nth(0).click()
from_input = page.locator("input")
await from_input.fill(params.departure)
to_input = page.locator("input")
await to_input.fill(params.destination)
date_input = page.locator("input")
await date_input.fill(params.departure_date)
这里使用 aria-label 定位输入框,比依赖 class 更稳定。日期格式需要与页面要求一致,例如 "2024-12-01"。
加载所有可用航班
Google Flights 默认只展示一部分结果,需要点击 "more flights" 按钮展开完整列表。通过循环不断点击,直到按钮消失:
async def _load_all_flights(self, page) -> None:
while True:
try:
more_button = await page.wait_for_selector(
'button', timeout=5000
)
if more_button:
await more_button.click()
await page.wait_for_timeout(2000)
else:
break
except Exception:
break
提取航班数据
当所有结果都展开后,使用 li.pIav2d 定位每条航班记录,再根据 SELECTORS 提取对应字段。
async def _extract_flight_data(self, page) -> list:
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)
if element:
flight_info = await element.inner_text()
else:
flight_info = ""
flights_data.append(FlightData(**flight_info))
return flights_data
如果某个元素不存在,这里置为空字符串,避免构造 FlightData 时因缺字段报错。
添加重试逻辑
抓取过程中可能遇到网络抖动或页面暂时拒绝响应,可以用 tenacity 装饰器设置最多重试 3 次、每次等待 5 秒。完整的抓取流程如下:
from tenacity import retry, stop_after_attempt, wait_fixed
from playwright.async_api import async_playwright
@retry(stop=stop_after_attempt(3), wait=wait_fixed(5))
async def search_flights(self, params: SearchParameters) -> list:
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)
await self._load_all_flights(page)
flights = await self._extract_flight_data(page)
await browser.close()
return flights
这里将 _load_all_flights 也纳入调用链,确保结果完整。
将结果保存为 JSON
抓取到的数据需要落地保存,便于后续分析。用时间戳生成唯一文件名,并同时记录搜索参数和航班列表:
import json
from datetime import datetime
def save_results(self, flights: list, 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": ,
}
with open(filename, "w", encoding="utf-8") as f:
json.dump(output_data, f, indent=2, ensure_ascii=False)
return filename
运行爬虫
下面是一段可执行的启动示例,搜索从 LAX 到 JFK 的单程航班:
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)
filename = scraper.save_results(flights, params)
print("Flights scraped successfully.")
print(f"Saved to {filename}")
except Exception as e:
print(f"Error during flight search: {str(e)}")
if __name__ == "__main__":
asyncio.run(main())
注意:FlightScraper 需要完整定义类的继承关系,实际运行时建议将上述方法都放在同一个类作用域中。
应对 IP 封锁和 CAPTCHA
Google Flights 对自动化流量检测很严格,常见的问题有两个:
一是 IP 被封禁。高频请求会被限制访问,解决办法是使用轮换代理,让每次请求的出口 IP 不同。
二是 CAPTCHA 验证。当被判断为机器人时,页面会弹出人机验证。这类问题很难只靠 Playwright 绕过,通常需要接入第三方打码服务或网络解锁器,例如 Bright Data 的 Web Unlocker 就能自动处理 CAPTCHA 挑战。
从工程角度,抓取频率应尽可能放缓,并实时监控页面是否出现异常元素,以便及时切换 IP 或验证码策略。
本教程给出的是一个可扩展的抓取框架,你可以根据实际业务需求调整选择器、增加代理池或改造数据存储层。Google Flights 页面版本更新频繁,选择器可能需要同步修改,但整体流程保持不变。
Re: Python爬虫实战:使用Playwright抓取Google Flig
感谢楼主分享,这个实战挺有参考价值的。Playwright 确实比 requests 适合这种动态页面,模拟用户行为也相对不容易被简单识别。有几个想请教的地方: 1. 搜索表单里的日期输入,Google Flights 现在好像有些版本是只读的,直接 fill 不一定有效,楼主在代码里有没有额外处理? 2. CSS 选择器部分,因为 Google Flights 前端改版挺勤的,有没有考虑过备用选择器或者更宽松的定位方式? 3. 关于 IP 封锁和 CAPTCHA,方便再展开说说实际踩坑时用到的具体策略吗?比如代理池、请求间隔等是怎么调参的? 另外,如果能把代码里 `_load_all_flights` 后面的部分也贴出来就更好啦,想看看重试机制和结果整理这块是怎么写的。总之学到了,收藏备用。Re: Python爬虫实战:使用Playwright抓取Google Flig
感谢分享!正好最近也在研究用Playwright抓动态页面,这篇把数据结构、选择器管理和重试机制都讲得很清楚,对我很有帮助。 想请教一下,Google Flights 对自动化操作检测挺严的,楼主实际测试下来,IP 封锁和 CAPTCHA 出现的频率大概多高?文中提到的应对思路有没有比较有效的方法?另外这些 CSS 选择器会不会很容易失效?大概多久需要维护一次?Re: Python爬虫实战:使用Playwright抓取Google Flig
楼主这篇实战帖干货很足,正好最近我也在折腾类似的需求,传统 requests 拿不到动态内容太头疼了。你用 Playwright 模拟操作这条路确实稳,特别是把 CSS 选择器集中管理这点,后续页面改版也好维护。想请教一下:日期输入那块,Google Flights 有时候对浏览器语言环境敏感,直接用 `2024-12-01` 这种格式会不会在某些区域设置下失效?有没有做过本地化适配?另外重试机制里 Tenacity 配合异步页面的等待策略,具体是监控哪些异常类型?希望楼主能再展开说说,感谢分享!
页:
[1]