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

refactor: Remove CrossfitBooker class and simplify booking system

Browse Source
This commit is contained in:
kbe
2025-10-01 17:48:41 +02:00
parent 5f89af44aa
commit b6ea2a4ff1
12 changed files with 207 additions and 668 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/__init__.py
View File

@@ -3,7 +3,6 @@
# Import all public modules to make them available as src.module
from .auth import AuthHandler
from .booker import Booker
from .crossfit_booker import CrossFitBooker
from .session_config import PREFERRED_SESSIONS
from .session_manager import SessionManager
from .session_notifier import SessionNotifier
@@ -11,7 +10,6 @@ from .session_notifier import SessionNotifier
__all__ = [
"AuthHandler",
"Booker",
"CrossFitBooker",
"PREFERRED_SESSIONS",
"SessionManager",
"SessionNotifier"

46
src/book_crossfit.py
View File

@@ -1,8 +1,15 @@
#!/usr/bin/env python3
import logging
import os
import traceback
from crossfit_booker import CrossFitBooker
from src.auth import AuthHandler
from src.booker import Booker
from src.session_notifier import SessionNotifier
# Load environment variables
from dotenv import load_dotenv
load_dotenv()
if __name__ == "__main__":
# Configure logging once at script startup
@@ -19,16 +26,45 @@ if __name__ == "__main__":
logging.getLogger("requests").setLevel(logging.WARNING)
logging.info("Logging enhanced with request library noise reduction")
# Create an instance of the CrossFitBooker class
booker = CrossFitBooker()
# Initialize components
auth_handler = AuthHandler(
os.environ.get("CROSSFIT_USERNAME"),
os.environ.get("CROSSFIT_PASSWORD")
)
# Initialize notification system
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")
notifier = SessionNotifier(
email_credentials,
telegram_credentials,
enable_email=enable_email,
enable_telegram=enable_telegram
)
# Create an instance of the Booker class
booker = Booker(auth_handler, notifier)
# Attempt to log in to the CrossFit booking system
# TODO: Make a authentication during running request to not get kicked out
if not booker.login():
if not auth_handler.login():
# If login fails, log the error and exit the script
logging.error("Failed to login - Traceback: %s", traceback.format_exc())
exit(1)
# Start the continuous booking loop
booker.run()
import asyncio
asyncio.run(booker.run())
logging.info("Script completed")

193
src/crossfit_booker.py
View File

@@ -1,193 +0,0 @@
# Native modules
import os
from typing import Dict, Any, Optional, List
from datetime import date, datetime
# Third-party modules
import requests
# Import the AuthHandler class
from src.auth import AuthHandler
# Import the SessionManager class
from src.session_manager import SessionManager
# Import the Booker class
from src.booker import Booker
# Import the SessionNotifier class
from src.session_notifier import SessionNotifier
# Load environment variables
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")
class CrossFitBooker:
"""
A simple orchestrator class for the CrossFit booking system.
This class is responsible for initializing and coordinating the other components
(AuthHandler, SessionManager, and Booker) and provides a unified interface for
interacting with the booking system.
"""
def __init__(self) -> None:
"""
Initialize the CrossFitBooker with necessary components.
"""
# Initialize the AuthHandler with credentials from environment variables
self.auth_handler = AuthHandler(USERNAME, PASSWORD)
# Initialize the SessionManager with the AuthHandler
self.session_manager = SessionManager(self.auth_handler)
# 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
)
# Initialize the Booker with the AuthHandler and SessionNotifier
self.booker = Booker(self.auth_handler, self.notifier)
# Initialize a session for direct API calls
self.session = requests.Session()
def run(self) -> None:
"""
Start the booking process.
This method initiates the booking process by running the Booker's main execution loop.
"""
import asyncio
try:
asyncio.run(self.booker.run())
except Exception as e:
logging.error(f"Error in booking process: {str(e)}")
def get_auth_headers(self) -> Dict[str, str]:
"""
Return headers with authorization from the AuthHandler.
Returns:
Dict[str, str]: Headers dictionary with authorization if available.
"""
return self.auth_handler.get_auth_headers()
def login(self) -> bool:
"""
Authenticate and get the bearer token.
Returns:
bool: True if login is successful, False otherwise.
"""
return self.auth_handler.login()
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.
"""
return self.session_manager.get_available_sessions(start_date, end_date)
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.
"""
return self.session_manager.book_session(session_id)
def get_booked_sessions(self) -> List[Dict[str, Any]]:
"""
Get a list of booked sessions.
Returns:
A list of dictionaries containing information about the booked sessions.
"""
return self.session_manager.get_booked_sessions()
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.
"""
return self.session_manager.is_session_bookable(session, current_time)
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.
"""
return self.session_manager.matches_preferred_session(session, current_time)
def display_upcoming_sessions(self, sessions: List[Dict[str, Any]], current_time: datetime) -> None:
"""
Display upcoming sessions with ID, name, date, and time.
Args:
sessions (List[Dict[str, Any]]): List of session data.
current_time (datetime): Current time for comparison.
"""
self.session_manager.display_upcoming_sessions(sessions, current_time)
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.
"""
await self.booker.booker(current_time)
def quit(self) -> None:
"""
Clean up resources and exit the script.
"""
self.booker.quit()