diff --git a/core/nexus_api.py b/core/nexus_api.py index 8b1c5ba..9826c21 100644 --- a/core/nexus_api.py +++ b/core/nexus_api.py @@ -1,13 +1,15 @@ """ -Entropia Nexus API Client - Production Implementation +Entropia Nexus API Client - Full Implementation -Uses the real Entropia Nexus API endpoints: -- https://api.entropianexus.com/weapons -- https://api.entropianexus.com/armors -- https://api.entropianexus.com/tools +Base URL: https://api.entropianexus.com +Docs: https://api.entropianexus.com/docs + +Endpoints: +- /weapons, /armors, /finders, /excavators +- /blueprints, /mobs, /materials +- /search, /items/{id} """ -import asyncio import json import logging from dataclasses import dataclass, field @@ -15,13 +17,6 @@ from decimal import Decimal from typing import Dict, List, Optional, Any from datetime import datetime -# Optional HTTP libraries -try: - import aiohttp - HAS_AIOHTTP = True -except ImportError: - HAS_AIOHTTP = False - try: import urllib.request HAS_URLLIB = True @@ -31,6 +26,10 @@ except ImportError: logger = logging.getLogger(__name__) +# ============================================================================= +# Data Classes +# ============================================================================= + @dataclass class WeaponStats: """Weapon statistics from Entropia Nexus API.""" @@ -66,30 +65,24 @@ class WeaponStats: dmg_skill_start: Optional[int] = None dmg_skill_end: Optional[int] = None ammo_name: Optional[str] = None - api_url: str = "" def __post_init__(self): - """Calculate derived statistics.""" self.total_damage = ( self.damage_stab + self.damage_cut + self.damage_impact + self.damage_penetration + self.damage_shrapnel + self.damage_burn + self.damage_cold + self.damage_acid + self.damage_electric ) - if self.decay and self.decay > 0: ammo_cost = Decimal(self.ammo_burn) if self.ammo_burn else Decimal("0") total_cost_pec = self.decay + ammo_cost if total_cost_pec > 0: self.dpp = self.total_damage / total_cost_pec - if self.uses_per_minute and self.uses_per_minute > 0 and self.decay: uses_per_hour = self.uses_per_minute * 60 - total_decay_per_hour = self.decay * uses_per_hour - self.cost_per_hour = total_decay_per_hour / 100 + self.cost_per_hour = (self.decay * uses_per_hour) / 100 @classmethod def from_api_data(cls, data: Dict[str, Any]) -> 'WeaponStats': - """Create WeaponStats from API response.""" props = data.get('Properties', {}) economy = props.get('Economy', {}) damage = props.get('Damage', {}) @@ -97,7 +90,6 @@ class WeaponStats: hit_skill = skill.get('Hit', {}) dmg_skill = skill.get('Dmg', {}) ammo = data.get('Ammo', {}) - links = data.get('Links', {}) return cls( id=data.get('Id', 0), @@ -129,12 +121,146 @@ class WeaponStats: dmg_skill_start=dmg_skill.get('LearningIntervalStart'), dmg_skill_end=dmg_skill.get('LearningIntervalEnd'), ammo_name=ammo.get('Name'), - api_url=links.get('$Url', ''), ) +@dataclass +class ArmorStats: + """Armor statistics.""" + id: int + item_id: int + name: str + weight: Decimal + durability: int + protection_stab: Decimal + protection_cut: Decimal + protection_impact: Decimal + protection_penetration: Decimal + protection_shrapnel: Decimal + protection_burn: Decimal + protection_cold: Decimal + protection_acid: Decimal + protection_electric: Decimal + total_protection: Decimal = field(default_factory=lambda: Decimal("0")) + + def __post_init__(self): + self.total_protection = ( + self.protection_stab + self.protection_cut + self.protection_impact + + self.protection_penetration + self.protection_shrapnel + + self.protection_burn + self.protection_cold + self.protection_acid + self.protection_electric + ) + + @classmethod + def from_api_data(cls, data: Dict[str, Any]) -> 'ArmorStats': + props = data.get('Properties', {}) + protection = props.get('Protection', {}) + + return cls( + id=data.get('Id', 0), + item_id=data.get('ItemId', 0), + name=data.get('Name', 'Unknown'), + weight=Decimal(str(props.get('Weight', 0))) if props.get('Weight') else Decimal('0'), + durability=props.get('Durability', 0) or 0, + protection_stab=Decimal(str(protection.get('Stab', 0))) if protection.get('Stab') else Decimal('0'), + protection_cut=Decimal(str(protection.get('Cut', 0))) if protection.get('Cut') else Decimal('0'), + protection_impact=Decimal(str(protection.get('Impact', 0))) if protection.get('Impact') else Decimal('0'), + protection_penetration=Decimal(str(protection.get('Penetration', 0))) if protection.get('Penetration') else Decimal('0'), + protection_shrapnel=Decimal(str(protection.get('Shrapnel', 0))) if protection.get('Shrapnel') else Decimal('0'), + protection_burn=Decimal(str(protection.get('Burn', 0))) if protection.get('Burn') else Decimal('0'), + protection_cold=Decimal(str(protection.get('Cold', 0))) if protection.get('Cold') else Decimal('0'), + protection_acid=Decimal(str(protection.get('Acid', 0))) if protection.get('Acid') else Decimal('0'), + protection_electric=Decimal(str(protection.get('Electric', 0))) if protection.get('Electric') else Decimal('0'), + ) + + +@dataclass +class FinderStats: + """Mining finder statistics.""" + id: int + item_id: int + name: str + type: str + category: str + depth: Decimal + radius: Decimal + decay: Decimal + uses_per_minute: int + + @classmethod + def from_api_data(cls, data: Dict[str, Any]) -> 'FinderStats': + props = data.get('Properties', {}) + economy = props.get('Economy', {}) + + return cls( + id=data.get('Id', 0), + item_id=data.get('ItemId', 0), + name=data.get('Name', 'Unknown'), + type=props.get('Type', 'Unknown'), + category=props.get('Category', 'Unknown'), + depth=Decimal(str(props.get('Depth', 0))) if props.get('Depth') else Decimal('0'), + radius=Decimal(str(props.get('Radius', 0))) if props.get('Radius') else Decimal('0'), + decay=Decimal(str(economy.get('Decay', 0))) if economy.get('Decay') else Decimal('0'), + uses_per_minute=props.get('UsesPerMinute', 0) or 0, + ) + + +@dataclass +class ExcavatorStats: + """Mining excavator statistics.""" + id: int + item_id: int + name: str + decay: Decimal + max_tt: Decimal + + @classmethod + def from_api_data(cls, data: Dict[str, Any]) -> 'ExcavatorStats': + props = data.get('Properties', {}) + economy = props.get('Economy', {}) + + return cls( + id=data.get('Id', 0), + item_id=data.get('ItemId', 0), + name=data.get('Name', 'Unknown'), + decay=Decimal(str(economy.get('Decay', 0))) if economy.get('Decay') else Decimal('0'), + max_tt=Decimal(str(economy.get('MaxTT', 0))) if economy.get('MaxTT') else Decimal('0'), + ) + + +@dataclass +class MobStats: + """Creature/Mob statistics.""" + id: int + name: str + hp: int + damage: int + agility: int + intelligence: int + psyche: int + stamina: int + strength: int + + @classmethod + def from_api_data(cls, data: Dict[str, Any]) -> 'MobStats': + return cls( + id=data.get('Id', 0), + name=data.get('Name', 'Unknown'), + hp=data.get('HP', 0) or 0, + damage=data.get('Damage', 0) or 0, + agility=data.get('Agility', 0) or 0, + intelligence=data.get('Intelligence', 0) or 0, + psyche=data.get('Psyche', 0) or 0, + stamina=data.get('Stamina', 0) or 0, + strength=data.get('Strength', 0) or 0, + ) + + +# ============================================================================= +# API Client +# ============================================================================= + class EntropiaNexusAPI: - """Client for Entropia Nexus API.""" + """Full client for Entropia Nexus API.""" BASE_URL = "https://api.entropianexus.com" @@ -142,18 +268,10 @@ class EntropiaNexusAPI: self._cache: Dict[str, Any] = {} self._cache_timestamps: Dict[str, float] = {} self._cache_ttl = cache_ttl - self._session = None - self._weapons_cache: Optional[List[WeaponStats]] = None logger.info("EntropiaNexusAPI initialized") - async def __aenter__(self): - return self - - async def __aexit__(self, exc_type, exc_val, exc_tb): - return False - - async def _make_request(self, endpoint: str) -> Optional[List[Dict]]: - """Make API request.""" + def _make_request(self, endpoint: str) -> Optional[List[Dict]]: + """Make API request with caching.""" cache_key = endpoint if cache_key in self._cache: @@ -161,47 +279,150 @@ class EntropiaNexusAPI: if (datetime.now().timestamp() - timestamp) < self._cache_ttl: return self._cache[cache_key] + if not HAS_URLLIB: + return None + url = f"{self.BASE_URL}/{endpoint}" try: - if HAS_URLLIB: - req = urllib.request.Request( - url, - headers={'Accept': 'application/json', 'User-Agent': 'Lemontropia-Suite/0.2.0'} - ) - with urllib.request.urlopen(req, timeout=30) as response: - data = json.loads(response.read().decode('utf-8')) - self._cache[cache_key] = data - self._cache_timestamps[cache_key] = datetime.now().timestamp() - return data + req = urllib.request.Request( + url, + headers={'Accept': 'application/json', 'User-Agent': 'Lemontropia-Suite/0.2.0'} + ) + with urllib.request.urlopen(req, timeout=30) as response: + data = json.loads(response.read().decode('utf-8')) + self._cache[cache_key] = data + self._cache_timestamps[cache_key] = datetime.now().timestamp() + logger.info(f"Fetched {len(data)} items from {endpoint}") + return data except Exception as e: - logger.error(f"API request failed: {e}") + logger.error(f"API request failed for {endpoint}: {e}") return None - async def get_all_weapons(self, force_refresh: bool = False) -> List[WeaponStats]: - """Get all weapons.""" - if not force_refresh and self._weapons_cache: - return self._weapons_cache + def _get_item(self, endpoint: str, item_id: int) -> Optional[Dict]: + """Get single item by ID.""" + cache_key = f"{endpoint}/{item_id}" - data = await self._make_request('weapons') + if cache_key in self._cache: + timestamp = self._cache_timestamps.get(cache_key, 0) + if (datetime.now().timestamp() - timestamp) < self._cache_ttl: + return self._cache[cache_key] + + if not HAS_URLLIB: + return None + + url = f"{self.BASE_URL}/{endpoint}/{item_id}" + + try: + req = urllib.request.Request( + url, + headers={'Accept': 'application/json', 'User-Agent': 'Lemontropia-Suite/0.2.0'} + ) + with urllib.request.urlopen(req, timeout=30) as response: + data = json.loads(response.read().decode('utf-8')) + self._cache[cache_key] = data + self._cache_timestamps[cache_key] = datetime.now().timestamp() + return data + except Exception as e: + logger.error(f"API request failed for {url}: {e}") + return None + + # ======================================================================== + # Weapons + # ======================================================================== + + def get_all_weapons(self) -> List[WeaponStats]: + """Get all weapons.""" + data = self._make_request('weapons') if data: - self._weapons_cache = [WeaponStats.from_api_data(w) for w in data] - return self._weapons_cache + return [WeaponStats.from_api_data(w) for w in data] return [] - async def search_weapons(self, query: str) -> List[WeaponStats]: + def get_weapon(self, weapon_id: int) -> Optional[WeaponStats]: + """Get specific weapon by ID.""" + data = self._get_item('weapons', weapon_id) + if data: + return WeaponStats.from_api_data(data) + return None + + def search_weapons(self, query: str) -> List[WeaponStats]: """Search weapons by name.""" - weapons = await self.get_all_weapons() + weapons = self.get_all_weapons() query_lower = query.lower() return [w for w in weapons if query_lower in w.name.lower()] - async def get_weapon_by_id(self, weapon_id: int) -> Optional[WeaponStats]: - """Get weapon by ID.""" - weapons = await self.get_all_weapons() - for w in weapons: - if w.id == weapon_id or w.item_id == weapon_id: - return w + # ======================================================================== + # Armors + # ======================================================================== + + def get_all_armors(self) -> List[ArmorStats]: + """Get all armors.""" + data = self._make_request('armors') + if data: + return [ArmorStats.from_api_data(a) for a in data] + return [] + + def get_armor(self, armor_id: int) -> Optional[ArmorStats]: + """Get specific armor by ID.""" + data = self._get_item('armors', armor_id) + if data: + return ArmorStats.from_api_data(data) return None + + # ======================================================================== + # Mining Tools + # ======================================================================== + + def get_all_finders(self) -> List[FinderStats]: + """Get all mining finders.""" + data = self._make_request('finders') + if data: + return [FinderStats.from_api_data(f) for f in data] + return [] + + def get_all_excavators(self) -> List[ExcavatorStats]: + """Get all excavators.""" + data = self._make_request('excavators') + if data: + return [ExcavatorStats.from_api_data(e) for e in data] + return [] + + # ======================================================================== + # Mobs + # ======================================================================== + + def get_all_mobs(self) -> List[MobStats]: + """Get all creatures/mobs.""" + data = self._make_request('mobs') + if data: + return [MobStats.from_api_data(m) for m in data] + return [] + + def get_mob(self, mob_id: int) -> Optional[MobStats]: + """Get specific mob by ID.""" + data = self._get_item('mobs', mob_id) + if data: + return MobStats.from_api_data(data) + return None + + def search_mobs(self, query: str) -> List[MobStats]: + """Search mobs by name.""" + mobs = self.get_all_mobs() + query_lower = query.lower() + return [m for m in mobs if query_lower in m.name.lower()] + + # ======================================================================== + # Cache Management + # ======================================================================== + + def clear_cache(self): + """Clear all cached data.""" + self._cache.clear() + self._cache_timestamps.clear() + logger.info("Cache cleared") -__all__ = ['WeaponStats', 'EntropiaNexusAPI'] +__all__ = [ + 'WeaponStats', 'ArmorStats', 'FinderStats', 'ExcavatorStats', 'MobStats', + 'EntropiaNexusAPI' +]