552 lines
19 KiB
Python
552 lines
19 KiB
Python
import asyncio
|
|
import json
|
|
import logging
|
|
import time
|
|
from abc import ABC, abstractmethod
|
|
from pathlib import Path
|
|
from urllib.parse import urljoin
|
|
|
|
import config
|
|
import pendulum
|
|
from aiohttp import ClientResponse, ClientSession
|
|
from bs4 import BeautifulSoup
|
|
from models import Booking, BookingFilter, Club, Court, Sport, User
|
|
from payload_builders import PayloadBuilder
|
|
from pendulum import DateTime
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
POST_HEADERS = config.get_post_headers("gestion-sports")
|
|
|
|
|
|
class Connector(ABC):
|
|
"""
|
|
Abstract class that defines the method a connector
|
|
to a website for sport booking should have
|
|
"""
|
|
|
|
@abstractmethod
|
|
async def book(self, user: User, booking_filter: BookingFilter) -> Court | None:
|
|
"""
|
|
Book a court matching the filter for a user
|
|
|
|
:param user: the user who will have the booking
|
|
:param booking_filter: the conditions to book (date, time, court)
|
|
:return: the court booked
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def has_user_ongoing_booking(self, user: User) -> bool:
|
|
"""
|
|
Test whether the user has ongoing bookings
|
|
|
|
:param user: the user who will have the booking
|
|
:return: true if the user has at least one ongoing booking, false otherwise
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def cancel_booking_id(self, user: User, booking_id: int) -> None:
|
|
"""
|
|
Cancel the booking for a given user
|
|
|
|
:param user: the user who has the booking
|
|
:param booking_id: the id of the booking
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def cancel_booking(self, user: User, booking_filter: BookingFilter) -> None:
|
|
"""
|
|
Cancel the booking that meet some conditions for a given user
|
|
|
|
:param user: the user who has the booking
|
|
:param booking_filter: the booking conditions to meet to cancel the booking
|
|
"""
|
|
pass
|
|
|
|
|
|
class GestionSportsConnector(Connector):
|
|
"""
|
|
The connector for the Gestion Sports platform.
|
|
It handles all the requests to the website.
|
|
"""
|
|
|
|
def __init__(self, club: Club):
|
|
if club is None:
|
|
raise ValueError("A connector cannot be instantiated without a club")
|
|
if club.booking_platform.id != "gestion-sports":
|
|
raise ValueError(
|
|
"Gestion Sports connector was instantiated with a club not handled"
|
|
" by gestions sports. Club id is {} instead of gestion-sports".format(
|
|
club.id
|
|
)
|
|
)
|
|
|
|
self.club = club
|
|
|
|
def _get_url_path(self, name: str) -> str:
|
|
"""
|
|
Get the URL path for the service with the given name
|
|
|
|
:param name: the name of the service
|
|
:return: the URL path
|
|
"""
|
|
self._check_url_path_exists(name)
|
|
|
|
return urljoin(
|
|
self.club.booking_platform.url,
|
|
self.club.booking_platform.urls.get(name).path,
|
|
)
|
|
|
|
def _get_payload_template(self, name: str) -> Path:
|
|
"""
|
|
Get the path to the template file for the service with the given name
|
|
|
|
:param name: the name of the service
|
|
:return: the path to the template file
|
|
"""
|
|
self._check_payload_template_exists(name)
|
|
|
|
return (
|
|
config.get_resources_folder()
|
|
/ self.club.booking_platform.urls.get(name).payload_template
|
|
)
|
|
|
|
def _check_url_path_exists(self, name: str) -> None:
|
|
"""
|
|
Check that the URL path for the given service is defined
|
|
|
|
:param name: the name of the service
|
|
"""
|
|
if (
|
|
self.club.booking_platform.urls is None
|
|
or self.club.booking_platform.urls.get(name) is None
|
|
or self.club.booking_platform.urls.get(name).path is None
|
|
):
|
|
raise ValueError(
|
|
f"The booking platform internal URL path for page {name} of club "
|
|
f"{self.club.name} are not set"
|
|
)
|
|
|
|
def _check_payload_template_exists(self, name: str) -> None:
|
|
"""
|
|
Check that the payload template for the given service is defined
|
|
|
|
:param name: the name of the service
|
|
"""
|
|
if (
|
|
self.club.booking_platform.urls is None
|
|
or self.club.booking_platform.urls.get(name) is None
|
|
or self.club.booking_platform.urls.get(name).path is None
|
|
):
|
|
raise ValueError(
|
|
f"The booking platform internal URL path for page {name} of club "
|
|
f"{self.club.name} are not set"
|
|
)
|
|
|
|
@property
|
|
def landing_url(self) -> str:
|
|
"""
|
|
Get the URL to the landing page of Gestion-Sports
|
|
|
|
:return: the URL to the landing page
|
|
"""
|
|
return self._get_url_path("landing-page")
|
|
|
|
@property
|
|
def login_url(self) -> str:
|
|
"""
|
|
Get the URL to the connection login of Gestion-Sports
|
|
|
|
:return: the URL to the login page
|
|
"""
|
|
return self._get_url_path("login")
|
|
|
|
@property
|
|
def login_template(self) -> Path:
|
|
"""
|
|
Get the payload template to send to log in the website
|
|
|
|
:return: the payload template for logging in
|
|
"""
|
|
return self._get_payload_template("login")
|
|
|
|
@property
|
|
def booking_url(self) -> str:
|
|
"""
|
|
Get the URL to the booking page of Gestion-Sports
|
|
|
|
:return: the URL to the booking page
|
|
"""
|
|
return self._get_url_path("booking")
|
|
|
|
@property
|
|
def booking_template(self) -> Path:
|
|
"""
|
|
Get the payload template to send to book a court
|
|
|
|
:return: the payload template for booking a court
|
|
"""
|
|
return self._get_payload_template("booking")
|
|
|
|
@property
|
|
def user_bookings_url(self) -> str:
|
|
"""
|
|
Get the URL where all the user's bookings are available
|
|
|
|
:return: the URL to the user's bookings
|
|
"""
|
|
return self._get_url_path("user-bookings")
|
|
|
|
@property
|
|
def user_bookings_template(self) -> Path:
|
|
"""
|
|
Get the payload template to send to get all the user's bookings that are
|
|
available
|
|
|
|
:return: the payload template for the user's bookings
|
|
"""
|
|
return self._get_payload_template("user-bookings")
|
|
|
|
@property
|
|
def booking_cancellation_url(self) -> str:
|
|
"""
|
|
Get the URL where all the user's bookings are available
|
|
|
|
:return: the URL to the user's bookings
|
|
"""
|
|
return self._get_url_path("cancellation")
|
|
|
|
@property
|
|
def booking_cancel_template(self) -> Path:
|
|
"""
|
|
Get the payload template to send to get all the user's bookings that are
|
|
available
|
|
|
|
:return: the payload template for the user's bookings
|
|
"""
|
|
return self._get_payload_template("cancellation")
|
|
|
|
@property
|
|
def available_sports(self) -> dict[str, Sport]:
|
|
"""
|
|
Get a dictionary of all sports, the key is the sport name lowered case
|
|
:return: the dictionary of all sports
|
|
"""
|
|
return {
|
|
sport.name.lower(): sport for sport in self.club.booking_platform.sports
|
|
}
|
|
|
|
async def land(self, session: ClientSession) -> ClientResponse:
|
|
"""
|
|
Perform the request to the landing page in order to get the cookie PHPSESSIONID
|
|
|
|
:return: the response from the landing page
|
|
"""
|
|
LOGGER.info("Connecting to GestionSports API at %s", self.login_url)
|
|
async with session.get(self.landing_url) as response:
|
|
await response.text()
|
|
return response
|
|
|
|
async def login(self, session: ClientSession, user: User) -> ClientResponse:
|
|
"""
|
|
Perform the request to the log in the user
|
|
|
|
:return: the response from the login
|
|
"""
|
|
LOGGER.info("Logging in to GestionSports API at %s", self.login_url)
|
|
payload = PayloadBuilder.build(self.login_template, user=user, club=self.club)
|
|
|
|
async with session.post(
|
|
self.login_url, data=payload, headers=POST_HEADERS
|
|
) as response:
|
|
resp_text = await response.text()
|
|
LOGGER.debug("Connexion request response:\n%s", resp_text)
|
|
return response
|
|
|
|
async def book(self, user: User, booking_filter: BookingFilter) -> Court | None:
|
|
"""
|
|
Perform a request for each court at the same time to increase the chances to get
|
|
a booking.
|
|
The gestion-sports backend does not allow several bookings at the same time
|
|
so there is no need to make each request one after the other
|
|
|
|
:param user: the user that wants to book the court
|
|
:param booking_filter: the booking conditions to meet
|
|
:return: the booked court, or None if no court was booked
|
|
"""
|
|
LOGGER.info(
|
|
"Booking any available court from GestionSports API at %s", self.booking_url
|
|
)
|
|
|
|
sport = self.available_sports.get(booking_filter.sport_name)
|
|
|
|
async with ClientSession() as session:
|
|
# use asyncio to request a booking on every court
|
|
# the gestion-sports backend is able to book only one court for a user
|
|
await self.land(session)
|
|
await self.login(session, user)
|
|
self.wait_until_booking_time(booking_filter)
|
|
bookings = await asyncio.gather(
|
|
*[
|
|
self.send_booking_request(
|
|
session, booking_filter.date, court.id, sport.id
|
|
)
|
|
for court in sport.courts
|
|
],
|
|
return_exceptions=True,
|
|
)
|
|
|
|
LOGGER.debug("Booking results:\n'%s'", bookings)
|
|
return self.get_booked_court(bookings, sport.name)
|
|
|
|
async def send_booking_request(
|
|
self,
|
|
session: ClientSession,
|
|
date: DateTime,
|
|
court_id: int,
|
|
sport_id: int,
|
|
) -> tuple[ClientResponse, int, bool]:
|
|
"""
|
|
Book a single court that meets the conditions from the booking filter
|
|
|
|
:param session: the HTTP session that contains the user information and cookies
|
|
:param date: the booking date
|
|
:param court_id: the id of the court to book
|
|
:param sport_id: the id of the sport
|
|
:return: a tuple containing the court id and the booking status
|
|
"""
|
|
LOGGER.debug("Booking court %s at %s", court_id, date.to_w3c_string())
|
|
payload = PayloadBuilder.build(
|
|
self.booking_template,
|
|
date=date,
|
|
court_id=court_id,
|
|
sport_id=sport_id,
|
|
)
|
|
|
|
LOGGER.debug("Booking court %s request:\n'%s'", court_id, payload)
|
|
|
|
async with session.post(
|
|
self.booking_url, data=payload, headers=POST_HEADERS
|
|
) as response:
|
|
assert response.status == 200
|
|
resp_json = await response.text()
|
|
LOGGER.debug("Response from booking request:\n'%s'", resp_json)
|
|
return response, court_id, self.is_booking_response_status_ok(resp_json)
|
|
|
|
def get_booked_court(
|
|
self, bookings: list[tuple[ClientSession, int, bool]], sport_name: str
|
|
) -> Court | None:
|
|
"""
|
|
Parse the booking list and return the court that was booked
|
|
|
|
:param bookings: a list of bookings
|
|
:param sport_name: the sport name
|
|
:return: the id of the booked court if any, None otherwise
|
|
"""
|
|
for _, court_id, is_booked in bookings:
|
|
if is_booked:
|
|
LOGGER.debug("Court %d is booked", court_id)
|
|
court_booked = self.find_court(court_id, sport_name)
|
|
LOGGER.info("Court '%s' is booked", court_booked.name)
|
|
return court_booked
|
|
LOGGER.debug("No booked court found")
|
|
return None
|
|
|
|
def find_court(self, court_id: int, sport_name: str) -> Court:
|
|
"""
|
|
Get all the court information based on the court id and the sport name
|
|
|
|
:param court_id: the court id
|
|
:param sport_name: the sport name
|
|
:return: the court that has the given id and sport name
|
|
"""
|
|
sport = self.available_sports.get(sport_name.lower())
|
|
for court in sport.courts:
|
|
if court.id == court_id:
|
|
return court
|
|
|
|
@staticmethod
|
|
def is_booking_response_status_ok(response: str) -> bool:
|
|
"""
|
|
Check if the booking response is OK
|
|
|
|
:param response: the response as a string
|
|
:return: true if the status is ok, false otherwise
|
|
"""
|
|
formatted_result = response.removeprefix('"').removesuffix('"')
|
|
result_json = json.loads(formatted_result)
|
|
return result_json["status"] == "ok"
|
|
|
|
def build_booking_datetime(self, booking_filter: BookingFilter) -> DateTime:
|
|
"""
|
|
Build the date and time when the booking is open for a given match date.
|
|
The booking filter contains the date and time of the booking.
|
|
The club has the information about when the booking is open for that date.
|
|
|
|
:param booking_filter: the booking information
|
|
:return: the date and time when the booking is open
|
|
"""
|
|
date_to_book = booking_filter.date
|
|
booking_opening = self.club.booking_platform.booking_opening
|
|
booking_date = date_to_book.subtract(days=booking_opening.days_before)
|
|
|
|
opening_time = pendulum.parse(booking_opening.opening_time)
|
|
booking_hour = opening_time.hour
|
|
booking_minute = opening_time.minute
|
|
|
|
return booking_date.at(booking_hour, booking_minute)
|
|
|
|
def wait_until_booking_time(self, booking_filter: BookingFilter) -> None:
|
|
"""
|
|
Wait until the booking is open.
|
|
The booking filter contains the date and time of the booking.
|
|
The club has the information about when the booking is open for that date.
|
|
|
|
:param booking_filter: the booking information
|
|
"""
|
|
LOGGER.info("Waiting for booking time")
|
|
booking_datetime = self.build_booking_datetime(booking_filter)
|
|
now = pendulum.now()
|
|
duration_until_booking = booking_datetime - now
|
|
LOGGER.debug(f"Current time: {now}, Datetime to book: {booking_datetime}")
|
|
LOGGER.debug(
|
|
f"Time to wait before booking: {duration_until_booking.hours:0>2}"
|
|
f":{duration_until_booking.minutes:0>2}"
|
|
f":{duration_until_booking.seconds:0>2}"
|
|
)
|
|
|
|
while now < booking_datetime:
|
|
time.sleep(1)
|
|
now = pendulum.now()
|
|
LOGGER.info("It's booking time!")
|
|
|
|
async def has_user_ongoing_booking(self, user: User) -> bool:
|
|
"""
|
|
Check if the user currently has bookings in the future
|
|
:param user: the user to check the bookings
|
|
:return: true if the user has some bookings, false otherwise
|
|
"""
|
|
async with ClientSession() as session:
|
|
await self.land(session)
|
|
await self.login(session, user)
|
|
return bool(await self.get_ongoing_bookings(session))
|
|
|
|
async def get_ongoing_bookings(self, session: ClientSession) -> list[Booking]:
|
|
"""
|
|
Get the list of all ongoing bookings of a user.
|
|
The steps to perform this are to get the user's bookings page and get a hidden
|
|
property in the HTML to get a hash that will be used in the payload of the
|
|
POST request (sic) to get the user's bookings.
|
|
Gestion sports is really a mess!!
|
|
|
|
:return: the list of all ongoing bookings of a user
|
|
"""
|
|
hash_value = await self.send_hash_request(session)
|
|
LOGGER.debug(f"Hash value: {hash_value}")
|
|
payload = PayloadBuilder.build(self.user_bookings_template, hash=hash_value)
|
|
LOGGER.debug(f"Payload to get ongoing bookings: {payload}")
|
|
return await self.send_user_bookings_request(session, payload)
|
|
|
|
async def send_hash_request(self, session: ClientSession) -> str:
|
|
"""
|
|
Get the hash value used in the request to get the user's bookings
|
|
|
|
:param session: the session in which the user logged in
|
|
:return: the value of the hash
|
|
"""
|
|
async with session.get(self.user_bookings_url) as response:
|
|
html = await response.text()
|
|
LOGGER.debug("Get bookings response: %s\n", html)
|
|
return self.get_hash_input(html)
|
|
|
|
@staticmethod
|
|
def get_hash_input(html_doc: str) -> str:
|
|
"""
|
|
There is a secret hash generated by Gestion sports that is reused when trying to get
|
|
users bookings. This hash is stored in a hidden input with name "mesresas-hash"
|
|
|
|
:param html_doc: the html document when getting the page mes-resas.html
|
|
:return: the value of the hash in the page
|
|
"""
|
|
soup = BeautifulSoup(html_doc, "html.parser")
|
|
inputs = soup.find_all("input")
|
|
for input_tag in inputs:
|
|
if input_tag.get("name") == "mesresas-hash":
|
|
return input_tag.get("value").strip()
|
|
|
|
async def send_user_bookings_request(
|
|
self, session: ClientSession, payload: str
|
|
) -> list[Booking]:
|
|
"""
|
|
Perform the HTTP request to get all bookings
|
|
|
|
:param session: the session in which the user logged in
|
|
:param payload: the HTTP payload for the request
|
|
:return: a dictionary containing all the bookings
|
|
"""
|
|
async with session.post(
|
|
self.user_bookings_url, data=payload, headers=POST_HEADERS
|
|
) as response:
|
|
resp = await response.text()
|
|
LOGGER.debug("ongoing bookings response: %s\n", resp)
|
|
return [Booking(**booking) for booking in json.loads(resp)]
|
|
|
|
async def cancel_booking_id(self, user: User, booking_id: int) -> ClientResponse:
|
|
"""
|
|
Cancel a booking based on its id for a given user
|
|
|
|
:param user: the user that has the booking
|
|
:param booking_id: the id of the booking to cancel
|
|
:return: the response from the client
|
|
"""
|
|
async with ClientSession() as session:
|
|
await self.land(session)
|
|
await self.login(session, user)
|
|
|
|
return await self.send_cancellation_request(session, booking_id)
|
|
|
|
async def send_cancellation_request(
|
|
self, session: ClientSession, booking_id: int
|
|
) -> ClientResponse:
|
|
"""
|
|
Send the HTTP request to cancel the booking
|
|
|
|
:param session: the HTTP session that contains the user information and cookies
|
|
:param booking_id: the id of the booking to cancel
|
|
:return: the response from the client
|
|
"""
|
|
hash_value = await self.send_hash_request(session)
|
|
|
|
payload = PayloadBuilder.build(
|
|
self.booking_cancel_template,
|
|
booking_id=booking_id,
|
|
hash=hash_value,
|
|
)
|
|
|
|
async with session.post(
|
|
self.booking_cancellation_url, data=payload, headers=POST_HEADERS
|
|
) as response:
|
|
await response.text()
|
|
return response
|
|
|
|
async def cancel_booking(
|
|
self, user: User, booking_filter: BookingFilter
|
|
) -> ClientResponse | None:
|
|
"""
|
|
Cancel the booking that meets some conditions
|
|
|
|
:param user: the user who owns the booking
|
|
:param booking_filter: the conditions the booking to cancel should meet
|
|
"""
|
|
async with ClientSession() as session:
|
|
await self.land(session)
|
|
await self.login(session, user)
|
|
|
|
bookings = await self.get_ongoing_bookings(session)
|
|
|
|
for booking in bookings:
|
|
if booking.matches(booking_filter):
|
|
return await self.send_cancellation_request(session, booking.id)
|