#!/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}") if not beh["AttemptSchedule"]: raise ValueError(f"ReservationBehavior '{beh['Name']}' must have at least one AttemptSchedule entry") schedules = [] prev_time_minutes = -1 for i, sched in enumerate(beh["AttemptSchedule"]): if "Time" not in sched or "Attempts" not in sched or "DelaySeconds" not in sched: raise ValueError(f"Invalid attempt schedule in '{beh['Name']}': {sched}") # Validate time format and check ascending order time_str = sched["Time"] try: # Parse time "HH:MM:SS-TZ:00" format if len(time_str) < 8: raise ValueError(f"Invalid time format") time_part = time_str[:8] parts = time_part.split(':') if len(parts) != 3: raise ValueError(f"Invalid time format") hour = int(parts[0]) minute = int(parts[1]) second = int(parts[2]) if not (0 <= hour <= 23 and 0 <= minute <= 59 and 0 <= second <= 59): raise ValueError(f"Invalid time values") # Convert to minutes for comparison current_time_minutes = hour * 60 + minute if current_time_minutes <= prev_time_minutes: raise ValueError( f"AttemptSchedule times in '{beh['Name']}' must be in ascending order. " f"Entry {i+1} time '{time_str}' is not after previous entry. " f"Please sort AttemptSchedule entries by time." ) prev_time_minutes = current_time_minutes except ValueError as e: if "ascending order" in str(e) or "AttemptSchedule" in str(e): raise raise ValueError(f"Invalid time format '{time_str}' in '{beh['Name']}': {e}") 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 _parse_schedule_time(self, time_str: str) -> tuple: """Parse a schedule time string like '04:29:00-08:00' into (hour, minute, second, tz_offset_hours).""" # Parse time portion time_part = time_str[:8] # "04:29:00" tz_part = time_str[8:] # "-08:00" parts = time_part.split(':') hour = int(parts[0]) minute = int(parts[1]) second = int(parts[2]) # Parse timezone offset tz_sign = 1 if tz_part[0] == '+' else -1 tz_hours = int(tz_part[1:3]) tz_offset = tz_sign * tz_hours return (hour, minute, second, tz_offset) def _get_schedule_datetime(self, schedule_time: str, for_date: datetime) -> datetime: """Get a datetime object for a schedule time on a specific date.""" hour, minute, second, tz_offset = self._parse_schedule_time(schedule_time) # Create timezone-aware datetime tz = timezone(timedelta(hours=tz_offset)) return datetime( for_date.year, for_date.month, for_date.day, hour, minute, second, tzinfo=tz ) def _get_current_time_in_tz(self, tz_offset_hours: int) -> datetime: """Get current time in the specified timezone offset.""" tz = timezone(timedelta(hours=tz_offset_hours)) return datetime.now(tz) def _wait_until(self, target_time: datetime, dry_run: bool = False) -> bool: """ Wait until the target time. Returns True if we reached the time, False if interrupted. """ while True: now = datetime.now(target_time.tzinfo) if now >= target_time: return True wait_seconds = (target_time - now).total_seconds() if wait_seconds > 60: # Long wait - sleep in chunks and show status if dry_run: print(f" [DRY RUN] Would wait {wait_seconds:.0f}s until {target_time.strftime('%H:%M:%S')}") return True print(f" [WAIT] {wait_seconds:.0f}s until {target_time.strftime('%H:%M:%S %Z')}") time.sleep(min(60, wait_seconds)) elif wait_seconds > 0: if dry_run: print(f" [DRY RUN] Would wait {wait_seconds:.1f}s") return True time.sleep(wait_seconds) else: return True def execute_reservation_attempts( self, reservation: Reservation, behavior: ReservationBehavior, target_date: datetime, dry_run: bool = True ): """ Execute reservation attempts according to the behavior schedule. Waits for the scheduled times before making attempts. 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}") # Get timezone from reservation time for scheduling _, _, _, tz_offset = self._parse_schedule_time(reservation.time) today = self._get_current_time_in_tz(tz_offset).date() for schedule in behavior.attempt_schedule: # Calculate the exact time for this attempt window schedule_datetime = self._get_schedule_datetime(schedule.time, datetime.combine(today, datetime.min.time())) print(f"\n[INFO] Attempt window: {schedule.time} - {schedule.attempts} attempts with {schedule.delay_seconds}s delay") # Wait for the scheduled time now = self._get_current_time_in_tz(tz_offset) if now < schedule_datetime: print(f"[INFO] Waiting for scheduled time: {schedule_datetime.strftime('%H:%M:%S %Z')}") self._wait_until(schedule_datetime, dry_run) elif (now - schedule_datetime).total_seconds() > 300: # More than 5 minutes past print(f"[SKIP] Attempt window {schedule.time} has passed (>5 min ago)") continue 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_daemon(self, dry_run: bool = True): """ Run as a continuous daemon, checking for and executing reservations. Args: dry_run: If True, don't actually make reservations, just print what would happen """ print(f"\n{'='*60}") print(f"Harbor Bot Daemon - {'DRY RUN' if dry_run else 'LIVE MODE'}") print(f"{'='*60}") print(f"[INFO] Started at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"[INFO] Configured reservations: {len(self.config.reservations)}") if not self.ensure_authenticated(): print("[ERROR] Failed to authenticate. Exiting.") return # Print initial status showing next scheduled attempts self._print_upcoming_reservations() last_status_print = datetime.now() last_date = datetime.now().date() while True: try: # Check for day change current_date = datetime.now().date() if current_date != last_date: print(f"\n[INFO] {'='*50}") print(f"[INFO] New day detected: {current_date.strftime('%A, %B %d, %Y')}") print(f"[INFO] {'='*50}") self._print_upcoming_reservations() last_date = current_date last_status_print = datetime.now() self._daemon_iteration(dry_run) # Print status every 30 minutes if (datetime.now() - last_status_print).total_seconds() > 1800: self._print_upcoming_reservations() last_status_print = datetime.now() # Sleep until next check (check every 30 seconds for better timing accuracy) time.sleep(30) except KeyboardInterrupt: print("\n[INFO] Daemon stopped by user") break except Exception as e: print(f"[ERROR] Daemon error: {e}") import traceback traceback.print_exc() print("[INFO] Sleeping 60s before retry...") time.sleep(60) time.sleep(60) def _print_upcoming_reservations(self): """Print status of upcoming reservations.""" print(f"\n[STATUS] {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - Checking upcoming reservations:") upcoming = [] for reservation in self.config.reservations: behavior = self.config.reservation_behaviors[reservation.reservation_behavior] target_date = self.calculate_reservation_date(reservation.day, behavior.days_in_advance) if not target_date: continue # Get timezone from reservation time _, _, _, tz_offset = self._parse_schedule_time(reservation.time) now = self._get_current_time_in_tz(tz_offset) today = now.date() # Find next upcoming attempt window next_attempt_time = None for schedule in behavior.attempt_schedule: sched_time = self._get_schedule_datetime(schedule.time, datetime.combine(today, datetime.min.time())) if sched_time > now: next_attempt_time = sched_time break if next_attempt_time: wait_seconds = (next_attempt_time - now).total_seconds() upcoming.append({ 'reservation': reservation, 'target_date': target_date, 'next_attempt': next_attempt_time, 'wait_seconds': wait_seconds }) if not upcoming: print(" No reservations scheduled for today") else: # Sort by next attempt time upcoming.sort(key=lambda x: x['next_attempt']) for item in upcoming: res = item['reservation'] wait = item['wait_seconds'] if wait < 60: wait_str = f"{wait:.0f}s" elif wait < 3600: wait_str = f"{wait/60:.1f}m" else: wait_str = f"{wait/3600:.1f}h" print(f" - {res.day} @ {res.time} -> {item['target_date'].strftime('%Y-%m-%d')}: " f"Next attempt at {item['next_attempt'].strftime('%H:%M:%S')} (in {wait_str})") def _daemon_iteration(self, dry_run: bool): """Single iteration of the daemon loop.""" # Re-check authentication periodically if self._needs_login(): print("[INFO] Refreshing authentication...") if not self.login(): print("[ERROR] Failed to refresh authentication") return # Check each reservation 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 not target_date: continue # Not scheduled for today # Check if we're within the attempt window for today _, _, _, tz_offset = self._parse_schedule_time(reservation.time) now = self._get_current_time_in_tz(tz_offset) today = now.date() # Find if we're within any attempt window (within 60 seconds of a scheduled time) for schedule in behavior.attempt_schedule: sched_time = self._get_schedule_datetime(schedule.time, datetime.combine(today, datetime.min.time())) time_diff = (now - sched_time).total_seconds() # Trigger if we're within 0-60 seconds after the scheduled time if 0 <= time_diff <= 60: print(f"\n[TRIGGER] {reservation.day} @ {reservation.time} for {target_date.strftime('%Y-%m-%d')}") print(f"[INFO] Attempt window: {schedule.time} (triggered at {now.strftime('%H:%M:%S')})") self._execute_single_attempt_window(reservation, schedule, target_date, dry_run) break # Only execute once per iteration def _execute_single_attempt_window( self, reservation: Reservation, schedule: AttemptSchedule, target_date: datetime, dry_run: bool ): """Execute attempts for a single attempt window.""" 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"[INFO] Booking for: {reservation_datetime}") print(f"[INFO] Attempts: {schedule.attempts}, Delay: {schedule.delay_seconds}s") 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}") 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 not dry_run: time.sleep(schedule.delay_seconds) # After executing, we'll skip this reservation until tomorrow # (the success tracking prevents re-booking, and time checks prevent re-triggering) def run_scheduler(self, dry_run: bool = True): """ One-time scheduler check (for backward compatibility). Use run_daemon() for continuous operation. 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 once and check for reservations (one-shot mode)") parser.add_argument("--daemon", action="store_true", help="Run continuously as a daemon (recommended)") parser.add_argument("--live", action="store_true", help="Actually make reservations (use with --run, --daemon, 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_val = args.test_book dry_run = not args.live bot.test_booking(court, account, date, time_val, dry_run=dry_run) if args.run: dry_run = not args.live bot.run_scheduler(dry_run=dry_run) if args.daemon: dry_run = not args.live bot.run_daemon(dry_run=dry_run) if not any([args.show_config, args.test_login, args.run, args.test_book, args.daemon]): 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()