From 553365c108e1d22ae0371d2a7670049e3371e239 Mon Sep 17 00:00:00 2001 From: Marty Oehme Date: Wed, 3 Sep 2025 17:51:03 +0200 Subject: [PATCH 1/4] Extract request to start connection from getting init token --- src/nj/main.py | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/src/nj/main.py b/src/nj/main.py index b1c62c4..00da28f 100644 --- a/src/nj/main.py +++ b/src/nj/main.py @@ -26,7 +26,7 @@ def dprint(txt) -> None: print(f"{datetime.now()}: {txt}") -def request_init_token(endpoint: str = "/nj-booking-ocp/init/start") -> str: +def request_start(endpoint: str = "/nj-booking-ocp/init/start") -> dict: headers = { "Referer": "https://www.nightjet.com", "Content-Type": "application/json", @@ -35,8 +35,11 @@ def request_init_token(endpoint: str = "/nj-booking-ocp/init/start") -> str: resp_json = requests.post( f"{BASE_URL}{endpoint}", data=json.dumps(body), headers=headers ).json() - token = resp_json["token"] + return resp_json + +def get_init_token(endpoint: str = "/nj-booking-ocp/init/start") -> str: + token = request_start(endpoint)["token"] dprint(f"Received init token: {token}") return token @@ -266,8 +269,13 @@ def notify_user(previous: Price, new: Price, channel: str) -> None: ) -def query(start_station: int, end_station: int, travel_date: datetime, traveller_birthdate: datetime) -> list[Price]: - token = request_init_token() +def query( + start_station: int, + end_station: int, + travel_date: datetime, + traveller_birthdate: datetime, +) -> list[Price]: + token = get_init_token() connections = request_connections(token, start_station, end_station, travel_date) booking_requests = connection_data_to_booking_requests(connections, traveller_birthdate=traveller_birthdate) bookings = [request_bookings(token, req) for req in booking_requests] From cf968da14815bfdc712f9f18d67ac5e3d41130ad Mon Sep 17 00:00:00 2001 From: Marty Oehme Date: Wed, 3 Sep 2025 17:51:03 +0200 Subject: [PATCH 2/4] Format main file --- src/nj/main.py | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/src/nj/main.py b/src/nj/main.py index 00da28f..f2a5f03 100644 --- a/src/nj/main.py +++ b/src/nj/main.py @@ -8,6 +8,7 @@ from typing import Annotated, Any import requests import typer +from rich import print BASE_URL = "https://www.nightjet.com" BASE_DIR = "out" @@ -277,7 +278,9 @@ def query( ) -> list[Price]: token = get_init_token() connections = request_connections(token, start_station, end_station, travel_date) - booking_requests = connection_data_to_booking_requests(connections, traveller_birthdate=traveller_birthdate) + booking_requests = connection_data_to_booking_requests( + connections, traveller_birthdate=traveller_birthdate + ) bookings = [request_bookings(token, req) for req in booking_requests] prices = extract_prices(bookings) @@ -299,7 +302,10 @@ def main( end_station: int = typer.Option( END_STATION, help="Destination station number. (default: Paris Est)" ), - birthdate: str = typer.Option(TRAVELLER_BIRTHDATE.strftime("%Y-%m-%d"), help="Traveller birthdate, may be important for discounts. (YYYY-MM-DD)"), + birthdate: str = typer.Option( + TRAVELLER_BIRTHDATE.strftime("%Y-%m-%d"), + help="Traveller birthdate, may be important for discounts. (YYYY-MM-DD)", + ), notification_channel: str = typer.Option( NOTIFICATION_CHANNEL, help="ntfy channel to inform user on." ), @@ -338,7 +344,10 @@ def main( while True: prices = query( - start_station=start_station, end_station=end_station, travel_date=travel_date_obj, traveller_birthdate=birth_date_obj, + start_station=start_station, + end_station=end_station, + travel_date=travel_date_obj, + traveller_birthdate=birth_date_obj, ) # create a snapshot of all current prices From 823451f00a7b4645129d9e5b825f666ca4ac4b68 Mon Sep 17 00:00:00 2001 From: Marty Oehme Date: Wed, 3 Sep 2025 18:00:17 +0200 Subject: [PATCH 3/4] Add option to query latest bookable date --- src/nj/main.py | 20 +++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/src/nj/main.py b/src/nj/main.py index f2a5f03..4699bda 100644 --- a/src/nj/main.py +++ b/src/nj/main.py @@ -1,7 +1,8 @@ import csv import json +import sys from dataclasses import dataclass -from datetime import datetime +from datetime import date, datetime from pathlib import Path from time import sleep from typing import Annotated, Any @@ -287,6 +288,13 @@ def query( return prices +def query_max_booking_date(endpoint: str = "/nj-booking-ocp/init/start") -> date | None: + latest_date = request_start().get("maxBookableDate") + if not latest_date: + return None + return date.fromisoformat(latest_date) + + ## CLI app = typer.Typer() @@ -330,7 +338,17 @@ def main( dump_price_snapshot: bool = typer.Option( True, help="Dump _all_ queried prices into a timestamped csv file." ), + latest_bookable_date: bool = typer.Option( + False, help="Check for latest currently possible booking date only." + ), ): + if latest_bookable_date: + latest_date = query_max_booking_date() + if not latest_date: + dprint("Could not determine max bookable date.") + dprint(f"Latest currently bookable date: {latest_date}") + sys.exit(0) + base_output_directory.mkdir(exist_ok=True, parents=True) lowest_prices_path = base_output_directory.joinpath(lowest_prices_filename) price_snapshot_path = base_output_directory.joinpath(price_snapshot_pattern) From eca5f2e4393b3f9bd0da256ff1ff331c0bc71bb8 Mon Sep 17 00:00:00 2001 From: Marty Oehme Date: Wed, 3 Sep 2025 18:07:57 +0200 Subject: [PATCH 4/4] Separate latest bookable date and price query into two commands --- src/nj/main.py | 44 ++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 42 insertions(+), 2 deletions(-) diff --git a/src/nj/main.py b/src/nj/main.py index 4699bda..ab638b3 100644 --- a/src/nj/main.py +++ b/src/nj/main.py @@ -299,7 +299,7 @@ def query_max_booking_date(endpoint: str = "/nj-booking-ocp/init/start") -> date app = typer.Typer() -@app.command() +@app.command("query") def main( travel_date: Annotated[ str, typer.Argument(help="Travel day to search from. (YYYY-MM-DD)") @@ -342,6 +342,12 @@ def main( False, help="Check for latest currently possible booking date only." ), ): + """Check the (lowest) prices for a single connection. + + Will run repeatedly or one-shot and query the prices for a specific connection. + Can be set to output all available prices or just output the lowest found. + Outputs will (curently) always be given as csv files. + """ if latest_bookable_date: latest_date = query_max_booking_date() if not latest_date: @@ -397,7 +403,41 @@ def main( if not monitor_mode: break dprint( - f"Query complete. Monitoring mode active, sleeping for {monitor_frequency} seconds..." + f"Checked for connection prices. Monitoring mode active, sleeping for {monitor_frequency} seconds..." + ) + sleep(monitor_frequency) + + +@app.command("lastdate") +def latestbookable_command( + notification_channel: str = typer.Option( + NOTIFICATION_CHANNEL, help="ntfy channel to inform user on." + ), + monitor_mode: bool = typer.Option( + True, + help="Run queries repeatedly over time. If False only runs a single query (oneshot mode).", + ), + monitor_frequency: int = typer.Option( + MONITOR_FREQUENCY, + help="How often to run price queries if in monitoring mode, in seconds.", + ), +): + """Check for the currently latest possible booking date. + + Will run repeatedly or one-shot and simply query for the date that is + currently the _last_ day which can be selected for any booking query. + """ + + while True: + latest_date = query_max_booking_date() + if not latest_date: + dprint("Could not determine max bookable date.") + dprint(f"Latest currently bookable date: {latest_date}") + + if not monitor_mode: + break + dprint( + f"Checked for latest bookable date. Monitoring mode active, sleeping for {monitor_frequency} seconds..." ) sleep(monitor_frequency)