Big refactoring.

- clubs, booking platforms and user are now defined in customization files -> there are less environment variables
- the responsibility of the session moved
- booking cancellation is available
This commit is contained in:
Stanislas Jouffroy 2024-03-17 23:57:50 +01:00 committed by stanislas
parent dbda5a158e
commit 0938fb98b7
27 changed files with 3050 additions and 696 deletions

View file

@ -2,62 +2,86 @@ import asyncio
import logging
import config
from gestion_sports.gestion_sports_platform import GestionSportsPlatform
from models import BookingFilter, Club, User
from connectors import Connector, GestionSportsConnector
from models import Action, BookingFilter, Club, Court, User
LOGGER = logging.getLogger(__name__)
async def book(club: Club, user: User, booking_filter: BookingFilter) -> int | None:
def get_connector(club: Club) -> Connector:
if club.booking_platform.id == "gestion-sports":
return GestionSportsConnector(club)
async def book_court(
club: Club, users: list[User], booking_filter: BookingFilter
) -> tuple[Court, User]:
"""
Book a court for a user to a club following a booking filter
Book any court that meets the condition from the filter. IThe user that will make
the booking is chosen among a list of users and should not have any ongoing bookings
:param club: the club where to book a court
:param user: the user information
:param booking_filter: the information related to the booking
:return: the id of the booked court, or None if no court was booked
:param club: the club in which the booking will be made
:param users: the list of users who have an account in the club
:param booking_filter: the conditions the court to book should meet
:return: a tuple containing the court that was booked and the user who made the
booking
"""
async with GestionSportsPlatform(club) as platform:
return await platform.book(user, booking_filter)
connector = get_connector(club)
for user in users:
if not await connector.has_user_ongoing_booking(user):
return await connector.book(user, booking_filter), user
async def get_user_without_booking(club: Club, users: list[User]) -> User | None:
async def cancel_booking(club: Club, user: User, booking_filter: BookingFilter) -> None:
"""
Return the first user who has no booking
Cancel the booking that matches the specified filter
:param club: the club where to book
:param users: the list of users
:return: any user who has no booking
:param club: the club in which the booking was made
:param user: the user who made the booking
:param booking_filter: the conditions to meet to cancel the booking
"""
async with GestionSportsPlatform(club) as platform:
for user in users:
if await platform.user_has_no_ongoing_booking(user):
return user
return None
connector = get_connector(club)
await connector.cancel_booking(user, booking_filter)
def main() -> int | None:
async def cancel_booking_id(club: Club, user: User, booking_id: int) -> None:
"""
Cancel a booking that matches the booking id
:param club: the club in which the booking was made
:param user: the user who made the booking
:param booking_id: the id of the booking to cancel
"""
connector = get_connector(club)
await connector.cancel_booking_id(user, booking_id)
def main() -> tuple[Court, User] | None:
"""
Main function used to book a court
:return: the id of the booked court, or None if no court was booked
"""
booking_filter = config.get_booking_filter()
club = config.get_club()
user = asyncio.run(get_user_without_booking(club, config.get_available_users()))
action = config.get_action()
LOGGER.info(
"Starting booking court of sport %s for user %s at club %s at %s",
booking_filter.sport_id,
user.login,
club.id,
booking_filter.date,
)
court_booked = asyncio.run(book(club, user, booking_filter))
if court_booked:
LOGGER.info(
"Court %s booked successfully at %s", court_booked, booking_filter.date
)
else:
LOGGER.info("Booking did not work")
return court_booked
if action == Action.BOOK:
club = config.get_club()
users = config.get_users(club.id)
booking_filter = config.get_booking_filter()
court_booked, user = asyncio.run(book_court(club, users, booking_filter))
if court_booked:
LOGGER.info(
"Court %s booked successfully at %s for user %s",
court_booked,
booking_filter.date,
user,
)
return court_booked, user
else:
LOGGER.info("Booking did not work")
elif action == Action.CANCEL:
user = config.get_user()
club = config.get_club()
booking_filter = config.get_booking_filter()
asyncio.run(cancel_booking(club, user, booking_filter))

View file

