Spaces:
Paused
Paused
File size: 1,515 Bytes
68388c0 e3cd7d9 68388c0 e3cd7d9 68388c0 e3cd7d9 68388c0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 | from __future__ import annotations
import time
from collections.abc import Callable
from datetime import datetime
from typing import Optional
def scheduled_interval_seconds(now: datetime | None = None) -> int:
current = now or datetime.now()
hour = current.hour
if 8 <= hour < 11:
return 60
if 11 <= hour < 13:
return 120
if 13 <= hour < 18:
return 300
return 3600
def run_loop(
check_once: Callable[[], None],
interval_seconds: int,
error_interval_seconds: int = 30,
interval_provider: Optional[Callable[[], int]] = None,
sleep: Callable[[int], None] = time.sleep,
run_once: bool = False,
max_runs: int | None = None,
) -> None:
runs = 0
while True:
failed = False
failed_retry_after_seconds = error_interval_seconds
try:
check_once()
except Exception as exc:
failed = True
failed_retry_after_seconds = int(
getattr(exc, "retry_after_seconds", error_interval_seconds)
)
print(f"Check failed: {exc}")
runs += 1
if run_once or (max_runs is not None and runs >= max_runs):
return
if failed:
wait_seconds = failed_retry_after_seconds
elif interval_provider:
wait_seconds = interval_provider()
else:
wait_seconds = interval_seconds
print(f"Waiting {wait_seconds} seconds before the next check.")
sleep(wait_seconds)
|