# -*- coding: utf-8 -*- """ 关卡配置管理器 负责从Excel读取关卡配置数据，并智能合并到现有JSON文件中 只修改或新增策划表中配置的字段，不会完全覆盖JSON文件 """ import pandas as pd import json import os from datetime import datetime from pathlib import Path import re class LevelConfigManager: def __init__(self, excel_path=None, levels_dir=None): """ 初始化关卡配置管理器 Args: excel_path: Excel文件路径 levels_dir: 关卡JSON文件目录 """ self.excel_path = excel_path self.levels_dir = levels_dir or "d:/CocosGame/Pong/assets/resources/data/levels" self.backup_dir = os.path.join(os.path.dirname(self.levels_dir), "backups", "levels") # 确保备份目录存在 os.makedirs(self.backup_dir, exist_ok=True) # 配置数据存储 self.basic_config = None self.weapon_config = None self.wave_config = None self.enemy_config = None self.energy_exp_config = None self.energy_upgrade_config = None def read_excel_data(self, excel_path=None): """ 读取Excel文件中的所有工作表数据 Args: excel_path: Excel文件路径，如果不提供则使用初始化时的路径 Returns: bool: 是否成功读取数据 """ if excel_path: self.excel_path = excel_path if not self.excel_path or not os.path.exists(self.excel_path): print(f"错误: Excel文件不存在: {self.excel_path}") return False try: # 读取所有工作表 all_sheets = pd.read_excel(self.excel_path, sheet_name=None) # 解析各个工作表 self._parse_basic_config(all_sheets) self._parse_weapon_config(all_sheets) self._parse_wave_config(all_sheets) self._parse_enemy_config(all_sheets) self._parse_energy_exp_config(all_sheets) self._parse_energy_upgrade_config(all_sheets) print(f"成功读取Excel数据: {self.excel_path}") return True except Exception as e: print(f"读取Excel文件失败: {e}") return False def _parse_basic_config(self, all_sheets): """ 解析关卡基础配置工作表 """ sheet_names = ['关卡基础配置', 'Level Config', 'levels', '关卡配置'] for sheet_name in sheet_names: if sheet_name in all_sheets: self.basic_config = all_sheets[sheet_name] print(f"找到关卡基础配置工作表: {sheet_name}") return print("警告: 未找到关卡基础配置工作表") def _parse_weapon_config(self, all_sheets): """ 解析关卡武器配置工作表 """ sheet_names = ['关卡武器配置', 'Level Weapon Config', 'level_weapons', '武器配置'] for sheet_name in sheet_names: if sheet_name in all_sheets: self.weapon_config = all_sheets[sheet_name] print(f"找到关卡武器配置工作表: {sheet_name}") return print("警告: 未找到关卡武器配置工作表") def _parse_wave_config(self, all_sheets): """ 解析关卡波次配置工作表 """ sheet_names = ['关卡波次配置', 'Wave Config', 'waves', '波次配置'] for sheet_name in sheet_names: if sheet_name in all_sheets: self.wave_config = all_sheets[sheet_name] print(f"找到关卡波次配置工作表: {sheet_name}") return print("警告: 未找到关卡波次配置工作表") def _parse_enemy_config(self, all_sheets): """ 解析敌人详细配置工作表 """ sheet_names = ['敌人详细配置', 'Enemy Detail Config', 'enemy_details', '敌人配置'] for sheet_name in sheet_names: if sheet_name in all_sheets: self.enemy_config = all_sheets[sheet_name] print(f"找到敌人详细配置工作表: {sheet_name}") return print("警告: 未找到敌人详细配置工作表") def _parse_energy_exp_config(self, all_sheets): """ 解析能量条经验值配置工作表 """ sheet_names = ['能量经验配置', 'Energy Exp Config', 'energy_exp', '能量经验'] for sheet_name in sheet_names: if sheet_name in all_sheets: self.energy_exp_config = all_sheets[sheet_name] print(f"找到能量条经验值配置工作表: {sheet_name}") return print("警告: 未找到能量条经验值配置工作表") def _parse_energy_upgrade_config(self, all_sheets): """ 解析能量条升级配置工作表 """ sheet_names = ['能量最大值升级', 'Energy Upgrade Config', 'energy_upgrade', '能量升级'] for sheet_name in sheet_names: if sheet_name in all_sheets: self.energy_upgrade_config = all_sheets[sheet_name] print(f"找到能量条升级配置工作表: {sheet_name}") return print("警告: 未找到能量条升级配置工作表") def get_all_level_ids(self): """ 获取所有关卡ID Returns: set: 所有关卡ID的集合 """ level_ids = set() # 从基础配置中获取关卡ID if self.basic_config is not None: for _, row in self.basic_config.iterrows(): if pd.notna(row['关卡ID']): level_ids.add(str(row['关卡ID'])) # 从武器配置中获取关卡ID if self.weapon_config is not None: for _, row in self.weapon_config.iterrows(): if pd.notna(row['关卡ID']): level_ids.add(str(row['关卡ID'])) # 从波次配置中获取关卡ID if self.wave_config is not None: for _, row in self.wave_config.iterrows(): if pd.notna(row['关卡ID']): level_ids.add(str(row['关卡ID'])) return level_ids def merge_configurations(self): """ 合并Excel配置到现有JSON文件 Returns: dict: 合并结果统计 """ if not any([self.basic_config is not None, self.weapon_config is not None, self.wave_config is not None, self.enemy_config is not None]): print("错误: 没有可用的配置数据") return {"success": False, "message": "没有可用的配置数据"} level_ids = self.get_all_level_ids() if not level_ids: print("错误: 没有找到任何关卡ID") return {"success": False, "message": "没有找到任何关卡ID"} results = { "success": True, "processed_levels": [], "created_levels": [], "updated_levels": [], "errors": [] } for level_id in level_ids: try: result = self._update_level_config(level_id) results["processed_levels"].append(level_id) if result["created"]: results["created_levels"].append(level_id) else: results["updated_levels"].append(level_id) except Exception as e: error_msg = f"处理关卡 {level_id} 时出错: {str(e)}" print(error_msg) results["errors"].append(error_msg) return results def _update_level_config(self, level_id): """ 更新单个关卡的配置 Args: level_id: 关卡ID Returns: dict: 更新结果 """ json_path = os.path.join(self.levels_dir, f"{level_id}.json") # 读取现有配置或创建新配置 if os.path.exists(json_path): with open(json_path, 'r', encoding='utf-8') as f: existing_config = json.load(f) created = False else: existing_config = { "levelId": level_id, "name": "", "scene": "grassland", "description": "", "backgroundImage": "images/LevelBackground/BG1", "availableWeapons": [], "coinReward": 100, "diamondReward": 0, "initialCoins": 50, "timeLimit": 300, "difficulty": "normal", "healthMultiplier": 1.0, "levelSettings": { "energyMax": 5 # 默认能量最大值 }, "waves": [] } created = True # 备份现有文件 if not created: self._backup_level_file(json_path, level_id) # 更新基础配置 if self.basic_config is not None: basic_row = self.basic_config[self.basic_config['关卡ID'] == level_id] if not basic_row.empty: row = basic_row.iloc[0] if pd.notna(row['关卡名称']): existing_config['name'] = str(row['关卡名称']) if pd.notna(row['场景']): existing_config['scene'] = str(row['场景']) if pd.notna(row['描述']): existing_config['description'] = str(row['描述']) if pd.notna(row['关卡背景图路径']): existing_config['backgroundImage'] = str(row['关卡背景图路径']) if pd.notna(row['钞票奖励']): existing_config['coinReward'] = int(row['钞票奖励']) if pd.notna(row['钻石奖励']): existing_config['diamondReward'] = int(row['钻石奖励']) if pd.notna(row['生命倍数']): existing_config['healthMultiplier'] = float(row['生命倍数']) if pd.notna(row['难度']): existing_config['difficulty'] = str(row['难度']) if pd.notna(row['初始金币']): existing_config['initialCoins'] = int(row['初始金币']) # 确保levelSettings存在（用于后续的energyMaxUpgrades配置） if 'levelSettings' not in existing_config: existing_config['levelSettings'] = {} # 注释：energyMax配置已移除，现在只使用energyMaxUpgrades数组管理能量最大值 # 注释：energyExpValues配置已移除，现在只使用energyMaxUpgrades配置 # 能量条经验值现在直接在游戏代码中硬编码为递增模式 # 更新能量条升级配置 if self.energy_upgrade_config is not None: energy_upgrade_row = self.energy_upgrade_config[self.energy_upgrade_config['关卡ID'] == level_id] if not energy_upgrade_row.empty: row = energy_upgrade_row.iloc[0] if 'levelSettings' not in existing_config: existing_config['levelSettings'] = {} # 读取20次升级后的最大值配置 energy_max_values = [] for i in range(1, 21): # 1到20次升级 col_name = f'第{i}次升级后最大值' if col_name in row and pd.notna(row[col_name]): energy_max_values.append(int(row[col_name])) else: # 如果没有配置，使用默认值（基于固定初始值6递增） base_max = 6 # 固定初始值，与energyMaxUpgrades数组第一个值保持一致 energy_max_values.append(base_max + (i - 1)) # 修正递增逻辑 existing_config['levelSettings']['energyMaxUpgrades'] = energy_max_values # 更新武器配置 if self.weapon_config is not None: weapon_rows = self.weapon_config[self.weapon_config['关卡ID'] == level_id] available_weapons = [] for _, weapon_row in weapon_rows.iterrows(): if pd.notna(weapon_row['可用武器']): weapons_str = str(weapon_row['可用武器']) # 支持多种分隔符 weapons = re.split(r'[、，,;；]', weapons_str) available_weapons.extend([w.strip() for w in weapons if w.strip()]) if available_weapons: existing_config['availableWeapons'] = available_weapons # 更新波次配置 if self.wave_config is not None and self.enemy_config is not None: level_waves = self.wave_config[self.wave_config['关卡ID'] == level_id] waves_data = [] for _, wave_row in level_waves.iterrows(): wave_id = int(wave_row['波次ID']) if pd.notna(wave_row['波次ID']) else 1 # 获取该波次的敌人配置 wave_enemies = [] wave_enemy_data = self.enemy_config[ (self.enemy_config['关卡ID'] == level_id) & (self.enemy_config['波次ID'] == wave_id) ] for _, enemy_row in wave_enemy_data.iterrows(): enemy_data = { 'enemyType': str(enemy_row['敌人类型']) if pd.notna(enemy_row['敌人类型']) else 'normal_zombie', 'count': int(enemy_row['数量']) if pd.notna(enemy_row['数量']) else 1, 'spawnInterval': float(enemy_row['生成间隔']) if pd.notna(enemy_row['生成间隔']) else 2.0, 'spawnDelay': float(enemy_row['生成延迟']) if pd.notna(enemy_row['生成延迟']) else 0.0, 'characteristics': str(enemy_row['特征描述']) if pd.notna(enemy_row['特征描述']) else '' } wave_enemies.append(enemy_data) wave_data = { 'waveId': wave_id, 'enemies': wave_enemies } waves_data.append(wave_data) if waves_data: # 按波次ID排序 waves_data.sort(key=lambda x: x['waveId']) existing_config['waves'] = waves_data # 保存更新后的配置 with open(json_path, 'w', encoding='utf-8') as f: json.dump(existing_config, f, ensure_ascii=False, indent=2) print(f"{'创建' if created else '更新'}关卡配置: {level_id}") return {"created": created, "updated": not created} def _backup_level_file(self, json_path, level_id): """ 备份关卡JSON文件 Args: json_path: JSON文件路径 level_id: 关卡ID """ try: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backup_filename = f"{level_id}_{timestamp}.json" backup_path = os.path.join(self.backup_dir, backup_filename) with open(json_path, 'r', encoding='utf-8') as src: with open(backup_path, 'w', encoding='utf-8') as dst: dst.write(src.read()) print(f"备份关卡文件: {backup_path}") except Exception as e: print(f"备份关卡文件失败: {e}") def import_from_excel(self, excel_path=None, levels_dir=None): """ 从Excel导入关卡配置的完整流程 Args: excel_path: Excel文件路径 levels_dir: 关卡JSON文件目录 Returns: dict: 导入结果 """ if excel_path: self.excel_path = excel_path if levels_dir: self.levels_dir = levels_dir print("开始导入关卡配置...") # 读取Excel数据 if not self.read_excel_data(): return {"success": False, "message": "读取Excel数据失败"} # 合并配置 results = self.merge_configurations() if results["success"]: print(f"关卡配置导入完成:") print(f" 处理关卡数: {len(results['processed_levels'])}") print(f" 新建关卡数: {len(results['created_levels'])}") print(f" 更新关卡数: {len(results['updated_levels'])}") if results["errors"]: print(f" 错误数: {len(results['errors'])}") for error in results["errors"]: print(f" {error}") return results # 使用示例 if __name__ == "__main__": # 创建关卡配置管理器 manager = LevelConfigManager( excel_path="d:/CocosGame/Pong/assets/resources/data/excel/关卡配置/关卡配置表.xlsx", levels_dir="d:/CocosGame/Pong/assets/resources/data/levels" ) # 导入配置 result = manager.import_from_excel() if result["success"]: print("关卡配置导入成功!") else: print(f"关卡配置导入失败: {result['message']}")