@ -0,0 +1,42 @@
import logging
from aiohttp import ClientSession
from connectors import Connector
from models import BookingFilter, Club, User
LOGGER = logging.getLogger(__name__)
class BookingService:
def __init__(self, club: Club, connector: Connector):
LOGGER.info("Initializing booking service at for club", club.name)
self.club = club
self.connector = connector
self.session: ClientSession | None = None
async def __aenter__(self):
self.session = ClientSession()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.session.close()
async def book(self, user: User, booking_filter: BookingFilter) -> int | None:
"""
Book a court matching the booking filters for a user.
The steps to perform a booking are to go to the landing page, to log in, wait
and for the time when booking is open and then actually book the court
:param user: the user that wants to book a court
:param booking_filter: the booking criteria
:return: the court number if the booking is successful, None otherwise
"""
if self.connector is None:
LOGGER.error("No connection to Gestion Sports is available")
return None
if user is None or booking_filter is None:
LOGGER.error("Not enough information available to book a court")
return None
self.connector.book(user, booking_filter)

View file

@ -6,36 +6,11 @@ from pathlib import Path
import pendulum
import yaml
from dotenv import load_dotenv
from models import BookingFilter, Club, User
from models import Action, BookingFilter, Club, User
load_dotenv()
def get_club() -> Club:
"""
Read the environment variables related to the current club
and build the Club object
:return: the club
"""
club_url = os.environ.get("CLUB_URL")
court_ids_tmp = os.environ.get("COURT_IDS") or ""
court_ids = (
[int(court_id) for court_id in court_ids_tmp.split(",")]
if court_ids_tmp
else []
)
club_id = os.environ.get("CLUB_ID")
booking_open_days_before = int(os.environ.get("BOOKING_OPEN_DAYS_BEFORE", "7"))
booking_opening_time_str = os.environ.get("BOOKING_OPENING_TIME", "00:00")
booking_opening_time = pendulum.parse(booking_opening_time_str)
return Club(
id=club_id,
url=club_url,
courts_ids=court_ids,
booking_open_days_before=booking_open_days_before,
booking_opening_time=booking_opening_time.time(),
)
ROOT_DIR = Path(__file__).parent
def get_booking_filter() -> BookingFilter:
@ -45,11 +20,10 @@ def get_booking_filter() -> BookingFilter:
:return: the club
"""
sport_id_tmp = os.environ.get("SPORT_ID")
sport_id = int(sport_id_tmp) if sport_id_tmp else None
sport_name = os.environ.get("SPORT_NAME")
date_time_tmp = os.environ.get("DATE_TIME")
date_time = pendulum.parse(date_time_tmp) if date_time_tmp else None
return BookingFilter(sport_id=sport_id, date=date_time)
return BookingFilter(sport_name=sport_name.lower(), date=date_time)
def get_user() -> User:
@ -64,25 +38,6 @@ def get_user() -> User:
return User(login=login, password=password)
def get_available_users() -> list[User]:
"""
Read the environment variables to get all the available users in order
to increase the chance of having a user with a free slot for a booking
:return: the list of all users that can book a court
"""
available_users_credentials = os.environ.get("AVAILABLE_USERS_CREDENTIALS")
available_users = [
credential for credential in available_users_credentials.split(",")
]
users = []
for user in available_users:
login, password = user.split(":")
users.append(User(login=login, password=password))
return users
def get_post_headers(platform_id: str) -> dict:
"""
Get the headers for the POST endpoint related to a specific booking platform
@ -102,13 +57,90 @@ def init_log_config():
"""
Read the logging.yaml file to initialize the logging configuration
"""
root_dir = os.path.realpath(os.path.dirname(__file__))
logging_file = root_dir + "/logging.yaml"
logging_file = ROOT_DIR / "logging.yaml"
with open(logging_file, "r") as f:
with logging_file.open(mode="r", encoding="utf-8") as f:
logging_config = yaml.safe_load(f.read())
logging.config.dictConfig(logging_config)
logging.config.dictConfig(logging_config)
ROOT_PATH = Path(__file__).parent.resolve()
RESOURCES_DIR = Path(ROOT_PATH, "resources")
def _build_urls(platform_urls: dict) -> dict:
return {url["name"]: url for url in platform_urls}
def _read_clubs(platforms_data: dict, clubs_data: dict) -> dict[str, Club]:
platforms = {platform["id"]: platform for platform in platforms_data}
for club in clubs_data["clubs"]:
club_platform = club["bookingPlatform"]
platform_id = club_platform["id"]
club_platform["urls"] = _build_urls(platforms[platform_id]["urls"])
return {club["id"]: Club(**club) for club in clubs_data["clubs"]}
def get_clubs():
platforms_file = ROOT_DIR / "resources" / "platforms.yaml"
with platforms_file.open(mode="r", encoding="utf-8") as fp:
platforms_data = yaml.safe_load(fp)
clubs_file = ROOT_DIR / "resources" / "clubs.yaml"
with clubs_file.open(mode="r", encoding="utf-8") as fp:
clubs_data = yaml.safe_load(fp)
return _read_clubs(platforms_data["platforms"], clubs_data)
def get_club() -> Club:
"""
Get the club from an environment variable
:return: the club
"""
club_id = os.environ.get("CLUB_ID")
clubs = get_clubs()
return clubs[club_id]
def read_users(data: dict, club_id: str) -> list[User]:
"""
Deserialize users
:param data: the dictionnary of users
:param club_id: the club id
:return: a list if users from the club
"""
for club in data.get("clubs"):
if club.get("id") == club_id:
return [User(**user) for user in club.get("users")]
def get_users(club_id: str) -> list[User]:
"""
Get a list of users from a club
:param club_id: the club to which the users should have an account
:return: the list of all users for that club
"""
users_file = ROOT_DIR / "resources" / "users.yaml"
with users_file.open(mode="r", encoding="utf-8") as fp:
data = yaml.safe_load(fp)
return read_users(data, club_id)
def get_resources_folder() -> Path:
"""
Compute the path to the resources used by the program
:return: the path to the resources folder
"""
default_resources_folder = Path(__file__).parent / "resources"
return Path(os.environ.get("RESOURCES_FOLDER", default_resources_folder))
def get_action() -> Action:
"""
Get the action to perform from an environment variable
:return: the action to perform
"""
return Action(os.environ.get("ACTION"))

