423 lines
14 KiB
Python
423 lines
14 KiB
Python
import asyncio
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from urllib.parse import urljoin
|
|
|
|
import config
|
|
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 GestionSportsConnector:
|
|
"""
|
|
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_any_court(
|
|
self, session: ClientSession, booking_filter: BookingFilter
|
|
) -> list[tuple[int, dict]]:
|
|
"""
|
|
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 session: the session to use
|
|
: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)
|
|
|
|
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 bookings
|
|
|
|
async def send_booking_request(
|
|
self,
|
|
session: ClientSession,
|
|
date: DateTime,
|
|
court_id: int,
|
|
sport_id: int,
|
|
) -> tuple[int, dict]:
|
|
"""
|
|
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 response
|
|
"""
|
|
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 = json.loads(await response.text())
|
|
LOGGER.debug("Response from booking request:\n'%s'", resp_json)
|
|
return court_id, resp_json
|
|
|
|
def get_booked_court(
|
|
self, bookings: list[tuple[int, dict]], 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, response in bookings:
|
|
if self.is_booking_response_status_ok(response):
|
|
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: dict) -> bool:
|
|
"""
|
|
Check if the booking response is OK
|
|
|
|
:param response: the response as a string
|
|
:return: true if the status is ok, false otherwise
|
|
"""
|
|
return response["status"] == "ok"
|
|
|
|
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, 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, session: ClientSession, booking_filter: BookingFilter
|
|
) -> ClientResponse | None:
|
|
"""
|
|
Cancel the booking that meets some conditions
|
|
|
|
:param session: the session
|
|
:param booking_filter: the conditions the booking to cancel should meet
|
|
"""
|
|
bookings = await self.get_ongoing_bookings(session)
|
|
|
|
for booking in bookings:
|
|
if booking.matches(booking_filter):
|
|
return await self.cancel_booking_id(session, booking.id)
|