323 lines
13 KiB
Python
Executable File
323 lines
13 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
import requests
|
|
import json
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
import pytz
|
|
from urllib.parse import urlencode
|
|
from typing import List, Dict, Optional
|
|
|
|
# Configuration
|
|
USERNAME = "Kevin8407"
|
|
PASSWORD = "9vx03OSE"
|
|
APPLICATION_ID = "81560887"
|
|
CATEGORY_ID = "677" # Activity category ID for CrossFit
|
|
TIMEZONE = "Europe/Paris" # Adjust to your timezone
|
|
TARGET_RESERVATION_TIME = "20:00" # When bookings open (8 PM)
|
|
DEVICE_TYPE = "3"
|
|
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
|
|
]
|
|
|
|
class CrossFitBooker:
|
|
def __init__(self):
|
|
self.auth_token = None
|
|
self.user_id = None
|
|
self.session = requests.Session()
|
|
self.base_headers = {
|
|
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:140.0) Gecko/20100101 Firefox/140.0",
|
|
"Accept": "application/json, text/plain, */*",
|
|
"Accept-Language": "en-GB,en;q=0.8,fr-FR;q=0.5,fr;q=0.3",
|
|
"Nubapp-Origin": "user_apps",
|
|
"Origin": "https://box.resawod.com",
|
|
"Referer": "https://box.resawod.com/",
|
|
}
|
|
self.session.headers.update(self.base_headers)
|
|
|
|
# Define mandatory parameters for API calls
|
|
self.mandatory_params = {
|
|
"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()
|
|
if self.auth_token:
|
|
headers["Authorization"] = f"Bearer {self.auth_token}"
|
|
return headers
|
|
|
|
def login(self) -> bool:
|
|
"""Authenticate and get the bearer token"""
|
|
try:
|
|
# First login endpoint
|
|
login_params = {
|
|
"app_version": APP_VERSION,
|
|
"device_type": DEVICE_TYPE,
|
|
"username": USERNAME,
|
|
"password": PASSWORD
|
|
}
|
|
|
|
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:
|
|
print(f"First login step failed: {response.status_code} - {response.text}")
|
|
return False
|
|
|
|
login_data = response.json()
|
|
self.user_id = str(login_data["data"]["user"]["id_user"])
|
|
|
|
# Second login endpoint
|
|
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:
|
|
login_data = response.json()
|
|
self.auth_token = login_data.get("token")
|
|
|
|
if self.auth_token and self.user_id:
|
|
print("Successfully logged in")
|
|
return True
|
|
print(f"Login failed: {response.status_code} - {response.text}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"Login error: {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")
|
|
return None
|
|
|
|
url = "https://sport.nubapp.com/api/v4/activities/getActivitiesCalendar.php"
|
|
|
|
# Prepare request with mandatory parameters
|
|
request_data = 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")
|
|
})
|
|
|
|
try:
|
|
# Debug output
|
|
print("\n--- Request Details ---")
|
|
print(f"URL: {url}")
|
|
print(f"Headers: {json.dumps(self.get_auth_headers(), indent=2)}")
|
|
print(f"Payload: {request_data}")
|
|
|
|
# Make the request
|
|
response = self.session.post(
|
|
url,
|
|
headers=self.get_auth_headers(),
|
|
data=urlencode(request_data),
|
|
timeout=10
|
|
)
|
|
|
|
# Debug raw response
|
|
print("\n--- Response ---")
|
|
print(f"Status Code: {response.status_code}")
|
|
print(f"Headers: {response.headers}")
|
|
print(f"Content: {response.text}")
|
|
|
|
# Handle response
|
|
if response.status_code == 200:
|
|
try:
|
|
json_response = response.json()
|
|
if json_response.get("success", False):
|
|
return json_response
|
|
else:
|
|
print(f"API reported failure: {json_response.get('message')}")
|
|
return None
|
|
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)}")
|
|
return None
|
|
except Exception as e:
|
|
print(f"Unexpected error: {str(e)}")
|
|
return None
|
|
|
|
def book_session(self, session_id: str) -> bool:
|
|
"""Book a specific session"""
|
|
if not self.auth_token or not self.user_id:
|
|
print("Not authenticated")
|
|
return False
|
|
|
|
try:
|
|
# Prepare request with mandatory parameters
|
|
request_data = self.mandatory_params.copy()
|
|
request_data.update({
|
|
"id_user": self.user_id,
|
|
"id_activity": session_id
|
|
})
|
|
|
|
response = self.session.post(
|
|
"https://sport.nubapp.com/api/v4/activities/bookActivity.php",
|
|
headers=self.get_auth_headers(),
|
|
data=urlencode(request_data))
|
|
|
|
if response.ok:
|
|
print(f"Successfully booked session {session_id}")
|
|
return True
|
|
else:
|
|
print(f"Error booking session {session_id}: {response.status_code} - {response.text}")
|
|
return False
|
|
except Exception as e:
|
|
print(f"Error booking session: {str(e)}")
|
|
return False
|
|
|
|
def is_session_bookable(self, session: Dict, current_time: datetime) -> bool:
|
|
"""Check if a session is bookable based on user_info"""
|
|
user_info = session.get("user_info", {})
|
|
|
|
# First check if can_join is true
|
|
if user_info.get("can_join", False):
|
|
return True
|
|
|
|
# Otherwise check booking availability time
|
|
booking_date_str = user_info.get("unableToBookUntilDate", "")
|
|
booking_time_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}", "%d-%m-%Y %H:%M")
|
|
booking_datetime = pytz.timezone(TIMEZONE).localize(booking_datetime)
|
|
return current_time >= booking_datetime
|
|
except ValueError:
|
|
pass
|
|
return False
|
|
|
|
def matches_preferred_session(self, session: Dict, current_time: datetime) -> bool:
|
|
"""Check if session matches one of your preferred sessions"""
|
|
try:
|
|
session_time = datetime.strptime(session["start_datetime"], "%Y-%m-%d %H:%M:%S")
|
|
session_time = pytz.timezone(TIMEZONE).localize(session_time)
|
|
|
|
# Check if session is exactly 2 days from now
|
|
two_days_from_now = current_time + timedelta(days=2)
|
|
if session_time.date() != two_days_from_now.date():
|
|
return False
|
|
|
|
# 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:
|
|
if (day_of_week == preferred_day and
|
|
session_time_str == preferred_time and
|
|
preferred_name in session_name):
|
|
return True
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"Error checking session: {str(e)}")
|
|
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)
|
|
|
|
# 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")
|
|
return
|
|
|
|
activities = sessions_data.get("data", {}).get("activities_calendar", [])
|
|
|
|
# Find sessions to book (both preferred and any available)
|
|
sessions_to_book = []
|
|
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))
|
|
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:
|
|
print("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']})")
|
|
if self.book_session(session["id_activity_calendar"]):
|
|
print(f"Successfully booked {session_type} session at {session_time}")
|
|
else:
|
|
print(f"Failed to book {session_type} session at {session_time}")
|
|
|
|
def run(self):
|
|
"""Main execution loop"""
|
|
# Set up timezone
|
|
tz = pytz.timezone(TIMEZONE)
|
|
|
|
# Initial login
|
|
if not self.login():
|
|
print("Failed to login, exiting")
|
|
return
|
|
|
|
while True:
|
|
current_time = datetime.now(tz)
|
|
print(f"\nCurrent 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)
|
|
|
|
if __name__ == "__main__":
|
|
booker = CrossFitBooker()
|
|
booker.login()
|
|
sessions = booker.get_available_sessions(datetime.strptime("21-07-2025", "%d-%m-%Y"), datetime.strptime("27-07-2025", "%d-%m-%Y"))
|
|
# print(sessions)
|
|
# booker.run_booking_cycle(datetime.now())
|
|
# booker.run() |