brittany-ferries/main.py

538 lines
14 KiB
Python
Raw Normal View History

2022-09-03 21:38:46 +01:00
#!/usr/bin/python3
"""Check prices of ferries to France."""
2023-09-11 07:08:45 +01:00
import collections
2022-09-03 21:38:46 +01:00
import inspect
2023-01-08 11:49:57 +00:00
import json
import os
import os.path
import re
2023-04-16 19:56:26 +01:00
import sys
import traceback
2023-01-08 11:49:57 +00:00
from datetime import date, datetime, timedelta
2022-09-03 21:38:46 +01:00
from typing import Any
2023-01-08 12:22:36 +00:00
import dateutil.parser
2022-09-03 21:38:46 +01:00
import flask
2023-01-08 11:49:57 +00:00
import pytz
2023-04-16 19:56:26 +01:00
import werkzeug
from werkzeug.debug.tbtools import DebugTraceback
2023-01-08 11:49:57 +00:00
from werkzeug.wrappers import Response
2022-09-03 21:38:46 +01:00
2023-01-08 12:22:36 +00:00
import ferry
2023-01-08 11:49:57 +00:00
from ferry.api import get_accommodations, get_prices
from ferry.read_config import ferry_config, vehicle_from_config
2022-09-03 21:38:46 +01:00
app = flask.Flask(__name__)
2023-04-16 19:56:26 +01:00
app.debug = True
routes = {
"outbound": [
("PORTSMOUTH", "CAEN"),
("PORTSMOUTH", "CHERBOURG"),
("PORTSMOUTH", "ST MALO"),
("PORTSMOUTH", "LE HAVRE"),
("POOLE", "CHERBOURG"),
2023-09-11 07:00:20 +01:00
("PLYMOUTH", "ROSCOFF"),
2023-04-16 19:56:26 +01:00
],
"return": [
("CAEN", "PORTSMOUTH"),
("CHERBOURG", "PORTSMOUTH"),
("ST MALO", "PORTSMOUTH"),
("LE HAVRE", "PORTSMOUTH"),
("CHERBOURG", "POOLE"),
2023-09-11 07:00:20 +01:00
("ROSCOFF", "PLYMOUTH"),
2023-04-16 19:56:26 +01:00
],
}
2023-01-08 11:49:57 +00:00
def cache_location() -> str:
2023-01-08 13:03:23 +00:00
"""Cache location for the current user."""
2023-01-08 11:49:57 +00:00
return os.path.expanduser(ferry_config.get("cache", "location"))
def cache_filename(params: str) -> str:
"""Get a filename to use for caching."""
now_str = datetime.utcnow().strftime("%Y-%m-%d_%H%M")
params = params.replace(":", "_").replace(".", "_")
2022-09-03 21:38:46 +01:00
2023-01-08 11:49:57 +00:00
return os.path.join(cache_location(), now_str + "_" + params + ".json")
2022-09-03 21:38:46 +01:00
2023-01-08 11:49:57 +00:00
def time_to_minutes(t: str) -> int:
"""Convert time (22:50) into minutes since midnight."""
m = re.match(r"^(\d\d):(\d\d)$", t)
assert m
hours, minutes = [int(j) for j in m.groups()]
2022-09-03 21:38:46 +01:00
2023-01-08 11:49:57 +00:00
return (60 * hours) + minutes
2022-09-03 21:38:46 +01:00
2023-01-08 11:49:57 +00:00
def get_duration(depart: str, arrive: str, time_delta: int) -> str:
"""Given two times calculate the duration and return as string."""
depart_min = time_to_minutes(depart)
arrive_min = time_to_minutes(arrive) + time_delta
duration = arrive_min - depart_min
if depart_min > arrive_min:
duration += 60 * 24
return f"{duration // 60}h{ duration % 60:02d}m"
2022-09-03 21:38:46 +01:00
@app.errorhandler(werkzeug.exceptions.InternalServerError)
2023-09-11 07:08:45 +01:00
def exception_handler(e: werkzeug.exceptions.InternalServerError) -> tuple[str, int]:
"""Handle exception."""
2023-04-16 19:56:26 +01:00
exec_type, exc_value, current_traceback = sys.exc_info()
assert exc_value
tb = DebugTraceback(exc_value)
summary = tb.render_traceback_html(include_title=False)
2023-09-11 07:08:45 +01:00
exc_lines = "".join(tb._te.format_exception_only())
2023-04-16 19:56:26 +01:00
2023-09-11 07:08:45 +01:00
last_frame = list(traceback.walk_tb(current_traceback))[-1][0]
2023-04-16 19:56:26 +01:00
last_frame_args = inspect.getargs(last_frame.f_code)
2022-09-03 21:38:46 +01:00
return (
flask.render_template(
"show_error.html",
2023-04-16 19:56:26 +01:00
plaintext=tb.render_traceback_text(),
2023-09-11 07:08:45 +01:00
exception=exc_lines,
2023-04-16 19:56:26 +01:00
exception_type=tb._te.exc_type.__name__,
summary=summary,
2022-09-03 21:38:46 +01:00
last_frame=last_frame,
last_frame_args=last_frame_args,
),
500,
)
def parse_date(d: str) -> date:
2023-01-08 11:49:57 +00:00
"""Parse a date from a string in ISO format."""
2022-09-03 21:38:46 +01:00
return datetime.strptime(d, "%Y-%m-%d").date()
@app.route("/")
2023-01-08 12:22:36 +00:00
def start() -> Response | str:
2022-09-03 21:38:46 +01:00
"""Start page."""
2023-01-08 11:49:57 +00:00
return flask.render_template("index.html")
2022-09-03 21:38:46 +01:00
2023-09-11 07:00:20 +01:00
def cabins_url(
dep: str,
arr: str,
crossing: dict[str, Any],
ticket_tier: str,
rear_mounted_bike_carrier: bool,
) -> str:
2023-01-08 13:03:23 +00:00
"""Generate a URL for the cabins on a given crossing."""
2022-09-03 21:38:46 +01:00
dt = datetime.fromisoformat(crossing["departureDateTime"]["iso"])
utc_dt = dt.astimezone(pytz.utc)
2023-08-10 14:05:07 +01:00
adults_str = flask.request.args.get("adults")
small_dogs_str = flask.request.args.get("small_dogs")
2022-09-03 21:38:46 +01:00
return flask.url_for(
"cabins",
2023-04-16 19:56:26 +01:00
sailing_id=int(crossing["sailingId"]),
2023-01-08 12:22:36 +00:00
departure_port=ferry.ports[dep],
arrival_port=ferry.ports[arr],
2022-09-03 21:38:46 +01:00
departure_date=utc_dt.strftime("%Y-%m-%dT%H:%M:%S.000Z"),
ticket_tier=ticket_tier,
2023-08-10 14:05:07 +01:00
adults=adults_str,
small_dogs=small_dogs_str,
2023-09-11 07:00:20 +01:00
rear_mounted_bike_carrier=("true" if rear_mounted_bike_carrier else None),
2022-09-03 21:38:46 +01:00
)
2023-01-08 11:49:57 +00:00
def get_days_until_start() -> int:
"""How long until the travel date."""
2022-09-03 21:38:46 +01:00
start = date.fromisoformat(ferry_config.get("dates", "start"))
return (start - date.today()).days
2023-04-16 19:56:26 +01:00
PriceData = list[tuple[str, str, list[dict[str, Any]]]]
2023-01-08 13:03:23 +00:00
def check_cache_for_prices(params: str) -> PriceData | None:
"""Look for prices in cache."""
existing_files = os.listdir(cache_location())
existing = [f for f in existing_files if f.endswith(params + ".json")]
if not existing:
return None
recent_filename = max(existing)
recent = datetime.strptime(recent_filename, f"%Y-%m-%d_%H%M_{params}.json")
now = datetime.utcnow()
delta = now - recent
if delta < timedelta(hours=1):
full = os.path.join(cache_location(), recent_filename)
data: PriceData = json.load(open(full))
return data
return None
2023-01-08 11:49:57 +00:00
def get_prices_with_cache(
2023-04-16 19:56:26 +01:00
direction: str,
2023-01-08 11:49:57 +00:00
start: str,
end: str,
selection: list[tuple[str, str]],
2023-08-10 14:05:07 +01:00
adults: int,
small_dogs: int,
2023-01-08 11:49:57 +00:00
refresh: bool = False,
2023-09-11 07:00:20 +01:00
rear_mounted_bike_carrier: bool = False,
2023-01-08 13:03:23 +00:00
) -> PriceData:
"""Get price data using cache."""
2023-09-11 07:00:20 +01:00
params = (
f"{direction}_{start}_{end}_{adults}_{small_dogs}_{rear_mounted_bike_carrier}"
)
2023-01-08 13:03:23 +00:00
if not refresh:
data = check_cache_for_prices(params)
if data:
2023-01-08 11:49:57 +00:00
return data
vehicle = vehicle_from_config(ferry_config)
2022-09-03 21:38:46 +01:00
all_data = [
2023-01-08 12:22:36 +00:00
(
dep,
arr,
2023-04-16 19:56:26 +01:00
get_prices(
ferry.ports[dep],
ferry.ports[arr],
start,
end,
vehicle,
adults,
small_dogs,
2023-09-11 07:00:20 +01:00
rear_mounted_bike_carrier,
2023-04-16 19:56:26 +01:00
)["crossings"],
2023-01-08 12:22:36 +00:00
)
2022-09-03 21:38:46 +01:00
for dep, arr in selection
]
2023-01-08 11:49:57 +00:00
2023-01-08 13:03:23 +00:00
with open(cache_filename(params), "w") as out:
2023-01-08 11:49:57 +00:00
json.dump(all_data, out, indent=2)
return all_data
2023-09-11 07:00:20 +01:00
Pax = collections.namedtuple("Pax", ["adults", "small_dogs"])
def read_pax() -> Pax:
2023-08-10 14:05:07 +01:00
"""Get the number of adults and dogs that are travelling."""
config_adults = int(ferry_config.get("pax", "adults"))
config_dogs = int(ferry_config.get("pax", "dogs"))
adults_str = flask.request.args.get("adults")
adults = int(adults_str) if adults_str else config_adults
small_dogs_str = flask.request.args.get("small_dogs")
small_dogs = int(small_dogs_str) if small_dogs_str else config_dogs
2023-09-11 07:00:20 +01:00
return Pax(adults=adults, small_dogs=small_dogs)
pax_options = [
{"pax": Pax(adults=1, small_dogs=0), "label": "1 adult"},
{"pax": Pax(adults=2, small_dogs=0), "label": "2 adults"},
{"pax": Pax(adults=2, small_dogs=1), "label": "2 adults and a dog"},
]
2023-08-10 14:05:07 +01:00
2023-01-08 11:49:57 +00:00
def build_outbound(section: str) -> str:
"""Show all routes on one page."""
start = ferry_config.get(section, "from")
end = ferry_config.get(section, "to")
refresh = bool(flask.request.args.get("refresh"))
2023-09-11 07:00:20 +01:00
rear_mounted_bike_carrier = (
flask.request.args.get("rear_mounted_bike_carrier") == "true"
)
2023-01-08 11:49:57 +00:00
2023-08-10 14:05:07 +01:00
pax = read_pax()
2023-04-16 19:56:26 +01:00
direction = section[:-1] if section[-1].isdigit() else section
all_data = get_prices_with_cache(
2023-08-10 14:05:07 +01:00
direction,
start,
end,
routes["outbound"],
2023-09-11 07:00:20 +01:00
pax.adults,
pax.small_dogs,
2023-08-10 14:05:07 +01:00
refresh,
2023-09-11 07:00:20 +01:00
rear_mounted_bike_carrier,
2023-04-16 19:56:26 +01:00
)
2023-01-08 11:49:57 +00:00
2022-09-03 21:38:46 +01:00
return flask.render_template(
"all_routes.html",
data=all_data,
days_until_start=get_days_until_start(),
2023-01-08 12:22:36 +00:00
ports=ferry.ports,
2022-09-03 21:38:46 +01:00
parse_date=parse_date,
2023-01-08 11:49:57 +00:00
from_date=start,
to_date=end,
2022-09-03 21:38:46 +01:00
cabins_url=cabins_url,
2023-01-08 11:49:57 +00:00
get_duration=get_duration,
time_delta=-60,
format_pet_options=format_pet_options,
2023-09-11 07:00:20 +01:00
pax=pax,
pax_options=pax_options,
rear_mounted_bike_carrier=rear_mounted_bike_carrier,
2022-09-03 21:38:46 +01:00
)
2023-04-16 19:56:26 +01:00
def build_return(section: str) -> str:
"""Show all routes on one page."""
start = ferry_config.get(section, "from")
end = ferry_config.get(section, "to")
refresh = bool(flask.request.args.get("refresh"))
2023-08-10 14:05:07 +01:00
pax = read_pax()
2023-09-11 07:00:20 +01:00
rear_mounted_bike_carrier = (
flask.request.args.get("rear_mounted_bike_carrier") == "true"
)
2023-04-16 19:56:26 +01:00
direction = section[:-1] if section[-1].isdigit() else section
all_data = get_prices_with_cache(
2023-08-10 14:05:07 +01:00
direction,
start,
end,
routes["return"],
2023-09-11 07:00:20 +01:00
pax.adults,
pax.small_dogs,
2023-08-10 14:05:07 +01:00
refresh,
2023-09-11 07:00:20 +01:00
rear_mounted_bike_carrier,
2023-04-16 19:56:26 +01:00
)
return flask.render_template(
"all_routes.html",
data=all_data,
ports=ferry.ports,
days_until_start=get_days_until_start(),
parse_date=parse_date,
from_date=start,
to_date=end,
cabins_url=cabins_url,
get_duration=get_duration,
time_delta=60,
format_pet_options=format_pet_options,
2023-09-11 07:00:20 +01:00
pax=pax,
pax_options=pax_options,
rear_mounted_bike_carrier=rear_mounted_bike_carrier,
2023-04-16 19:56:26 +01:00
)
2023-01-08 11:49:57 +00:00
@app.route("/outbound1")
def outbound1_page() -> str:
return build_outbound("outbound1")
@app.route("/outbound2")
def outbound2_page() -> str:
return build_outbound("outbound2")
2023-04-16 19:56:26 +01:00
@app.route("/outbound3")
def outbound3_page() -> str:
return build_outbound("outbound3")
@app.route("/return1")
def return1_page() -> str:
return build_return("return1")
@app.route("/return2")
def return2_page() -> str:
return build_return("return2")
2023-01-08 11:49:57 +00:00
def format_pet_options(o: dict[str, bool]) -> list[str]:
2023-04-16 19:56:26 +01:00
"""Read pet options from API and format for display."""
2023-01-08 11:49:57 +00:00
ret = []
if o.get("petCabinAvailable"):
ret.append("pet cabins")
if o.get("smallKennelAvailable"):
ret.append("small kennel")
if o.get("stayInCarAvailable"):
ret.append("pets stay in car")
return ret
def get_accommodations_with_cache(
2023-09-11 07:00:20 +01:00
dep: str,
arr: str,
d: str,
ticket_tier: str,
refresh: bool = False,
rear_mounted_bike_carrier: bool = False,
2023-01-08 11:49:57 +00:00
) -> dict[str, list[dict[str, Any]]]:
2023-08-10 14:05:07 +01:00
pax = read_pax()
2023-09-11 07:00:20 +01:00
params = f"{dep}_{arr}_{d}_{ticket_tier}_{pax.adults}_{pax.small_dogs}_{rear_mounted_bike_carrier}"
2023-01-08 11:49:57 +00:00
existing_files = os.listdir(cache_location())
existing = [f for f in existing_files if f.endswith(params + ".json")]
if not refresh and existing:
recent_filename = max(existing)
recent = datetime.strptime(recent_filename, f"%Y-%m-%d_%H%M_{params}.json")
now = datetime.utcnow()
delta = now - recent
if delta < timedelta(hours=1):
full = os.path.join(cache_location(), recent_filename)
data = json.load(open(full))
return data
vehicle = vehicle_from_config(ferry_config)
filename = cache_filename(params)
2023-08-10 14:05:07 +01:00
data = get_accommodations(
2023-09-11 07:00:20 +01:00
dep,
arr,
d,
ticket_tier,
vehicle,
pax.adults,
pax.small_dogs,
rear_mounted_bike_carrier,
2023-08-10 14:05:07 +01:00
)
2023-01-08 11:49:57 +00:00
with open(filename, "w") as out:
print(filename)
json.dump(data, out, indent=2)
return data
2023-04-16 19:56:26 +01:00
def get_start_end(section: str) -> tuple[str, str]:
start = ferry_config.get(section, "from")
end = ferry_config.get(section, "to")
return start, end
def get_return_section(departure_date: str) -> tuple[str, str, str]:
section = "return1"
start, end = get_start_end(section)
dep = departure_date[:10]
if start <= dep <= end:
return section, start, end
section = "return2"
start, end = get_start_end(section)
if start <= dep <= end:
return section, start, end
def get_outbound_section(departure_date: str) -> tuple[str, str, str]:
section = "outbound1"
start, end = get_start_end(section)
dep = departure_date[:10]
if start <= dep <= end:
return section, start, end
section = "outbound2"
start, end = get_start_end(section)
if start <= dep <= end:
return section, start, end
section = "outbound3"
start, end = get_start_end(section)
if start <= dep <= end:
return section, start, end
def lookup_sailing_id(prices: PriceData, sailing_id: int) -> dict[str, Any] | None:
for dep, arr, price_list in prices:
for i in price_list:
for j in i["prices"]:
crossing: dict[str, Any] = j["crossingPrices"]
if int(crossing["sailingId"]) == sailing_id:
return crossing
return None
@app.route(
"/cabins/<int:sailing_id>/<departure_port>/<arrival_port>/<departure_date>/<ticket_tier>"
)
2023-01-08 11:49:57 +00:00
def cabins(
2023-04-16 19:56:26 +01:00
sailing_id: int,
departure_port: str,
arrival_port: str,
departure_date: str,
ticket_tier: str,
2023-01-08 11:49:57 +00:00
) -> str:
2023-04-16 19:56:26 +01:00
"""Get details of cabins."""
direction = {"GB": "outbound", "FR": "return"}[departure_port[:2]]
if direction == "return":
section, start, end = get_return_section(departure_date[:10])
start, end = get_start_end(section)
time_delta = 60
else:
section, start, end = get_outbound_section(departure_date[:10])
time_delta = -60
2023-08-10 14:05:07 +01:00
pax = read_pax()
2023-09-11 07:00:20 +01:00
rear_mounted_bike_carrier = (
flask.request.args.get("rear_mounted_bike_carrier") == "true"
)
2023-08-10 14:05:07 +01:00
prices = get_prices_with_cache(
2023-09-11 07:00:20 +01:00
direction,
start,
end,
routes[direction],
pax.adults,
pax.small_dogs,
False,
rear_mounted_bike_carrier,
2023-08-10 14:05:07 +01:00
)
2023-04-16 19:56:26 +01:00
crossing = lookup_sailing_id(prices, sailing_id)
2023-01-08 11:49:57 +00:00
cabin_data = get_accommodations_with_cache(
2023-09-11 07:00:20 +01:00
departure_port,
arrival_port,
departure_date,
ticket_tier,
False,
rear_mounted_bike_carrier,
2023-01-08 11:49:57 +00:00
)
accommodations = [
a
for a in cabin_data["accommodations"]
if a["quantityAvailable"] > 0 and a["code"] != "RS"
# and "Inside" not in a["description"]
]
2023-01-08 12:22:36 +00:00
2023-09-11 07:00:20 +01:00
pax_labels = {i["pax"]: i["label"] for i in pax_options}
2023-01-08 12:22:36 +00:00
dep = dateutil.parser.isoparse(departure_date)
2022-09-03 21:38:46 +01:00
return flask.render_template(
"cabins.html",
2023-01-08 12:22:36 +00:00
port_lookup=ferry.port_lookup,
2022-09-03 21:38:46 +01:00
departure_port=departure_port,
arrival_port=arrival_port,
2023-01-08 12:22:36 +00:00
departure_date=dep,
2022-09-03 21:38:46 +01:00
ticket_tier=ticket_tier,
2023-01-08 11:49:57 +00:00
accommodations=accommodations,
pet_accommodations=cabin_data["petAccommodations"],
2023-04-16 19:56:26 +01:00
crossing=crossing,
get_duration=get_duration,
time_delta=time_delta,
format_pet_options=format_pet_options,
section=section,
2023-09-11 07:00:20 +01:00
pax=pax,
pax_label=pax_labels[pax],
rear_mounted_bike_carrier=rear_mounted_bike_carrier,
2022-09-03 21:38:46 +01:00
)
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5001)