345 lines
11 KiB
Python
345 lines
11 KiB
Python
import asyncio
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
import config
|
|
from aiohttp import ClientResponse, ClientSession
|
|
from exceptions import WrongResponseStatus
|
|
from models import BookingFilter, Club, 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 handles all the HTTP requests to the
|
|
Gestion sports 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
|
|
|
|
@property
|
|
def landing_url(self) -> str:
|
|
"""
|
|
Get the URL to for landing to the website
|
|
|
|
:return: the URL to landing
|
|
"""
|
|
return self.club.landing_url
|
|
|
|
@property
|
|
def login_url(self) -> str:
|
|
"""
|
|
Get the URL to for logging in the website
|
|
|
|
:return: the URL for logging in
|
|
"""
|
|
return self.club.login_url
|
|
|
|
@property
|
|
def login_template(self) -> Path:
|
|
"""
|
|
Get the payload template for logging in the website
|
|
|
|
:return: the payload template for logging
|
|
"""
|
|
return self.club.login_template
|
|
|
|
@property
|
|
def booking_url(self) -> str:
|
|
"""
|
|
Get the URL used to book a court
|
|
|
|
:return: the URL to book a court
|
|
"""
|
|
return self.club.booking_url
|
|
|
|
@property
|
|
def booking_template(self) -> Path:
|
|
"""
|
|
Get the payload template for booking a court
|
|
|
|
:return: the payload template for booking a court
|
|
"""
|
|
return self.club.booking_template
|
|
|
|
@property
|
|
def user_bookings_url(self) -> str:
|
|
"""
|
|
Get the URL of the bookings related to a user that are not yet passed
|
|
|
|
:return: the URL to get the bookings related to a user
|
|
"""
|
|
return self.club.user_bookings_url
|
|
|
|
@property
|
|
def user_bookings_template(self) -> Path:
|
|
"""
|
|
Get the payload template to get the bookings related to a user that are not yet
|
|
passed
|
|
|
|
:return: the template for requesting the bookings related to a user
|
|
"""
|
|
return self.club.user_bookings_template
|
|
|
|
@property
|
|
def cancel_url(self) -> str:
|
|
"""
|
|
Get the URL used to cancel a booking
|
|
|
|
:return: the URL to cancel a booking
|
|
"""
|
|
return self.club.cancel_url
|
|
|
|
@property
|
|
def cancel_template(self) -> Path:
|
|
"""
|
|
Get the payload template for cancelling a booking
|
|
|
|
:return: the template for cancelling a booking
|
|
"""
|
|
return self.club.cancel_template
|
|
|
|
@property
|
|
def sessions_url(self) -> str:
|
|
"""
|
|
Get the URL of the session containing all the tournaments
|
|
|
|
:return: the URL to get the session
|
|
"""
|
|
return self.club.sessions_url
|
|
|
|
@property
|
|
def sessions_template(self) -> Path:
|
|
"""
|
|
Get the payload template for requesting the session containing all the
|
|
tournaments
|
|
|
|
:return: the template for requesting the session
|
|
"""
|
|
return self.club.sessions_template
|
|
|
|
@property
|
|
def tournaments_url(self) -> str:
|
|
"""
|
|
Get the URL of all the tournaments list
|
|
|
|
:return: the URL to get the tournaments list
|
|
"""
|
|
return self.club.tournaments_url
|
|
|
|
@property
|
|
def 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 self.club.sports
|
|
|
|
@staticmethod
|
|
def check_response_status(response_status: int) -> None:
|
|
if response_status != 200:
|
|
raise WrongResponseStatus("GestionSports request failed")
|
|
|
|
async def land(self, session: ClientSession) -> ClientResponse:
|
|
"""
|
|
Perform the request to the landing page in order to get the cookie PHPSESSIONID
|
|
|
|
:param session: the client session shared among all connections
|
|
: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:
|
|
self.check_response_status(response.status)
|
|
await response.text()
|
|
return response
|
|
|
|
async def login(self, session: ClientSession, user: User) -> ClientResponse:
|
|
"""
|
|
Perform the request to the log in the user
|
|
|
|
:param session: the client session shared among all connections
|
|
:param user: the user to log in
|
|
: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, allow_redirects=False
|
|
) as response:
|
|
self.check_response_status(response.status)
|
|
resp_text = await response.text()
|
|
LOGGER.debug("Connexion request response:\n%s", resp_text)
|
|
return response
|
|
|
|
async def send_all_booking_requests(
|
|
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 client session shared among all connections
|
|
: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.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 client session shared among all connections
|
|
: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:
|
|
self.check_response_status(response.status)
|
|
resp_json = json.loads(await response.text())
|
|
|
|
LOGGER.debug("Response from booking request:\n'%s'", resp_json)
|
|
return court_id, resp_json
|
|
|
|
async def send_hash_request(self, session: ClientSession) -> ClientResponse:
|
|
"""
|
|
Get the hash value used in some other requests
|
|
|
|
:param session: the client session shared among all connections
|
|
:return: the value of the hash
|
|
"""
|
|
async with session.get(self.user_bookings_url) as response:
|
|
self.check_response_status(response.status)
|
|
html = await response.text()
|
|
LOGGER.debug("Get bookings response: %s\n", html)
|
|
return response
|
|
|
|
async def send_user_bookings_request(
|
|
self, session: ClientSession, hash_value: str
|
|
) -> ClientResponse:
|
|
"""
|
|
Send a request to the platform to get all bookings of a user
|
|
|
|
:param session: the client session shared among all connections
|
|
:param hash_value: the hash value to put in the payload
|
|
:return: a dictionary containing all the bookings
|
|
"""
|
|
|
|
payload = PayloadBuilder.build(self.user_bookings_template, hash=hash_value)
|
|
async with session.post(
|
|
self.user_bookings_url, data=payload, headers=POST_HEADERS
|
|
) as response:
|
|
self.check_response_status(response.status)
|
|
await response.text()
|
|
return response
|
|
|
|
async def send_cancellation_request(
|
|
self, session: ClientSession, booking_id: int, hash_value: str
|
|
) -> ClientResponse:
|
|
"""
|
|
Send the HTTP request to cancel the booking
|
|
|
|
:param session: the client session shared among all connections
|
|
:param booking_id: the id of the booking to cancel
|
|
:return: the response from the client
|
|
"""
|
|
payload = PayloadBuilder.build(
|
|
self.cancel_template,
|
|
booking_id=booking_id,
|
|
hash=hash_value,
|
|
)
|
|
|
|
async with session.post(
|
|
self.cancel_url, data=payload, headers=POST_HEADERS
|
|
) as response:
|
|
self.check_response_status(response.status)
|
|
await response.text()
|
|
return response
|
|
|
|
async def send_session_request(self, session: ClientSession) -> ClientResponse:
|
|
"""
|
|
Send a request to the platform to get the session id
|
|
|
|
:param session: the client session shared among all connections
|
|
:return: a client response containing HTML which has the session id
|
|
"""
|
|
payload = self.sessions_template.read_text()
|
|
|
|
async with session.post(
|
|
self.sessions_url, data=payload, headers=POST_HEADERS
|
|
) as response:
|
|
self.check_response_status(response.status)
|
|
LOGGER.debug("tournament sessions: \n%s", await response.text())
|
|
return response
|
|
|
|
async def send_tournaments_request(
|
|
self, session: ClientSession, session_id: str
|
|
) -> ClientResponse:
|
|
"""
|
|
Send a request to the platform to get the next tournaments
|
|
|
|
:param session: the client session shared among all connections
|
|
:param session_id: the tournaments are grouped in a session
|
|
:return: a client response containing the list of all the nex tournaments
|
|
"""
|
|
final_url = self.tournaments_url + session_id
|
|
LOGGER.debug("Getting tournaments list at %s", final_url)
|
|
async with session.get(final_url) as response:
|
|
self.check_response_status(response.status)
|
|
LOGGER.debug("tournaments: %s\n", await response.text())
|
|
return response
|