598
resa_padel/connectors.py Normal file
View file

@ -0,0 +1,598 @@
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.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_cancellation_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_cancellation_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) -> 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 self.is_booking_matching_filter(booking, booking_filter):
await self.send_cancellation_request(session, booking.id)
def is_booking_matching_filter(
self, booking: Booking, booking_filter: BookingFilter
) -> bool:
"""
Check if the booking matches the booking filter
:param booking: the booking to be checked
:param booking_filter: the conditions the booking should meet
:return: true if the booking matches the conditions, false otherwise
"""
return (
self._is_same_sport(booking, booking_filter)
and self._is_date_matching(booking, booking_filter)
and self._is_time_matching(booking, booking_filter)
)
@staticmethod
def _is_same_sport(booking: Booking, booking_filter: BookingFilter) -> bool:
"""
Check if the booking and the booking filter are about the same sport
:param booking: the booking to be checked
:param booking_filter: the conditions the booking should meet
:return: true if the booking sport matches the filter sport, false otherwise
"""
return booking.sport == booking_filter.sport_name
@staticmethod
def _is_date_matching(booking: Booking, booking_filter: BookingFilter) -> bool:
"""
Check if the booking and the booking filter are at the same date
:param booking: the booking to be checked
:param booking_filter: the conditions the booking should meet
:return: true if the booking date matches the filter date, false otherwise
"""
return booking.booking_date.date() == booking_filter.date.date()
@staticmethod
def _is_time_matching(booking: Booking, booking_filter: BookingFilter) -> bool:
"""
Check if the booking and the booking filter are at the same time
:param booking: the booking to be checked
:param booking_filter: the conditions the booking should meet
:return: true if the booking time matches the filter time, false otherwise
"""
return booking.start_time.time() == booking_filter.date.time()

View file

@ -1,9 +0,0 @@
from pathlib import Path
import config
RESOURCES_DIR = Path(config.RESOURCES_DIR, "gestion-sports")
BOOKING_TEMPLATE = Path(RESOURCES_DIR, "booking-payload.txt")
LOGIN_TEMPLATE = Path(RESOURCES_DIR, "login-payload.txt")
USERS_BOOKINGS_TEMPLATE = Path(RESOURCES_DIR, "users_bookings.txt")

