From 30eb9863a0e6f91a62f2ab716e5193881146da95 Mon Sep 17 00:00:00 2001 From: kbe Date: Fri, 8 Aug 2025 21:54:12 +0200 Subject: [PATCH] refactor: does not notify if no session found --- crossfit_booker.py | 208 ++++++++++++----------- docs/crossfit.txt | 16 +- old/crossfit_booker.py | 357 ++++++++++++++++++++++++++++++++++++++++ preferred_sessions.json | 5 + 4 files changed, 473 insertions(+), 113 deletions(-) create mode 100644 old/crossfit_booker.py diff --git a/crossfit_booker.py b/crossfit_booker.py index edd0612..f275fa4 100644 --- a/crossfit_booker.py +++ b/crossfit_booker.py @@ -55,7 +55,7 @@ class CrossFitBooker: 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]: """ @@ -173,67 +173,13 @@ class CrossFitBooker: 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}") + 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 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 and r.get("success", False): - logging.info(f"Successfully booked session {session_id}") - return True - logging.error(f"Booking failed: {r}" if r is not None else "Booking call failed") - 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.strptime(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"]) @@ -246,77 +192,127 @@ class CrossFitBooker: logging.error(f"Failed to check session: {e} - Session: {session}") return False - async def run_booking_cycle(self, current_time: datetime) -> None: + 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 current_time >= booking_datetime: + return True # Booking window is open + except ValueError: + pass # Ignore invalid date formats + # Default case: not bookable + logging.debug(f"Session: {session.get('id_activity_calendar')} ({session.get('name_activity')}) is not bookable") + return False + + def book_session(self, session_id: str) -> bool: + """ + Book a specific session. + """ + url = "https://sport.nubapp.com/api/v4/activities/bookActivityCalendar.php" + 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 + } + + 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): + logging.info(f"Successfully booked session {session_id}") + return True + else: + logging.error(f"API returned success:false: {json_response} - Session ID: {session_id}") + return False + + logging.error(f"HTTP {response.status_code}: {response.text[:100]}") + return False + + 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 False + + # Script main entry point + async def execute_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]] = [] + found_preferred_sessions: List[Tuple[str, 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 + for session in activities: + start_timestamp = self._parse_local(session["start_timestamp"]) + days_diff = (start_timestamp.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')}") + is_preferred = self.matches_preferred_session(session, current_time) - 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) + if self.is_session_bookable(session, current_time): + session_type = "Preferred" if is_preferred else "Regular" + found_preferred_sessions.append((session_type, session)) - # 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") + if not found_preferred_sessions: + logging.info("No preferred sessions bookable found in the booking window") return - for s in found_preferred_sessions: + for session_type, 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}") + logging.info(f"Notified about found {session_type.lower()} 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}") + # Sort by preferred sessions first + found_preferred_sessions.sort(key=lambda x: 0 if x[0] == "Preferred" else 1) - 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 self.book_session(s["id_activity_calendar"]): + for session_type, s in found_preferred_sessions: + st_dt = self._parse_local(s["start_timestamp"]) + logging.info(f"Attempting to book {session_type} 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}") + logging.info(f"Successfully booked {session_type} session at {st_dt}") else: - logging.error(f"Failed to book {stype} session at {st_dt}") + logging.error(f"Failed to book {session_type} 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}") + logging.info(f"Notified about impossible booking for {session_type} session at {st_dt}") async def run(self) -> None: tz = pytz.timezone(TIMEZONE) @@ -331,7 +327,7 @@ class CrossFitBooker: 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) + await self.execute_cycle(now); time.sleep(60) else: time.sleep(300) except Exception as e: @@ -340,4 +336,4 @@ class CrossFitBooker: self.quit() def quit(self) -> None: - logging.info("Script interrupted by user. Quitting..."); exit(0) + logging.info("Script interrupted by user. Quitting..."); exit(0) \ No newline at end of file diff --git a/docs/crossfit.txt b/docs/crossfit.txt index 83703b8..ad809ae 100644 --- a/docs/crossfit.txt +++ b/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' diff --git a/old/crossfit_booker.py b/old/crossfit_booker.py new file mode 100644 index 0000000..6ec3348 --- /dev/null +++ b/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) diff --git a/preferred_sessions.json b/preferred_sessions.json index 0fe230b..cee8690 100644 --- a/preferred_sessions.json +++ b/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" } ]