#!/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/////" ) 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")