kbe/crossfit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'A lot of features' (#1) from develop into main

Browse Source 
Reviewed-on: #1
This commit was merged in pull request #1.
This commit is contained in:
Kevin BATAILLE
2025-07-20 14:32:06 +00:00
parent f67b3ac36f 5dcc2a89ae
commit 04a255dd56
8 changed files with 995 additions and 410 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
.env.example
View File

@@ -1,3 +1,17 @@
# Configuration
CROSSFIT_USERNAME=Kevin8407
CROSSFIT_PASSWORD=9vx03OSE
# CrossFit booking credentials
CROSSFIT_USERNAME=your_username
CROSSFIT_PASSWORD=your_password
# Notification settings
ENABLE_EMAIL_NOTIFICATIONS=true
ENABLE_TELEGRAM_NOTIFICATIONS=false
# Email notification credentials
SMTP_SERVER=mail.infomaniak.com
EMAIL_FROM=your_email
EMAIL_TO=recipient_email
EMAIL_PASSWORD=email_password
# Telegram notification credentials
TELEGRAM_TOKEN=your_telegram_token
TELEGRAM_CHAT_ID=your_chat_id

183
.gitignore vendored
View File

@@ -1,3 +1,186 @@
.env
log/
!log/.gitkeep
# Created by https://www.toptal.com/developers/gitignore/api/python
# Edit at https://www.toptal.com/developers/gitignore?templates=python
### Python ###
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
#pdm.lock
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
# in version control.
# https://pdm.fming.dev/#use-with-ide
.pdm.toml
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/
### Python Patch ###
# Poetry local configuration file - https://python-poetry.org/docs/configuration/#local-configuration
poetry.toml
# ruff
.ruff_cache/
# LSP config files
pyrightconfig.json
# End of https://www.toptal.com/developers/gitignore/api/python
# Docker
docker-compose.override.yml
docker-compose.yml
Dockerfile
.dockerignore

1
.kilocode/mcp.json Normal file
View File

@@ -0,0 +1 @@
{"mcpServers":{}}

92
README.md Normal file
View File

@@ -0,0 +1,92 @@
# Crossfit Application
This is a Python application for managing Crossfit bookings and notifications. The application automates the process of booking Crossfit sessions and sends notifications via email and Telegram when a booking is successful.
## Features
- Automated booking of Crossfit sessions
- Email and Telegram notifications for successful bookings
- Configurable preferred sessions
- Retry logic for booking failures
- Detailed logging
## Prerequisites
- Docker
- Docker Compose
## Setup
1. Create a `.env` file based on `.env.example` and fill in the required credentials.
2. Build and run the application using Docker Compose:
```bash
docker-compose up --build
```
3. The application will run in a Docker container, and the logs will be stored in the `./log` directory.
## Usage
The application will automatically check for available sessions and book them based on your preferences. It will send notifications via email and Telegram when a booking is successful.
### Environment Variables
The following environment variables are required:
- `CROSSFIT_USERNAME`: Your Crossfit username
- `CROSSFIT_PASSWORD`: Your Crossfit password
- `EMAIL_FROM`: Your email address
- `EMAIL_TO`: Recipient email address
- `EMAIL_PASSWORD`: Your email password
- `TELEGRAM_TOKEN`: Your Telegram bot token
- `TELEGRAM_CHAT_ID`: Your Telegram chat ID
### Preferred Sessions
You can configure your preferred sessions in the `crossfit_booker.py` file. The preferred sessions are defined as a list of tuples, where each tuple contains the day of the week, start time, and session name.
```python
PREFERRED_SESSIONS = [
(4, "17:00", "WEIGHTLIFTING"), # Friday 17:00 WEIGHTLIFTING
(5, "12:30", "HYROX"), # Saturday 12:30 HYROX
(2, "18:30", "CONDITIONING"), # Wednesday 18:30 CONDITIONING
]
```
## Files
- `Dockerfile`: Docker image definition
- `docker-compose.yml`: Docker Compose service definition
- `.env.example`: Example environment variables file
- `.dockerignore`: Docker ignore file
- `.gitignore`: Git ignore file
- `book_crossfit.py`: Main application script
- `crossfit_booker.py`: Crossfit booking script
- `session_notifier.py`: Session notification script
- `requirements.txt`: Python dependencies
## Project Structure
```
.
├── Dockerfile
├── docker-compose.yml
├── .env.example
├── .dockerignore
├── .gitignore
├── book_crossfit.py
├── crossfit_booker.py
├── session_notifier.py
├── requirements.txt
└── log
└── crossfit_booking.log
```
## Contributing
Contributions are welcome! Please open an issue or submit a pull request.
## License
This project is licensed under the MIT License. See the [LICENSE](LICENSE) file for details.

413
book_crossfit.py
View File

@@ -1,415 +1,14 @@
#!/usr/bin/env python3
# Native modules
import json
import logging
from datetime import datetime, timedelta
import os
import sys
import time
# Third-party modules
import requests
from dateutil.parser import parse
import pytz
from dotenv import load_dotenv
from urllib.parse import urlencode
from typing import List, Dict, Optional
from datetime import datetime, timedelta
# Parse session time (handles timezones if present)
from dateutil.parser import parse
import pytz
from urllib.parse import urlencode
from typing import List, Dict, Optional
from dotenv import load_dotenv
load_dotenv()
# Configuration
USERNAME = os.environ.get("CROSSFIT_USERNAME")
PASSWORD = os.environ.get("CROSSFIT_PASSWORD")
if not all([USERNAME, PASSWORD]):
raise ValueError("Missing environment variables: CROSSFIT_USERNAME and/or CROSSFIT_PASSWORD")
APPLICATION_ID = "81560887"
CATEGORY_ID = "677" # Activity category ID for CrossFit
TIMEZONE = "Europe/Paris" # Adjust to your timezone
TARGET_RESERVATION_TIME = "20:01" # When bookings open (8 PM)
DEVICE_TYPE = "3"
APP_VERSION = "5.09.21"
# Define your preferred sessions
# Format: List of tuples (day_of_week, start_time, session_name_contains)
# day_of_week: 0=Monday, 6=Sunday
PREFERRED_SESSIONS = [
(4, "17:00", "WEIGHTLIFTING"), # Friday 17:00 WEIGHTLIFTING
(5, "12:30", "HYROX"), # Saturday 12:30 HYROX
(2, "18:30", "CONDITIONING"), # Wednesday 18:30 CONDITIONING
# (5, "15:15", "BIG WOD") # Saturday 15:15 BIG WOD
]
# Configure logging once at script startup
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler("log/crossfit_booking.log"),
logging.StreamHandler()
]
)
class CrossFitBooker:
def __init__(self):
self.auth_token = None
self.user_id = None
self.session = requests.Session()
self.base_headers = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:140.0) Gecko/20100101 Firefox/140.0",
"Content-Type": "application/x-www-form-urlencoded",
"Nubapp-Origin": "user_apps",
}
self.session.headers.update(self.base_headers)
# Define mandatory parameters for API calls
self.mandatory_params = {
"app_version": APP_VERSION,
"device_type": DEVICE_TYPE,
"id_application": APPLICATION_ID,
"id_category_activity": CATEGORY_ID
}
def get_auth_headers(self) -> Dict:
"""Return headers with authorization if available"""
headers = self.base_headers.copy()
if self.auth_token:
headers["Authorization"] = f"Bearer {self.auth_token}"
return headers
def login(self) -> bool:
"""Authenticate and get the bearer token"""
try:
# First login endpoint
login_params = {
"app_version": APP_VERSION,
"device_type": DEVICE_TYPE,
"username": USERNAME,
"password": PASSWORD
}
response = self.session.post(
"https://sport.nubapp.com/api/v4/users/checkUser.php",
headers={"Content-Type": "application/x-www-form-urlencoded"},
data=urlencode(login_params))
if not response.ok:
print(f"First login step failed: {response.status_code} - {response.text}")
return False
login_data = response.json()
self.user_id = str(login_data["data"]["user"]["id_user"])
# Second login endpoint
response = self.session.post(
"https://sport.nubapp.com/api/v4/login",
headers={"Content-Type": "application/x-www-form-urlencoded;charset=UTF-8"},
data=urlencode({
"device_type": DEVICE_TYPE,
"username": USERNAME,
"password": PASSWORD
}))
if response.ok:
login_data = response.json()
self.auth_token = login_data.get("token")
if self.auth_token and self.user_id:
print("Successfully logged in")
return True
print(f"Login failed: {response.status_code} - {response.text}")
return False
except Exception as e:
print(f"Login error: {str(e)}")
return False
def get_available_sessions(self, start_date: datetime, end_date: datetime) -> Optional[Dict]:
"""Fetch available sessions from the API with comprehensive error handling"""
if not self.auth_token or not self.user_id:
print("Authentication required - missing token or user ID")
return None
url = "https://sport.nubapp.com/api/v4/activities/getActivitiesCalendar.php"
# Prepare request with mandatory parameters
request_data = self.mandatory_params.copy()
request_data.update({
"id_user": self.user_id,
"start_timestamp": start_date.strftime("%d-%m-%Y"),
"end_timestamp": end_date.strftime("%d-%m-%Y")
})
# Debugging logs
# print(f"Request Data: {request_data}")
# print(f"Headers: {self.get_auth_headers()}")
try:
# Make the request
response = self.session.post(
url,
headers=self.get_auth_headers(),
data=urlencode(request_data),
timeout=10
)
# Debug raw response
# print(f"Response Status Code: {response.status_code}")
# print(f"Response Content: {response.text}")
# Handle response
if response.status_code == 200:
try:
json_response = response.json()
return json_response
except ValueError:
print("Failed to decode JSON response")
return None
elif response.status_code == 400:
print("400 Bad Request - likely missing or invalid parameters")
print("Verify these parameters:")
for param, value in request_data.items():
print(f"- {param}: {value}")
return None
elif response.status_code == 401:
print("401 Unauthorized - token may be expired or invalid")
return None
else:
print(f"Unexpected status code: {response.status_code}")
return None
except requests.exceptions.RequestException as e:
print(f"Request failed: {str(e)}")
return None
except Exception as e:
print(f"Unexpected error: {str(e)}")
return None
def book_session(self, session_id: str) -> bool:
"""Book a specific session with debug logging."""
logging.info(f"Attempting to book session_id: {session_id}")
if not self.auth_token or not self.user_id:
logging.error("Not authenticated: missing auth_token or user_id")
return False
try:
# Prepare headers
headers = self.get_auth_headers()
# print(f"[DEBUG] Request headers: {headers}")
# Prepare the exact request data from cURL
request_data = self.mandatory_params.copy()
request_data.update({
"id_activity_calendar": session_id, # Note the different parameter name
"id_user": self.user_id,
"action_by": self.user_id, # Same as id_user in this case
"n_guests": "0",
"booked_on": "3" # 3 likely means "booked via app"
})
print(f"[DEBUG] Final request data: {request_data}")
# Use the correct endpoint
response = self.session.post(
"https://sport.nubapp.com/api/v4/activities/bookActivityCalendar.php",
headers=headers,
data=urlencode(request_data)
)
logging.debug(f"Response status: {response.status_code}")
logging.debug(f"API response: {response.text}")
if response.ok:
try:
json_response = 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}")
return False
except ValueError:
logging.error("Invalid JSON response")
return False
else:
logging.error(f"HTTP {response.status_code}: {response.text}")
return False
except Exception as e:
logging.critical(f"Unexpected error: {str(e)}", exc_info=True)
return False
def is_session_bookable(self, session: Dict, current_time: datetime) -> bool:
"""Check if a session is bookable based on user_info, ignoring error codes."""
user_info = session.get("user_info", {})
# First check if can_join is true (primary condition)
if user_info.get("can_join", False):
return True
# If can_join is False, check if there's a booking window
booking_date_str = user_info.get("unableToBookUntilDate", "")
booking_time_str = user_info.get("unableToBookUntilTime", "")
if booking_date_str and booking_time_str:
try:
booking_datetime = datetime.strptime(
f"{booking_date_str} {booking_time_str}",
"%d-%m-%Y %H:%M"
)
booking_datetime = pytz.timezone(TIMEZONE).localize(booking_datetime)
if current_time >= booking_datetime:
return True # Booking window is open
else:
return False # Still waiting for booking to open
except ValueError:
pass # Ignore invalid date formats
# Default case: not bookable
return False
def matches_preferred_session(self, session: Dict, current_time: datetime) -> bool:
"""Check if session matches one of your preferred sessions."""
try:
session_time = parse(session["start_timestamp"])
if not session_time.tzinfo:
session_time = pytz.timezone(TIMEZONE).localize(session_time)
# Get day of week (0=Monday, 6=Sunday) and time
day_of_week = session_time.weekday()
session_time_str = session_time.strftime("%H:%M")
session_name = session.get("name_activity", "").upper()
# Check against preferred sessions
for preferred_day, preferred_time, preferred_name in PREFERRED_SESSIONS:
if (day_of_week == preferred_day and
session_time_str == preferred_time and
preferred_name in session_name): # Partial match
return True
return False
except Exception as e:
logging.error(f"Failed to check session: {str(e)}")
return False
def run_booking_cycle(self, current_time: datetime):
"""Run one cycle of checking and booking sessions"""
# Calculate date range to check (next 3 days)
start_date = current_time.date()
end_date = start_date + timedelta(days=3)
# Get available sessions
sessions_data = self.get_available_sessions(start_date, end_date)
if not sessions_data or not sessions_data.get("success", False):
print("No sessions available or error fetching sessions")
return
activities = sessions_data.get("data", {}).get("activities_calendar", [])
# Find sessions to book (both preferred and any available)
sessions_to_book = []
for session in activities:
if not self.is_session_bookable(session, current_time):
continue
if self.matches_preferred_session(session, current_time):
sessions_to_book.append(("Preferred", session))
elif current_time.strftime("%H:%M") == TARGET_RESERVATION_TIME:
# At booking time, consider all available sessions
sessions_to_book.append(("Available", session))
if not sessions_to_book:
print("No matching sessions found to book")
return
# Book sessions (preferred first)
sessions_to_book.sort(key=lambda x: 0 if x[0] == "Preferred" else 1)
for session_type, session in sessions_to_book:
session_time = datetime.strptime(session["start_datetime"], "%Y-%m-%d %H:%M:%S")
print(f"Attempting to book {session_type} session at {session_time} ({session['name_activity']})")
if self.book_session(session["id_activity_calendar"]):
print(f"Successfully booked {session_type} session at {session_time}")
else:
print(f"Failed to book {session_type} session at {session_time}")
def run(self):
"""Main execution loop"""
# Set up timezone
tz = pytz.timezone(TIMEZONE)
# Initial login
if not self.login():
logging.error("Authentication failed - exiting program")
return
while True:
current_time = datetime.now(tz)
print(f"\nCurrent time: {current_time}")
# Run booking cycle at the target time or if it's a test
if current_time.strftime("%H:%M") == TARGET_RESERVATION_TIME:
self.run_booking_cycle(current_time)
# Wait a minute to avoid checking again immediately
time.sleep(60)
else:
# Check again in 30 seconds
time.sleep(30)
import traceback
from crossfit_booker import CrossFitBooker
if __name__ == "__main__":
booker = CrossFitBooker()
if not booker.login():
print("Failed to login")
logging.error("Failed to login - Traceback: %s", traceback.format_exc())
exit(1)
# Set timezone for current_time
tz = pytz.timezone(TIMEZONE)
current_time = datetime.now(tz)
# Get sessions for the next 7 days
start_date = datetime.now()
end_date = start_date + timedelta(days=3)
session_data = booker.get_available_sessions(start_date, end_date)
if not session_data or not session_data.get("success", False):
logging.error("Failed to get session data")
exit(1)
activities = session_data.get("data", {}).get("activities_calendar", [])
bookable_sessions = []
for session in activities:
# Assuming the string is stored in a variable named session_time_str
session_time_str = "2025-07-19 20:01:08.858174+02:00"
session_time = datetime.strptime(session_time_str, "%Y-%m-%d %H:%M:%S.%f%z")
if booker.is_session_bookable(session, current_time):
# if booker.is_session_bookable(session, session_time):
bookable_sessions.append(session)
# print(f"Bookable sessions: {json.dumps(bookable_sessions, indent=2)}")
print(f"Found {len(bookable_sessions)} sessions to book")
for session in bookable_sessions:
is_prefered_session = booker.matches_preferred_session(session, current_time)
# print(session.get("name_activity") + " / " + str(is_prefered_session))
if is_prefered_session:
booker.book_session(session.get("id_activity_calendar"))
# Start continuous booking loop
booker.run()
logging.info("Script completed")

