""" 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)