""" Rate Limit Handler für HTTP 429 und ähnliche Fehler. Dieses Modul implementiert exponentielles Backoff für Rate-Limiting, um automatisch auf zu viele Anfragen zu reagieren und Sperren zu vermeiden. """ import logging import time import random from typing import Callable, Any, Optional, List logger = logging.getLogger("rate_limit_handler") class RateLimitHandler: """ Behandelt Rate-Limits mit exponentiellem Backoff. Diese Klasse implementiert eine robuste Strategie zum Umgang mit Rate-Limiting durch soziale Netzwerke. Bei Erkennung eines Rate-Limits wird exponentiell länger gewartet, um Sperren zu vermeiden. Beispiel: handler = RateLimitHandler() # Option 1: Manuelles Handling if rate_limit_detected: handler.handle_rate_limit() # Option 2: Automatisches Retry result = handler.execute_with_backoff(my_function, arg1, arg2) """ # Bekannte Rate-Limit-Indikatoren RATE_LIMIT_INDICATORS = [ # HTTP Status Codes '429', 'rate limit', 'rate_limit', 'ratelimit', # Englische Meldungen 'too many requests', 'too many attempts', 'slow down', 'try again later', 'temporarily blocked', 'please wait', 'request blocked', # Deutsche Meldungen 'zu viele anfragen', 'zu viele versuche', 'später erneut versuchen', 'vorübergehend gesperrt', 'bitte warten', # Plattform-spezifische Meldungen 'challenge_required', # Instagram 'checkpoint_required', # Instagram/Facebook 'feedback_required', # Instagram 'spam', # Generisch 'suspicious activity', # Generisch 'unusual activity', # Generisch ] def __init__(self, initial_delay: float = 60.0, max_delay: float = 600.0, backoff_multiplier: float = 2.0, max_retries: int = 5, jitter_factor: float = 0.2): """ Initialisiert den Rate-Limit-Handler. Args: initial_delay: Anfängliche Wartezeit in Sekunden (Standard: 60s = 1 Minute) max_delay: Maximale Wartezeit in Sekunden (Standard: 600s = 10 Minuten) backoff_multiplier: Multiplikator für exponentielles Backoff (Standard: 2.0) max_retries: Maximale Anzahl an Wiederholungsversuchen (Standard: 5) jitter_factor: Faktor für zufällige Variation (Standard: 0.2 = ±20%) """ self.initial_delay = initial_delay self.max_delay = max_delay self.backoff_multiplier = backoff_multiplier self.max_retries = max_retries self.jitter_factor = jitter_factor # Status-Tracking self.current_retry = 0 self.last_rate_limit_time = 0 self.total_rate_limits = 0 self.consecutive_successes = 0 def is_rate_limited(self, response_text: str) -> bool: """ Prüft, ob eine Antwort auf ein Rate-Limit hindeutet. Args: response_text: Text der Antwort (z.B. Seiteninhalt, Fehlermeldung) Returns: True wenn Rate-Limit erkannt wurde, sonst False """ if not response_text: return False response_lower = response_text.lower() for indicator in self.RATE_LIMIT_INDICATORS: if indicator in response_lower: logger.warning(f"Rate-Limit Indikator gefunden: '{indicator}'") return True return False def calculate_delay(self, retry_count: int = None) -> float: """ Berechnet die Backoff-Verzögerung. Args: retry_count: Aktueller Wiederholungsversuch (optional) Returns: Verzögerung in Sekunden """ if retry_count is None: retry_count = self.current_retry # Exponentielles Backoff berechnen delay = self.initial_delay * (self.backoff_multiplier ** retry_count) # Jitter hinzufügen (zufällige Variation) jitter = delay * random.uniform(-self.jitter_factor, self.jitter_factor) delay = delay + jitter # Maximum nicht überschreiten delay = min(delay, self.max_delay) return delay def handle_rate_limit(self, retry_count: int = None, on_waiting: Optional[Callable[[float, int], None]] = None) -> float: """ Behandelt ein erkanntes Rate-Limit mit Backoff. Args: retry_count: Aktueller Wiederholungsversuch on_waiting: Optionaler Callback während des Wartens (delay, retry) Returns: Tatsächlich gewartete Zeit in Sekunden """ if retry_count is None: retry_count = self.current_retry delay = self.calculate_delay(retry_count) logger.warning( f"Rate-Limit erkannt! Warte {delay:.1f}s " f"(Versuch {retry_count + 1}/{self.max_retries})" ) # Callback aufrufen falls vorhanden if on_waiting: on_waiting(delay, retry_count + 1) # Warten time.sleep(delay) # Status aktualisieren self.current_retry = retry_count + 1 self.last_rate_limit_time = time.time() self.total_rate_limits += 1 self.consecutive_successes = 0 return delay def execute_with_backoff(self, func: Callable, *args, on_retry: Optional[Callable[[int, Exception], None]] = None, **kwargs) -> Any: """ Führt eine Funktion mit automatischem Backoff bei Rate-Limits aus. Args: func: Auszuführende Funktion *args: Positionsargumente für die Funktion on_retry: Optionaler Callback bei Retry (retry_count, exception) **kwargs: Keyword-Argumente für die Funktion Returns: Rückgabewert der Funktion oder None bei Fehler Raises: Exception: Wenn max_retries erreicht oder nicht-Rate-Limit-Fehler """ last_exception = None for attempt in range(self.max_retries): try: result = func(*args, **kwargs) # Erfolg - Reset Retry-Zähler self.current_retry = 0 self.consecutive_successes += 1 # Nach mehreren Erfolgen: Backoff-Zähler langsam reduzieren if self.consecutive_successes >= 3: self.total_rate_limits = max(0, self.total_rate_limits - 1) return result except Exception as e: last_exception = e error_str = str(e).lower() # Prüfe auf Rate-Limit-Indikatoren is_rate_limit = any( indicator in error_str for indicator in self.RATE_LIMIT_INDICATORS ) if is_rate_limit: logger.warning(f"Rate-Limit Exception erkannt: {e}") if on_retry: on_retry(attempt, e) self.handle_rate_limit(attempt) else: # Anderer Fehler - nicht durch Backoff lösbar logger.error(f"Nicht-Rate-Limit Fehler: {e}") raise # Maximum erreicht logger.error( f"Maximale Wiederholungsversuche ({self.max_retries}) erreicht. " f"Letzter Fehler: {last_exception}" ) return None def should_slow_down(self) -> bool: """ Prüft, ob die Geschwindigkeit reduziert werden sollte. Basierend auf der Anzahl der kürzlichen Rate-Limits wird empfohlen, ob zusätzliche Verzögerungen eingebaut werden sollten. Returns: True wenn Verlangsamung empfohlen, sonst False """ # Wenn kürzlich (< 5 min) ein Rate-Limit war time_since_last = time.time() - self.last_rate_limit_time if time_since_last < 300 and self.last_rate_limit_time > 0: return True # Wenn viele Rate-Limits insgesamt if self.total_rate_limits >= 3: return True return False def get_recommended_delay(self) -> float: """ Gibt eine empfohlene zusätzliche Verzögerung zurück. Basierend auf dem aktuellen Status wird eine Verzögerung empfohlen, die zwischen Aktionen eingefügt werden sollte. Returns: Empfohlene Verzögerung in Sekunden """ if not self.should_slow_down(): return 0.0 # Basis-Verzögerung basierend auf Anzahl der Rate-Limits base_delay = 5.0 * self.total_rate_limits # Zusätzliche Verzögerung wenn kürzlich Rate-Limit war time_since_last = time.time() - self.last_rate_limit_time if time_since_last < 300: # Je kürzer her, desto länger warten recency_factor = 1.0 - (time_since_last / 300) base_delay += 10.0 * recency_factor return min(base_delay, 30.0) # Maximum 30 Sekunden def reset(self): """Setzt den Handler auf Anfangszustand zurück.""" self.current_retry = 0 self.last_rate_limit_time = 0 self.total_rate_limits = 0 self.consecutive_successes = 0 logger.info("Rate-Limit Handler zurückgesetzt") def get_status(self) -> dict: """ Gibt den aktuellen Status des Handlers zurück. Returns: Dictionary mit Status-Informationen """ return { "current_retry": self.current_retry, "total_rate_limits": self.total_rate_limits, "consecutive_successes": self.consecutive_successes, "last_rate_limit_time": self.last_rate_limit_time, "should_slow_down": self.should_slow_down(), "recommended_delay": self.get_recommended_delay(), } # Globale Instanz für einfache Verwendung _default_handler: Optional[RateLimitHandler] = None def get_default_handler() -> RateLimitHandler: """ Gibt die globale Standard-Instanz des Rate-Limit-Handlers zurück. Returns: RateLimitHandler-Instanz """ global _default_handler if _default_handler is None: _default_handler = RateLimitHandler() return _default_handler def handle_rate_limit(retry_count: int = None) -> float: """ Convenience-Funktion für Rate-Limit-Handling mit Standard-Handler. Args: retry_count: Aktueller Wiederholungsversuch Returns: Gewartete Zeit in Sekunden """ return get_default_handler().handle_rate_limit(retry_count) def is_rate_limited(response_text: str) -> bool: """ Convenience-Funktion für Rate-Limit-Erkennung. Args: response_text: Zu prüfender Text Returns: True wenn Rate-Limit erkannt """ return get_default_handler().is_rate_limited(response_text)