311 lines
11 KiB
Python
311 lines
11 KiB
Python
from selenium import webdriver
|
|
from selenium.webdriver.firefox.options import Options
|
|
from models import Booking
|
|
import json
|
|
from selenium.webdriver.common.by import By
|
|
from selenium.webdriver.support.ui import WebDriverWait
|
|
from selenium.webdriver.support import expected_conditions as EC
|
|
import time
|
|
import requests
|
|
from selenium.webdriver.chrome.service import Service
|
|
|
|
|
|
class LoginFailedException(Exception):
|
|
def __init__(self, *args: object) -> None:
|
|
super().__init__(*args)
|
|
|
|
|
|
class XClient:
|
|
def __init__(self, username, password, on_access_token_change=lambda x: x):
|
|
self.username = username
|
|
self.password = password
|
|
self.access_token = None
|
|
self.id = -1
|
|
self.on_access_token_change = on_access_token_change
|
|
|
|
self.session = requests.Session()
|
|
# Default headers for session that mimicks browser
|
|
self.session.headers.update(
|
|
{
|
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3"
|
|
}
|
|
)
|
|
|
|
def fetch_access_token(self):
|
|
options = webdriver.FirefoxOptions()
|
|
options.add_argument("--headless")
|
|
|
|
driver = webdriver.Firefox(options=options)
|
|
|
|
driver.get("https://x.tudelft.nl")
|
|
|
|
button = WebDriverWait(driver, 30).until(
|
|
EC.element_to_be_clickable(
|
|
(By.XPATH, "//span[contains(text(), 'TUDelft')]")
|
|
)
|
|
)
|
|
button.click()
|
|
|
|
button = WebDriverWait(driver, 30).until(
|
|
EC.element_to_be_clickable(
|
|
(
|
|
By.XPATH,
|
|
"//h3[contains(text(), 'Delft University of Technology')]",
|
|
)
|
|
)
|
|
)
|
|
|
|
button.click()
|
|
|
|
# Input the username
|
|
username_input = driver.find_element(By.ID, "username")
|
|
username_input.send_keys(self.username) # Replace with your actual username
|
|
|
|
password_input = driver.find_element(By.ID, "password")
|
|
password_input.send_keys(self.password)
|
|
|
|
password_input.submit()
|
|
|
|
time.sleep(2)
|
|
|
|
cookie = driver.execute_script(
|
|
"return window.localStorage.getItem('delcom_auth');"
|
|
)
|
|
|
|
if cookie is None:
|
|
raise LoginFailedException("Logging in to X has failed")
|
|
|
|
delcom_auth = json.loads(cookie)
|
|
|
|
driver.quit()
|
|
|
|
if delcom_auth.get("tokenResponse") is None:
|
|
raise LoginFailedException(
|
|
"Logging in to X has failed, missing tokenResponse"
|
|
)
|
|
|
|
access_token = delcom_auth["tokenResponse"].get("accessToken", None)
|
|
return access_token
|
|
|
|
def list_slots(self, start, end, tagIDs=[28]):
|
|
"""
|
|
URl:
|
|
https://backbone-web-api.production.delft.delcom.nl/bookable-slots
|
|
Get parameters:
|
|
s: {"startDate":"2024-09-09T14:23:24.934Z","endDate":"2024-09-09T22:00:00.000Z","tagIds":{"$in":[28]},"availableFromDate":{"$gt":"2024-09-09T14:23:24.935Z"},"availableTillDate":{"$gte":"2024-09-09T14:23:24.934Z"}}
|
|
join: product
|
|
"""
|
|
|
|
self.login_if_necessary()
|
|
start_stamp = start.strftime("%Y-%m-%dT%H:%M:%S.000Z")
|
|
end_stamp = end.strftime("%Y-%m-%dT%H:%M:%S.000Z")
|
|
|
|
response = self.session.get(
|
|
"https://backbone-web-api.production.delft.delcom.nl/bookable-slots",
|
|
params={
|
|
"s": json.dumps(
|
|
{
|
|
"startDate": start_stamp,
|
|
"endDate": end_stamp,
|
|
"tagIds": {"$in": tagIDs},
|
|
"availableFromDate": {"$gt": start_stamp},
|
|
"availableTillDate": {"$gte": start_stamp},
|
|
}
|
|
),
|
|
"join": ["product", "linkedProduct"],
|
|
},
|
|
headers={"Authorization": f"Bearer {self.access_token}"},
|
|
)
|
|
return [Booking(**x) for x in response.json()["data"]]
|
|
|
|
def my_bookings(self, start, end):
|
|
# Ensure user is logged in
|
|
self.login_if_necessary()
|
|
|
|
# Replace this with the actual member ID of the logged-in user
|
|
member_id = self.user_id # Assuming it's stored when the user logs in
|
|
|
|
# Define the URL
|
|
url = "https://backbone-web-api.production.delft.delcom.nl/participations"
|
|
|
|
# Create the parameters for the request
|
|
params = {
|
|
"s": json.dumps(
|
|
{
|
|
"$or": [
|
|
{"memberId": member_id}, # Fetch bookings based on the user ID
|
|
],
|
|
"booking.startDate": {
|
|
"$gte": start.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
|
|
},
|
|
"booking.endDate": {"$lte": end.strftime("%Y-%m-%dT%H:%M:%S.%fZ")},
|
|
},
|
|
),
|
|
"join": ["booking", "linkedProduct", "product"],
|
|
"fetchOptimizedCustomerDashboard": "true",
|
|
"fetchTicket": "false",
|
|
"sort": "booking.startDate,ASC",
|
|
}
|
|
|
|
# Make the GET request to the API
|
|
response = self.session.get(
|
|
url,
|
|
params=params,
|
|
headers={"Authorization": f"Bearer {self.access_token}"},
|
|
)
|
|
|
|
# Check if the request was successful
|
|
if response.status_code == 200:
|
|
bookings = [
|
|
Booking(**(x | {"bookingId": x["id"], "bookableProductId": None}))
|
|
for x in response.json()["data"]["bookings"]
|
|
]
|
|
return bookings
|
|
elif response.status_code == 403:
|
|
raise Exception(f"Failed to fetch bookings: {response.json()['message']}")
|
|
else:
|
|
raise Exception(
|
|
f"Failed to fetch bookings: {response.status_code}, {response.text}"
|
|
)
|
|
|
|
def make_booking(self, booking):
|
|
# Ensure that we're logged in
|
|
self.login_if_necessary()
|
|
|
|
# Extract necessary details from the booking object
|
|
booking_data = {
|
|
"organizationId": None, # This is None in the example curl
|
|
"memberId": self.user_id, # This is fetched during login
|
|
"bookingId": booking.booking_id, # Booking ID from the booking object
|
|
"primaryPurchaseMessage": None,
|
|
"secondaryPurchaseMessage": None,
|
|
"params": {
|
|
"startDate": booking.start.strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
|
|
"endDate": booking.end.strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
|
|
"bookableProductId": booking.bookable_product_id,
|
|
"bookableLinkedProductId": booking.linked_product_id,
|
|
"bookingId": booking.booking_id, # Booking ID again
|
|
"invitedMemberEmails": [],
|
|
"invitedGuests": [],
|
|
"invitedOthers": [],
|
|
"primaryPurchaseMessage": None,
|
|
"secondaryPurchaseMessage": None,
|
|
},
|
|
"dateOfRegistration": None,
|
|
}
|
|
|
|
# Send the booking request
|
|
response = self.session.post(
|
|
"https://backbone-web-api.production.delft.delcom.nl/participations",
|
|
json=booking_data,
|
|
headers={"Authorization": f"Bearer {self.access_token}"},
|
|
)
|
|
|
|
# Check the response status
|
|
if response.status_code == 201:
|
|
return response.json() # Return response if successful
|
|
elif response.status_code == 403:
|
|
raise Exception(f"Failed to make booking: {response.json()['message']}")
|
|
else:
|
|
raise Exception(
|
|
f"Failed to make booking: {response.status_code}, {response.text}"
|
|
)
|
|
|
|
def check_booking_availability(self, booking):
|
|
# Ensure user is logged in
|
|
self.login_if_necessary()
|
|
|
|
# Define the URL
|
|
url = "https://backbone-web-api.production.delft.delcom.nl/bookable-slots"
|
|
|
|
# Create the parameters for the request
|
|
params = {
|
|
"s": json.dumps(
|
|
{
|
|
"bookingId": booking.booking_id,
|
|
"tagIds": {"$in": [28]},
|
|
},
|
|
),
|
|
"join": ["product", "linkedProduct"],
|
|
}
|
|
|
|
# Make the GET request to the API
|
|
response = self.session.get(
|
|
url,
|
|
params=params,
|
|
headers={"Authorization": f"Bearer {self.access_token}"},
|
|
)
|
|
|
|
# Check if the request was successful
|
|
if response.status_code == 200:
|
|
if len(response.json()["data"]) != 1:
|
|
return False
|
|
booking = Booking(**response.json()["data"][0])
|
|
return booking.available
|
|
|
|
elif response.status_code == 403:
|
|
raise Exception(
|
|
f"Failed to check booking availability: {response.json()['message']}"
|
|
)
|
|
else:
|
|
raise Exception(
|
|
f"Failed to check booking availability: {response.status_code}, {response.text}"
|
|
)
|
|
|
|
@property
|
|
def user_id(self):
|
|
if self.id != -1:
|
|
return self.id
|
|
|
|
self.login_if_necessary()
|
|
|
|
response = self.session.get(
|
|
"https://backbone-web-api.production.delft.delcom.nl/auth?cf=0",
|
|
headers={"Authorization": f"Bearer {self.access_token}"},
|
|
)
|
|
self.id = response.json()["id"]
|
|
return self.id
|
|
|
|
def cancel_booking(self, booking):
|
|
participation_id = None
|
|
for i in booking.booking_dict.get("participations", []):
|
|
if i["memberId"] == self.user_id:
|
|
participation_id = i["id"]
|
|
break
|
|
|
|
assert (
|
|
participation_id is not None
|
|
), "Booking not found in user's participations"
|
|
|
|
## Send a DELETE request to the URL https://backbone-web-api.production.delft.delcom.nl/participations/{id}
|
|
response = self.session.delete(
|
|
f"https://backbone-web-api.production.delft.delcom.nl/participations/{participation_id}",
|
|
headers={"Authorization": f"Bearer {self.access_token}"},
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
return True
|
|
elif response.status_code == 403:
|
|
raise Exception(f"Failed to cancel booking: {response.json()['message']}")
|
|
else:
|
|
raise Exception(
|
|
f"Failed to fetch bookings: {response.status_code}, {response.text}"
|
|
)
|
|
|
|
def login(self):
|
|
self.access_token = self.fetch_access_token()
|
|
self.on_access_token_change(self.access_token)
|
|
|
|
def is_logged_in(self):
|
|
# Send request to 'https://backbone-web-api.production.delft.delcom.nl/auth?cf=0 with access token as bearer, check status code
|
|
response = self.session.get(
|
|
"https://backbone-web-api.production.delft.delcom.nl/auth?cf=0",
|
|
headers={"Authorization": f"Bearer {self.access_token}"},
|
|
)
|
|
|
|
return response.status_code == 200
|
|
|
|
def login_if_necessary(self):
|
|
if not self.is_logged_in():
|
|
self.login()
|