feat: New script for booking test function in shell and python

.env.example

@@ -1,6 +1,9 @@
 # CrossFit booking credentials
 CROSSFIT_USERNAME=your_username
 CROSSFIT_PASSWORD=your_password
+TARGET_RESERVATION_TIME="21:01"
+BOOKING_WINDOW_END_DELTA_MINUTES="59"
+
 
 # Notification settings
 ENABLE_EMAIL_NOTIFICATIONS=true

crossfit_booker.py

@@ -1,60 +1,36 @@
 # Native modules
-import logging
+import logging, traceback, os, time
-import traceback
-import os
-import time
-import difflib
 from datetime import datetime, timedelta, date
 
 # Third-party modules
-import requests
+import requests, pytz
 from dateutil.parser import parse
-import pytz
 from dotenv import load_dotenv
 from urllib.parse import urlencode
 from typing import List, Dict, Optional, Any, Tuple
 
-# 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
+CATEGORY_ID = "677"
-TIMEZONE = "Europe/Paris"  # Adjust to your timezone
+TIMEZONE = "Europe/Paris"
-TARGET_RESERVATION_TIME = "20:01"  # When bookings open (8 PM)
+TARGET_RESERVATION_TIME = os.environ.get("TARGET_RESERVATION_TIME", "20:01")
-BOOKING_WINDOW_END_DELTA_MINUTES = 10 # End window book session
+BOOKING_WINDOW_END_DELTA_MINUTES = int(os.environ.get("BOOKING_WINDOW_END_DELTA_MINUTES", "10"))
 DEVICE_TYPE = "3"
-
-# Retry configuration
 RETRY_MAX = 3
 RETRY_BACKOFF = 1
 APP_VERSION = "5.09.21"
 
-
 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()
@@ -64,174 +40,188 @@ class CrossFitBooker:
             "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,
+            "app_version": APP_VERSION, "device_type": DEVICE_TYPE,
-            "device_type": DEVICE_TYPE,
+            "id_application": APPLICATION_ID, "id_category_activity": CATEGORY_ID
-            "id_application": APPLICATION_ID,
-            "id_category_activity": CATEGORY_ID
         }
-
+        email_credentials = {"from": os.environ.get("EMAIL_FROM"), "to": os.environ.get("EMAIL_TO"), "password": os.environ.get("EMAIL_PASSWORD")}
-        # Initialize the SessionNotifier with credentials from environment variables
+        telegram_credentials = {"token": os.environ.get("TELEGRAM_TOKEN"), "chat_id": os.environ.get("TELEGRAM_CHAT_ID")}
-        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)
 
-        self.notifier = SessionNotifier(
+    def _auth_headers(self) -> Dict[str, str]:
-            email_credentials,
+        h = self.base_headers.copy()
-            telegram_credentials,
+        if self.auth_token:
-            enable_email=enable_email,
+            h["Authorization"] = f"Bearer {self.auth_token}"
-            enable_telegram=enable_telegram
+        return h
-        )
     
+    # Public method expected by tests
     def get_auth_headers(self) -> Dict[str, str]:
         """
-        Return headers with authorization if available.
+        Return headers with Authorization when token is present.
-        Returns:
+        This wraps _auth_headers() to satisfy tests expecting a public method.
-            Dict[str, str]: Headers dictionary with authorization if available.
         """
-        headers: Dict[str, str] = self.base_headers.copy()
+        return self._auth_headers()
-        if self.auth_token:
+
-            headers["Authorization"] = f"Bearer {self.auth_token}"
+    def _post(self, url: str, data: Dict[str, Any], headers: Optional[Dict[str, str]] = None, expect_json: bool = True) -> Optional[Any]:
-        return headers
+        for retry in range(RETRY_MAX):
+            try:
+                resp = self.session.post(url, headers=(headers or self._auth_headers()), data=urlencode(data), timeout=10)
+                sc = resp.status_code
+                if sc == 200:
+                    return resp.json() if expect_json else resp
+                if sc == 401:
+                    logging.error("401 Unauthorized")
+                    return None
+                # Guard sc to ensure it's an int for comparison (fix tests using Mock without status_code int semantics)
+                if isinstance(sc, int) and 500 <= sc < 600:
+                    logging.error(f"Server error {sc}")
+                    raise requests.exceptions.ConnectionError(f"Server error {sc}")
+                logging.error(f"HTTP {sc}: {getattr(resp, 'text', '')[:100]}")
+                return None
+            except requests.exceptions.RequestException as e:
+                if retry == RETRY_MAX - 1:
+                    logging.error(f"Final retry failed: {e}")
+                    raise
+                wt = RETRY_BACKOFF * (2 ** retry)
+                logging.warning(f"Request failed ({retry+1}/{RETRY_MAX}): {e}. Retrying in {wt}s...")
+                time.sleep(wt)
+        return None
+
+    def _parse_local(self, ts: str) -> datetime:
+        dt = parse(ts)
+        return dt if dt.tzinfo else pytz.timezone(TIMEZONE).localize(dt)
+
+    def _fmt_session(self, s: Dict[str, Any], dt: Optional[datetime] = None) -> str:
+        dt = dt or self._parse_local(s["start_timestamp"])
+        return f"{s['id_activity_calendar']} {s['name_activity']} at {dt.strftime('%Y-%m-%d %H:%M')}"
 
     def login(self) -> bool:
-        """
-        Authenticate and get the bearer token.
-        Returns:
-            bool: True if login is successful, False otherwise.
-        """
         try:
-            # First login endpoint
+            # Directly use requests to align with tests that mock requests.Session.post
-            login_params: Dict[str, str] = {
+            a = {"app_version": APP_VERSION, "device_type": DEVICE_TYPE, "username": USERNAME, "password": PASSWORD}
-                "app_version": APP_VERSION,
+            resp1 = self.session.post(
-                "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))
+                data=urlencode(a)
-            
+            )
-            if not response.ok:
+            if not getattr(resp1, "ok", False):
-                logging.error(f"First login step failed: {response.status_code} - {response.text[:100]}")
+                logging.error("First login step failed")
                 return False
-
             try:
-                login_data: Dict[str, Any] = response.json()
+                r1 = resp1.json()
-                self.user_id = str(login_data["data"]["user"]["id_user"])
+                self.user_id = str(r1["data"]["user"]["id_user"])
-            except (KeyError, ValueError) as e:
+            except Exception as e:
-                logging.error(f"Error during login: {str(e)} - Response: {response.text}")
+                logging.error(f"Error during login: {e} - Response: {getattr(resp1, 'text', '')}")
                 return False
 
-            # Second login endpoint
+            resp2 = 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({
+                data=urlencode({"device_type": DEVICE_TYPE, "username": USERNAME, "password": PASSWORD})
-                    "device_type": DEVICE_TYPE,
+            )
-                    "username": USERNAME,
+            if getattr(resp2, "ok", False):
-                    "password": PASSWORD
-                }))
-            
-            if response.ok:
                 try:
