brittany-ferries/main.py
2023-04-16 19:56:26 +01:00

462 lines
13 KiB
Python
Executable file

#!/usr/bin/python3
"""Check prices of ferries to France."""
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
# import werkzeug.exceptions
# from werkzeug.debug.tbtools import DebugTraceback
# from werkzeug.debug.tbtools import get_current_traceback
app = flask.Flask(__name__)
app.debug = True
routes = {
"outbound": [
("PORTSMOUTH", "CAEN"),
("PORTSMOUTH", "CHERBOURG"),
("PORTSMOUTH", "ST MALO"),
("PORTSMOUTH", "LE HAVRE"),
("POOLE", "CHERBOURG"),
],
"return": [
("CAEN", "PORTSMOUTH"),
("CHERBOURG", "PORTSMOUTH"),
("ST MALO", "PORTSMOUTH"),
("LE HAVRE", "PORTSMOUTH"),
("CHERBOURG", "POOLE"),
],
}
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: Exception) -> tuple[str, int]:
"""Show error page."""
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 = list(tb._te.format_exception_only())
frames = [f for f, lineno in traceback.walk_tb(current_traceback)]
last_frame = frames[-1]
last_frame_args = inspect.getargs(last_frame.f_code)
return (
flask.render_template(
"show_error.html",
plaintext=tb.render_traceback_text(),
exception="".join(exc_lines),
exception_type=tb._te.exc_type.__name__,
summary=summary,
last_frame=last_frame,
last_frame_args=last_frame_args,
),
500,
)
# @app.errorhandler(werkzeug.exceptions.InternalServerError)
# def exception_handler(e):
# # tb = get_current_traceback()
# tb = DebugTraceback()
# last_frame = next(frame for frame in reversed(tb.frames) if not frame.is_library)
# last_frame_args = inspect.getargs(last_frame.code)
# return (
# flask.render_template(
# "show_error.html",
# tb=tb,
# 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) -> str:
"""Generate a URL for the cabins on a given crossing."""
dt = datetime.fromisoformat(crossing["departureDateTime"]["iso"])
utc_dt = dt.astimezone(pytz.utc)
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,
)
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]],
refresh: bool = False,
adults: int = 2,
small_dogs: int = 1,
) -> PriceData:
"""Get price data using cache."""
params = f"{direction}_{start}_{end}_{adults}_{small_dogs}"
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,
)["crossings"],
)
for dep, arr in selection
]
with open(cache_filename(params), "w") as out:
json.dump(all_data, out, indent=2)
return all_data
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"))
adults_str = flask.request.args.get("adults")
adults = int(adults_str) if adults_str else 2
small_dogs_str = flask.request.args.get("small_dogs")
small_dogs = int(small_dogs_str) if small_dogs_str else 2
direction = section[:-1] if section[-1].isdigit() else section
all_data = get_prices_with_cache(
direction, start, end, routes["outbound"], refresh, adults, small_dogs
)
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,
)
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"))
direction = section[:-1] if section[-1].isdigit() else section
adults_str = flask.request.args.get("adults")
adults = int(adults_str) if adults_str else 2
small_dogs_str = flask.request.args.get("small_dogs")
small_dogs = int(small_dogs_str) if small_dogs_str else 2
all_data = get_prices_with_cache(
direction, start, end, routes["return"], refresh, adults, small_dogs
)
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,
)
@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
) -> dict[str, list[dict[str, Any]]]:
params = f"{dep}_{arr}_{d}_{ticket_tier}"
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)
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
prices = get_prices_with_cache(direction, start, end, routes[direction])
crossing = lookup_sailing_id(prices, sailing_id)
cabin_data = get_accommodations_with_cache(
departure_port, arrival_port, departure_date, ticket_tier
)
accommodations = [
a
for a in cabin_data["accommodations"]
if a["quantityAvailable"] > 0 and a["code"] != "RS"
# and "Inside" not in a["description"]
]
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,
)
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5001)