EU-Utility/core/nexus_api.py

741 lines
23 KiB
Python

"""
EU-Utility - Entropia Nexus API Client
API client for https://api.entropianexus.com
Provides access to game data: items, mobs, market info.
Part of core - plugins access via PluginAPI.
"""
import time
import json
from typing import Dict, Any, List, Optional, Union
from dataclasses import dataclass
from enum import Enum
from datetime import datetime, timedelta
class EntityType(Enum):
"""Types of entities that can be searched."""
# Core types
ITEM = "items"
MOB = "mobs"
ALL = "all"
# Equipment
WEAPON = "weapons"
ARMOR = "armors"
ENHANCER = "enhancers"
# Tools
MEDICAL_TOOL = "medicaltools"
FINDER = "finders"
EXCAVATOR = "excavators"
REFINER = "refiners"
# Crafting & Materials
BLUEPRINT = "blueprints"
MATERIAL = "materials"
# Creatures
PET = "pets"
# Locations
LOCATION = "locations"
TELEPORTER = "teleporters"
SHOP = "shops"
VENDOR = "vendors"
PLANET = "planets"
AREA = "areas"
# Other
SKILL = "skills"
VEHICLE = "vehicles"
DECORATION = "decorations"
FURNITURE = "furniture"
STORAGE_CONTAINER = "storagecontainers"
STRONGBOX = "strongboxes"
@classmethod
def get_all_types(cls) -> List[str]:
"""Get list of all entity type values."""
return [e.value for e in cls if e != cls.ALL]
@classmethod
def from_string(cls, type_str: str) -> Optional["EntityType"]:
"""Get EntityType from string value."""
type_str = type_str.lower()
for e in cls:
if e.value.lower() == type_str:
return e
return None
class NexusAPIError(Exception):
"""Custom exception for Nexus API errors."""
pass
class RateLimitError(NexusAPIError):
"""Raised when rate limit is exceeded."""
pass
@dataclass
class SearchResult:
"""Result from a search operation."""
id: str
name: str
type: str
category: Optional[str] = None
icon_url: Optional[str] = None
data: Dict[str, Any] = None
def __post_init__(self):
if self.data is None:
self.data = {}
@dataclass
class ItemDetails:
"""Detailed item information."""
id: str
name: str
description: Optional[str] = None
category: Optional[str] = None
weight: Optional[float] = None
tt_value: Optional[float] = None
decay: Optional[float] = None
ammo_consumption: Optional[int] = None
damage: Optional[float] = None
range: Optional[float] = None
accuracy: Optional[float] = None
durability: Optional[int] = None
requirements: Dict[str, Any] = None
materials: List[Dict[str, Any]] = None
raw_data: Dict[str, Any] = None
def __post_init__(self):
if self.requirements is None:
self.requirements = {}
if self.materials is None:
self.materials = []
if self.raw_data is None:
self.raw_data = {}
@dataclass
class MarketData:
"""Market data for an item."""
item_id: str
item_name: str
current_markup: Optional[float] = None
avg_markup_7d: Optional[float] = None
avg_markup_30d: Optional[float] = None
volume_24h: Optional[int] = None
volume_7d: Optional[int] = None
buy_orders: List[Dict[str, Any]] = None
sell_orders: List[Dict[str, Any]] = None
last_updated: Optional[datetime] = None
raw_data: Dict[str, Any] = None
def __post_init__(self):
if self.buy_orders is None:
self.buy_orders = []
if self.sell_orders is None:
self.sell_orders = []
if self.raw_data is None:
self.raw_data = {}
class NexusAPI:
"""
Singleton client for Entropia Nexus API.
Features:
- Auto-retry on transient failures
- Rate limiting (max requests per second)
- Caching for frequently accessed data
- Proper error handling
Usage:
api = get_nexus_api()
results = api.search_items("ArMatrix")
details = api.get_item_details("item_id")
"""
_instance = None
# API Configuration
BASE_URL = "https://api.entropianexus.com"
API_VERSION = "v1"
# Rate limiting
MAX_REQUESTS_PER_SECOND = 5
MIN_REQUEST_INTERVAL = 1.0 / MAX_REQUESTS_PER_SECOND
# Retry configuration
MAX_RETRIES = 3
RETRY_DELAY = 1.0 # seconds
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance._initialized = False
return cls._instance
def __init__(self):
if self._initialized:
return
self._last_request_time = 0
self._request_count = 0
self._cache: Dict[str, Any] = {}
self._cache_ttl = 300 # 5 minutes default cache
self._cache_timestamps: Dict[str, float] = {}
self._session = None
self._initialized = True
self._available = True
print("[NexusAPI] Initialized")
def _get_session(self):
"""Get or create HTTP session."""
if self._session is None:
try:
import requests
self._session = requests.Session()
self._session.headers.update({
'User-Agent': 'EU-Utility/1.0 (Entropia Universe Utility Tool)',
'Accept': 'application/json'
})
except ImportError:
raise NexusAPIError("requests library not installed. Run: pip install requests")
return self._session
def _rate_limit(self):
"""Enforce rate limiting between requests."""
current_time = time.time()
time_since_last = current_time - self._last_request_time
if time_since_last < self.MIN_REQUEST_INTERVAL:
sleep_time = self.MIN_REQUEST_INTERVAL - time_since_last
time.sleep(sleep_time)
self._last_request_time = time.time()
self._request_count += 1
def _make_request(
self,
endpoint: str,
params: Dict[str, Any] = None,
use_cache: bool = True
) -> Dict[str, Any]:
"""
Make HTTP request with retry logic and rate limiting.
Args:
endpoint: API endpoint path
params: Query parameters
use_cache: Whether to use caching
Returns:
JSON response as dict
Raises:
NexusAPIError: On API errors
RateLimitError: On rate limit exceeded
"""
# Check cache
cache_key = f"{endpoint}:{json.dumps(params, sort_keys=True) if params else ''}"
if use_cache and self._is_cache_valid(cache_key):
return self._cache[cache_key]
url = f"{self.BASE_URL}/{self.API_VERSION}/{endpoint}"
last_error = None
for attempt in range(self.MAX_RETRIES):
try:
# Rate limit
self._rate_limit()
# Make request
session = self._get_session()
response = session.get(url, params=params, timeout=30)
# Handle rate limiting
if response.status_code == 429:
retry_after = int(response.headers.get('Retry-After', 60))
if attempt < self.MAX_RETRIES - 1:
print(f"[NexusAPI] Rate limited. Waiting {retry_after}s...")
time.sleep(retry_after)
continue
else:
raise RateLimitError(f"Rate limit exceeded. Retry after {retry_after}s")
# Handle other HTTP errors
if response.status_code >= 500:
if attempt < self.MAX_RETRIES - 1:
wait_time = self.RETRY_DELAY * (2 ** attempt) # Exponential backoff
print(f"[NexusAPI] Server error {response.status_code}. Retrying in {wait_time}s...")
time.sleep(wait_time)
continue
response.raise_for_status()
# Parse response
data = response.json()
# Cache successful response
if use_cache:
self._cache[cache_key] = data
self._cache_timestamps[cache_key] = time.time()
return data
except RateLimitError:
raise
except Exception as e:
last_error = e
if attempt < self.MAX_RETRIES - 1:
wait_time = self.RETRY_DELAY * (2 ** attempt)
print(f"[NexusAPI] Request failed: {e}. Retrying in {wait_time}s...")
time.sleep(wait_time)
else:
break
# All retries exhausted
error_msg = f"Request failed after {self.MAX_RETRIES} attempts: {last_error}"
print(f"[NexusAPI] {error_msg}")
raise NexusAPIError(error_msg)
def _is_cache_valid(self, key: str) -> bool:
"""Check if cached data is still valid."""
if key not in self._cache_timestamps:
return False
age = time.time() - self._cache_timestamps[key]
return age < self._cache_ttl
def clear_cache(self):
"""Clear all cached data."""
self._cache.clear()
self._cache_timestamps.clear()
print("[NexusAPI] Cache cleared")
def is_available(self) -> bool:
"""Check if API is available."""
return self._available
# ========== Search Methods ==========
def search_items(self, query: str, limit: int = 20) -> List[SearchResult]:
"""
Search for items by name.
Args:
query: Search term (e.g., "ArMatrix", "Omegaton")
limit: Maximum results (default 20, max 100)
Returns:
List of SearchResult objects
Example:
results = api.search_items("ArMatrix")
for r in results:
print(f"{r.name} ({r.type})")
"""
try:
params = {
'q': query,
'limit': min(limit, 100),
'type': 'item'
}
data = self._make_request('search', params)
results = []
for item in data.get('results', []):
results.append(SearchResult(
id=item.get('id', ''),
name=item.get('name', ''),
type=item.get('type', 'item'),
category=item.get('category'),
icon_url=item.get('icon_url'),
data=item
))
return results
except Exception as e:
print(f"[NexusAPI] search_items error: {e}")
return []
def search_mobs(self, query: str, limit: int = 20) -> List[SearchResult]:
"""
Search for creatures/mobs by name.
Args:
query: Search term (e.g., "Atrox", "Hispidus")
limit: Maximum results (default 20, max 100)
Returns:
List of SearchResult objects
"""
try:
params = {
'q': query,
'limit': min(limit, 100),
'type': 'mob'
}
data = self._make_request('search', params)
results = []
for item in data.get('results', []):
results.append(SearchResult(
id=item.get('id', ''),
name=item.get('name', ''),
type=item.get('type', 'mob'),
category=item.get('category'),
icon_url=item.get('icon_url'),
data=item
))
return results
except Exception as e:
print(f"[NexusAPI] search_mobs error: {e}")
return []
def search_all(self, query: str, limit: int = 20) -> List[SearchResult]:
"""
Search across all entity types (items, mobs, etc.).
Args:
query: Search term
limit: Maximum results per type (default 20)
Returns:
List of SearchResult objects
"""
try:
params = {
'q': query,
'limit': min(limit, 100)
}
data = self._make_request('search', params)
results = []
for item in data.get('results', []):
results.append(SearchResult(
id=item.get('id', ''),
name=item.get('name', ''),
type=item.get('type', 'unknown'),
category=item.get('category'),
icon_url=item.get('icon_url'),
data=item
))
return results
except Exception as e:
print(f"[NexusAPI] search_all error: {e}")
return []
def search_by_type(self, query: str, entity_type: str, limit: int = 20) -> List[SearchResult]:
"""
Search for entities of a specific type.
Args:
query: Search term
entity_type: Entity type (e.g., 'weapons', 'blueprints', 'mobs')
limit: Maximum results (default 20, max 100)
Returns:
List of SearchResult objects
Example:
results = api.search_by_type("ArMatrix", "weapons")
results = api.search_by_type("Atrox", "mobs")
"""
try:
# Normalize entity type
entity_type = entity_type.lower().replace(' ', '').replace('_', '')
# Map common aliases
type_mapping = {
'item': 'items',
'weapon': 'weapons',
'armor': 'armors',
'mob': 'mobs',
'blueprint': 'blueprints',
'location': 'locations',
'skill': 'skills',
'material': 'materials',
'enhancer': 'enhancers',
'medicaltool': 'medicaltools',
'medical_tool': 'medicaltools',
'finder': 'finders',
'excavator': 'excavators',
'refiner': 'refiners',
'vehicle': 'vehicles',
'pet': 'pets',
'decoration': 'decorations',
'furniture': 'furniture',
'storage': 'storagecontainers',
'storagecontainer': 'storagecontainers',
'storage_container': 'storagecontainers',
'strongbox': 'strongboxes',
'teleporter': 'teleporters',
'shop': 'shops',
'vendor': 'vendors',
'planet': 'planets',
'area': 'areas',
}
endpoint = type_mapping.get(entity_type, entity_type)
params = {
'q': query,
'limit': min(limit, 100),
'fuzzy': 'true'
}
data = self._make_request(f'{endpoint}', params)
results = []
items = data if isinstance(data, list) else data.get('results', [])
for item in items:
results.append(SearchResult(
id=item.get('id', item.get('Id', '')),
name=item.get('name', item.get('Name', 'Unknown')),
type=item.get('type', entity_type),
category=item.get('category', item.get('Category')),
icon_url=item.get('icon_url', item.get('IconUrl')),
data=item
))
return results
except Exception as e:
print(f"[NexusAPI] search_by_type error ({entity_type}): {e}")
return []
def get_entity_details(self, entity_id: str, entity_type: str) -> Optional[Dict[str, Any]]:
"""
Get detailed information about any entity type.
Args:
entity_id: The entity's unique identifier
entity_type: Entity type (e.g., 'mobs', 'locations', 'blueprints')
Returns:
Dict with entity details or None if not found
Example:
mob = api.get_entity_details("atrox", "mobs")
location = api.get_entity_details("fort-izzuk", "locations")
"""
try:
# Normalize entity type
entity_type = entity_type.lower().replace(' ', '').replace('_', '')
# Map common aliases
type_mapping = {
'item': 'items',
'weapon': 'weapons',
'armor': 'armors',
'mob': 'mobs',
'blueprint': 'blueprints',
'location': 'locations',
'skill': 'skills',
'material': 'materials',
'enhancer': 'enhancers',
'medicaltool': 'medicaltools',
'finder': 'finders',
'excavator': 'excavators',
'refiner': 'refiners',
'vehicle': 'vehicles',
'pet': 'pets',
'decoration': 'decorations',
'furniture': 'furniture',
'storage': 'storagecontainers',
'storagecontainer': 'storagecontainers',
'strongbox': 'strongboxes',
'teleporter': 'teleporters',
'shop': 'shops',
'vendor': 'vendors',
'planet': 'planets',
'area': 'areas',
}
endpoint = type_mapping.get(entity_type, entity_type)
data = self._make_request(f'{endpoint}/{entity_id}')
if not data or 'error' in data:
return None
return data
except Exception as e:
print(f"[NexusAPI] get_entity_details error ({entity_type}): {e}")
return None
# ========== Detail Methods ==========
def get_item_details(self, item_id: str) -> Optional[ItemDetails]:
"""
Get detailed information about a specific item.
Args:
item_id: The item's unique identifier
Returns:
ItemDetails object or None if not found
Example:
details = api.get_item_details("armatrix_lp-35")
print(f"TT Value: {details.tt_value} PED")
"""
try:
data = self._make_request(f'items/{item_id}')
if not data or 'error' in data:
return None
return ItemDetails(
id=data.get('id', item_id),
name=data.get('name', 'Unknown'),
description=data.get('description'),
category=data.get('category'),
weight=data.get('weight'),
tt_value=data.get('tt_value'),
decay=data.get('decay'),
ammo_consumption=data.get('ammo_consumption'),
damage=data.get('damage'),
range=data.get('range'),
accuracy=data.get('accuracy'),
durability=data.get('durability'),
requirements=data.get('requirements', {}),
materials=data.get('materials', []),
raw_data=data
)
except Exception as e:
print(f"[NexusAPI] get_item_details error: {e}")
return None
def get_market_data(self, item_id: str) -> Optional[MarketData]:
"""
Get market data for a specific item.
Args:
item_id: The item's unique identifier
Returns:
MarketData object or None if not found
Example:
market = api.get_market_data("armatrix_lp-35")
print(f"Current markup: {market.current_markup:.1f}%")
print(f"24h Volume: {market.volume_24h}")
"""
try:
data = self._make_request(f'items/{item_id}/market')
if not data or 'error' in data:
return None
# Parse timestamp if present
last_updated = None
if 'last_updated' in data:
try:
last_updated = datetime.fromisoformat(data['last_updated'].replace('Z', '+00:00'))
except:
pass
return MarketData(
item_id=item_id,
item_name=data.get('item_name', 'Unknown'),
current_markup=data.get('current_markup'),
avg_markup_7d=data.get('avg_markup_7d'),
avg_markup_30d=data.get('avg_markup_30d'),
volume_24h=data.get('volume_24h'),
volume_7d=data.get('volume_7d'),
buy_orders=data.get('buy_orders', []),
sell_orders=data.get('sell_orders', []),
last_updated=last_updated,
raw_data=data
)
except Exception as e:
print(f"[NexusAPI] get_market_data error: {e}")
return None
# ========== Batch Methods ==========
def get_items_batch(self, item_ids: List[str]) -> Dict[str, Optional[ItemDetails]]:
"""
Get details for multiple items efficiently.
Args:
item_ids: List of item IDs
Returns:
Dict mapping item_id to ItemDetails (or None if failed)
"""
results = {}
for item_id in item_ids:
results[item_id] = self.get_item_details(item_id)
return results
def get_market_batch(self, item_ids: List[str]) -> Dict[str, Optional[MarketData]]:
"""
Get market data for multiple items efficiently.
Args:
item_ids: List of item IDs
Returns:
Dict mapping item_id to MarketData (or None if failed)
"""
results = {}
for item_id in item_ids:
results[item_id] = self.get_market_data(item_id)
return results
# Singleton instance
_nexus_api = None
def get_nexus_api() -> NexusAPI:
"""Get the global NexusAPI instance."""
global _nexus_api
if _nexus_api is None:
_nexus_api = NexusAPI()
return _nexus_api
# Convenience functions for quick access
def search_items(query: str, limit: int = 20) -> List[SearchResult]:
"""Quick search for items."""
return get_nexus_api().search_items(query, limit)
def search_mobs(query: str, limit: int = 20) -> List[SearchResult]:
"""Quick search for mobs."""
return get_nexus_api().search_mobs(query, limit)
def search_all(query: str, limit: int = 20) -> List[SearchResult]:
"""Quick search across all types."""
return get_nexus_api().search_all(query, limit)
def get_item_details(item_id: str) -> Optional[ItemDetails]:
"""Quick get item details."""
return get_nexus_api().get_item_details(item_id)
def get_market_data(item_id: str) -> Optional[MarketData]:
"""Quick get market data."""
return get_nexus_api().get_market_data(item_id)