-                    login_data: Dict[str, Any] = response.json()
+                    r2 = resp2.json()
-                    self.auth_token = login_data.get("token")
+                    self.auth_token = r2.get("token")
-                except (KeyError, ValueError) as e:
+                except Exception as e:
-                    logging.error(f"Error during login: {str(e)} - Response: {response.text}")
+                    logging.error(f"Error during login: {e} - Response: {getattr(resp2, '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[:100]}")
-                return False
 
+            logging.error("Login failed")
+            return False
         except requests.exceptions.RequestException as e:
-            logging.error(f"Request error during login: {str(e)}")
+            logging.error(f"Request error during login: {e}")
             return False
         except Exception as e:
-            logging.error(f"Unexpected error during login: {str(e)}")
+            logging.error(f"Unexpected error during login: {e}")
             return False
 
     def get_available_sessions(self, start_date: date, end_date: date) -> Optional[Dict[str, Any]]:
-        """
-        Fetch available sessions from the API.
-        Args:
-            start_date (date): Start date for fetching sessions.
-            end_date (date): 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
-
+        data = {
-        url: str = "https://sport.nubapp.com/api/v4/activities/getActivitiesCalendar.php"
+            **self.mandatory_params,
-        
-        # 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")
+            "end_timestamp": end_date.strftime("%d-%m-%Y"),
-        })
+        }
-        
+        logging.debug(f"[get_available_sessions] URL=https://sport.nubapp.com/api/v4/activities/getActivitiesCalendar.php "
-        # Add retry logic
+                      f"method=POST content_type=application/x-www-form-urlencoded "
-        for retry in range(RETRY_MAX):
+                      f"keys={list(data.keys())} id_user_present={bool(self.user_id)}")
+        r = self._post("https://sport.nubapp.com/api/v4/activities/getActivitiesCalendar.php", data)
+        # Display available sessions in debug console
+        if r is not None:
+            success = r.get("success", False) if isinstance(r, dict) else None
+            count = 0
             try:
-                response: requests.Response = self.session.post(
+                if isinstance(r, dict):
-                    url,
+                    activities = r.get("data", {}).get("activities_calendar", [])
-                    headers=self.get_auth_headers(),
+                    count = len(activities) if isinstance(activities, list) else 0
-                    data=urlencode(request_data),
+            except Exception:
-                    timeout=10
+                pass
+            logging.debug(f"[get_available_sessions] success={success} activities_count={count}")
+            # Log concise session summary to aid debugging
+            if success and count:
+                for s in r.get("data", {}).get("activities_calendar", [])[:50]:
+                    try:
+                        summary = self._fmt_session(s)
+                    except Exception:
+                        summary = f"{s.get('id_activity_calendar')} {s.get('name_activity')} at {s.get('start_timestamp')}"
+                    logging.debug(f"Session: {summary}")
+            else:
+                logging.debug(f"[get_available_sessions] raw_response_preview={str(r)[:500]}")
+        else:
+            logging.debug("[get_available_sessions] No response (None) from API")
+        return r
+
+    def matches_preferred_session(self, session: Dict[str, Any], current_time: datetime) -> bool:
+        try:
+            st = self._parse_local(session["start_timestamp"])
+            dow, hhmm = st.weekday(), st.strftime("%H:%M")
+            name = session.get("name_activity", "").upper()
+            for pd, pt, pn in PREFERRED_SESSIONS:
+                if dow == pd and hhmm == pt and pn in name: return True
+            return False
+        except Exception as e:
+            logging.error(f"Failed to check session: {e} - Session: {session}")
+            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.
+        """
+        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 response.status_code == 200:
+                if current_time >= booking_datetime:
-                    return response.json()
+                    return True  # Booking window is open
-                elif response.status_code == 401:
+            except ValueError:
-                    logging.error("401 Unauthorized - token may be expired or invalid")
+                pass  # Ignore invalid date formats
-                    return None
+        # Default case: not bookable
-                elif 500 <= response.status_code < 600:
+        logging.debug(f"Session: {session.get('id_activity_calendar')} ({session.get('name_activity')}) is not bookable")
-                    logging.error(f"Server error {response.status_code}")
+        return False
-                    raise requests.exceptions.ConnectionError(f"Server error {response.status_code}")
-                else:
-                    logging.error(f"Unexpected status code: {response.status_code} - {response.text[:100]}")
-                    return None
-                    
-            except requests.exceptions.RequestException as e:
-                if retry == RETRY_MAX - 1:
-                    logging.error(f"Final retry failed: {str(e)}")
-                    raise
-                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)
-
-        return None
 
     def book_session(self, session_id: str) -> bool:
         """
         Book a specific session.
-        Args:
-            session_id (str): ID of the session to book.
-        Returns:
-            bool: True if booking is successful, False otherwise.
         """
         url = "https://sport.nubapp.com/api/v4/activities/bookActivityCalendar.php"
         data = {
@@ -277,242 +267,73 @@ class CrossFitBooker:
         logging.error(f"Failed to complete request after {RETRY_MAX} attempts")
         return False
 
-    def get_booked_sessions(self) -> List[Dict[str, Any]]:
+    # Script main entry point
-        """
+    async def execute_cycle(self, current_time: datetime) -> None:
-        Get a list of booked sessions.
+        start_date, end_date = current_time.date(), current_time.date() + timedelta(days=2)
-        
+        sessions_data = self.get_available_sessions(start_date, end_date)
-        Returns:
-            A list of dictionaries containing information about the booked sessions.
-        """
-        url = "https://sport.nubapp.com/api/v4/activities/getBookedActivities.php"
-        data = {
-            **self.mandatory_params,
-            "id_user": self.user_id,
-            "action_by": self.user_id
-        }
-        
-        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):
-                        return json_response.get("data", [])
-                    logging.error(f"API returned success:false: {json_response}")
-                    return []
-
-                logging.error(f"HTTP {response.status_code}: {response.text[:100]}")
-                return []
-
-            except requests.exceptions.RequestException as e:
-                if retry == RETRY_MAX - 1:
-                    logging.error(f"Final retry failed: {str(e)}")
-                    raise
-                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 []
-
-    def is_session_bookable(self, session: Dict[str, Any], current_time: datetime) -> bool:
-        """
-        Check if a session is bookable based on user_info.
-        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):
-            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
-            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 exact 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
-                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:
-            logging.error(f"Failed to check session: {str(e)} - Session: {session}")
-            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")
             return
-
         activities: List[Dict[str, Any]] = sessions_data.get("data", {}).get("activities_calendar", [])
-
+        found_preferred_sessions: List[Tuple[str, Dict[str, Any]]] = []
-        # Find sessions to book (preferred only) within allowed date range
-        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:
-            session_time: datetime = parse(session["start_timestamp"])
+            start_timestamp = self._parse_local(session["start_timestamp"])
-            if not session_time.tzinfo:
+            days_diff = (start_timestamp.date() - current_time.date()).days
-                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
             if not (0 <= days_diff <= 2):
-                continue  # Skip sessions outside the allowed date range
+                continue
+
+            is_preferred = self.matches_preferred_session(session, current_time)
 
-            # Check if session is preferred and bookable
             if self.is_session_bookable(session, current_time):
-                if self.matches_preferred_session(session, current_time):
+                session_type = "Preferred" if is_preferred else "Regular"
-                    sessions_to_book.append(("Preferred", session))
+                found_preferred_sessions.append((session_type, 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 (day + 1)
-                    if days_diff == 1:
-                        upcoming_sessions.append(session)
 
-        if not sessions_to_book and not upcoming_sessions:
+        if not found_preferred_sessions:
-            logging.info("No matching sessions found to book")
+            logging.info("No preferred sessions bookable found in the booking window")
             return
 
-        # Notify about all found preferred sessions, regardless of bookability
+        for session_type, s in found_preferred_sessions:
-        for session in found_preferred_sessions:
+            details = self._fmt_session(s)
-            session_time: datetime = parse(session["start_timestamp"])
+            await self.notifier.notify_session_booking(details)
-            if not session_time.tzinfo:
+            logging.info(f"Notified about found {session_type.lower()} session: {details}")
-                session_time = pytz.timezone(TIMEZONE).localize(session_time)
-            session_details = f"{session['id_activity_calendar']} {session['name_activity']} at {session_time.strftime('%Y-%m-%d %H:%M')}"
-            await self.notifier.notify_session_booking(session_details)
-            logging.info(f"Notified about found preferred session: {session_details}")
 
-        # Notify about upcoming sessions
+        # Sort by preferred sessions first
-        for session in upcoming_sessions:
+        found_preferred_sessions.sort(key=lambda x: 0 if x[0] == "Preferred" else 1)
-            session_time: datetime = parse(session["start_timestamp"])
-            if not session_time.tzinfo:
-                session_time = pytz.timezone(TIMEZONE).localize(session_time)
-            session_details = f"{session['id_activity_calendar']} {session['name_activity']} at {session_time.strftime('%Y-%m-%d %H:%M')}"
-            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 (preferred first)
+        for session_type, s in found_preferred_sessions:
-        sessions_to_book.sort(key=lambda x: 0 if x[0] == "Preferred" else 1)
+            st_dt = self._parse_local(s["start_timestamp"])
-        for session_type, session in sessions_to_book:
+            logging.info(f"Attempting to book {session_type} session at {st_dt} ({s['name_activity']})")
-            session_time: datetime = datetime.strptime(session["start_timestamp"], "%Y-%m-%d %H:%M:%S")
+            if self.book_session(s["id_activity_calendar"]):
-            logging.info(f"Attempting to book {session_type} session at {session_time} ({session['name_activity']})")
+                details = f"{s['name_activity']} at {st_dt.strftime('%Y-%m-%d %H:%M')}"
-            if self.book_session(session["id_activity_calendar"]):
+                await self.notifier.notify_session_booking(details)
-                # Send notification after successful booking
+                logging.info(f"Successfully booked {session_type} session at {st_dt}")
-                session_details = f"{session['name_activity']} at {session_time.strftime('%Y-%m-%d %H:%M')}"
-                await 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}")
+                logging.error(f"Failed to book {session_type} session at {st_dt}")
-                # Send notification about the failed booking
+                details = f"{s['name_activity']} at {st_dt.strftime('%Y-%m-%d %H:%M')}"
-                session_details = f"{session['name_activity']} at {session_time.strftime('%Y-%m-%d %H:%M')}"
+                await self.notifier.notify_impossible_booking(details)
-                await self.notifier.notify_impossible_booking(session_details)
+                logging.info(f"Notified about impossible booking for {session_type} session at {st_dt}")
-                logging.info(f"Notified about impossible booking for {session_type} session at {session_time}")
 
     async def run(self) -> None:
-        """
+        tz = pytz.timezone(TIMEZONE)
-        Main execution loop.
+        th, tm = map(int, TARGET_RESERVATION_TIME.split(":"))
-        """
+        target_time = datetime.now(tz).replace(hour=th, minute=tm, second=0, microsecond=0)
-        # 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(minutes=BOOKING_WINDOW_END_DELTA_MINUTES)
-
-        # Initial login
         if not self.login():
-            logging.error("Authentication failed - exiting program")
+            logging.error("Authentication failed - exiting program"); return
-            return
-
         try:
             while True:
                 try:
-                    current_time: datetime = datetime.now(tz)
+                    now = datetime.now(tz)
-                    logging.info(f"Current time: {current_time}")
+                    logging.info(f"Current time: {now}")
-
+                    if target_time <= now <= booking_window_end:
-                    # Only book sessions if current time is within the booking window
+                        await self.execute_cycle(now); time.sleep(60)
-                    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()}")
+                    logging.error(f"Unexpected error in booking cycle: {e} - Traceback: {traceback.format_exc()}"); time.sleep(60)
-                    time.sleep(60)  # Wait before retrying after error
         except KeyboardInterrupt:
             self.quit()
 
     def quit(self) -> None:
-        """
+        logging.info("Script interrupted by user. Quitting..."); exit(0)
-        Clean up resources and exit the script.
-        """
-        logging.info("Script interrupted by user. Quitting...")
-        exit(0)

docs/crossfit.txt

@@ -23,21 +23,23 @@ curl 'https://sport.nubapp.com/api/v4/activities/getActivitiesCalendar.php' \
   -X POST \
   -H 'User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:140.0) Gecko/20100101 Firefox/140.0' \
   -H 'Accept: application/json, text/plain, */*' \
-  -H 'Accept-Language: en-GB,en;q=0.8,fr-FR;q=0.5,fr;q=0.3' \
+  -H 'Accept-Language: fr-FR,fr;q=0.8,en-GB;q=0.5,en;q=0.3' \
   -H 'Accept-Encoding: gzip, deflate, br, zstd' \
-  -H 'Referer: https://box.resawod.com/' \
   -H 'Content-Type: application/x-www-form-urlencoded' \
-  -H 'Authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiJ9.eyJpYXQiOjE3NTI3NDI5ODYsImV4cCI6MTc2ODY0NDE4Niwicm9sZXMiOlsiUk9MRV9VU0VSIl0sImlkX3VzZXIiOjMxOTE0MjksImlkX2FwcGxpY2F0aW9uIjo4MTU2MDg4NywicGFzc3dvcmRfaGFzaCI6ImEzZGZiOGQzY2NhNTA5NmJhYzYxMWM4MmEwMzYyYTg2ODc3MjQ4MjIiLCJhdXRoZW50aWNhdGlvbl90eXBlIjoiYXBpIiwidXNlcm5hbWUiOiJLZXZpbjg0MDcifQ.BQ-o2-f0-Pj36nvnWz_ZVag6nWIlus5YPCh5tJSrt2XdpxvU6RYVbx4uobyicqS8jyK6G5OmQ1dJU8H5l3KRdK7enSFC4ClbyX7hXCRfQ0EW2bndNj_eIeR5qxbU8jgELCi6JTiCOouICdx6gkqvT9uYk4jSVdDWjYP9lATnzvNpwEIg2Aac1aqXflZtgFPSgwxEuotknLYFxRgMB6nTnMS34iIcvY4WVqsN_geYAtvhZM40NgEqK3Q4XMJusg7wStfiR7d8sqk-9Gqm3thE7Qsu-EjO9T7840zVyTWlLRa5TtUZiqDHjex-KYIh3p8xATbg_Ssct62o_FYkpN0XNA' \
+  -H 'Authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiJ9.eyJpYXQiOjE3NTQ2NjQ4ODYsImV4cCI6MTc3MDU2NjA4Niwicm9sZXMiOlsiUk9MRV9VU0VSIl0sImlkX3VzZXIiOjMxOTE0MjksImlkX2FwcGxpY2F0aW9uIjo4MTU2MDg4NywicGFzc3dvcmRfaGFzaCI6ImEzZGZiOGQzY2NhNTA5NmJhYzYxMWM4MmEwMzYyYTg2ODc3MjQ4MjIiLCJhdXRoZW50aWNhdGlvbl90eXBlIjoiYXBpIiwidXNlcm5hbWUiOiJLZXZpbjg0MDcifQ.AuFvEW5dWWD1SVWwj6dVNOKpTCNeBzSY6HLtbz2z6KGh30VhMoswICWv6KcKqMNtJmt-UV5kpQkdfWAKOHZy843QGEnpI4scMAe1KUAl--Tn7pvwmTYxpx4Ta5E7f4DEV7m6YqZmdClOy-6YbDmf1qb-bR_v20gORMmutkdmcS8btteW11Lc2Uk6k8I2AWMulygQkV1PURzTIm7cmMNX0_R1Z_7M0HWsq_PLuZ0h0gaxKBS_BUqiMNeh8Hvu_sS56OPeDNRXQmFHuDI7CLMTNWBbmNAhfauIP293I8-cwC7eqjP_pp-v6zLTObHndJgNFEoce82iX11KCINyLryqcQ' \
   -H 'Nubapp-Origin: user_apps' \
   -H 'Origin: https://box.resawod.com' \
+  -H 'Sec-GPC: 1' \
   -H 'Connection: keep-alive' \
+  -H 'Referer: https://box.resawod.com/' \
   -H 'Sec-Fetch-Dest: empty' \
   -H 'Sec-Fetch-Mode: cors' \
   -H 'Sec-Fetch-Site: cross-site' \
+  -H 'Priority: u=0' \
   -H 'TE: trailers' \
-  --data-raw 'app_version=5.09.21&id_application=81560887&id_user=3191429&start_timestamp=21-07-2025&end_timestamp=27-07-2025&id_category_activity=677'
+  --data-raw 'app_version=5.09.25&id_application=81560887&id_user=3191429&start_timestamp=04-08-2025&end_timestamp=10-08-2025&id_category_activity=677'
 
-Book activity
+## Book activity
 
   curl 'https://sport.nubapp.com/api/v4/activities/bookActivityCalendar.php' \
   -X POST \
@@ -47,7 +49,7 @@ Book activity
   -H 'Accept-Encoding: gzip, deflate, br, zstd' \
   -H 'Referer: https://box.resawod.com/' \
   -H 'Content-Type: application/x-www-form-urlencoded' \
-  -H 'Authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiJ9.eyJpYXQiOjE3NTI2NzYwNDksImV4cCI6MTc2ODU3NzI0OSwicm9sZXMiOlsiUk9MRV9VU0VSIl0sImlkX3VzZXIiOjMxOTE0MjksImlkX2FwcGxpY2F0aW9uIjo4MTU2MDg4NywicGFzc3dvcmRfaGFzaCI6ImEzZGZiOGQzY2NhNTA5NmJhYzYxMWM4MmEwMzYyYTg2ODc3MjQ4MjIiLCJhdXRoZW50aWNhdGlvbl90eXBlIjoiYXBpIiwidXNlcm5hbWUiOiJLZXZpbjg0MDcifQ.mQhKOTtEt1QUwDzK7BGzA3yk1R4RozhQW1xqfhmfmk_mesrkz5RM9IOnLFbP-ZMvx4_9ZlWv4qvgx3XjAzDWf8M86BPWhY6nNAoKAdTbD_Pg-fsjmRVVEadNv0pizavue5--K2XvU50AQinkzHawihm7HtTqcWOJgH3J7PM5NQO0Y1azd2nkt9mqTBf1l5MrvDZPWR_KbiztNavacr5SY9vSk1pfnf1A9jbR9ca3wCxZNKhXfxCWxNHCBqh_VnXP3Wwh518xL94xCx0nziKDR5VQYNQCLTO3cb9lDTmCILgZSnvSXIpFVNw1mTkz34MKS2WCkF540okzIFTooNhf_A' \
+  -H 'Authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiJ9.eyJpYXQiOjE3NTQ2NjQ4ODYsImV4cCI6MTc3MDU2NjA4Niwicm9sZXMiOlsiUk9MRV9VU0VSIl0sImlkX3VzZXIiOjMxOTE0MjksImlkX2FwcGxpY2F0aW9uIjo4MTU2MDg4NywicGFzc3dvcmRfaGFzaCI6ImEzZGZiOGQzY2NhNTA5NmJhYzYxMWM4MmEwMzYyYTg2ODc3MjQ4MjIiLCJhdXRoZW50aWNhdGlvbl90eXBlIjoiYXBpIiwidXNlcm5hbWUiOiJLZXZpbjg0MDcifQ.AuFvEW5dWWD1SVWwj6dVNOKpTCNeBzSY6HLtbz2z6KGh30VhMoswICWv6KcKqMNtJmt-UV5kpQkdfWAKOHZy843QGEnpI4scMAe1KUAl--Tn7pvwmTYxpx4Ta5E7f4DEV7m6YqZmdClOy-6YbDmf1qb-bR_v20gORMmutkdmcS8btteW11Lc2Uk6k8I2AWMulygQkV1PURzTIm7cmMNX0_R1Z_7M0HWsq_PLuZ0h0gaxKBS_BUqiMNeh8Hvu_sS56OPeDNRXQmFHuDI7CLMTNWBbmNAhfauIP293I8-cwC7eqjP_pp-v6zLTObHndJgNFEoce82iX11KCINyLryqcQ' \
   -H 'Nubapp-Origin: user_apps' \
   -H 'Origin: https://box.resawod.com' \
   -H 'Connection: keep-alive' \
@@ -56,7 +58,7 @@ Book activity
   -H 'Sec-Fetch-Site: cross-site' \
   -H 'Priority: u=0' \
   -H 'TE: trailers' \
-  --data-raw 'app_version=5.09.21&id_application=81560887^&id_activity_calendar=19291304&id_user=3191429&action_by=3191429&n_guests=0&booked_on=3'
+  --data-raw 'app_version=5.09.21&id_application=81560887^&id_activity_calendar=19291766&id_user=3191429&action_by=3191429&n_guests=0&booked_on=3'
 
 
 

old/crossfit_booker.py

@@ -0,0 +1,357 @@
+# Native modules
+import logging, traceback, os, time
+from datetime import datetime, timedelta, date
+
+# Third-party modules
+import requests, pytz
+from dateutil.parser import parse
+from dotenv import load_dotenv
+from urllib.parse import urlencode
+from typing import List, Dict, Optional, Any, Tuple
+
+from session_notifier import SessionNotifier
+from session_config import PREFERRED_SESSIONS
+
+load_dotenv()
+
+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"
+TIMEZONE = "Europe/Paris"
+TARGET_RESERVATION_TIME = os.environ.get("TARGET_RESERVATION_TIME", "20:01")
+BOOKING_WINDOW_END_DELTA_MINUTES = int(os.environ.get("BOOKING_WINDOW_END_DELTA_MINUTES", "10"))
+DEVICE_TYPE = "3"
+RETRY_MAX = 3
+RETRY_BACKOFF = 1
+APP_VERSION = "5.09.21"
+
+class CrossFitBooker:
+    def __init__(self) -> None:
+        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)
+        self.mandatory_params: Dict[str, str] = {
+            "app_version": APP_VERSION, "device_type": DEVICE_TYPE,
+            "id_application": APPLICATION_ID, "id_category_activity": CATEGORY_ID
+        }
+        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")}
+        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 _auth_headers(self) -> Dict[str, str]:
+        h = self.base_headers.copy()
+        if self.auth_token:
+            h["Authorization"] = f"Bearer {self.auth_token}"
+        return h
+
+    # Public method expected by tests
+    def get_auth_headers(self) -> Dict[str, str]:
+        """
+        Return headers with Authorization when token is present.
+        This wraps _auth_headers() to satisfy tests expecting a public method.
+        """
+        return self._auth_headers()
+
+    def _post(self, url: str, data: Dict[str, Any], headers: Optional[Dict[str, str]] = None, expect_json: bool = True) -> Optional[Any]:
+        for retry in range(RETRY_MAX):
+            try:
+                resp = self.session.post(url, headers=(headers or self._auth_headers()), data=urlencode(data), timeout=10)
+                sc = resp.status_code
+                if sc == 200:
+                    return resp.json() if expect_json else resp
+                if sc == 401:
+                    logging.error("401 Unauthorized")
+                    return None
+                # Guard sc to ensure it's an int for comparison (fix tests using Mock without status_code int semantics)
+                if isinstance(sc, int) and 500 <= sc < 600:
+                    logging.error(f"Server error {sc}")
+                    raise requests.exceptions.ConnectionError(f"Server error {sc}")
+                logging.error(f"HTTP {sc}: {getattr(resp, 'text', '')[:100]}")
+                return None
+            except requests.exceptions.RequestException as e:
+                if retry == RETRY_MAX - 1:
+                    logging.error(f"Final retry failed: {e}")
+                    raise
+                wt = RETRY_BACKOFF * (2 ** retry)
+                logging.warning(f"Request failed ({retry+1}/{RETRY_MAX}): {e}. Retrying in {wt}s...")
+                time.sleep(wt)
+        return None
+
+    def _parse_local(self, ts: str) -> datetime:
+        dt = parse(ts)
+        return dt if dt.tzinfo else pytz.timezone(TIMEZONE).localize(dt)
+
+    def _fmt_session(self, s: Dict[str, Any], dt: Optional[datetime] = None) -> str:
+        dt = dt or self._parse_local(s["start_timestamp"])
+        return f"{s['id_activity_calendar']} {s['name_activity']} at {dt.strftime('%Y-%m-%d %H:%M')}"
+
+    def login(self) -> bool:
+        try:
+            # Directly use requests to align with tests that mock requests.Session.post
+            a = {"app_version": APP_VERSION, "device_type": DEVICE_TYPE, "username": USERNAME, "password": PASSWORD}
+            resp1 = self.session.post(
+                "https://sport.nubapp.com/api/v4/users/checkUser.php",
+                headers={"Content-Type": "application/x-www-form-urlencoded"},
+                data=urlencode(a)
+            )
+            if not getattr(resp1, "ok", False):
+                logging.error("First login step failed")
+                return False
+            try:
+                r1 = resp1.json()
+                self.user_id = str(r1["data"]["user"]["id_user"])
+            except Exception as e:
+                logging.error(f"Error during login: {e} - Response: {getattr(resp1, 'text', '')}")
+                return False
+
+            resp2 = 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 getattr(resp2, "ok", False):
+                try:
+                    r2 = resp2.json()
+                    self.auth_token = r2.get("token")
+                except Exception as e:
+                    logging.error(f"Error during login: {e} - Response: {getattr(resp2, 'text', '')}")
+                    return False
+                if self.auth_token and self.user_id:
+                    logging.info("Successfully logged in")
+                    return True
+
+            logging.error("Login failed")
+            return False
+        except requests.exceptions.RequestException as e:
+            logging.error(f"Request error during login: {e}")
+            return False
+        except Exception as e:
+            logging.error(f"Unexpected error during login: {e}")
+            return False
+
+    def get_available_sessions(self, start_date: date, end_date: date) -> Optional[Dict[str, Any]]:
+        if not self.auth_token or not self.user_id:
+            logging.error("Authentication required - missing token or user ID")
+            return None
+        data = {
+            **self.mandatory_params,
+            "id_user": self.user_id,
+            "start_timestamp": start_date.strftime("%d-%m-%Y"),
+            "end_timestamp": end_date.strftime("%d-%m-%Y"),
+        }
+        logging.debug(f"[get_available_sessions] URL=https://sport.nubapp.com/api/v4/activities/getActivitiesCalendar.php "
+                      f"method=POST content_type=application/x-www-form-urlencoded "
+                      f"keys={list(data.keys())} id_user_present={bool(self.user_id)}")
+        r = self._post("https://sport.nubapp.com/api/v4/activities/getActivitiesCalendar.php", data)
+        # Display available sessions in debug console
+        if r is not None:
+            success = r.get("success", False) if isinstance(r, dict) else None
+            count = 0
+            try:
+                if isinstance(r, dict):
+                    activities = r.get("data", {}).get("activities_calendar", [])
+                    count = len(activities) if isinstance(activities, list) else 0
+            except Exception:
+                pass
+            logging.debug(f"[get_available_sessions] success={success} activities_count={count}")
+            # Log concise session summary to aid debugging
+            if success and count:
+                for s in r.get("data", {}).get("activities_calendar", [])[:50]:
+                    try:
+                        summary = self._fmt_session(s)
+                    except Exception:
+                        summary = f"{s.get('id_activity_calendar')} {s.get('name_activity')} at {s.get('start_timestamp')}"
+                    logging.debug(f"[session] {summary}")
+            else:
+                logging.debug(f"[get_available_sessions] raw_response_preview={str(r)[:500]}")
+        else:
+            logging.debug("[get_available_sessions] No response (None) from API")
+        return r
+
+    async def book_session(self, session_id: str) -> bool:
+        # Debug payload composition (without secrets)
+        safe_user_id = str(self.user_id) if self.user_id else None
+        debug_payload = {
+            **self.mandatory_params,
+            "id_activity_calendar": session_id,
+            "id_user": safe_user_id,
+            "action_by": safe_user_id,
+            "n_guests": "0",
+            "booked_on": "1",
+            "device_type": self.mandatory_params.get("device_type"),
+            "token_present": bool(self.auth_token),
+        }
+        logging.debug(f"[book_session] URL=https://sport.nubapp.com/api/v4/activities/bookActivityCalendar.php "
+                      f"method=POST content_type=application/x-www-form-urlencoded "
+                      f"keys={list(debug_payload.keys())} "
+                      f"id_activity_calendar_present={bool(session_id)} "
+                      f"user_id_present={bool(safe_user_id)} token_present={debug_payload['token_present']}")
+        data = {
+            **self.mandatory_params,
+            "id_activity_calendar": session_id,
+            "id_user": self.user_id,
+            "action_by": self.user_id,
+            "n_guests": "0",
+            "booked_on": "1",
+            "device_type": self.mandatory_params["device_type"],
+            "token": self.auth_token
+        }
+        r = self._post("https://sport.nubapp.com/api/v4/activities/bookActivityCalendar.php", data)
+
+        if r is None:
+            logging.error("Booking call failed")
+            return False
+
+        # Check booking status and notify accordingly
+        if r.get("success", False):
+            logging.info(f"Successfully booked session {session_id}")
+            details = f"{session_id} at {datetime.now().strftime('%Y-%m-%d %H:%M')}"
+            await self.notifier.notify_session_booking(details)
+            return True
+
+        # Handle failure case with error message
+        error_message = r.get("message", "Unknown error")
+        error_code = r.get("error", "unknown")
+        logging.error(f"Booking failed: {error_message} (error code: {error_code})")
+        details = f"{session_id} at {datetime.now().strftime('%Y-%m-%d %H:%M')}"
+        await self.notifier.notify_impossible_booking(details)
+        return False
+
+    def get_booked_sessions(self) -> List[Dict[str, Any]]:
+        data = {**self.mandatory_params, "id_user": self.user_id, "action_by": self.user_id}
+        r = self._post("https://sport.nubapp.com/api/v4/activities/getBookedActivities.php", data)
+        if r and r.get("success", False):
+            return r.get("data", [])
+        logging.error(f"Failed to retrieve booked sessions: {r}" if r is not None else "Call failed")
+        return []
+
+    def is_session_bookable(self, session: Dict[str, Any], current_time: datetime) -> bool:
+        ui = session.get("user_info", {})
+        if ui.get("can_join", False): return True
+        d, t = ui.get("unableToBookUntilDate", ""), ui.get("unableToBookUntilTime", "")
+        if d and t:
+            try:
+                bd = pytz.timezone(TIMEZONE).localize(datetime.strftime(f"{d} {t}", "%d-%m-%Y %H:%M"))
+                if current_time >= bd: return True
+            except ValueError: pass
+        return False
+
+    def matches_preferred_session(self, session: Dict[str, Any], current_time: datetime) -> bool:
+        try:
+            st = self._parse_local(session["start_timestamp"])
+            dow, hhmm = st.weekday(), st.strftime("%H:%M")
+            name = session.get("name_activity", "").upper()
+            for pd, pt, pn in PREFERRED_SESSIONS:
+                if dow == pd and hhmm == pt and pn in name: return True
+            return False
+        except Exception as e:
+            logging.error(f"Failed to check session: {e} - Session: {session}")
+            return False
+
+    async def run_booking_cycle(self, current_time: datetime) -> None:
+        start_date, end_date = current_time.date(), current_time.date() + timedelta(days=2)
+        sessions_data = 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")
+            return
+        activities: List[Dict[str, Any]] = sessions_data.get("data", {}).get("activities_calendar", [])
+        sessions_to_book: List[Tuple[str, Dict[str, Any]]] = []
+        upcoming_sessions: List[Dict[str, Any]] = []
+        found_preferred_sessions: List[Dict[str, Any]] = []
+
+        # Debug: list all preferred sessions detected (bookable or not)
+        preferred_debug: List[str] = []
+
+        for s in activities:
+            st = self._parse_local(s["start_timestamp"])
+            days_diff = (st.date() - current_time.date()).days
+            if not (0 <= days_diff <= 2):
+                continue
+
+            is_preferred = self.matches_preferred_session(s, current_time)
+            if is_preferred:
+                # Collect concise summaries to debug output
+                try:
+                    preferred_debug.append(self._fmt_session(s, st))
+                except Exception:
+                    preferred_debug.append(f"{s.get('id_activity_calendar')} {s.get('name_activity')} at {s.get('start_timestamp')}")
+
+            if self.is_session_bookable(s, current_time):
+                if is_preferred:
+                    sessions_to_book.append(("Preferred", s))
+                    found_preferred_sessions.append(s)
+            else:
+                if is_preferred:
+                    found_preferred_sessions.append(s)
+                    if days_diff == 1:
+                        upcoming_sessions.append(s)
+
+        # Emit debug of preferred sessions
+        if preferred_debug:
+            logging.debug("[preferred_sessions] " + " | ".join(preferred_debug[:50]))
+        else:
+            logging.debug("[preferred_sessions] none found in window")
+
+        if not sessions_to_book and not upcoming_sessions:
+            logging.info("No matching sessions found to book")
+            return
+
+        for s in found_preferred_sessions:
+            details = self._fmt_session(s)
+            await self.notifier.notify_session_booking(details)
+            logging.info(f"Notified about found preferred session: {details}")
+
+        for s in upcoming_sessions:
+            details = self._fmt_session(s)
+            await self.notifier.notify_upcoming_session(details, 1)
+            logging.info(f"Notified about upcoming session: {details}")
+
+        sessions_to_book.sort(key=lambda x: 0 if x[0] == "Preferred" else 1)
+        for stype, s in sessions_to_book:
+            st_dt = datetime.strptime(s["start_timestamp"], "%Y-%m-%d %H:%M:%S")
+            logging.info(f"Attempting to book {stype} session at {st_dt} ({s['name_activity']})")
+            if await self.book_session(s["id_activity_calendar"]):
+                details = f"{s['name_activity']} at {st_dt.strftime('%Y-%m-%d %H:%M')}"
+                await self.notifier.notify_session_booking(details)
+                logging.info(f"Successfully booked {stype} session at {st_dt}")
+            else:
+                logging.error(f"Failed to book {stype} session at {st_dt}")
+                details = f"{s['name_activity']} at {st_dt.strftime('%Y-%m-%d %H:%M')}"
+                await self.notifier.notify_impossible_booking(details)
+                logging.info(f"Notified about impossible booking for {stype} session at {st_dt}")
+
+    async def run(self) -> None:
+        tz = pytz.timezone(TIMEZONE)
+        th, tm = map(int, TARGET_RESERVATION_TIME.split(":"))
+        target_time = datetime.now(tz).replace(hour=th, minute=tm, second=0, microsecond=0)
+        booking_window_end = target_time + timedelta(minutes=BOOKING_WINDOW_END_DELTA_MINUTES)
+        if not self.login():
+            logging.error("Authentication failed - exiting program"); return
+        try:
+            while True:
+                try:
+                    now = datetime.now(tz)
+                    logging.info(f"Current time: {now}")
+                    if target_time <= now <= booking_window_end:
+                        await self.run_booking_cycle(now); time.sleep(60)
+                    else:
+                        time.sleep(300)
+                except Exception as e:
+                    logging.error(f"Unexpected error in booking cycle: {e} - Traceback: {traceback.format_exc()}"); time.sleep(60)
+        except KeyboardInterrupt:
+            self.quit()
+
+    def quit(self) -> None:
+        logging.info("Script interrupted by user. Quitting..."); exit(0)

preferred_sessions.json

@@ -13,5 +13,10 @@
         "day_of_week": 5,
         "start_time": "12:30",
         "session_name_contains": "HYROX"
+    },
+    {
+        "day_of_week": 5,
+        "start_time": "12:30",
+        "session_name_contains": "CONDITIONING"
     }
 ]

test/test_crossfit_booker_final.py

@@ -1,236 +0,0 @@
-#!/usr/bin/env python3
-"""
-Comprehensive unit tests for the CrossFitBooker class in crossfit_booker.py
-"""
-
-import os
-import sys
-from unittest.mock import Mock, patch
-from datetime import date
-import requests
-
-# Add the parent directory to the path to import crossfit_booker
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-
-from crossfit_booker import CrossFitBooker
-
-
-class TestCrossFitBookerInit:
-    """Test cases for CrossFitBooker initialization"""
-    
-    def test_init_success(self):
-        """Test successful initialization with all required env vars"""
-        with patch.dict(os.environ, {
-            'CROSSFIT_USERNAME': 'test_user',
-            'CROSSFIT_PASSWORD': 'test_pass',
-            'EMAIL_FROM': 'from@test.com',
-            'EMAIL_TO': 'to@test.com',
-            'EMAIL_PASSWORD': 'email_pass',
-            'TELEGRAM_TOKEN': 'telegram_token',
-            'TELEGRAM_CHAT_ID': '12345'
-        }):
-            booker = CrossFitBooker()
-            assert booker.auth_token is None
-            assert booker.user_id is None
-            assert booker.session is not None
-            assert booker.notifier is not None
-    
-    def test_init_missing_credentials(self):
-        """Test initialization fails with missing credentials"""
-        with patch.dict(os.environ, {}, clear=True):
-            try:
-                CrossFitBooker()
-            except ValueError as e:
-                assert str(e) == "Missing environment variables"
-    
-    def test_init_partial_credentials(self):
-        """Test initialization fails with partial credentials"""
-        with patch.dict(os.environ, {
-            'CROSSFIT_USERNAME': 'test_user'
-            # Missing PASSWORD
-        }, clear=True):
-            try:
-                CrossFitBooker()
-            except ValueError as e:
-                assert str(e) == "Missing environment variables"
-
-
-class TestCrossFitBookerAuthHeaders:
-    """Test cases for get_auth_headers method"""
-    
-    def test_get_auth_headers_without_token(self):
-        """Test headers without auth token"""
-        with patch.dict(os.environ, {
-            'CROSSFIT_USERNAME': 'test_user',
-            'CROSSFIT_PASSWORD': 'test_pass'
-        }):
-            booker = CrossFitBooker()
-            headers = booker.get_auth_headers()
-            assert "Authorization" not in headers
-            assert headers["User-Agent"] == "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:140.0) Gecko/20100101 Firefox/140.0"
-    
-    def test_get_auth_headers_with_token(self):
-        """Test headers with auth token"""
-        with patch.dict(os.environ, {
-            'CROSSFIT_USERNAME': 'test_user',
-            'CROSSFIT_PASSWORD': 'test_pass'
-        }):
-            booker = CrossFitBooker()
-            booker.auth_token = "test_token_123"
-            headers = booker.get_auth_headers()
-            assert headers["Authorization"] == "Bearer test_token_123"
-
-
-class TestCrossFitBookerLogin:
-    """Test cases for login method"""
-    
-    @patch('requests.Session.post')
-    def test_login_success(self, mock_post):
-        """Test successful login flow"""
-        # Mock first login response
-        mock_response1 = Mock()
-        mock_response1.ok = True
-        mock_response1.json.return_value = {
-            "data": {
-                "user": {
-                    "id_user": "12345"
-                }
-            }
-        }
-        
-        # Mock second login response
-        mock_response2 = Mock()
-        mock_response2.ok = True
-        mock_response2.json.return_value = {
-            "token": "test_bearer_token"
-        }
-        
-        mock_post.side_effect = [mock_response1, mock_response2]
-        
-        with patch.dict(os.environ, {
-            'CROSSFIT_USERNAME': 'test_user',
-            'CROSSFIT_PASSWORD': 'test_pass'
-        }):
-            booker = CrossFitBooker()
-            result = booker.login()
-            
-            assert result is True
-            assert booker.user_id == "12345"
-            assert booker.auth_token == "test_bearer_token"
-    
-    @patch('requests.Session.post')
-    def test_login_first_step_failure(self, mock_post):
-        """Test login failure on first step"""
-        mock_response = Mock()
-        mock_response.ok = False
-        mock_response.status_code = 400
-        mock_response.text = "Bad Request"
-        
-        mock_post.return_value = mock_response
-        
-        with patch.dict(os.environ, {
-            'CROSSFIT_USERNAME': 'test_user',
-            'CROSSFIT_PASSWORD': 'test_pass'
-        }):
-            booker = CrossFitBooker()
-            result = booker.login()
-            
-            assert result is False
-            assert booker.user_id is None
-            assert booker.auth_token is None
-    
-    @patch('requests.Session.post')
-    def test_login_json_parsing_error(self, mock_post):
-        """Test login with JSON parsing error"""
-        mock_response = Mock()
-        mock_response.ok = True
-        mock_response.json.side_effect = ValueError("Invalid JSON")
-        
-        mock_post.return_value = mock_response
-        
-        with patch.dict(os.environ, {
-            'CROSSFIT_USERNAME': 'test_user',
-            'CROSSFIT_PASSWORD': 'test_pass'
-        }):
-            booker = CrossFitBooker()
-            result = booker.login()
-            
-            assert result is False
-    
-    @patch('requests.Session.post')
-    def test_login_request_exception(self, mock_post):
-        """Test login with request exception"""
-        mock_post.side_effect = requests.exceptions.ConnectionError("Network error")
-        
-        with patch.dict(os.environ, {
-            'CROSSFIT_USERNAME': 'test_user',
-            'CROSSFIT_PASSWORD': 'test_pass'
-        }):
-            booker = CrossFitBooker()
-            result = booker.login()
-            
-            assert result is False
-
-
-class TestCrossFitBookerGetAvailableSessions:
-    """Test cases for get_available_sessions method"""
-    
-    def test_get_available_sessions_no_auth(self):
-        """Test get_available_sessions without authentication"""
-        with patch.dict(os.environ, {
-            'CROSSFIT_USERNAME': 'test_user',
-            'CROSSFIT_PASSWORD': 'test_pass'
-        }):
-            booker = CrossFitBooker()
-            result = booker.get_available_sessions(date(2025, 7, 24), date(2025, 7, 25))
-            assert result is None
-    
-    @patch('requests.Session.post')
-    def test_get_available_sessions_success(self, mock_post):
-        """Test successful get_available_sessions"""
-        mock_response = Mock()
-        mock_response.status_code = 200
-        mock_response.json.return_value = {
-            "success": True,
-            "data": {
-                "activities_calendar": [
-                    {"id": "1", "name": "Test Session"}
-                ]
-            }
-        }
-        
-        mock_post.return_value = mock_response
-        
-        with patch.dict(os.environ, {
-            'CROSSFIT_USERNAME': 'test_user',
-            'CROSSFIT_PASSWORD': 'test_pass'
-        }):
-            booker = CrossFitBooker()
-            booker.auth_token = "test_token"
-            booker.user_id = "12345"
-            
-            result = booker.get_available_sessions(date(2025, 7, 24), date(2025, 7, 25))
-            
-            assert result is not None
-            assert result["success"] is True
-    
-    @patch('requests.Session.post')
-    def test_get_available_sessions_failure(self, mock_post):
-        """Test get_available_sessions with API failure"""
-        mock_response = Mock()
-        mock_response.status_code = 400
-        mock_response.text = "Bad Request"
-        
-        mock_post.return_value = mock_response
-        
-        with patch.dict(os.environ, {
-            'CROSSFIT_USERNAME': 'test_user',
-            'CROSSFIT_PASSWORD': 'test_pass'
-        }):
-            booker = CrossFitBooker()
-            booker.auth_token = "test_token"
-            booker.user_id = "12345"
-            
-            result = booker.get_available_sessions(date(2025, 7, 24), date(2025, 7, 25))
-            
-            assert result is None

test/test_crossfit_booker_init.py

@@ -1,17 +0,0 @@
-#!/usr/bin/env python3
-"""
-Unit tests for CrossFitBooker initialization
-"""
-
-import pytest
-import os
-import sys
-
-# Add the parent directory to the path
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-
-
-
-
-if __name__ == "__main__":
-    pytest.main([__file__, "-v"])

test/test_crossfit_booker_sessions.py

@@ -190,18 +190,18 @@ class TestCrossFitBookerIsSessionBookable:
             assert result is False
 
 
-class TestCrossFitBookerRunBookingCycle:
+class TestCrossFitBookerExcuteCycle:
-    """Test cases for run_booking_cycle method"""
+    """Test cases for execute_cycle method"""
 
     @patch('crossfit_booker.CrossFitBooker.get_available_sessions')
     @patch('crossfit_booker.CrossFitBooker.is_session_bookable')
     @patch('crossfit_booker.CrossFitBooker.matches_preferred_session')
     @patch('crossfit_booker.CrossFitBooker.book_session')
-    async def test_run_booking_cycle_no_sessions(self, mock_book_session, mock_matches_preferred, mock_is_bookable, mock_get_sessions):
+    async def test_execute_cycle_no_sessions(self, mock_book_session, mock_matches_preferred, mock_is_bookable, mock_get_sessions):
-        """Test run_booking_cycle with no available sessions"""
+        """Test execute_cycle with no available sessions"""
         mock_get_sessions.return_value = {"success": False}
         booker = CrossFitBooker()
-        await booker.run_booking_cycle(datetime.now(pytz.timezone("Europe/Paris")))
+        await booker.execute_cycle(datetime.now(pytz.timezone("Europe/Paris")))
         mock_get_sessions.assert_called_once()
         mock_book_session.assert_not_called()
 
@@ -209,8 +209,8 @@ class TestCrossFitBookerRunBookingCycle:
     @patch('crossfit_booker.CrossFitBooker.is_session_bookable')
     @patch('crossfit_booker.CrossFitBooker.matches_preferred_session')
     @patch('crossfit_booker.CrossFitBooker.book_session')
-    async def test_run_booking_cycle_with_sessions(self, mock_book_session, mock_matches_preferred, mock_is_bookable, mock_get_sessions):
+    async def test_execute_cycle_with_sessions(self, mock_book_session, mock_matches_preferred, mock_is_bookable, mock_get_sessions):
-        """Test run_booking_cycle with available sessions"""
+        """Test execute_cycle with available sessions"""
         # Use current date for the session to ensure it falls within 0-2 day window
         current_time = datetime.now(pytz.timezone("Europe/Paris"))
         session_date = current_time.date()
@@ -233,7 +233,7 @@ class TestCrossFitBookerRunBookingCycle:
         mock_book_session.return_value = True
 
         booker = CrossFitBooker()
-        await booker.run_booking_cycle(current_time)
+        await booker.execute_cycle(current_time)
 
         mock_get_sessions.assert_called_once()
         mock_is_bookable.assert_called_once()
@@ -245,22 +245,22 @@ class TestCrossFitBookerRun:
     """Test cases for run method"""
 
     @patch('crossfit_booker.CrossFitBooker.login')
-    @patch('crossfit_booker.CrossFitBooker.run_booking_cycle')
+    @patch('crossfit_booker.CrossFitBooker.execute_cycle')
-    async def test_run_auth_failure(self, mock_run_booking_cycle, mock_login):
+    async def test_run_auth_failure(self, mock_execute_cycle, mock_login):
         """Test run with authentication failure"""
         mock_login.return_value = False
         booker = CrossFitBooker()
         with patch.object(booker, 'run', new=booker.run) as mock_run:
             await booker.run()
             mock_login.assert_called_once()
-            mock_run_booking_cycle.assert_not_called()
+            mock_execute_cycle.assert_not_called()
 
     @patch('crossfit_booker.CrossFitBooker.login')
-    @patch('crossfit_booker.CrossFitBooker.run_booking_cycle')
+    @patch('crossfit_booker.CrossFitBooker.execute_cycle')
     @patch('crossfit_booker.CrossFitBooker.quit')
     @patch('time.sleep')
     @patch('datetime.datetime')
-    async def test_run_booking_outside_window(self, mock_datetime, mock_sleep, mock_quit, mock_run_booking_cycle, mock_login):
+    async def test_run_booking_outside_window(self, mock_datetime, mock_sleep, mock_quit, mock_execute_cycle, mock_login):
         """Test run with booking outside window"""
         mock_login.return_value = True
         mock_quit.return_value = None  # Prevent actual exit
@@ -292,8 +292,8 @@ class TestCrossFitBookerRun:
         # Verify login was called
         mock_login.assert_called_once()
         
-        # Verify run_booking_cycle was NOT called since we're outside the booking window
+        # Verify execute_cycle was NOT called since we're outside the booking window
-        mock_run_booking_cycle.assert_not_called()
+        mock_execute_cycle.assert_not_called()
         
         # Verify quit was called (due to KeyboardInterrupt)
         mock_quit.assert_called_once()