300 lines
11 KiB
Python
300 lines
11 KiB
Python
import json
|
|
import logging
|
|
import time
|
|
|
|
import pendulum
|
|
from aiohttp import ClientSession
|
|
from bs4 import BeautifulSoup
|
|
from gestion_sport_connector import GestionSportsConnector
|
|
from models import (
|
|
Booking,
|
|
BookingFilter,
|
|
BookingOpening,
|
|
Club,
|
|
Court,
|
|
Sport,
|
|
Tournament,
|
|
User,
|
|
)
|
|
from pendulum import DateTime
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
class GestionSportsServices:
|
|
async def book(
|
|
self, club: Club, 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 club: the club in which the booking will be made
|
|
: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
|
|
"""
|
|
connector = GestionSportsConnector(club)
|
|
LOGGER.info(
|
|
"Booking any available court from GestionSports API at %s",
|
|
connector.booking_url,
|
|
)
|
|
|
|
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 connector.land(session)
|
|
await connector.login(session, user)
|
|
|
|
booking_opening = club.booking_platform.booking_opening
|
|
GestionSportsServices.wait_until_booking_time(
|
|
booking_filter, booking_opening
|
|
)
|
|
|
|
bookings = await connector.send_all_booking_requests(
|
|
session, booking_filter
|
|
)
|
|
|
|
LOGGER.debug("Booking results:\n'%s'", bookings)
|
|
|
|
sport = club.sports.get(booking_filter.sport_name)
|
|
|
|
return self.get_booked_court(bookings, sport)
|
|
|
|
def get_booked_court(
|
|
self, bookings: list[tuple[int, dict]], sport: Sport
|
|
) -> Court | None:
|
|
"""
|
|
Parse the booking list and return the court that was booked
|
|
|
|
:param bookings: a list of bookings
|
|
:param sport: the sport of the club and all the courts it has
|
|
: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)
|
|
LOGGER.info("Court '%s' is booked", court_booked.name)
|
|
return court_booked
|
|
LOGGER.debug("No booked court found")
|
|
return None
|
|
|
|
@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"
|
|
|
|
@staticmethod
|
|
def find_court(court_id: int, sport: Sport) -> Court:
|
|
"""
|
|
Get all the court information based on the court id and the sport name
|
|
|
|
:param court_id: the court id
|
|
:param sport: the sport
|
|
:return: the court that has the given id and sport name
|
|
"""
|
|
for court in sport.courts:
|
|
if court.id == court_id:
|
|
return court
|
|
|
|
async def has_user_available_slots(self, user: User, club: Club) -> bool:
|
|
"""
|
|
Checks if a user has available booking slot.
|
|
If a user already has an ongoing booking, it is considered as no slot is
|
|
available
|
|
|
|
:param user: The user to check the booking availability
|
|
:param club: The club of the user
|
|
:return: True if the user has no ongoing booking, False otherwise
|
|
"""
|
|
connector = GestionSportsConnector(club)
|
|
async with ClientSession() as session:
|
|
await connector.land(session)
|
|
await connector.login(session, user)
|
|
bookings = await self.get_ongoing_bookings(session, connector)
|
|
|
|
return bool(bookings)
|
|
|
|
@staticmethod
|
|
def wait_until_booking_time(
|
|
booking_filter: BookingFilter, booking_opening: BookingOpening
|
|
) -> 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_opening:
|
|
:param booking_filter: the booking information
|
|
"""
|
|
LOGGER.info("Waiting for booking time")
|
|
booking_datetime = GestionSportsServices.build_booking_datetime(
|
|
booking_filter, booking_opening
|
|
)
|
|
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!")
|
|
|
|
@staticmethod
|
|
def build_booking_datetime(
|
|
booking_filter: BookingFilter, booking_opening: BookingOpening
|
|
) -> 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_opening:the booking opening conditions
|
|
:param booking_filter: the booking information
|
|
:return: the date and time when the booking is open
|
|
"""
|
|
date_to_book = booking_filter.date
|
|
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)
|
|
|
|
async def cancel_booking(
|
|
self, user: User, club: Club, booking_filter: BookingFilter
|
|
):
|
|
connector = GestionSportsConnector(club)
|
|
async with ClientSession() as session:
|
|
await connector.land(session)
|
|
await connector.login(session, user)
|
|
|
|
bookings = await self.get_ongoing_bookings(session, connector)
|
|
for booking in bookings:
|
|
if booking.matches(booking_filter):
|
|
return await self.cancel_booking_id(session, connector, booking.id)
|
|
|
|
async def get_ongoing_bookings(
|
|
self, session: ClientSession, connector: GestionSportsConnector
|
|
) -> 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!!
|
|
|
|
:param session: the client session shared among all connections
|
|
:param connector: the connector used to send the requests
|
|
:return: the list of all ongoing bookings of a user
|
|
"""
|
|
response = await connector.send_hash_request(session)
|
|
hash_value = self.get_hash_input(await response.text())
|
|
LOGGER.debug(f"Hash value: {hash_value}")
|
|
response = await connector.send_user_bookings_request(session, hash_value)
|
|
return [Booking(**booking) for booking in json.loads(await response.text())]
|
|
|
|
@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 cancel_booking_id(
|
|
self, session: ClientSession, connector: GestionSportsConnector, booking_id: int
|
|
) -> None:
|
|
"""
|
|
Send the HTTP request to cancel the booking
|
|
|
|
:param session: the client session shared among all connections
|
|
:param connector: the connector used to send the requests
|
|
:param booking_id: the id of the booking to cancel
|
|
:return: the response from the client
|
|
"""
|
|
response = await connector.send_hash_request(session)
|
|
hash_value = self.get_hash_input(await response.text())
|
|
LOGGER.debug(f"Hash value: {hash_value}")
|
|
|
|
await connector.send_cancellation_request(session, booking_id, hash_value)
|
|
|
|
@staticmethod
|
|
async def get_all_tournaments(user: User, club: Club) -> list[Tournament]:
|
|
connector = GestionSportsConnector(club)
|
|
async with ClientSession() as session:
|
|
await connector.land(session)
|
|
await connector.login(session, user)
|
|
|
|
session_html = await connector.send_session_request(session)
|
|
tournaments_id = GestionSportsServices.retrieve_tournament_session(
|
|
await session_html.text()
|
|
)
|
|
|
|
tournaments = await connector.send_tournaments_request(
|
|
session, tournaments_id
|
|
)
|
|
return GestionSportsServices.retrieve_tournaments(await tournaments.text())
|
|
|
|
@staticmethod
|
|
def retrieve_tournament_session(sessions: str) -> str:
|
|
session_object = json.loads(sessions).get("Inscription tournois:school-outline")
|
|
return list(session_object.keys())[0]
|
|
|
|
@staticmethod
|
|
def retrieve_tournaments(html: str) -> list[Tournament]:
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
tournaments = []
|
|
|
|
cards = soup.find_all("div", {"class": "card-body"})
|
|
for card in cards:
|
|
title = card.find("h5")
|
|
price = title.find("span").get_text().strip()
|
|
name = title.get_text().strip().removesuffix(price).strip()
|
|
elements = card.find("div", {"class": "row"}).find_all("li")
|
|
date = elements[0].get_text().strip()
|
|
start_time, end_time = (
|
|
elements[2].get_text().strip().replace("h", ":").split(" - ")
|
|
)
|
|
start_datetime = pendulum.from_format(
|
|
f"{date} {start_time}", "DD/MM/YYYY HH:mm"
|
|
)
|
|
end_datetime = pendulum.from_format(
|
|
f"{date} {end_time}", "DD/MM/YYYY HH:mm"
|
|
)
|
|
gender = elements[1].get_text().strip()
|
|
places_left = (
|
|
card.find("span", {"class": "nb_place_libre"}).get_text().strip()
|
|
)
|
|
tournament = Tournament(
|
|
name=name,
|
|
price=price,
|
|
start_date=start_datetime,
|
|
end_date=end_datetime,
|
|
gender=gender,
|
|
places_left=places_left,
|
|
)
|
|
tournaments.append(tournament)
|
|
|
|
return tournaments
|