diff options
Diffstat (limited to 'backend/src/esi/api.py')
| -rw-r--r-- | backend/src/esi/api.py | 324 |
1 files changed, 243 insertions, 81 deletions
diff --git a/backend/src/esi/api.py b/backend/src/esi/api.py index 862b4bf..d8bed26 100644 --- a/backend/src/esi/api.py +++ b/backend/src/esi/api.py @@ -6,9 +6,12 @@ import datetime import json import logging import os +from dataclasses import asdict, dataclass, field import queue import threading import time +import asyncio +import aiohttp from pathlib import Path from typing import Any @@ -30,7 +33,35 @@ ERROR_LIMIT_THRESHOLD = 50 # ESI Error limit threshold REQUEST_RATE_LIMIT = 0.05 # Time in seconds between requests to avoid rate limiting -def get_location() -> int: +@dataclass +class WaypointSubmissionResult: + requested_count: int + success_count: int = 0 + applied_waypoints: list[int] = field(default_factory=list) + failed_waypoints: list[int] = field(default_factory=list) + errors: list[str] = field(default_factory=list) + + def to_dict(self) -> dict[str, Any]: + data = asdict(self) + data["success"] = self.requested_count > 0 and self.success_count == self.requested_count + return data + + +@dataclass(frozen=True) +class WalletTransaction: + transaction_id: int + date: datetime.datetime + location_id: int + type_id: int + unit_price: float + quantity: int + client_id: int + is_buy: bool + is_personal: bool + journal_ref_id: int + + +async def get_location() -> int: """ Gets the current location of the character. @@ -55,42 +86,48 @@ def get_location() -> int: "datasource": DATASOURCE, } - response = requests.get(url, params=params, headers=headers) - logger.debug(f"Request URL: {response.url}") - logger.debug(f"Response Status Code: {response.status_code}") + async with aiohttp.ClientSession() as session: + async with session.get(url, params=params, headers=headers) as response: + logger.debug(f"Request URL: {response.url}") + logger.debug(f"Response Status Code: {response.status}") - if response.status_code == 401: - logger.info("Access token expired, refreshing token.") - access_token(force_refresh=True) - headers["Authorization"] = f"Bearer {access_token()}" - response = requests.get(url, params=params, headers=headers) + if response.status == 401: + logger.info("Access token expired, refreshing token.") + # Using run_in_executor for synchronous access_token + loop = asyncio.get_running_loop() + await loop.run_in_executor(None, access_token, True) + headers["Authorization"] = f"Bearer {access_token()}" + async with session.get(url, params=params, headers=headers) as retry_response: + response = retry_response - if int(response.headers.get("X-Esi-Error-Limit-Remain", 100)) < ERROR_LIMIT_THRESHOLD: - logger.warning("Approaching ESI Error Limit.") - raise Exception("ESI Error Limit approaching threshold.") + if int(response.headers.get("X-Esi-Error-Limit-Remain", 100)) < ERROR_LIMIT_THRESHOLD: + logger.warning("Approaching ESI Error Limit.") + raise Exception("ESI Error Limit approaching threshold.") - response.raise_for_status() - data: dict[str, Any] = response.json() + response.raise_for_status() + data = await response.json() - location_id = data.get("station_id") or data.get("solar_system_id") - if not location_id: - logger.error("Location ID not found in response.") - raise ValueError("Unable to determine character's location.") + location_id = data.get("station_id") or data.get("solar_system_id") + if not location_id: + logger.error("Location ID not found in response.") + raise ValueError("Unable to determine character's location.") - logger.info(f"Character location ID: {location_id}") - return location_id + logger.info(f"Character location ID: {location_id}") + return location_id -def set_waypoints(waypoints: list[int]) -> None: +async def set_waypoints(waypoints: list[int]) -> dict[str, Any]: """ Sets the waypoints for the character's autopilot. Args: waypoints: A list of waypoint IDs to set. """ + result = WaypointSubmissionResult(requested_count=len(waypoints)) + if not waypoints: logger.warning("No waypoints provided.") - return + return result.to_dict() url = f"{ESI_BASE_URL}/ui/autopilot/waypoint/" @@ -101,25 +138,58 @@ def set_waypoints(waypoints: list[int]) -> None: **USER_AGENT_HEADER, } - for index, waypoint in enumerate(waypoints): - params = { - "add_to_beginning": False, - "clear_other_waypoints": index == 0, - "destination_id": waypoint, - "datasource": DATASOURCE, - } + async with aiohttp.ClientSession() as session: + for index, waypoint in enumerate(waypoints): + params = { + "add_to_beginning": "false", + "clear_other_waypoints": "true" if index == 0 else "false", + "destination_id": waypoint, + "datasource": DATASOURCE, + } + waypoint_headers = dict(headers) + waypoint_error = None - try: - response = requests.post(url, params=params, headers=headers) - response.raise_for_status() - logger.info(f"Waypoint {waypoint} set successfully.") - time.sleep(1) # Rate limiting - except requests.HTTPError as e: - logger.error(f"Failed to set waypoint {waypoint}: {e}") - continue + for attempt in range(2): + try: + async with session.post(url, params=params, headers=waypoint_headers) as response: + if response.status == 401 and attempt == 0: + logger.info("Access token expired while setting waypoint %s, refreshing token.", waypoint) + loop = asyncio.get_running_loop() + refreshed_token = await loop.run_in_executor(None, access_token, True) + waypoint_headers["Authorization"] = f"Bearer {refreshed_token}" + continue + + if response.status == 401: + waypoint_error = f"Waypoint {waypoint} unauthorized after token refresh" + logger.error(waypoint_error) + break + + response.raise_for_status() + logger.info(f"Waypoint {waypoint} set successfully.") + result.success_count += 1 + result.applied_waypoints.append(waypoint) + await asyncio.sleep(1) # Rate limiting + waypoint_error = None + break + except (aiohttp.ClientError, TypeError, ValueError) as error: + waypoint_error = f"Failed to set waypoint {waypoint}: {error}" + logger.error(waypoint_error) + break + except Exception as error: + waypoint_error = f"Unexpected error while setting waypoint {waypoint}: {error}" + logger.exception(waypoint_error) + break + else: + waypoint_error = f"Failed to set waypoint {waypoint}: retries exhausted" + + if waypoint_error is not None: + result.failed_waypoints.append(waypoint) + result.errors.append(waypoint_error) + return result.to_dict() -def open_market_window(type_id: int) -> None: + +async def open_market_window(type_id: int) -> None: """ Opens the market window for a specific item type in the game client. @@ -140,62 +210,154 @@ def open_market_window(type_id: int) -> None: "datasource": DATASOURCE, } - try: - response = requests.post(url, params=params, headers=headers) - response.raise_for_status() - logger.info(f"Market window opened for type ID {type_id}.") - except requests.HTTPError as e: - logger.error(f"Failed to open market window for type ID {type_id}: {e}") + async with aiohttp.ClientSession() as session: + try: + async with session.post(url, params=params, headers=headers) as response: + response.raise_for_status() + logger.info(f"Market window opened for type ID {type_id}.") + except aiohttp.ClientError as e: + logger.error(f"Failed to open market window for type ID {type_id}: {e}") -def get_region_orders(region_id: int) -> tuple[list[dict[str, Any]], str]: +async def get_wallet_transactions(from_id: int | None = None) -> list[WalletTransaction]: """ - Retrieves all market orders for a given region. + Retrieves recent wallet transactions for the configured character. Args: - region_id: The ID of the region. + from_id: Optional transaction ID cursor. Returns: - A tuple containing the list of orders and the expiry time. + A list of normalized wallet transactions. """ - def request_orders(region: int, page: int, q: queue.Queue) -> None: - url = f"{ESI_BASE_URL}/markets/{region}/orders/" - params = { - 'datasource': DATASOURCE, - 'order_type': 'all', - 'page': page, - } - headers = USER_AGENT_HEADER + character_id = os.getenv('character_id') + if not character_id: + logger.error("Character ID not found in environment variables.") + raise EnvironmentError("Character ID not set in environment variables.") - try: - response = requests.get(url, params=params, headers=headers) - response.raise_for_status() - logger.debug(f"Retrieved page {page} for region {region}.") - q.put(response) - except requests.HTTPError as e: - logger.error(f"Error fetching page {page} for region {region}: {e}") + url = f"{ESI_BASE_URL}/characters/{character_id}/wallet/transactions/" + headers = { + "Accept": "application/json", + "Cache-Control": "no-cache", + "Authorization": f"Bearer {access_token()}", + **USER_AGENT_HEADER, + } + params = { + "datasource": DATASOURCE, + } + if from_id is not None: + params["from_id"] = from_id + + async with aiohttp.ClientSession() as session: + for attempt in range(2): + async with session.get(url, params=params, headers=headers) as response: + if response.status == 401 and attempt == 0: + logger.info("Access token expired while loading wallet transactions, refreshing token.") + loop = asyncio.get_running_loop() + refreshed_token = await loop.run_in_executor(None, access_token, True) + headers["Authorization"] = f"Bearer {refreshed_token}" + continue + + if response.status == 403: + error_body = await response.text() + logger.error("Wallet transaction access denied: %s", error_body) + raise PermissionError( + "Wallet transaction access denied. Re-run scripts/setup_auth.py to grant esi-wallet.read_character_wallet.v1." + ) + + if int(response.headers.get("X-Esi-Error-Limit-Remain", 100)) < ERROR_LIMIT_THRESHOLD: + logger.warning("Approaching ESI Error Limit while loading wallet transactions.") + raise Exception("ESI Error Limit approaching threshold.") + + response.raise_for_status() + payload = await response.json() + transactions: list[WalletTransaction] = [] + for entry in payload: + issued_at = datetime.datetime.fromisoformat(entry["date"].replace("Z", "+00:00")) + transactions.append( + WalletTransaction( + transaction_id=entry["transaction_id"], + date=issued_at, + location_id=entry["location_id"], + type_id=entry["type_id"], + unit_price=entry["unit_price"], + quantity=entry["quantity"], + client_id=entry["client_id"], + is_buy=entry["is_buy"], + is_personal=entry["is_personal"], + journal_ref_id=entry["journal_ref_id"], + ) + ) - order_queue = queue.Queue() - request_orders(region_id, 1, order_queue) # Get the first page - first_response = order_queue.get() - pages = int(first_response.headers.get('X-Pages', '1')) - expires = first_response.headers.get('Expires', '') + logger.info("Retrieved %d wallet transactions.", len(transactions)) + return transactions + + return [] + + +async def get_region_orders(region_id: int) -> tuple[list[dict[str, Any]], str]: + """ + Retrieves all market orders for a given region using asyncio and aiohttp. + + Args: + region_id: The ID of the region. + + Returns: + A tuple containing the list of orders and the expiry time. + """ + url_template = f"{ESI_BASE_URL}/markets/{{region}}/orders/" + headers = USER_AGENT_HEADER - if pages > 1: - threads = [] - for page in range(2, pages + 1): - thread = threading.Thread(target=request_orders, args=(region_id, page, order_queue)) - threads.append(thread) - thread.start() - time.sleep(REQUEST_RATE_LIMIT) # Rate limiting + async def fetch_page(session: aiohttp.ClientSession, region: int, page: int, sem: asyncio.Semaphore) -> tuple[list[dict[str, Any]], str, int] | None: + async with sem: + params = { + 'datasource': DATASOURCE, + 'order_type': 'all', + 'page': page, + } + for attempt in range(3): + try: + async with session.get(url_template.format(region=region), params=params, headers=headers) as response: + if response.status == 429: + logger.warning(f"Rate limited (429) on page {page} for region {region}. Retrying...") + await asyncio.sleep(2 * (attempt + 1)) + continue - for thread in threads: - thread.join() + response.raise_for_status() + logger.debug(f"Retrieved page {page} for region {region}.") + + pages = int(response.headers.get('X-Pages', '1')) + expires = response.headers.get('Expires', '') + data = await response.json() + return data, expires, pages + except aiohttp.ClientError as e: + logger.error(f"Error fetching page {page} for region {region} (attempt {attempt + 1}): {e}") + if attempt == 2: + return None + await asyncio.sleep(1) + return None - results: list[dict[str, Any]] = first_response.json() - while not order_queue.empty(): - response = order_queue.get() - results.extend(response.json()) + results: list[dict[str, Any]] = [] + expires = '' + sem = asyncio.Semaphore(15) # Bound concurrency to 15 active requests + + async with aiohttp.ClientSession() as session: + # Fetch page 1 to get the total number of pages + first_page_result = await fetch_page(session, region_id, 1, sem) + if not first_page_result: + return [], '' + + data, expires, total_pages = first_page_result + results.extend(data) + + if total_pages > 1: + tasks = [ + fetch_page(session, region_id, page, sem) + for page in range(2, total_pages + 1) + ] + pages_data = await asyncio.gather(*tasks) + for page_result in pages_data: + if page_result: + results.extend(page_result[0]) # Filter out orders with zero remaining volume results = [order for order in results if order.get('volume_remain', 0) > 0] |