View file

@ -1,199 +0,0 @@
import asyncio
import json
import logging
from urllib.parse import urljoin
import config
from aiohttp import ClientResponse, ClientSession
from gestion_sports import gestion_sports_html_parser as html_parser
from gestion_sports.payload_builders import (
GestionSportsBookingPayloadBuilder,
GestionSportsLoginPayloadBuilder,
GestionSportsUsersBookingsPayloadBuilder,
)
from models import BookingFilter, Club, User
LOGGER = logging.getLogger(__name__)
POST_HEADERS = config.get_post_headers("gestion-sports")
class GestionSportsConnector:
"""
Handle the specific booking requests to Gestion-Sports
"""
def __init__(self, session: ClientSession, url: str):
LOGGER.info("Initializing connection to GestionSports API")
self.url = url
self.session = session
@property
def landing_url(self) -> str:
"""
Get the URL to the landing page of Gestion-Sports
:return: the URL to the landing page
"""
return urljoin(self.url, "/connexion.php")
@property
def login_url(self) -> str:
"""
Get the URL to the connection login of Gestion-Sports
:return: the URL to the login page
"""
return urljoin(self.url, "/connexion.php")
@property
def booking_url(self) -> str:
"""
Get the URL to the booking page of Gestion-Sports
:return: the URL to the booking page
"""
return urljoin(self.url, "/membre/reservation.html")
@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 urljoin(self.url, "/membre/mesresas.html")
async def land(self) -> 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 self.session.get(self.landing_url) as response:
await response.text()
return response
async def login(self, user: User, club: Club) -> 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_builder = GestionSportsLoginPayloadBuilder()
payload = payload_builder.user(user).club(club).build()
async with self.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, booking_filter: BookingFilter, club: Club) -> int | 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 booking_filter: the booking information
:param club: the club where to book the court
: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
)
# use asyncio to request a booking on every court
# the gestion-sports backend is able to book only one court for a user
bookings = await asyncio.gather(
*[
self.book_one_court(booking_filter, court_id)
for court_id in club.courts_ids
],
return_exceptions=True,
)
LOGGER.debug("Booking results:\n'%s'", bookings)
return self.get_booked_court(bookings)
async def book_one_court(
self, booking_filter: BookingFilter, court_id: int
) -> tuple[int, bool]:
"""
Book a single court according to the information provided in the booking filter
:param booking_filter: the booking information
:param court_id: the id of the court to book
:return: a tuple containing the court id and the booking status
"""
LOGGER.debug(
"Booking court %s at %s",
court_id,
booking_filter.date.to_w3c_string(),
)
payload_builder = GestionSportsBookingPayloadBuilder()
payload = (
payload_builder.booking_filter(booking_filter).court_id(court_id).build()
)
LOGGER.debug("Booking court %s request:\n'%s'", court_id, payload)
async with self.session.post(
self.booking_url, data=payload, headers=POST_HEADERS
) as response:
resp_json = await response.text()
LOGGER.debug("Response from booking request:\n'%s'", resp_json)
return court_id, self.is_booking_response_status_ok(resp_json)
@staticmethod
def get_booked_court(bookings: list[tuple[int, bool]]) -> int | None:
"""
Parse the booking list and return the court that was booked
:param bookings: a list of bookings
:return: the id of the booked court if any, None otherwise
"""
for court, is_booked in bookings:
if is_booked:
LOGGER.debug("Court %s is booked", court)
return court
LOGGER.debug("No booked court found")
return None
@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"
async def get_ongoing_bookings(self) -> dict:
"""
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
"""
async with self.session.get(self.user_bookings_url) as get_resp:
html = await get_resp.text()
hash_value = html_parser.get_hash_input(html)
payload_builder = GestionSportsUsersBookingsPayloadBuilder()
payload_builder.hash(hash_value)
payload = payload_builder.build()
async with self.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 json.loads(resp)

View file

@ -1,16 +0,0 @@
from bs4 import BeautifulSoup
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()

View file

