add a functional file

crossfit_booker_functional.py

@@ -0,0 +1,721 @@
+# Native modules
+import logging
+import traceback
+import os
+import time
+import difflib
+from datetime import datetime, timedelta, date
+from typing import List, Dict, Optional, Any, Tuple
+from functools import reduce
+
+# Third-party modules
+import requests
+from dateutil.parser import parse
+import pytz
+from dotenv import load_dotenv
+from urllib.parse import urlencode
+
+# Import the SessionNotifier class
+from session_notifier import SessionNotifier
+
+# Import the preferred sessions from the session_config module
+from session_config import PREFERRED_SESSIONS
+
+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)
+DEVICE_TYPE = "3"  # Crossfit Louvre 3
+
+# Retry configuration
+RETRY_MAX = 3
+RETRY_BACKOFF = 1
+APP_VERSION = "5.09.21"
+
+
+# Pure functions for data processing
+def get_auth_headers(base_headers: Dict[str, str], auth_token: Optional[str]) -> Dict[str, str]:
+    """
+    Return headers with authorization if available.
+    
+    Args:
+        base_headers (Dict[str, str]): Base headers dictionary
+        auth_token (Optional[str]): Authorization token if available
+        
+    Returns:
+        Dict[str, str]: Headers dictionary with authorization if available.
+    """
+    headers: Dict[str, str] = base_headers.copy()
+    if auth_token:
+        headers["Authorization"] = f"Bearer {auth_token}"
+    return headers
+
+
+def is_session_bookable(session: Dict[str, Any], current_time: datetime, timezone: str) -> 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.
+        timezone (str): Timezone string for localization.
+        
+    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):
+        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:
+                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(session: Dict[str, Any], preferred_sessions: List[Tuple[int, str, str]], 
+                             timezone: str) -> bool:
+    """
+    Check if session matches one of your preferred sessions with fuzzy matching.
+    
+    Args:
+        session (Dict[str, Any]): Session data.
+        preferred_sessions (List[Tuple[int, str, str]]): List of preferred sessions.
+        timezone (str): Timezone string for localization.
+        
+    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 prepare_booking_data(mandatory_params: Dict[str, str], session_id: str, user_id: str) -> Dict[str, str]:
+    """
+    Prepare request data for booking a session.
+    
+    Args:
+        mandatory_params (Dict[str, str]): Mandatory parameters for API calls
+        session_id (str): ID of the session to book
+        user_id (str): User ID for the booking
+        
+    Returns:
+        Dict[str, str]: Dictionary containing request data for booking a session.
+    """
+    return {
+        **mandatory_params,
+        "id_activity_calendar": session_id,
+        "id_user": user_id,
+        "action_by": user_id,
+        "n_guests": "0",
+        "booked_on": "3"  # Target CrossFit Louvre 3 ?
+    }
+
+
+def is_bookable_and_preferred(session: Dict[str, Any], current_time: datetime, 
+                             preferred_sessions: List[Tuple[int, str, str]], timezone: str) -> bool:
+    """
+    Check if a session is both bookable and matches preferred sessions.
+    
+    Args:
+        session (Dict[str, Any]): Session data
+        current_time (datetime): Current time for comparison
+        preferred_sessions (List[Tuple[int, str, str]]): List of preferred sessions
+        timezone (str): Timezone string for localization
+        
+    Returns:
+        bool: True if session is bookable and preferred, False otherwise
+    """
+    return (is_session_bookable(session, current_time, timezone) and 
+            matches_preferred_session(session, preferred_sessions, timezone))
+
+
+def filter_bookable_sessions(sessions: List[Dict[str, Any]], current_time: datetime, 
+                           preferred_sessions: List[Tuple[int, str, str]], timezone: str) -> List[Dict[str, Any]]:
+    """
+    Filter sessions to find those that are both bookable and match preferred sessions.
+    
+    Args:
+        sessions (List[Dict[str, Any]]): List of session data
+        current_time (datetime): Current time for comparison
+        preferred_sessions (List[Tuple[int, str, str]]): List of preferred sessions
+        timezone (str): Timezone string for localization
+        
+    Returns:
+        List[Dict[str, Any]]: List of sessions that are bookable and match preferences
+    """
+    return list(filter(
+        lambda session: is_bookable_and_preferred(session, current_time, preferred_sessions, timezone),
+        sessions
+    ))
+
+
+def is_upcoming_preferred(session: Dict[str, Any], current_time: datetime,
+                         preferred_sessions: List[Tuple[int, str, str]], timezone: str) -> bool:
+    """
+    Check if a session is an upcoming preferred session.
+    
+    Args:
+        session (Dict[str, Any]): Session data
+        current_time (datetime): Current time for comparison
+        preferred_sessions (List[Tuple[int, str, str]]): List of preferred sessions
+        timezone (str): Timezone string for localization
+        
+    Returns:
+        bool: True if session is an upcoming 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)
+        
+        # Check if session is within allowed date range (current day, day + 1, or day + 2)
+        days_diff = (session_time.date() - current_time.date()).days
+        is_in_range = 0 <= days_diff <= 2
+        
+        # Check if it's a preferred session that's not bookable yet
+        is_preferred = matches_preferred_session(session, preferred_sessions, timezone)
+        is_tomorrow = days_diff == 1
+        
+        return is_in_range and is_preferred and is_tomorrow
+    except Exception:
+        return False
+
+
+def filter_upcoming_sessions(sessions: List[Dict[str, Any]], current_time: datetime,
+                           preferred_sessions: List[Tuple[int, str, str]], timezone: str) -> List[Dict[str, Any]]:
+    """
+    Filter sessions to find upcoming preferred sessions.
+    
+    Args:
+        sessions (List[Dict[str, Any]]): List of session data
+        current_time (datetime): Current time for comparison
+        preferred_sessions (List[Tuple[int, str, str]]): List of preferred sessions
+        timezone (str): Timezone string for localization
+        
+    Returns:
+        List[Dict[str, Any]]: List of upcoming preferred sessions
+    """
+    return list(filter(
+        lambda session: is_upcoming_preferred(session, current_time, preferred_sessions, timezone),
+        sessions
+    ))
+
+
+def filter_preferred_sessions(sessions: List[Dict[str, Any]], 
+                             preferred_sessions: List[Tuple[int, str, str]], 
+                             timezone: str) -> List[Dict[str, Any]]:
+    """
+    Filter sessions to find all preferred sessions.
+    
+    Args:
+        sessions (List[Dict[str, Any]]): List of session data
+        preferred_sessions (List[Tuple[int, str, str]]): List of preferred sessions
+        timezone (str): Timezone string for localization
+        
+    Returns:
+        List[Dict[str, Any]]: List of preferred sessions
+    """
+    return list(filter(
+        lambda session: matches_preferred_session(session, preferred_sessions, timezone),
+        sessions
+    ))
+
+
+def format_session_details(session: Dict[str, Any], timezone: str) -> str:
+    """
+    Format session details for notifications.
+    
+    Args:
+        session (Dict[str, Any]): Session data
+        timezone (str): Timezone string for localization
+        
+    Returns:
+        str: Formatted session details
+    """
+    try:
+        session_time: datetime = parse(session["start_timestamp"])
+        if not session_time.tzinfo:
+            session_time = pytz.timezone(timezone).localize(session_time)
+        return f"{session['name_activity']} at {session_time.strftime('%Y-%m-%d %H:%M')}"
+    except Exception:
+        return f"{session.get('name_activity', 'Unknown session')} at Unknown time"
+
+
+def categorize_sessions(activities: List[Dict[str, Any]], current_time: datetime,
+                       preferred_sessions: List[Tuple[int, str, str]], timezone: str) -> Dict[str, List[Dict[str, Any]]]:
+    """
+    Categorize sessions into bookable, upcoming, and all preferred sessions.
+    
+    Args:
+        activities (List[Dict[str, Any]]): List of session data
+        current_time (datetime): Current time for comparison
+        preferred_sessions (List[Tuple[int, str, str]]): List of preferred sessions
+        timezone (str): Timezone string for localization
+        
+    Returns:
+        Dict[str, List[Dict[str, Any]]]: Dictionary with categorized sessions
+    """
+    return {
+        "bookable": filter_bookable_sessions(activities, current_time, preferred_sessions, timezone),
+        "upcoming": filter_upcoming_sessions(activities, current_time, preferred_sessions, timezone),
+        "all_preferred": filter_preferred_sessions(activities, preferred_sessions, timezone)
+    }
+
+
+def process_booking_results(session: Dict[str, Any], booking_success: bool, timezone: str) -> Dict[str, Any]:
+    """
+    Process the results of a booking attempt.
+    
+    Args:
+        session (Dict[str, Any]): Session data
+        booking_success (bool): Whether the booking was successful
+        timezone (str): Timezone string for localization
+        
+    Returns:
+        Dict[str, Any]: Dictionary with session and booking result information
+    """
+    return {
+        "session": session,
+        "success": booking_success,
+        "details": format_session_details(session, timezone)
+    }
+
+
+class CrossFitBooker:
+    """
+    A class for automating the booking of CrossFit sessions.
+    
+    This class handles authentication, session availability checking,
+    booking, and notifications for CrossFit sessions.
+    """
+
+    def __init__(self) -> None:
+        """
+        Initialize the CrossFitBooker with necessary attributes.
+        
+        Sets up authentication tokens, session headers, mandatory parameters,
+        and initializes the SessionNotifier for sending notifications.
+        """
+        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")
+        }
+
+        # 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 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=get_auth_headers(self.base_headers, self.auth_token),
+                        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=prepare_booking_data(self.mandatory_params, session_id, self.user_id),
+            success_msg=f"Successfully booked session {session_id}"
+        )
+
+    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=get_auth_headers(self.base_headers, self.auth_token),
+                    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
+
+    async 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 (current day, day + 1, and day + 2)
+        start_date: date = current_time.date()
+        end_date: date = start_date + timedelta(days=2)  # Only go up to day + 2
+
+        # 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", [])
+
+        # Categorize sessions
+        categorized_sessions = categorize_sessions(activities, current_time, PREFERRED_SESSIONS, TIMEZONE)
+
+        if not categorized_sessions["bookable"] and not categorized_sessions["upcoming"]:
+            logging.info("No matching sessions found to book")
+            return
+
+        # Notify about all found preferred sessions, regardless of bookability
+        for session in categorized_sessions["all_preferred"]:
+            session_details = format_session_details(session, TIMEZONE)
+            await self.notifier.notify_session_booking(session_details)
+            logging.info(f"Notified about found preferred session: {session_details}")
+
+        # Notify about upcoming sessions
+        for session in categorized_sessions["upcoming"]:
+            session_details = format_session_details(session, TIMEZONE)
+            await 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
+        for session in categorized_sessions["bookable"]:
+            session_time: datetime = datetime.strptime(session["start_timestamp"], "%Y-%m-%d %H:%M:%S")
+            logging.info(f"Attempting to book Preferred session at {session_time} ({session['name_activity']})")
+            booking_success = self.book_session(session["id_activity_calendar"])
+            
+            # Process booking result
+            result = process_booking_results(session, booking_success, TIMEZONE)
+            
+            if result["success"]:
+                # Send notification after successful booking
+                await self.notifier.notify_session_booking(result["details"])
+                logging.info(f"Successfully booked Preferred session at {session_time}")
+            else:
+                logging.error(f"Failed to book Preferred session at {session_time} - Session: {session}")
+                # Send notification about the failed booking
+                await self.notifier.notify_impossible_booking(result["details"])
+                logging.info(f"Notified about impossible booking for Preferred session at {session_time}")
+
+    async def run(self) -> None:
+        """
+        Main execution loop.
+        """
+        # Set up timezone
+        tz: pytz.timezone = pytz.timezone(TIMEZONE)
+
+        # Parse TARGET_RESERVATION_TIME to get the target hour and minute
+        target_hour, target_minute = map(int, TARGET_RESERVATION_TIME.split(":"))
+        target_time = datetime.now(tz).replace(hour=target_hour, minute=target_minute, second=0, microsecond=0)
+        booking_window_end = target_time + timedelta(hours=1)
+
+        # 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}")
+
+                    # Only book sessions if current time is within the booking window
+                    if target_time <= current_time <= booking_window_end:
+                        # Run booking cycle to check for preferred sessions and book
+                        await self.run_booking_cycle(current_time)
+                        # Wait for a short time before next check
+                        time.sleep(60)
+                    else:
+                        # Check again in 5 minutes if outside booking window
+                        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)

