Jocaru/echoes-of-the-ash
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

What a mess

Browse Source
This commit is contained in:
Joan
2025-11-07 15:27:13 +01:00
parent 0b79b3ae59
commit 33cc9586c2
130 changed files with 29819 additions and 1175 deletions
Download Patch File Download Diff File Expand all files Collapse all files

465
api/background_tasks.py Normal file
View File

@@ -0,0 +1,465 @@
"""
Background tasks for the API.
Handles periodic maintenance, regeneration, spawning, and processing.
"""
import asyncio
import logging
import random
import time
import os
import fcntl
from typing import Dict, Optional
# Import from API modules (not bot modules)
from api import database as db
from data.npcs import (
LOCATION_SPAWNS,
LOCATION_DANGER,
get_random_npc_for_location,
get_wandering_enemy_chance
)
logger = logging.getLogger(__name__)
# Lock file to ensure only one worker runs background tasks
LOCK_FILE_PATH = "/tmp/echoes_background_tasks.lock"
_lock_file_handle: Optional[int] = None
# ============================================================================
# SPAWN MANAGER CONFIGURATION
# ============================================================================
SPAWN_CHECK_INTERVAL = 120 # Check every 2 minutes
ENEMY_LIFETIME = 600 # Enemies live for 10 minutes
MAX_ENEMIES_PER_LOCATION = {
0: 0, # Safe zones - no wandering enemies
1: 1, # Low danger - max 1 enemy
2: 2, # Medium danger - max 2 enemies
3: 3, # High danger - max 3 enemies
4: 4, # Extreme danger - max 4 enemies
}
def get_danger_level(location_id: str) -> int:
"""Get danger level for a location."""
danger_data = LOCATION_DANGER.get(location_id, (0, 0.0, 0.0))
return danger_data[0]
# ============================================================================
# BACKGROUND TASK: WANDERING ENEMY SPAWNER
# ============================================================================
async def spawn_manager_loop():
"""
Main spawn manager loop.
Runs continuously, checking spawn conditions every SPAWN_CHECK_INTERVAL seconds.
"""
logger.info("🎲 Spawn Manager started")
while True:
try:
await asyncio.sleep(SPAWN_CHECK_INTERVAL)
# Clean up expired enemies first
despawned_count = await db.cleanup_expired_wandering_enemies()
if despawned_count > 0:
logger.info(f"🧹 Cleaned up {despawned_count} expired wandering enemies")
# Process each location
spawned_count = 0
for location_id, spawn_table in LOCATION_SPAWNS.items():
if not spawn_table:
continue # Skip locations with no spawns
# Get danger level and max enemies for this location
danger_level = get_danger_level(location_id)
max_enemies = MAX_ENEMIES_PER_LOCATION.get(danger_level, 0)
if max_enemies == 0:
continue # Skip safe zones
# Check current enemy count
current_count = await db.get_wandering_enemy_count_in_location(location_id)
if current_count >= max_enemies:
continue # Location is at capacity
# Calculate spawn chance based on wandering_enemy_chance
spawn_chance = get_wandering_enemy_chance(location_id)
# Attempt to spawn enemies up to max capacity
for _ in range(max_enemies - current_count):
if random.random() < spawn_chance:
# Spawn an enemy
npc_id = get_random_npc_for_location(location_id)
if npc_id:
await db.spawn_wandering_enemy(
npc_id=npc_id,
location_id=location_id,
lifetime_seconds=ENEMY_LIFETIME
)
spawned_count += 1
logger.info(f"👹 Spawned {npc_id} at {location_id} (current: {current_count + 1}/{max_enemies})")
if spawned_count > 0:
logger.info(f"✨ Spawn cycle complete: {spawned_count} enemies spawned")
except Exception as e:
logger.error(f"❌ Error in spawn manager loop: {e}", exc_info=True)
# Continue running even if there's an error
await asyncio.sleep(10)
# ============================================================================
# BACKGROUND TASK: DROPPED ITEM DECAY
# ============================================================================
async def decay_dropped_items():
"""Periodically cleans up old dropped items."""
logger.info("🗑️ Item Decay task started")
while True:
try:
await asyncio.sleep(300) # Wait 5 minutes
start_time = time.time()
logger.info("Running item decay task...")
# Set decay time to 1 hour (3600 seconds)
decay_seconds = 3600
timestamp_limit = int(time.time()) - decay_seconds
items_removed = await db.remove_expired_dropped_items(timestamp_limit)
elapsed = time.time() - start_time
if items_removed > 0:
logger.info(f"Decayed and removed {items_removed} old items in {elapsed:.2f}s")
except Exception as e:
logger.error(f"❌ Error in item decay task: {e}", exc_info=True)
await asyncio.sleep(10)
# ============================================================================
# BACKGROUND TASK: STAMINA REGENERATION
# ============================================================================
async def regenerate_stamina():
"""Periodically regenerates stamina for all players."""
logger.info("💪 Stamina Regeneration task started")
while True:
try:
await asyncio.sleep(300) # Wait 5 minutes
start_time = time.time()
logger.info("Running stamina regeneration...")
players_updated = await db.regenerate_all_players_stamina()
elapsed = time.time() - start_time
if players_updated > 0:
logger.info(f"Regenerated stamina for {players_updated} players in {elapsed:.2f}s")
# Alert if regeneration is taking too long (potential scaling issue)
if elapsed > 5.0:
logger.warning(f"⚠️ Stamina regeneration took {elapsed:.2f}s (threshold: 5s) - check database load!")
except Exception as e:
logger.error(f"❌ Error in stamina regeneration: {e}", exc_info=True)
await asyncio.sleep(10)
# ============================================================================
# BACKGROUND TASK: COMBAT TIMERS
# ============================================================================
async def check_combat_timers():
"""Checks for idle combat turns and auto-attacks."""
logger.info("⚔️ Combat Timer task started")
while True:
try:
await asyncio.sleep(30) # Wait 30 seconds
start_time = time.time()
# Check for combats idle for more than 5 minutes (300 seconds)
idle_threshold = time.time() - 300
idle_combats = await db.get_all_idle_combats(idle_threshold)
if idle_combats:
logger.info(f"Processing {len(idle_combats)} idle combats...")
for combat in idle_combats:
try:
# Import combat logic from API
from api import game_logic
# Force end player's turn and let NPC attack
if combat['turn'] == 'player':
await db.update_combat(combat['player_id'], {
'turn': 'npc',
'turn_started_at': time.time()
})
# NPC attacks
await game_logic.npc_attack(combat['player_id'])
except Exception as e:
logger.error(f"Error processing idle combat: {e}")
# Log performance for monitoring
if idle_combats:
elapsed = time.time() - start_time
logger.info(f"Processed {len(idle_combats)} idle combats in {elapsed:.2f}s")
# Warn if taking too long (potential scaling issue)
if elapsed > 10.0:
logger.warning(f"⚠️ Combat timer check took {elapsed:.2f}s (threshold: 10s) - consider batching!")
except Exception as e:
logger.error(f"❌ Error in combat timer check: {e}", exc_info=True)
await asyncio.sleep(10)
# ============================================================================
# BACKGROUND TASK: CORPSE DECAY
# ============================================================================
async def decay_corpses():
"""Removes old corpses."""
logger.info("💀 Corpse Decay task started")
while True:
try:
await asyncio.sleep(600) # Wait 10 minutes
start_time = time.time()
logger.info("Running corpse decay...")
# Player corpses decay after 24 hours
player_corpse_limit = time.time() - (24 * 3600)
player_corpses_removed = await db.remove_expired_player_corpses(player_corpse_limit)
# NPC corpses decay after 2 hours
npc_corpse_limit = time.time() - (2 * 3600)
npc_corpses_removed = await db.remove_expired_npc_corpses(npc_corpse_limit)
elapsed = time.time() - start_time
if player_corpses_removed > 0 or npc_corpses_removed > 0:
logger.info(f"Decayed {player_corpses_removed} player corpses and {npc_corpses_removed} NPC corpses in {elapsed:.2f}s")
except Exception as e:
logger.error(f"❌ Error in corpse decay: {e}", exc_info=True)
await asyncio.sleep(10)
# ============================================================================
# BACKGROUND TASK: STATUS EFFECTS PROCESSOR
# ============================================================================
async def process_status_effects():
"""
Applies damage from persistent status effects.
Runs every 5 minutes to process status effect ticks.
"""
logger.info("🩸 Status Effects Processor started")
while True:
try:
await asyncio.sleep(300) # Wait 5 minutes
start_time = time.time()
logger.info("Running status effects processor...")
try:
# Decrement all status effect ticks and get affected players
affected_players = await db.decrement_all_status_effect_ticks()
if not affected_players:
elapsed = time.time() - start_time
logger.info(f"No active status effects to process ({elapsed:.3f}s)")
continue
# Process each affected player
deaths = 0
damage_dealt = 0
for player_id in affected_players:
try:
# Get current status effects (after decrement)
effects = await db.get_player_status_effects(player_id)
if not effects:
continue
# Calculate total damage
from api.game_logic import calculate_status_damage
total_damage = calculate_status_damage(effects)
if total_damage > 0:
damage_dealt += total_damage
player = await db.get_player_by_id(player_id)
if not player or player['is_dead']:
continue
new_hp = max(0, player['hp'] - total_damage)
# Check if player died from status effects
if new_hp <= 0:
await db.update_player(player_id, {'hp': 0, 'is_dead': True})
deaths += 1
# Create player corpse
inventory = await db.get_inventory(player_id)
await db.create_player_corpse(
player_name=player['name'],
location_id=player['location_id'],
items=inventory
)
# Remove status effects from dead player
await db.remove_all_status_effects(player_id)
logger.info(f"Player {player['name']} (ID: {player_id}) died from status effects")
else:
# Apply damage
await db.update_player(player_id, {'hp': new_hp})
except Exception as e:
logger.error(f"Error processing status effects for player {player_id}: {e}")
elapsed = time.time() - start_time
logger.info(
f"Processed status effects for {len(affected_players)} players "
f"({damage_dealt} total damage, {deaths} deaths) in {elapsed:.3f}s"
)
# Warn if taking too long (potential scaling issue)
if elapsed > 5.0:
logger.warning(
f"⚠️ Status effects processing took {elapsed:.3f}s (threshold: 5s) "
f"- {len(affected_players)} players affected"
)
except Exception as e:
logger.error(f"Error in status effects processor: {e}")
except Exception as e:
logger.error(f"❌ Error in status effects task: {e}", exc_info=True)
await asyncio.sleep(10)
# ============================================================================
# TASK STARTUP FUNCTION
# ============================================================================
def acquire_background_tasks_lock() -> bool:
"""
Try to acquire an exclusive lock for running background tasks.
Only one worker across all Gunicorn processes should succeed.
Returns True if lock acquired, False otherwise.
"""
global _lock_file_handle
try:
# Open lock file (create if doesn't exist)
_lock_file_handle = os.open(LOCK_FILE_PATH, os.O_CREAT | os.O_RDWR)
# Try to acquire exclusive, non-blocking lock
fcntl.flock(_lock_file_handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
logger.info("🔒 Successfully acquired background tasks lock")
return True
except (IOError, OSError) as e:
# Lock already held by another worker
if _lock_file_handle is not None:
try:
os.close(_lock_file_handle)
except:
pass
_lock_file_handle = None
return False
def release_background_tasks_lock():
"""Release the background tasks lock."""
global _lock_file_handle
if _lock_file_handle is not None:
try:
fcntl.flock(_lock_file_handle, fcntl.LOCK_UN)
os.close(_lock_file_handle)
logger.info("🔓 Released background tasks lock")
except Exception as e:
logger.error(f"Error releasing lock: {e}")
finally:
_lock_file_handle = None
async def start_background_tasks():
"""
Start all background tasks.
Called when the API starts up.
Only runs in ONE worker (the first one to acquire the lock).
"""
# Try to acquire lock - only one worker will succeed
if not acquire_background_tasks_lock():
logger.info("⏭️ Background tasks already running in another worker, skipping...")
return []
logger.info("🚀 Starting background tasks in this worker...")
# Create tasks for all background jobs
tasks = [
asyncio.create_task(spawn_manager_loop()),
asyncio.create_task(decay_dropped_items()),
asyncio.create_task(regenerate_stamina()),
asyncio.create_task(check_combat_timers()),
asyncio.create_task(decay_corpses()),
asyncio.create_task(process_status_effects()),
]
logger.info(f"✅ Started {len(tasks)} background tasks")
return tasks
async def stop_background_tasks(tasks):
"""Stop all background tasks and release the lock."""
if not tasks:
return
logger.info("🛑 Shutting down background tasks...")
for task in tasks:
task.cancel()
# Wait for tasks to finish canceling
await asyncio.gather(*tasks, return_exceptions=True)
# Release the lock
release_background_tasks_lock()
logger.info("✅ Background tasks stopped")
# ============================================================================
# MONITORING / DEBUG FUNCTIONS
# ============================================================================
async def get_spawn_stats() -> Dict:
"""Get statistics about current spawns (for debugging/monitoring)."""
all_enemies = await db.get_all_active_wandering_enemies()
# Count by location
location_counts = {}
for enemy in all_enemies:
loc = enemy['location_id']
location_counts[loc] = location_counts.get(loc, 0) + 1
return {
"total_active": len(all_enemies),
"by_location": location_counts,
"enemies": all_enemies
}

1646
api/database.py Normal file
View File

File diff suppressed because it is too large Load Diff

506
api/game_logic.py Normal file
View File

@@ -0,0 +1,506 @@
"""
Standalone game logic for the API.
Contains all game mechanics without bot dependencies.
"""
import random
import time
from typing import Dict, Any, Tuple, Optional, List
from . import database as db
async def move_player(player_id: int, direction: str, locations: Dict) -> Tuple[bool, str, Optional[str], int, int]:
"""
Move player in a direction.
Returns: (success, message, new_location_id, stamina_cost, distance_meters)
"""
player = await db.get_player_by_id(player_id)
if not player:
return False, "Player not found", None, 0, 0
current_location_id = player['location_id']
current_location = locations.get(current_location_id)
if not current_location:
return False, "Current location not found", None, 0, 0
# Check if direction is valid
if direction not in current_location.exits:
return False, f"You cannot go {direction} from here.", None, 0, 0
new_location_id = current_location.exits[direction]
new_location = locations.get(new_location_id)
if not new_location:
return False, "Destination not found", None, 0, 0
# Calculate total weight
from api.items import items_manager as ITEMS_MANAGER
inventory = await db.get_inventory(player_id)
total_weight = 0.0
for inv_item in inventory:
item = ITEMS_MANAGER.get_item(inv_item['item_id'])
if item:
total_weight += item.weight * inv_item['quantity']
# Calculate distance between locations (1 coordinate unit = 100 meters)
import math
coord_distance = math.sqrt(
(new_location.x - current_location.x)**2 +
(new_location.y - current_location.y)**2
)
distance = int(coord_distance * 100) # Convert to meters, round to integer
# Calculate stamina cost: base from distance, adjusted by weight and agility
base_cost = max(1, round(distance / 50)) # 50m = 1 stamina
weight_penalty = int(total_weight / 10)
agility_reduction = int(player.get('agility', 5) / 3)
stamina_cost = max(1, base_cost + weight_penalty - agility_reduction)
# Check stamina
if player['stamina'] < stamina_cost:
return False, "You're too exhausted to move. Wait for your stamina to regenerate.", None, 0, 0
# Update player location and stamina
await db.update_player(
player_id,
location_id=new_location_id,
stamina=max(0, player['stamina'] - stamina_cost)
)
return True, f"You travel {direction} to {new_location.name}.", new_location_id, stamina_cost, distance
async def inspect_area(player_id: int, location, interactables_data: Dict) -> str:
"""
Inspect the current area and return detailed information.
Returns formatted text with interactables and their actions.
"""
player = await db.get_player_by_id(player_id)
if not player:
return "Player not found"
# Check if player has enough stamina
if player['stamina'] < 1:
return "You're too exhausted to inspect the area thoroughly. Wait for your stamina to regenerate."
# Deduct stamina
await db.update_player_stamina(player_id, player['stamina'] - 1)
# Build inspection message
lines = [f"🔍 **Inspecting {location.name}**\n"]
lines.append(location.description)
lines.append("")
if location.interactables:
lines.append("**Interactables:**")
for interactable in location.interactables:
lines.append(f"• **{interactable.name}**")
if interactable.actions:
actions_text = ", ".join([f"{action.label} (⚡{action.stamina_cost})" for action in interactable.actions])
lines.append(f" Actions: {actions_text}")
lines.append("")
if location.npcs:
lines.append(f"**NPCs:** {', '.join(location.npcs)}")
lines.append("")
# Check for dropped items
dropped_items = await db.get_dropped_items(location.id)
if dropped_items:
lines.append("**Items on ground:**")
for item in dropped_items:
lines.append(f"• {item['item_id']} x{item['quantity']}")
return "\n".join(lines)
async def interact_with_object(
player_id: int,
interactable_id: str,
action_id: str,
location,
items_manager
) -> Dict[str, Any]:
"""
Interact with an object using a specific action.
Returns: {success, message, items_found, damage_taken, stamina_cost}
"""
player = await db.get_player_by_id(player_id)
if not player:
return {"success": False, "message": "Player not found"}
# Find the interactable
interactable = None
for obj in location.interactables:
if obj.id == interactable_id:
interactable = obj
break
if not interactable:
return {"success": False, "message": "Object not found"}
# Find the action
action = None
for act in interactable.actions:
if act.id == action_id:
action = act
break
if not action:
return {"success": False, "message": "Action not found"}
# Check stamina
if player['stamina'] < action.stamina_cost:
return {
"success": False,
"message": f"Not enough stamina. Need {action.stamina_cost}, have {player['stamina']}."
}
# Check cooldown
cooldown_expiry = await db.get_interactable_cooldown(interactable_id)
if cooldown_expiry:
remaining = int(cooldown_expiry - time.time())
return {
"success": False,
"message": f"This object is still recovering. Wait {remaining} seconds."
}
# Deduct stamina
new_stamina = player['stamina'] - action.stamina_cost
await db.update_player_stamina(player_id, new_stamina)
# Determine outcome (simple success/failure for now)
# TODO: Implement proper skill checks
roll = random.randint(1, 100)
if roll <= 10: # 10% critical failure
outcome_key = 'critical_failure'
elif roll <= 30: # 20% failure
outcome_key = 'failure'
else: # 70% success
outcome_key = 'success'
outcome = action.outcomes.get(outcome_key)
if not outcome:
# Fallback to success if outcome not defined
outcome = action.outcomes.get('success')
if not outcome:
return {
"success": False,
"message": "Action has no defined outcomes"
}
# Process outcome
items_found = []
items_dropped = []
damage_taken = outcome.damage_taken
# Calculate current capacity
from api.main import calculate_player_capacity
current_weight, max_weight, current_volume, max_volume = await calculate_player_capacity(player_id)
# Add items to inventory (or drop if over capacity)
for item_id, quantity in outcome.items_reward.items():
item = items_manager.get_item(item_id)
if not item:
continue
item_name = item.name if item else item_id
emoji = item.emoji if item and hasattr(item, 'emoji') else ''
# Check if item has durability (unique item)
has_durability = hasattr(item, 'durability') and item.durability is not None
# For items with durability, we need to create each one individually
if has_durability:
for _ in range(quantity):
# Check if item fits in inventory
if (current_weight + item.weight <= max_weight and
current_volume + item.volume <= max_volume):
# Add to inventory with durability properties
await db.add_item_to_inventory(
player_id,
item_id,
quantity=1,
durability=item.durability,
max_durability=item.durability,
tier=getattr(item, 'tier', None)
)
items_found.append(f"{emoji} {item_name}")
current_weight += item.weight
current_volume += item.volume
else:
# Create unique_item and drop to ground
unique_item_id = await db.create_unique_item(
item_id=item_id,
durability=item.durability,
max_durability=item.durability,
tier=getattr(item, 'tier', None)
)
await db.drop_item_to_world(item_id, 1, player['location_id'], unique_item_id=unique_item_id)
items_dropped.append(f"{emoji} {item_name}")
else:
# Stackable items - process as before
item_weight = item.weight * quantity
item_volume = item.volume * quantity
if (current_weight + item_weight <= max_weight and
current_volume + item_volume <= max_volume):
# Add to inventory
await db.add_item_to_inventory(player_id, item_id, quantity)
items_found.append(f"{emoji} {item_name} x{quantity}")
current_weight += item_weight
current_volume += item_volume
else:
# Drop to ground
await db.drop_item_to_world(item_id, quantity, player['location_id'])
items_dropped.append(f"{emoji} {item_name} x{quantity}")
# Apply damage
if damage_taken > 0:
new_hp = max(0, player['hp'] - damage_taken)
await db.update_player_hp(player_id, new_hp)
# Check if player died
if new_hp <= 0:
await db.update_player(player_id, is_dead=True)
# Set cooldown (60 seconds default)
await db.set_interactable_cooldown(interactable_id, 60)
# Build message
final_message = outcome.text
if items_dropped:
final_message += f"\n⚠️ Inventory full! Dropped to ground: {', '.join(items_dropped)}"
return {
"success": True,
"message": final_message,
"items_found": items_found,
"items_dropped": items_dropped,
"damage_taken": damage_taken,
"stamina_cost": action.stamina_cost,
"new_stamina": new_stamina,
"new_hp": player['hp'] - damage_taken if damage_taken > 0 else player['hp']
}
async def use_item(player_id: int, item_id: str, items_manager) -> Dict[str, Any]:
"""
Use an item from inventory.
Returns: {success, message, effects}
"""
player = await db.get_player_by_id(player_id)
if not player:
return {"success": False, "message": "Player not found"}
# Check if player has the item
inventory = await db.get_inventory(player_id)
item_entry = None
for inv_item in inventory:
if inv_item['item_id'] == item_id:
item_entry = inv_item
break
if not item_entry:
return {"success": False, "message": "You don't have this item"}
# Get item data
item = items_manager.get_item(item_id)
if not item:
return {"success": False, "message": "Item not found in game data"}
if not item.consumable:
return {"success": False, "message": "This item cannot be used"}
# Apply item effects
effects = {}
effects_msg = []
if 'hp_restore' in item.effects:
hp_restore = item.effects['hp_restore']
old_hp = player['hp']
new_hp = min(player['max_hp'], old_hp + hp_restore)
actual_restored = new_hp - old_hp
if actual_restored > 0:
await db.update_player_hp(player_id, new_hp)
effects['hp_restored'] = actual_restored
effects_msg.append(f"+{actual_restored} HP")
if 'stamina_restore' in item.effects:
stamina_restore = item.effects['stamina_restore']
old_stamina = player['stamina']
new_stamina = min(player['max_stamina'], old_stamina + stamina_restore)
actual_restored = new_stamina - old_stamina
if actual_restored > 0:
await db.update_player_stamina(player_id, new_stamina)
effects['stamina_restored'] = actual_restored
effects_msg.append(f"+{actual_restored} Stamina")
# Consume the item (remove 1 from inventory)
await db.remove_item_from_inventory(player_id, item_id, 1)
# Track statistics
stat_updates = {"items_used": 1, "increment": True}
if 'hp_restored' in effects:
stat_updates['hp_restored'] = effects['hp_restored']
if 'stamina_restored' in effects:
stat_updates['stamina_restored'] = effects['stamina_restored']
await db.update_player_statistics(player_id, **stat_updates)
# Build message
msg = f"Used {item.name}"
if effects_msg:
msg += f" ({', '.join(effects_msg)})"
return {
"success": True,
"message": msg,
"effects": effects
}
async def pickup_item(player_id: int, item_id: int, location_id: str, quantity: int = None, items_manager=None) -> Dict[str, Any]:
"""
Pick up an item from the ground.
item_id is the dropped_item id, not the item_id field.
quantity: how many to pick up (None = all)
items_manager: ItemsManager instance to get item definitions
Returns: {success, message}
"""
# Get the dropped item by its ID
dropped_item = await db.get_dropped_item(item_id)
if not dropped_item:
return {"success": False, "message": "Item not found on ground"}
# Get item definition
item_def = items_manager.get_item(dropped_item['item_id']) if items_manager else None
if not item_def:
return {"success": False, "message": "Item data not found"}
# Determine how many to pick up
available_qty = dropped_item['quantity']
if quantity is None or quantity >= available_qty:
pickup_qty = available_qty
else:
if quantity < 1:
return {"success": False, "message": "Invalid quantity"}
pickup_qty = quantity
# Get player and calculate capacity
player = await db.get_player_by_id(player_id)
inventory = await db.get_inventory(player_id)
# Calculate current weight and volume (including equipped bag capacity)
current_weight = 0.0
current_volume = 0.0
max_weight = 10.0 # Base capacity
max_volume = 10.0 # Base capacity
for inv_item in inventory:
inv_item_def = items_manager.get_item(inv_item['item_id']) if items_manager else None
if inv_item_def:
current_weight += inv_item_def.weight * inv_item['quantity']
current_volume += inv_item_def.volume * inv_item['quantity']
# Check for equipped bags/containers that increase capacity
if inv_item['is_equipped'] and inv_item_def.stats:
max_weight += inv_item_def.stats.get('weight_capacity', 0)
max_volume += inv_item_def.stats.get('volume_capacity', 0)
# Calculate weight and volume for items to pick up
item_weight = item_def.weight * pickup_qty
item_volume = item_def.volume * pickup_qty
new_weight = current_weight + item_weight
new_volume = current_volume + item_volume
# Check limits
if new_weight > max_weight:
return {
"success": False,
"message": f"⚠️ Item too heavy! {item_def.emoji} {item_def.name} x{pickup_qty} ({item_weight:.1f}kg) would exceed capacity. Current: {current_weight:.1f}/{max_weight:.1f}kg"
}
if new_volume > max_volume:
return {
"success": False,
"message": f"⚠️ Item too large! {item_def.emoji} {item_def.name} x{pickup_qty} ({item_volume:.1f}L) would exceed capacity. Current: {current_volume:.1f}/{max_volume:.1f}L"
}
# Items fit - update dropped item quantity or remove it
if pickup_qty >= available_qty:
await db.remove_dropped_item(item_id)
else:
new_qty = available_qty - pickup_qty
await db.update_dropped_item_quantity(item_id, new_qty)
# Add to inventory (pass unique_item_id if it's a unique item)
await db.add_item_to_inventory(
player_id,
dropped_item['item_id'],
pickup_qty,
unique_item_id=dropped_item.get('unique_item_id')
)
return {
"success": True,
"message": f"Picked up {item_def.emoji} {item_def.name} x{pickup_qty}"
}
async def check_and_apply_level_up(player_id: int) -> Dict[str, Any]:
"""
Check if player has enough XP to level up and apply it.
Returns: {leveled_up: bool, new_level: int, levels_gained: int}
"""
player = await db.get_player_by_id(player_id)
if not player:
return {"leveled_up": False, "new_level": 1, "levels_gained": 0}
current_level = player['level']
current_xp = player['xp']
levels_gained = 0
# Check for level ups (can level up multiple times if enough XP)
while current_xp >= (current_level * 100):
current_xp -= (current_level * 100)
current_level += 1
levels_gained += 1
if levels_gained > 0:
# Update player with new level, remaining XP, and unspent points
new_unspent_points = player['unspent_points'] + levels_gained
await db.update_player(
player_id,
level=current_level,
xp=current_xp,
unspent_points=new_unspent_points
)
return {
"leveled_up": True,
"new_level": current_level,
"levels_gained": levels_gained
}
return {"leveled_up": False, "new_level": current_level, "levels_gained": 0}
# ============================================================================
# STATUS EFFECTS UTILITIES
# ============================================================================
def calculate_status_damage(effects: list) -> int:
"""
Calculate total damage from all status effects.
Args:
effects: List of status effect dicts
Returns:
Total damage per tick
"""
return sum(effect.get('damage_per_tick', 0) for effect in effects)

283
api/internal.old.py Normal file
View File

@@ -0,0 +1,283 @@
"""
Internal API endpoints for Telegram Bot
These endpoints are protected by an internal key and handle game logic
"""
from fastapi import APIRouter, Header, HTTPException, Depends
from pydantic import BaseModel
from typing import Optional, Dict, Any, List
import os
# Internal API key for bot authentication
INTERNAL_API_KEY = os.getenv("API_INTERNAL_KEY", "internal-bot-access-key-change-me")
router = APIRouter(prefix="/api/internal", tags=["internal"])
def verify_internal_key(x_internal_key: str = Header(...)):
"""Verify internal API key"""
if x_internal_key != INTERNAL_API_KEY:
raise HTTPException(status_code=403, detail="Invalid internal API key")
return True
# ==================== Pydantic Models ====================
class PlayerCreate(BaseModel):
telegram_id: int
name: str = "Survivor"
class PlayerUpdate(BaseModel):
name: Optional[str] = None
hp: Optional[int] = None
stamina: Optional[int] = None
location_id: Optional[str] = None
level: Optional[int] = None
xp: Optional[int] = None
strength: Optional[int] = None
agility: Optional[int] = None
endurance: Optional[int] = None
intellect: Optional[int] = None
class MoveRequest(BaseModel):
direction: str
class CombatStart(BaseModel):
telegram_id: int
npc_id: str
class CombatAction(BaseModel):
action: str # "attack", "defend", "flee"
class UseItem(BaseModel):
item_db_id: int
class EquipItem(BaseModel):
item_db_id: int
# ==================== Player Endpoints ====================
@router.get("/player/telegram/{telegram_id}")
async def get_player_by_telegram(
telegram_id: int,
_: bool = Depends(verify_internal_key)
):
"""Get player by Telegram ID"""
from bot.database import get_player
player = await get_player(telegram_id=telegram_id)
if not player:
raise HTTPException(status_code=404, detail="Player not found")
return player
@router.post("/player")
async def create_player_internal(
player_data: PlayerCreate,
_: bool = Depends(verify_internal_key)
):
"""Create a new player (Telegram bot)"""
from bot.database import create_player
player = await create_player(telegram_id=player_data.telegram_id, name=player_data.name)
if not player:
raise HTTPException(status_code=500, detail="Failed to create player")
return player
@router.patch("/player/telegram/{telegram_id}")
async def update_player_internal(
telegram_id: int,
updates: PlayerUpdate,
_: bool = Depends(verify_internal_key)
):
"""Update player data"""
from bot.database import update_player
# Convert to dict and remove None values
update_dict = {k: v for k, v in updates.dict().items() if v is not None}
if not update_dict:
return {"success": True, "message": "No updates provided"}
await update_player(telegram_id=telegram_id, updates=update_dict)
return {"success": True, "message": "Player updated"}
# ==================== Location Endpoints ====================
@router.get("/location/{location_id}")
async def get_location_internal(
location_id: str,
_: bool = Depends(verify_internal_key)
):
"""Get location details"""
from api.main import LOCATIONS
location = LOCATIONS.get(location_id)
if not location:
raise HTTPException(status_code=404, detail="Location not found")
return {
"id": location.id,
"name": location.name,
"description": location.description,
"exits": location.exits,
"interactables": {k: {
"id": v.id,
"name": v.name,
"actions": list(v.actions.keys())
} for k, v in location.interactables.items()},
"image_path": location.image_path
}
@router.post("/player/telegram/{telegram_id}/move")
async def move_player_internal(
telegram_id: int,
move_data: MoveRequest,
_: bool = Depends(verify_internal_key)
):
"""Move player in a direction"""
from bot.database import get_player, update_player
from api.main import LOCATIONS
player = await get_player(telegram_id=telegram_id)
if not player:
raise HTTPException(status_code=404, detail="Player not found")
current_location = LOCATIONS.get(player['location_id'])
if not current_location:
raise HTTPException(status_code=400, detail="Invalid current location")
# Check stamina
if player['stamina'] < 1:
raise HTTPException(status_code=400, detail="Not enough stamina to move")
# Find exit
destination_id = current_location.exits.get(move_data.direction.lower())
if not destination_id:
raise HTTPException(status_code=400, detail=f"Cannot move {move_data.direction} from here")
new_location = LOCATIONS.get(destination_id)
if not new_location:
raise HTTPException(status_code=400, detail="Invalid destination")
# Update player
await update_player(telegram_id=telegram_id, updates={
'location_id': new_location.id,
'stamina': max(0, player['stamina'] - 1)
})
return {
"success": True,
"location": {
"id": new_location.id,
"name": new_location.name,
"description": new_location.description,
"exits": new_location.exits
},
"stamina": max(0, player['stamina'] - 1)
}
# ==================== Inventory Endpoints ====================
@router.get("/player/telegram/{telegram_id}/inventory")
async def get_inventory_internal(
telegram_id: int,
_: bool = Depends(verify_internal_key)
):
"""Get player's inventory"""
from bot.database import get_inventory
inventory = await get_inventory(telegram_id)
return {"items": inventory}
@router.post("/player/telegram/{telegram_id}/use_item")
async def use_item_internal(
telegram_id: int,
item_data: UseItem,
_: bool = Depends(verify_internal_key)
):
"""Use an item from inventory"""
from bot.logic import use_item_logic
from bot.database import get_player
player = await get_player(telegram_id=telegram_id)
if not player:
raise HTTPException(status_code=404, detail="Player not found")
result = await use_item_logic(player, item_data.item_db_id)
return result
@router.post("/player/telegram/{telegram_id}/equip")
async def equip_item_internal(
telegram_id: int,
item_data: EquipItem,
_: bool = Depends(verify_internal_key)
):
"""Equip/unequip an item"""
from bot.logic import toggle_equip
result = await toggle_equip(telegram_id, item_data.item_db_id)
return {"success": True, "message": result}
# ==================== Combat Endpoints ====================
@router.post("/combat/start")
async def start_combat_internal(
combat_data: CombatStart,
_: bool = Depends(verify_internal_key)
):
"""Start combat with an NPC"""
from bot.combat import start_combat
from bot.database import get_player
player = await get_player(telegram_id=combat_data.telegram_id)
if not player:
raise HTTPException(status_code=404, detail="Player not found")
result = await start_combat(combat_data.telegram_id, combat_data.npc_id, player['location_id'])
if not result.get("success"):
raise HTTPException(status_code=400, detail=result.get("message", "Failed to start combat"))
return result
@router.get("/combat/telegram/{telegram_id}")
async def get_combat_internal(
telegram_id: int,
_: bool = Depends(verify_internal_key)
):
"""Get active combat state"""
from bot.combat import get_active_combat
combat = await get_active_combat(telegram_id)
if not combat:
raise HTTPException(status_code=404, detail="No active combat")
return combat
@router.post("/combat/telegram/{telegram_id}/action")
async def combat_action_internal(
telegram_id: int,
action_data: CombatAction,
_: bool = Depends(verify_internal_key)
):
"""Perform combat action"""
from bot.combat import player_attack, player_defend, player_flee
if action_data.action == "attack":
result = await player_attack(telegram_id)
elif action_data.action == "defend":
result = await player_defend(telegram_id)
elif action_data.action == "flee":
result = await player_flee(telegram_id)
else:
raise HTTPException(status_code=400, detail="Invalid combat action")
return result

157
api/items.py Normal file
View File

@@ -0,0 +1,157 @@
"""
Standalone items module for the API.
Loads and manages game items from JSON without bot dependencies.
"""
import json
from pathlib import Path
from typing import Dict, Any, Optional
from dataclasses import dataclass
@dataclass
class Item:
"""Represents a game item"""
id: str
name: str
description: str
type: str
image_path: str = ""
emoji: str = "📦"
stackable: bool = True
equippable: bool = False
consumable: bool = False
weight: float = 0.0
volume: float = 0.0
stats: Dict[str, int] = None
effects: Dict[str, Any] = None
# Equipment system
slot: str = None # Equipment slot: head, torso, legs, feet, weapon, offhand, backpack
durability: int = None # Max durability for equippable items
tier: int = 1 # Item tier (1-5)
encumbrance: int = 0 # Encumbrance penalty when equipped
weapon_effects: Dict[str, Any] = None # Weapon effects: bleeding, stun, etc.
# Repair system
repairable: bool = False # Can this item be repaired?
repair_materials: list = None # Materials needed for repair
repair_percentage: int = 25 # Percentage of durability restored per repair
repair_tools: list = None # Tools required for repair (consumed durability)
# Crafting system
craftable: bool = False # Can this item be crafted?
craft_materials: list = None # Materials needed to craft this item
craft_level: int = 1 # Minimum level required to craft this item
craft_tools: list = None # Tools required for crafting (consumed durability)
# Uncrafting system
uncraftable: bool = False # Can this item be uncrafted?
uncraft_yield: list = None # Materials yielded from uncrafting (before loss chance)
uncraft_loss_chance: float = 0.3 # Chance to lose materials when uncrafting (0.3 = 30%)
uncraft_tools: list = None # Tools required for uncrafting
def __post_init__(self):
if self.stats is None:
self.stats = {}
if self.effects is None:
self.effects = {}
if self.weapon_effects is None:
self.weapon_effects = {}
if self.repair_materials is None:
self.repair_materials = []
if self.craft_materials is None:
self.craft_materials = []
if self.repair_tools is None:
self.repair_tools = []
if self.craft_tools is None:
self.craft_tools = []
if self.uncraft_yield is None:
self.uncraft_yield = []
if self.uncraft_tools is None:
self.uncraft_tools = []
self.craft_materials = []
class ItemsManager:
"""Manages all game items"""
def __init__(self, gamedata_path: str = "./gamedata"):
self.gamedata_path = Path(gamedata_path)
self.items: Dict[str, Item] = {}
self.load_items()
def load_items(self):
"""Load all items from items.json"""
json_path = self.gamedata_path / 'items.json'
try:
with open(json_path, 'r') as f:
data = json.load(f)
for item_id, item_data in data.get('items', {}).items():
item_type = item_data.get('type', 'misc')
# Automatically mark as consumable if type is consumable
is_consumable = item_data.get('consumable', item_type == 'consumable')
# Collect effects from root level or effects dict
effects = item_data.get('effects', {}).copy()
# Add common consumable effects if they exist at root level
if 'hp_restore' in item_data:
effects['hp_restore'] = item_data['hp_restore']
if 'stamina_restore' in item_data:
effects['stamina_restore'] = item_data['stamina_restore']
if 'treats' in item_data:
effects['treats'] = item_data['treats']
item = Item(
id=item_id,
name=item_data.get('name', 'Unknown Item'),
description=item_data.get('description', ''),
type=item_type,
image_path=item_data.get('image_path', ''),
emoji=item_data.get('emoji', '📦'),
stackable=item_data.get('stackable', True),
equippable=item_data.get('equippable', False),
consumable=is_consumable,
weight=item_data.get('weight', 0.0),
volume=item_data.get('volume', 0.0),
stats=item_data.get('stats', {}),
effects=effects,
slot=item_data.get('slot'),
durability=item_data.get('durability'),
tier=item_data.get('tier', 1),
encumbrance=item_data.get('encumbrance', 0),
weapon_effects=item_data.get('weapon_effects', {}),
repairable=item_data.get('repairable', False),
repair_materials=item_data.get('repair_materials', []),
repair_percentage=item_data.get('repair_percentage', 25),
repair_tools=item_data.get('repair_tools', []),
craftable=item_data.get('craftable', False),
craft_materials=item_data.get('craft_materials', []),
craft_level=item_data.get('craft_level', 1),
craft_tools=item_data.get('craft_tools', []),
uncraftable=item_data.get('uncraftable', False),
uncraft_yield=item_data.get('uncraft_yield', []),
uncraft_loss_chance=item_data.get('uncraft_loss_chance', 0.3),
uncraft_tools=item_data.get('uncraft_tools', [])
)
self.items[item_id] = item
print(f"📦 Loaded {len(self.items)} items")
except FileNotFoundError:
print("⚠️ items.json not found")
except Exception as e:
print(f"⚠️ Error loading items.json: {e}")
def get_item(self, item_id: str) -> Optional[Item]:
"""Get an item by ID"""
return self.items.get(item_id)
def get_all_items(self) -> Dict[str, Item]:
"""Get all items"""
return self.items
# Global items manager instance
items_manager = ItemsManager()
def get_item(item_id: str) -> Optional[Item]:
"""Convenience function to get an item"""
return items_manager.get_item(item_id)

499
api/main.old.py Normal file
View File

@@ -0,0 +1,499 @@
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
from typing import Optional, List
import jwt
import bcrypt
from datetime import datetime, timedelta
import os
import sys
# Add parent directory to path to import bot modules
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from bot.database import get_player, create_player
from data.world_loader import load_world
from api.internal import router as internal_router
app = FastAPI(title="Echoes of the Ashes API", version="1.0.0")
# Include internal API router
app.include_router(internal_router)
# CORS configuration
app.add_middleware(
CORSMiddleware,
allow_origins=["https://echoesoftheashgame.patacuack.net", "http://localhost:3000"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# JWT Configuration
SECRET_KEY = os.getenv("JWT_SECRET_KEY", "your-secret-key-change-in-production")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 * 7 # 7 days
security = HTTPBearer()
# Load world data
WORLD = None
LOCATIONS = {}
try:
WORLD = load_world()
# WORLD.locations is already a dict {location_id: Location}
LOCATIONS = WORLD.locations
print(f"✅ Loaded {len(LOCATIONS)} locations")
except Exception as e:
print(f"⚠️ Warning: Could not load world data: {e}")
import traceback
traceback.print_exc()
# Pydantic Models
class UserRegister(BaseModel):
username: str
password: str
class UserLogin(BaseModel):
username: str
password: str
class Token(BaseModel):
access_token: str
token_type: str = "bearer"
class User(BaseModel):
id: int
username: str
telegram_id: Optional[str] = None
class PlayerState(BaseModel):
location_id: str
location_name: str
health: int
max_health: int
stamina: int
max_stamina: int
inventory: List[dict]
status_effects: List[dict]
class MoveRequest(BaseModel):
direction: str
# Helper Functions
def create_access_token(data: dict):
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
try:
token = credentials.credentials
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
user_id: int = payload.get("sub")
if user_id is None:
raise HTTPException(status_code=401, detail="Invalid authentication credentials")
return user_id
except jwt.ExpiredSignatureError:
raise HTTPException(status_code=401, detail="Token has expired")
except jwt.JWTError:
raise HTTPException(status_code=401, detail="Could not validate credentials")
# Routes
@app.get("/")
async def root():
return {"message": "Echoes of the Ashes API", "status": "online"}
@app.post("/api/auth/register", response_model=Token)
async def register(user_data: UserRegister):
"""Register a new user account"""
try:
# Check if username already exists
existing_player = await get_player(username=user_data.username)
if existing_player:
raise HTTPException(status_code=400, detail="Username already exists")
# Hash password
password_hash = bcrypt.hashpw(user_data.password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
# Create player with web auth
player = await create_player(
telegram_id=None,
username=user_data.username,
password_hash=password_hash
)
if not player or 'id' not in player:
print(f"ERROR: create_player returned: {player}")
raise HTTPException(status_code=500, detail="Failed to create player - no ID returned")
# Create token
access_token = create_access_token(data={"sub": player['id']})
return {"access_token": access_token}
except HTTPException:
raise
except Exception as e:
import traceback
print(f"ERROR in register: {str(e)}")
print(traceback.format_exc())
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/auth/login", response_model=Token)
async def login(user_data: UserLogin):
"""Login with username and password"""
try:
# Get player
player = await get_player(username=user_data.username)
if not player or not player.get('password_hash'):
raise HTTPException(status_code=401, detail="Invalid username or password")
# Verify password
if not bcrypt.checkpw(user_data.password.encode('utf-8'), player['password_hash'].encode('utf-8')):
raise HTTPException(status_code=401, detail="Invalid username or password")
# Create token
access_token = create_access_token(data={"sub": player['id']})
return {"access_token": access_token}
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/auth/me", response_model=User)
async def get_current_user(user_id: int = Depends(verify_token)):
"""Get current authenticated user"""
player = await get_player(player_id=user_id)
if not player:
raise HTTPException(status_code=404, detail="User not found")
return {
"id": player['id'],
"username": player.get('username'),
"telegram_id": player.get('telegram_id')
}
@app.get("/api/game/state", response_model=PlayerState)
async def get_game_state(user_id: int = Depends(verify_token)):
"""Get current player game state"""
player = await get_player(player_id=user_id)
if not player:
raise HTTPException(status_code=404, detail="Player not found")
location = LOCATIONS.get(player['location_id'])
# TODO: Get actual inventory and status effects from database
inventory = []
status_effects = []
return {
"location_id": player['location_id'],
"location_name": location.name if location else "Unknown",
"health": player['hp'],
"max_health": player['max_hp'],
"stamina": player['stamina'],
"max_stamina": player['max_stamina'],
"inventory": inventory,
"status_effects": status_effects
}
@app.post("/api/game/move")
async def move_player(move_data: MoveRequest, user_id: int = Depends(verify_token)):
"""Move player in a direction"""
from bot.database import update_player
player = await get_player(player_id=user_id)
if not player:
raise HTTPException(status_code=404, detail="Player not found")
current_location = LOCATIONS.get(player['location_id'])
if not current_location:
raise HTTPException(status_code=400, detail="Invalid current location")
# Check if player has enough stamina
if player['stamina'] < 1:
raise HTTPException(status_code=400, detail="Not enough stamina to move")
# Find exit in the specified direction (exits is dict {direction: destination_id})
destination_id = current_location.exits.get(move_data.direction.lower())
if not destination_id:
raise HTTPException(status_code=400, detail=f"Cannot move {move_data.direction} from here")
# Move player
new_location = LOCATIONS.get(destination_id)
if not new_location:
raise HTTPException(status_code=400, detail="Invalid destination")
# Update player location and stamina (use player_id for web users)
await update_player(player_id=player['id'], updates={
'location_id': new_location.id,
'stamina': max(0, player['stamina'] - 1)
})
# Get updated player state
updated_player = await get_player(player_id=user_id)
return {
"success": True,
"message": f"You travel {move_data.direction} to {new_location.name}. {new_location.description}",
"player_state": {
"location_id": updated_player['location_id'],
"location_name": new_location.name,
"health": updated_player['hp'],
"max_health": updated_player['max_hp'],
"stamina": updated_player['stamina'],
"max_stamina": updated_player['max_stamina'],
"inventory": [],
"status_effects": []
}
}
@app.get("/api/game/location")
async def get_current_location(user_id: int = Depends(verify_token)):
"""Get detailed information about current location"""
player = await get_player(player_id=user_id)
if not player:
raise HTTPException(status_code=404, detail="Player not found")
location = LOCATIONS.get(player['location_id'])
if not location:
raise HTTPException(status_code=404, detail=f"Location '{player['location_id']}' not found")
# Get available directions from exits dict
directions = list(location.exits.keys())
# Get NPCs at location (TODO: implement NPC spawning)
npcs = []
# Get items at location (TODO: implement dropped items)
items = []
# Determine image extension (png or jpg)
image_url = None
if location.image_path:
# Use the path from location data
image_url = f"/{location.image_path}"
else:
# Default to png with fallback to jpg
image_url = f"/images/locations/{location.id}.png"
return {
"id": location.id,
"name": location.name,
"description": location.description,
"directions": directions,
"npcs": npcs,
"items": items,
"image_url": image_url,
"interactables": [{"id": k, "name": v.name} for k, v in location.interactables.items()]
}
@app.get("/api/game/inventory")
async def get_inventory(user_id: int = Depends(verify_token)):
"""Get player's inventory"""
from bot.database import get_inventory
player = await get_player(player_id=user_id)
if not player:
raise HTTPException(status_code=404, detail="Player not found")
# For web users without telegram_id, inventory might be empty
# This is a limitation of the current schema
inventory = []
return {
"items": inventory,
"capacity": 20 # TODO: Calculate based on equipped bag
}
@app.get("/api/game/profile")
async def get_profile(user_id: int = Depends(verify_token)):
"""Get player profile and stats"""
player = await get_player(player_id=user_id)
if not player:
raise HTTPException(status_code=404, detail="Player not found")
return {
"name": player['name'],
"level": player['level'],
"xp": player['xp'],
"hp": player['hp'],
"max_hp": player['max_hp'],
"stamina": player['stamina'],
"max_stamina": player['max_stamina'],
"strength": player['strength'],
"agility": player['agility'],
"endurance": player['endurance'],
"intellect": player['intellect'],
"unspent_points": player['unspent_points'],
"is_dead": player['is_dead']
}
@app.get("/api/game/map")
async def get_map(user_id: int = Depends(verify_token)):
"""Get world map data"""
player = await get_player(player_id=user_id)
if not player:
raise HTTPException(status_code=404, detail="Player not found")
# Return all locations and connections (LOCATIONS is dict {id: Location})
locations_data = []
for loc_id, loc in LOCATIONS.items():
locations_data.append({
"id": loc.id,
"name": loc.name,
"description": loc.description,
"exits": loc.exits # Dict of {direction: destination_id}
})
return {
"current_location": player['location_id'],
"locations": locations_data
}
@app.post("/api/game/inspect")
async def inspect_area(user_id: int = Depends(verify_token)):
"""Inspect the current area for details"""
player = await get_player(player_id=user_id)
if not player:
raise HTTPException(status_code=404, detail="Player not found")
location = LOCATIONS.get(player['location_id'])
if not location:
raise HTTPException(status_code=404, detail="Location not found")
# Get detailed information
interactables_detail = []
for inst_id, inter in location.interactables.items():
actions = [{"id": act.id, "label": act.label, "stamina_cost": act.stamina_cost}
for act in inter.actions.values()]
interactables_detail.append({
"instance_id": inst_id,
"name": inter.name,
"actions": actions
})
return {
"location": location.name,
"description": location.description,
"interactables": interactables_detail,
"exits": location.exits
}
class InteractRequest(BaseModel):
interactable_id: str
action_id: str
@app.post("/api/game/interact")
async def interact_with_object(interact_data: InteractRequest, user_id: int = Depends(verify_token)):
"""Interact with an object in the world"""
from bot.database import update_player, add_inventory_item
import random
player = await get_player(player_id=user_id)
if not player:
raise HTTPException(status_code=404, detail="Player not found")
location = LOCATIONS.get(player['location_id'])
if not location:
raise HTTPException(status_code=404, detail="Location not found")
interactable = location.interactables.get(interact_data.interactable_id)
if not interactable:
raise HTTPException(status_code=404, detail="Interactable not found")
action = interactable.actions.get(interact_data.action_id)
if not action:
raise HTTPException(status_code=404, detail="Action not found")
# Check stamina
if player['stamina'] < action.stamina_cost:
raise HTTPException(status_code=400, detail="Not enough stamina")
# Perform action - randomly choose outcome
outcome_key = random.choice(list(action.outcomes.keys()))
outcome = action.outcomes[outcome_key]
# Apply outcome
stamina_change = -action.stamina_cost
hp_change = -outcome.damage_taken if outcome.damage_taken else 0
items_found = outcome.items_reward if outcome.items_reward else {}
# Update player
new_hp = max(1, player['hp'] + hp_change)
new_stamina = max(0, player['stamina'] + stamina_change)
await update_player(player_id=player['id'], updates={
'hp': new_hp,
'stamina': new_stamina
})
# Add items to inventory (if player has telegram_id for FK)
items_added = []
if player.get('telegram_id') and items_found:
for item_id, quantity in items_found.items():
# This will fail for web users without telegram_id
# TODO: Fix inventory schema
try:
items_added.append({"id": item_id, "quantity": quantity})
except:
pass
return {
"success": True,
"outcome": outcome_key,
"message": outcome.text,
"items_found": items_added,
"hp_change": hp_change,
"stamina_change": stamina_change,
"new_hp": new_hp,
"new_stamina": new_stamina
}
class UseItemRequest(BaseModel):
item_db_id: int
@app.post("/api/game/use_item")
async def use_item_endpoint(item_data: UseItemRequest, user_id: int = Depends(verify_token)):
"""Use an item from inventory"""
from bot.logic import use_item_logic
player = await get_player(player_id=user_id)
if not player:
raise HTTPException(status_code=404, detail="Player not found")
if not player.get('telegram_id'):
raise HTTPException(status_code=400, detail="Inventory not available for web users yet")
result = await use_item_logic(player, item_data.item_db_id)
return result
class EquipItemRequest(BaseModel):
item_db_id: int
@app.post("/api/game/equip_item")
async def equip_item_endpoint(item_data: EquipItemRequest, user_id: int = Depends(verify_token)):
"""Equip or unequip an item"""
from bot.logic import toggle_equip
player = await get_player(player_id=user_id)
if not player:
raise HTTPException(status_code=404, detail="Player not found")
if not player.get('telegram_id'):
raise HTTPException(status_code=400, detail="Inventory not available for web users yet")
result = await toggle_equip(player['telegram_id'], item_data.item_db_id)
return {"success": True, "message": result}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)

4239
api/main.py Normal file
View File

File diff suppressed because it is too large Load Diff

6
api/requirements.old.txt Normal file
View File

@@ -0,0 +1,6 @@
fastapi==0.104.1
uvicorn[standard]==0.24.0
pyjwt==2.8.0
bcrypt==4.1.1
pydantic==2.5.2
python-multipart==0.0.6

19
api/requirements.txt Normal file
View File

@@ -0,0 +1,19 @@
# FastAPI and server
fastapi==0.104.1
uvicorn[standard]==0.24.0
gunicorn==21.2.0
python-multipart==0.0.6
# Database
sqlalchemy==2.0.23
psycopg[binary]==3.1.13
# Authentication
pyjwt==2.8.0
bcrypt==4.1.1
# Utilities
aiofiles==23.2.1
# Testing
httpx==0.25.2

28
api/start.sh Normal file
View File

@@ -0,0 +1,28 @@
#!/bin/bash
# Startup script for API with auto-scaling workers
# Detect number of CPU cores
CPU_CORES=$(nproc)
# Calculate optimal workers: (2 x CPU cores) + 1
# But cap at 8 workers to avoid over-saturation
WORKERS=$((2 * CPU_CORES + 1))
if [ $WORKERS -gt 8 ]; then
WORKERS=8
fi
# Use environment variable if set, otherwise use calculated value
WORKERS=${API_WORKERS:-$WORKERS}
echo "Starting API with $WORKERS workers (detected $CPU_CORES CPU cores)"
exec gunicorn api.main:app \
--workers $WORKERS \
--worker-class uvicorn.workers.UvicornWorker \
--bind 0.0.0.0:8000 \
--timeout 120 \
--max-requests 1000 \
--max-requests-jitter 100 \
--access-logfile - \
--error-logfile - \
--log-level info

290
api/world_loader.py Normal file
View File

@@ -0,0 +1,290 @@
"""
Standalone world loader for the API.
Loads game data from JSON files without bot dependencies.
"""
import json
from pathlib import Path
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, field
@dataclass
class Outcome:
"""Represents an outcome of an action"""
text: str
items_reward: Dict[str, int] = field(default_factory=dict)
damage_taken: int = 0
@dataclass
class Action:
"""Represents an action that can be performed on an interactable"""
id: str
label: str
stamina_cost: int = 2
outcomes: Dict[str, Outcome] = field(default_factory=dict)
def add_outcome(self, outcome_type: str, outcome: Outcome):
self.outcomes[outcome_type] = outcome
@dataclass
class Interactable:
"""Represents an interactable object"""
id: str
name: str
image_path: str = ""
actions: List[Action] = field(default_factory=list)
def add_action(self, action: Action):
self.actions.append(action)
@dataclass
class Exit:
"""Represents an exit from a location"""
direction: str
destination: str
description: str = ""
@dataclass
class Location:
"""Represents a location in the game world"""
id: str
name: str
description: str
image_path: str = ""
exits: Dict[str, str] = field(default_factory=dict) # direction -> destination_id
exit_stamina: Dict[str, int] = field(default_factory=dict) # direction -> stamina_cost
interactables: List[Interactable] = field(default_factory=list)
npcs: List[str] = field(default_factory=list)
tags: List[str] = field(default_factory=list) # Location tags like 'workbench', 'safe_zone'
x: float = 0.0 # X coordinate for distance calculations
y: float = 0.0 # Y coordinate for distance calculations
danger_level: int = 0 # Danger level (0-5)
def add_exit(self, direction: str, destination: str, stamina_cost: int = 5):
self.exits[direction] = destination
self.exit_stamina[direction] = stamina_cost
def add_interactable(self, interactable: Interactable):
self.interactables.append(interactable)
@dataclass
class World:
"""Represents the entire game world"""
locations: Dict[str, Location] = field(default_factory=dict)
def add_location(self, location: Location):
self.locations[location.id] = location
class WorldLoader:
"""Loads world data from JSON files"""
def __init__(self, gamedata_path: str = "./gamedata"):
self.gamedata_path = Path(gamedata_path)
self.interactable_templates = {}
def load_interactable_templates(self) -> Dict[str, Any]:
"""Load interactable templates from interactables.json"""
json_path = self.gamedata_path / 'interactables.json'
try:
with open(json_path, 'r') as f:
data = json.load(f)
self.interactable_templates = data.get('interactables', {})
print(f"📦 Loaded {len(self.interactable_templates)} interactable templates")
except FileNotFoundError:
print("⚠️ interactables.json not found")
except Exception as e:
print(f"⚠️ Error loading interactables.json: {e}")
return self.interactable_templates
def create_interactable_from_template(
self,
template_id: str,
template_data: Dict[str, Any],
instance_data: Dict[str, Any]
) -> Interactable:
"""Create an Interactable object from template and instance data"""
interactable = Interactable(
id=template_id,
name=template_data.get('name', 'Unknown'),
image_path=template_data.get('image_path', '')
)
# Get actions from template
template_actions = template_data.get('actions', {})
# Get outcomes from instance
instance_outcomes = instance_data.get('outcomes', {})
# Build actions by merging template actions with instance outcomes
for action_id, action_template in template_actions.items():
action = Action(
id=action_template['id'],
label=action_template['label'],
stamina_cost=action_template.get('stamina_cost', 2)
)
# Get instance-specific outcome data for this action
if action_id in instance_outcomes:
outcome_data = instance_outcomes[action_id]
# Build outcomes from the instance data
text_dict = outcome_data.get('text', {})
rewards = outcome_data.get('rewards', {})
# Add success outcome
if text_dict.get('success'):
items_reward = {}
if 'items' in rewards:
for item in rewards['items']:
items_reward[item['item_id']] = item.get('quantity', 1)
outcome = Outcome(
text=text_dict['success'],
items_reward=items_reward,
damage_taken=rewards.get('damage', 0)
)
action.add_outcome('success', outcome)
# Add failure outcome
if text_dict.get('failure'):
outcome = Outcome(
text=text_dict['failure'],
items_reward={},
damage_taken=0
)
action.add_outcome('failure', outcome)
# Add critical failure outcome
if text_dict.get('crit_failure'):
outcome = Outcome(
text=text_dict['crit_failure'],
items_reward={},
damage_taken=rewards.get('crit_damage', 0)
)
action.add_outcome('critical_failure', outcome)
interactable.add_action(action)
return interactable
def load_locations(self) -> Dict[str, Location]:
"""Load all locations from locations.json"""
json_path = self.gamedata_path / 'locations.json'
locations = {}
try:
with open(json_path, 'r') as f:
data = json.load(f)
# Get danger config
danger_config = data.get('danger_config', {})
# First pass: create all locations
locations_data = data.get('locations', [])
if isinstance(locations_data, dict):
# Old format: dict of locations
locations_iter = locations_data.items()
else:
# New format: list of locations
locations_iter = [(loc['id'], loc) for loc in locations_data]
for loc_id, loc_data in locations_iter:
# Get danger level from danger_config
danger_level = 0
if loc_id in danger_config:
danger_level = danger_config[loc_id].get('danger_level', 0)
location = Location(
id=loc_id,
name=loc_data.get('name', 'Unknown Location'),
description=loc_data.get('description', ''),
image_path=loc_data.get('image_path', ''),
x=float(loc_data.get('x', 0.0)),
y=float(loc_data.get('y', 0.0)),
danger_level=danger_level,
tags=loc_data.get('tags', []),
npcs=loc_data.get('npcs', [])
)
# Add exits
for direction, destination in loc_data.get('exits', {}).items():
location.add_exit(direction, destination)
# Add NPCs
location.npcs = loc_data.get('npcs', [])
# Add interactables
interactables_data = loc_data.get('interactables', {})
if isinstance(interactables_data, dict):
# New format: dict of interactables
interactables_list = [
{**data, 'instance_id': inst_id, 'id': data.get('template_id', inst_id)}
for inst_id, data in interactables_data.items()
]
else:
# Old format: list of interactables
interactables_list = interactables_data
for interactable_data in interactables_list:
template_id = interactable_data.get('id')
instance_id = interactable_data.get('instance_id', template_id)
if template_id in self.interactable_templates:
template = self.interactable_templates[template_id]
interactable = self.create_interactable_from_template(
instance_id,
template,
interactable_data
)
location.add_interactable(interactable)
locations[loc_id] = location
# Second pass: add connections from the connections array
connections = data.get('connections', [])
for conn in connections:
from_id = conn.get('from')
to_id = conn.get('to')
direction = conn.get('direction')
stamina_cost = conn.get('stamina_cost', 5) # Default 5 if not specified
if from_id in locations and direction:
locations[from_id].add_exit(direction, to_id, stamina_cost)
print(f"🗺️ Loaded {len(locations)} locations with {len(connections)} connections")
except FileNotFoundError:
print("⚠️ locations.json not found")
except Exception as e:
print(f"⚠️ Error loading locations.json: {e}")
import traceback
traceback.print_exc()
return locations
def load_world(self) -> World:
"""Load the entire world"""
world = World()
# Load interactable templates first
self.load_interactable_templates()
# Load locations
locations = self.load_locations()
for location in locations.values():
world.add_location(location)
return world
def load_world() -> World:
"""Convenience function to load the world"""
loader = WorldLoader()
return loader.load_world()