From e8cd2d3d960cd9f20f3d5956f8df81099413f76f Mon Sep 17 00:00:00 2001 From: kbe Date: Fri, 18 Jul 2025 16:03:38 +0200 Subject: [PATCH 01/11] It looks like it works? --- book_crossfit.py | 177 +++++++++++++++++++++++++---------------------- 1 file changed, 94 insertions(+), 83 deletions(-) diff --git a/book_crossfit.py b/book_crossfit.py index 6593f73..5f68c3c 100755 --- a/book_crossfit.py +++ b/book_crossfit.py @@ -33,12 +33,16 @@ PASSWORD = os.environ.get("CROSSFIT_PASSWORD") if not all([USERNAME, PASSWORD]): raise ValueError("Missing environment variables: CROSSFIT_USERNAME and/or CROSSFIT_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) DEVICE_TYPE = "3" + +# Retry configuration +RETRY_MAX = 3 +RETRY_BACKOFF = 1 APP_VERSION = "5.09.21" # Define your preferred sessions @@ -156,102 +160,109 @@ class CrossFitBooker: # print(f"Request Data: {request_data}") # print(f"Headers: {self.get_auth_headers()}") - try: - # Make the request - response = self.session.post( - url, - headers=self.get_auth_headers(), - data=urlencode(request_data), - timeout=10 - ) - - # Debug raw response - # print(f"Response Status Code: {response.status_code}") - # print(f"Response Content: {response.text}") - - # Handle response - if response.status_code == 200: - try: - json_response = response.json() - return json_response - except ValueError: - print("Failed to decode JSON response") - return None - elif response.status_code == 400: - print("400 Bad Request - likely missing or invalid parameters") - print("Verify these parameters:") - for param, value in request_data.items(): - print(f"- {param}: {value}") - return None - elif response.status_code == 401: - print("401 Unauthorized - token may be expired or invalid") - return None - else: - print(f"Unexpected status code: {response.status_code}") - return None - - except requests.exceptions.RequestException as e: - print(f"Request failed: {str(e)}") + # Add retry logic with exponential backoff + for retry in range(RETRY_MAX): + try: + response = self.session.post( + url, + headers=self.get_auth_headers(), + data=urlencode(request_data), + timeout=10 + ) + break # Success, exit retry loop + except (requests.exceptions.ConnectionError, + requests.exceptions.Timeout, + requests.exceptions.ReadTimeout) as e: + if retry == RETRY_MAX - 1: + raise # Final retry failed, propagate error + wait_time = 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 + print(f"Failed after {RETRY_MAX} attempts") return None - except Exception as e: - print(f"Unexpected error: {str(e)}") + + # Debug raw response + # print(f"Response Status Code: {response.status_code}") + # print(f"Response Content: {response.text}") + + # Handle response + if response.status_code == 200: + try: + json_response = response.json() + return json_response + except ValueError: + print("Failed to decode JSON response") + return None + elif response.status_code == 400: + print("400 Bad Request - likely missing or invalid parameters") + print("Verify these parameters:") + for param, value in request_data.items(): + print(f"- {param}: {value}") + return None + elif response.status_code == 401: + print("401 Unauthorized - token may be expired or invalid") + return None + elif 500 <= response.status_code < 600: + raise requests.exceptions.ConnectionError(f"Server error {response.status_code}") + else: + print(f"Unexpected status code: {response.status_code}") return None def book_session(self, session_id: str) -> bool: """Book a specific session with debug logging.""" - - logging.info(f"Attempting to book session_id: {session_id}") - if not self.auth_token or not self.user_id: - logging.error("Not authenticated: missing auth_token or user_id") - return False + 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}" + ) - try: - # Prepare headers - headers = self.get_auth_headers() + def _prepare_booking_data(self, session_id: str) -> Dict: + """Prepare request data for booking a session""" + return { + **self.mandatory_params, + "id_activity_calendar": session_id, + "id_user": self.user_id, + "action_by": self.user_id, + "n_guests": "0", + "booked_on": "3" + } - # print(f"[DEBUG] Request headers: {headers}") - - # Prepare the exact request data from cURL - request_data = self.mandatory_params.copy() - request_data.update({ - "id_activity_calendar": session_id, # Note the different parameter name - "id_user": self.user_id, - "action_by": self.user_id, # Same as id_user in this case - "n_guests": "0", - "booked_on": "3" # 3 likely means "booked via app" - }) - - print(f"[DEBUG] Final request data: {request_data}") - - # Use the correct endpoint - response = self.session.post( - "https://sport.nubapp.com/api/v4/activities/bookActivityCalendar.php", - headers=headers, - data=urlencode(request_data) - ) - - logging.debug(f"Response status: {response.status_code}") - logging.debug(f"API response: {response.text}") - - if response.ok: - try: + def _make_request(self, url: str, data: Dict, success_msg: str) -> bool: + """Handle API requests with retry logic and response processing""" + for retry in range(RETRY_MAX): + try: + response = self.session.post( + url, + headers=self.get_auth_headers(), + data=urlencode(data), + timeout=10 + ) + + if response.status_code == 200: json_response = response.json() if json_response.get("success", False): - logging.info(f"Successfully booked session {session_id}") + logging.info(success_msg) return True - else: - logging.error(f"API returned success:false: {json_response}") - return False - except ValueError: - logging.error("Invalid JSON response") + logging.error(f"API returned success:false: {json_response}") return False - else: + logging.error(f"HTTP {response.status_code}: {response.text}") return False - except Exception as e: - logging.critical(f"Unexpected error: {str(e)}", exc_info=True) - return False + except (requests.exceptions.ConnectionError, + requests.exceptions.Timeout, + requests.exceptions.ReadTimeout) as e: + if retry == RETRY_MAX - 1: + logging.error(f"All {RETRY_MAX} retry attempts failed") + raise + wait_time = 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.""" -- 2.49.1 From 3a0af18b730a1729ecec6da9e77d8a5c9010c70e Mon Sep 17 00:00:00 2001 From: kbe Date: Fri, 18 Jul 2025 21:24:23 +0200 Subject: [PATCH 02/11] feat: More logging information in code All methods functions are working properly. Now includes logging and log file. --- .kilocode/mcp.json | 1 + book_crossfit.py | 227 +++++++++++++++++++++++---------------------- 2 files changed, 116 insertions(+), 112 deletions(-) create mode 100644 .kilocode/mcp.json diff --git a/.kilocode/mcp.json b/.kilocode/mcp.json new file mode 100644 index 0000000..6b0a486 --- /dev/null +++ b/.kilocode/mcp.json @@ -0,0 +1 @@ +{"mcpServers":{}} \ No newline at end of file diff --git a/book_crossfit.py b/book_crossfit.py index 5f68c3c..7f0999b 100755 --- a/book_crossfit.py +++ b/book_crossfit.py @@ -7,6 +7,7 @@ from datetime import datetime, timedelta import os import sys import time +import difflib # Third-party modules import requests @@ -16,18 +17,9 @@ from dotenv import load_dotenv from urllib.parse import urlencode from typing import List, Dict, Optional -from datetime import datetime, timedelta - -# Parse session time (handles timezones if present) -from dateutil.parser import parse -import pytz -from urllib.parse import urlencode -from typing import List, Dict, Optional -from dotenv import load_dotenv -load_dotenv() +load_dotenv() # Configuration - USERNAME = os.environ.get("CROSSFIT_USERNAME") PASSWORD = os.environ.get("CROSSFIT_PASSWORD") @@ -57,13 +49,15 @@ PREFERRED_SESSIONS = [ # Configure logging once at script startup logging.basicConfig( - level=logging.INFO, + level=logging.DEBUG, # Change to DEBUG for more detailed logs format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler("log/crossfit_booking.log"), logging.StreamHandler() ] ) +logging.getLogger("requests").setLevel(logging.WARNING) +logging.info("Logging enhanced with request library noise reduction") class CrossFitBooker: @@ -110,11 +104,18 @@ class CrossFitBooker: data=urlencode(login_params)) if not response.ok: - print(f"First login step failed: {response.status_code} - {response.text}") + logging.error(f"First login step failed: {response.status_code} - {response.text} - Response: {response.text[:100]}") return False - login_data = response.json() - self.user_id = str(login_data["data"]["user"]["id_user"]) + try: + login_data = 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}") + return False # Second login endpoint response = self.session.post( @@ -127,23 +128,37 @@ class CrossFitBooker: })) if response.ok: - login_data = response.json() - self.auth_token = login_data.get("token") + try: + login_data = 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}") + return False if self.auth_token and self.user_id: - print("Successfully logged in") + logging.info("Successfully logged in") return True - print(f"Login failed: {response.status_code} - {response.text}") - return False + 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 + except requests.exceptions.RequestException as e: + logging.error(f"Request error during login: {str(e)}") + return False except Exception as e: - print(f"Login error: {str(e)}") + 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""" if not self.auth_token or not self.user_id: - print("Authentication required - missing token or user ID") + logging.error("Authentication required - missing token or user ID") return None url = "https://sport.nubapp.com/api/v4/activities/getActivitiesCalendar.php" @@ -156,58 +171,60 @@ class CrossFitBooker: "end_timestamp": end_date.strftime("%d-%m-%Y") }) - # Debugging logs - # print(f"Request Data: {request_data}") - # print(f"Headers: {self.get_auth_headers()}") - # Add retry logic with exponential backoff for retry in range(RETRY_MAX): try: - response = self.session.post( - url, - headers=self.get_auth_headers(), - data=urlencode(request_data), - timeout=10 - ) + try: + 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}") + 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.ConnectionError, - requests.exceptions.Timeout, - requests.exceptions.ReadTimeout) as e: + except requests.exceptions.JSONDecodeError: + logging.error("Failed to decode JSON response") + return None + except requests.exceptions.RequestException as e: if retry == RETRY_MAX - 1: - raise # Final retry failed, propagate error + logging.error(f"Final retry failed: {str(e)}") + raise # Propagate error wait_time = 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 - print(f"Failed after {RETRY_MAX} attempts") + logging.error(f"Failed after {RETRY_MAX} attempts") return None - - # Debug raw response - # print(f"Response Status Code: {response.status_code}") - # print(f"Response Content: {response.text}") - + # Handle response if response.status_code == 200: try: json_response = response.json() return json_response except ValueError: - print("Failed to decode JSON response") + logging.error("Failed to decode JSON response") return None elif response.status_code == 400: - print("400 Bad Request - likely missing or invalid parameters") - print("Verify these parameters:") - for param, value in request_data.items(): - print(f"- {param}: {value}") + 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: - print("401 Unauthorized - token may be expired or invalid") + 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: - print(f"Unexpected status code: {response.status_code}") + logging.error(f"Unexpected status code: {response.status_code}") return None def book_session(self, session_id: str) -> bool: @@ -248,15 +265,16 @@ class CrossFitBooker: logging.error(f"API returned success:false: {json_response}") return False - logging.error(f"HTTP {response.status_code}: {response.text}") + logging.error(f"HTTP {response.status_code}: {response.text[:100]}") return False - except (requests.exceptions.ConnectionError, - requests.exceptions.Timeout, - requests.exceptions.ReadTimeout) as e: + 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"All {RETRY_MAX} retry attempts failed") - raise + logging.error(f"Final retry failed: {str(e)}") + raise # Propagate error wait_time = 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) @@ -270,6 +288,7 @@ 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") return True # If can_join is False, check if there's a booking window @@ -285,6 +304,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 @@ -295,27 +315,40 @@ class CrossFitBooker: return False def matches_preferred_session(self, session: Dict, current_time: datetime) -> bool: - """Check if session matches one of your preferred sessions.""" + """Check if session matches one of your preferred sessions with fuzzy matching.""" try: session_time = parse(session["start_timestamp"]) if not session_time.tzinfo: session_time = pytz.timezone(TIMEZONE).localize(session_time) - # Get day of week (0=Monday, 6=Sunday) and time day_of_week = session_time.weekday() session_time_str = session_time.strftime("%H:%M") session_name = session.get("name_activity", "").upper() - # Check against preferred sessions 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): # Partial match + preferred_name in session_name): return True + + # Fuzzy match fallback (80% similarity) + ratio = difflib.SequenceMatcher( + None, + session_name.lower(), + preferred_name.lower() + ).ratio() + + if (day_of_week == preferred_day and + abs(session_time.hour - int(preferred_time.split(':')[0])) <= 1 and + 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)}") + logging.error(f"Failed to check session: {str(e)} - Session: {session}") return False def run_booking_cycle(self, current_time: datetime): @@ -327,7 +360,7 @@ class CrossFitBooker: # Get available sessions sessions_data = self.get_available_sessions(start_date, end_date) if not sessions_data or not sessions_data.get("success", False): - print("No sessions available or error fetching sessions") + logging.error("No sessions available or error fetching sessions - Sessions Data: {sessions_data}") return activities = sessions_data.get("data", {}).get("activities_calendar", []) @@ -345,18 +378,18 @@ class CrossFitBooker: sessions_to_book.append(("Available", session)) if not sessions_to_book: - print("No matching sessions found 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_datetime"], "%Y-%m-%d %H:%M:%S") - print(f"Attempting to book {session_type} session at {session_time} ({session['name_activity']})") + logging.info(f"Attempting to book {session_type} session at {session_time} ({session['name_activity']})") if self.book_session(session["id_activity_calendar"]): - print(f"Successfully booked {session_type} session at {session_time}") + logging.info(f"Successfully booked {session_type} session at {session_time}") else: - print(f"Failed to book {session_type} session at {session_time}") + logging.error(f"Failed to book {session_type} session at {session_time} - Session: {session}") def run(self): """Main execution loop""" @@ -369,58 +402,28 @@ class CrossFitBooker: return while True: - current_time = datetime.now(tz) - print(f"\nCurrent time: {current_time}") + try: + current_time = datetime.now(tz) + logging.info(f"Current time: {current_time}") + + # Run booking cycle at the target time or if it's a test + 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) + else: + # Check again in 30 seconds + time.sleep(30) + 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 - # Run booking cycle at the target time or if it's a test - 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) - else: - # Check again in 30 seconds - time.sleep(30) if __name__ == "__main__": booker = CrossFitBooker() if not booker.login(): - print("Failed to login") + logging.error("Failed to login - Traceback: {traceback.format_exc()}") exit(1) - # Set timezone for current_time - tz = pytz.timezone(TIMEZONE) - current_time = datetime.now(tz) - - # Get sessions for the next 7 days - start_date = datetime.now() - end_date = start_date + timedelta(days=3) - - session_data = booker.get_available_sessions(start_date, end_date) - if not session_data or not session_data.get("success", False): - logging.error("Failed to get session data") - exit(1) - - activities = session_data.get("data", {}).get("activities_calendar", []) - - bookable_sessions = [] - for session in activities: - # Assuming the string is stored in a variable named session_time_str - session_time_str = "2025-07-19 20:01:08.858174+02:00" - session_time = datetime.strptime(session_time_str, "%Y-%m-%d %H:%M:%S.%f%z") - - if booker.is_session_bookable(session, current_time): - # if booker.is_session_bookable(session, session_time): - bookable_sessions.append(session) - - # print(f"Bookable sessions: {json.dumps(bookable_sessions, indent=2)}") - print(f"Found {len(bookable_sessions)} sessions to book") - - - for session in bookable_sessions: - is_prefered_session = booker.matches_preferred_session(session, current_time) - # print(session.get("name_activity") + " / " + str(is_prefered_session)) - if is_prefered_session: - booker.book_session(session.get("id_activity_calendar")) - - - + # Start continuous booking loop + booker.run() -- 2.49.1 From 870a65301d70ccb24cc29c5476fa551d5b71bae1 Mon Sep 17 00:00:00 2001 From: kbe Date: Fri, 18 Jul 2025 21:41:16 +0200 Subject: [PATCH 03/11] feat: Only books preferred activities I changed the runner to only book preferred sessions and ignore all others. --- book_crossfit.py | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/book_crossfit.py b/book_crossfit.py index 7f0999b..98279bf 100755 --- a/book_crossfit.py +++ b/book_crossfit.py @@ -29,7 +29,8 @@ 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 = "20:01" # When bookings open (8 PM) +TARGET_RESERVATION_TIME = "21:40" # When bookings open (8 PM) DEVICE_TYPE = "3" # Retry configuration @@ -365,7 +366,7 @@ class CrossFitBooker: activities = sessions_data.get("data", {}).get("activities_calendar", []) - # Find sessions to book (both preferred and any available) + # Find sessions to book (prefered only) sessions_to_book = [] for session in activities: if not self.is_session_bookable(session, current_time): @@ -373,9 +374,6 @@ class CrossFitBooker: if self.matches_preferred_session(session, current_time): sessions_to_book.append(("Preferred", session)) - elif current_time.strftime("%H:%M") == TARGET_RESERVATION_TIME: - # At booking time, consider all available sessions - sessions_to_book.append(("Available", session)) if not sessions_to_book: logging.info("No matching sessions found to book") @@ -384,7 +382,7 @@ class CrossFitBooker: # 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_datetime"], "%Y-%m-%d %H:%M:%S") + session_time = 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}") -- 2.49.1 From cba4299b9a7fcbd80032215b9e80ea477a267ebc Mon Sep 17 00:00:00 2001 From: kbe Date: Sun, 20 Jul 2025 03:06:07 +0200 Subject: [PATCH 04/11] 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 -- 2.49.1 From 4b27a6f4a3a5c1e450f00f6d6df850b29d8fef6a Mon Sep 17 00:00:00 2001 From: kbe Date: Sun, 20 Jul 2025 03:12:22 +0200 Subject: [PATCH 05/11] feat: Program interrupt key quit --- book_crossfit.py | 49 ++++++++++++++++++++++++++++++------------------ 1 file changed, 31 insertions(+), 18 deletions(-) diff --git a/book_crossfit.py b/book_crossfit.py index 2ba345b..b93b376 100755 --- a/book_crossfit.py +++ b/book_crossfit.py @@ -15,7 +15,7 @@ from dateutil.parser import parse import pytz from dotenv import load_dotenv from urllib.parse import urlencode -from typing import List, Dict, Optional +from typing import List, Dict, Optional, Any load_dotenv() @@ -474,23 +474,34 @@ class CrossFitBooker: logging.error("Authentication failed - exiting program") return - while True: - try: - 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 - 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()) - else: - # 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 + try: + while True: + try: + 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 + 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()) + else: + # 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 + 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) if __name__ == "__main__": @@ -501,3 +512,5 @@ if __name__ == "__main__": # Start continuous booking loop booker.run() + logging.info("Script completed") + -- 2.49.1 From 6c8647b11cf0e7eac6388e7319cafed7583283d4 Mon Sep 17 00:00:00 2001 From: kbe Date: Sun, 20 Jul 2025 03:27:13 +0200 Subject: [PATCH 06/11] feat: Prepare for notification trough mail and Telegram --- requirements.txt | 1 + session_notifier.py | 76 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 77 insertions(+) create mode 100644 session_notifier.py diff --git a/requirements.txt b/requirements.txt index 5b5bc16..95da2be 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,6 +8,7 @@ pytz==2025.2 requests==2.32.4 setuptools==80.9.0 six==1.17.0 +telegram==0.0.1 typing==3.7.4.3 urllib3==2.5.0 zope.interface==7.2 diff --git a/session_notifier.py b/session_notifier.py new file mode 100644 index 0000000..f6b063d --- /dev/null +++ b/session_notifier.py @@ -0,0 +1,76 @@ +import smtplib +from email.message import EmailMessage +from telegram import Bot + +class SessionNotifier: + """ + A class to handle notifications for session bookings. + Supports sending notifications via email and Telegram. + + Attributes: + email_credentials (dict): Dictionary containing email credentials + (e.g., 'from', 'to', 'password') + telegram_credentials (dict): Dictionary containing Telegram credentials + (e.g., 'token', 'chat_id') + """ + + def __init__(self, email_credentials, telegram_credentials): + """ + Initialize the SessionNotifier with email and Telegram credentials. + + Args: + email_credentials (dict): Email credentials for authentication + telegram_credentials (dict): Telegram credentials for authentication + """ + self.email_credentials = email_credentials + self.telegram_credentials = telegram_credentials + + def send_email_notification(self, message): + """ + Send an email notification with the given message. + + Args: + message (str): The message content to be sent in the email + """ + # Create an EmailMessage object + email = EmailMessage() + email.set_content(message) + + # Set the email sender and recipient + email['From'] = self.email_credentials['from'] + email['To'] = self.email_credentials['to'] + + # Set the email subject + email['Subject'] = 'Session Booking Notification' + + # Send the email using smtplib + with smtplib.SMTP_SSL('smtp.gmail.com', 465) as smtp: + smtp.login(self.email_credentials['from'], self.email_credentials['password']) + smtp.send_message(email) + + def send_telegram_notification(self, message): + """ + Send a Telegram notification with the given message. + + Args: + message (str): The message content to be sent in the Telegram chat + """ + # Create a Bot instance with the provided token + bot = Bot(token=self.telegram_credentials['token']) + # Send the message to the specified chat ID + bot.send_message(chat_id=self.telegram_credentials['chat_id'], text=message) + + def notify_session_booking(self, session_details): + """ + Notify about a session booking via email and Telegram. + + Args: + session_details (str): Details about the booked session + """ + # Create messages for both email and Telegram + email_message = f"Session booked: {session_details}" + telegram_message = f"Session booked: {session_details}" + + # Send notifications through both channels + self.send_email_notification(email_message) + self.send_telegram_notification(telegram_message) \ No newline at end of file -- 2.49.1 From 1fd446331496a25cca31e762a52ac1a8a7a0e198 Mon Sep 17 00:00:00 2001 From: kbe Date: Sun, 20 Jul 2025 03:43:27 +0200 Subject: [PATCH 07/11] chore: moved out classes to files --- .env.example | 7 + .gitignore | 177 +++++++++++++++ book_crossfit.py | 510 +----------------------------------------- crossfit_booker.py | 523 ++++++++++++++++++++++++++++++++++++++++++++ requirements.txt | 6 + session_notifier.py | 2 +- 6 files changed, 718 insertions(+), 507 deletions(-) create mode 100644 crossfit_booker.py diff --git a/.env.example b/.env.example index cfc1581..4424a4a 100644 --- a/.env.example +++ b/.env.example @@ -1,3 +1,10 @@ # Configuration CROSSFIT_USERNAME=Kevin8407 CROSSFIT_PASSWORD=9vx03OSE + +EMAIL_FROM=kevin@mistergeek.fr +EMAIL_TO=kbataille@vivaldi.net +EMAIL_PASSWORD= + +TELEGRAM_TOKEN= +TELEGRAM_CHAT_ID \ No newline at end of file diff --git a/.gitignore b/.gitignore index 25c7260..c925db7 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,180 @@ .env log/ !log/.gitkeep + +# Created by https://www.toptal.com/developers/gitignore/api/python +# Edit at https://www.toptal.com/developers/gitignore?templates=python + +### Python ### +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ +cover/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +.pybuilder/ +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +# For a library or package, you might want to ignore these files since the code is +# intended to run in multiple environments; otherwise, check them in: +# .python-version + +# pipenv +# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. +# However, in case of collaboration, if having platform-specific dependencies or dependencies +# having no cross-platform support, pipenv may install dependencies that don't work, or not +# install all needed dependencies. +#Pipfile.lock + +# poetry +# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control +#poetry.lock + +# pdm +# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. +#pdm.lock +# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it +# in version control. +# https://pdm.fming.dev/#use-with-ide +.pdm.toml + +# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# pytype static type analyzer +.pytype/ + +# Cython debug symbols +cython_debug/ + +# PyCharm +# JetBrains specific template is maintained in a separate JetBrains.gitignore that can +# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore +# and can be added to the global gitignore or merged into this file. For a more nuclear +# option (not recommended) you can uncomment the following to ignore the entire idea folder. +#.idea/ + +### Python Patch ### +# Poetry local configuration file - https://python-poetry.org/docs/configuration/#local-configuration +poetry.toml + +# ruff +.ruff_cache/ + +# LSP config files +pyrightconfig.json + +# End of https://www.toptal.com/developers/gitignore/api/python diff --git a/book_crossfit.py b/book_crossfit.py index b93b376..9d98b3b 100755 --- a/book_crossfit.py +++ b/book_crossfit.py @@ -1,516 +1,14 @@ #!/usr/bin/env python3 - -# Native modules -import json import logging -from datetime import datetime, timedelta -import os -import sys -import time -import difflib - -# Third-party modules -import requests -from dateutil.parser import parse -import pytz -from dotenv import load_dotenv -from urllib.parse import urlencode -from typing import List, Dict, Optional, Any - -load_dotenv() - -# Configuration -USERNAME = os.environ.get("CROSSFIT_USERNAME") -PASSWORD = os.environ.get("CROSSFIT_PASSWORD") - -if not all([USERNAME, PASSWORD]): - raise ValueError("Missing environment variables: CROSSFIT_USERNAME and/or CROSSFIT_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) -DEVICE_TYPE = "3" - -# Retry configuration -RETRY_MAX = 3 -RETRY_BACKOFF = 1 -APP_VERSION = "5.09.21" - -# Define your preferred sessions -# Format: List of tuples (day_of_week, start_time, session_name_contains) -# day_of_week: 0=Monday, 6=Sunday -PREFERRED_SESSIONS = [ - (4, "17:00", "WEIGHTLIFTING"), # Friday 17:00 WEIGHTLIFTING - (5, "12:30", "HYROX"), # Saturday 12:30 HYROX - (2, "18:30", "CONDITIONING"), # Wednesday 18:30 CONDITIONING - # (5, "15:15", "BIG WOD") # Saturday 15:15 BIG WOD -] - -# Configure logging once at script startup -logging.basicConfig( - level=logging.DEBUG, # Change to DEBUG for more detailed logs - format='%(asctime)s - %(levelname)s - %(message)s', - handlers=[ - logging.FileHandler("log/crossfit_booking.log"), - logging.StreamHandler() - ] -) -logging.getLogger("requests").setLevel(logging.WARNING) -logging.info("Logging enhanced with request library noise reduction") - - -class CrossFitBooker: - 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", - } - self.session.headers.update(self.base_headers) - - # Define mandatory parameters for API calls - 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[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. - - Returns: - bool: True if login is successful, False otherwise. - """ - try: - # First login endpoint - login_params: Dict[str, str] = { - "app_version": APP_VERSION, - "device_type": DEVICE_TYPE, - "username": USERNAME, - "password": PASSWORD - } - - 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: 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}") - return False - - # Second login endpoint - 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({ - "device_type": DEVICE_TYPE, - "username": USERNAME, - "password": PASSWORD - })) - - if response.ok: - 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}") - 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 - except requests.exceptions.RequestException as e: - logging.error(f"Request error during login: {str(e)}") - return False - except Exception as e: - 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]]: - """ - 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: str = "https://sport.nubapp.com/api/v4/activities/getActivitiesCalendar.php" - - # Prepare request with mandatory parameters - 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 and more informative error messages - 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}") - 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)}") - 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 - 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 - def book_session(self, session_id: str) -> bool: - """ - 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[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, - "id_user": self.user_id, - "action_by": self.user_id, - "n_guests": "0", - "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( - url, - headers=self.get_auth_headers(), - data=urlencode(data), - timeout=10 - ) - - if response.status_code == 200: - json_response: Dict[str, Any] = response.json() - if json_response.get("success", False): - logging.info(success_msg) - return True - logging.error(f"API returned success:false: {json_response}") - return False - - 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 - 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[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: 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: - 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 - - # Default case: not bookable - return False - - 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: datetime = parse(session["start_timestamp"]) - if not session_time.tzinfo: - session_time = pytz.timezone(TIMEZONE).localize(session_time) - - 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: float = difflib.SequenceMatcher( - None, - session_name.lower(), - preferred_name.lower() - ).ratio() - - if (day_of_week == preferred_day and - abs(session_time.hour - int(preferred_time.split(':')[0])) <= 1 and - 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) -> 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: 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: List[Dict[str, Any]] = sessions_data.get("data", {}).get("activities_calendar", []) - - # Find sessions to book (prefered only) - 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 = 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) -> None: - """ - Main execution loop. - """ - # Set up timezone - tz: pytz.timezone = pytz.timezone(TIMEZONE) - - # 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}") - - # 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 until the next booking window - wait_until = current_time + timedelta(minutes=60) - time.sleep((wait_until - current_time).total_seconds()) - else: - # 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 - 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) - +import traceback +from crossfit_booker import CrossFitBooker if __name__ == "__main__": booker = CrossFitBooker() if not booker.login(): - logging.error("Failed to login - Traceback: {traceback.format_exc()}") + logging.error("Failed to login - Traceback: %s", traceback.format_exc()) exit(1) - + # Start continuous booking loop booker.run() logging.info("Script completed") - diff --git a/crossfit_booker.py b/crossfit_booker.py new file mode 100644 index 0000000..f9b69f6 --- /dev/null +++ b/crossfit_booker.py @@ -0,0 +1,523 @@ +# Native modules +import json +import logging +from datetime import datetime, timedelta +import os +import sys +import time +import difflib + +# Third-party modules +import requests +from dateutil.parser import parse +import pytz +from dotenv import load_dotenv +from urllib.parse import urlencode +from typing import List, Dict, Optional, Any + +# Import the SessionNotifier class +from session_notifier import SessionNotifier + +load_dotenv() + +# Configuration +USERNAME = os.environ.get("CROSSFIT_USERNAME") +PASSWORD = os.environ.get("CROSSFIT_PASSWORD") + +if not all([USERNAME, PASSWORD]): + raise ValueError("Missing environment variables: CROSSFIT_USERNAME and/or CROSSFIT_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) +DEVICE_TYPE = "3" + +# Retry configuration +RETRY_MAX = 3 +RETRY_BACKOFF = 1 +APP_VERSION = "5.09.21" + +# Define your preferred sessions +# Format: List of tuples (day_of_week, start_time, session_name_contains) +# day_of_week: 0=Monday, 6=Sunday +PREFERRED_SESSIONS = [ + (4, "17:00", "WEIGHTLIFTING"), # Friday 17:00 WEIGHTLIFTING + (5, "12:30", "HYROX"), # Saturday 12:30 HYROX + (2, "18:30", "CONDITIONING"), # Wednesday 18:30 CONDITIONING + # (5, "15:15", "BIG WOD") # Saturday 15:15 BIG WOD +] + +# Configure logging once at script startup +logging.basicConfig( + level=logging.DEBUG, # Change to DEBUG for more detailed logs + format='%(asctime)s - %(levelname)s - %(message)s', + handlers=[ + logging.FileHandler("log/crossfit_booking.log"), + logging.StreamHandler() + ] +) + +logging.getLogger("requests").setLevel(logging.WARNING) +logging.info("Logging enhanced with request library noise reduction") + +class CrossFitBooker: + 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", + } + self.session.headers.update(self.base_headers) + + # Define mandatory parameters for API calls + self.mandatory_params: Dict[str, str] = { + "app_version": APP_VERSION, + "device_type": DEVICE_TYPE, + "id_application": APPLICATION_ID, + "id_category_activity": CATEGORY_ID + } + + # Initialize the SessionNotifier with credentials from environment variables + email_credentials = { + "from": os.environ.get("EMAIL_FROM"), + "to": os.environ.get("EMAIL_TO"), + "password": os.environ.get("EMAIL_PASSWORD") + } + + telegram_credentials = { + "token": os.environ.get("TELEGRAM_TOKEN"), + "chat_id": os.environ.get("TELEGRAM_CHAT_ID") + } + + self.notifier = SessionNotifier(email_credentials, telegram_credentials) + + 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. + + Returns: + bool: True if login is successful, False otherwise. + """ + try: + # First login endpoint + login_params: Dict[str, str] = { + "app_version": APP_VERSION, + "device_type": DEVICE_TYPE, + "username": USERNAME, + "password": PASSWORD + } + + 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: 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}") + return False + + # Second login endpoint + 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({ + "device_type": DEVICE_TYPE, + "username": USERNAME, + "password": PASSWORD + })) + + if response.ok: + 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}") + 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 + except requests.exceptions.RequestException as e: + logging.error(f"Request error during login: {str(e)}") + return False + except Exception as e: + 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]]: + """ + 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: str = "https://sport.nubapp.com/api/v4/activities/getActivitiesCalendar.php" + + # Prepare request with mandatory parameters + 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 and more informative error messages + 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}") + 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)}") + 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 + 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 + + def book_session(self, session_id: str) -> bool: + """ + 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[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, + "id_user": self.user_id, + "action_by": self.user_id, + "n_guests": "0", + "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( + url, + headers=self.get_auth_headers(), + data=urlencode(data), + timeout=10 + ) + + if response.status_code == 200: + json_response: Dict[str, Any] = response.json() + if json_response.get("success", False): + logging.info(success_msg) + return True + logging.error(f"API returned success:false: {json_response}") + return False + + 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 + 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[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: 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: + 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 + + # Default case: not bookable + return False + + 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: datetime = parse(session["start_timestamp"]) + if not session_time.tzinfo: + session_time = pytz.timezone(TIMEZONE).localize(session_time) + + 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: float = difflib.SequenceMatcher( + None, + session_name.lower(), + preferred_name.lower() + ).ratio() + + if (day_of_week == preferred_day and + abs(session_time.hour - int(preferred_time.split(':')[0])) <= 1 and + 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) -> 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: 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: List[Dict[str, Any]] = sessions_data.get("data", {}).get("activities_calendar", []) + + # Find sessions to book (prefered only) + 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 = 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"]): + # 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) + 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: + """ + Main execution loop. + """ + # Set up timezone + tz: pytz.timezone = pytz.timezone(TIMEZONE) + + # 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}") + + # 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 until the next booking window + wait_until = current_time + timedelta(minutes=60) + time.sleep((wait_until - current_time).total_seconds()) + else: + # 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 + 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) \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 95da2be..5e5e25f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,13 +1,19 @@ +anyio==4.9.0 certifi==2025.7.14 charset-normalizer==3.4.2 DateTime==5.5 +h11==0.16.0 +httpcore==1.0.9 +httpx==0.28.1 idna==3.10 python-dateutil==2.9.0.post0 python-dotenv==1.1.1 +python-telegram-bot==22.2 pytz==2025.2 requests==2.32.4 setuptools==80.9.0 six==1.17.0 +sniffio==1.3.1 telegram==0.0.1 typing==3.7.4.3 urllib3==2.5.0 diff --git a/session_notifier.py b/session_notifier.py index f6b063d..40d590d 100644 --- a/session_notifier.py +++ b/session_notifier.py @@ -56,7 +56,7 @@ class SessionNotifier: message (str): The message content to be sent in the Telegram chat """ # Create a Bot instance with the provided token - bot = Bot(token=self.telegram_credentials['token']) + bot = Bot(token=self.telegram_credentials['token'], base_url=self.telegram_credentials.get('base_url', 'https://api.telegram.org')) # Send the message to the specified chat ID bot.send_message(chat_id=self.telegram_credentials['chat_id'], text=message) -- 2.49.1 From 35160bc0330481981a37f44f66db742fb636d1d5 Mon Sep 17 00:00:00 2001 From: kbe Date: Sun, 20 Jul 2025 03:47:12 +0200 Subject: [PATCH 08/11] feat: Enable (or not) email or Telegram notifications --- .env.example | 4 ++++ crossfit_booker.py | 13 +++++++++++-- session_notifier.py | 18 ++++++++++++++---- 3 files changed, 29 insertions(+), 6 deletions(-) diff --git a/.env.example b/.env.example index 4424a4a..020e235 100644 --- a/.env.example +++ b/.env.example @@ -2,6 +2,10 @@ CROSSFIT_USERNAME=Kevin8407 CROSSFIT_PASSWORD=9vx03OSE +# Notification settings +ENABLE_EMAIL_NOTIFICATIONS=true +ENABLE_TELEGRAM_NOTIFICATIONS=true + EMAIL_FROM=kevin@mistergeek.fr EMAIL_TO=kbataille@vivaldi.net EMAIL_PASSWORD= diff --git a/crossfit_booker.py b/crossfit_booker.py index f9b69f6..357c180 100644 --- a/crossfit_booker.py +++ b/crossfit_booker.py @@ -91,13 +91,22 @@ class CrossFitBooker: "to": os.environ.get("EMAIL_TO"), "password": os.environ.get("EMAIL_PASSWORD") } - + telegram_credentials = { "token": os.environ.get("TELEGRAM_TOKEN"), "chat_id": os.environ.get("TELEGRAM_CHAT_ID") } - self.notifier = SessionNotifier(email_credentials, telegram_credentials) + # Get notification settings from environment variables + enable_email = os.environ.get("ENABLE_EMAIL_NOTIFICATIONS", "true").lower() in ("true", "1", "yes") + enable_telegram = os.environ.get("ENABLE_TELEGRAM_NOTIFICATIONS", "true").lower() in ("true", "1", "yes") + + self.notifier = SessionNotifier( + email_credentials, + telegram_credentials, + enable_email=enable_email, + enable_telegram=enable_telegram + ) def get_auth_headers(self) -> Dict[str, str]: """ diff --git a/session_notifier.py b/session_notifier.py index 40d590d..5aba62f 100644 --- a/session_notifier.py +++ b/session_notifier.py @@ -1,4 +1,5 @@ import smtplib +import os from email.message import EmailMessage from telegram import Bot @@ -12,18 +13,24 @@ class SessionNotifier: (e.g., 'from', 'to', 'password') telegram_credentials (dict): Dictionary containing Telegram credentials (e.g., 'token', 'chat_id') + enable_email (bool): Whether to enable email notifications + enable_telegram (bool): Whether to enable Telegram notifications """ - def __init__(self, email_credentials, telegram_credentials): + def __init__(self, email_credentials, telegram_credentials, enable_email=True, enable_telegram=True): """ Initialize the SessionNotifier with email and Telegram credentials. Args: email_credentials (dict): Email credentials for authentication telegram_credentials (dict): Telegram credentials for authentication + enable_email (bool): Whether to enable email notifications + enable_telegram (bool): Whether to enable Telegram notifications """ self.email_credentials = email_credentials self.telegram_credentials = telegram_credentials + self.enable_email = enable_email + self.enable_telegram = enable_telegram def send_email_notification(self, message): """ @@ -71,6 +78,9 @@ class SessionNotifier: email_message = f"Session booked: {session_details}" telegram_message = f"Session booked: {session_details}" - # Send notifications through both channels - self.send_email_notification(email_message) - self.send_telegram_notification(telegram_message) \ No newline at end of file + # Send notifications through enabled channels + if self.enable_email: + self.send_email_notification(email_message) + + if self.enable_telegram: + self.send_telegram_notification(telegram_message) \ No newline at end of file -- 2.49.1 From be27046d6c8d9495720f12b96249b8006b868490 Mon Sep 17 00:00:00 2001 From: kbe Date: Sun, 20 Jul 2025 14:19:20 +0200 Subject: [PATCH 09/11] Add email credentials --- .env.example | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/.env.example b/.env.example index 020e235..78104a7 100644 --- a/.env.example +++ b/.env.example @@ -4,11 +4,11 @@ CROSSFIT_PASSWORD=9vx03OSE # Notification settings ENABLE_EMAIL_NOTIFICATIONS=true -ENABLE_TELEGRAM_NOTIFICATIONS=true +ENABLE_TELEGRAM_NOTIFICATIONS=false -EMAIL_FROM=kevin@mistergeek.fr +EMAIL_FROM=no-reply@cyanet.fr EMAIL_TO=kbataille@vivaldi.net -EMAIL_PASSWORD= +EMAIL_PASSWORD=BV3GzqHjsSx5A6TE TELEGRAM_TOKEN= -TELEGRAM_CHAT_ID \ No newline at end of file +TELEGRAM_CHAT_ID -- 2.49.1 From 28b8d57b279a79a631c566a64e62792a9ee4297e Mon Sep 17 00:00:00 2001 From: kbe Date: Sun, 20 Jul 2025 15:44:52 +0200 Subject: [PATCH 10/11] fix: Use SMTP server instead of gmail The script was using gmail as SMTP server. Or the script uses a custom SMTP one. --- .env.example | 19 +++++++++++-------- .gitignore | 6 ++++++ book_crossfit.py | 2 +- session_notifier.py | 23 ++++++++++++++++++++--- 4 files changed, 38 insertions(+), 12 deletions(-) diff --git a/.env.example b/.env.example index 78104a7..036cc04 100644 --- a/.env.example +++ b/.env.example @@ -1,14 +1,17 @@ -# Configuration -CROSSFIT_USERNAME=Kevin8407 -CROSSFIT_PASSWORD=9vx03OSE +# CrossFit booking credentials +CROSSFIT_USERNAME=your_username +CROSSFIT_PASSWORD=your_password # Notification settings ENABLE_EMAIL_NOTIFICATIONS=true ENABLE_TELEGRAM_NOTIFICATIONS=false -EMAIL_FROM=no-reply@cyanet.fr -EMAIL_TO=kbataille@vivaldi.net -EMAIL_PASSWORD=BV3GzqHjsSx5A6TE +# Email notification credentials +SMTP_SERVER=mail.infomaniak.com +EMAIL_FROM=your_email +EMAIL_TO=recipient_email +EMAIL_PASSWORD=email_password -TELEGRAM_TOKEN= -TELEGRAM_CHAT_ID +# Telegram notification credentials +TELEGRAM_TOKEN=your_telegram_token +TELEGRAM_CHAT_ID=your_chat_id diff --git a/.gitignore b/.gitignore index c925db7..4544309 100644 --- a/.gitignore +++ b/.gitignore @@ -178,3 +178,9 @@ poetry.toml pyrightconfig.json # End of https://www.toptal.com/developers/gitignore/api/python + +# Docker +docker-compose.override.yml +docker-compose.yml +Dockerfile +.dockerignore diff --git a/book_crossfit.py b/book_crossfit.py index 9d98b3b..584a109 100755 --- a/book_crossfit.py +++ b/book_crossfit.py @@ -11,4 +11,4 @@ if __name__ == "__main__": # Start continuous booking loop booker.run() - logging.info("Script completed") + logging.info("Script completed") \ No newline at end of file diff --git a/session_notifier.py b/session_notifier.py index 5aba62f..924cd6d 100644 --- a/session_notifier.py +++ b/session_notifier.py @@ -1,8 +1,10 @@ import smtplib import os +import logging from email.message import EmailMessage from telegram import Bot + class SessionNotifier: """ A class to handle notifications for session bookings. @@ -39,6 +41,9 @@ class SessionNotifier: Args: message (str): The message content to be sent in the email """ + logging.debug("Sending email notification") + logging.debug(f"Email credentials: {self.email_credentials}") + # Create an EmailMessage object email = EmailMessage() email.set_content(message) @@ -51,9 +56,21 @@ class SessionNotifier: email['Subject'] = 'Session Booking Notification' # Send the email using smtplib - with smtplib.SMTP_SSL('smtp.gmail.com', 465) as smtp: - smtp.login(self.email_credentials['from'], self.email_credentials['password']) - smtp.send_message(email) + try: + smtp_server = os.environ.get("SMTP_SERVER") + if not smtp_server: + logging.error("SMTP server not configured in environment variables") + raise ValueError("SMTP server not configured") + + with smtplib.SMTP_SSL(smtp_server, 465) as smtp: + logging.debug(f"Connecting to SMTP server: {smtp_server}") + smtp.login(self.email_credentials['from'], self.email_credentials['password']) + logging.debug("Logged in to SMTP server") + smtp.send_message(email) + logging.debug("Email sent successfully") + except Exception as e: + logging.error(f"Failed to send email: {str(e)}") + raise def send_telegram_notification(self, message): """ -- 2.49.1 From 5dcc2a89ae5e29920246b8c66d4c28577becd8a9 Mon Sep 17 00:00:00 2001 From: kbe Date: Sun, 20 Jul 2025 16:17:15 +0200 Subject: [PATCH 11/11] feat: Add notification for upcoming sessions --- README.md | 92 +++++++++++++++++++++++++++++++++++++++++++++ crossfit_booker.py | 57 ++++++++++++++++++++++------ session_notifier.py | 19 ++++++++++ 3 files changed, 157 insertions(+), 11 deletions(-) create mode 100644 README.md diff --git a/README.md b/README.md new file mode 100644 index 0000000..d68a12d --- /dev/null +++ b/README.md @@ -0,0 +1,92 @@ +# Crossfit Application + +This is a Python application for managing Crossfit bookings and notifications. The application automates the process of booking Crossfit sessions and sends notifications via email and Telegram when a booking is successful. + +## Features + +- Automated booking of Crossfit sessions +- Email and Telegram notifications for successful bookings +- Configurable preferred sessions +- Retry logic for booking failures +- Detailed logging + +## Prerequisites + +- Docker +- Docker Compose + +## Setup + +1. Create a `.env` file based on `.env.example` and fill in the required credentials. +2. Build and run the application using Docker Compose: + +```bash +docker-compose up --build +``` + +3. The application will run in a Docker container, and the logs will be stored in the `./log` directory. + +## Usage + +The application will automatically check for available sessions and book them based on your preferences. It will send notifications via email and Telegram when a booking is successful. + +### Environment Variables + +The following environment variables are required: + +- `CROSSFIT_USERNAME`: Your Crossfit username +- `CROSSFIT_PASSWORD`: Your Crossfit password +- `EMAIL_FROM`: Your email address +- `EMAIL_TO`: Recipient email address +- `EMAIL_PASSWORD`: Your email password +- `TELEGRAM_TOKEN`: Your Telegram bot token +- `TELEGRAM_CHAT_ID`: Your Telegram chat ID + +### Preferred Sessions + +You can configure your preferred sessions in the `crossfit_booker.py` file. The preferred sessions are defined as a list of tuples, where each tuple contains the day of the week, start time, and session name. + +```python +PREFERRED_SESSIONS = [ + (4, "17:00", "WEIGHTLIFTING"), # Friday 17:00 WEIGHTLIFTING + (5, "12:30", "HYROX"), # Saturday 12:30 HYROX + (2, "18:30", "CONDITIONING"), # Wednesday 18:30 CONDITIONING +] +``` + +## Files + +- `Dockerfile`: Docker image definition +- `docker-compose.yml`: Docker Compose service definition +- `.env.example`: Example environment variables file +- `.dockerignore`: Docker ignore file +- `.gitignore`: Git ignore file +- `book_crossfit.py`: Main application script +- `crossfit_booker.py`: Crossfit booking script +- `session_notifier.py`: Session notification script +- `requirements.txt`: Python dependencies + +## Project Structure + +``` +. +├── Dockerfile +├── docker-compose.yml +├── .env.example +├── .dockerignore +├── .gitignore +├── book_crossfit.py +├── crossfit_booker.py +├── session_notifier.py +├── requirements.txt +└── log + └── crossfit_booking.log +``` + +## Contributing + +Contributions are welcome! Please open an issue or submit a pull request. + +## License + +This project is licensed under the MIT License. See the [LICENSE](LICENSE) file for details. \ No newline at end of file diff --git a/crossfit_booker.py b/crossfit_booker.py index 357c180..d17a9b1 100644 --- a/crossfit_booker.py +++ b/crossfit_booker.py @@ -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()) diff --git a/session_notifier.py b/session_notifier.py index 924cd6d..eadb98b 100644 --- a/session_notifier.py +++ b/session_notifier.py @@ -95,6 +95,25 @@ class SessionNotifier: email_message = f"Session booked: {session_details}" telegram_message = f"Session booked: {session_details}" + # Send notifications through enabled channels + if self.enable_email: + self.send_email_notification(email_message) + + if self.enable_telegram: + self.send_telegram_notification(telegram_message) + + def notify_upcoming_session(self, session_details, days_until): + """ + Notify about an upcoming session via email and Telegram. + + Args: + session_details (str): Details about the upcoming session + days_until (int): Number of days until the session + """ + # Create messages for both email and Telegram + email_message = f"Session available soon: {session_details} (in {days_until} days)" + telegram_message = f"Session available soon: {session_details} (in {days_until} days)" + # Send notifications through enabled channels if self.enable_email: self.send_email_notification(email_message) -- 2.49.1