How to Scrape Google Flights With Python in 2026

https://www.google.com/travel/flights/search?tfs=CBwQAhoeEgoyMDI2LTA2LTAxagcIARIDSkZLcgcIARIDTEhS&hl=en

pip install fast-flights

pip install playwright beautifulsoup4 httpx
playwright install chromium

from fast_flights import FlightData, Passengers, Result, get_flights

# Define your search
result: Result = get_flights(
    flight_data=[
        FlightData(
            date="2026-06-15",       # YYYY-MM-DD format
            from_airport="JFK",    # IATA departure code
            to_airport="LHR",      # IATA arrival code
        )
    ],
    trip="one-way",
    seat="economy",           # economy, premium-economy, business, first
    passengers=Passengers(
        adults=1,
        children=0,
        infants_in_seat=0,
        infants_on_lap=0,
    ),
    fetch_mode="fallback",    # Use fallback if primary parse fails
)

# Check if the price is low, typical, or high vs. historical data
print(f"Price trend: {result.current_price}")

# Loop through flight results
for flight in result.flights:
    print(f"Airline:    {flight.name}")
    print(f"Departure:  {flight.departure}")
    print(f"Arrival:    {flight.arrival}")
    print(f"Duration:   {flight.duration}")
    print(f"Stops:      {flight.stops}")
    print(f"Is best:    {flight.is_best}")
    print("---")

from fast_flights import FlightData, Passengers, Result, get_flights

result: Result = get_flights(
    flight_data=[
        FlightData(
            date="2026-06-15",
            from_airport="JFK",
            to_airport="NRT",    # Tokyo Narita
        ),
        FlightData(
            date="2026-06-29",    # Return leg: add a second FlightData
            from_airport="NRT",
            to_airport="JFK",
        ),
    ],
    trip="round-trip",
    seat="economy",
    passengers=Passengers(adults=1, children=0, infants_in_seat=0, infants_on_lap=0),
    fetch_mode="fallback",
)

print(f"Found {len(result.flights)} results")
print(f"Price trend: {result.current_price}")

for f in result.flights[:5]:  # First 5 results
    print(f"{f.name} | {f.departure} -> {f.arrival} | {f.duration} | {f.stops} stops | Best: {f.is_best}")

import asyncio
import json
import random
from datetime import datetime, timezone
from playwright.async_api import async_playwright, TimeoutError as PlaywrightTimeoutError

# TorchProxies rotating residential credentials
PROXY_HOST = "residential.torchproxies.com"
PROXY_PORT = "31112"
PROXY_USER = "your_username"
PROXY_PASS = "your_password"

COUNTRY_TIMEZONES = {
    "us": "America/New_York",
    "gb": "Europe/London",
    "de": "Europe/Berlin",
    "jp": "Asia/Tokyo",
    "kr": "Asia/Seoul",
    "in": "Asia/Kolkata",
    "id": "Asia/Jakarta",
    "nl": "Europe/Amsterdam",
    "hk": "Asia/Hong_Kong",
    "ca": "America/Toronto",
}

def is_blocked_or_empty(html: str, flights: list) -> bool:
    block_signals = [
        "Our systems have detected unusual traffic",
        "g-recaptcha",
        "sorry.google.com",
    ]
    if any(s in html for s in block_signals):
        return True
    return len(flights) == 0

async def parse_flights_from_dom(page) -> list:
    """
    Prefer ARIA/text-based locators over brittle class names.
    This still may need maintenance when Google changes UI.
    """
    flights = []
    cards = page.locator("li").filter(has_text="Select flight")
    count = await cards.count()

    for i in range(min(count, 20)):
        card = cards.nth(i)
        try:
            text = await card.inner_text(timeout=1000)
            flights.append({
                "raw_text": text,
                "scraped_at": datetime.now(timezone.utc).isoformat(),
            })
        except Exception:
            continue

    return flights

async def scrape_google_flights(
    from_airport: str,
    to_airport: str,
    departure_date_iso: str,  # YYYY-MM-DD
    country_code: str = "us",
    lang: str = "en",
    max_retries: int = 3,
) -> list:
    """
    Note: Google Flights UI is dynamic and localized.
    Validate selectors in your target locales before scaling.
    """
    results = []
    url = f"https://www.google.com/travel/flights?hl={lang}&gl={country_code}"

    async with async_playwright() as pw:
        browser = await pw.chromium.launch(
            headless=True,
            args=["--no-sandbox", "--disable-dev-shm-usage"],
        )

        for attempt in range(1, max_retries + 1):
            context = await browser.new_context(
                proxy={
                    "server": f"http://{PROXY_HOST}:{PROXY_PORT}",
                    "username": PROXY_USER,
                    "password": PROXY_PASS,
                },
                locale=f"{lang}-{country_code.upper()}",
                timezone_id=COUNTRY_TIMEZONES.get(country_code, "America/New_York"),
            )
            await context.add_init_script(
                "Object.defineProperty(navigator, 'webdriver', {get: () => undefined})"
            )
            page = await context.new_page()

            try:
                await page.goto(url, wait_until="domcontentloaded", timeout=45000)

                # Use role/name locators (safer than fragile CSS class selectors)
                await page.get_by_role("button", name="Where from?").click(timeout=5000)
                await page.keyboard.press("Control+A")
                await page.keyboard.type(from_airport, delay=60)
                await page.keyboard.press("Enter")

                await page.get_by_role("button", name="Where to?").click(timeout=5000)
                await page.keyboard.type(to_airport, delay=60)
                await page.keyboard.press("Enter")

                # ISO date (YYYY-MM-DD) is locale-independent
                await page.get_by_role("button", name="Departure").click(timeout=5000)
                await page.keyboard.type(departure_date_iso, delay=50)
                await page.keyboard.press("Enter")

                await page.wait_for_timeout(random.randint(2500, 4500))

                html = await page.content()
                flights = await parse_flights_from_dom(page)

                if is_blocked_or_empty(html, flights):
                    await asyncio.sleep(random.uniform(3, 7))
                    continue

                results = flights
                break

            except PlaywrightTimeoutError:
                await asyncio.sleep(random.uniform(2, 5))
            except Exception:
                await asyncio.sleep(random.uniform(2, 5))
            finally:
                await context.close()

        await browser.close()

    return results

