It looks like it works?
This commit is contained in:
kbe
177
book_crossfit.py
177
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."""
|
||||
|
||||
Reference in New Issue
Block a user