567
crossfit_booker.py Normal file
View File

@@ -0,0 +1,567 @@
# Native modules
import json
import logging
from datetime import datetime, timedelta
import os
import sys
import time
import difflib
# Third-party modules
import requests
from dateutil.parser import parse
import pytz
from dotenv import load_dotenv
from urllib.parse import urlencode
from typing import List, Dict, Optional, Any
# Import the SessionNotifier class
from session_notifier import SessionNotifier
load_dotenv()
# Configuration
USERNAME = os.environ.get("CROSSFIT_USERNAME")
PASSWORD = os.environ.get("CROSSFIT_PASSWORD")
if not all([USERNAME, PASSWORD]):
raise ValueError("Missing environment variables: CROSSFIT_USERNAME and/or CROSSFIT_PASSWORD")
APPLICATION_ID = "81560887"
CATEGORY_ID = "677" # Activity category ID for CrossFit
TIMEZONE = "Europe/Paris" # Adjust to your timezone
TARGET_RESERVATION_TIME = "20:01" # When bookings open (8 PM)
DEVICE_TYPE = "3"
# Retry configuration
RETRY_MAX = 3
RETRY_BACKOFF = 1
APP_VERSION = "5.09.21"
# Define your preferred sessions
# Format: List of tuples (day_of_week, start_time, session_name_contains)
# day_of_week: 0=Monday, 6=Sunday
PREFERRED_SESSIONS = [
# (0, "17:00", "HYROX"), # Monday 17:00 HYROX
(4, "17:00", "WEIGHTLIFTING"), # Friday 17:00 WEIGHTLIFTING
(5, "12:30", "HYROX"), # Saturday 12:30 HYROX
(2, "18:30", "CONDITIONING"), # Wednesday 18:30 CONDITIONING
# (5, "15:15", "BIG WOD") # Saturday 15:15 BIG WOD
]
# Configure logging once at script startup
logging.basicConfig(
level=logging.DEBUG, # Change to DEBUG for more detailed logs
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler("log/crossfit_booking.log"),
logging.StreamHandler()
]
)
logging.getLogger("requests").setLevel(logging.WARNING)
logging.info("Logging enhanced with request library noise reduction")
class CrossFitBooker:
def __init__(self) -> None:
"""
Initialize the CrossFitBooker with necessary attributes.
"""
self.auth_token: Optional[str] = None
self.user_id: Optional[str] = None
self.session: requests.Session = requests.Session()
self.base_headers: Dict[str, str] = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:140.0) Gecko/20100101 Firefox/140.0",
"Content-Type": "application/x-www-form-urlencoded",
"Nubapp-Origin": "user_apps",
}
self.session.headers.update(self.base_headers)
# Define mandatory parameters for API calls
self.mandatory_params: Dict[str, str] = {
"app_version": APP_VERSION,
"device_type": DEVICE_TYPE,
"id_application": APPLICATION_ID,
"id_category_activity": CATEGORY_ID
}
# Initialize the SessionNotifier with credentials from environment variables
email_credentials = {
"from": os.environ.get("EMAIL_FROM"),
"to": os.environ.get("EMAIL_TO"),
"password": os.environ.get("EMAIL_PASSWORD")
}
telegram_credentials = {
"token": os.environ.get("TELEGRAM_TOKEN"),
"chat_id": os.environ.get("TELEGRAM_CHAT_ID")
}
# Get notification settings from environment variables
enable_email = os.environ.get("ENABLE_EMAIL_NOTIFICATIONS", "true").lower() in ("true", "1", "yes")
enable_telegram = os.environ.get("ENABLE_TELEGRAM_NOTIFICATIONS", "true").lower() in ("true", "1", "yes")
self.notifier = SessionNotifier(
email_credentials,
telegram_credentials,
enable_email=enable_email,
enable_telegram=enable_telegram
)
def get_auth_headers(self) -> Dict[str, str]:
"""
Return headers with authorization if available.
Returns:
Dict[str, str]: Headers dictionary with authorization if available.
"""
headers: Dict[str, str] = self.base_headers.copy()
if self.auth_token:
headers["Authorization"] = f"Bearer {self.auth_token}"
return headers
def login(self) -> bool:
"""
Authenticate and get the bearer token.
Returns:
bool: True if login is successful, False otherwise.
"""
try:
# First login endpoint
login_params: Dict[str, str] = {
"app_version": APP_VERSION,
"device_type": DEVICE_TYPE,
"username": USERNAME,
"password": PASSWORD
}
response: requests.Response = self.session.post(
"https://sport.nubapp.com/api/v4/users/checkUser.php",
headers={"Content-Type": "application/x-www-form-urlencoded"},
data=urlencode(login_params))
if not response.ok:
logging.error(f"First login step failed: {response.status_code} - {response.text} - Response: {response.text[:100]}")
return False
try:
login_data: Dict[str, Any] = response.json()
self.user_id = str(login_data["data"]["user"]["id_user"])
except KeyError as ke:
logging.error(f"Key error during login: {str(ke)} - Response: {response.text}")
return False
except ValueError as ve:
logging.error(f"Value error during login: {str(ve)} - Response: {response.text}")
return False
# Second login endpoint
response: requests.Response = self.session.post(
"https://sport.nubapp.com/api/v4/login",
headers={"Content-Type": "application/x-www-form-urlencoded;charset=UTF-8"},
data=urlencode({
"device_type": DEVICE_TYPE,
"username": USERNAME,
"password": PASSWORD
}))
if response.ok:
try:
login_data: Dict[str, Any] = response.json()
self.auth_token = login_data.get("token")
except KeyError as ke:
logging.error(f"Key error during login: {str(ke)} - Response: {response.text}")
return False
except ValueError as ve:
logging.error(f"Value error during login: {str(ve)} - Response: {response.text}")
return False
if self.auth_token and self.user_id:
logging.info("Successfully logged in")
return True
else:
logging.error(f"Login failed: {response.status_code} - {response.text} - Response: {response.text[:100]}")
return False
except requests.exceptions.JSONDecodeError:
logging.error("Failed to decode JSON response during login")
return False
except requests.exceptions.RequestException as e:
logging.error(f"Request error during login: {str(e)}")
return False
except Exception as e:
logging.error(f"Unexpected error during login: {str(e)}")
return False
def get_available_sessions(self, start_date: datetime, end_date: datetime) -> Optional[Dict[str, Any]]:
"""
Fetch available sessions from the API with comprehensive error handling.
Args:
start_date (datetime): Start date for fetching sessions.
end_date (datetime): End date for fetching sessions.
Returns:
Optional[Dict[str, Any]]: Dictionary containing available sessions if successful, None otherwise.
"""
if not self.auth_token or not self.user_id:
logging.error("Authentication required - missing token or user ID")
return None
url: str = "https://sport.nubapp.com/api/v4/activities/getActivitiesCalendar.php"
# Prepare request with mandatory parameters
request_data: Dict[str, str] = self.mandatory_params.copy()
request_data.update({
"id_user": self.user_id,
"start_timestamp": start_date.strftime("%d-%m-%Y"),
"end_timestamp": end_date.strftime("%d-%m-%Y")
})
# Add retry logic with exponential backoff and more informative error messages
for retry in range(RETRY_MAX):
try:
try:
response: requests.Response = self.session.post(
url,
headers=self.get_auth_headers(),
data=urlencode(request_data),
timeout=10
)
except requests.exceptions.Timeout:
logging.error(f"Request timed out after 10 seconds for URL: {url}. Retry {retry+1}/{RETRY_MAX}")
return None
except requests.exceptions.ConnectionError as e:
logging.error(f"Connection error for URL: {url} - Error: {str(e)}")
return None
except requests.exceptions.RequestException as e:
logging.error(f"Request failed for URL: {url} - Error: {str(e)}")
return None
break # Success, exit retry loop
except requests.exceptions.JSONDecodeError:
logging.error("Failed to decode JSON response")
return None
except requests.exceptions.RequestException as e:
if retry == RETRY_MAX - 1:
logging.error(f"Final retry failed: {str(e)}")
raise # Propagate error
wait_time: int = RETRY_BACKOFF * (2 ** retry)
logging.warning(f"Request failed (attempt {retry+1}/{RETRY_MAX}): {str(e)}. Retrying in {wait_time}s...")
time.sleep(wait_time)
else:
# All retries exhausted
logging.error(f"Failed after {RETRY_MAX} attempts")
return None
# Handle response
if response.status_code == 200:
try:
json_response: Dict[str, Any] = response.json()
return json_response
except ValueError:
logging.error("Failed to decode JSON response")
return None
elif response.status_code == 400:
logging.error("400 Bad Request - likely missing or invalid parameters")
logging.error(f"Request Data: {request_data}")
logging.error(f"Response: {response.text[:100]}")
return None
elif response.status_code == 401:
logging.error("401 Unauthorized - token may be expired or invalid")
logging.error(f"Response: {response.text[:100]}")
return None
elif 500 <= response.status_code < 600:
logging.error(f"Server error {response.status_code} - Response: {response.text[:100]}")
raise requests.exceptions.ConnectionError(f"Server error {response.status_code}")
else:
logging.error(f"Unexpected status code: {response.status_code}")
return None
def book_session(self, session_id: str) -> bool:
"""
Book a specific session with debug logging.
Args:
session_id (str): ID of the session to book.
Returns:
bool: True if booking is successful, False otherwise.
"""
return self._make_request(
url="https://sport.nubapp.com/api/v4/activities/bookActivityCalendar.php",
data=self._prepare_booking_data(session_id),
success_msg=f"Successfully booked session {session_id}"
)
def _prepare_booking_data(self, session_id: str) -> Dict[str, str]:
"""
Prepare request data for booking a session.
Args:
session_id (str): ID of the session to book.
Returns:
Dict[str, str]: Dictionary containing request data for booking a session.
"""
return {
**self.mandatory_params,
"id_activity_calendar": session_id,
"id_user": self.user_id,
"action_by": self.user_id,
"n_guests": "0",
"booked_on": "3"
}
def _make_request(self, url: str, data: Dict[str, str], success_msg: str) -> bool:
"""
Handle API requests with retry logic and response processing.
Args:
url (str): URL for the API request.
data (Dict[str, str]): Data to send with the request.
success_msg (str): Message to log on successful request.
Returns:
bool: True if request is successful, False otherwise.
"""
for retry in range(RETRY_MAX):
try:
response: requests.Response = self.session.post(
url,
headers=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(success_msg)
return True
logging.error(f"API returned success:false: {json_response}")
return False
logging.error(f"HTTP {response.status_code}: {response.text[:100]}")
return False
except requests.exceptions.JSONDecodeError:
logging.error("Failed to decode JSON response")
return False
except requests.exceptions.RequestException as e:
if retry == RETRY_MAX - 1:
logging.error(f"Final retry failed: {str(e)}")
raise # Propagate error
wait_time: int = RETRY_BACKOFF * (2 ** retry)
logging.warning(f"Request failed (attempt {retry+1}/{RETRY_MAX}): {str(e)}. Retrying in {wait_time}s...")
time.sleep(wait_time)
logging.error(f"Failed to complete request after {RETRY_MAX} attempts")
return False
def is_session_bookable(self, session: Dict[str, Any], current_time: datetime) -> bool:
"""
Check if a session is bookable based on user_info, ignoring error codes.
Args:
session (Dict[str, Any]): Session data.
current_time (datetime): Current time for comparison.
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):
activity_name = session.get("name_activity")
logging.debug(f"Session is bookable: {activity_name} can_join is True")
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:
logging.debug(f"Session is bookable: current_time {current_time} >= booking_datetime {booking_datetime}")
return True # Booking window is open
else:
return False # Still waiting for booking to open
except ValueError:
pass # Ignore invalid date formats
# Default case: not bookable
return False
def matches_preferred_session(self, session: Dict[str, Any], current_time: datetime) -> bool:
"""
Check if session matches one of your preferred sessions with fuzzy 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 first
if (day_of_week == preferred_day and
session_time_str == preferred_time and
preferred_name in session_name):
return True
# Fuzzy match fallback (80% similarity)
ratio: float = difflib.SequenceMatcher(
None,
session_name.lower(),
preferred_name.lower()
).ratio()
if (day_of_week == preferred_day and
abs(session_time.hour - int(preferred_time.split(':')[0])) <= 1 and
ratio >= 0.8):
logging.debug(f"Fuzzy match: {session_name} → {preferred_name} ({ratio:.2%})")
return True
return False
except Exception as e:
logging.error(f"Failed to check session: {str(e)} - Session: {session}")
return False
def 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 (next 3 days)
start_date: date = current_time.date()
end_date: date = start_date + timedelta(days=3)
# Get available sessions
sessions_data: Optional[Dict[str, Any]] = self.get_available_sessions(start_date, end_date)
if not sessions_data or not sessions_data.get("success", False):
logging.error("No sessions available or error fetching sessions - Sessions Data: {sessions_data}")
return
activities: List[Dict[str, Any]] = sessions_data.get("data", {}).get("activities_calendar", [])
# Find sessions to book (preferred only)
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"])
if not session_time.tzinfo:
session_time = pytz.timezone(TIMEZONE).localize(session_time)
# Check if session is preferred and bookable
if self.is_session_bookable(session, current_time):
if self.matches_preferred_session(session, current_time):
sessions_to_book.append(("Preferred", 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
if (session_time.date() - current_time.date()).days == 1:
upcoming_sessions.append(session)
if not sessions_to_book and not upcoming_sessions:
logging.info("No matching sessions found to book")
return
# Notify about all found preferred sessions, regardless of bookability
for session in found_preferred_sessions:
session_time: datetime = parse(session["start_timestamp"])
if not session_time.tzinfo:
session_time = pytz.timezone(TIMEZONE).localize(session_time)
session_details = f"{session['name_activity']} at {session_time.strftime('%Y-%m-%d %H:%M')}"
self.notifier.notify_session_booking(session_details)
logging.info(f"Notified about found preferred session: {session_details}")
# Notify about upcoming sessions
for session in upcoming_sessions:
session_time: datetime = parse(session["start_timestamp"])
if not session_time.tzinfo:
session_time = pytz.timezone(TIMEZONE).localize(session_time)
session_details = f"{session['name_activity']} at {session_time.strftime('%Y-%m-%d %H:%M')}"
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)
sessions_to_book.sort(key=lambda x: 0 if x[0] == "Preferred" else 1)
for session_type, session in sessions_to_book:
session_time: datetime = datetime.strptime(session["start_timestamp"], "%Y-%m-%d %H:%M:%S")
logging.info(f"Attempting to book {session_type} session at {session_time} ({session['name_activity']})")
if self.book_session(session["id_activity_calendar"]):
# Send notification after successful booking
session_details = f"{session['name_activity']} at {session_time.strftime('%Y-%m-%d %H:%M')}"
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} - Session: {session}")
def run(self) -> None:
"""
Main execution loop.
"""
# Set up timezone
tz: pytz.timezone = pytz.timezone(TIMEZONE)
# Initial login
if not self.login():
logging.error("Authentication failed - exiting program")
return
try:
while True:
try:
current_time: datetime = datetime.now(tz)
logging.info(f"Current time: {current_time}")
# Always run booking cycle to check for preferred sessions and notify
self.run_booking_cycle(current_time)
# Run booking cycle at the target time for actual booking
if current_time.strftime("%H:%M") == TARGET_RESERVATION_TIME:
# Wait until the next booking window
wait_until = current_time + timedelta(minutes=60)
time.sleep((wait_until - current_time).total_seconds())
else:
# Check again in 5 minutes
time.sleep(300)
except Exception as e:
logging.error(f"Unexpected error in booking cycle: {str(e)} - Traceback: {traceback.format_exc()}")
time.sleep(60) # Wait before retrying after error
except KeyboardInterrupt:
self.quit()
def quit(self) -> None:
"""
Clean up resources and exit the script.
"""
logging.info("Script interrupted by user. Quitting...")
# Add any cleanup code here
exit(0)

7
requirements.txt
View File

@@ -1,13 +1,20 @@
anyio==4.9.0
certifi==2025.7.14
charset-normalizer==3.4.2
DateTime==5.5
h11==0.16.0
httpcore==1.0.9
httpx==0.28.1
idna==3.10
python-dateutil==2.9.0.post0
python-dotenv==1.1.1
python-telegram-bot==22.2
pytz==2025.2
requests==2.32.4
setuptools==80.9.0
six==1.17.0
sniffio==1.3.1
telegram==0.0.1
typing==3.7.4.3
urllib3==2.5.0
zope.interface==7.2

122
session_notifier.py Normal file
View File

@@ -0,0 +1,122 @@
import smtplib
import os
import logging
from email.message import EmailMessage
from telegram import Bot
class SessionNotifier:
"""
A class to handle notifications for session bookings.
Supports sending notifications via email and Telegram.
Attributes:
email_credentials (dict): Dictionary containing email credentials
(e.g., 'from', 'to', 'password')
telegram_credentials (dict): Dictionary containing Telegram credentials
(e.g., 'token', 'chat_id')
enable_email (bool): Whether to enable email notifications
enable_telegram (bool): Whether to enable Telegram notifications
"""
def __init__(self, email_credentials, telegram_credentials, enable_email=True, enable_telegram=True):
"""
Initialize the SessionNotifier with email and Telegram credentials.
Args:
email_credentials (dict): Email credentials for authentication
telegram_credentials (dict): Telegram credentials for authentication
enable_email (bool): Whether to enable email notifications
enable_telegram (bool): Whether to enable Telegram notifications
"""
self.email_credentials = email_credentials
self.telegram_credentials = telegram_credentials
self.enable_email = enable_email
self.enable_telegram = enable_telegram
def send_email_notification(self, message):
"""
Send an email notification with the given message.
Args:
message (str): The message content to be sent in the email
"""
logging.debug("Sending email notification")
logging.debug(f"Email credentials: {self.email_credentials}")
# Create an EmailMessage object
email = EmailMessage()
email.set_content(message)
# Set the email sender and recipient
email['From'] = self.email_credentials['from']
email['To'] = self.email_credentials['to']
# Set the email subject
email['Subject'] = 'Session Booking Notification'
# Send the email using smtplib
try:
smtp_server = os.environ.get("SMTP_SERVER")
if not smtp_server:
logging.error("SMTP server not configured in environment variables")
raise ValueError("SMTP server not configured")
with smtplib.SMTP_SSL(smtp_server, 465) as smtp:
logging.debug(f"Connecting to SMTP server: {smtp_server}")
smtp.login(self.email_credentials['from'], self.email_credentials['password'])
logging.debug("Logged in to SMTP server")
smtp.send_message(email)
logging.debug("Email sent successfully")
except Exception as e:
logging.error(f"Failed to send email: {str(e)}")
raise
def send_telegram_notification(self, message):
"""
Send a Telegram notification with the given message.
Args:
message (str): The message content to be sent in the Telegram chat
"""
# Create a Bot instance with the provided token
bot = Bot(token=self.telegram_credentials['token'], base_url=self.telegram_credentials.get('base_url', 'https://api.telegram.org'))
# Send the message to the specified chat ID
bot.send_message(chat_id=self.telegram_credentials['chat_id'], text=message)
def notify_session_booking(self, session_details):
"""
Notify about a session booking via email and Telegram.
Args:
session_details (str): Details about the booked session
"""
# Create messages for both email and Telegram
email_message = f"Session booked: {session_details}"
telegram_message = f"Session booked: {session_details}"
# Send notifications through enabled channels
if self.enable_email:
self.send_email_notification(email_message)
if self.enable_telegram:
self.send_telegram_notification(telegram_message)
def notify_upcoming_session(self, session_details, days_until):
"""
Notify about an upcoming session via email and Telegram.
Args:
session_details (str): Details about the upcoming session
days_until (int): Number of days until the session
"""
# Create messages for both email and Telegram
email_message = f"Session available soon: {session_details} (in {days_until} days)"
telegram_message = f"Session available soon: {session_details} (in {days_until} days)"
# Send notifications through enabled channels
if self.enable_email:
self.send_email_notification(email_message)
if self.enable_telegram:
self.send_telegram_notification(telegram_message)