if __name__ == "__main__":
    data = asyncio.run(
        scrape_google_flights(
            from_airport="JFK",
            to_airport="LHR",
            departure_date_iso="2026-06-15",
            country_code="us",
            lang="en",
        )
    )
    print(json.dumps(data[:3], indent=2))
    print(f"Total parsed cards: {len(data)}")

import json
import os
import re
from datetime import datetime
from fast_flights import FlightData, Passengers, get_flights

HISTORY_FILE = "flight_prices.json"
DROP_THRESHOLD_PCT = 10.0  # Alert if price drops >= 10%

def load_history() -> dict:
    if os.path.exists(HISTORY_FILE):
        with open(HISTORY_FILE, "r", encoding="utf-8") as f:
            return json.load(f)
    return {}

def save_history(history: dict) -> None:
    with open(HISTORY_FILE, "w", encoding="utf-8") as f:
        json.dump(history, f, indent=2, ensure_ascii=False)

def parse_price(value) -> float:
    """
    Handles ints/floats or strings like '$823', 'EUR 1,245', 'JPY 42,000'
    """
    if value is None:
        return 0.0
    if isinstance(value, (int, float)):
        return float(value)
    s = str(value)
    cleaned = re.sub(r"[^\d.,]", "", s).replace(",", "")
    try:
        return float(cleaned) if cleaned else 0.0
    except ValueError:
        return 0.0

def check_route(from_airport: str, to_airport: str, departure_date: str, route_key=None):
    key = route_key or f"{from_airport}-{to_airport}-{departure_date}"

    try:
        result = get_flights(
            flight_data=[
                FlightData(
                    date=departure_date,
                    from_airport=from_airport,
                    to_airport=to_airport,
                )
            ],
            trip="one-way",
            seat="economy",
            passengers=Passengers(adults=1, children=0, infants_in_seat=0, infants_on_lap=0),
            fetch_mode="fallback",
        )
    except Exception as e:
        print(f"Error fetching {key}: {e}")
        return

    if not result.flights:
        print(f"{key}: No results (possible block or no inventory).")
        return

    # Filter to flights with a parseable numeric price
    priced = []
    for f in result.flights:
        p = parse_price(getattr(f, "price", None))
        if p > 0:
            priced.append((f, p))

    if not priced:
        print(f"{key}: Flights returned but no parseable numeric price.")
        return

    best_flight, best_price = min(priced, key=lambda x: x[1])
    timestamp = datetime.now().isoformat()

    entry = {
        "timestamp": timestamp,
        "airline": best_flight.name,
        "departure": best_flight.departure,
        "arrival": best_flight.arrival,
        "duration": best_flight.duration,
        "stops": best_flight.stops,
        "price": best_price,
        "price_raw": str(getattr(best_flight, "price", "")),
        "price_trend": getattr(result, "current_price", None),
        "is_best": bool(getattr(best_flight, "is_best", False)),
    }

    history = load_history()
    history.setdefault(key, [])
    previous_price = history[key][-1]["price"] if history[key] else None

    history[key].append(entry)
    save_history(history)

    print(f"{key}: {best_flight.name} | {best_price:.2f} | trend={entry['price_trend']}")

    # Real numeric drop detection
    if previous_price and previous_price > 0:
        drop_pct = ((previous_price - best_price) / previous_price) * 100
        if drop_pct >= DROP_THRESHOLD_PCT:
            print(
                f"ALERT: Price dropped {drop_pct:.1f}% "
                f"({previous_price:.2f} -> {best_price:.2f}) for {key}"
            )

if __name__ == "__main__":
    routes = [
        ("JFK", "LHR", "2026-06-15"),
        ("LHR", "NRT", "2026-07-01"),
        ("AMS", "HKG", "2026-08-10"),
    ]
    for dep, arr, date in routes:
        check_route(dep, arr, date)

# Add to crontab (run: crontab -e)
0 */6 * * * /usr/bin/python3 /your/path/flight_tracker.py >> /your/path/tracker.log 2>&1