371 lines
11 KiB
Python
371 lines
11 KiB
Python
"""
|
|
Internal/Admin API router.
|
|
Endpoints for internal services (bot, admin tools, etc.)
|
|
Requires API_INTERNAL_KEY for authentication.
|
|
"""
|
|
from fastapi import APIRouter, HTTPException, Depends
|
|
from typing import Dict, Any
|
|
import json
|
|
|
|
from ..core.security import verify_internal_key
|
|
from .. import database as db
|
|
from ..items import ItemsManager
|
|
|
|
# These will be injected by main.py
|
|
LOCATIONS = None
|
|
ITEMS_MANAGER = None
|
|
WORLD = None
|
|
IMAGES_DIR = None
|
|
|
|
def init_router_dependencies(locations, items_manager, world, images_dir):
|
|
"""Initialize router with game data dependencies"""
|
|
global LOCATIONS, ITEMS_MANAGER, WORLD, IMAGES_DIR
|
|
LOCATIONS = locations
|
|
ITEMS_MANAGER = items_manager
|
|
WORLD = world
|
|
IMAGES_DIR = images_dir
|
|
|
|
router = APIRouter(prefix="/api/internal", tags=["internal"], dependencies=[Depends(verify_internal_key)])
|
|
|
|
|
|
# Player endpoints
|
|
@router.get("/player/by_id/{player_id}")
|
|
async def get_player_by_id(player_id: int):
|
|
"""Get player data by ID"""
|
|
player = await db.get_player_by_id(player_id)
|
|
if not player:
|
|
raise HTTPException(status_code=404, detail="Player not found")
|
|
return player
|
|
|
|
|
|
@router.patch("/player/{player_id}")
|
|
async def update_player(player_id: int, data: dict):
|
|
"""Update player"""
|
|
await db.update_player(player_id, **data)
|
|
return {"success": True}
|
|
|
|
|
|
@router.get("/player/{player_id}/inventory")
|
|
async def get_inventory(player_id: int):
|
|
"""Get player inventory"""
|
|
inventory = await db.get_inventory(player_id)
|
|
return inventory
|
|
|
|
|
|
@router.get("/player/{player_id}/status-effects")
|
|
async def get_player_status_effects(player_id: int):
|
|
"""Get player's active status effects"""
|
|
effects = await db.get_active_status_effects(player_id)
|
|
return effects
|
|
|
|
|
|
# Combat endpoints
|
|
@router.get("/player/{player_id}/combat")
|
|
async def get_player_combat(player_id: int):
|
|
"""Get player's active combat"""
|
|
combat = await db.get_active_combat(player_id)
|
|
return combat
|
|
|
|
|
|
@router.post("/combat/create")
|
|
async def create_combat(data: dict):
|
|
"""Create combat"""
|
|
return await db.create_combat(**data)
|
|
|
|
|
|
@router.patch("/combat/{player_id}")
|
|
async def update_combat(player_id: int, data: dict):
|
|
"""Update combat"""
|
|
await db.update_combat(player_id, **data)
|
|
return {"success": True}
|
|
|
|
|
|
@router.delete("/combat/{player_id}")
|
|
async def end_combat(player_id: int):
|
|
"""End combat"""
|
|
await db.end_combat(player_id)
|
|
return {"success": True}
|
|
|
|
|
|
# Game action endpoints
|
|
@router.post("/player/{player_id}/move")
|
|
async def move_player(player_id: int, data: dict):
|
|
"""Move player"""
|
|
from .. import game_logic
|
|
success, message, new_location_id, stamina_cost, distance = await game_logic.move_player(
|
|
player_id,
|
|
data['direction'],
|
|
LOCATIONS
|
|
)
|
|
return {"success": success, "message": message, "new_location_id": new_location_id}
|
|
|
|
|
|
@router.get("/player/{player_id}/inspect")
|
|
async def inspect_player(player_id: int):
|
|
"""Inspect area for player"""
|
|
player = await db.get_player_by_id(player_id)
|
|
location = LOCATIONS.get(player['location_id'])
|
|
from .. import game_logic
|
|
message = await game_logic.inspect_area(player_id, location, {})
|
|
return {"message": message}
|
|
|
|
|
|
@router.post("/player/{player_id}/interact")
|
|
async def interact_player(player_id: int, data: dict):
|
|
"""Interact for player"""
|
|
player = await db.get_player_by_id(player_id)
|
|
location = LOCATIONS.get(player['location_id'])
|
|
from .. import game_logic
|
|
result = await game_logic.interact_with_object(
|
|
player_id,
|
|
data['interactable_id'],
|
|
data['action_id'],
|
|
location,
|
|
ITEMS_MANAGER
|
|
)
|
|
return result
|
|
|
|
|
|
@router.post("/player/{player_id}/use_item")
|
|
async def use_item(player_id: int, data: dict):
|
|
"""Use item"""
|
|
from .. import game_logic
|
|
result = await game_logic.use_item(player_id, data['item_id'], ITEMS_MANAGER)
|
|
return result
|
|
|
|
|
|
@router.post("/player/{player_id}/pickup")
|
|
async def pickup_item(player_id: int, data: dict):
|
|
"""Pickup item"""
|
|
player = await db.get_player_by_id(player_id)
|
|
from .. import game_logic
|
|
result = await game_logic.pickup_item(
|
|
player_id,
|
|
data['item_id'],
|
|
player['location_id'],
|
|
data.get('quantity'),
|
|
ITEMS_MANAGER
|
|
)
|
|
return result
|
|
|
|
|
|
@router.post("/player/{player_id}/drop_item")
|
|
async def drop_item(player_id: int, data: dict):
|
|
"""Drop item"""
|
|
player = await db.get_player_by_id(player_id)
|
|
await db.drop_item(player_id, data['item_id'], data['quantity'], player['location_id'])
|
|
return {"success": True}
|
|
|
|
|
|
# Equipment endpoints
|
|
@router.post("/player/{player_id}/equip")
|
|
async def equip_item(player_id: int, data: dict):
|
|
"""Equip item"""
|
|
inv_item = await db.get_inventory_item_by_id(data['inventory_id'])
|
|
if not inv_item:
|
|
raise HTTPException(status_code=404, detail="Item not found")
|
|
|
|
item_def = ITEMS_MANAGER.get_item(inv_item['item_id'])
|
|
if not item_def or not item_def.equippable:
|
|
raise HTTPException(status_code=400, detail="Item not equippable")
|
|
|
|
# Unequip current item in slot if any
|
|
current = await db.get_equipped_item_in_slot(player_id, item_def.slot)
|
|
if current:
|
|
await db.unequip_item(player_id, item_def.slot)
|
|
await db.update_inventory_item(current['item_id'], is_equipped=False)
|
|
|
|
# Equip new item
|
|
await db.equip_item(player_id, item_def.slot, data['inventory_id'])
|
|
await db.update_inventory_item(data['inventory_id'], is_equipped=True)
|
|
|
|
return {"success": True, "slot": item_def.slot}
|
|
|
|
|
|
@router.post("/player/{player_id}/unequip")
|
|
async def unequip_item(player_id: int, data: dict):
|
|
"""Unequip item"""
|
|
equipped = await db.get_equipped_item_in_slot(player_id, data['slot'])
|
|
if not equipped:
|
|
raise HTTPException(status_code=400, detail="No item in slot")
|
|
|
|
await db.unequip_item(player_id, data['slot'])
|
|
await db.update_inventory_item(equipped['item_id'], is_equipped=False)
|
|
|
|
return {"success": True}
|
|
|
|
|
|
# Dropped items endpoints
|
|
@router.post("/dropped-items")
|
|
async def create_dropped_item(data: dict):
|
|
"""Create dropped item"""
|
|
await db.drop_item(None, data['item_id'], data['quantity'], data['location_id'])
|
|
return {"success": True}
|
|
|
|
|
|
@router.get("/dropped-items/{dropped_item_id}")
|
|
async def get_dropped_item(dropped_item_id: int):
|
|
"""Get dropped item"""
|
|
item = await db.get_dropped_item(dropped_item_id)
|
|
return item
|
|
|
|
|
|
@router.get("/location/{location_id}/dropped-items")
|
|
async def get_location_dropped_items(location_id: str):
|
|
"""Get location's dropped items"""
|
|
items = await db.get_dropped_items(location_id)
|
|
return items
|
|
|
|
|
|
@router.patch("/dropped-items/{dropped_item_id}")
|
|
async def update_dropped_item(dropped_item_id: int, data: dict):
|
|
"""Update dropped item"""
|
|
return {"success": True}
|
|
|
|
|
|
@router.delete("/dropped-items/{dropped_item_id}")
|
|
async def delete_dropped_item(dropped_item_id: int):
|
|
"""Delete dropped item"""
|
|
await db.delete_dropped_item(dropped_item_id)
|
|
return {"success": True}
|
|
|
|
|
|
# Corpse endpoints - Player
|
|
@router.post("/corpses/player")
|
|
async def create_player_corpse(data: dict):
|
|
"""Create player corpse"""
|
|
corpse_id = await db.create_player_corpse(**data)
|
|
return {"id": corpse_id}
|
|
|
|
|
|
@router.get("/corpses/player/{corpse_id}")
|
|
async def get_player_corpse(corpse_id: int):
|
|
"""Get player corpse"""
|
|
corpse = await db.get_player_corpse(corpse_id)
|
|
return corpse
|
|
|
|
|
|
@router.patch("/corpses/player/{corpse_id}")
|
|
async def update_player_corpse(corpse_id: int, data: dict):
|
|
"""Update player corpse"""
|
|
await db.update_player_corpse(corpse_id, **data)
|
|
return {"success": True}
|
|
|
|
|
|
@router.delete("/corpses/player/{corpse_id}")
|
|
async def delete_player_corpse(corpse_id: int):
|
|
"""Delete player corpse"""
|
|
await db.delete_player_corpse(corpse_id)
|
|
return {"success": True}
|
|
|
|
|
|
@router.get("/location/{location_id}/corpses/player")
|
|
async def get_player_corpses_in_location(location_id: str):
|
|
"""Get player corpses in location"""
|
|
corpses = await db.get_player_corpses_in_location(location_id)
|
|
return corpses
|
|
|
|
|
|
# Corpse endpoints - NPC
|
|
@router.post("/corpses/npc")
|
|
async def create_npc_corpse(data: dict):
|
|
"""Create NPC corpse"""
|
|
corpse_id = await db.create_npc_corpse(
|
|
npc_id=data['npc_id'],
|
|
location_id=data['location_id'],
|
|
loot=json.dumps(data['loot'])
|
|
)
|
|
return {"id": corpse_id}
|
|
|
|
|
|
@router.get("/corpses/npc/{corpse_id}")
|
|
async def get_npc_corpse(corpse_id: int):
|
|
"""Get NPC corpse"""
|
|
corpse = await db.get_npc_corpse(corpse_id)
|
|
return corpse
|
|
|
|
|
|
@router.patch("/corpses/npc/{corpse_id}")
|
|
async def update_npc_corpse(corpse_id: int, data: dict):
|
|
"""Update NPC corpse"""
|
|
await db.update_npc_corpse(corpse_id, **data)
|
|
return {"success": True}
|
|
|
|
|
|
@router.delete("/corpses/npc/{corpse_id}")
|
|
async def delete_npc_corpse(corpse_id: int):
|
|
"""Delete NPC corpse"""
|
|
await db.delete_npc_corpse(corpse_id)
|
|
return {"success": True}
|
|
|
|
|
|
@router.get("/location/{location_id}/corpses/npc")
|
|
async def get_npc_corpses_in_location(location_id: str):
|
|
"""Get NPC corpses in location"""
|
|
corpses = await db.get_npc_corpses_in_location(location_id)
|
|
return corpses
|
|
|
|
|
|
# Wandering enemies endpoints
|
|
@router.post("/wandering-enemies")
|
|
async def create_wandering_enemy(data: dict):
|
|
"""Create wandering enemy"""
|
|
enemy_id = await db.create_wandering_enemy(**data)
|
|
return {"id": enemy_id}
|
|
|
|
|
|
@router.get("/location/{location_id}/wandering-enemies")
|
|
async def get_wandering_enemies(location_id: str):
|
|
"""Get wandering enemies in location"""
|
|
enemies = await db.get_wandering_enemies_in_location(location_id)
|
|
return enemies
|
|
|
|
|
|
@router.delete("/wandering-enemies/{enemy_id}")
|
|
async def delete_wandering_enemy(enemy_id: int):
|
|
"""Delete wandering enemy"""
|
|
await db.delete_wandering_enemy(enemy_id)
|
|
return {"success": True}
|
|
|
|
|
|
# Inventory item endpoint
|
|
@router.get("/inventory/item/{item_db_id}")
|
|
async def get_inventory_item(item_db_id: int):
|
|
"""Get inventory item"""
|
|
item = await db.get_inventory_item_by_id(item_db_id)
|
|
return item
|
|
|
|
|
|
# Cooldown endpoints
|
|
@router.get("/cooldown/{cooldown_key}")
|
|
async def get_cooldown(cooldown_key: str):
|
|
"""Get cooldown"""
|
|
parts = cooldown_key.split(':')
|
|
if len(parts) >= 3:
|
|
expiry = await db.get_interactable_cooldown(parts[1], parts[2])
|
|
return {"expiry": expiry}
|
|
return {"expiry": None}
|
|
|
|
|
|
@router.post("/cooldown/{cooldown_key}")
|
|
async def set_cooldown(cooldown_key: str, data: dict):
|
|
"""Set cooldown"""
|
|
parts = cooldown_key.split(':')
|
|
if len(parts) >= 3:
|
|
await db.set_interactable_cooldown(parts[1], parts[2], data['duration'])
|
|
return {"success": True}
|
|
|
|
|
|
# Image cache endpoints
|
|
@router.get("/image-cache/{image_path:path}")
|
|
async def get_image_cache(image_path: str):
|
|
"""Check if image exists"""
|
|
full_path = IMAGES_DIR / image_path
|
|
return {"exists": full_path.exists()}
|
|
|
|
|
|
@router.post("/image-cache")
|
|
async def create_image_cache(data: dict):
|
|
"""Cache image"""
|
|
return {"success": True}
|