""" A simplified API for interacting with the EVE Online ESI API. """ 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 import requests from .refresh_token import get_access_token_from_refresh_token # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Constants ESI_BASE_URL = "https://esi.evetech.net/latest" DATASOURCE = "tranquility" USER_AGENT_HEADER = {'User-Agent': 'Haul (Hyperspace Asset Unloading and Loading)'} CACHE_DIR = Path(__file__).resolve().parent.parent / 'cache' AUTH_FILE = CACHE_DIR / 'auth.json' ERROR_LIMIT_THRESHOLD = 50 # ESI Error limit threshold REQUEST_RATE_LIMIT = 0.05 # Time in seconds between requests to avoid rate limiting @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. Returns: The location ID of the character (station ID or solar system ID). """ 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.") url = f"{ESI_BASE_URL}/characters/{character_id}/location/" headers = { "Accept": "application/json", "Cache-Control": "no-cache", "Authorization": f"Bearer {access_token()}", **USER_AGENT_HEADER, } params = { "datasource": DATASOURCE, } 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 == 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.") 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.") logger.info(f"Character location ID: {location_id}") return location_id 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 result.to_dict() url = f"{ESI_BASE_URL}/ui/autopilot/waypoint/" headers = { "Accept": "application/json", "Cache-Control": "no-cache", "Authorization": f"Bearer {access_token()}", **USER_AGENT_HEADER, } 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 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() async def open_market_window(type_id: int) -> None: """ Opens the market window for a specific item type in the game client. Args: type_id: The type ID of the item. """ url = f"{ESI_BASE_URL}/ui/openwindow/marketdetails/" headers = { "Accept": "application/json", "Cache-Control": "no-cache", "Authorization": f"Bearer {access_token()}", **USER_AGENT_HEADER, } params = { "type_id": type_id, "datasource": DATASOURCE, } 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}") async def get_wallet_transactions(from_id: int | None = None) -> list[WalletTransaction]: """ Retrieves recent wallet transactions for the configured character. Args: from_id: Optional transaction ID cursor. Returns: A list of normalized wallet transactions. """ 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.") 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"], ) ) 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 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 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]] = [] 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] # Add region ID to each order for order in results: order['region_id'] = region_id logger.info(f"Retrieved {len(results)} orders for region {region_id}.") return results, expires def access_token(force_refresh: bool = False) -> str: """ Retrieves an active access token, refreshing it if necessary. Args: force_refresh: If True, forces a refresh of the access token. Returns: The active access token as a string. """ try: with AUTH_FILE.open("r", encoding="utf-8") as f: auth: dict[str, Any] = json.load(f) except FileNotFoundError: logger.error(f"Authentication file not found at {AUTH_FILE}.") raise FileNotFoundError("Authentication file is missing.") expires_at = auth.get("expires_at", 0) current_time = datetime.datetime.utcnow().timestamp() if not force_refresh and expires_at > current_time: return auth.get("access_token", "") logger.info("Access token expired or force refresh requested, refreshing token.") new_auth = get_access_token_from_refresh_token(auth["refresh_token"]) # Update expiry time (assuming token is valid for 20 minutes) new_auth["expires_at"] = current_time + 1200 with AUTH_FILE.open("w", encoding="utf-8") as f: json.dump(new_auth, f) logger.info("Access token refreshed successfully.") return new_auth["access_token"]