@ -1,116 +0,0 @@
import logging
import time
import pendulum
from aiohttp import ClientSession
from gestion_sports.gestion_sports_connector import GestionSportsConnector
from models import BookingFilter, Club, User
from pendulum import DateTime
LOGGER = logging.getLogger(__name__)
class GestionSportsPlatform:
def __init__(self, club: Club):
LOGGER.info("Initializing Gestion Sports platform at url %s", club.url)
self.connector: GestionSportsConnector | None = None
self.club: Club = club
self.session: ClientSession | None = None
async def __aenter__(self):
self.session = ClientSession()
self.connector = GestionSportsConnector(self.session, self.club.url)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.session.close()
async def book(self, user: User, booking_filter: BookingFilter) -> int | None:
"""
Book a court matching the booking filters for a user.
The steps to perform a booking are to go to the landing page, to log in, wait
and for the time when booking is open and then actually book the court
:param user: the user that wants to book a court
:param booking_filter: the booking criteria
:return: the court number if the booking is successful, None otherwise
"""
if self.connector is None:
LOGGER.error("No connection to Gestion Sports is available")
return None
if user is None or booking_filter is None:
LOGGER.error("Not enough information available to book a court")
return None
await self.connector.land()
await self.connector.login(user, self.club)
wait_until_booking_time(self.club, booking_filter)
return await self.connector.book(booking_filter, self.club)
async def user_has_no_ongoing_booking(self, user: User) -> bool:
"""
Check if the user has any ongoing booking.
The steps to perform this task are to go to the landing page, to log in and
then retrieve user information and extract the ongoing bookings
:param user: the user to check the bookings
:return: True if the user has ongoing bookings, false otherwise
"""
if self.connector is None:
LOGGER.error("No connection to Gestion Sports is available")
return False
if user is None:
LOGGER.error("Not enough information available to book a court")
return False
await self.connector.land()
await self.connector.login(user, self.club)
bookings = await self.connector.get_ongoing_bookings()
return bookings == []
def wait_until_booking_time(club: Club, booking_filter: BookingFilter):
"""
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 club: the club where to book a court
:param booking_filter: the booking information
"""
LOGGER.info("Waiting for booking time")
booking_datetime = build_booking_datetime(booking_filter, club)
now = pendulum.now()
duration_until_booking = booking_datetime - now
LOGGER.debug(
"Time to wait before booking: %s:%s:%s",
"{:0>2}".format(duration_until_booking.hours),
"{:0>2}".format(duration_until_booking.minutes),
"{:0>2}".format(duration_until_booking.seconds),
)
while now < booking_datetime:
time.sleep(1)
now = pendulum.now()
LOGGER.info("It's booking time!")
def build_booking_datetime(booking_filter: BookingFilter, club: Club) -> 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
:param club: the club where to book a court
:return: the date and time when the booking is open
"""
date_to_book = booking_filter.date
booking_date = date_to_book.subtract(days=club.booking_open_days_before)
booking_hour = club.booking_opening_time.hour
booking_minute = club.booking_opening_time.minute
return booking_date.at(booking_hour, booking_minute)

View file

@ -1,128 +0,0 @@
from exceptions import ArgumentMissing
from gestion_sports.gestion_sports_config import (
BOOKING_TEMPLATE,
LOGIN_TEMPLATE,
USERS_BOOKINGS_TEMPLATE,
)
from jinja2 import Environment, FileSystemLoader
from models import BookingFilter, Club, User
class GestionSportsLoginPayloadBuilder:
"""
Build the payload for the login page
"""
def __init__(self):
self._user: User | None = None
self._club: Club | None = None
def user(self, user: User):
"""
Set the user
:param user: the user
:return: the class itself
"""
self._user = user
return self
def club(self, club: Club):
"""
Set the club
:param club: the club
:return: the class itself
"""
self._club = club
return self
def build(self) -> str:
"""
Build the payload
:return: the string representation of the payload
"""
if self._user is None:
raise ArgumentMissing("No user was provided")
if self._club is None:
raise ArgumentMissing("No club was provided")
environment = Environment(loader=FileSystemLoader(LOGIN_TEMPLATE.parent))
template = environment.get_template(LOGIN_TEMPLATE.name)
return template.render(club=self._club, user=self._user)
class GestionSportsBookingPayloadBuilder:
def __init__(self):
self._booking_filter: BookingFilter | None = None
self._court_id: int | None = None
def booking_filter(self, booking_filter: BookingFilter):
"""
Set the booking filter
:param booking_filter: the booking filter
:return: the class itself
"""
self._booking_filter = booking_filter
return self
def court_id(self, court_id: int):
"""
Set the court id
:param court_id: the court id
:return: the class itself
"""
self._court_id = court_id
return self
def build(self) -> str:
"""
Build the payload
:return: the string representation of the payload
"""
if self._booking_filter is None:
raise ArgumentMissing("No booking filter was provided")
if self.court_id is None:
raise ArgumentMissing("No court id was provided")
environment = Environment(loader=FileSystemLoader(BOOKING_TEMPLATE.parent))
template = environment.get_template(BOOKING_TEMPLATE.name)
return template.render(
court_id=self._court_id, booking_filter=self._booking_filter
)
class GestionSportsUsersBookingsPayloadBuilder:
def __init__(self):
self._hash: str | None = None
def hash(self, hash_value: str):
"""
Set the hash
:param hash_value: the hash
:return: the class itself
"""
self._hash = hash_value
def build(self) -> str:
"""
Build the payload
:return: the string representation of the payload
"""
if self._hash is None:
raise ArgumentMissing("No hash was provided")
environment = Environment(
loader=FileSystemLoader(USERS_BOOKINGS_TEMPLATE.parent)
)
template = environment.get_template(USERS_BOOKINGS_TEMPLATE.name)
return template.render(hash=self._hash)

View file

@ -1,22 +1,171 @@
from pendulum import Time
from pydantic import BaseModel, ConfigDict, Field
from enum import Enum
from typing import Optional
import pendulum
from pydantic import BaseModel, ConfigDict, Field, field_validator
from pydantic_extra_types.pendulum_dt import DateTime
class Club(BaseModel):
class User(BaseModel):
login: str
password: str = Field(repr=False)
club_id: Optional[str] = Field(default=None)
class BookingOpening(BaseModel):
model_config = ConfigDict(arbitrary_types_allowed=True)
id: str = Field()
url: str = Field()
courts_ids: list[int] = Field(default_factory=list)
booking_open_days_before: int = Field(default=7)
booking_opening_time: Time = Field(default=Time(hour=0, minute=0))
days_before: Optional[int] = Field(default=7, alias="daysBefore")
opening_time: Optional[str] = Field(alias="time", default=None, repr=False)
time_after_booking: Optional[str] = Field(
alias="timeAfterBookingTime", default=None, repr=False
)
def __repr__(self):
base = super().__repr__()
time = f", time: {self.time})" if self.time else ""
time_after_booking = (
f", time_after_booking_time: {self.time_after_booking_time})"
if self.time_after_booking_time
else ""
)
return base.removesuffix(")") + time + time_after_booking
@property
def time(self):
return pendulum.parse(self.opening_time).time()
@property
def time_after_booking_time(self):
return (
pendulum.parse(self.time_after_booking).time()
if self.time_after_booking
else None
)
class TotalBookings(BaseModel):
peak_hours: int | str = Field(alias="peakHours")
off_peak_hours: int | str = Field(alias="offPeakHours")
class Court(BaseModel):
id: int
name: str
number: int
is_indoor: Optional[bool] = Field(alias="isIndoor")
class Sport(BaseModel):
name: str
id: int
duration: int
price: int
players: int
courts: list[Court]
class Url(BaseModel):
name: str
path: str
payload_template: Optional[str] = Field(default=None, alias="payloadTemplate")
class BookingPlatform(BaseModel):
id: str
club_id: int = Field(alias="clubId")
url: str
hours_before_cancellation: int = Field(alias="hoursBeforeCancellation")
booking_opening: BookingOpening = Field(alias="bookingOpening")
total_bookings: TotalBookings = Field(alias="totalBookings")
sports: list[Sport]
urls: dict[str, Url]
class Club(BaseModel):
id: str
name: str
url: str
booking_platform: BookingPlatform = Field(alias="bookingPlatform")
class PlatformDefinition(BaseModel):
id: str
name: str
url: str
urls: list[Url]
class BookingFilter(BaseModel):
sport_id: int = Field()
date: DateTime = Field()
date: DateTime
sport_name: str
@field_validator("sport_name", mode="before")
@classmethod
def to_lower_case(cls, d: str) -> str:
return d.lower()
class User(BaseModel):
login: str = Field()
password: str = Field(repr=False)
class Booking(BaseModel):
id: int
booking_date: DateTime = Field(alias="dateResa")
start_time: DateTime = Field(alias="startTime")
sport: str
court: str
game_creation: Optional[int] = Field(default=None, alias="creaPartie")
game_creation_limit: Optional[DateTime] = Field(
default=None, alias="limitCreaPartie"
)
cancel: Optional[bool] = Field(default=True)
block_player_replacement: Optional[int] = Field(
default=None, alias="bloquerRemplacementJoueur"
)
can_remove_parteners: bool = Field(default=True, alias="canRemovePartners")
end_time: Optional[DateTime] = Field(default=None, alias="endTime")
day_fr: Optional[str] = Field(default=None, alias="dayFr")
live_xperience_code: Optional[str] = Field(default=None, alias="codeLiveXperience")
spartime_qr_code: Optional[str] = Field(default=None, alias="qrCodeSpartime")
remaining_places: int = Field(default=3, alias="remainingPlaces")
is_captain: bool = Field(default=True, alias="isCaptain")
dt_start: Optional[DateTime] = Field(default=None, alias="dtStart")
credit_card_guaranty: Optional[str] = Field(default=None, alias="garantieCb")
certificate_validity_duration: Optional[int] = Field(
alias="dureeValidCertif", default=None
)
charge_id: Optional[str] = Field(default=None, alias="chargeId")
partners: Optional[list] = Field(default=[])
player_status: Optional[int] = Field(default=None, alias="playerStatus")
products: Optional[list] = Field(default=[])
@field_validator("booking_date", mode="before")
@classmethod
def validate_date(cls, d: str) -> DateTime:
return pendulum.from_format(
d, "DD/MM/YYYY", tz=pendulum.timezone("Europe/Paris")
)
@field_validator("start_time", "end_time", mode="before")
@classmethod
def validate_time(cls, t: str) -> DateTime:
return pendulum.from_format(t, "HH:mm", tz=pendulum.timezone("Europe/Paris"))
@field_validator("game_creation_limit", mode="before")
@classmethod
def validate_datetime_add_tz(cls, dt: str) -> DateTime:
return pendulum.parse(dt, tz=pendulum.timezone("Europe/Paris"))
@field_validator("dt_start", mode="before")
@classmethod
def validate_datetime(cls, dt: str) -> DateTime:
return pendulum.parse(dt)
@field_validator("sport", mode="before")
@classmethod
def to_lower_case(cls, d: str) -> str:
return d.lower()
class Action(Enum):
BOOK = "book"
CANCEL = "cancel"

View file

@ -0,0 +1,12 @@
from pathlib import Path
from jinja2 import Environment, FileSystemLoader
class PayloadBuilder:
@staticmethod
def build(template_path: Path, **kwargs) -> str:
environment = Environment(loader=FileSystemLoader(template_path.parent))
template = environment.get_template(template_path.name)
return template.render(**kwargs)

View file

@ -0,0 +1,160 @@
clubs:
- name: Toulouse Padel Club
url: https://www.toulousepadelclub.com
id: tpc
bookingPlatform:
id: gestion-sports
clubId: 88
url: https://toulousepadelclub.gestion-sports.com
hoursBeforeCancellation: 4
bookingOpening:
daysBefore: 7
time: 00:00
totalBookings:
peakHours: 1
offPeakHours: unlimited
sports:
- name: Padel
id: 217
duration: 90
price: 48
players: 4
courts:
- name: Court 1
number: 1
id: 598
isIndoor: True
- name: Court 2
number: 2
id: 599
isIndoor: True
- name: Court 3
number: 3
id: 600
isIndoor: True
- name: Court 4
number: 4
id: 601
isIndoor: True
- name: Court 5
number: 5
id: 602
isIndoor: True
- name: Court 6
number: 6
id: 603
isIndoor: True
- name: Court 7
number: 7
id: 604
isIndoor: True
- name: Court 8
number: 8
id: 605
isIndoor: True
- name: Court 9
number: 9
id: 606
isIndoor: True
- name: Court 10
number: 10
id: 607
isIndoor: True
- name: Court 11
number: 11
id: 608
isIndoor: True
- name: Court 12
number: 12
id: 609
isIndoor: True
- name: Court 13
number: 13
id: 610
isIndoor: True
- name: Court 14
number: 14
id: 611
isIndoor: True
- name: Squash
id: 218
duration: 45
price: 18
players: 2
courts:
- name: Court 1
id: 613
number: 1
isIndoor: True
- name: Court 2
number: 2
id: 614
isIndoor: True
- name: Court 3
number: 3
id: 615
isIndoor: True
- name: Court 4
number: 4
id: 616
isIndoor: True
- name: Padel Tolosa
url: https://www.padeltolosa.fr/
id: padeltolosa
bookingPlatform:
id: gestion-sports
clubId: 89
url: https://padeltolosa.gestion-sports.com/
hoursBeforeCancellation: 24
bookingOpening:
daysBefore: 7
time: 00:00
totalBookings:
peakHours: 4
offPeakHours: unlimited
sports:
- name: Padel
id: 262
duration: 90
price: 48
players: 4
courts:
- name: Court 1 M.F IMMOBILLIER
number: 1
id: 746
isIndoor: True
- name: Court 2 PAQUITO
number: 2
id: 747
isIndoor: True
- name: Court 3 Seven Sisters PUB TOULOUSE
number: 3
id: 748
isIndoor: True
- name: Court MADRID
number: 4
id: 749
isIndoor: True
- name: Squash
id: 218
duration: 45
price: 18
players: 2
courts:
- name: Court 1
id: 613
isIndoor: True
number: 1
- name: Court 2
id: 614
isIndoor: True
number: 2
- name: Court 3
id: 615
isIndoor: True
number: 3
- name: Court 4
id: 616
isIndoor: True
number: 4

View file

@ -0,0 +1 @@
ajax=removeResa&hash={{ hash }}&id={{ booking_id }}

View file

@ -1 +1 @@
ajax=addResa&date={{ booking_filter.date.date().strftime("%d/%m/%Y") }}&hour={{ booking_filter.date.time().strftime("%H:%M") }}&duration=90&partners=null|null|null&paiement=facultatif&idSport={{ booking_filter.sport_id }}&creaPartie=false&idCourt={{ court_id }}&pay=false&token=undefined&totalPrice=44&saveCard=0&foodNumber=0
ajax=addResa&date={{ date.date().strftime("%d/%m/%Y") }}&hour={{ date.time().strftime("%H:%M") }}&duration=90&partners=null|null|null&paiement=facultatif&idSport={{ sport_id }}&creaPartie=false&idCourt={{ court_id }}&pay=false&token=undefined&totalPrice=48&saveCard=0&foodNumber=0

View file

@ -1 +1 @@
ajax=connexionUser&id_club={{ club.id }}&email={{ user.login }}&form_ajax=1&pass={{ user.password }}&compte=user&playeridonesignal=0&identifiant=identifiant&externCo=true
ajax=connexionUser&id_club={{ club.booking_platform.club_id }}&email={{ user.login }}&form_ajax=1&pass={{ user.password }}&compte=user&playeridonesignal=0&identifiant=identifiant&externCo=true

View file

@ -0,0 +1,19 @@
platforms:
- name: Gestion sports
url: https://gestion-sports.fr/
id: gestion-sports
urls:
- name: landing-page
path: /connexion.php
- name: login
path: /connexion.php
payloadTemplate: gestion-sports/login-payload.txt
- name: booking
path: /membre/reservation.html
payloadTemplate: gestion-sports/booking-payload.txt
- name: user-bookings
path: /membre/mesresas.html
payloadTemplate: gestion-sports/user-bookings-payload.txt
- name: cancellation
path: /membre/mesresas.html
payloadTemplate: gestion-sports/booking-cancellation-payload.txt

View file

@ -0,0 +1,13 @@
clubs:
- id: tpc
users:
- login: padel.testing@jouf.fr
password: ridicule
- login: mateochal31@gmail.com
password: pleanyakis
- id: padeltolosa
users:
- login: padel.testing@jouf.fr
password: ridicule
- login: mateochal31@gmail.com
password: pleanyakis