528 lines
19 KiB
Python
528 lines
19 KiB
Python
"""
|
||
Combat system logic for turn-based NPC encounters.
|
||
"""
|
||
|
||
import random
|
||
import json
|
||
import time
|
||
from typing import Dict, List, Tuple, Optional
|
||
from bot.api_client import api_client
|
||
from bot.utils import format_stat_bar
|
||
from data.npcs import NPCS, STATUS_EFFECTS
|
||
from data.items import ITEMS
|
||
|
||
|
||
# XP curve for leveling
|
||
def xp_for_level(level: int) -> int:
|
||
"""Calculate XP needed to reach a level."""
|
||
if level <= 1:
|
||
return 0 # Level 1 starts at 0 XP
|
||
return int(100 * (level ** 1.5))
|
||
|
||
|
||
async def calculate_player_damage(player: dict) -> int:
|
||
"""Calculate player's damage output based on stats and equipped weapon."""
|
||
base_damage = 5
|
||
strength_bonus = player['strength'] // 2
|
||
level_bonus = player['level']
|
||
|
||
# Check for equipped weapon
|
||
inventory = await api_client.get_inventory(player['telegram_id'])
|
||
weapon_damage = 0
|
||
|
||
for item in inventory:
|
||
if item.get('is_equipped'):
|
||
item_def = ITEMS.get(item['item_id'], {})
|
||
if item_def.get('type') == 'weapon':
|
||
# Get weapon damage range
|
||
damage_min = item_def.get('damage_min', 0)
|
||
damage_max = item_def.get('damage_max', 0)
|
||
weapon_damage = random.randint(damage_min, damage_max)
|
||
break
|
||
|
||
# Random variance
|
||
variance = random.randint(-2, 2)
|
||
|
||
return max(1, base_damage + strength_bonus + level_bonus + weapon_damage + variance)
|
||
|
||
|
||
def calculate_npc_damage(npc_def, npc_hp: int, npc_max_hp: int) -> int:
|
||
"""Calculate NPC's damage output."""
|
||
base_damage = random.randint(npc_def.damage_min, npc_def.damage_max)
|
||
|
||
# Enraged bonus if low HP
|
||
hp_percent = npc_hp / npc_max_hp
|
||
if hp_percent < 0.3:
|
||
base_damage = int(base_damage * 1.5)
|
||
|
||
return max(1, base_damage)
|
||
|
||
|
||
async def initiate_combat(player_id: int, npc_id: str, location_id: str, from_wandering_enemy: bool = False) -> Dict:
|
||
"""
|
||
Start a new combat encounter.
|
||
Args:
|
||
player_id: Telegram user ID
|
||
npc_id: NPC definition ID
|
||
location_id: Where combat is happening
|
||
from_wandering_enemy: If True, enemy will respawn if player flees or dies
|
||
Returns combat state dict.
|
||
"""
|
||
npc_def = NPCS.get(npc_id)
|
||
if not npc_def:
|
||
return None
|
||
|
||
# Randomize NPC HP
|
||
npc_hp = random.randint(npc_def.hp_min, npc_def.hp_max)
|
||
|
||
# Create combat in database
|
||
combat_id = await api_client.create_combat(
|
||
player_id=player_id,
|
||
npc_id=npc_id,
|
||
npc_hp=npc_hp,
|
||
npc_max_hp=npc_hp,
|
||
location_id=location_id,
|
||
from_wandering_enemy=from_wandering_enemy
|
||
)
|
||
|
||
return await api_client.get_combat(player_id)
|
||
|
||
|
||
async def player_attack(player_id: int) -> Tuple[str, bool, bool]:
|
||
"""
|
||
Player attacks the NPC.
|
||
Returns: (message, npc_died, player_turn_ended)
|
||
"""
|
||
combat = await api_client.get_combat(player_id)
|
||
if not combat or combat['turn'] != 'player':
|
||
return ("It's not your turn!", False, False)
|
||
|
||
player = await api_client.get_player(player_id)
|
||
npc_def = NPCS.get(combat['npc_id'])
|
||
|
||
if not player or not npc_def:
|
||
return ("Combat error!", False, False)
|
||
|
||
# Check if player is stunned
|
||
player_effects = json.loads(combat['player_status_effects'])
|
||
is_stunned = any(effect['name'] == 'Stunned' for effect in player_effects)
|
||
if is_stunned:
|
||
# Update status effects
|
||
player_effects = update_status_effects(player_effects)
|
||
await api_client.update_combat(player_id, {
|
||
'turn': 'npc',
|
||
'turn_started_at': time.time(),
|
||
'player_status_effects': json.dumps(player_effects)
|
||
})
|
||
return ("⚠️ You're stunned and cannot attack! The enemy seizes the opportunity!", False, True)
|
||
|
||
# Calculate damage
|
||
raw_damage = await calculate_player_damage(player)
|
||
actual_damage = max(1, raw_damage - npc_def.defense)
|
||
|
||
new_npc_hp = max(0, combat['npc_hp'] - actual_damage)
|
||
|
||
# Check for critical hit (10% chance)
|
||
is_crit = random.random() < 0.1
|
||
if is_crit:
|
||
actual_damage = int(actual_damage * 1.5)
|
||
new_npc_hp = max(0, combat['npc_hp'] - actual_damage)
|
||
|
||
message = "━━━ YOUR TURN ━━━\n"
|
||
message += f"⚔️ You attack the {npc_def.name} for {actual_damage} damage!"
|
||
if is_crit:
|
||
message += " 💥 CRITICAL HIT!"
|
||
|
||
# Check for status effect infliction (5% chance to stun)
|
||
npc_effects = json.loads(combat['npc_status_effects'])
|
||
if random.random() < 0.05:
|
||
npc_effects.append({
|
||
'name': 'Stunned',
|
||
'turns_remaining': 1,
|
||
'damage_per_turn': 0
|
||
})
|
||
message += f"\n🌟 You stunned the {npc_def.name}!"
|
||
|
||
# Apply status effect damage to player
|
||
player_effects, status_damage, status_messages = apply_status_effects(player_effects)
|
||
if status_damage > 0:
|
||
new_player_hp = max(0, player['hp'] - status_damage)
|
||
await api_client.update_player(player_id, {'hp': new_player_hp})
|
||
message += f"\n{status_messages}"
|
||
|
||
if new_player_hp <= 0:
|
||
await handle_player_death(player_id)
|
||
return (message + "\n\n💀 You have died from your wounds...", True, True)
|
||
|
||
# Check if NPC died
|
||
if new_npc_hp <= 0:
|
||
await api_client.update_combat(player_id, {
|
||
'npc_hp': 0,
|
||
'npc_status_effects': json.dumps(npc_effects),
|
||
'player_status_effects': json.dumps(player_effects)
|
||
})
|
||
|
||
# Handle victory
|
||
victory_msg = await handle_npc_death(player_id, combat, npc_def)
|
||
return (message + "\n\n" + victory_msg, True, True)
|
||
|
||
# Update combat - switch to NPC turn
|
||
await api_client.update_combat(player_id, {
|
||
'npc_hp': new_npc_hp,
|
||
'turn': 'npc',
|
||
'turn_started_at': time.time(),
|
||
'npc_status_effects': json.dumps(npc_effects),
|
||
'player_status_effects': json.dumps(player_effects)
|
||
})
|
||
|
||
# Show both health bars after player's turn
|
||
message += "\n━━━━━━━━━━━━━━━━━━━━\n"
|
||
message += format_stat_bar("Your HP", "❤️", player['hp'], player['max_hp']) + "\n"
|
||
message += format_stat_bar(npc_def.name, npc_def.emoji, new_npc_hp, combat['npc_max_hp'])
|
||
|
||
return (message, False, True)
|
||
|
||
|
||
|
||
async def npc_attack(player_id: int) -> Tuple[str, bool]:
|
||
"""
|
||
NPC attacks the player.
|
||
Returns: (message, player_died)
|
||
"""
|
||
combat = await api_client.get_combat(player_id)
|
||
if not combat or combat['turn'] != 'npc':
|
||
return ("", False)
|
||
|
||
player = await api_client.get_player(player_id)
|
||
npc_def = NPCS.get(combat['npc_id'])
|
||
|
||
if not player or not npc_def:
|
||
return ("Combat error!", False)
|
||
|
||
# Check if NPC is stunned
|
||
npc_effects = json.loads(combat['npc_status_effects'])
|
||
is_stunned = any(effect['name'] == 'Stunned' for effect in npc_effects)
|
||
if is_stunned:
|
||
# Update status effects
|
||
npc_effects = update_status_effects(npc_effects)
|
||
await api_client.update_combat(player_id, {
|
||
'turn': 'player',
|
||
'turn_started_at': time.time(),
|
||
'npc_status_effects': json.dumps(npc_effects)
|
||
})
|
||
return (f"⚠️ The {npc_def.name} is stunned and cannot attack!", False)
|
||
|
||
# Calculate damage
|
||
damage = calculate_npc_damage(npc_def, combat['npc_hp'], combat['npc_max_hp'])
|
||
|
||
# Apply damage to player
|
||
new_player_hp = max(0, player['hp'] - damage)
|
||
await api_client.update_player(player_id, {'hp': new_player_hp})
|
||
|
||
message = "━━━ ENEMY TURN ━━━\n"
|
||
message += f"💥 The {npc_def.name} attacks you for {damage} damage!"
|
||
|
||
# Check for status effect infliction
|
||
player_effects = json.loads(combat['player_status_effects'])
|
||
if random.random() < npc_def.status_inflict_chance:
|
||
# Bleeding is most common
|
||
player_effects.append({
|
||
'name': 'Bleeding',
|
||
'turns_remaining': 3,
|
||
'damage_per_turn': 2
|
||
})
|
||
message += "\n🩸 You're bleeding!"
|
||
|
||
# Apply status effect damage to NPC
|
||
npc_effects, status_damage, status_messages = apply_status_effects(npc_effects)
|
||
if status_damage > 0:
|
||
new_npc_hp = max(0, combat['npc_hp'] - status_damage)
|
||
await api_client.update_combat(player_id, {'npc_hp': new_npc_hp})
|
||
message += f"\n{status_messages}"
|
||
|
||
if new_npc_hp <= 0:
|
||
victory_msg = await handle_npc_death(player_id, combat, npc_def)
|
||
return (message + "\n\n" + victory_msg, False)
|
||
|
||
# Check if player died
|
||
if new_player_hp <= 0:
|
||
await handle_player_death(player_id)
|
||
return (message + "\n\n💀 You have been slain...", True)
|
||
|
||
# Update combat - switch to player turn
|
||
await api_client.update_combat(player_id, {
|
||
'turn': 'player',
|
||
'turn_started_at': time.time(),
|
||
'player_status_effects': json.dumps(player_effects),
|
||
'npc_status_effects': json.dumps(npc_effects)
|
||
})
|
||
|
||
# Show both health bars after enemy's turn
|
||
message += "\n━━━━━━━━━━━━━━━━━━━━\n"
|
||
message += format_stat_bar("Your HP", "❤️", new_player_hp, player['max_hp']) + "\n"
|
||
message += format_stat_bar(npc_def.name, npc_def.emoji, combat['npc_hp'], combat['npc_max_hp'])
|
||
|
||
return (message, False)
|
||
|
||
|
||
async def flee_attempt(player_id: int) -> Tuple[str, bool, bool]:
|
||
"""
|
||
Player attempts to flee from combat.
|
||
Returns: (message, fled_successfully, turn_ended)
|
||
"""
|
||
combat = await api_client.get_combat(player_id)
|
||
if not combat or combat['turn'] != 'player':
|
||
return ("It's not your turn!", False, False)
|
||
|
||
player = await api_client.get_player(player_id)
|
||
npc_def = NPCS.get(combat['npc_id'])
|
||
|
||
# Base flee chance is 50%, modified by agility
|
||
flee_chance = 0.5 + (player['agility'] / 100)
|
||
|
||
if random.random() < flee_chance:
|
||
# Success! Check if we need to respawn the wandering enemy
|
||
if combat.get('from_wandering_enemy', False):
|
||
# Respawn the enemy at the same location with full HP
|
||
await api_client.spawn_wandering_enemy(
|
||
npc_id=combat['npc_id'],
|
||
location_id=combat['location_id'],
|
||
current_hp=npc_def.hp,
|
||
max_hp=npc_def.hp
|
||
)
|
||
|
||
await api_client.end_combat(player_id)
|
||
return (f"🏃 You successfully flee from the {npc_def.name}!", True, True)
|
||
else:
|
||
# Failed - lose turn and NPC attacks
|
||
message = f"❌ You failed to escape! The {npc_def.name} takes advantage!"
|
||
|
||
# NPC gets a free attack
|
||
await api_client.update_combat(player_id, {
|
||
'turn': 'npc',
|
||
'turn_started_at': time.time()
|
||
})
|
||
|
||
return (message, False, True)
|
||
|
||
|
||
def update_status_effects(effects: List[Dict]) -> List[Dict]:
|
||
"""Decrease turn counters on status effects."""
|
||
new_effects = []
|
||
for effect in effects:
|
||
effect['turns_remaining'] -= 1
|
||
if effect['turns_remaining'] > 0:
|
||
new_effects.append(effect)
|
||
return new_effects
|
||
|
||
|
||
def apply_status_effects(effects: List[Dict]) -> Tuple[List[Dict], int, str]:
|
||
"""
|
||
Apply status effect damage with stacking.
|
||
Returns: (updated_effects, total_damage, message)
|
||
"""
|
||
from bot.status_utils import stack_status_effects
|
||
|
||
if not effects:
|
||
return effects, 0, ""
|
||
|
||
# Convert effect format if needed (name -> effect_name, damage_per_turn -> damage_per_tick)
|
||
normalized_effects = []
|
||
for effect in effects:
|
||
normalized = {
|
||
'effect_name': effect.get('name', effect.get('effect_name', 'Unknown')),
|
||
'effect_icon': effect.get('icon', effect.get('effect_icon', '❓')),
|
||
'damage_per_tick': effect.get('damage_per_turn', effect.get('damage_per_tick', 0)),
|
||
'ticks_remaining': effect.get('turns_remaining', effect.get('ticks_remaining', 0))
|
||
}
|
||
normalized_effects.append(normalized)
|
||
|
||
# Stack effects
|
||
stacked = stack_status_effects(normalized_effects)
|
||
|
||
total_damage = 0
|
||
messages = []
|
||
|
||
for name, data in stacked.items():
|
||
if data['total_damage'] > 0:
|
||
total_damage += data['total_damage']
|
||
# Show stacked damage
|
||
if data['stacks'] > 1:
|
||
messages.append(f"{data['icon']} {name.replace('_', ' ').title()}: -{data['total_damage']} HP (×{data['stacks']})")
|
||
else:
|
||
messages.append(f"{data['icon']} {name.replace('_', ' ').title()}: -{data['total_damage']} HP")
|
||
|
||
return effects, total_damage, "\n".join(messages)
|
||
|
||
|
||
async def handle_npc_death(player_id: int, combat: Dict, npc_def) -> str:
|
||
"""Handle NPC death - give XP, drop loot, create corpse."""
|
||
player = await api_client.get_player(player_id)
|
||
|
||
# Give XP
|
||
new_xp = player['xp'] + npc_def.xp_reward
|
||
level_up_msg = ""
|
||
|
||
# Check for level up
|
||
current_level = player['level']
|
||
xp_needed = xp_for_level(current_level + 1)
|
||
|
||
if new_xp >= xp_needed:
|
||
new_level = current_level + 1
|
||
# Give stat points instead of auto-allocating
|
||
# Players get 5 points per level to spend as they wish
|
||
points_gained = 5
|
||
new_unspent_points = player.get('unspent_points', 0) + points_gained
|
||
|
||
await api_client.update_player(player_id, {
|
||
'xp': new_xp,
|
||
'level': new_level,
|
||
'hp': player['max_hp'], # Heal on level up
|
||
'stamina': player['max_stamina'], # Restore stamina on level up
|
||
'unspent_points': new_unspent_points
|
||
})
|
||
|
||
level_up_msg = f"\n\n🎉 LEVEL UP! You are now level {new_level}!"
|
||
level_up_msg += f"\n⭐ You have {points_gained} stat points to spend!"
|
||
level_up_msg += f"\n❤️ Fully healed and stamina restored!"
|
||
level_up_msg += f"\n\n💡 Check your profile to spend your points!"
|
||
else:
|
||
await api_client.update_player(player_id, {'xp': new_xp})
|
||
|
||
# Drop loot
|
||
loot_msg = "\n\n💰 Loot dropped:"
|
||
loot_items = []
|
||
for loot_item in npc_def.loot_table:
|
||
if random.random() < loot_item.drop_chance:
|
||
quantity = random.randint(loot_item.quantity_min, loot_item.quantity_max)
|
||
await api_client.drop_item_to_world(
|
||
loot_item.item_id,
|
||
quantity,
|
||
combat['location_id']
|
||
)
|
||
item_def = ITEMS.get(loot_item.item_id, {})
|
||
loot_msg += f"\n{item_def.get('emoji', '❔')} {item_def.get('name', 'Unknown')} x{quantity}"
|
||
loot_items.append(loot_item.item_id)
|
||
|
||
if not loot_items:
|
||
loot_msg += "\nNothing..."
|
||
|
||
# Create corpse if it has corpse loot
|
||
if npc_def.corpse_loot:
|
||
corpse_loot_json = json.dumps([{
|
||
'item_id': cl.item_id,
|
||
'quantity_min': cl.quantity_min,
|
||
'quantity_max': cl.quantity_max,
|
||
'required_tool': cl.required_tool
|
||
} for cl in npc_def.corpse_loot])
|
||
|
||
await api_client.create_npc_corpse(
|
||
npc_id=combat['npc_id'],
|
||
location_id=combat['location_id'],
|
||
loot_remaining=corpse_loot_json
|
||
)
|
||
loot_msg += f"\n\n{npc_def.emoji} The corpse can be scavenged for more resources."
|
||
|
||
# End combat
|
||
await api_client.end_combat(player_id)
|
||
|
||
message = f"🏆 Victory! {npc_def.death_message}"
|
||
message += f"\n+{npc_def.xp_reward} XP"
|
||
message += level_up_msg
|
||
message += loot_msg
|
||
|
||
return message
|
||
|
||
|
||
async def handle_player_death(player_id: int):
|
||
"""Handle player death - create corpse bag with all items."""
|
||
player = await api_client.get_player(player_id)
|
||
inventory_items = await api_client.get_inventory(player_id)
|
||
|
||
# Check if combat was with a wandering enemy that should respawn
|
||
combat = await api_client.get_combat(player_id)
|
||
if combat and combat.get('from_wandering_enemy', False):
|
||
# Respawn the enemy at the same location with full HP
|
||
npc_def = NPCS.get(combat['npc_id'])
|
||
await api_client.spawn_wandering_enemy(
|
||
npc_id=combat['npc_id'],
|
||
location_id=combat['location_id'],
|
||
current_hp=npc_def.hp,
|
||
max_hp=npc_def.hp
|
||
)
|
||
|
||
# Create corpse bag if player has items
|
||
if inventory_items:
|
||
items_json = json.dumps([{
|
||
'item_id': item['item_id'],
|
||
'quantity': item['quantity']
|
||
} for item in inventory_items])
|
||
|
||
await api_client.create_player_corpse(
|
||
player_name=player['name'],
|
||
location_id=player['location_id'],
|
||
items=items_json
|
||
)
|
||
|
||
# Remove all items from player
|
||
for item in inventory_items:
|
||
await api_client.remove_item_from_inventory(item['id'], item['quantity'])
|
||
|
||
# Mark player as dead and end any combat
|
||
await api_client.update_player(player_id, {'is_dead': True, 'hp': 0})
|
||
await api_client.end_combat(player_id)
|
||
|
||
|
||
async def use_item_in_combat(player_id: int, item_db_id: int) -> Tuple[str, bool]:
|
||
"""
|
||
Use a consumable item during combat.
|
||
Returns: (message, turn_ended)
|
||
"""
|
||
combat = await api_client.get_combat(player_id)
|
||
if not combat or combat['turn'] != 'player':
|
||
return ("It's not your turn!", False)
|
||
|
||
item_data = await api_client.get_inventory_item(item_db_id)
|
||
if not item_data or item_data['player_id'] != player_id:
|
||
return ("You don't have that item!", False)
|
||
|
||
item_def = ITEMS.get(item_data['item_id'])
|
||
if not item_def or item_def.get('type') != 'consumable':
|
||
return ("That item cannot be used in combat!", False)
|
||
|
||
player = await api_client.get_player(player_id)
|
||
|
||
# Apply consumable effects
|
||
message = f"💊 Used {item_def['name']}!"
|
||
|
||
hp_restore = item_def.get('hp_restore', 0)
|
||
stamina_restore = item_def.get('stamina_restore', 0)
|
||
|
||
updates = {}
|
||
if hp_restore > 0:
|
||
new_hp = min(player['hp'] + hp_restore, player['max_hp'])
|
||
updates['hp'] = new_hp
|
||
message += f"\n❤️ +{hp_restore} HP"
|
||
|
||
if stamina_restore > 0:
|
||
new_stamina = min(player['stamina'] + stamina_restore, player['max_stamina'])
|
||
updates['stamina'] = new_stamina
|
||
message += f"\n⚡ +{stamina_restore} Stamina"
|
||
|
||
if updates:
|
||
await api_client.update_player(player_id, updates)
|
||
|
||
# Remove item from inventory
|
||
if item_data['quantity'] > 1:
|
||
await api_client.update_inventory_item(item_db_id, item_data['quantity'] - 1)
|
||
else:
|
||
await api_client.remove_item_from_inventory(item_db_id, 1)
|
||
|
||
# Using an item ends your turn
|
||
await api_client.update_combat(player_id, {
|
||
'turn': 'npc',
|
||
'turn_started_at': time.time()
|
||
})
|
||
|
||
return (message, True)
|