From cba4299b9a7fcbd80032215b9e80ea477a267ebc Mon Sep 17 00:00:00 2001 From: kbe Date: Sun, 20 Jul 2025 03:06:07 +0200 Subject: [PATCH] Add more self documenting comments --- book_crossfit.py | 252 ++++++++++++++++++++++++++++++----------------- 1 file changed, 164 insertions(+), 88 deletions(-) diff --git a/book_crossfit.py b/book_crossfit.py index 98279bf..2ba345b 100755 --- a/book_crossfit.py +++ b/book_crossfit.py @@ -62,11 +62,14 @@ logging.info("Logging enhanced with request library noise reduction") class CrossFitBooker: - def __init__(self): - self.auth_token = None - self.user_id = None - self.session = requests.Session() - self.base_headers = { + def __init__(self) -> None: + """ + Initialize the CrossFitBooker with necessary attributes. + """ + self.auth_token: Optional[str] = None + self.user_id: Optional[str] = None + self.session: requests.Session = requests.Session() + self.base_headers: Dict[str, str] = { "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:140.0) Gecko/20100101 Firefox/140.0", "Content-Type": "application/x-www-form-urlencoded", "Nubapp-Origin": "user_apps", @@ -74,42 +77,52 @@ class CrossFitBooker: self.session.headers.update(self.base_headers) # Define mandatory parameters for API calls - self.mandatory_params = { + self.mandatory_params: Dict[str, str] = { "app_version": APP_VERSION, "device_type": DEVICE_TYPE, "id_application": APPLICATION_ID, "id_category_activity": CATEGORY_ID } - def get_auth_headers(self) -> Dict: - """Return headers with authorization if available""" - headers = self.base_headers.copy() + def get_auth_headers(self) -> Dict[str, str]: + """ + Return headers with authorization if available. + + Returns: + Dict[str, str]: Headers dictionary with authorization if available. + """ + headers: Dict[str, str] = self.base_headers.copy() if self.auth_token: headers["Authorization"] = f"Bearer {self.auth_token}" return headers def login(self) -> bool: - """Authenticate and get the bearer token""" + """ + Authenticate and get the bearer token. + + Returns: + bool: True if login is successful, False otherwise. + """ try: # First login endpoint - login_params = { + login_params: Dict[str, str] = { "app_version": APP_VERSION, "device_type": DEVICE_TYPE, "username": USERNAME, "password": PASSWORD } - - response = self.session.post( + + response: requests.Response = self.session.post( "https://sport.nubapp.com/api/v4/users/checkUser.php", headers={"Content-Type": "application/x-www-form-urlencoded"}, data=urlencode(login_params)) - + if not response.ok: logging.error(f"First login step failed: {response.status_code} - {response.text} - Response: {response.text[:100]}") return False - + try: - login_data = response.json() + 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}") @@ -117,9 +130,9 @@ class CrossFitBooker: except ValueError as ve: logging.error(f"Value error during login: {str(ve)} - Response: {response.text}") return False - + # Second login endpoint - response = self.session.post( + response: requests.Response = self.session.post( "https://sport.nubapp.com/api/v4/login", headers={"Content-Type": "application/x-www-form-urlencoded;charset=UTF-8"}, data=urlencode({ @@ -127,10 +140,10 @@ class CrossFitBooker: "username": USERNAME, "password": PASSWORD })) - + if response.ok: try: - login_data = response.json() + 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}") @@ -138,14 +151,14 @@ class CrossFitBooker: except ValueError as ve: logging.error(f"Value error during login: {str(ve)} - 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]}") return False - + except requests.exceptions.JSONDecodeError: logging.error("Failed to decode JSON response during login") return False @@ -156,34 +169,46 @@ 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]: - """Fetch available sessions from the API with comprehensive error handling""" + def get_available_sessions(self, start_date: datetime, end_date: datetime) -> Optional[Dict[str, Any]]: + """ + Fetch available sessions from the API with comprehensive error handling. + + Args: + start_date (datetime): Start date for fetching sessions. + end_date (datetime): End date for fetching sessions. + + Returns: + Optional[Dict[str, Any]]: Dictionary containing available sessions if successful, None otherwise. + """ if not self.auth_token or not self.user_id: logging.error("Authentication required - missing token or user ID") return None - - url = "https://sport.nubapp.com/api/v4/activities/getActivitiesCalendar.php" - + + url: str = "https://sport.nubapp.com/api/v4/activities/getActivitiesCalendar.php" + # Prepare request with mandatory parameters - request_data = self.mandatory_params.copy() + request_data: Dict[str, str] = self.mandatory_params.copy() request_data.update({ "id_user": self.user_id, "start_timestamp": start_date.strftime("%d-%m-%Y"), "end_timestamp": end_date.strftime("%d-%m-%Y") }) - - # Add retry logic with exponential backoff + + # Add retry logic with exponential backoff and more informative error messages for retry in range(RETRY_MAX): try: try: - response = self.session.post( + 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}") + logging.error(f"Request timed out after 10 seconds for URL: {url}. Retry {retry+1}/{RETRY_MAX}") + return None + except requests.exceptions.ConnectionError as e: + logging.error(f"Connection error for URL: {url} - Error: {str(e)}") return None except requests.exceptions.RequestException as e: logging.error(f"Request failed for URL: {url} - Error: {str(e)}") @@ -196,7 +221,7 @@ class CrossFitBooker: if retry == RETRY_MAX - 1: logging.error(f"Final retry failed: {str(e)}") raise # Propagate error - wait_time = RETRY_BACKOFF * (2 ** retry) + 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: @@ -207,7 +232,7 @@ class CrossFitBooker: # Handle response if response.status_code == 200: try: - json_response = response.json() + json_response: Dict[str, Any] = response.json() return json_response except ValueError: logging.error("Failed to decode JSON response") @@ -227,17 +252,32 @@ class CrossFitBooker: else: logging.error(f"Unexpected status code: {response.status_code}") return None - def book_session(self, session_id: str) -> bool: - """Book a specific session with debug logging.""" + """ + Book a specific session with debug logging. + + 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: - """Prepare request data for booking a session""" + 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 { **self.mandatory_params, "id_activity_calendar": session_id, @@ -247,11 +287,21 @@ class CrossFitBooker: "booked_on": "3" } - def _make_request(self, url: str, data: Dict, success_msg: str) -> bool: - """Handle API requests with retry logic and response processing""" + 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 = self.session.post( + response: requests.Response = self.session.post( url, headers=self.get_auth_headers(), data=urlencode(data), @@ -259,7 +309,7 @@ class CrossFitBooker: ) if response.status_code == 200: - json_response = response.json() + json_response: Dict[str, Any] = response.json() if json_response.get("success", False): logging.info(success_msg) return True @@ -268,7 +318,7 @@ 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 @@ -276,30 +326,39 @@ class CrossFitBooker: if retry == RETRY_MAX - 1: logging.error(f"Final retry failed: {str(e)}") raise # Propagate error - wait_time = RETRY_BACKOFF * (2 ** retry) + 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) logging.error(f"Failed to complete request after {RETRY_MAX} attempts") return False - def is_session_bookable(self, session: Dict, current_time: datetime) -> bool: - """Check if a session is bookable based on user_info, ignoring error codes.""" - user_info = session.get("user_info", {}) + 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. + Args: + session (Dict[str, Any]): Session data. + current_time (datetime): Current time for comparison. + + Returns: + bool: True if the session is bookable, False otherwise. + """ + user_info: Dict[str, Any] = session.get("user_info", {}) + # 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") return True # If can_join is False, check if there's a booking window - booking_date_str = user_info.get("unableToBookUntilDate", "") - booking_time_str = user_info.get("unableToBookUntilTime", "") + 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.strptime( - f"{booking_date_str} {booking_time_str}", + 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) @@ -315,26 +374,35 @@ class CrossFitBooker: # Default case: not bookable return False - def matches_preferred_session(self, session: Dict, current_time: datetime) -> bool: - """Check if session matches one of your preferred sessions with fuzzy matching.""" + 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. + """ try: - session_time = parse(session["start_timestamp"]) + session_time: datetime = parse(session["start_timestamp"]) if not session_time.tzinfo: session_time = pytz.timezone(TIMEZONE).localize(session_time) - - day_of_week = session_time.weekday() - session_time_str = session_time.strftime("%H:%M") - session_name = session.get("name_activity", "").upper() - + + day_of_week: int = session_time.weekday() + session_time_str: str = session_time.strftime("%H:%M") + session_name: str = session.get("name_activity", "").upper() + for preferred_day, preferred_time, preferred_name in PREFERRED_SESSIONS: # Exact match first if (day_of_week == preferred_day and session_time_str == preferred_time and preferred_name in session_name): return True - + # Fuzzy match fallback (80% similarity) - ratio = difflib.SequenceMatcher( + ratio: float = difflib.SequenceMatcher( None, session_name.lower(), preferred_name.lower() @@ -345,73 +413,81 @@ class CrossFitBooker: ratio >= 0.8): logging.debug(f"Fuzzy match: {session_name} → {preferred_name} ({ratio:.2%})") return True - + return False - + except Exception as e: logging.error(f"Failed to check session: {str(e)} - Session: {session}") return False - def run_booking_cycle(self, current_time: datetime): - """Run one cycle of checking and booking sessions""" - # Calculate date range to check (next 3 days) - start_date = current_time.date() - end_date = start_date + timedelta(days=3) + 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. + """ + # Calculate date range to check (next 3 days) + start_date: date = current_time.date() + end_date: date = start_date + timedelta(days=3) + # Get available sessions - sessions_data = self.get_available_sessions(start_date, end_date) + 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}") return - - activities = sessions_data.get("data", {}).get("activities_calendar", []) - + + activities: List[Dict[str, Any]] = sessions_data.get("data", {}).get("activities_calendar", []) + # Find sessions to book (prefered only) - sessions_to_book = [] + sessions_to_book: List[Tuple[str, Dict[str, Any]]] = [] for session in activities: if not self.is_session_bookable(session, current_time): continue - + if self.matches_preferred_session(session, current_time): sessions_to_book.append(("Preferred", session)) - + if not sessions_to_book: logging.info("No matching sessions found to book") return - + # 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: - session_time = datetime.strptime(session["start_timestamp"], "%Y-%m-%d %H:%M:%S") + session_time: datetime = datetime.strptime(session["start_timestamp"], "%Y-%m-%d %H:%M:%S") logging.info(f"Attempting to book {session_type} session at {session_time} ({session['name_activity']})") if self.book_session(session["id_activity_calendar"]): 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): - """Main execution loop""" + def run(self) -> None: + """ + Main execution loop. + """ # Set up timezone - tz = pytz.timezone(TIMEZONE) - + tz: pytz.timezone = pytz.timezone(TIMEZONE) + # Initial login if not self.login(): logging.error("Authentication failed - exiting program") return - + while True: try: - current_time = datetime.now(tz) + 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 + # Run booking cycle at the target time or if it's a test, with optimized checking if current_time.strftime("%H:%M") == TARGET_RESERVATION_TIME: self.run_booking_cycle(current_time) - # Wait a minute to avoid checking again immediately - time.sleep(60) + # Wait until the next booking window + wait_until = current_time + timedelta(minutes=60) + time.sleep((wait_until - current_time).total_seconds()) else: - # Check again in 30 seconds - time.sleep(30) + # Check again in 5 minutes + 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