538 lines
14 KiB
Python
Executable file
538 lines
14 KiB
Python
Executable file
#!/usr/bin/python3
|
|
"""Check prices of ferries to France."""
|
|
|
|
import collections
|
|
import inspect
|
|
import json
|
|
import os
|
|
import os.path
|
|
import re
|
|
import sys
|
|
import traceback
|
|
from datetime import date, datetime, timedelta
|
|
from typing import Any
|
|
|
|
import dateutil.parser
|
|
import flask
|
|
import pytz
|
|
import werkzeug
|
|
from werkzeug.debug.tbtools import DebugTraceback
|
|
from werkzeug.wrappers import Response
|
|
|
|
import ferry
|
|
from ferry.api import get_accommodations, get_prices
|
|
from ferry.read_config import ferry_config, vehicle_from_config
|
|
|
|
app = flask.Flask(__name__)
|
|
app.debug = True
|
|
|
|
routes = {
|
|
"outbound": [
|
|
("PORTSMOUTH", "CAEN"),
|
|
("PORTSMOUTH", "CHERBOURG"),
|
|
("PORTSMOUTH", "ST MALO"),
|
|
("PORTSMOUTH", "LE HAVRE"),
|
|
("POOLE", "CHERBOURG"),
|
|
("PLYMOUTH", "ROSCOFF"),
|
|
],
|
|
"return": [
|
|
("CAEN", "PORTSMOUTH"),
|
|
("CHERBOURG", "PORTSMOUTH"),
|
|
("ST MALO", "PORTSMOUTH"),
|
|
("LE HAVRE", "PORTSMOUTH"),
|
|
("CHERBOURG", "POOLE"),
|
|
("ROSCOFF", "PLYMOUTH"),
|
|
],
|
|
}
|
|
|
|
|
|
def cache_location() -> str:
|
|
"""Cache location for the current user."""
|
|
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(".", "_")
|
|
|
|
return os.path.join(cache_location(), now_str + "_" + params + ".json")
|
|
|
|
|
|
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()]
|
|
|
|
return (60 * hours) + minutes
|
|
|
|
|
|
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"
|
|
|
|
|
|
@app.errorhandler(werkzeug.exceptions.InternalServerError)
|
|
def exception_handler(e: werkzeug.exceptions.InternalServerError) -> tuple[str, int]:
|
|
"""Handle exception."""
|
|
exec_type, exc_value, current_traceback = sys.exc_info()
|
|
assert exc_value
|
|
tb = DebugTraceback(exc_value)
|
|
|
|
summary = tb.render_traceback_html(include_title=False)
|
|
exc_lines = "".join(tb._te.format_exception_only())
|
|
|
|
last_frame = list(traceback.walk_tb(current_traceback))[-1][0]
|
|
last_frame_args = inspect.getargs(last_frame.f_code)
|
|
|
|
return (
|
|
flask.render_template(
|
|
"show_error.html",
|
|
plaintext=tb.render_traceback_text(),
|
|
exception=exc_lines,
|
|
exception_type=tb._te.exc_type.__name__,
|
|
summary=summary,
|
|
last_frame=last_frame,
|
|
last_frame_args=last_frame_args,
|
|
),
|
|
500,
|
|
)
|
|
|
|
|
|
def parse_date(d: str) -> date:
|
|
"""Parse a date from a string in ISO format."""
|
|
return datetime.strptime(d, "%Y-%m-%d").date()
|
|
|
|
|
|
@app.route("/")
|
|
def start() -> Response | str:
|
|
"""Start page."""
|
|
return flask.render_template("index.html")
|
|
|
|
|
|
def cabins_url(
|
|
dep: str,
|
|
arr: str,
|
|
crossing: dict[str, Any],
|
|
ticket_tier: str,
|
|
rear_mounted_bike_carrier: bool,
|
|
) -> str:
|
|
"""Generate a URL for the cabins on a given crossing."""
|
|
dt = datetime.fromisoformat(crossing["departureDateTime"]["iso"])
|
|
utc_dt = dt.astimezone(pytz.utc)
|
|
adults_str = flask.request.args.get("adults")
|
|
small_dogs_str = flask.request.args.get("small_dogs")
|
|
|
|
return flask.url_for(
|
|
"cabins",
|
|
sailing_id=int(crossing["sailingId"]),
|
|
departure_port=ferry.ports[dep],
|
|
arrival_port=ferry.ports[arr],
|
|
departure_date=utc_dt.strftime("%Y-%m-%dT%H:%M:%S.000Z"),
|
|
ticket_tier=ticket_tier,
|
|
adults=adults_str,
|
|
small_dogs=small_dogs_str,
|
|
rear_mounted_bike_carrier=("true" if rear_mounted_bike_carrier else None),
|
|
)
|
|
|
|
|
|
def get_days_until_start() -> int:
|
|
"""How long until the travel date."""
|
|
start = date.fromisoformat(ferry_config.get("dates", "start"))
|
|
return (start - date.today()).days
|
|
|
|
|
|
PriceData = list[tuple[str, str, list[dict[str, Any]]]]
|
|
|
|
|
|
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
|
|
|
|
|
|
def get_prices_with_cache(
|
|
direction: str,
|
|
start: str,
|
|
end: str,
|
|
selection: list[tuple[str, str]],
|
|
adults: int,
|
|
small_dogs: int,
|
|
refresh: bool = False,
|
|
rear_mounted_bike_carrier: bool = False,
|
|
) -> PriceData:
|
|
"""Get price data using cache."""
|
|
params = (
|
|
f"{direction}_{start}_{end}_{adults}_{small_dogs}_{rear_mounted_bike_carrier}"
|
|
)
|
|
if not refresh:
|
|
data = check_cache_for_prices(params)
|
|
if data:
|
|
return data
|
|
|
|
vehicle = vehicle_from_config(ferry_config)
|
|
|
|
all_data = [
|
|
(
|
|
dep,
|
|
arr,
|
|
get_prices(
|
|
ferry.ports[dep],
|
|
ferry.ports[arr],
|
|
start,
|
|
end,
|
|
vehicle,
|
|
adults,
|
|
small_dogs,
|
|
rear_mounted_bike_carrier,
|
|
)["crossings"],
|
|
)
|
|
for dep, arr in selection
|
|
]
|
|
|
|
with open(cache_filename(params), "w") as out:
|
|
json.dump(all_data, out, indent=2)
|
|
|
|
return all_data
|
|
|
|
|
|
Pax = collections.namedtuple("Pax", ["adults", "small_dogs"])
|
|
|
|
|
|
def read_pax() -> Pax:
|
|
"""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
|
|
|
|
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"},
|
|
]
|
|
|
|
|
|
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"))
|
|
rear_mounted_bike_carrier = (
|
|
flask.request.args.get("rear_mounted_bike_carrier") == "true"
|
|
)
|
|
|
|
pax = read_pax()
|
|
|
|
direction = section[:-1] if section[-1].isdigit() else section
|
|
|
|
all_data = get_prices_with_cache(
|
|
direction,
|
|
start,
|
|
end,
|
|
routes["outbound"],
|
|
pax.adults,
|
|
pax.small_dogs,
|
|
refresh,
|
|
rear_mounted_bike_carrier,
|
|
)
|
|
|
|
return flask.render_template(
|
|
"all_routes.html",
|
|
data=all_data,
|
|
days_until_start=get_days_until_start(),
|
|
ports=ferry.ports,
|
|
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,
|
|
pax=pax,
|
|
pax_options=pax_options,
|
|
rear_mounted_bike_carrier=rear_mounted_bike_carrier,
|
|
)
|
|
|
|
|
|
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"))
|
|
pax = read_pax()
|
|
rear_mounted_bike_carrier = (
|
|
flask.request.args.get("rear_mounted_bike_carrier") == "true"
|
|
)
|
|
|
|
direction = section[:-1] if section[-1].isdigit() else section
|
|
|
|
all_data = get_prices_with_cache(
|
|
direction,
|
|
start,
|
|
end,
|
|
routes["return"],
|
|
pax.adults,
|
|
pax.small_dogs,
|
|
refresh,
|
|
rear_mounted_bike_carrier,
|
|
)
|
|
|
|
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,
|
|
pax=pax,
|
|
pax_options=pax_options,
|
|
rear_mounted_bike_carrier=rear_mounted_bike_carrier,
|
|
)
|
|
|
|
|
|
@app.route("/outbound1")
|
|
def outbound1_page() -> str:
|
|
return build_outbound("outbound1")
|
|
|
|
|
|
@app.route("/outbound2")
|
|
def outbound2_page() -> str:
|
|
return build_outbound("outbound2")
|
|
|
|
|
|
@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")
|
|
|
|
|
|
def format_pet_options(o: dict[str, bool]) -> list[str]:
|
|
"""Read pet options from API and format for display."""
|
|
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(
|
|
dep: str,
|
|
arr: str,
|
|
d: str,
|
|
ticket_tier: str,
|
|
refresh: bool = False,
|
|
rear_mounted_bike_carrier: bool = False,
|
|
) -> dict[str, list[dict[str, Any]]]:
|
|
pax = read_pax()
|
|
|
|
params = f"{dep}_{arr}_{d}_{ticket_tier}_{pax.adults}_{pax.small_dogs}_{rear_mounted_bike_carrier}"
|
|
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)
|
|
data = get_accommodations(
|
|
dep,
|
|
arr,
|
|
d,
|
|
ticket_tier,
|
|
vehicle,
|
|
pax.adults,
|
|
pax.small_dogs,
|
|
rear_mounted_bike_carrier,
|
|
)
|
|
|
|
with open(filename, "w") as out:
|
|
print(filename)
|
|
json.dump(data, out, indent=2)
|
|
|
|
return data
|
|
|
|
|
|
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>"
|
|
)
|
|
def cabins(
|
|
sailing_id: int,
|
|
departure_port: str,
|
|
arrival_port: str,
|
|
departure_date: str,
|
|
ticket_tier: str,
|
|
) -> str:
|
|
"""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
|
|
|
|
pax = read_pax()
|
|
rear_mounted_bike_carrier = (
|
|
flask.request.args.get("rear_mounted_bike_carrier") == "true"
|
|
)
|
|
|
|
prices = get_prices_with_cache(
|
|
direction,
|
|
start,
|
|
end,
|
|
routes[direction],
|
|
pax.adults,
|
|
pax.small_dogs,
|
|
False,
|
|
rear_mounted_bike_carrier,
|
|
)
|
|
crossing = lookup_sailing_id(prices, sailing_id)
|
|
|
|
cabin_data = get_accommodations_with_cache(
|
|
departure_port,
|
|
arrival_port,
|
|
departure_date,
|
|
ticket_tier,
|
|
False,
|
|
rear_mounted_bike_carrier,
|
|
)
|
|
accommodations = [
|
|
a
|
|
for a in cabin_data["accommodations"]
|
|
if a["quantityAvailable"] > 0 and a["code"] != "RS"
|
|
# and "Inside" not in a["description"]
|
|
]
|
|
|
|
pax_labels = {i["pax"]: i["label"] for i in pax_options}
|
|
|
|
dep = dateutil.parser.isoparse(departure_date)
|
|
|
|
return flask.render_template(
|
|
"cabins.html",
|
|
port_lookup=ferry.port_lookup,
|
|
departure_port=departure_port,
|
|
arrival_port=arrival_port,
|
|
departure_date=dep,
|
|
ticket_tier=ticket_tier,
|
|
accommodations=accommodations,
|
|
pet_accommodations=cabin_data["petAccommodations"],
|
|
crossing=crossing,
|
|
get_duration=get_duration,
|
|
time_delta=time_delta,
|
|
format_pet_options=format_pet_options,
|
|
section=section,
|
|
pax=pax,
|
|
pax_label=pax_labels[pax],
|
|
rear_mounted_bike_carrier=rear_mounted_bike_carrier,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="0.0.0.0")
|