624 lines
24 KiB
Python
624 lines
24 KiB
Python
"""
|
|
API client for the bot to communicate with the standalone API.
|
|
All database operations now go through the API.
|
|
"""
|
|
import httpx
|
|
import os
|
|
from typing import Optional, Dict, Any, List
|
|
|
|
|
|
class APIClient:
|
|
"""Client for bot-to-API communication"""
|
|
|
|
def __init__(self):
|
|
self.api_url = os.getenv("API_BASE_URL", os.getenv("API_URL", "http://echoes_of_the_ashes_api:8000"))
|
|
self.internal_key = os.getenv("API_INTERNAL_KEY", "change-this-internal-key")
|
|
self.client = httpx.AsyncClient(timeout=30.0)
|
|
self.headers = {
|
|
"Authorization": f"Bearer {self.internal_key}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
async def close(self):
|
|
"""Close the HTTP client"""
|
|
await self.client.aclose()
|
|
|
|
# Player operations
|
|
async def get_player(self, telegram_id: int) -> Optional[Dict[str, Any]]:
|
|
"""Get player by Telegram ID"""
|
|
try:
|
|
response = await self.client.get(
|
|
f"{self.api_url}/api/internal/player/{telegram_id}",
|
|
headers=self.headers
|
|
)
|
|
if response.status_code == 404:
|
|
return None
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error getting player: {e}")
|
|
return None
|
|
|
|
async def get_player_by_id(self, player_id: int) -> Optional[Dict[str, Any]]:
|
|
"""Get player by unique database ID"""
|
|
try:
|
|
response = await self.client.get(
|
|
f"{self.api_url}/api/internal/player/by_id/{player_id}",
|
|
headers=self.headers
|
|
)
|
|
if response.status_code == 404:
|
|
return None
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error getting player by id: {e}")
|
|
return None
|
|
|
|
async def create_player(self, telegram_id: int, name: str = "Survivor") -> Optional[Dict[str, Any]]:
|
|
"""Create a new player"""
|
|
try:
|
|
response = await self.client.post(
|
|
f"{self.api_url}/api/internal/player",
|
|
headers=self.headers,
|
|
params={"telegram_id": telegram_id, "name": name}
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error creating player: {e}")
|
|
return None
|
|
|
|
# Movement operations
|
|
async def move_player(self, player_id: int, direction: str) -> Dict[str, Any]:
|
|
"""Move player in a direction"""
|
|
try:
|
|
response = await self.client.post(
|
|
f"{self.api_url}/api/internal/player/{player_id}/move",
|
|
headers=self.headers,
|
|
params={"direction": direction}
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error moving player: {e}")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
# Inspection operations
|
|
async def inspect_area(self, player_id: int) -> Dict[str, Any]:
|
|
"""Inspect current area"""
|
|
try:
|
|
response = await self.client.get(
|
|
f"{self.api_url}/api/internal/player/{player_id}/inspect",
|
|
headers=self.headers
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error inspecting area: {e}")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
# Interaction operations
|
|
async def interact(self, player_id: int, interactable_id: str, action_id: str) -> Dict[str, Any]:
|
|
"""Interact with an object"""
|
|
try:
|
|
response = await self.client.post(
|
|
f"{self.api_url}/api/internal/player/{player_id}/interact",
|
|
headers=self.headers,
|
|
params={"interactable_id": interactable_id, "action_id": action_id}
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error interacting: {e}")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
# Inventory operations
|
|
async def get_inventory(self, player_id: int) -> Dict[str, Any]:
|
|
"""Get player inventory"""
|
|
try:
|
|
response = await self.client.get(
|
|
f"{self.api_url}/api/internal/player/{player_id}/inventory",
|
|
headers=self.headers
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error getting inventory: {e}")
|
|
return {"success": False, "inventory": []}
|
|
|
|
async def use_item(self, player_id: int, item_id: str) -> Dict[str, Any]:
|
|
"""Use an item"""
|
|
try:
|
|
response = await self.client.post(
|
|
f"{self.api_url}/api/internal/player/{player_id}/use_item",
|
|
headers=self.headers,
|
|
params={"item_id": item_id}
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error using item: {e}")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
async def pickup_item(self, player_id: int, item_id: str) -> Dict[str, Any]:
|
|
"""Pick up an item"""
|
|
try:
|
|
response = await self.client.post(
|
|
f"{self.api_url}/api/internal/player/{player_id}/pickup",
|
|
headers=self.headers,
|
|
params={"item_id": item_id}
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error picking up item: {e}")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
async def drop_item(self, player_id: int, item_id: str, quantity: int = 1) -> Dict[str, Any]:
|
|
"""Drop an item"""
|
|
try:
|
|
response = await self.client.post(
|
|
f"{self.api_url}/api/internal/player/{player_id}/drop_item",
|
|
headers=self.headers,
|
|
params={"item_id": item_id, "quantity": quantity}
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error dropping item: {e}")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
async def equip_item(self, player_id: int, item_id: str) -> Dict[str, Any]:
|
|
"""Equip an item"""
|
|
try:
|
|
response = await self.client.post(
|
|
f"{self.api_url}/api/internal/player/{player_id}/equip",
|
|
headers=self.headers,
|
|
params={"item_id": item_id}
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error equipping item: {e}")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
async def unequip_item(self, player_id: int, item_id: str) -> Dict[str, Any]:
|
|
"""Unequip an item"""
|
|
try:
|
|
response = await self.client.post(
|
|
f"{self.api_url}/api/internal/player/{player_id}/unequip",
|
|
headers=self.headers,
|
|
params={"item_id": item_id}
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error unequipping item: {e}")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
# Combat operations
|
|
async def get_combat(self, player_id: int) -> Optional[Dict[str, Any]]:
|
|
"""Get active combat for player"""
|
|
try:
|
|
response = await self.client.get(
|
|
f"{self.api_url}/api/internal/player/{player_id}/combat",
|
|
headers=self.headers
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error getting combat: {e}")
|
|
return None
|
|
|
|
async def create_combat(self, player_id: int, npc_id: str, npc_hp: int, npc_max_hp: int, location_id: str, from_wandering: bool = False) -> Optional[Dict[str, Any]]:
|
|
"""Create new combat"""
|
|
try:
|
|
response = await self.client.post(
|
|
f"{self.api_url}/api/internal/combat/create",
|
|
headers=self.headers,
|
|
params={
|
|
"player_id": player_id,
|
|
"npc_id": npc_id,
|
|
"npc_hp": npc_hp,
|
|
"npc_max_hp": npc_max_hp,
|
|
"location_id": location_id,
|
|
"from_wandering": from_wandering
|
|
}
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error creating combat: {e}")
|
|
return None
|
|
|
|
async def update_combat(self, player_id: int, updates: Dict[str, Any]) -> bool:
|
|
"""Update combat state"""
|
|
try:
|
|
response = await self.client.patch(
|
|
f"{self.api_url}/api/internal/combat/{player_id}",
|
|
headers=self.headers,
|
|
json=updates
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
return result.get('success', False)
|
|
except Exception as e:
|
|
print(f"Error updating combat: {e}")
|
|
return False
|
|
|
|
async def end_combat(self, player_id: int) -> bool:
|
|
"""End combat"""
|
|
try:
|
|
response = await self.client.delete(
|
|
f"{self.api_url}/api/internal/combat/{player_id}",
|
|
headers=self.headers
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
return result.get('success', False)
|
|
except Exception as e:
|
|
print(f"Error ending combat: {e}")
|
|
return False
|
|
|
|
# Player update operations
|
|
async def update_player(self, player_id: int, updates: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
"""Update player fields"""
|
|
try:
|
|
response = await self.client.patch(
|
|
f"{self.api_url}/api/internal/player/{player_id}",
|
|
headers=self.headers,
|
|
json=updates
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error updating player: {e}")
|
|
return None
|
|
|
|
# Dropped items operations
|
|
async def drop_item_to_world(self, item_id: str, quantity: int, location_id: str) -> bool:
|
|
"""Drop an item to the world"""
|
|
try:
|
|
response = await self.client.post(
|
|
f"{self.api_url}/api/internal/dropped-items",
|
|
headers=self.headers,
|
|
params={"item_id": item_id, "quantity": quantity, "location_id": location_id}
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
return result.get('success', False)
|
|
except Exception as e:
|
|
print(f"Error dropping item: {e}")
|
|
return False
|
|
|
|
async def get_dropped_item(self, dropped_item_id: int) -> Optional[Dict[str, Any]]:
|
|
"""Get a specific dropped item"""
|
|
try:
|
|
response = await self.client.get(
|
|
f"{self.api_url}/api/internal/dropped-items/{dropped_item_id}",
|
|
headers=self.headers
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error getting dropped item: {e}")
|
|
return None
|
|
|
|
async def get_dropped_items_in_location(self, location_id: str) -> List[Dict[str, Any]]:
|
|
"""Get all dropped items in a location"""
|
|
try:
|
|
response = await self.client.get(
|
|
f"{self.api_url}/api/internal/location/{location_id}/dropped-items",
|
|
headers=self.headers
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error getting dropped items: {e}")
|
|
return []
|
|
|
|
async def update_dropped_item(self, dropped_item_id: int, quantity: int) -> bool:
|
|
"""Update dropped item quantity"""
|
|
try:
|
|
response = await self.client.patch(
|
|
f"{self.api_url}/api/internal/dropped-items/{dropped_item_id}",
|
|
headers=self.headers,
|
|
params={"quantity": quantity}
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
return result.get('success', False)
|
|
except Exception as e:
|
|
print(f"Error updating dropped item: {e}")
|
|
return False
|
|
|
|
async def remove_dropped_item(self, dropped_item_id: int) -> bool:
|
|
"""Remove a dropped item"""
|
|
try:
|
|
response = await self.client.delete(
|
|
f"{self.api_url}/api/internal/dropped-items/{dropped_item_id}",
|
|
headers=self.headers
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
return result.get('success', False)
|
|
except Exception as e:
|
|
print(f"Error removing dropped item: {e}")
|
|
return False
|
|
|
|
# Corpse operations
|
|
async def create_player_corpse(self, player_name: str, location_id: str, items: str) -> Optional[int]:
|
|
"""Create a player corpse"""
|
|
try:
|
|
response = await self.client.post(
|
|
f"{self.api_url}/api/internal/corpses/player",
|
|
headers=self.headers,
|
|
params={"player_name": player_name, "location_id": location_id, "items": items}
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
return result.get('corpse_id')
|
|
except Exception as e:
|
|
print(f"Error creating player corpse: {e}")
|
|
return None
|
|
|
|
async def get_player_corpse(self, corpse_id: int) -> Optional[Dict[str, Any]]:
|
|
"""Get a player corpse"""
|
|
try:
|
|
response = await self.client.get(
|
|
f"{self.api_url}/api/internal/corpses/player/{corpse_id}",
|
|
headers=self.headers
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error getting player corpse: {e}")
|
|
return None
|
|
|
|
async def update_player_corpse(self, corpse_id: int, items: str) -> bool:
|
|
"""Update player corpse items"""
|
|
try:
|
|
response = await self.client.patch(
|
|
f"{self.api_url}/api/internal/corpses/player/{corpse_id}",
|
|
headers=self.headers,
|
|
params={"items": items}
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
return result.get('success', False)
|
|
except Exception as e:
|
|
print(f"Error updating player corpse: {e}")
|
|
return False
|
|
|
|
async def remove_player_corpse(self, corpse_id: int) -> bool:
|
|
"""Remove a player corpse"""
|
|
try:
|
|
response = await self.client.delete(
|
|
f"{self.api_url}/api/internal/corpses/player/{corpse_id}",
|
|
headers=self.headers
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
return result.get('success', False)
|
|
except Exception as e:
|
|
print(f"Error removing player corpse: {e}")
|
|
return False
|
|
|
|
async def create_npc_corpse(self, npc_id: str, location_id: str, loot_remaining: str) -> Optional[int]:
|
|
"""Create an NPC corpse"""
|
|
try:
|
|
response = await self.client.post(
|
|
f"{self.api_url}/api/internal/corpses/npc",
|
|
headers=self.headers,
|
|
params={"npc_id": npc_id, "location_id": location_id, "loot_remaining": loot_remaining}
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
return result.get('corpse_id')
|
|
except Exception as e:
|
|
print(f"Error creating NPC corpse: {e}")
|
|
return None
|
|
|
|
async def get_npc_corpse(self, corpse_id: int) -> Optional[Dict[str, Any]]:
|
|
"""Get an NPC corpse"""
|
|
try:
|
|
response = await self.client.get(
|
|
f"{self.api_url}/api/internal/corpses/npc/{corpse_id}",
|
|
headers=self.headers
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error getting NPC corpse: {e}")
|
|
return None
|
|
|
|
async def update_npc_corpse(self, corpse_id: int, loot_remaining: str) -> bool:
|
|
"""Update NPC corpse loot"""
|
|
try:
|
|
response = await self.client.patch(
|
|
f"{self.api_url}/api/internal/corpses/npc/{corpse_id}",
|
|
headers=self.headers,
|
|
params={"loot_remaining": loot_remaining}
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
return result.get('success', False)
|
|
except Exception as e:
|
|
print(f"Error updating NPC corpse: {e}")
|
|
return False
|
|
|
|
async def remove_npc_corpse(self, corpse_id: int) -> bool:
|
|
"""Remove an NPC corpse"""
|
|
try:
|
|
response = await self.client.delete(
|
|
f"{self.api_url}/api/internal/corpses/npc/{corpse_id}",
|
|
headers=self.headers
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
return result.get('success', False)
|
|
except Exception as e:
|
|
print(f"Error removing NPC corpse: {e}")
|
|
return False
|
|
|
|
# Wandering enemies operations
|
|
async def spawn_wandering_enemy(self, npc_id: str, location_id: str, current_hp: int, max_hp: int) -> Optional[int]:
|
|
"""Spawn a wandering enemy"""
|
|
try:
|
|
response = await self.client.post(
|
|
f"{self.api_url}/api/internal/wandering-enemies",
|
|
headers=self.headers,
|
|
params={"npc_id": npc_id, "location_id": location_id, "current_hp": current_hp, "max_hp": max_hp}
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
return result.get('enemy_id')
|
|
except Exception as e:
|
|
print(f"Error spawning wandering enemy: {e}")
|
|
return None
|
|
|
|
async def get_wandering_enemies_in_location(self, location_id: str) -> List[Dict[str, Any]]:
|
|
"""Get all wandering enemies in a location"""
|
|
try:
|
|
response = await self.client.get(
|
|
f"{self.api_url}/api/internal/location/{location_id}/wandering-enemies",
|
|
headers=self.headers
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error getting wandering enemies: {e}")
|
|
return []
|
|
|
|
async def remove_wandering_enemy(self, enemy_id: int) -> bool:
|
|
"""Remove a wandering enemy"""
|
|
try:
|
|
response = await self.client.delete(
|
|
f"{self.api_url}/api/internal/wandering-enemies/{enemy_id}",
|
|
headers=self.headers
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
return result.get('success', False)
|
|
except Exception as e:
|
|
print(f"Error removing wandering enemy: {e}")
|
|
return False
|
|
|
|
async def get_inventory_item(self, item_db_id: int) -> Optional[Dict[str, Any]]:
|
|
"""Get a specific inventory item by database ID"""
|
|
try:
|
|
response = await self.client.get(
|
|
f"{self.api_url}/api/internal/inventory/item/{item_db_id}",
|
|
headers=self.headers
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error getting inventory item: {e}")
|
|
return None
|
|
|
|
# Cooldown operations
|
|
async def get_cooldown(self, cooldown_key: str) -> int:
|
|
"""Get remaining cooldown time in seconds"""
|
|
try:
|
|
response = await self.client.get(
|
|
f"{self.api_url}/api/internal/cooldown/{cooldown_key}",
|
|
headers=self.headers
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
return result.get('remaining_seconds', 0)
|
|
except Exception as e:
|
|
print(f"Error getting cooldown: {e}")
|
|
return 0
|
|
|
|
async def set_cooldown(self, cooldown_key: str, duration_seconds: int = 600) -> bool:
|
|
"""Set a cooldown"""
|
|
try:
|
|
response = await self.client.post(
|
|
f"{self.api_url}/api/internal/cooldown/{cooldown_key}",
|
|
headers=self.headers,
|
|
params={"duration_seconds": duration_seconds}
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
return result.get('success', False)
|
|
except Exception as e:
|
|
print(f"Error setting cooldown: {e}")
|
|
return False
|
|
|
|
# Corpse list operations
|
|
async def get_player_corpses_in_location(self, location_id: str) -> List[Dict[str, Any]]:
|
|
"""Get all player corpses in a location"""
|
|
try:
|
|
response = await self.client.get(
|
|
f"{self.api_url}/api/internal/location/{location_id}/corpses/player",
|
|
headers=self.headers
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error getting player corpses: {e}")
|
|
return []
|
|
|
|
async def get_npc_corpses_in_location(self, location_id: str) -> List[Dict[str, Any]]:
|
|
"""Get all NPC corpses in a location"""
|
|
try:
|
|
response = await self.client.get(
|
|
f"{self.api_url}/api/internal/location/{location_id}/corpses/npc",
|
|
headers=self.headers
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error getting NPC corpses: {e}")
|
|
return []
|
|
|
|
# Image cache operations
|
|
async def get_cached_image(self, image_path: str) -> Optional[str]:
|
|
"""Get cached telegram file ID for an image"""
|
|
try:
|
|
response = await self.client.get(
|
|
f"{self.api_url}/api/internal/image-cache/{image_path}",
|
|
headers=self.headers
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
return result.get('telegram_file_id')
|
|
except Exception as e:
|
|
# Not found is expected, not an error
|
|
return None
|
|
|
|
async def cache_image(self, image_path: str, telegram_file_id: str) -> bool:
|
|
"""Cache a telegram file ID for an image"""
|
|
try:
|
|
response = await self.client.post(
|
|
f"{self.api_url}/api/internal/image-cache",
|
|
headers=self.headers,
|
|
params={"image_path": image_path, "telegram_file_id": telegram_file_id}
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
return result.get('success', False)
|
|
except Exception as e:
|
|
print(f"Error caching image: {e}")
|
|
return False
|
|
|
|
# Status effects operations
|
|
async def get_player_status_effects(self, player_id: int) -> List[Dict[str, Any]]:
|
|
"""Get player status effects"""
|
|
try:
|
|
response = await self.client.get(
|
|
f"{self.api_url}/api/internal/player/{player_id}/status-effects",
|
|
headers=self.headers
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error getting status effects: {e}")
|
|
return []
|
|
|
|
|
|
# Global API client instance
|
|
api_client = APIClient()
|