diff --git a/main.py b/main.py index c098478..571ef45 100755 --- a/main.py +++ b/main.py @@ -25,6 +25,7 @@ app.debug = False def cache_location() -> str: + """Cache location for the current user.""" return os.path.expanduser(ferry_config.get("cache", "location")) @@ -79,29 +80,14 @@ def parse_date(d: str) -> date: return datetime.strptime(d, "%Y-%m-%d").date() -@app.route("/route////") -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) - - return flask.render_template( - "route.html", - departure_port=ferry.port_lookup[departure_port], - arrival_port=ferry.port_lookup[arrival_port], - days=prices["crossings"], - parse_date=parse_date, - ) - - @app.route("/") def start() -> Response | str: """Start page.""" return flask.render_template("index.html") -def cabins_url(dep, arr, crossing, ticket_tier): +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) @@ -120,31 +106,44 @@ def get_days_until_start() -> int: return (start - date.today()).days +PriceData = list[tuple[str, str, 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( name: str, start: str, end: str, selection: list[tuple[str, str]], refresh: bool = False, -) -> list[tuple[str, str, dict[str, Any]]]: - +) -> PriceData: + """Get price data using cache.""" 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)) + if not refresh: + data = check_cache_for_prices(params) + if data: return data vehicle = vehicle_from_config(ferry_config) - filename = cache_filename(params) all_data = [ ( @@ -157,8 +156,7 @@ def get_prices_with_cache( for dep, arr in selection ] - with open(filename, "w") as out: - print(filename) + with open(cache_filename(params), "w") as out: json.dump(all_data, out, indent=2) return all_data