feat: Add notification for upcoming sessions
This commit is contained in:
@@ -30,8 +30,7 @@ if not all([USERNAME, PASSWORD]):
|
||||
APPLICATION_ID = "81560887"
|
||||
CATEGORY_ID = "677" # Activity category ID for CrossFit
|
||||
TIMEZONE = "Europe/Paris" # Adjust to your timezone
|
||||
# TARGET_RESERVATION_TIME = "20:01" # When bookings open (8 PM)
|
||||
TARGET_RESERVATION_TIME = "21:40" # When bookings open (8 PM)
|
||||
TARGET_RESERVATION_TIME = "20:01" # When bookings open (8 PM)
|
||||
DEVICE_TYPE = "3"
|
||||
|
||||
# Retry configuration
|
||||
@@ -43,6 +42,7 @@ APP_VERSION = "5.09.21"
|
||||
# Format: List of tuples (day_of_week, start_time, session_name_contains)
|
||||
# day_of_week: 0=Monday, 6=Sunday
|
||||
PREFERRED_SESSIONS = [
|
||||
# (0, "17:00", "HYROX"), # Monday 17:00 HYROX
|
||||
(4, "17:00", "WEIGHTLIFTING"), # Friday 17:00 WEIGHTLIFTING
|
||||
(5, "12:30", "HYROX"), # Saturday 12:30 HYROX
|
||||
(2, "18:30", "CONDITIONING"), # Wednesday 18:30 CONDITIONING
|
||||
@@ -373,7 +373,8 @@ class CrossFitBooker:
|
||||
|
||||
# First check if can_join is true (primary condition)
|
||||
if user_info.get("can_join", False):
|
||||
logging.debug("Session is bookable: can_join is True")
|
||||
activity_name = session.get("name_activity")
|
||||
logging.debug(f"Session is bookable: {activity_name} can_join is True")
|
||||
return True
|
||||
|
||||
# If can_join is False, check if there's a booking window
|
||||
@@ -464,19 +465,51 @@ class CrossFitBooker:
|
||||
|
||||
activities: List[Dict[str, Any]] = sessions_data.get("data", {}).get("activities_calendar", [])
|
||||
|
||||
# Find sessions to book (prefered only)
|
||||
# Find sessions to book (preferred only)
|
||||
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:
|
||||
if not self.is_session_bookable(session, current_time):
|
||||
continue
|
||||
session_time: datetime = parse(session["start_timestamp"])
|
||||
if not session_time.tzinfo:
|
||||
session_time = pytz.timezone(TIMEZONE).localize(session_time)
|
||||
|
||||
if self.matches_preferred_session(session, current_time):
|
||||
sessions_to_book.append(("Preferred", session))
|
||||
# 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
|
||||
if (session_time.date() - current_time.date()).days == 1:
|
||||
upcoming_sessions.append(session)
|
||||
|
||||
if not sessions_to_book:
|
||||
if not sessions_to_book and not upcoming_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['name_activity']} at {session_time.strftime('%Y-%m-%d %H:%M')}"
|
||||
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['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
|
||||
logging.info(f"Notified about upcoming session: {session_details}")
|
||||
|
||||
# Book sessions (preferred first)
|
||||
sessions_to_book.sort(key=lambda x: 0 if x[0] == "Preferred" else 1)
|
||||
for session_type, session in sessions_to_book:
|
||||
@@ -508,9 +541,11 @@ class CrossFitBooker:
|
||||
current_time: datetime = datetime.now(tz)
|
||||
logging.info(f"Current time: {current_time}")
|
||||
|
||||
# Run booking cycle at the target time or if it's a test, with optimized checking
|
||||
# Always run booking cycle to check for preferred sessions and notify
|
||||
self.run_booking_cycle(current_time)
|
||||
|
||||
# Run booking cycle at the target time for actual booking
|
||||
if current_time.strftime("%H:%M") == TARGET_RESERVATION_TIME:
|
||||
self.run_booking_cycle(current_time)
|
||||
# Wait until the next booking window
|
||||
wait_until = current_time + timedelta(minutes=60)
|
||||
time.sleep((wait_until - current_time).total_seconds())
|
||||
|
||||
Reference in New Issue
Block a user