Parse Google Flights shopping XHR responses

fast_flights/__init__.py

@@ -8,6 +8,7 @@
     create_query as create_filter,  # alias
 )
 from .fetcher import get_flights, fetch_flights_html
+from .fetch_result import FetchResult
 
 __all__ = [
     "FlightQuery",
@@ -17,5 +18,6 @@
     "create_filter",
     "get_flights",
     "fetch_flights_html",
+    "FetchResult",
     "integrations",
 ]

fast_flights/fetch_result.py

@@ -0,0 +1,12 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+
+@dataclass
+class FetchResult:
+    """Result returned by integrations that capture both HTML and XHR data."""
+
+    html: str = ""
+    xhr_bodies: list[str | bytes] = field(default_factory=list)
+    url: str | None = None

fast_flights/fetcher.py

@@ -2,6 +2,7 @@
 
 from primp import Client
 
+from .fetch_result import FetchResult
 from .integrations.base import Integration
 from .parser import MetaList, parse
 from .querying import Query
@@ -57,8 +58,8 @@ def get_flights(
         q: The query.
         proxy (str, optional): Proxy.
     """
-    html = fetch_flights_html(q, proxy=proxy, integration=integration)
-    return parse(html)
+    fetched = fetch_flights_html(q, proxy=proxy, integration=integration)
+    return parse(fetched)
 
 
 def fetch_flights_html(
@@ -67,7 +68,7 @@ def fetch_flights_html(
     *,
     proxy: str | None = None,
     integration: Integration | None = None,
-) -> str:
+) -> str | FetchResult:
     """Fetch flights and get the **HTML**.
 
     Args:

fast_flights/integrations/__init__.py

@@ -1,4 +1,5 @@
 from .base import Integration
 from .bright_data import BrightData
+from .playwright import Playwright
 
-__all__ = ["Integration", "BrightData"]
+__all__ = ["Integration", "BrightData", "Playwright"]

fast_flights/integrations/base.py

@@ -1,6 +1,7 @@
 import os
 from abc import ABC
 
+from ..fetch_result import FetchResult
 from ..querying import Query
 
 try:
@@ -15,7 +16,7 @@
 class Integration(ABC):
     """Represents an integration."""
 
-    def fetch_html(self, q: Query | str, /) -> str:
+    def fetch_html(self, q: Query | str, /) -> str | FetchResult:
         """Fetch the flights page HTML from a query.
 
         Args:

fast_flights/integrations/playwright.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+from typing import Any
+
+from ..fetch_result import FetchResult
+from ..querying import Query
+from .base import Integration
+
+GOOGLE_FLIGHTS_URL = "https://www.google.com/travel/flights"
+XHR_MARKER = "FlightsFrontendService/GetShoppingResults"
+
+
+class Playwright(Integration):
+    """Local Chromium integration that captures Google Flights shopping XHRs."""
+
+    def fetch_html(self, q: Query | str, /) -> FetchResult:
+        from playwright.sync_api import sync_playwright
+
+        url = q.url() if isinstance(q, Query) else GOOGLE_FLIGHTS_URL + "?q=" + q
+        response_objects: list[Any] = []
+
+        with sync_playwright() as p:
+            browser = p.chromium.launch(headless=True)
+            context = browser.new_context(
+                user_agent=(
+                    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
+                    "AppleWebKit/537.36 (KHTML, like Gecko) "
+                    "Chrome/145.0.0.0 Safari/537.36"
+                ),
+                locale="en-US",
+                viewport={"width": 1280, "height": 900},
+            )
+            page = context.new_page()
+            page.add_init_script("""
+                Object.defineProperty(navigator, 'webdriver', {get: () => undefined});
+                window.chrome = {runtime: {}};
+                Object.defineProperty(navigator, 'plugins', {get: () => [1, 2, 3, 4, 5]});
+                Object.defineProperty(navigator, 'languages', {get: () => ['en-US', 'en']});
+            """)
+
+            def on_response(response: Any) -> None:
+                if XHR_MARKER in response.url:
+                    response_objects.append(response)
+
+            page.on("response", on_response)
+            try:
+                page.goto(
+                    GOOGLE_FLIGHTS_URL,
+                    timeout=25_000,
+                    wait_until="domcontentloaded",
+                )
+                page.wait_for_timeout(1_000)
+            except Exception:
+                pass
+
+            page.goto(url, timeout=25_000, wait_until="domcontentloaded")
+            try:
+                page.wait_for_load_state("networkidle", timeout=12_000)
+            except Exception:
+                pass
+            page.wait_for_timeout(2_000)
+
+            xhr_bodies: list[bytes] = []
+            for response in response_objects:
+                try:
+                    xhr_bodies.append(response.body())
+                except Exception:
+                    pass
+
+            html = page.content()
+            browser.close()
+
+        return FetchResult(html=html, xhr_bodies=xhr_bodies, url=url)

fast_flights/model.py

@@ -18,6 +18,7 @@ class Alliance:
 class JsMetadata:
     airlines: list[Airline]
     alliances: list[Alliance]
+    diagnostics: dict[str, object] | None = None
 
 
 @dataclass