stanislas/resa-padel
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Added some logs

Browse source
This commit is contained in:
Stanislas Jouffroy 2024-02-22 20:48:55 +01:00
parent 552db2aa8a
commit a8322d6be0
8 changed files with 140 additions and 72 deletions
Download patch file Download diff file Expand all files Collapse all files

2
resa_padel/booking.py
View file

@ -32,7 +32,7 @@ def main() -> int | None:
club = config.get_club()
LOGGER.info(
"Starting booking court of %s for user %s at club %s at %s",
"Starting booking court of sport %s for user %s at club %s at %s",
booking_filter.sport_id,
user.login,
club.id,

51
resa_padel/gestion_sports/gestion_sports_connector.py
View file

@ -58,7 +58,7 @@ class GestionSportsConnector:
:return: the response from the landing page
"""
LOGGER.info("Connecting to GestionSports API")
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
@ -69,18 +69,21 @@ class GestionSportsConnector:
: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:
await response.text()
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.
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
@ -88,6 +91,9 @@ class GestionSportsConnector:
: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(
@ -98,36 +104,51 @@ class GestionSportsConnector:
return_exceptions=True,
)
LOGGER.debug("Booking results:\n'%s'", bookings)
return self.get_booked_court(bookings)
@staticmethod
def get_booked_court(bookings):
LOGGER.info(bookings)
for court, is_booked in bookings:
if is_booked:
return court
return None
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.info(payload)
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_response_status_ok(resp_json)
@staticmethod
def is_response_status_ok(response: str) -> bool:
def get_booked_court(bookings):
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
LOGGER.info(response)
@staticmethod
def is_response_status_ok(response: str) -> bool:
formatted_result = response.removeprefix('"').removesuffix('"')
result_json = json.loads(formatted_result)
return result_json["status"] == "ok"
def get_user_information(self):
pass

26
resa_padel/gestion_sports/gestion_sports_platform.py
View file

@ -12,6 +12,7 @@ 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
@ -25,14 +26,24 @@ class GestionSportsPlatform:
await self.session.close()
async def book(self, user: User, booking_filter: BookingFilter) -> int | None:
if self.connector is None or user is None or booking_filter is None:
return None
if self.connector is None:
LOGGER.error("No connection to Gestion Sports is available")
return
if user is None or booking_filter is None:
LOGGER.error("Not enough information available to book a court")
return
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_can_book(self, user: User, club: Club):
await self.connector.land()
await self.connector.login(user, self.club)
await self.connector.get_user_information()
def wait_until_booking_time(club: Club, booking_filter: BookingFilter):
"""
@ -43,12 +54,21 @@ def wait_until_booking_time(club: Club, booking_filter: BookingFilter):
:param club: the club where to book a court
:param booking_filter: the booking information
"""
LOGGER.info("Waiting booking time")
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: