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

feat: Scripts display preferred sessions during execution

Browse Source
This commit is contained in:
kbe
2025-08-12 00:09:56 +02:00
parent d31976084a
commit 7161a11905
5 changed files with 98 additions and 777 deletions
Download Patch File Download Diff File Expand all files Collapse all files

90
crossfit_booker.py
View File

@@ -339,23 +339,6 @@ class CrossFitBooker:
if user_info.get("can_join", False):
return True
# If can_join is False, check if there's a booking window
booking_date_str: str = user_info.get("unableToBookUntilDate", "")
booking_time_str: str = user_info.get("unableToBookUntilTime", "")
if booking_date_str and booking_time_str:
try:
booking_datetime: datetime = datetime.strptime(
f"{booking_date_str} {booking_time_str}",
"%d-%m-%Y %H:%M"
)
booking_datetime = pytz.timezone(TIMEZONE).localize(booking_datetime)
if current_time >= booking_datetime:
return True # Booking window is open
except ValueError:
pass # Ignore invalid date formats
# Default case: not bookable
return False
@@ -390,7 +373,37 @@ class CrossFitBooker:
logging.error(f"Failed to check session: {str(e)} - Session: {session}")
return False
async def run_booking_cycle(self, current_time: datetime) -> None:
def display_upcoming_sessions(self, sessions: List[Dict[str, Any]], current_time: datetime) -> None:
"""
Display upcoming sessions with ID, name, date, and time.
Args:
sessions (List[Dict[str, Any]]): List of session data.
current_time (datetime): Current time for comparison.
"""
if not sessions:
logging.info("No sessions to display")
return
logging.info("Upcoming sessions:")
logging.info("ID\t\tName\t\tDate\t\tTime")
logging.info("="*50)
for session in sessions:
session_time: datetime = parse(session["start_timestamp"])
if not session_time.tzinfo:
session_time = pytz.timezone(TIMEZONE).localize(session_time)
# Format session details
session_id = session.get("id_activity_calendar", "N/A")
session_name = session.get("name_activity", "N/A")
session_date = session_time.strftime("%Y-%m-%d")
session_time_str = session_time.strftime("%H:%M")
# Display session details
logging.info(f"{session_id}\t{session_name}\t{session_date}\t{session_time_str}")
async def booker(self, current_time: datetime) -> None:
"""
Run one cycle of checking and booking sessions.
Args:
@@ -408,9 +421,14 @@ class CrossFitBooker:
activities: List[Dict[str, Any]] = sessions_data.get("data", {}).get("activities_calendar", [])
# Display all available sessions within the date range
self.display_upcoming_sessions(activities, current_time)
# Display preferred sessions found
# Display booked session(s)
# Find sessions to book (preferred only) within allowed date range
sessions_to_book: List[Tuple[str, Dict[str, Any]]] = []
upcoming_sessions: List[Dict[str, Any]] = []
found_preferred_sessions: List[Dict[str, Any]] = []
for session in activities:
@@ -426,39 +444,14 @@ class CrossFitBooker:
# Check if session is preferred and bookable
if self.is_session_bookable(session, current_time):
if self.matches_preferred_session(session, current_time):
sessions_to_book.append(("Preferred", session))
found_preferred_sessions.append(session)
else:
# Check if it's a preferred session that's not bookable yet
if self.matches_preferred_session(session, current_time):
found_preferred_sessions.append(session)
# Check if it's available tomorrow (day + 1)
if days_diff == 1:
upcoming_sessions.append(session)
if not sessions_to_book and not upcoming_sessions:
if not found_preferred_sessions:
logging.info("No matching sessions found to book")
return
# Notify about all found preferred sessions, regardless of bookability
for session in found_preferred_sessions:
session_time: datetime = parse(session["start_timestamp"])
if not session_time.tzinfo:
session_time = pytz.timezone(TIMEZONE).localize(session_time)
session_details = f"{session['id_activity_calendar']} {session['name_activity']} at {session_time.strftime('%Y-%m-%d %H:%M')}"
await self.notifier.notify_session_booking(session_details)
logging.info(f"Notified about found preferred session: {session_details}")
# Notify about upcoming sessions
for session in upcoming_sessions:
session_time: datetime = parse(session["start_timestamp"])
if not session_time.tzinfo:
session_time = pytz.timezone(TIMEZONE).localize(session_time)
session_details = f"{session['id_activity_calendar']} {session['name_activity']} at {session_time.strftime('%Y-%m-%d %H:%M')}"
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)
sessions_to_book = [(("Preferred", session) for session in found_preferred_sessions)]
sessions_to_book.sort(key=lambda x: 0 if x[0] == "Preferred" else 1)
for session_type, session in sessions_to_book:
session_time: datetime = datetime.strptime(session["start_timestamp"], "%Y-%m-%d %H:%M:%S")
@@ -475,6 +468,7 @@ class CrossFitBooker:
await self.notifier.notify_impossible_booking(session_details)
logging.info(f"Notified about impossible booking for {session_type} session at {session_time}")
async def run(self) -> None:
"""
Main execution loop.
@@ -501,7 +495,7 @@ class CrossFitBooker:
# Only book sessions if current time is within the booking window
if target_time <= current_time <= booking_window_end:
# Run booking cycle to check for preferred sessions and book
await self.run_booking_cycle(current_time)
await self.booker(current_time)
# Wait for a short time before next check
time.sleep(60)
else: