paddington-eurostar/scraper/gwr_fares.py

"""
Fetch GWR walk-on single fares from any station to London Paddington.

Uses the GWR journey search API (same API as www.gwr.com ticket search).
Returns per-train cheapest standard-class fare with restrictions already applied.
Cache for 30 days — fares rarely change.
"""

import httpx

_API_URL = "https://api.gwr.com/api/shopping/journeysearch"
# API key is embedded in the GWR web app (appvalues.prod.json)
_API_KEY = "OgovGqAlLp4gWAhL7DQLo7pMCt8GHi2U4SPFiZgG"
_PAD_CODE = "GBQQP"  # London Paddington cluster code as used by GWR website
_WALKON_CODES = {"SSS", "SVS", "SDS", "CDS"}
_MAX_PAGES = 20


def _headers() -> dict:
    return {
        "user-agent": (
            "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
            "(KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36"
        ),
        "accept": "application/json, text/plain, */*",
        "channel": "WEB",
        "content-type": "application/json",
        "apikey": _API_KEY,
        "origin": "https://www.gwr.com",
        "referer": "https://www.gwr.com/",
    }


def _request_body(
    station_crs: str,
    travel_date: str,
    conversation_token: str | None,
    later: bool,
) -> dict:
    return {
        "IsNextOutward": False,
        "IsPreviousOutward": False,
        "IsNextReturn": False,
        "IsPreviousReturn": False,
        "campaignCode": "",
        "validationCode": "",
        "locfrom": f"GB{station_crs}",
        "locto": _PAD_CODE,
        "datetimedepart": f"{travel_date}T00:00:00",
        "outwarddepartafter": True,
        "datetimereturn": None,
        "returndepartafter": False,
        "directServicesOnly": False,
        "firstclass": False,
        "standardclass": True,
        "adults": 1,
        "children": 0,
        "openreturn": False,
        "via": None,
        "avoid": None,
        "isEarlierSearch": False,
        "isLaterSearch": later,
        "isEarlierSearchReturn": False,
        "isLaterSearchReturn": False,
        "railcards": [],
        "conversationToken": conversation_token,
    }


def _run_pages(station_crs: str, travel_date: str, first_class: bool = False):
    """
    Iterate all pages of GWR journey search results.

    Yields (dep_time, fares_list) for each unique departure time seen.
    first_class=True switches the request to first class fares.
    """
    seen: set[str] = set()
    with httpx.Client(headers=_headers(), timeout=30) as client:
        conversation_token = None
        later = False
        for _ in range(_MAX_PAGES):
            body = _request_body(station_crs, travel_date, conversation_token, later)
            if first_class:
                body["firstclass"] = True
                body["standardclass"] = False
            resp = client.post(_API_URL, json=body)
            resp.raise_for_status()
            data = resp.json().get("data", {})
            conversation_token = data.get("conversationToken")
            for journey in data.get("outwardOpenPureReturnFare", []):
                dep_iso = journey.get("departureTime", "")
                dep_time = dep_iso[11:16]  # "HH:MM" from "2026-04-10T09:08:00"
                if not dep_time or dep_time in seen:
                    continue
                seen.add(dep_time)
                yield dep_time, journey.get("journeyFareDetails", [])
            if not data.get("showLaterOutward", False):
                break
            later = True


def fetch(station_crs: str, travel_date: str) -> dict[str, dict]:
    """
    Fetch GWR walk-on single fares from station_crs to London Paddington on travel_date.

    Returns {departure_time: {'ticket': name, 'price': float, 'code': code}}
    where price is in £ and only the cheapest available standard-class walk-on
    ticket per departure (with restrictions already applied by GWR) is kept.
    """
    result: dict[str, dict] = {}
    for dep_time, fares in _run_pages(station_crs, travel_date):
        cheapest = None
        for fare in fares:
            code = fare.get("ticketTypeCode")
            if code not in _WALKON_CODES:
                continue
            if not fare.get("isStandardClass"):
                continue
            price_pence = fare.get("fare", 0)
            if cheapest is None or price_pence < cheapest["price_pence"]:
                cheapest = {
                    "ticket": fare.get("ticketType", ""),
                    "price": price_pence / 100,
                    "price_pence": price_pence,
                    "code": code,
                }
        if cheapest:
            result[dep_time] = {
                "ticket": cheapest["ticket"],
                "price": cheapest["price"],
                "code": cheapest["code"],
            }
    return result


def fetch_advance(station_crs: str, travel_date: str) -> dict[str, dict]:
    """
    Fetch advance fares: cheapest standard advance and first-class advance per departure.

    Makes two sets of paginated API calls (standard class, then first class).
    Returns {departure_time: {'advance_std': dict or None, 'advance_1st': dict or None}}
    where each sub-dict has keys 'ticket', 'price', 'code'.
    """
    std_advance: dict[str, dict] = {}
    for dep_time, fares in _run_pages(station_crs, travel_date, first_class=False):
        cheapest = None
        for fare in fares:
            code = fare.get("ticketTypeCode")
            if code in _WALKON_CODES:
                continue  # skip walk-on fares
            if not fare.get("isStandardClass"):
                continue
            price_pence = fare.get("fare", 0)
            if cheapest is None or price_pence < cheapest["price_pence"]:
                cheapest = {
                    "ticket": fare.get("ticketType", ""),
                    "price": price_pence / 100,
                    "price_pence": price_pence,
                    "code": code,
                }
        if cheapest:
            std_advance[dep_time] = {
                "ticket": cheapest["ticket"],
                "price": cheapest["price"],
                "code": cheapest["code"],
            }

    first_advance: dict[str, dict] = {}
    for dep_time, fares in _run_pages(station_crs, travel_date, first_class=True):
        cheapest = None
        for fare in fares:
            price_pence = fare.get("fare", 0)
            if cheapest is None or price_pence < cheapest["price_pence"]:
                cheapest = {
                    "ticket": fare.get("ticketType", ""),
                    "price": price_pence / 100,
                    "price_pence": price_pence,
                    "code": fare.get("ticketTypeCode"),
                }
        if cheapest:
            first_advance[dep_time] = {
                "ticket": cheapest["ticket"],
                "price": cheapest["price"],
                "code": cheapest["code"],
            }

    all_times = set(std_advance) | set(first_advance)
    return {
        t: {
            "advance_std": std_advance.get(t),
            "advance_1st": first_advance.get(t),
        }
        for t in all_times
    }