#!/usr/bin/env python3 """ Harbor Isles Pickleball Court Reservation Bot This bot automates court reservations using curl_cffi to mimic browser behavior. """ import json import os import sys import time import re from datetime import datetime, timedelta, timezone from typing import Dict, List, Optional, Any from dataclasses import dataclass, field from pathlib import Path from curl_cffi import requests # Constants TOKEN_FILE = "token_data.json" SUCCESS_TRACKING_FILE = "reservation_success.json" TOKEN_MAX_AGE_DAYS = 3 # Browser-like headers DEFAULT_HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:146.0) Gecko/20100101 Firefox/146.0", "Accept": "application/json, text/plain, */*", "Accept-Language": "en-US,en;q=0.5", "Accept-Encoding": "gzip, deflate, br, zstd", "Content-Type": "application/json", "If-Modified-Since": "Mon, 26 Jul 1997 05:00:00 GMT", "Cache-Control": "no-cache", "Pragma": "no-cache", "Connection": "keep-alive", "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-origin", "Priority": "u=0", } @dataclass class Account: name: str id: int @dataclass class Court: court: str id: int name: str resource_type_id: int assigned_resource_id: int is_assigned_resource_selectable: bool @dataclass class AttemptSchedule: time: str attempts: int delay_seconds: int @dataclass class ReservationBehavior: name: str days_in_advance: int attempt_schedule: List[AttemptSchedule] @dataclass class CourtReservation: court: str account: str @dataclass class Reservation: day: str time: str reservation_behavior: str courts: List[CourtReservation] @dataclass class Config: base_api_url: str company_id: str club_id: int appointment_item_id: int login_username: str login_password: str logged_in_customer_id: int accounts: Dict[str, Account] courts: Dict[str, Court] reservation_behaviors: Dict[str, ReservationBehavior] reservations: List[Reservation] @dataclass class TokenData: token: str token_expiration: str generated_at: str def escape_password_for_curl_display(password: str) -> str: """Convert password for display in curl command (octal escape sequences). Example: "Goliath00!!@" becomes "Goliath00\\041\\041@" for curl $'...' syntax. Note: The API itself expects the raw password, not escaped. """ result = [] for char in password: if char == '!': result.append('\\041') else: result.append(char) return ''.join(result) def load_config(config_path: str = "config.json") -> Config: """Load and validate configuration from JSON file.""" if not os.path.exists(config_path): raise FileNotFoundError(f"Configuration file not found: {config_path}") with open(config_path, 'r') as f: data = json.load(f) # Validate required fields required_fields = [ "BaseAPIUrl", "CompanyId", "ClubId", "AppointmentItemId", "Login", "LoggedInCustomerId", "Accounts", "Courts", "ReservationBehaviors", "Reservations" ] for field_name in required_fields: if field_name not in data: raise ValueError(f"Missing required configuration field: {field_name}") # Validate Login fields if "Username" not in data["Login"] or "Password" not in data["Login"]: raise ValueError("Login must contain Username and Password") # Parse accounts accounts = {} for acc in data["Accounts"]: if "Name" not in acc or "Id" not in acc: raise ValueError(f"Invalid account configuration: {acc}") accounts[acc["Name"]] = Account(name=acc["Name"], id=acc["Id"]) # Parse courts courts = {} for ct in data["Courts"]: required_court_fields = ["Court", "Id", "Name", "ResourceTypeId", "AssignedResourceId", "IsAssignedResourceSelectable"] for cf in required_court_fields: if cf not in ct: raise ValueError(f"Court missing required field '{cf}': {ct}") courts[ct["Court"]] = Court( court=ct["Court"], id=ct["Id"], name=ct["Name"], resource_type_id=ct["ResourceTypeId"], assigned_resource_id=ct["AssignedResourceId"], is_assigned_resource_selectable=ct["IsAssignedResourceSelectable"] ) # Parse reservation behaviors behaviors = {} for beh in data["ReservationBehaviors"]: if "Name" not in beh or "DaysInAdvance" not in beh or "AttemptSchedule" not in beh: raise ValueError(f"Invalid reservation behavior: {beh}") schedules = [] for sched in beh["AttemptSchedule"]: if "Time" not in sched or "Attempts" not in sched or "DelaySeconds" not in sched: raise ValueError(f"Invalid attempt schedule: {sched}") schedules.append(AttemptSchedule( time=sched["Time"], attempts=sched["Attempts"], delay_seconds=sched["DelaySeconds"] )) behaviors[beh["Name"]] = ReservationBehavior( name=beh["Name"], days_in_advance=beh["DaysInAdvance"], attempt_schedule=schedules ) # Parse reservations reservations = [] valid_days = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"] for res in data["Reservations"]: if "Day" not in res or "Time" not in res or "ReservationBehavior" not in res or "Courts" not in res: raise ValueError(f"Invalid reservation configuration: {res}") if res["Day"] not in valid_days: raise ValueError(f"Invalid day '{res['Day']}'. Must be one of: {valid_days}") if res["ReservationBehavior"] not in behaviors: raise ValueError(f"Unknown reservation behavior: {res['ReservationBehavior']}") court_reservations = [] for cr in res["Courts"]: if "Court" not in cr or "Account" not in cr: raise ValueError(f"Invalid court reservation: {cr}") if cr["Court"] not in courts: raise ValueError(f"Unknown court: {cr['Court']}") if cr["Account"] not in accounts: raise ValueError(f"Unknown account: {cr['Account']}") court_reservations.append(CourtReservation(court=cr["Court"], account=cr["Account"])) reservations.append(Reservation( day=res["Day"], time=res["Time"], reservation_behavior=res["ReservationBehavior"], courts=court_reservations )) return Config( base_api_url=data["BaseAPIUrl"], company_id=data["CompanyId"], club_id=data["ClubId"], appointment_item_id=data["AppointmentItemId"], login_username=data["Login"]["Username"], login_password=data["Login"]["Password"], logged_in_customer_id=data["LoggedInCustomerId"], accounts=accounts, courts=courts, reservation_behaviors=behaviors, reservations=reservations ) class HarborBot: """Main bot class for handling pickleball court reservations.""" def __init__(self, config: Config, config_dir: str = "."): self.config = config self.config_dir = config_dir self.token_file = os.path.join(config_dir, TOKEN_FILE) self.success_file = os.path.join(config_dir, SUCCESS_TRACKING_FILE) self.session = requests.Session(impersonate="firefox") self.token_data: Optional[TokenData] = None self._load_token() self._load_success_tracking() def _get_base_headers(self) -> Dict[str, str]: """Get base headers for requests.""" headers = DEFAULT_HEADERS.copy() headers["x-companyid"] = self.config.company_id headers["Origin"] = "https://future.ourclublogin.com" headers["Referer"] = f"https://future.ourclublogin.com/login/{self.config.company_id}" return headers def _get_auth_headers(self) -> Dict[str, str]: """Get headers with authentication token.""" if not self.token_data: raise RuntimeError("Not authenticated. Call login() first.") headers = self._get_base_headers() headers["Authorization"] = f"Bearer {self.token_data.token}" headers["x-customerid"] = str(self.config.logged_in_customer_id) headers["Referer"] = "https://future.ourclublogin.com/Appointments" return headers def _load_token(self): """Load token from file if it exists.""" if os.path.exists(self.token_file): try: with open(self.token_file, 'r') as f: data = json.load(f) self.token_data = TokenData( token=data["token"], token_expiration=data["token_expiration"], generated_at=data["generated_at"] ) print(f"[INFO] Loaded existing token from {self.token_file}") except (json.JSONDecodeError, KeyError) as e: print(f"[WARN] Failed to load token file: {e}") self.token_data = None def _save_token(self): """Save token to file.""" if self.token_data: with open(self.token_file, 'w') as f: json.dump({ "token": self.token_data.token, "token_expiration": self.token_data.token_expiration, "generated_at": self.token_data.generated_at }, f, indent=2) print(f"[INFO] Token saved to {self.token_file}") def _load_success_tracking(self): """Load success tracking data.""" self.success_tracking: Dict[str, bool] = {} if os.path.exists(self.success_file): try: with open(self.success_file, 'r') as f: self.success_tracking = json.load(f) except (json.JSONDecodeError, KeyError): self.success_tracking = {} def _save_success_tracking(self): """Save success tracking data.""" with open(self.success_file, 'w') as f: json.dump(self.success_tracking, f, indent=2) def _get_reservation_key(self, reservation_date: str, court: str, account: str) -> str: """Generate unique key for tracking reservation success.""" return f"{reservation_date}_{court}_{account}" def _is_reservation_successful(self, reservation_date: str, court: str, account: str) -> bool: """Check if a reservation was already successful.""" key = self._get_reservation_key(reservation_date, court, account) return self.success_tracking.get(key, False) def _mark_reservation_successful(self, reservation_date: str, court: str, account: str): """Mark a reservation as successful.""" key = self._get_reservation_key(reservation_date, court, account) self.success_tracking[key] = True self._save_success_tracking() def _needs_login(self) -> bool: """Check if login is needed (token expired or too old).""" if not self.token_data: return True # Check if token is expired try: expiration = datetime.fromisoformat(self.token_data.token_expiration.replace('Z', '+00:00')) if datetime.now(timezone.utc) >= expiration: print("[INFO] Token expired") return True except ValueError: print("[WARN] Could not parse token expiration") return True # Check if token is older than TOKEN_MAX_AGE_DAYS try: generated = datetime.fromisoformat(self.token_data.generated_at.replace('Z', '+00:00')) age = datetime.now(timezone.utc) - generated if age.days >= TOKEN_MAX_AGE_DAYS: print(f"[INFO] Token is {age.days} days old, refreshing") return True except ValueError: print("[WARN] Could not parse token generation time") return True return False def login(self) -> bool: """Perform login to get authentication token.""" print("[INFO] Logging in...") headers = self._get_base_headers() headers["x-customerid"] = "0" # API expects raw password - curl uses $'...' syntax with \041 for display payload = { "UserLogin": self.config.login_username, "Pswd": self.config.login_password } url = f"{self.config.base_api_url}/CustomerAuth/CustomerLogin" try: response = self.session.post( url, headers=headers, json=payload, cookies={"coid": self.config.company_id} ) response.raise_for_status() data = response.json() if data.get("LoginResult") != 1: print(f"[ERROR] Login failed: {data.get('LoginError', 'Unknown error')}") return False # Extract token data token_info = data.get("data", {}) if not token_info.get("token"): print("[ERROR] No token in login response") return False self.token_data = TokenData( token=token_info["token"], token_expiration=token_info["tokenExpiration"], generated_at=datetime.now(timezone.utc).isoformat() ) self._save_token() print(f"[INFO] Login successful. Customer: {data.get('CustomerName', {}).get('DisplayName', 'Unknown')}") return True except Exception as e: print(f"[ERROR] Login request failed: {e}") return False def ensure_authenticated(self) -> bool: """Ensure we have a valid authentication token.""" if self._needs_login(): return self.login() print("[INFO] Using existing valid token") return True def book_appointment( self, court: Court, account: Account, start_date: str ) -> Dict[str, Any]: """ Book a court appointment. Args: court: Court to book account: Account to book under (PrimaryCustomerId) start_date: ISO format datetime string (e.g., "2026-01-15T17:30:00-08:00") Returns: API response as dict """ if not self.ensure_authenticated(): return {"Success": False, "Message": "Authentication failed"} headers = self._get_auth_headers() payload = { "ClubId": self.config.club_id, "LoggedInCustomerId": self.config.logged_in_customer_id, "PrimaryCustomerId": account.id, "AdditionalCustomerIds": [], "AppointmentItemId": self.config.appointment_item_id, "SelectedBooks": [{ "Id": court.id, "Name": court.name, "ResourceTypeId": court.resource_type_id, "AssignedResourceId": court.assigned_resource_id, "IsAssignedResourceSelectable": court.is_assigned_resource_selectable }], "PackageItemId": 0, "PackageQuantity": 0, "ChangeFeeId": 0, "StartDate": start_date, "UserDisplayedPayNowGrandTotal": 0, "DisplayedAmountDueAtTimeOfService": 0, "CancellationAppointmentId": 0 } url = f"{self.config.base_api_url}/TransactionProcessing/BookAppointmentOnAccount" try: response = self.session.post( url, headers=headers, json=payload, cookies={"coid": self.config.company_id} ) response.raise_for_status() return response.json() except Exception as e: print(f"[ERROR] Book appointment request failed: {e}") return {"Success": False, "Message": str(e)} def calculate_reservation_date(self, day_name: str, days_in_advance: int) -> Optional[datetime]: """ Calculate the target reservation date if we should make it today. For example: if days_in_advance=7 and day_name="Wednesday", we can only book Wednesday's court on the Wednesday that is 7 days before it. Returns the target date if today is the trigger day, None otherwise. """ day_mapping = { "Monday": 0, "Tuesday": 1, "Wednesday": 2, "Thursday": 3, "Friday": 4, "Saturday": 5, "Sunday": 6 } target_day = day_mapping.get(day_name) if target_day is None: return None today = datetime.now() today_weekday = today.weekday() # The target date is days_in_advance days from today target_date = today + timedelta(days=days_in_advance) # Check if that target date falls on the correct day of week if target_date.weekday() == target_day: return target_date return None def get_next_reservation_info(self, reservation: Reservation) -> Dict[str, Any]: """Get information about when a reservation would next trigger.""" behavior = self.config.reservation_behaviors[reservation.reservation_behavior] day_mapping = { "Monday": 0, "Tuesday": 1, "Wednesday": 2, "Thursday": 3, "Friday": 4, "Saturday": 5, "Sunday": 6 } target_day = day_mapping.get(reservation.day) today = datetime.now() today_weekday = today.weekday() # Find next occurrence of target day that is days_in_advance away for offset in range(14): # Check next 2 weeks check_date = today + timedelta(days=offset) target_date = check_date + timedelta(days=behavior.days_in_advance) if target_date.weekday() == target_day: return { "trigger_date": check_date.strftime("%Y-%m-%d"), "reservation_date": target_date.strftime("%Y-%m-%d"), "days_until_trigger": offset } return {"error": "Could not calculate next reservation"} def execute_reservation_attempts( self, reservation: Reservation, behavior: ReservationBehavior, target_date: datetime, dry_run: bool = True ): """ Execute reservation attempts according to the behavior schedule. Args: reservation: The reservation configuration behavior: The reservation behavior with attempt schedule target_date: The target date for the reservation dry_run: If True, just print what would happen instead of actually booking """ # Format the reservation datetime reservation_datetime = target_date.strftime(f"%Y-%m-%dT{reservation.time[:8]}") + reservation.time[8:] reservation_date_str = target_date.strftime("%Y-%m-%d") print(f"\n[INFO] Reservation execution for {reservation.day} at {reservation.time}") print(f"[INFO] Target date: {reservation_datetime}") print(f"[INFO] Using behavior: {behavior.name}") for schedule in behavior.attempt_schedule: print(f"\n[INFO] Attempt window: {schedule.time} - {schedule.attempts} attempts with {schedule.delay_seconds}s delay") for court_res in reservation.courts: court = self.config.courts[court_res.court] account = self.config.accounts[court_res.account] # Check if already successful if self._is_reservation_successful(reservation_date_str, court_res.court, court_res.account): print(f" [SKIP] {court.name} for {account.name} - already booked successfully") continue for attempt in range(schedule.attempts): if dry_run: print(f" [DRY RUN] Attempt {attempt + 1}/{schedule.attempts}: " f"Would book {court.name} for {account.name} at {reservation_datetime}") else: print(f" [ATTEMPT] {attempt + 1}/{schedule.attempts}: " f"Booking {court.name} for {account.name}...") result = self.book_appointment(court, account, reservation_datetime) if result.get("Success"): print(f" [SUCCESS] Booked {court.name} for {account.name}") self._mark_reservation_successful(reservation_date_str, court_res.court, court_res.account) break else: print(f" [FAILED] {result.get('Message', 'Unknown error')}") if result.get("AllowRetry") is False and "no longer available" in result.get("Message", ""): print(f" [INFO] Slot no longer available, moving to next court") break if attempt < schedule.attempts - 1 and schedule.delay_seconds > 0: if dry_run: print(f" [DRY RUN] Would wait {schedule.delay_seconds}s") else: time.sleep(schedule.delay_seconds) def run_scheduler(self, dry_run: bool = True): """ Main scheduler loop that checks for reservations to make. Args: dry_run: If True, don't actually make reservations, just print what would happen """ print(f"\n{'='*60}") print(f"Harbor Bot Scheduler - {'DRY RUN' if dry_run else 'LIVE MODE'}") print(f"{'='*60}") if not self.ensure_authenticated(): print("[ERROR] Failed to authenticate. Exiting.") return print(f"\n[INFO] Checking {len(self.config.reservations)} configured reservations...") for reservation in self.config.reservations: behavior = self.config.reservation_behaviors[reservation.reservation_behavior] # Calculate if we should make this reservation today target_date = self.calculate_reservation_date(reservation.day, behavior.days_in_advance) if target_date: print(f"\n[TRIGGER] Reservation for {reservation.day} at {reservation.time}") print(f"[INFO] Target date: {target_date.strftime('%Y-%m-%d')}") self.execute_reservation_attempts(reservation, behavior, target_date, dry_run) else: print(f"\n[SKIP] {reservation.day} at {reservation.time} - not scheduled for today") def test_login(self): """Test the login functionality.""" print("\n[TEST] Testing login...") if self.login(): print("[TEST] Login successful!") print(f"[TEST] Token expires: {self.token_data.token_expiration}") else: print("[TEST] Login failed!") def show_config(self): """Display the loaded configuration.""" print("\n" + "="*60) print("CONFIGURATION SUMMARY") print("="*60) print(f"\nAPI URL: {self.config.base_api_url}") print(f"Company ID: {self.config.company_id}") print(f"Club ID: {self.config.club_id}") print(f"Login User: {self.config.login_username}") print(f"\nAccounts ({len(self.config.accounts)}):") for name, acc in self.config.accounts.items(): print(f" - {name}: ID={acc.id}") print(f"\nCourts ({len(self.config.courts)}):") for name, court in self.config.courts.items(): print(f" - {name}: {court.name} (ID={court.id})") print(f"\nReservation Behaviors ({len(self.config.reservation_behaviors)}):") for name, beh in self.config.reservation_behaviors.items(): print(f" - {name}: {beh.days_in_advance} days in advance") for sched in beh.attempt_schedule: print(f" @ {sched.time}: {sched.attempts} attempts, {sched.delay_seconds}s delay") print(f"\nReservations ({len(self.config.reservations)}):") for res in self.config.reservations: next_info = self.get_next_reservation_info(res) trigger_info = "" if "trigger_date" in next_info: if next_info["days_until_trigger"] == 0: trigger_info = f" [TRIGGERS TODAY for {next_info['reservation_date']}]" else: trigger_info = f" [Next trigger: {next_info['trigger_date']} for {next_info['reservation_date']}]" print(f" - {res.day} @ {res.time} ({res.reservation_behavior}){trigger_info}") for cr in res.courts: print(f" {cr.court} -> {cr.account}") def test_booking(self, court_name: str, account_name: str, date_str: str, time_str: str, dry_run: bool = True): """ Test booking a specific court for a specific account. Args: court_name: Court identifier (e.g., "Court_1") account_name: Account name (e.g., "Saleh") date_str: Date in YYYY-MM-DD format time_str: Time with timezone (e.g., "19:00:00-08:00") dry_run: If True, just print what would happen """ if court_name not in self.config.courts: print(f"[ERROR] Unknown court: {court_name}") print(f"[INFO] Available courts: {', '.join(self.config.courts.keys())}") return if account_name not in self.config.accounts: print(f"[ERROR] Unknown account: {account_name}") print(f"[INFO] Available accounts: {', '.join(self.config.accounts.keys())}") return court = self.config.courts[court_name] account = self.config.accounts[account_name] reservation_datetime = f"{date_str}T{time_str}" print(f"\n[TEST BOOKING] {'DRY RUN' if dry_run else 'LIVE'}") print(f" Court: {court.name} (ID={court.id})") print(f" Account: {account.name} (ID={account.id})") print(f" DateTime: {reservation_datetime}") if dry_run: print(f"\n[DRY RUN] Would book {court.name} for {account.name} at {reservation_datetime}") else: if not self.ensure_authenticated(): print("[ERROR] Authentication failed") return result = self.book_appointment(court, account, reservation_datetime) if result.get("Success"): print(f"\n[SUCCESS] Booked {court.name} for {account.name}") else: print(f"\n[FAILED] {result.get('Message', 'Unknown error')}") print(f"[DEBUG] Full response: {json.dumps(result, indent=2)}") def main(): """Main entry point.""" import argparse parser = argparse.ArgumentParser(description="Harbor Isles Pickleball Court Reservation Bot") parser.add_argument("--config", default="config.json", help="Path to configuration file") parser.add_argument("--test-login", action="store_true", help="Test login functionality") parser.add_argument("--show-config", action="store_true", help="Display loaded configuration") parser.add_argument("--run", action="store_true", help="Run the scheduler (dry run by default)") parser.add_argument("--live", action="store_true", help="Actually make reservations (use with --run or --test-book)") parser.add_argument("--test-book", nargs=4, metavar=("COURT", "ACCOUNT", "DATE", "TIME"), help="Test booking: --test-book Court_1 Saleh 2026-01-19 19:00:00-08:00") args = parser.parse_args() # Determine config directory config_path = Path(args.config) config_dir = str(config_path.parent) if config_path.parent != Path(".") else "." try: print(f"[INFO] Loading configuration from {args.config}...") config = load_config(args.config) print("[INFO] Configuration loaded successfully!") bot = HarborBot(config, config_dir) if args.show_config: bot.show_config() if args.test_login: bot.test_login() if args.test_book: court, account, date, time = args.test_book dry_run = not args.live bot.test_booking(court, account, date, time, dry_run=dry_run) if args.run: dry_run = not args.live bot.run_scheduler(dry_run=dry_run) if not any([args.show_config, args.test_login, args.run, args.test_book]): parser.print_help() except FileNotFoundError as e: print(f"[ERROR] {e}") sys.exit(1) except ValueError as e: print(f"[ERROR] Configuration error: {e}") sys.exit(1) except Exception as e: print(f"[ERROR] Unexpected error: {e}") raise if __name__ == "__main__": main()