diff --git a/src/scraper.py b/src/scraper.py index f36ba55..c7a948f 100644 --- a/src/scraper.py +++ b/src/scraper.py @@ -2,7 +2,7 @@ import logging import time import threading -from typing import Dict, List, Optional +from typing import Dict, List, Optional, Union from src.config import Config logger = logging.getLogger(__name__) @@ -58,80 +58,110 @@ def _rate_limit(self): time.sleep(delay) self.last_request_time = time.time() + def _fetch_with_backoff(self, url: str, max_retries: int = 5) -> Optional[Union[Dict, List[Dict]]]: + """Fetch URL with exponential backoff on 429 errors + + Rate limiting is applied once before the first request. + Exponential backoff handles retry delays independently. + + Returns either a dict or list depending on the endpoint. + """ + # Apply rate limiting once before first request attempt + self._rate_limit() + + for attempt in range(max_retries): + try: + response = self.session.get(url, timeout=self.timeout) + response.raise_for_status() + return response.json() + except requests.HTTPError as e: + if e.response.status_code == 429: + if attempt < max_retries - 1: + # Exponential backoff: 5s, 10s, 20s, 40s, 80s + # Aligns with API's 10-second rate limit window + backoff_delay = (2 ** attempt) * 5 + logger.warning( + f"Rate limited (429). Attempt {attempt + 1}/{max_retries}. " + f"Backing off for {backoff_delay}s before retry..." + ) + time.sleep(backoff_delay) + else: + logger.error(f"Rate limited after {max_retries} attempts: {e}") + return None + else: + logger.error(f"HTTP error {e.response.status_code}: {e}") + return None + except requests.RequestException as e: + logger.error(f"Request failed: {e}") + return None + return None + def get_war_status(self) -> Optional[Dict]: """Fetch current war status""" - try: - self._rate_limit() - response = self.session.get(f"{self.base_url}/war", timeout=self.timeout) - response.raise_for_status() - return response.json() - except requests.RequestException as e: - logger.error(f"Failed to fetch war status: {e}") + result = self._fetch_with_backoff(f"{self.base_url}/war") + if result is None: return None + if isinstance(result, dict): + return result + logger.warning(f"Expected dict from war status endpoint, got {type(result).__name__}") + return None def get_campaign_info(self) -> Optional[List[Dict]]: """Fetch active campaigns information""" - try: - self._rate_limit() - response = self.session.get(f"{self.base_url}/campaigns", timeout=self.timeout) - response.raise_for_status() - return response.json() - except requests.RequestException as e: - logger.error(f"Failed to fetch campaign info: {e}") + result = self._fetch_with_backoff(f"{self.base_url}/campaigns") + if result is None: return None + if not isinstance(result, list): + logger.warning(f"Expected list from campaigns endpoint, got {type(result).__name__}") + return None + return result def get_assignments(self) -> Optional[List[Dict]]: """Fetch current assignments (Major Orders)""" - try: - self._rate_limit() - response = self.session.get(f"{self.base_url}/assignments", timeout=self.timeout) - response.raise_for_status() - return response.json() - except requests.RequestException as e: - logger.error(f"Failed to fetch assignments: {e}") + result = self._fetch_with_backoff(f"{self.base_url}/assignments") + if result is None: return None + if not isinstance(result, list): + logger.warning(f"Expected list from assignments endpoint, got {type(result).__name__}") + return None + return result def get_dispatches(self) -> Optional[List[Dict]]: """Fetch news dispatches and announcements""" - try: - self._rate_limit() - response = self.session.get(f"{self.base_url}/dispatches", timeout=self.timeout) - response.raise_for_status() - return response.json() - except requests.RequestException as e: - logger.error(f"Failed to fetch dispatches: {e}") + result = self._fetch_with_backoff(f"{self.base_url}/dispatches") + if result is None: return None + if not isinstance(result, list): + logger.warning(f"Expected list from dispatches endpoint, got {type(result).__name__}") + return None + return result def get_planets(self) -> Optional[List[Dict]]: """Fetch all planets information""" - try: - self._rate_limit() - response = self.session.get(f"{self.base_url}/planets", timeout=self.timeout) - response.raise_for_status() - return response.json() - except requests.RequestException as e: - logger.error(f"Failed to fetch planets: {e}") + result = self._fetch_with_backoff(f"{self.base_url}/planets") + if result is None: return None + if not isinstance(result, list): + logger.warning(f"Expected list from planets endpoint, got {type(result).__name__}") + return None + return result def get_planet_status(self, planet_index: int) -> Optional[Dict]: """Fetch status of a specific planet""" - try: - self._rate_limit() - response = self.session.get( - f"{self.base_url}/planets/{planet_index}", timeout=self.timeout - ) - response.raise_for_status() - return response.json() - except requests.RequestException as e: - logger.error(f"Failed to fetch planet status for {planet_index}: {e}") + result = self._fetch_with_backoff(f"{self.base_url}/planets/{planet_index}") + if result is None: return None + if isinstance(result, dict): + return result + logger.warning(f"Expected dict from planet status endpoint, got {type(result).__name__}") + return None def get_statistics(self) -> Optional[Dict]: """Fetch global game statistics (part of war status)""" try: # Statistics are included in war status, return the statistics subset war_data = self.get_war_status() - if war_data and "statistics" in war_data: + if war_data and isinstance(war_data, dict) and "statistics" in war_data: return war_data["statistics"] return None except requests.RequestException as e: @@ -140,14 +170,13 @@ def get_statistics(self) -> Optional[Dict]: def get_planet_events(self) -> Optional[List[Dict]]: """Fetch planet events""" - try: - self._rate_limit() - response = self.session.get(f"{self.base_url}/planet-events", timeout=self.timeout) - response.raise_for_status() - return response.json() - except requests.RequestException as e: - logger.error(f"Failed to fetch planet events: {e}") + result = self._fetch_with_backoff(f"{self.base_url}/planet-events") + if result is None: + return None + if not isinstance(result, list): + logger.warning(f"Expected list from planet-events endpoint, got {type(result).__name__}") return None + return result def get_factions(self) -> Optional[List[Dict]]: """Fetch all factions (from war data)"""