brittany-ferries/main.py
2023-01-08 11:49:57 +00:00

308 lines
8.5 KiB
Python
Executable file

#!/usr/bin/python3
"""Check prices of ferries to France."""
import inspect
import json
import os
import os.path
import re
from datetime import date, datetime, timedelta
from typing import Any
import flask
import pytz
import routes
import werkzeug.exceptions
from werkzeug.debug.tbtools import get_current_traceback
from werkzeug.wrappers import Response
from ferry import ports
from ferry.api import get_accommodations, get_prices
from ferry.read_config import ferry_config, vehicle_from_config
app = flask.Flask(__name__)
app.debug = False
def cache_location() -> str:
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):
tb = get_current_traceback()
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("/route/<departure_port>/<arrival_port>/<from_date>/<to_date>")
def show_route(
departure_port: str, arrival_port: str, from_date: str, to_date: str
) -> str:
"""Page showing list of prices."""
prices = get_prices(departure_port, arrival_port)
port_lookup = {code: name for name, code in ports.items()}
return flask.render_template(
"route.html",
departure_port=port_lookup[departure_port],
arrival_port=port_lookup[arrival_port],
days=prices["crossings"],
parse_date=parse_date,
)
@app.route("/")
def start() -> Response:
"""Start page."""
return flask.render_template("index.html")
return flask.redirect(flask.url_for("outbound_page"))
def cabins_url(dep, arr, crossing, ticket_tier):
dt = datetime.fromisoformat(crossing["departureDateTime"]["iso"])
utc_dt = dt.astimezone(pytz.utc)
return flask.url_for(
"cabins",
departure_port=ports[dep],
arrival_port=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
def get_prices_with_cache(
name: str,
start: str,
end: str,
selection: list[tuple[str, str]],
refresh: bool = False,
) -> list[tuple[str, str, dict[str, Any]]]:
params = f"{name}_{start}_{end}"
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)
all_data = [
(dep, arr, get_prices(ports[dep], ports[arr], start, end, vehicle)["crossings"])
for dep, arr in selection
]
with open(filename, "w") as out:
print(filename)
json.dump(all_data, out, indent=2)
return all_data
def build_outbound(section: str) -> str:
"""Show all routes on one page."""
selection = [
("PORTSMOUTH", "CAEN"),
("PORTSMOUTH", "CHERBOURG"),
("PORTSMOUTH", "ST MALO"),
# ("POOLE", "CHERBOURG"),
]
start = ferry_config.get(section, "from")
end = ferry_config.get(section, "to")
refresh = bool(flask.request.args.get("refresh"))
all_data = get_prices_with_cache(section, start, end, selection, refresh)
return flask.render_template(
"all_routes.html",
data=all_data,
days_until_start=get_days_until_start(),
ports=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,
)
@app.route("/outbound1")
def outbound1_page() -> str:
return build_outbound("outbound1")
@app.route("/outbound2")
def outbound2_page() -> str:
return build_outbound("outbound2")
def format_pet_options(o: dict[str, bool]) -> list[str]:
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
@app.route("/return")
def return_page() -> str:
"""Show all routes on one page."""
selection = [
("CAEN", "PORTSMOUTH"),
("CHERBOURG", "PORTSMOUTH"),
("ST MALO", "PORTSMOUTH"),
# ("CHERBOURG", "POOLE"),
]
start = ferry_config.get("return", "from")
end = ferry_config.get("return", "to")
refresh = bool(flask.request.args.get("refresh"))
all_data = get_prices_with_cache("return", start, end, selection, refresh)
return flask.render_template(
"all_routes.html",
data=all_data,
ports=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,
)
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
@app.route("/cabins/<departure_port>/<arrival_port>/<departure_date>/<ticket_tier>")
def cabins(
departure_port: str, arrival_port: str, departure_date: str, ticket_tier: str
) -> str:
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"]
]
return flask.render_template(
"cabins.html",
departure_port=departure_port,
arrival_port=arrival_port,
departure_date=departure_date,
ticket_tier=ticket_tier,
accommodations=accommodations,
pet_accommodations=cabin_data["petAccommodations"],
)
@app.route("/routes")
def route_list() -> str:
"""List of routes."""
return flask.render_template("index.html", routes=routes, ports=ports)
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5001)