بوت دعم لمتجرك بلغة Python: الكود المصدري

1"""Per-user sliding-window rate limiter."""23from __future__ import annotations45import threading6import time7from collections import deque8from collections.abc import Callable91011class RateLimiter:12    """Allows at most `limit` events per `window_s` seconds for each key."""1314    def __init__(self, limit: int, window_s: float = 60.0, clock: Callable[[], float] = time.monotonic) -> None:15        if limit < 1:16            raise ValueError("limit must be at least 1")17        self._limit = limit18        self._window = window_s19        self._clock = clock20        self._events: dict[int, deque[float]] = {}21        self._lock = threading.Lock()2223    def check(self, key: int) -> float:24        """Record an event and return 0, or return the seconds to wait if the key is over the limit."""25        now = self._clock()26        with self._lock:27            events = self._events.setdefault(key, deque())28            while events and now - events[0] >= self._window:29                events.popleft()30            if len(events) >= self._limit:31                return max(0.0, self._window - (now - events[0]))32            events.append(now)33            self._prune(now)34            return 0.03536    def _prune(self, now: float) -> None:37        if len(self._events) < 5_000:38            return39        for key in [k for k, ev in self._events.items() if not ev or now - ev[-1] >= self._window]:40            del self._events[key]