diff --git a/crossfit_booker.py b/crossfit_booker.py index cdc3fb4..7cda75f 100644 --- a/crossfit_booker.py +++ b/crossfit_booker.py @@ -33,6 +33,7 @@ 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) +BOOKING_WINDOW_END_DELTA_MINUTES = 50 # End window book session DEVICE_TYPE = "3" # Retry configuration @@ -44,7 +45,6 @@ APP_VERSION = "5.09.21" class CrossFitBooker: """ A class for automating the booking of CrossFit sessions. - This class handles authentication, session availability checking, booking, and notifications for CrossFit sessions. """ @@ -52,7 +52,6 @@ class CrossFitBooker: def __init__(self) -> None: """ Initialize the CrossFitBooker with necessary attributes. - Sets up authentication tokens, session headers, mandatory parameters, and initializes the SessionNotifier for sending notifications. """ @@ -100,7 +99,6 @@ class CrossFitBooker: def get_auth_headers(self) -> Dict[str, str]: """ Return headers with authorization if available. - Returns: Dict[str, str]: Headers dictionary with authorization if available. """ @@ -112,7 +110,6 @@ class CrossFitBooker: def login(self) -> bool: """ Authenticate and get the bearer token. - Returns: bool: True if login is successful, False otherwise. """ @@ -131,17 +128,14 @@ class CrossFitBooker: data=urlencode(login_params)) if not response.ok: - logging.error(f"First login step failed: {response.status_code} - {response.text} - Response: {response.text[:100]}") + logging.error(f"First login step failed: {response.status_code} - {response.text[:100]}") return False try: login_data: Dict[str, Any] = response.json() self.user_id = str(login_data["data"]["user"]["id_user"]) - except KeyError as ke: - logging.error(f"Key error during login: {str(ke)} - Response: {response.text}") - return False - except ValueError as ve: - logging.error(f"Value error during login: {str(ve)} - Response: {response.text}") + except (KeyError, ValueError) as e: + logging.error(f"Error during login: {str(e)} - Response: {response.text}") return False # Second login endpoint @@ -158,23 +152,17 @@ class CrossFitBooker: try: login_data: Dict[str, Any] = response.json() self.auth_token = login_data.get("token") - except KeyError as ke: - logging.error(f"Key error during login: {str(ke)} - Response: {response.text}") - return False - except ValueError as ve: - logging.error(f"Value error during login: {str(ve)} - Response: {response.text}") + except (KeyError, ValueError) as e: + logging.error(f"Error during login: {str(e)} - Response: {response.text}") return False if self.auth_token and self.user_id: logging.info("Successfully logged in") return True else: - logging.error(f"Login failed: {response.status_code} - {response.text} - Response: {response.text[:100]}") + logging.error(f"Login failed: {response.status_code} - {response.text[:100]}") return False - except requests.exceptions.JSONDecodeError: - logging.error("Failed to decode JSON response during login") - return False except requests.exceptions.RequestException as e: logging.error(f"Request error during login: {str(e)}") return False @@ -182,14 +170,12 @@ class CrossFitBooker: logging.error(f"Unexpected error during login: {str(e)}") return False - def get_available_sessions(self, start_date: datetime, end_date: datetime) -> Optional[Dict[str, Any]]: + def get_available_sessions(self, start_date: date, end_date: date) -> Optional[Dict[str, Any]]: """ - Fetch available sessions from the API with comprehensive error handling. - + Fetch available sessions from the API. Args: - start_date (datetime): Start date for fetching sessions. - end_date (datetime): End date for fetching sessions. - + start_date (date): Start date for fetching sessions. + end_date (date): End date for fetching sessions. Returns: Optional[Dict[str, Any]]: Dictionary containing available sessions if successful, None otherwise. """ @@ -207,112 +193,56 @@ class CrossFitBooker: "end_timestamp": end_date.strftime("%d-%m-%Y") }) - # Add retry logic with exponential backoff and more informative error messages + # Add retry logic for retry in range(RETRY_MAX): try: - try: - response: requests.Response = self.session.post( - url, - headers=self.get_auth_headers(), - data=urlencode(request_data), - timeout=10 - ) - except requests.exceptions.Timeout: - logging.error(f"Request timed out after 10 seconds for URL: {url}. Retry {retry+1}/{RETRY_MAX}") + response: requests.Response = self.session.post( + url, + headers=self.get_auth_headers(), + data=urlencode(request_data), + timeout=10 + ) + + if response.status_code == 200: + return response.json() + elif response.status_code == 401: + logging.error("401 Unauthorized - token may be expired or invalid") return None - except requests.exceptions.ConnectionError as e: - logging.error(f"Connection error for URL: {url} - Error: {str(e)}") + elif 500 <= response.status_code < 600: + logging.error(f"Server error {response.status_code}") + raise requests.exceptions.ConnectionError(f"Server error {response.status_code}") + else: + logging.error(f"Unexpected status code: {response.status_code} - {response.text[:100]}") return None - except requests.exceptions.RequestException as e: - logging.error(f"Request failed for URL: {url} - Error: {str(e)}") - return None - break # Success, exit retry loop - except requests.exceptions.JSONDecodeError: - logging.error("Failed to decode JSON response") - return None + except requests.exceptions.RequestException as e: if retry == RETRY_MAX - 1: logging.error(f"Final retry failed: {str(e)}") - raise # Propagate error + raise wait_time: int = RETRY_BACKOFF * (2 ** retry) logging.warning(f"Request failed (attempt {retry+1}/{RETRY_MAX}): {str(e)}. Retrying in {wait_time}s...") time.sleep(wait_time) - else: - # All retries exhausted - logging.error(f"Failed after {RETRY_MAX} attempts") - return None - # Handle response - if response.status_code == 200: - try: - json_response: Dict[str, Any] = response.json() - return json_response - except ValueError: - logging.error("Failed to decode JSON response") - return None - elif response.status_code == 400: - logging.error("400 Bad Request - likely missing or invalid parameters") - logging.error(f"Request Data: {request_data}") - logging.error(f"Response: {response.text[:100]}") - return None - elif response.status_code == 401: - logging.error("401 Unauthorized - token may be expired or invalid") - logging.error(f"Response: {response.text[:100]}") - return None - elif 500 <= response.status_code < 600: - logging.error(f"Server error {response.status_code} - Response: {response.text[:100]}") - raise requests.exceptions.ConnectionError(f"Server error {response.status_code}") - else: - logging.error(f"Unexpected status code: {response.status_code}") - return None + return None def book_session(self, session_id: str) -> bool: """ - Book a specific session with debug logging. - + Book a specific session. Args: session_id (str): ID of the session to book. - Returns: bool: True if booking is successful, False otherwise. """ - return self._make_request( - url="https://sport.nubapp.com/api/v4/activities/bookActivityCalendar.php", - data=self._prepare_booking_data(session_id), - success_msg=f"Successfully booked session {session_id}" - ) - - def _prepare_booking_data(self, session_id: str) -> Dict[str, str]: - """ - Prepare request data for booking a session. - - Args: - session_id (str): ID of the session to book. - - Returns: - Dict[str, str]: Dictionary containing request data for booking a session. - """ - return { + url = "https://sport.nubapp.com/api/v4/activities/bookActivityCalendar.php" + data = { **self.mandatory_params, "id_activity_calendar": session_id, "id_user": self.user_id, "action_by": self.user_id, "n_guests": "0", - "booked_on": "3" # Target CrossFit Louvre 3 ? + "booked_on": "3" } - - def _make_request(self, url: str, data: Dict[str, str], success_msg: str) -> bool: - """ - Handle API requests with retry logic and response processing. - - Args: - url (str): URL for the API request. - data (Dict[str, str]): Data to send with the request. - success_msg (str): Message to log on successful request. - - Returns: - bool: True if request is successful, False otherwise. - """ + for retry in range(RETRY_MAX): try: response: requests.Response = self.session.post( @@ -325,7 +255,7 @@ class CrossFitBooker: if response.status_code == 200: json_response: Dict[str, Any] = response.json() if json_response.get("success", False): - logging.info(success_msg) + logging.info(f"Successfully booked session {session_id}") return True logging.error(f"API returned success:false: {json_response}") return False @@ -333,13 +263,10 @@ class CrossFitBooker: logging.error(f"HTTP {response.status_code}: {response.text[:100]}") return False - except requests.exceptions.JSONDecodeError: - logging.error("Failed to decode JSON response") - return False except requests.exceptions.RequestException as e: if retry == RETRY_MAX - 1: logging.error(f"Final retry failed: {str(e)}") - raise # Propagate error + raise wait_time: int = RETRY_BACKOFF * (2 ** retry) logging.warning(f"Request failed (attempt {retry+1}/{RETRY_MAX}): {str(e)}. Retrying in {wait_time}s...") time.sleep(wait_time) @@ -349,12 +276,10 @@ class CrossFitBooker: def is_session_bookable(self, session: Dict[str, Any], current_time: datetime) -> bool: """ - Check if a session is bookable based on user_info, ignoring error codes. - + Check if a session is bookable based on user_info. Args: session (Dict[str, Any]): Session data. current_time (datetime): Current time for comparison. - Returns: bool: True if the session is bookable, False otherwise. """ @@ -362,10 +287,6 @@ class CrossFitBooker: # First check if can_join is true (primary condition) if user_info.get("can_join", False): - activity_name = session.get("name_activity") - 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 @@ -381,10 +302,7 @@ class CrossFitBooker: booking_datetime = pytz.timezone(TIMEZONE).localize(booking_datetime) if current_time >= booking_datetime: - logging.debug(f"Session is bookable: current_time {current_time} >= booking_datetime {booking_datetime}") return True # Booking window is open - else: - return False # Still waiting for booking to open except ValueError: pass # Ignore invalid date formats @@ -394,11 +312,9 @@ class CrossFitBooker: def matches_preferred_session(self, session: Dict[str, Any], current_time: datetime) -> bool: """ Check if session matches one of your preferred sessions with fuzzy matching. - Args: session (Dict[str, Any]): Session data. current_time (datetime): Current time for comparison. - Returns: bool: True if the session matches a preferred session, False otherwise. """ @@ -440,7 +356,6 @@ class CrossFitBooker: async def run_booking_cycle(self, current_time: datetime) -> None: """ Run one cycle of checking and booking sessions. - Args: current_time (datetime): Current time for comparison. """ @@ -451,7 +366,7 @@ class CrossFitBooker: # Get available sessions sessions_data: Optional[Dict[str, Any]] = self.get_available_sessions(start_date, end_date) if not sessions_data or not sessions_data.get("success", False): - logging.error("No sessions available or error fetching sessions - Sessions Data: {sessions_data}") + logging.error("No sessions available or error fetching sessions") return activities: List[Dict[str, Any]] = sessions_data.get("data", {}).get("activities_calendar", []) @@ -493,7 +408,7 @@ class CrossFitBooker: 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')}" + 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}") @@ -502,7 +417,7 @@ class CrossFitBooker: 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')}" + 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}") @@ -517,55 +432,53 @@ class CrossFitBooker: 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}") + logging.error(f"Failed to book {session_type} session at {session_time}") # Send notification about the failed booking session_details = f"{session['name_activity']} at {session_time.strftime('%Y-%m-%d %H:%M')}" 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. - """ - # Set up timezone - tz: pytz.timezone = pytz.timezone(TIMEZONE) + """ + Main execution loop. + """ + # Set up timezone + tz: pytz.timezone = pytz.timezone(TIMEZONE) - # Parse TARGET_RESERVATION_TIME to get the target hour and minute - target_hour, target_minute = map(int, TARGET_RESERVATION_TIME.split(":")) - target_time = datetime.now(tz).replace(hour=target_hour, minute=target_minute, second=0, microsecond=0) - booking_window_end = target_time + timedelta(hours=1) + # Parse TARGET_RESERVATION_TIME to get the target hour and minute + target_hour, target_minute = map(int, TARGET_RESERVATION_TIME.split(":")) + target_time = datetime.now(tz).replace(hour=target_hour, minute=target_minute, second=0, microsecond=0) + booking_window_end = target_time + timedelta(minutes=BOOKING_WINDOW_END_DELTA_MINUTES) - # Initial login - if not self.login(): - logging.error("Authentication failed - exiting program") - return + # Initial login + if not self.login(): + logging.error("Authentication failed - exiting program") + return - try: - while True: - try: - current_time: datetime = datetime.now(tz) - logging.info(f"Current time: {current_time}") - - # 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) - # Wait for a short time before next check - time.sleep(60) - else: - # Check again in 5 minutes if outside booking window - time.sleep(300) - except Exception as e: - logging.error(f"Unexpected error in booking cycle: {str(e)} - Traceback: {traceback.format_exc()}") - time.sleep(60) # Wait before retrying after error - except KeyboardInterrupt: - self.quit() + try: + while True: + try: + current_time: datetime = datetime.now(tz) + logging.info(f"Current time: {current_time}") + # 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) + # Wait for a short time before next check + time.sleep(60) + else: + # Check again in 5 minutes if outside booking window + time.sleep(300) + except Exception as e: + logging.error(f"Unexpected error in booking cycle: {str(e)} - Traceback: {traceback.format_exc()}") + time.sleep(60) # Wait before retrying after error + except KeyboardInterrupt: + self.quit() def quit(self) -> None: """ Clean up resources and exit the script. """ logging.info("Script interrupted by user. Quitting...") - # Add any cleanup code here exit(0) diff --git a/test_crossfit_booker.py b/test_crossfit_booker.py deleted file mode 100644 index 253c312..0000000 --- a/test_crossfit_booker.py +++ /dev/null @@ -1,205 +0,0 @@ -#!/usr/bin/env python3 -""" -Test script for the refactored CrossFitBooker functional implementation. -""" - -import sys -import os -from datetime import datetime -import pytz -from typing import List, Dict, Any, Tuple - -# Add the current directory to the path so we can import our modules -sys.path.append(os.path.dirname(os.path.abspath(__file__))) - -# Import the functional functions from our refactored code -from crossfit_booker_functional import ( - is_session_bookable, - matches_preferred_session, - filter_bookable_sessions, - filter_upcoming_sessions, - filter_preferred_sessions, - categorize_sessions, - format_session_details -) - -def test_is_session_bookable(): - """Test the is_session_bookable function.""" - print("Testing is_session_bookable...") - - # Test case 1: Session with can_join = True - session1 = { - "user_info": { - "can_join": True - } - } - current_time = datetime.now(pytz.timezone("Europe/Paris")) - assert is_session_bookable(session1, current_time, "Europe/Paris") == True - - # Test case 2: Session with booking window in the past - session2 = { - "user_info": { - "unableToBookUntilDate": "01-01-2020", - "unableToBookUntilTime": "10:00" - } - } - assert is_session_bookable(session2, current_time, "Europe/Paris") == True - - # Test case 3: Session with booking window in the future - session3 = { - "user_info": { - "unableToBookUntilDate": "01-01-2030", - "unableToBookUntilTime": "10:00" - } - } - assert is_session_bookable(session3, current_time, "Europe/Paris") == False - - print("✓ is_session_bookable tests passed") - - -def test_matches_preferred_session(): - """Test the matches_preferred_session function.""" - print("Testing matches_preferred_session...") - - # Define some preferred sessions (day_of_week, start_time, session_name_contains) - preferred_sessions: List[Tuple[int, str, str]] = [ - (2, "18:30", "CONDITIONING"), # Wednesday 18:30 CONDITIONING - (4, "17:00", "WEIGHTLIFTING"), # Friday 17:00 WEIGHTLIFTING - (5, "12:30", "HYROX"), # Saturday 12:30 HYROX - ] - - # Test case 1: Exact match - session1 = { - "start_timestamp": "2025-07-30 18:30:00", # Wednesday - "name_activity": "CONDITIONING" - } - assert matches_preferred_session(session1, preferred_sessions, "Europe/Paris") == True - - # Test case 2: No match - session2 = { - "start_timestamp": "2025-07-30 18:30:00", # Wednesday - "name_activity": "YOGA" - } - assert matches_preferred_session(session2, preferred_sessions, "Europe/Paris") == False - - print("✓ matches_preferred_session tests passed") - - -def test_filter_functions(): - """Test the filter functions.""" - print("Testing filter functions...") - - # Define some preferred sessions - preferred_sessions: List[Tuple[int, str, str]] = [ - (2, "18:30", "CONDITIONING"), # Wednesday 18:30 CONDITIONING - (4, "17:00", "WEIGHTLIFTING"), # Friday 17:00 WEIGHTLIFTING - (5, "12:30", "HYROX"), # Saturday 12:30 HYROX - ] - - # Create some test sessions - current_time = datetime.now(pytz.timezone("Europe/Paris")) - - sessions = [ - { - "start_timestamp": "2025-07-30 18:30:00", # Wednesday - "name_activity": "CONDITIONING", - "user_info": {"can_join": True} - }, - { - "start_timestamp": "2025-07-30 19:00:00", # Wednesday - "name_activity": "YOGA", - "user_info": {"can_join": True} - }, - { - "start_timestamp": "2025-08-01 17:00:00", # Friday - "name_activity": "WEIGHTLIFTING", - "user_info": {"can_join": True} - } - ] - - # Test filter_preferred_sessions - preferred = filter_preferred_sessions(sessions, preferred_sessions, "Europe/Paris") - assert len(preferred) == 2 # CONDITIONING and WEIGHTLIFTING sessions - - # Test filter_bookable_sessions - bookable = filter_bookable_sessions(sessions, current_time, preferred_sessions, "Europe/Paris") - assert len(bookable) == 2 # Both preferred sessions are bookable - - print("✓ Filter function tests passed") - - -def test_categorize_sessions(): - """Test the categorize_sessions function.""" - print("Testing categorize_sessions...") - - # Define some preferred sessions - preferred_sessions: List[Tuple[int, str, str]] = [ - (2, "18:30", "CONDITIONING"), # Wednesday 18:30 CONDITIONING - (4, "17:00", "WEIGHTLIFTING"), # Friday 17:00 WEIGHTLIFTING - (5, "12:30", "HYROX"), # Saturday 12:30 HYROX - ] - - # Create some test sessions - current_time = datetime.now(pytz.timezone("Europe/Paris")) - - sessions = [ - { - "start_timestamp": "2025-07-30 18:30:00", # Wednesday - "name_activity": "CONDITIONING", - "user_info": {"can_join": True} - }, - { - "start_timestamp": "2025-07-31 18:30:00", # Thursday (tomorrow relative to Wednesday) - "name_activity": "CONDITIONING", - "user_info": {"can_join": False, "unableToBookUntilDate": "01-08-2025", "unableToBookUntilTime": "10:00"} - } - ] - - # Test categorize_sessions - categorized = categorize_sessions(sessions, current_time, preferred_sessions, "Europe/Paris") - assert "bookable" in categorized - assert "upcoming" in categorized - assert "all_preferred" in categorized - - print("✓ categorize_sessions tests passed") - - -def test_format_session_details(): - """Test the format_session_details function.""" - print("Testing format_session_details...") - - # Test case 1: Valid session - session1 = { - "start_timestamp": "2025-07-30 18:30:00", - "name_activity": "CONDITIONING" - } - formatted = format_session_details(session1, "Europe/Paris") - assert "CONDITIONING" in formatted - assert "2025-07-30 18:30" in formatted - - # Test case 2: Session with missing data - session2 = { - "name_activity": "WEIGHTLIFTING" - } - formatted = format_session_details(session2, "Europe/Paris") - assert "WEIGHTLIFTING" in formatted - assert "Unknown time" in formatted - - print("✓ format_session_details tests passed") - - -def run_all_tests(): - """Run all tests.""" - print("Running all tests for CrossFitBooker functional implementation...\n") - - test_is_session_bookable() - test_matches_preferred_session() - test_filter_functions() - test_categorize_sessions() - test_format_session_details() - - print("\n✓ All tests passed!") - - -if __name__ == "__main__": - run_all_tests() \ No newline at end of file