74 lines
2.6 KiB
Python
74 lines
2.6 KiB
Python
|
import datetime
|
||
|
|
import requests
|
||
|
|
|
||
|
|
|
||
|
|
class SNCFJourney(dict):
|
||
|
|
def __repr__(self):
|
||
|
|
sections = []
|
||
|
|
for section in self["sections"]:
|
||
|
|
nice_dep_date = SNCF.api_date_to_datetime(section['departure_date_time']).strftime("%H:%M")
|
||
|
|
nice_arr_date = SNCF.api_date_to_datetime(section['arrival_date_time']).strftime("%H:%M")
|
||
|
|
sections.append(f"{nice_dep_date} {section['from']['name']} - {section['to']['name']} {nice_arr_date}")
|
||
|
|
return "\n".join(sections)
|
||
|
|
|
||
|
|
|
||
|
|
class SNCF:
|
||
|
|
BASE_URL = "https://api.sncf.com/v1/coverage/sncf"
|
||
|
|
|
||
|
|
def __init__(self, token: str):
|
||
|
|
self.token = token
|
||
|
|
self.source_stop_point: str = ""
|
||
|
|
self.dest_stop_point: str = ""
|
||
|
|
self.allowed_lines: list[str] = []
|
||
|
|
self.max_transfers: int = 0
|
||
|
|
self.max_duration_secs: int = 1800
|
||
|
|
self.prefered_date_format = "%H:%M"
|
||
|
|
|
||
|
|
def set_source_stop_point(self, stop_point: str):
|
||
|
|
self.source_stop_point = stop_point
|
||
|
|
|
||
|
|
def set_dest_stop_point(self, stop_point: str):
|
||
|
|
self.dest_stop_point = stop_point
|
||
|
|
|
||
|
|
def set_allowed_lines(self, allowed_lines: list[str]):
|
||
|
|
self.allowed_lines = allowed_lines
|
||
|
|
|
||
|
|
def get_next_journeys(self, count: int = 1):
|
||
|
|
valid_journeys = []
|
||
|
|
api_response = self.api_request(f"/journeys?from={self.source_stop_point}&to={self.dest_stop_point}&datetime={self.date_now()}&count={count}")
|
||
|
|
journeys = api_response["journeys"]
|
||
|
|
for journey in journeys:
|
||
|
|
if journey["nb_transfers"] <= self.max_transfers and journey["duration"] <= self.max_duration_secs:
|
||
|
|
valid_journeys.append(SNCFJourney(journey))
|
||
|
|
return valid_journeys
|
||
|
|
|
||
|
|
def api_request(self, url):
|
||
|
|
req = requests.get(SNCF.BASE_URL + url, headers={"Authorization": self.token})
|
||
|
|
req.raise_for_status()
|
||
|
|
return req.json()
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def date_now() -> str:
|
||
|
|
return datetime.datetime.now().strftime("%Y%m%dT%H%M%S")
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def api_date_to_datetime(api_date: str) -> datetime:
|
||
|
|
return datetime.datetime.strptime(api_date, "%Y%m%dT%H%M%S")
|
||
|
|
|
||
|
|
def test_api_key(self) -> bool:
|
||
|
|
try:
|
||
|
|
self.api_request("")
|
||
|
|
return True
|
||
|
|
except requests.HTTPError:
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
sncf = SNCF("5722269b-2e58-49ee-986b-21741438d5ff")
|
||
|
|
sncf.set_source_stop_point("stop_point:SNCF:87214056:Train")
|
||
|
|
sncf.set_dest_stop_point("stop_point:SNCF:87212027:Train")
|
||
|
|
next_journeys = sncf.get_next_journeys(count=5)
|
||
|
|
for next_journey in next_journeys:
|
||
|
|
print(next_journey)
|
||
|
|
#print(json.dumps(next_schedules, indent=4))
|