# pvp_battle.py import time import random import threading class PvPRoom: """真实PVP共享战斗房间，双方状态实时同步，支持自动挂机攻击""" def __init__(self, room_id, player_a_name, player_b_name, game_engine, socketio): self.lock = threading.Lock() self.room_id = room_id self.game_engine = game_engine self.socketio = socketio self.players = { player_a_name: {'state_ref': None, 'last_attack_time': 0, 'last_hero_attack_time': 0}, player_b_name: {'state_ref': None, 'last_attack_time': 0, 'last_hero_attack_time': 0} } self.active = True self.lock = threading.Lock() self.pending_logs = [] def get_opponent(self, player_name): for name in self.players: if name != player_name: return self.players[name]['state_ref'] return None def get_opponent_name(self, player_name): for name in self.players: if name != player_name: return name return None def _perform_attack(self, attacker_name, attacker_state, defender_state, is_hero=False): if is_hero: attacker = attacker_state['hero'] if not attacker.get('summoned') or attacker.get('hp', 0) <= 0: return attacker_type = f"{attacker_name}的英雄" ring = self.game_engine._get_special_ring_effect(attacker_state['equipment'], 'hero') weapon = attacker_state['equipment']['hero'].get('weapon') skill_db = self.game_engine.config['heroSkills'] else: attacker = attacker_state['player'] if attacker.get('hp', 0) <= 0: return attacker_type = f"{attacker_name}的玩家" ring = self.game_engine._get_special_ring_effect(attacker_state['equipment'], 'player') weapon = attacker_state['equipment']['player'].get('weapon') skill_db = self.game_engine.config['playerSkills'] # 获取攻击力 att_min, att_max = self.game_engine._get_attack(attacker_state, 'hero' if is_hero else 'player') # 确定目标（优先玩家，其次英雄） target = None target_type = None if defender_state['player'].get('hp', 0) > 0: target = defender_state['player'] target_type = 'player' elif defender_state['hero'].get('summoned') and defender_state['hero'].get('hp', 0) > 0: target = defender_state['hero'] target_type = 'hero' else: return # 无存活目标 # ---------- 技能选择 ---------- skill_used = None skills = attacker.get('skills', []) if skills and random.random() < 0.3: # 30%概率触发技能 skill_entry = random.choice(skills) full_skill = next((s for s in skill_db if s['id'] == skill_entry['id']), None) if full_skill and attacker.get('mp', 0) >= full_skill.get('mp_cost', 0): skill_used = full_skill.copy() skill_used['enhance_level'] = skill_entry.get('enhance_level', 0) skill_used['proficiency'] = skill_entry.get('proficiency', 0) attacker['mp'] -= full_skill.get('mp_cost', 0) attacker_state['log_messages'].append(('skill', f'{attacker_type}使用【{skill_used["name"]}】（消耗{full_skill["mp_cost"]}MP）')) # ---------- 处理群攻技能（AOE） ---------- if skill_used and skill_used.get('targetType') in ('aoe', 'all'): # 收集所有存活目标（对方玩家 + 英雄） targets = [] if defender_state['player'].get('hp', 0) > 0: targets.append(('player', defender_state['player'])) if defender_state['hero'].get('summoned') and defender_state['hero'].get('hp', 0) > 0: targets.append(('hero', defender_state['hero'])) for t_type, t in targets: # 对每个目标单独调用 _character_attack self.game_engine._character_attack( attacker_type, attacker, att_min, att_max, ring, weapon, t, skill_used, attacker_state ) # 检查目标死亡（注意：死亡处理中会添加日志到 defender_state['log_messages']） if t.get('hp', 0) <= 0: if t_type == 'player': self.game_engine._handle_player_death(defender_state, defender_state.get('log_messages', [])) else: self.game_engine._handle_hero_death(defender_state, defender_state.get('log_messages', [])) # 群攻结束，直接返回（日志已经在 _character_attack 中添加） return # ---------- 单攻 ---------- self.game_engine._character_attack( attacker_type, attacker, att_min, att_max, ring, weapon, target, skill_used, attacker_state ) # 检查目标死亡 if target.get('hp', 0) <= 0: if target_type == 'player': self.game_engine._handle_player_death(defender_state, defender_state.get('log_messages', [])) else: self.game_engine._handle_hero_death(defender_state, defender_state.get('log_messages', [])) def process_auto_combat(self, now): with self.lock: # 刷新所有玩家的状态引用（确保与 state_manager 同步） from state_manager import player_state_manager for name, info in self.players.items(): fresh_state = player_state_manager.get(name) if fresh_state is not None: info['state_ref'] = fresh_state if not self.active: return False, False # 为每个玩家单独准备日志列表 if not hasattr(self, '_pending_logs_by_player'): self._pending_logs_by_player = {} for name in self.players: self._pending_logs_by_player[name] = [] # 确保每个玩家都有日志长度记录（保留原有逻辑，用于增量获取，但这里不再使用全局 pending_logs） if not hasattr(self, '_log_lengths'): self._log_lengths = {} for name in self.players: self._log_lengths[name] = 0 any_attacked = False any_death = False # 自动用药冷却控制（每个玩家独立） if not hasattr(self, '_last_auto_potion_time'): self._last_auto_potion_time = {} for player_name, info in self.players.items(): attacker_state = info['state_ref'] if not attacker_state or attacker_state['player'].get('hp', 0) <= 0: continue # 自动用药（玩家 + 英雄） last_time = self._last_auto_potion_time.get(player_name, 0) if now - last_time >= 1.5: self.game_engine._auto_use_potion(attacker_state, 'player') if attacker_state['hero'].get('summoned') and attacker_state['hero'].get('hp', 0) > 0: self.game_engine._auto_use_potion(attacker_state, 'hero') self._last_auto_potion_time[player_name] = now defender_state = self.get_opponent(player_name) if not defender_state: continue # 检查对手是否已无存活单位 opponent_alive = (defender_state['player'].get('hp', 0) > 0) or \ (defender_state['hero'].get('summoned') and defender_state['hero'].get('hp', 0) > 0) if not opponent_alive: self.end_battle(player_name) return True, True opponent_name = self.get_opponent_name(player_name) # 玩家攻击 interval = self.game_engine._calc_attack_interval( attacker_state['player'], self.game_engine.player_base_attack_interval ) if now - info['last_attack_time'] >= interval: # 记录攻击前双方日志长度 before_len_att = len(attacker_state.get('log_messages', [])) before_len_def = len(defender_state.get('log_messages', [])) self._perform_attack(player_name, attacker_state, defender_state, is_hero=False) info['last_attack_time'] = now any_attacked = True # 收集攻击方新增日志 after_len_att = len(attacker_state.get('log_messages', [])) if after_len_att > before_len_att: new_logs_att = attacker_state['log_messages'][before_len_att:after_len_att] self._pending_logs_by_player[player_name].extend(new_logs_att) # 收集防御方新增日志（死亡等） after_len_def = len(defender_state.get('log_messages', [])) if after_len_def > before_len_def: new_logs_def = defender_state['log_messages'][before_len_def:after_len_def] self._pending_logs_by_player[opponent_name].extend(new_logs_def) # 英雄攻击 hero = attacker_state['hero'] if hero.get('summoned') and hero.get('hp', 0) > 0: hero_interval = self.game_engine._calc_attack_interval( hero, self.game_engine.hero_base_attack_interval ) if now - info['last_hero_attack_time'] >= hero_interval: # 记录攻击前双方日志长度 before_len_att = len(attacker_state.get('log_messages', [])) before_len_def = len(defender_state.get('log_messages', [])) self._perform_attack(player_name, attacker_state, defender_state, is_hero=True) info['last_hero_attack_time'] = now any_attacked = True # 收集攻击方新增日志 after_len_att = len(attacker_state.get('log_messages', [])) if after_len_att > before_len_att: new_logs_att = attacker_state['log_messages'][before_len_att:after_len_att] self._pending_logs_by_player[player_name].extend(new_logs_att) # 收集防御方新增日志 after_len_def = len(defender_state.get('log_messages', [])) if after_len_def > before_len_def: new_logs_def = defender_state['log_messages'][before_len_def:after_len_def] self._pending_logs_by_player[opponent_name].extend(new_logs_def) # 分别为每个玩家广播状态和自己的日志 for player_name, logs in self._pending_logs_by_player.items(): self.broadcast_state(logs=logs, target_player=player_name) # 清空临时日志字典，避免积累 self._pending_logs_by_player = {name: [] for name in self.players} return any_attacked, any_death def broadcast_state(self, logs=None, target_player=None): """广播PVP状态，如果指定 target_player 则只发给该玩家，且附带其专属日志""" players_to_send = [target_player] if target_player else list(self.players.keys()) for player_name in players_to_send: info = self.players.get(player_name) if not info: continue state = info['state_ref'] opponent_state = self.get_opponent(player_name) if not state or not opponent_state: continue opponent_name = self.get_opponent_name(player_name) data = { 'room_id': self.room_id, 'your': { 'hp': state['player']['hp'], 'maxHp': state['player']['maxHp'], 'mp': state['player'].get('mp', 0), 'maxMp': state['player'].get('maxMp', 100), 'level': state['player'].get('level', 1), 'gameSpeed': state['player'].get('gameSpeed', 1), 'hero_hp': state['hero'].get('hp', 0), 'hero_max_hp': state['hero'].get('maxHp', 0), 'hero_mp': state['hero'].get('mp', 0), 'hero_max_mp': state['hero'].get('maxMp', 100), 'pk_score': state['player'].get('pk_score', 0) }, 'opponent': { 'name': opponent_name, 'avatar': opponent_state['player'].get('avatar', ''), 'hp': opponent_state['player']['hp'], 'maxHp': opponent_state['player']['maxHp'], 'mp': opponent_state['player'].get('mp', 0), 'maxMp': opponent_state['player'].get('maxMp', 100), 'level': opponent_state['player'].get('level', 1), 'gameSpeed': opponent_state['player'].get('gameSpeed', 1), 'hero_hp': opponent_state['hero'].get('hp', 0), 'hero_max_hp': opponent_state['hero'].get('maxHp', 0), 'hero_mp': opponent_state['hero'].get('mp', 0), 'hero_max_mp': opponent_state['hero'].get('maxMp', 100), 'hero_name': opponent_state['hero'].get('name', '英雄'), 'pk_score': opponent_state['player'].get('pk_score', 0) } } if logs: data['logs'] = logs self.socketio.emit('pvp_state_update', data, room=player_name) def end_battle(self, winner_name): self.active = False loser_name = self.get_opponent_name(winner_name) # 通知双方战斗结束 self.socketio.emit('pvp_battle_end', {'winner': winner_name, 'loser': loser_name}, room=winner_name) self.socketio.emit('pvp_battle_end', {'winner': winner_name, 'loser': loser_name}, room=loser_name) from state_manager import player_state_manager for name in self.players: # 1. 清理房间内缓存的旧状态 cached_state = self.players[name]['state_ref'] if cached_state: cached_state.pop('_pvp_room_id', None) cached_state.pop('_pvp_opponent_state', None) # 2. 获取玩家当前实际状态（在线玩家的最新状态） real_state = player_state_manager.get(name) if real_state: real_state.pop('_pvp_room_id', None) real_state.pop('_pvp_opponent_state', None) # 强制重新生成普通怪物组 map_id = real_state.get('currentMap', 1) new_monsters = self.game_engine._select_monster_group_for_map(map_id) if new_monsters: import uuid real_state['monsterGroup'] = { 'monsters': new_monsters, 'uuid': str(uuid.uuid4()) } for m in real_state['monsterGroup']['monsters']: m['active'] = False m['last_attack_time'] = time.time() + random.uniform(0, 0.5) else: real_state['monsterGroup'] = None # 保存状态修改 player_state_manager.set(name, real_state) # 推送前端更新 self.socketio.emit('combat_update', { 'diff': {'monsterGroup': real_state['monsterGroup']} }, room=name) class PvPBattleManager: def __init__(self, game_engine, socketio): self.game_engine = game_engine self.socketio = socketio self.rooms = {} self.player_room = {} def create_room(self, player_a_name, player_b_name, state_a, state_b): room_id = f"pvp_{int(time.time())}_{player_a_name}_{player_b_name}" room = PvPRoom(room_id, player_a_name, player_b_name, self.game_engine, self.socketio) room.players[player_a_name]['state_ref'] = state_a room.players[player_b_name]['state_ref'] = state_b self.rooms[room_id] = room self.player_room[player_a_name] = room_id self.player_room[player_b_name] = room_id # 标记双方正在PVP state_a['_pvp_room_id'] = room_id state_b['_pvp_room_id'] = room_id # 清空怪物组 state_a['monsterGroup'] = None state_b['monsterGroup'] = None return room_id def get_room_by_player(self, player_name): room_id = self.player_room.get(player_name) return self.rooms.get(room_id) def remove_room(self, room_id): room = self.rooms.pop(room_id, None) if room: from state_manager import player_state_manager for name in room.players: # 清理房间缓存 cached = room.players[name]['state_ref'] if cached: cached.pop('_pvp_room_id', None) cached.pop('_pvp_opponent_state', None) # 清理实际状态 real_state = player_state_manager.get(name) if real_state: real_state.pop('_pvp_room_id', None) real_state.pop('_pvp_opponent_state', None) self.player_room.pop(name, None) def player_leave(self, player_name): room = self.get_room_by_player(player_name) if room: winner = self.get_opponent_name(room, player_name) room.end_battle(winner) self.remove_room(room.room_id) def get_opponent_name(self, room, player_name): for name in room.players: if name != player_name: return name return None