This commit is contained in:
Edward Betts 2023-01-08 13:03:23 +00:00
parent 944fe24662
commit 5bca7bb801

66
main.py
View file

@ -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/<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)
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