@ -521,6 +521,70 @@ class HarborBot:
}
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 ,
@ -530,6 +594,7 @@ class HarborBot:
) :
"""
Execute reservation attempts according to the behavior schedule .
Waits for the scheduled times before making attempts .
Args :
reservation : The reservation configuration
@ -545,9 +610,25 @@ class HarborBot:
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 ]
@ -583,9 +664,95 @@ class HarborBot:
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
while True :
try :
self . _daemon_iteration ( dry_run )
# Sleep until next check (check every minute)
time . sleep ( 60 )
except KeyboardInterrupt :
print ( " \n [INFO] Daemon stopped by user " )
break
except Exception as e :
print ( f " [ERROR] Daemon error: { e } " )
print ( " [INFO] Sleeping 60s before retry... " )
time . sleep ( 60 )
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 ( )
# Get first and last attempt times
first_schedule = behavior . attempt_schedule [ 0 ]
last_schedule = behavior . attempt_schedule [ - 1 ]
first_time = self . _get_schedule_datetime ( first_schedule . time , datetime . combine ( today , datetime . min . time ( ) ) )
last_time = self . _get_schedule_datetime ( last_schedule . time , datetime . combine ( today , datetime . min . time ( ) ) )
# Add buffer for last attempt window (assume max 10 minutes for attempts)
last_time_with_buffer = last_time + timedelta ( minutes = 10 )
# Check if we're before the first window (wait) or past the last window (skip)
if now < first_time - timedelta ( minutes = 5 ) :
# More than 5 minutes before first attempt - not time yet
wait_seconds = ( first_time - now ) . total_seconds ( )
if wait_seconds < 300 : # Less than 5 minutes away
print ( f " \n [INFO] { reservation . day } @ { reservation . time } : Starting in { wait_seconds : .0f } s " )
continue
if now > last_time_with_buffer :
# Past all attempt windows for today
continue
# We're within the attempt window - execute
print ( f " \n [TRIGGER] { reservation . day } @ { reservation . time } for { target_date . strftime ( ' % Y- % m- %d ' ) } " )
self . execute_reservation_attempts ( reservation , behavior , target_date , dry_run )
# 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 ) :
"""
Main scheduler loop that checks for reservations to make .
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
@ -716,8 +883,9 @@ def main():
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 ( " --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 " )
@ -741,15 +909,19 @@ def main():
bot . test_login ( )
if args . test_book :
court , account , date , time = args . test_book
court , account , date , time _val = args . test_book
dry_run = not args . live
bot . test_booking ( court , account , date , time , dry_run = dry_run )
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 not any ( [ args . show_config , args . test_login , args . run , args . test_book ] ) :
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 :
