418 lines
16 KiB
Python
418 lines
16 KiB
Python
"""
|
|
Action handlers for button callbacks.
|
|
This module contains organized handler functions for different types of player actions.
|
|
"""
|
|
import logging
|
|
import json
|
|
import random
|
|
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
|
|
from telegram.ext import ContextTypes
|
|
from . import keyboards, logic
|
|
from .api_client import api_client
|
|
from .utils import format_stat_bar
|
|
from data.world_loader import game_world
|
|
from data.items import ITEMS
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# ============================================================================
|
|
# UTILITY FUNCTIONS
|
|
# ============================================================================
|
|
|
|
async def check_and_redirect_if_in_combat(query, user_id: int, player: dict) -> bool:
|
|
"""
|
|
Check if player is in combat and redirect to combat view if so.
|
|
Returns True if player is in combat (and was redirected), False otherwise.
|
|
"""
|
|
combat_data = await api_client.get_combat(user_id)
|
|
if combat_data:
|
|
from data.npcs import NPCS
|
|
npc_def = NPCS.get(combat_data['npc_id'])
|
|
|
|
message = f"⚔️ You're in combat with {npc_def.emoji} {npc_def.name}!\n"
|
|
message += format_stat_bar("Your HP", "❤️", player['hp'], player['max_hp']) + "\n"
|
|
message += format_stat_bar("Enemy HP", npc_def.emoji, combat_data['npc_hp'], combat_data['npc_max_hp']) + "\n\n"
|
|
message += "🎯 Your turn!" if combat_data['turn'] == 'player' else "⏳ Enemy's turn..."
|
|
|
|
keyboard = await keyboards.combat_keyboard(user_id)
|
|
|
|
from .handlers import send_or_edit_with_image
|
|
await send_or_edit_with_image(
|
|
query,
|
|
text=message,
|
|
reply_markup=keyboard,
|
|
image_path=npc_def.image_url if npc_def else None
|
|
)
|
|
await query.answer("⚔️ You're in combat! Finish or flee first.", show_alert=True)
|
|
return True
|
|
return False
|
|
|
|
async def get_player_status_text(player_id: int) -> str:
|
|
"""Generate player status text with location and stats.
|
|
|
|
Args:
|
|
player_id: The unique database ID of the player (not telegram_id)
|
|
"""
|
|
from .api_client import api_client
|
|
|
|
player = await api_client.get_player_by_id(player_id)
|
|
if not player:
|
|
return "Could not find player data."
|
|
|
|
location = game_world.get_location(player["location_id"])
|
|
if not location:
|
|
return "Error: Player is in an unknown location."
|
|
|
|
# Get inventory from API
|
|
inv_result = await api_client.get_inventory(player_id)
|
|
inventory = inv_result.get('inventory', [])
|
|
weight, volume = logic.calculate_inventory_load(inventory)
|
|
max_weight, max_volume = logic.get_player_capacity(inventory, player)
|
|
|
|
# Get equipped items
|
|
equipped_items = []
|
|
for item in inventory:
|
|
if item.get('is_equipped'):
|
|
item_def = ITEMS.get(item['item_id'], {})
|
|
emoji = item_def.get('emoji', '❔')
|
|
equipped_items.append(f"{emoji} {item_def.get('name', 'Unknown')}")
|
|
|
|
# Build status with visual bars
|
|
status = f"<b>📍 Location:</b> {location.name}\n"
|
|
status += f"{format_stat_bar('HP', '❤️', player['hp'], player['max_hp'])}\n"
|
|
status += f"{format_stat_bar('Stamina', '⚡', player['stamina'], player['max_stamina'])}\n"
|
|
status += f"🎒 <b>Load:</b> {weight}/{max_weight} kg | {volume}/{max_volume} vol\n"
|
|
|
|
if equipped_items:
|
|
status += f"⚔️ <b>Equipped:</b> {', '.join(equipped_items)}\n"
|
|
|
|
status += "━━━━━━━━━━━━━━━━━━━━\n"
|
|
status += f"<i>{location.description}</i>"
|
|
return status
|
|
|
|
|
|
# ============================================================================
|
|
# INSPECTION & WORLD INTERACTION HANDLERS
|
|
# ============================================================================
|
|
|
|
async def handle_inspect_area(query, user_id: int, player: dict, data: list = None):
|
|
"""Handle inspect area action - show NPCs and interactables in current location."""
|
|
# Check if player is in combat and redirect if so
|
|
if await check_and_redirect_if_in_combat(query, user_id, player):
|
|
return
|
|
|
|
await query.answer()
|
|
location_id = player['location_id']
|
|
location = game_world.get_location(location_id)
|
|
dropped_items = await api_client.get_dropped_items_in_location(location_id)
|
|
wandering_enemies = await api_client.get_wandering_enemies_in_location(location_id)
|
|
|
|
keyboard = await keyboards.inspect_keyboard(location_id, dropped_items, wandering_enemies)
|
|
image_path = location.image_path if location else None
|
|
|
|
from .handlers import send_or_edit_with_image
|
|
await send_or_edit_with_image(
|
|
query,
|
|
text="You scan the area. You notice...",
|
|
reply_markup=keyboard,
|
|
image_path=image_path
|
|
)
|
|
|
|
|
|
async def handle_attack_wandering(query, user_id: int, player: dict, data: list):
|
|
"""Handle attacking a wandering enemy."""
|
|
enemy_db_id = int(data[1])
|
|
await query.answer()
|
|
|
|
# Get the enemy from database
|
|
wandering_enemies = await api_client.get_wandering_enemies_in_location(player['location_id'])
|
|
enemy_data = next((e for e in wandering_enemies if e['id'] == enemy_db_id), None)
|
|
|
|
if not enemy_data:
|
|
await query.answer("That enemy has already moved on!", show_alert=True)
|
|
# Refresh inspect menu
|
|
location_id = player['location_id']
|
|
location = game_world.get_location(location_id)
|
|
dropped_items = await api_client.get_dropped_items_in_location(location_id)
|
|
wandering_enemies = await api_client.get_wandering_enemies_in_location(location_id)
|
|
keyboard = await keyboards.inspect_keyboard(location_id, dropped_items, wandering_enemies)
|
|
image_path = location.image_path if location else None
|
|
|
|
from .handlers import send_or_edit_with_image
|
|
await send_or_edit_with_image(
|
|
query,
|
|
text="You scan the area. You notice...",
|
|
reply_markup=keyboard,
|
|
image_path=image_path
|
|
)
|
|
return
|
|
|
|
npc_id = enemy_data['npc_id']
|
|
|
|
# Remove enemy from wandering table (they're now in combat)
|
|
await api_client.remove_wandering_enemy(enemy_db_id)
|
|
|
|
from data.npcs import NPCS
|
|
from bot import combat
|
|
|
|
# Initiate combat
|
|
combat_data = await combat.initiate_combat(
|
|
user_id, npc_id, player['location_id'], from_wandering_enemy=True
|
|
)
|
|
|
|
if combat_data:
|
|
npc_def = NPCS.get(npc_id)
|
|
message = f"⚔️ You engage the {npc_def.emoji} {npc_def.name}!\n\n"
|
|
message += f"{npc_def.description}\n\n"
|
|
message += format_stat_bar("Your HP", "❤️", player['hp'], player['max_hp']) + "\n"
|
|
message += format_stat_bar("Enemy HP", npc_def.emoji, combat_data['npc_hp'], combat_data['npc_max_hp']) + "\n\n"
|
|
message += "🎯 Your turn! What will you do?"
|
|
|
|
keyboard = await keyboards.combat_keyboard(user_id)
|
|
|
|
from .handlers import send_or_edit_with_image
|
|
await send_or_edit_with_image(
|
|
query,
|
|
text=message,
|
|
reply_markup=keyboard,
|
|
image_path=npc_def.image_url if npc_def else None
|
|
)
|
|
else:
|
|
await query.answer("Failed to initiate combat.", show_alert=True)
|
|
|
|
|
|
async def handle_inspect_interactable(query, user_id: int, player: dict, data: list):
|
|
"""Handle inspecting an interactable object."""
|
|
# Check if player is in combat and redirect if so
|
|
if await check_and_redirect_if_in_combat(query, user_id, player):
|
|
return
|
|
|
|
location_id, instance_id = data[1], data[2]
|
|
|
|
location = game_world.get_location(location_id)
|
|
if not location:
|
|
await query.answer("Location not found.", show_alert=True)
|
|
return
|
|
|
|
interactable = location.get_interactable(instance_id)
|
|
if not interactable:
|
|
await query.answer("Object not found.", show_alert=False)
|
|
return
|
|
|
|
# Check if ALL actions are on cooldown
|
|
all_on_cooldown = True
|
|
for action_id in interactable.actions.keys():
|
|
cooldown_key = f"{instance_id}:{action_id}"
|
|
if await api_client.get_cooldown(cooldown_key) == 0:
|
|
all_on_cooldown = False
|
|
break
|
|
|
|
if all_on_cooldown and len(interactable.actions) > 0:
|
|
await query.answer(
|
|
f"The {interactable.name} has already been searched. Try again later.",
|
|
show_alert=False
|
|
)
|
|
return
|
|
|
|
# Show action menu
|
|
await query.answer()
|
|
image_path = interactable.image_path if interactable else None
|
|
|
|
from .handlers import send_or_edit_with_image
|
|
await send_or_edit_with_image(
|
|
query,
|
|
text=f"You focus on the {interactable.name}. What do you do?",
|
|
reply_markup=await keyboards.actions_keyboard(location_id, instance_id),
|
|
image_path=image_path
|
|
)
|
|
|
|
|
|
async def handle_action(query, user_id: int, player: dict, data: list):
|
|
"""Handle performing an action on an interactable object."""
|
|
# Check if player is in combat and redirect if so
|
|
if await check_and_redirect_if_in_combat(query, user_id, player):
|
|
return
|
|
|
|
location_id, instance_id, action_id = data[1], data[2], data[3]
|
|
cooldown_key = f"{instance_id}:{action_id}"
|
|
cooldown = await api_client.get_cooldown(cooldown_key)
|
|
|
|
if cooldown > 0:
|
|
await query.answer("Someone got to it just before you!", show_alert=False)
|
|
return
|
|
|
|
location = game_world.get_location(location_id)
|
|
if not location:
|
|
await query.answer("Location not found.", show_alert=True)
|
|
return
|
|
|
|
action_obj = location.get_interactable(instance_id).get_action(action_id)
|
|
|
|
if player['stamina'] < action_obj.stamina_cost:
|
|
await query.answer("You are too tired to do that!", show_alert=False)
|
|
return
|
|
|
|
await query.answer()
|
|
|
|
# Set cooldown
|
|
await api_client.set_cooldown(cooldown_key)
|
|
|
|
# Resolve action
|
|
outcome = logic.resolve_action(player, action_obj)
|
|
new_stamina = player['stamina'] - action_obj.stamina_cost
|
|
new_hp = player['hp'] - outcome.damage_taken
|
|
await api_client.update_player(user_id, {"stamina": new_stamina, "hp": new_hp})
|
|
|
|
# Build detailed action result
|
|
result_details = [f"<i>{outcome.text}</i>"]
|
|
|
|
if action_obj.stamina_cost > 0:
|
|
result_details.append(f"⚡️ <b>Stamina:</b> -{action_obj.stamina_cost}")
|
|
|
|
if outcome.damage_taken > 0:
|
|
result_details.append(f"❤️ <b>HP:</b> -{outcome.damage_taken}")
|
|
|
|
# Add items gained
|
|
if outcome.items_reward:
|
|
items_text = []
|
|
items_failed = []
|
|
for item_id, quantity in outcome.items_reward.items():
|
|
can_add, reason = await logic.can_add_item_to_inventory(user_id, item_id, quantity)
|
|
|
|
if can_add:
|
|
await api_client.add_item_to_inventory(user_id, item_id, quantity)
|
|
item_def = ITEMS.get(item_id, {})
|
|
emoji = item_def.get('emoji', '❔')
|
|
item_name = item_def.get('name', item_id)
|
|
items_text.append(f"{emoji} {item_name} x{quantity}")
|
|
else:
|
|
item_def = ITEMS.get(item_id, {})
|
|
item_name = item_def.get('name', item_id)
|
|
items_failed.append(f"{item_name} ({reason})")
|
|
|
|
if items_text:
|
|
result_details.append(f"🎁 <b>Gained:</b> {', '.join(items_text)}")
|
|
if items_failed:
|
|
result_details.append(f"⚠️ <b>Couldn't take:</b> {', '.join(items_failed)}")
|
|
|
|
final_text = await get_player_status_text(user_id)
|
|
final_text += f"\n\n<b>━━━ Action Result ━━━</b>\n" + "\n".join(result_details)
|
|
|
|
# Get location image for the result screen
|
|
current_location = game_world.get_location(player['location_id'])
|
|
location_image = current_location.image_path if current_location else None
|
|
|
|
from .handlers import send_or_edit_with_image
|
|
await send_or_edit_with_image(
|
|
query,
|
|
text=final_text,
|
|
reply_markup=keyboards.main_menu_keyboard(),
|
|
image_path=location_image
|
|
)
|
|
|
|
|
|
# ============================================================================
|
|
# NAVIGATION & MOVEMENT HANDLERS
|
|
# ============================================================================
|
|
|
|
async def handle_main_menu(query, user_id: int, player: dict, data: list = None):
|
|
"""Return to main menu."""
|
|
await query.answer()
|
|
status_text = await get_player_status_text(user_id)
|
|
location = game_world.get_location(player['location_id'])
|
|
location_image = location.image_path if location else None
|
|
|
|
from .handlers import send_or_edit_with_image
|
|
await send_or_edit_with_image(
|
|
query,
|
|
text=status_text,
|
|
reply_markup=keyboards.main_menu_keyboard(),
|
|
image_path=location_image
|
|
)
|
|
|
|
|
|
async def handle_move_menu(query, user_id: int, player: dict, data: list = None):
|
|
"""Show movement options menu."""
|
|
# Check if player is in combat and redirect if so
|
|
if await check_and_redirect_if_in_combat(query, user_id, player):
|
|
return
|
|
|
|
await query.answer()
|
|
location = game_world.get_location(player['location_id'])
|
|
location_image = location.image_path if location else None
|
|
|
|
from .handlers import send_or_edit_with_image
|
|
await send_or_edit_with_image(
|
|
query,
|
|
text="Where do you want to go?",
|
|
reply_markup=await keyboards.move_keyboard(player['location_id'], user_id),
|
|
image_path=location_image
|
|
)
|
|
|
|
|
|
async def handle_move(query, user_id: int, player: dict, data: list):
|
|
"""Handle player movement to a new location."""
|
|
# Check if player is in combat and redirect if so
|
|
if await check_and_redirect_if_in_combat(query, user_id, player):
|
|
return
|
|
|
|
destination_id = data[1]
|
|
|
|
# Use API to move player
|
|
from .api_client import api_client
|
|
result = await api_client.move_player(player['id'], destination_id)
|
|
|
|
if not result.get('success'):
|
|
await query.answer(result.get('message', 'Cannot move there!'), show_alert=True)
|
|
return
|
|
|
|
await query.answer(result.get('message', 'Moving...'), show_alert=False)
|
|
|
|
# Refresh player data from API using unique id
|
|
player = await api_client.get_player_by_id(user_id)
|
|
|
|
# Check for random NPC encounter
|
|
from data.npcs import NPCS, get_random_npc_for_location, get_location_encounter_rate
|
|
encounter_rate = get_location_encounter_rate(destination_id)
|
|
|
|
if random.random() < encounter_rate:
|
|
from bot import combat
|
|
logger.info(f"Encounter triggered at {destination_id} (rate: {encounter_rate})")
|
|
|
|
npc_id = get_random_npc_for_location(destination_id)
|
|
|
|
if npc_id:
|
|
combat_data = await combat.initiate_combat(user_id, npc_id, destination_id)
|
|
|
|
if combat_data:
|
|
npc_def = NPCS.get(npc_id)
|
|
message = f"⚠️ A {npc_def.emoji} {npc_def.name} appears!\n\n"
|
|
message += f"{npc_def.description}\n\n"
|
|
message += format_stat_bar("Your HP", "❤️", player['hp'], player['max_hp']) + "\n"
|
|
message += format_stat_bar("Enemy HP", npc_def.emoji, combat_data['npc_hp'], combat_data['npc_max_hp']) + "\n\n"
|
|
message += "🎯 Your turn! What will you do?"
|
|
|
|
keyboard = await keyboards.combat_keyboard(user_id)
|
|
|
|
from .handlers import send_or_edit_with_image
|
|
await send_or_edit_with_image(
|
|
query,
|
|
text=message,
|
|
reply_markup=keyboard,
|
|
image_path=npc_def.image_url if npc_def else None
|
|
)
|
|
return
|
|
|
|
status_text = await get_player_status_text(user_id)
|
|
new_location = game_world.get_location(destination_id)
|
|
location_image = new_location.image_path if new_location else None
|
|
|
|
from .handlers import send_or_edit_with_image
|
|
await send_or_edit_with_image(
|
|
query,
|
|
text=status_text,
|
|
reply_markup=keyboards.main_menu_keyboard(),
|
|
image_path=location_image
|
|
)
|