import time import uuid import threading from typing import Optional, Callable class RateLimiter: def __init__(self, rpm: int = 60): self.max_permits = rpm self.interval = 60.0 / rpm self.last_refill = time.monotonic() self.permits = rpm self._lock = threading.Lock() def acquire(self) -> float: with self._lock: now = time.monotonic() elapsed = now - self.last_refill self.permits += elapsed * (self.max_permits / 60.0) self.last_refill = now if self.permits > self.max_permits: self.permits = self.max_permits if self.permits < 1: wait = (1 - self.permits) * (60.0 / self.max_permits) self.permits = 0 return wait self.permits -= 1 return 0.0 def generate_ui_ref() -> str: import random chars = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" rand = "".join(random.choice(chars) for _ in range(13)) return f"W{int(time.time() * 1000)}{rand}" def retry(max_retries: int = 3, delay: float = 1.0, backoff: float = 2.0): def decorator(func): def wrapper(*args, **kwargs): last_error = None current_delay = delay for attempt in range(max_retries): try: return func(*args, **kwargs) except Exception as e: last_error = e if attempt < max_retries - 1: time.sleep(current_delay) current_delay *= backoff else: raise last_error return wrapper return decorator