Compare commits

..

No commits in common. "6b044b9493dfe84f2f224b633b3056a7be6398e5" and "4de4c1d556f8968acf0075b5eeebda8832f3675e" have entirely different histories.

9 changed files with 56 additions and 695 deletions

45
app.py
View file

@ -7,8 +7,7 @@ from datetime import date, timedelta
from cache import get_cached, set_cached from cache import get_cached, set_cached
import scraper.eurostar as eurostar_scraper import scraper.eurostar as eurostar_scraper
import scraper.realtime_trains as rtt_scraper import scraper.realtime_trains as rtt_scraper
from trip_planner import combine_trips, find_unreachable_morning_eurostars from trip_planner import combine_trips
from scraper.eurostar import fetch_prices as fetch_eurostar_prices
RTT_PADDINGTON_URL = ( RTT_PADDINGTON_URL = (
"https://www.realtimetrains.co.uk/search/detailed/" "https://www.realtimetrains.co.uk/search/detailed/"
@ -81,12 +80,10 @@ def results(slug, travel_date):
rtt_cache_key = f"rtt_{travel_date}" rtt_cache_key = f"rtt_{travel_date}"
es_cache_key = f"eurostar_{travel_date}_{destination}" es_cache_key = f"eurostar_{travel_date}_{destination}"
prices_cache_key = f"eurostar_prices_{travel_date}_{destination}"
cached_rtt = get_cached(rtt_cache_key) cached_rtt = get_cached(rtt_cache_key)
cached_es = get_cached(es_cache_key) cached_es = get_cached(es_cache_key)
cached_prices = get_cached(prices_cache_key, ttl=24 * 3600) from_cache = bool(cached_rtt and cached_es)
from_cache = bool(cached_rtt and cached_es and cached_prices)
error = None error = None
@ -111,44 +108,8 @@ def results(slug, travel_date):
msg = f"Could not fetch Eurostar times: {e}" msg = f"Could not fetch Eurostar times: {e}"
error = f"{error}; {msg}" if error else msg error = f"{error}; {msg}" if error else msg
if cached_prices:
eurostar_prices = cached_prices
else:
try:
eurostar_prices = fetch_eurostar_prices(destination, travel_date)
set_cached(prices_cache_key, eurostar_prices)
except Exception as e:
eurostar_prices = {}
msg = f"Could not fetch Eurostar prices: {e}"
error = f"{error}; {msg}" if error else msg
trips = combine_trips(gwr_trains, eurostar_trains, travel_date, min_connection, max_connection) trips = combine_trips(gwr_trains, eurostar_trains, travel_date, min_connection, max_connection)
# Annotate each trip with Eurostar Standard price and total cost
for trip in trips:
es_price = eurostar_prices.get(trip['depart_st_pancras'])
trip['eurostar_price'] = es_price
if es_price is not None:
trip['total_price'] = trip['ticket_price'] + es_price
else:
trip['total_price'] = None
unreachable_morning_services = find_unreachable_morning_eurostars(
gwr_trains,
eurostar_trains,
travel_date,
min_connection,
max_connection,
)
for svc in unreachable_morning_services:
svc['eurostar_price'] = eurostar_prices.get(svc['depart_st_pancras'])
result_rows = sorted(
[{'row_type': 'trip', **trip} for trip in trips]
+ [{'row_type': 'unreachable', **service} for service in unreachable_morning_services],
key=lambda row: row['depart_st_pancras'],
)
dt = date.fromisoformat(travel_date) dt = date.fromisoformat(travel_date)
prev_date = (dt - timedelta(days=1)).isoformat() prev_date = (dt - timedelta(days=1)).isoformat()
next_date = (dt + timedelta(days=1)).isoformat() next_date = (dt + timedelta(days=1)).isoformat()
@ -160,8 +121,6 @@ def results(slug, travel_date):
return render_template( return render_template(
'results.html', 'results.html',
trips=trips, trips=trips,
result_rows=result_rows,
unreachable_morning_services=unreachable_morning_services,
destinations=DESTINATIONS, destinations=DESTINATIONS,
destination=destination, destination=destination,
travel_date=travel_date, travel_date=travel_date,

View file

@ -1,6 +1,5 @@
import json import json
import os import os
import time
CACHE_DIR = os.path.join(os.path.dirname(__file__), 'cache') CACHE_DIR = os.path.join(os.path.dirname(__file__), 'cache')
@ -10,13 +9,10 @@ def _cache_path(key: str) -> str:
return os.path.join(CACHE_DIR, f"{safe_key}.json") return os.path.join(CACHE_DIR, f"{safe_key}.json")
def get_cached(key: str, ttl: int | None = None): def get_cached(key: str):
"""Return cached data, or None if missing or older than ttl seconds."""
path = _cache_path(key) path = _cache_path(key)
if not os.path.exists(path): if not os.path.exists(path):
return None return None
if ttl is not None and time.time() - os.path.getmtime(path) > ttl:
return None
with open(path) as f: with open(path) as f:
return json.load(f) return json.load(f)

View file

@ -1,7 +1,7 @@
""" """
Scrape Eurostar timetable via httpx and fetch prices via the GraphQL API. Scrape Eurostar timetable via httpx.
Timetable: route-specific pages are Next.js SSR all departure data is The route-specific timetable pages are Next.js SSR all departure data is
embedded in <script id="__NEXT_DATA__"> as JSON, so no browser / JS needed. embedded in <script id="__NEXT_DATA__"> as JSON, so no browser / JS needed.
URL pattern: URL pattern:
@ -12,19 +12,10 @@ Data path: props.pageProps.pageData.liveDepartures[]
.origin.model.scheduledDepartureDateTime London departure .origin.model.scheduledDepartureDateTime London departure
.destination.model.scheduledArrivalDateTime destination arrival .destination.model.scheduledArrivalDateTime destination arrival
(already filtered to the requested stop, not the final stop) (already filtered to the requested stop, not the final stop)
Prices: POST https://site-api.eurostar.com/gateway (GraphQL, operationName
NewBookingSearch). The `journeys[].fares[]` array contains one entry per
class of service; we extract the Eurostar Standard (classOfService.code ==
"STANDARD") displayPrice for 1 adult, in GBP.
""" """
import json import json
import random
import re import re
import string
import httpx import httpx
import requests
DEFAULT_UA = ( DEFAULT_UA = (
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
@ -99,105 +90,3 @@ def fetch(destination: str, travel_date: str,
r = client.get(url, params={'date': travel_date}) r = client.get(url, params={'date': travel_date})
r.raise_for_status() r.raise_for_status()
return _parse(r.text, destination) return _parse(r.text, destination)
# ---------------------------------------------------------------------------
# Price fetching via site-api.eurostar.com GraphQL
# ---------------------------------------------------------------------------
_GATEWAY_URL = 'https://site-api.eurostar.com/gateway'
# Minimal query requesting only timing + Eurostar Standard fare price.
# Variable names and inline argument names match what the site sends so the
# server-side query planner sees a familiar shape.
_GQL_PRICES = (
"query NewBookingSearch("
"$origin:String!,$destination:String!,$outbound:String!,"
"$currency:Currency!,$adult:Int,"
"$filteredClassesOfService:[ClassOfServiceEnum]"
"){"
"journeySearch("
"outboundDate:$outbound origin:$origin destination:$destination"
" adults:$adult currency:$currency"
" productFamilies:[\"PUB\"] contractCode:\"EIL_ALL\""
" adults16Plus:0 children:0 youths:0 children4Only:0 children5To11:0"
" infants:0 adultsWheelchair:0 childrenWheelchair:0 guideDogs:0"
" wheelchairCompanions:0 nonWheelchairCompanions:0"
" isAftersales:false multipleFlexibility:true showAllSummatedFares:false"
" seniorsAges:[] prioritiseShortHaulODTrains:true"
"){"
"outbound{"
"journeys("
"hideIndirectTrainsWhenDisruptedAndCancelled:false"
" hideDepartedTrains:true"
" hideExternalCarrierTrains:true"
" hideDirectExternalCarrierTrains:true"
"){"
"timing{departureTime:departs __typename}"
"fares(filteredClassesOfService:$filteredClassesOfService){"
"classOfService{code __typename}"
"prices{displayPrice __typename}"
"seats __typename"
"}"
"__typename"
"}"
"__typename"
"}"
"__typename"
"}"
"}"
)
def _generate_cid() -> str:
chars = string.ascii_letters + string.digits
return 'SRCH-' + ''.join(random.choices(chars, k=22))
def fetch_prices(destination: str, travel_date: str) -> dict[str, int | None]:
"""
Return Eurostar Standard prices for every departure on travel_date.
Result: {depart_st_pancras: price_gbp_int_or_None}
None means the class is sold out or unavailable for that departure.
"""
dest_id = DESTINATION_STATION_IDS[destination]
headers = {
'User-Agent': DEFAULT_UA,
'Content-Type': 'application/json',
'Accept': '*/*',
'Accept-Language': 'en-GB',
'Referer': 'https://www.eurostar.com/',
'x-platform': 'web',
'x-market-code': 'uk',
'x-source-url': 'search-app/',
'cid': _generate_cid(),
}
payload = {
'operationName': 'NewBookingSearch',
'variables': {
'origin': ORIGIN_STATION_ID,
'destination': dest_id,
'outbound': travel_date,
'currency': 'GBP',
'adult': 1,
'filteredClassesOfService': ['STANDARD'],
},
'query': _GQL_PRICES,
}
resp = requests.post(_GATEWAY_URL, json=payload, headers=headers, timeout=20)
resp.raise_for_status()
data = resp.json()
prices: dict[str, int | None] = {}
journeys = data['data']['journeySearch']['outbound']['journeys']
for journey in journeys:
dep = journey['timing']['departureTime']
price = None
for fare in journey['fares']:
if fare['classOfService']['code'] == 'STANDARD':
p = fare.get('prices')
if p and p.get('displayPrice'):
price = int(p['displayPrice'])
break
prices[dep] = price
return prices

View file

@ -3,15 +3,7 @@
<head> <head>
<meta charset="UTF-8"> <meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1"> <meta name="viewport" content="width=device-width, initial-scale=1">
<title>{% block title %}Bristol to Europe via Eurostar{% endblock %}</title> <title>Bristol to Europe via Eurostar</title>
<meta property="og:type" content="website">
<meta property="og:site_name" content="Bristol to Europe via Eurostar">
<meta property="og:title" content="{% block og_title %}Bristol to Europe via Eurostar{% endblock %}">
<meta property="og:description" content="{% block og_description %}Find Bristol Temple Meads to Europe itineraries via Paddington, St Pancras, and Eurostar.{% endblock %}">
<meta property="og:url" content="{{ request.url }}">
<meta name="twitter:card" content="summary">
<meta name="twitter:title" content="{% block twitter_title %}Bristol to Europe via Eurostar{% endblock %}">
<meta name="twitter:description" content="{% block twitter_description %}Find Bristol Temple Meads to Europe itineraries via Paddington, St Pancras, and Eurostar.{% endblock %}">
<link rel="icon" href="{{ url_for('static', filename='favicon.svg') }}" type="image/svg+xml"> <link rel="icon" href="{{ url_for('static', filename='favicon.svg') }}" type="image/svg+xml">
<style> <style>
*, *::before, *::after { box-sizing: border-box; } *, *::before, *::after { box-sizing: border-box; }

View file

@ -1,15 +1,5 @@
{% extends "base.html" %} {% extends "base.html" %}
{% block title %}Bristol to {{ destination }} via Eurostar{% endblock %}
{% block og_title %}Bristol to {{ destination }} via Eurostar{% endblock %}
{% block og_description %}Train options from Bristol Temple Meads to {{ destination }} on {{ travel_date_display }} via Paddington, St Pancras, and Eurostar.{% endblock %}
{% block twitter_title %}Bristol to {{ destination }} via Eurostar{% endblock %}
{% block twitter_description %}Train options from Bristol Temple Meads to {{ destination }} on {{ travel_date_display }} via Paddington, St Pancras, and Eurostar.{% endblock %}
{% block content %} {% block content %}
<style>
@media (max-width: 640px) {
.col-transfer { display: none; }
}
</style>
<p style="margin-bottom:1rem"> <p style="margin-bottom:1rem">
<a href="{{ url_for('index') }}">&larr; New search</a> <a href="{{ url_for('index') }}">&larr; New search</a>
@ -80,12 +70,6 @@
{{ gwr_count }} GWR service{{ 's' if gwr_count != 1 }} {{ gwr_count }} GWR service{{ 's' if gwr_count != 1 }}
&nbsp;&middot;&nbsp; &nbsp;&middot;&nbsp;
{{ eurostar_count }} Eurostar service{{ 's' if eurostar_count != 1 }} {{ eurostar_count }} Eurostar service{{ 's' if eurostar_count != 1 }}
{% if unreachable_morning_services %}
&nbsp;&middot;&nbsp;
<span style="color:#718096">
{{ unreachable_morning_services | length }} morning service{{ 's' if unreachable_morning_services | length != 1 }} unavailable from Bristol
</span>
{% endif %}
{% if from_cache %} {% if from_cache %}
&nbsp;&middot;&nbsp; <span style="color:#718096;font-size:0.85rem">(cached)</span> &nbsp;&middot;&nbsp; <span style="color:#718096;font-size:0.85rem">(cached)</span>
{% endif %} {% endif %}
@ -97,106 +81,62 @@
{% endif %} {% endif %}
</div> </div>
{% if trips or unreachable_morning_services %} {% if trips %}
<div class="card" style="overflow-x:auto"> <div class="card" style="overflow-x:auto">
<table style="width:100%;border-collapse:collapse;font-size:0.95rem"> <table style="width:100%;border-collapse:collapse;font-size:0.95rem">
<thead> <thead>
<tr style="border-bottom:2px solid #e2e8f0;text-align:left"> <tr style="border-bottom:2px solid #e2e8f0;text-align:left">
<th style="padding:0.6rem 0.8rem;white-space:nowrap">Bristol</th> <th style="padding:0.6rem 0.8rem;white-space:nowrap">Bristol</th>
<th style="padding:0.6rem 0.8rem;white-space:nowrap">Paddington</th> <th style="padding:0.6rem 0.8rem;white-space:nowrap">Paddington</th>
<th style="padding:0.6rem 0.8rem;white-space:nowrap">GWR Fare</th> <th style="padding:0.6rem 0.8rem;white-space:nowrap">Transfer</th>
<th class="col-transfer" style="padding:0.6rem 0.8rem;white-space:nowrap">Transfer</th>
<th style="padding:0.6rem 0.8rem;white-space:nowrap">Depart St&nbsp;Pancras</th> <th style="padding:0.6rem 0.8rem;white-space:nowrap">Depart St&nbsp;Pancras</th>
<th style="padding:0.6rem 0.8rem">{{ destination }}</th> <th style="padding:0.6rem 0.8rem">{{ destination }}
<th style="padding:0.6rem 0.8rem;white-space:nowrap">ES Std</th> </th>
<th style="padding:0.6rem 0.8rem;white-space:nowrap">Total</th> <th style="padding:0.6rem 0.8rem;white-space:nowrap">Total</th>
</tr> </tr>
</thead> </thead>
<tbody> <tbody>
{% if trips %}
{% set best_mins = trips | map(attribute='total_minutes') | min %} {% set best_mins = trips | map(attribute='total_minutes') | min %}
{% set worst_mins = trips | map(attribute='total_minutes') | max %} {% set worst_mins = trips | map(attribute='total_minutes') | max %}
{% endif %} {% for trip in trips %}
{% for row in result_rows %} {% if trip.total_minutes == best_mins and trips | length > 1 %}
{% if row.row_type == 'trip' and row.total_minutes <= best_mins + 5 and trips | length > 1 %}
{% set row_bg = 'background:#f0fff4' %} {% set row_bg = 'background:#f0fff4' %}
{% elif row.row_type == 'trip' and row.total_minutes >= worst_mins - 5 and trips | length > 1 %} {% elif trip.total_minutes == worst_mins and trips | length > 1 %}
{% set row_bg = 'background:#fff5f5' %} {% set row_bg = 'background:#fff5f5' %}
{% elif row.row_type == 'unreachable' %}
{% set row_bg = 'background:#f7fafc;color:#a0aec0' %}
{% elif loop.index is odd %} {% elif loop.index is odd %}
{% set row_bg = 'background:#f7fafc' %} {% set row_bg = 'background:#f7fafc' %}
{% else %} {% else %}
{% set row_bg = '' %} {% set row_bg = '' %}
{% endif %} {% endif %}
<tr style="border-bottom:1px solid #e2e8f0;{{ row_bg }}"> <tr style="border-bottom:1px solid #e2e8f0;{{ row_bg }}">
{% if row.row_type == 'trip' %}
<td style="padding:0.6rem 0.8rem;font-weight:600"> <td style="padding:0.6rem 0.8rem;font-weight:600">
{{ row.depart_bristol }} {{ trip.depart_bristol }}
{% if row.headcode %}<br><span style="font-size:0.75rem;font-weight:400;color:#718096">{{ row.headcode }}</span>{% endif %} {% if trip.headcode %}<br><span style="font-size:0.75rem;font-weight:400;color:#718096">{{ trip.headcode }}</span>{% endif %}
</td> </td>
<td style="padding:0.6rem 0.8rem"> <td style="padding:0.6rem 0.8rem">
{{ row.arrive_paddington }} {{ trip.arrive_paddington }}
<span style="font-size:0.8rem;color:#718096">({{ row.gwr_duration }})</span> <span style="font-size:0.8rem;color:#718096">({{ trip.gwr_duration }})</span>
</td> </td>
<td style="padding:0.6rem 0.8rem;white-space:nowrap"> <td style="padding:0.6rem 0.8rem;color:#4a5568">
£{{ "%.2f"|format(row.ticket_price) }} {{ trip.connection_duration }}
<br><span style="font-size:0.75rem;color:#718096">{{ row.ticket_name }}</span>
</td>
<td class="col-transfer" style="padding:0.6rem 0.8rem;color:#4a5568">
{{ row.connection_duration }}
</td> </td>
<td style="padding:0.6rem 0.8rem;font-weight:600"> <td style="padding:0.6rem 0.8rem;font-weight:600">
{{ row.depart_st_pancras }} {{ trip.depart_st_pancras }}
{% if row.train_number %}<br><span style="font-size:0.75rem;font-weight:400;color:#718096">{{ row.train_number }}</span>{% endif %} {% if trip.train_number %}<br><span style="font-size:0.75rem;font-weight:400;color:#718096">{{ trip.train_number }}</span>{% endif %}
</td> </td>
<td style="padding:0.6rem 0.8rem"> <td style="padding:0.6rem 0.8rem">
{{ row.arrive_destination }} {{ trip.arrive_destination }}
<span style="font-weight:400;color:#718096;font-size:0.85em">(CET)</span> <span style="font-weight:400;color:#718096;font-size:0.85em">(CET)</span>
</td> </td>
<td style="padding:0.6rem 0.8rem;white-space:nowrap">
{% if row.eurostar_price is not none %}
£{{ row.eurostar_price }}
{% else %}
<span style="color:#718096">&ndash;</span>
{% endif %}
</td>
<td style="padding:0.6rem 0.8rem;font-weight:600;white-space:nowrap">
{% if row.total_minutes <= best_mins + 5 and trips | length > 1 %}
<span style="color:#276749" title="Fastest option">{{ row.total_duration }} ⚡</span>
{% elif row.total_minutes >= worst_mins - 5 and trips | length > 1 %}
<span style="color:#c53030" title="Slowest option">{{ row.total_duration }} 🐢</span>
{% else %}
<span style="color:#00539f">{{ row.total_duration }}</span>
{% endif %}
{% if row.total_price is not none %}
<br><span style="font-size:0.8rem;font-weight:700;color:#276749">£{{ "%.2f"|format(row.total_price) }}</span>
{% endif %}
</td>
{% else %}
<td style="padding:0.6rem 0.8rem;font-weight:600">&mdash;</td>
<td style="padding:0.6rem 0.8rem">&mdash;</td>
<td class="col-transfer" style="padding:0.6rem 0.8rem">&mdash;</td>
<td style="padding:0.6rem 0.8rem">Unavailable</td>
<td style="padding:0.6rem 0.8rem;font-weight:600"> <td style="padding:0.6rem 0.8rem;font-weight:600">
{{ row.depart_st_pancras }} {% if trip.total_minutes == best_mins and trips | length > 1 %}
{% if row.train_number %}<br><span style="font-size:0.75rem;font-weight:400;color:#a0aec0">{{ row.train_number }}</span>{% endif %} <span style="color:#276749" title="Fastest option">{{ trip.total_duration }} ⚡</span>
</td> {% elif trip.total_minutes == worst_mins and trips | length > 1 %}
<td style="padding:0.6rem 0.8rem"> <span style="color:#c53030" title="Slowest option">{{ trip.total_duration }} 🐢</span>
{{ row.arrive_destination }}
<span style="font-weight:400;color:#a0aec0;font-size:0.85em">(CET)</span>
</td>
<td style="padding:0.6rem 0.8rem;white-space:nowrap">
{% if row.eurostar_price is not none %}
<span style="color:#a0aec0">£{{ row.eurostar_price }}</span>
{% else %} {% else %}
<span style="color:#a0aec0">&ndash;</span> <span style="color:#00539f">{{ trip.total_duration }}</span>
{% endif %} {% endif %}
</td> </td>
<td style="padding:0.6rem 0.8rem;font-weight:600">
<span title="No same-day Bristol connection" style="color:#a0aec0">Unavailable from Bristol</span>
</td>
{% endif %}
</tr> </tr>
{% endfor %} {% endfor %}
</tbody> </tbody>
@ -205,11 +145,7 @@
<p style="margin-top:1rem;font-size:0.82rem;color:#718096"> <p style="margin-top:1rem;font-size:0.82rem;color:#718096">
Paddington &rarr; St&nbsp;Pancras connection: {{ min_connection }}&ndash;{{ max_connection }}&nbsp;min. Paddington &rarr; St&nbsp;Pancras connection: {{ min_connection }}&ndash;{{ max_connection }}&nbsp;min.
{% if unreachable_morning_services %} Eurostar times are from the general timetable and may vary; always check
Morning means Eurostar departures before 12:00 from St&nbsp;Pancras.
{% endif %}
GWR walk-on single prices for Bristol Temple Meads&nbsp;&rarr;&nbsp;Paddington.
Eurostar Standard prices are for 1 adult in GBP; always check
<a href="{{ eurostar_url }}" target="_blank" rel="noopener">eurostar.com</a> to book. <a href="{{ eurostar_url }}" target="_blank" rel="noopener">eurostar.com</a> to book.
&nbsp;&middot;&nbsp; &nbsp;&middot;&nbsp;
<a href="{{ rtt_url }}" target="_blank" rel="noopener">Paddington arrivals on RTT</a> <a href="{{ rtt_url }}" target="_blank" rel="noopener">Paddington arrivals on RTT</a>

View file

@ -6,8 +6,8 @@ def _client():
return app_module.app.test_client() return app_module.app.test_client()
def _stub_data(monkeypatch, prices=None): def _stub_data(monkeypatch):
monkeypatch.setattr(app_module, 'get_cached', lambda key, ttl=None: None) monkeypatch.setattr(app_module, 'get_cached', lambda key: None)
monkeypatch.setattr(app_module, 'set_cached', lambda key, data: None) monkeypatch.setattr(app_module, 'set_cached', lambda key, data: None)
monkeypatch.setattr( monkeypatch.setattr(
app_module.rtt_scraper, app_module.rtt_scraper,
@ -33,8 +33,6 @@ def _stub_data(monkeypatch, prices=None):
'timetable_url', 'timetable_url',
lambda destination: f'https://example.test/{destination.lower().replace(" ", "-")}', lambda destination: f'https://example.test/{destination.lower().replace(" ", "-")}',
) )
_prices = prices if prices is not None else {}
monkeypatch.setattr(app_module, 'fetch_eurostar_prices', lambda dest, date: _prices)
def test_index_shows_fixed_departure_and_destination_radios(): def test_index_shows_fixed_departure_and_destination_radios():
@ -74,199 +72,3 @@ def test_results_shows_same_day_destination_switcher(monkeypatch):
assert '/results/brussels/2026-04-10?min_connection=60&amp;max_connection=120' in html assert '/results/brussels/2026-04-10?min_connection=60&amp;max_connection=120' in html
assert '/results/rotterdam/2026-04-10?min_connection=60&amp;max_connection=120' in html assert '/results/rotterdam/2026-04-10?min_connection=60&amp;max_connection=120' in html
assert 'ES 9014' in html assert 'ES 9014' in html
def test_results_title_and_social_meta_include_destination(monkeypatch):
_stub_data(monkeypatch)
client = _client()
resp = client.get('/results/lille/2026-04-10?min_connection=60&max_connection=120')
html = resp.get_data(as_text=True)
assert resp.status_code == 200
assert '<title>Bristol to Lille Europe via Eurostar</title>' in html
assert '<meta property="og:title" content="Bristol to Lille Europe via Eurostar">' in html
assert (
'<meta property="og:description" content="Train options from Bristol Temple Meads '
'to Lille Europe on Friday 10 April 2026 via Paddington, St Pancras, and Eurostar.">'
) in html
assert '<meta property="og:url" content="http://localhost/results/lille/2026-04-10?min_connection=60&amp;max_connection=120">' in html
def test_results_marks_trips_within_five_minutes_of_fastest_and_slowest(monkeypatch):
monkeypatch.setattr(app_module, 'get_cached', lambda key, ttl=None: None)
monkeypatch.setattr(app_module, 'set_cached', lambda key, data: None)
monkeypatch.setattr(app_module, 'fetch_eurostar_prices', lambda dest, date: {})
monkeypatch.setattr(
app_module.rtt_scraper,
'fetch',
lambda travel_date, user_agent: [
{'depart_bristol': '07:00', 'arrive_paddington': '08:30', 'headcode': '1A01'},
{'depart_bristol': '07:05', 'arrive_paddington': '08:36', 'headcode': '1A02'},
{'depart_bristol': '07:10', 'arrive_paddington': '08:46', 'headcode': '1A03'},
{'depart_bristol': '07:15', 'arrive_paddington': '08:56', 'headcode': '1A04'},
{'depart_bristol': '07:20', 'arrive_paddington': '09:06', 'headcode': '1A05'},
],
)
monkeypatch.setattr(
app_module.eurostar_scraper,
'fetch',
lambda destination, travel_date, user_agent: [
{
'depart_st_pancras': '09:30',
'arrive_destination': '11:50',
'destination': destination,
'train_number': 'ES 1001',
},
{
'depart_st_pancras': '09:40',
'arrive_destination': '12:00',
'destination': destination,
'train_number': 'ES 1002',
},
{
'depart_st_pancras': '09:50',
'arrive_destination': '12:20',
'destination': destination,
'train_number': 'ES 1003',
},
{
'depart_st_pancras': '10:00',
'arrive_destination': '12:35',
'destination': destination,
'train_number': 'ES 1004',
},
{
'depart_st_pancras': '10:10',
'arrive_destination': '12:45',
'destination': destination,
'train_number': 'ES 1005',
},
],
)
monkeypatch.setattr(
app_module.eurostar_scraper,
'timetable_url',
lambda destination: f'https://example.test/{destination.lower().replace(" ", "-")}',
)
client = _client()
resp = client.get('/results/paris/2026-04-10?min_connection=60&max_connection=120')
html = resp.get_data(as_text=True)
assert resp.status_code == 200
assert html.count('title="Fastest option"') == 2
assert html.count('title="Slowest option"') == 2
assert '4h 50m ⚡' in html
assert '4h 55m ⚡' in html
assert '5h 20m 🐢' in html
assert '5h 25m 🐢' in html
assert '5h 10m ⚡' not in html
assert '5h 10m 🐢' not in html
def test_results_shows_unreachable_morning_eurostar_services(monkeypatch):
monkeypatch.setattr(app_module, 'get_cached', lambda key, ttl=None: None)
monkeypatch.setattr(app_module, 'set_cached', lambda key, data: None)
monkeypatch.setattr(app_module, 'fetch_eurostar_prices', lambda dest, date: {})
monkeypatch.setattr(
app_module.rtt_scraper,
'fetch',
lambda travel_date, user_agent: [
{'depart_bristol': '07:00', 'arrive_paddington': '08:45', 'headcode': '1A23'},
],
)
monkeypatch.setattr(
app_module.eurostar_scraper,
'fetch',
lambda destination, travel_date, user_agent: [
{
'depart_st_pancras': '09:30',
'arrive_destination': '12:00',
'destination': destination,
'train_number': 'ES 9001',
},
{
'depart_st_pancras': '10:15',
'arrive_destination': '13:40',
'destination': destination,
'train_number': 'ES 9002',
},
{
'depart_st_pancras': '12:30',
'arrive_destination': '15:55',
'destination': destination,
'train_number': 'ES 9003',
},
],
)
monkeypatch.setattr(
app_module.eurostar_scraper,
'timetable_url',
lambda destination: f'https://example.test/{destination.lower().replace(" ", "-")}',
)
client = _client()
resp = client.get('/results/paris/2026-04-10?min_connection=60&max_connection=120')
html = resp.get_data(as_text=True)
assert resp.status_code == 200
assert '1 morning service unavailable from Bristol' in html
assert '09:30' in html
assert 'ES 9001' in html
assert 'Unavailable from Bristol' in html
assert 'Morning means Eurostar departures before 12:00 from St&nbsp;Pancras.' in html
assert html.index('09:30') < html.index('10:15')
def test_results_shows_eurostar_price_and_total(monkeypatch):
# 07:00 on Friday 2026-04-10 → Anytime £138.70 (weekday, 05:0508:25 window)
_stub_data(monkeypatch, prices={'10:01': 59})
client = _client()
resp = client.get('/results/paris/2026-04-10?min_connection=60&max_connection=120')
html = resp.get_data(as_text=True)
assert resp.status_code == 200
assert '£59' in html # Eurostar Standard price
assert '£197.70' in html # Anytime £138.70 + ES £59
def test_results_can_show_only_unreachable_morning_services(monkeypatch):
monkeypatch.setattr(app_module, 'get_cached', lambda key, ttl=None: None)
monkeypatch.setattr(app_module, 'set_cached', lambda key, data: None)
monkeypatch.setattr(app_module, 'fetch_eurostar_prices', lambda dest, date: {})
monkeypatch.setattr(
app_module.rtt_scraper,
'fetch',
lambda travel_date, user_agent: [
{'depart_bristol': '07:00', 'arrive_paddington': '08:45', 'headcode': '1A23'},
],
)
monkeypatch.setattr(
app_module.eurostar_scraper,
'fetch',
lambda destination, travel_date, user_agent: [
{
'depart_st_pancras': '09:30',
'arrive_destination': '12:00',
'destination': destination,
'train_number': 'ES 9001',
},
],
)
monkeypatch.setattr(
app_module.eurostar_scraper,
'timetable_url',
lambda destination: f'https://example.test/{destination.lower().replace(" ", "-")}',
)
client = _client()
resp = client.get('/results/paris/2026-04-10?min_connection=60&max_connection=120')
html = resp.get_data(as_text=True)
assert resp.status_code == 200
assert 'No valid journeys found.' not in html
assert '1 morning service unavailable from Bristol' in html
assert '09:30' in html
assert 'Unavailable from Bristol' in html

View file

@ -1,42 +0,0 @@
import os
import time
import pytest
from cache import get_cached, set_cached
@pytest.fixture
def tmp_cache(tmp_path, monkeypatch):
import cache as cache_module
monkeypatch.setattr(cache_module, 'CACHE_DIR', str(tmp_path))
return tmp_path
def test_get_cached_returns_none_for_missing_key(tmp_cache):
assert get_cached('no_such_key') is None
def test_set_and_get_cached_roundtrip(tmp_cache):
set_cached('my_key', {'a': 1})
assert get_cached('my_key') == {'a': 1}
def test_get_cached_no_ttl_never_expires(tmp_cache):
set_cached('k', [1, 2, 3])
# Backdate the file by 2 days
path = tmp_cache / 'k.json'
old = time.time() - 2 * 86400
os.utime(path, (old, old))
assert get_cached('k') == [1, 2, 3]
def test_get_cached_within_ttl(tmp_cache):
set_cached('k', 'fresh')
assert get_cached('k', ttl=3600) == 'fresh'
def test_get_cached_expired_returns_none(tmp_cache):
set_cached('k', 'stale')
path = tmp_cache / 'k.json'
old = time.time() - 25 * 3600 # 25 hours ago
os.utime(path, (old, old))
assert get_cached('k', ttl=24 * 3600) is None

View file

@ -1,5 +1,5 @@
import pytest import pytest
from trip_planner import combine_trips, find_unreachable_morning_eurostars, _fmt_duration, cheapest_gwr_ticket from trip_planner import combine_trips, _fmt_duration
DATE = '2026-03-30' DATE = '2026-03-30'
@ -129,88 +129,3 @@ def test_connection_duration_in_trip():
# arrive Paddington 08:45, depart St Pancras 10:01 → 1h 16m # arrive Paddington 08:45, depart St Pancras 10:01 → 1h 16m
trips = combine_trips([GWR_FAST], [ES_PARIS], DATE) trips = combine_trips([GWR_FAST], [ES_PARIS], DATE)
assert trips[0]['connection_duration'] == '1h 16m' assert trips[0]['connection_duration'] == '1h 16m'
def test_unreachable_morning_eurostars_lists_only_unreachable_morning_services():
gwr = [
{'depart_bristol': '07:00', 'arrive_paddington': '08:45'},
]
eurostar = [
{'depart_st_pancras': '09:30', 'arrive_destination': '12:00', 'destination': 'Paris Gare du Nord', 'train_number': 'ES 9001'},
{'depart_st_pancras': '10:15', 'arrive_destination': '13:40', 'destination': 'Paris Gare du Nord', 'train_number': 'ES 9002'},
{'depart_st_pancras': '12:30', 'arrive_destination': '15:55', 'destination': 'Paris Gare du Nord', 'train_number': 'ES 9003'},
]
unreachable = find_unreachable_morning_eurostars(gwr, eurostar, DATE)
assert [service['depart_st_pancras'] for service in unreachable] == ['09:30']
# ---------------------------------------------------------------------------
# cheapest_gwr_ticket — Bristol Temple Meads → Paddington
# ---------------------------------------------------------------------------
# 2026-03-30 is a Monday; 2026-03-28 is a Saturday
def test_cheapest_ticket_weekday_super_off_peak_morning():
# 05:00 on Monday: dep ≤ 05:04 → Super Off-Peak
t = cheapest_gwr_ticket('05:00', '2026-03-30')
assert t['ticket'] == 'Super Off-Peak'
assert t['price'] == 45.00
def test_cheapest_ticket_weekday_anytime_window():
# 07:00 on Monday: 05:0508:25 → Anytime only
t = cheapest_gwr_ticket('07:00', '2026-03-30')
assert t['ticket'] == 'Anytime'
assert t['price'] == 138.70
def test_cheapest_ticket_weekday_off_peak():
# 08:30 on Monday: dep ≥ 08:26 but < 09:58 → Off-Peak
t = cheapest_gwr_ticket('08:30', '2026-03-30')
assert t['ticket'] == 'Off-Peak'
assert t['price'] == 63.60
def test_cheapest_ticket_weekday_super_off_peak_late():
# 10:00 on Monday: dep ≥ 09:58 → Super Off-Peak
t = cheapest_gwr_ticket('10:00', '2026-03-30')
assert t['ticket'] == 'Super Off-Peak'
assert t['price'] == 45.00
def test_cheapest_ticket_boundary_super_off_peak_cutoff():
# 05:04 is last valid minute for early Super Off-Peak
assert cheapest_gwr_ticket('05:04', '2026-03-30')['ticket'] == 'Super Off-Peak'
# 05:05 falls into the Anytime window (off-peak starts at 08:26)
assert cheapest_gwr_ticket('05:05', '2026-03-30')['ticket'] == 'Anytime'
def test_cheapest_ticket_boundary_off_peak_start():
assert cheapest_gwr_ticket('08:25', '2026-03-30')['ticket'] == 'Anytime'
assert cheapest_gwr_ticket('08:26', '2026-03-30')['ticket'] == 'Off-Peak'
def test_cheapest_ticket_boundary_super_off_peak_resumes():
assert cheapest_gwr_ticket('09:57', '2026-03-30')['ticket'] == 'Off-Peak'
assert cheapest_gwr_ticket('09:58', '2026-03-30')['ticket'] == 'Super Off-Peak'
def test_cheapest_ticket_weekend_always_super_off_peak():
# Saturday — no restrictions
t = cheapest_gwr_ticket('07:00', '2026-03-28')
assert t['ticket'] == 'Super Off-Peak'
assert t['price'] == 45.00
def test_combine_trips_includes_ticket_fields():
trips = combine_trips([GWR_FAST], [ES_PARIS], DATE)
assert len(trips) == 1
t = trips[0]
assert 'ticket_name' in t
assert 'ticket_price' in t
assert 'ticket_code' in t
def test_unreachable_morning_eurostars_returns_empty_when_morning_service_is_reachable():
gwr = [
{'depart_bristol': '07:00', 'arrive_paddington': '08:45'},
]
eurostar = [
{'depart_st_pancras': '10:15', 'arrive_destination': '13:40', 'destination': 'Paris Gare du Nord', 'train_number': 'ES 9002'},
]
assert find_unreachable_morning_eurostars(gwr, eurostar, DATE) == []

View file

@ -1,47 +1,15 @@
""" """
Combine GWR BristolPaddington trains with Eurostar St Pancrasdestination trains. Combine GWR BristolPaddington trains with Eurostar St Pancrasdestination trains.
""" """
from datetime import datetime, timedelta, time as _time from datetime import datetime, timedelta
MIN_CONNECTION_MINUTES = 50 MIN_CONNECTION_MINUTES = 50
MAX_CONNECTION_MINUTES = 110 MAX_CONNECTION_MINUTES = 110
MAX_GWR_MINUTES = 110 MAX_GWR_MINUTES = 110
MORNING_CUTOFF_HOUR = 12
DATE_FMT = '%Y-%m-%d' DATE_FMT = '%Y-%m-%d'
TIME_FMT = '%H:%M' TIME_FMT = '%H:%M'
# Bristol Temple Meads → London Paddington walk-on single fares.
# Weekday restrictions (MonFri only):
# Super Off-Peak (£45.00, SSS): not valid if 05:05 ≤ dep ≤ 09:57
# Off-Peak (£63.60, SVS): not valid if dep < 08:26, except 02:0005:10
# Anytime (£138.70, SDS): no restrictions
_TICKET_SUPER_OFF_PEAK = {'ticket': 'Super Off-Peak', 'price': 45.00, 'code': 'SSS'}
_TICKET_OFF_PEAK = {'ticket': 'Off-Peak', 'price': 63.60, 'code': 'SVS'}
_TICKET_ANYTIME = {'ticket': 'Anytime', 'price': 138.70, 'code': 'SDS'}
def cheapest_gwr_ticket(depart_time: str, travel_date: str) -> dict:
"""
Return the cheapest walk-on single for Bristol Temple Meads Paddington.
Weekday (MonFri) restrictions derived from the SSS/SVS ticket conditions:
Super Off-Peak valid: dep 05:04 or dep 09:58
Off-Peak valid: 02:00 dep 05:10 or dep 08:26
Weekends: no restrictions always Super Off-Peak.
"""
dt = datetime.strptime(f"{travel_date} {depart_time}", f"{DATE_FMT} {TIME_FMT}")
if dt.weekday() >= 5: # Saturday or Sunday
return _TICKET_SUPER_OFF_PEAK
dep = dt.time()
if dep <= _time(5, 4) or dep >= _time(9, 58):
return _TICKET_SUPER_OFF_PEAK
if dep >= _time(8, 26):
return _TICKET_OFF_PEAK
return _TICKET_ANYTIME
def _parse_dt(date: str, time: str) -> datetime: def _parse_dt(date: str, time: str) -> datetime:
return datetime.strptime(f"{date} {time}", f"{DATE_FMT} {TIME_FMT}") return datetime.strptime(f"{date} {time}", f"{DATE_FMT} {TIME_FMT}")
@ -55,36 +23,6 @@ def _fmt_duration(minutes: int) -> str:
return f"{m}m" return f"{m}m"
def _is_viable_connection(
gwr: dict,
eurostar: dict,
travel_date: str,
min_connection_minutes: int,
max_connection_minutes: int,
) -> tuple[datetime, datetime, datetime, datetime] | None:
try:
arr_pad = _parse_dt(travel_date, gwr['arrive_paddington'])
dep_bri = _parse_dt(travel_date, gwr['depart_bristol'])
dep_stp = _parse_dt(travel_date, eurostar['depart_st_pancras'])
arr_dest = _parse_dt(travel_date, eurostar['arrive_destination'])
except (ValueError, KeyError):
return None
if int((arr_pad - dep_bri).total_seconds() / 60) > MAX_GWR_MINUTES:
return None
if arr_dest < dep_stp:
arr_dest += timedelta(days=1)
connection_minutes = (dep_stp - arr_pad).total_seconds() / 60
if connection_minutes < min_connection_minutes:
return None
if connection_minutes > max_connection_minutes:
return None
return dep_bri, arr_pad, dep_stp, arr_dest
def combine_trips( def combine_trips(
gwr_trains: list[dict], gwr_trains: list[dict],
eurostar_trains: list[dict], eurostar_trains: list[dict],
@ -108,21 +46,35 @@ def combine_trips(
trips = [] trips = []
for gwr in gwr_trains: for gwr in gwr_trains:
try:
arr_pad = _parse_dt(travel_date, gwr['arrive_paddington'])
dep_bri = _parse_dt(travel_date, gwr['depart_bristol'])
except (ValueError, KeyError):
continue
if int((arr_pad - dep_bri).total_seconds() / 60) > MAX_GWR_MINUTES:
continue
earliest_eurostar = arr_pad + timedelta(minutes=min_connection_minutes)
# Find only the earliest viable Eurostar for this GWR departure # Find only the earliest viable Eurostar for this GWR departure
for es in eurostar_trains: for es in eurostar_trains:
connection = _is_viable_connection( try:
gwr, dep_stp = _parse_dt(travel_date, es['depart_st_pancras'])
es, arr_dest = _parse_dt(travel_date, es['arrive_destination'])
travel_date, except (ValueError, KeyError):
min_connection_minutes, continue
max_connection_minutes,
) # Eurostar arrives next day? (e.g. night service — unlikely but handle it)
if not connection: if arr_dest < dep_stp:
arr_dest += timedelta(days=1)
if dep_stp < earliest_eurostar:
continue
if (dep_stp - arr_pad).total_seconds() / 60 > max_connection_minutes:
continue continue
dep_bri, arr_pad, dep_stp, arr_dest = connection
total_mins = int((arr_dest - dep_bri).total_seconds() / 60) total_mins = int((arr_dest - dep_bri).total_seconds() / 60)
ticket = cheapest_gwr_ticket(gwr['depart_bristol'], travel_date)
trips.append({ trips.append({
'depart_bristol': gwr['depart_bristol'], 'depart_bristol': gwr['depart_bristol'],
'arrive_paddington': gwr['arrive_paddington'], 'arrive_paddington': gwr['arrive_paddington'],
@ -135,46 +87,8 @@ def combine_trips(
'total_duration': _fmt_duration(total_mins), 'total_duration': _fmt_duration(total_mins),
'total_minutes': total_mins, 'total_minutes': total_mins,
'destination': es['destination'], 'destination': es['destination'],
'ticket_name': ticket['ticket'],
'ticket_price': ticket['price'],
'ticket_code': ticket['code'],
}) })
break # Only the earliest valid Eurostar per GWR departure break # Only the earliest valid Eurostar per GWR departure
trips.sort(key=lambda t: (t['depart_bristol'], t['depart_st_pancras'])) trips.sort(key=lambda t: (t['depart_bristol'], t['depart_st_pancras']))
return trips return trips
def find_unreachable_morning_eurostars(
gwr_trains: list[dict],
eurostar_trains: list[dict],
travel_date: str,
min_connection_minutes: int = MIN_CONNECTION_MINUTES,
max_connection_minutes: int = MAX_CONNECTION_MINUTES,
) -> list[dict]:
unreachable = []
for es in eurostar_trains:
try:
dep_stp = _parse_dt(travel_date, es['depart_st_pancras'])
except (ValueError, KeyError):
continue
if dep_stp.hour >= MORNING_CUTOFF_HOUR:
continue
if any(
_is_viable_connection(
gwr,
es,
travel_date,
min_connection_minutes,
max_connection_minutes,
)
for gwr in gwr_trains
):
continue
unreachable.append(es)
return sorted(unreachable, key=lambda s: s['depart_st_pancras'])