crossfit/book_crossfit.py

#!/usr/bin/env python3

import requests
import json
import time
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

# 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
    # (5, "15:15", "BIG WOD")       # Saturday 15:15 BIG WOD
]

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",
            "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 = {
            "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")
        })

        # Debugging logs
        # 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)}")
            return None
        except Exception as e:
            print(f"Unexpected error: {str(e)}")
            return None

        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()
                    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)}")
            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, ignoring error codes."""
        user_info = session.get("user_info", {})

        # First check if can_join is true (primary condition)
        if user_info.get("can_join", False):
            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", "")
        
        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)
                
                if current_time >= 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, current_time: datetime) -> bool:
        """Check if session matches one of your preferred sessions."""
        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:
                if (day_of_week == preferred_day and
                    session_time_str == preferred_time and
                    preferred_name in session_name):  # Partial match
                    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()
    if not booker.login():
        print("Failed to login")
        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):
        print("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"\nFound {len(bookable_sessions)} sessions to book")


    for session in bookable_sessions:
        result = booker.matches_preferred_session(session, current_time)
        print(session.get("name_activity") + " / " + str(result))