kbe/crossfit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Make notifier and program async

Browse Source 
Using asyncio library, notification are now trigerred
asynchronously.
This commit is contained in:
kbe
2025-07-21 20:29:44 +02:00
parent 3a378c03a6
commit d71bfff10e
3 changed files with 83 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
crossfit_booker.py
View File

@@ -363,7 +363,9 @@ class CrossFitBooker:
# First check if can_join is true (primary condition)
if user_info.get("can_join", False):
activity_name = session.get("name_activity")
logging.debug(f"Session is bookable: {activity_name} can_join is True")
activity_date = session.get("start_timestamp", "Unknown date")
activity_time = activity_date.split(" ")[1] if " " in activity_date else "Unknown time"
logging.debug(f"Session is bookable: {activity_name} on {activity_date} at {activity_time} - can_join is True")
return True
# If can_join is False, check if there's a booking window
@@ -435,7 +437,7 @@ class CrossFitBooker:
logging.error(f"Failed to check session: {str(e)} - Session: {session}")
return False
def run_booking_cycle(self, current_time: datetime) -> None:
async def run_booking_cycle(self, current_time: datetime) -> None:
"""
Run one cycle of checking and booking sessions.
@@ -487,7 +489,7 @@ class CrossFitBooker:
if not session_time.tzinfo:
session_time = pytz.timezone(TIMEZONE).localize(session_time)
session_details = f"{session['name_activity']} at {session_time.strftime('%Y-%m-%d %H:%M')}"
self.notifier.notify_session_booking(session_details)
await self.notifier.notify_session_booking(session_details)
logging.info(f"Notified about found preferred session: {session_details}")
# Notify about upcoming sessions
@@ -496,7 +498,7 @@ class CrossFitBooker:
if not session_time.tzinfo:
session_time = pytz.timezone(TIMEZONE).localize(session_time)
session_details = f"{session['name_activity']} at {session_time.strftime('%Y-%m-%d %H:%M')}"
self.notifier.notify_upcoming_session(session_details, 1) # Days until is 1 for tomorrow
await self.notifier.notify_upcoming_session(session_details, 1) # Days until is 1 for tomorrow
logging.info(f"Notified about upcoming session: {session_details}")
# Book sessions (preferred first)
@@ -507,12 +509,12 @@ class CrossFitBooker:
if self.book_session(session["id_activity_calendar"]):
# Send notification after successful booking
session_details = f"{session['name_activity']} at {session_time.strftime('%Y-%m-%d %H:%M')}"
self.notifier.notify_session_booking(session_details)
await self.notifier.notify_session_booking(session_details)
logging.info(f"Successfully booked {session_type} session at {session_time}")
else:
logging.error(f"Failed to book {session_type} session at {session_time} - Session: {session}")
def run(self) -> None:
async def run(self) -> None:
"""
Main execution loop.
"""
@@ -531,7 +533,7 @@ class CrossFitBooker:
logging.info(f"Current time: {current_time}")
# Always run booking cycle to check for preferred sessions and notify
self.run_booking_cycle(current_time)
await self.run_booking_cycle(current_time)
# Run booking cycle at the target time for actual booking
target_time_str = current_time.strftime("%H:%M")