test_crossfit_booker.py

@@ -0,0 +1,205 @@
+#!/usr/bin/env python3
+"""
+Test script for the refactored CrossFitBooker functional implementation.
+"""
+
+import sys
+import os
+from datetime import datetime
+import pytz
+from typing import List, Dict, Any, Tuple
+
+# Add the current directory to the path so we can import our modules
+sys.path.append(os.path.dirname(os.path.abspath(__file__)))
+
+# Import the functional functions from our refactored code
+from crossfit_booker_functional import (
+    is_session_bookable,
+    matches_preferred_session,
+    filter_bookable_sessions,
+    filter_upcoming_sessions,
+    filter_preferred_sessions,
+    categorize_sessions,
+    format_session_details
+)
+
+def test_is_session_bookable():
+    """Test the is_session_bookable function."""
+    print("Testing is_session_bookable...")
+    
+    # Test case 1: Session with can_join = True
+    session1 = {
+        "user_info": {
+            "can_join": True
+        }
+    }
+    current_time = datetime.now(pytz.timezone("Europe/Paris"))
+    assert is_session_bookable(session1, current_time, "Europe/Paris") == True
+    
+    # Test case 2: Session with booking window in the past
+    session2 = {
+        "user_info": {
+            "unableToBookUntilDate": "01-01-2020",
+            "unableToBookUntilTime": "10:00"
+        }
+    }
+    assert is_session_bookable(session2, current_time, "Europe/Paris") == True
+    
+    # Test case 3: Session with booking window in the future
+    session3 = {
+        "user_info": {
+            "unableToBookUntilDate": "01-01-2030",
+            "unableToBookUntilTime": "10:00"
+        }
+    }
+    assert is_session_bookable(session3, current_time, "Europe/Paris") == False
+    
+    print("✓ is_session_bookable tests passed")
+
+
+def test_matches_preferred_session():
+    """Test the matches_preferred_session function."""
+    print("Testing matches_preferred_session...")
+    
+    # Define some preferred sessions (day_of_week, start_time, session_name_contains)
+    preferred_sessions: List[Tuple[int, str, str]] = [
+        (2, "18:30", "CONDITIONING"),   # Wednesday 18:30 CONDITIONING
+        (4, "17:00", "WEIGHTLIFTING"),  # Friday 17:00 WEIGHTLIFTING
+        (5, "12:30", "HYROX"),          # Saturday 12:30 HYROX
+    ]
+    
+    # Test case 1: Exact match
+    session1 = {
+        "start_timestamp": "2025-07-30 18:30:00",  # Wednesday
+        "name_activity": "CONDITIONING"
+    }
+    assert matches_preferred_session(session1, preferred_sessions, "Europe/Paris") == True
+    
+    # Test case 2: No match
+    session2 = {
+        "start_timestamp": "2025-07-30 18:30:00",  # Wednesday
+        "name_activity": "YOGA"
+    }
+    assert matches_preferred_session(session2, preferred_sessions, "Europe/Paris") == False
+    
+    print("✓ matches_preferred_session tests passed")
+
+
+def test_filter_functions():
+    """Test the filter functions."""
+    print("Testing filter functions...")
+    
+    # Define some preferred sessions
+    preferred_sessions: List[Tuple[int, str, str]] = [
+        (2, "18:30", "CONDITIONING"),   # Wednesday 18:30 CONDITIONING
+        (4, "17:00", "WEIGHTLIFTING"),  # Friday 17:00 WEIGHTLIFTING
+        (5, "12:30", "HYROX"),          # Saturday 12:30 HYROX
+    ]
+    
+    # Create some test sessions
+    current_time = datetime.now(pytz.timezone("Europe/Paris"))
+    
+    sessions = [
+        {
+            "start_timestamp": "2025-07-30 18:30:00",  # Wednesday
+            "name_activity": "CONDITIONING",
+            "user_info": {"can_join": True}
+        },
+        {
+            "start_timestamp": "2025-07-30 19:00:00",  # Wednesday
+            "name_activity": "YOGA",
+            "user_info": {"can_join": True}
+        },
+        {
+            "start_timestamp": "2025-08-01 17:00:00",  # Friday
+            "name_activity": "WEIGHTLIFTING",
+            "user_info": {"can_join": True}
+        }
+    ]
+    
+    # Test filter_preferred_sessions
+    preferred = filter_preferred_sessions(sessions, preferred_sessions, "Europe/Paris")
+    assert len(preferred) == 2  # CONDITIONING and WEIGHTLIFTING sessions
+    
+    # Test filter_bookable_sessions
+    bookable = filter_bookable_sessions(sessions, current_time, preferred_sessions, "Europe/Paris")
+    assert len(bookable) == 2  # Both preferred sessions are bookable
+    
+    print("✓ Filter function tests passed")
+
+
+def test_categorize_sessions():
+    """Test the categorize_sessions function."""
+    print("Testing categorize_sessions...")
+    
+    # Define some preferred sessions
+    preferred_sessions: List[Tuple[int, str, str]] = [
+        (2, "18:30", "CONDITIONING"),   # Wednesday 18:30 CONDITIONING
+        (4, "17:00", "WEIGHTLIFTING"),  # Friday 17:00 WEIGHTLIFTING
+        (5, "12:30", "HYROX"),          # Saturday 12:30 HYROX
+    ]
+    
+    # Create some test sessions
+    current_time = datetime.now(pytz.timezone("Europe/Paris"))
+    
+    sessions = [
+        {
+            "start_timestamp": "2025-07-30 18:30:00",  # Wednesday
+            "name_activity": "CONDITIONING",
+            "user_info": {"can_join": True}
+        },
+        {
+            "start_timestamp": "2025-07-31 18:30:00",  # Thursday (tomorrow relative to Wednesday)
+            "name_activity": "CONDITIONING",
+            "user_info": {"can_join": False, "unableToBookUntilDate": "01-08-2025", "unableToBookUntilTime": "10:00"}
+        }
+    ]
+    
+    # Test categorize_sessions
+    categorized = categorize_sessions(sessions, current_time, preferred_sessions, "Europe/Paris")
+    assert "bookable" in categorized
+    assert "upcoming" in categorized
+    assert "all_preferred" in categorized
+    
+    print("✓ categorize_sessions tests passed")
+
+
+def test_format_session_details():
+    """Test the format_session_details function."""
+    print("Testing format_session_details...")
+    
+    # Test case 1: Valid session
+    session1 = {
+        "start_timestamp": "2025-07-30 18:30:00",
+        "name_activity": "CONDITIONING"
+    }
+    formatted = format_session_details(session1, "Europe/Paris")
+    assert "CONDITIONING" in formatted
+    assert "2025-07-30 18:30" in formatted
+    
+    # Test case 2: Session with missing data
+    session2 = {
+        "name_activity": "WEIGHTLIFTING"
+    }
+    formatted = format_session_details(session2, "Europe/Paris")
+    assert "WEIGHTLIFTING" in formatted
+    assert "Unknown time" in formatted
+    
+    print("✓ format_session_details tests passed")
+
+
+def run_all_tests():
+    """Run all tests."""
+    print("Running all tests for CrossFitBooker functional implementation...\n")
+    
+    test_is_session_bookable()
+    test_matches_preferred_session()
+    test_filter_functions()
+    test_categorize_sessions()
+    test_format_session_details()
+    
+    print("\n✓ All tests passed!")
+
+
+if __name__ == "__main__":
+    run_all_tests()