tanqiu/assets/data/excel/level_config_manager.py

# -*- coding: utf-8 -*-
"""
关卡配置管理器
负责从Excel读取关卡配置数据，并智能合并到现有JSON文件中
只修改或新增策划表中配置的字段，不会完全覆盖JSON文件
"""

import pandas as pd
import json
import os
from datetime import datetime
from pathlib import Path
import re

# 是否启用大小写不敏感解析（由启动脚本设置 CFG_CASE_INSENSITIVE=1）
CASE_INSENSITIVE = str(os.environ.get('CFG_CASE_INSENSITIVE', '0')).strip().lower() in ('1', 'true', 'yes', 'y')

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/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 _normalize_key(self, s):
        if s is None:
            return ""
        return str(s).strip().lower()

    # 统一关卡ID的规范形式，例如 level12 -> Level12
    def _canonical_level_id(self, s):
        if s is None:
            return ""
        v = str(s).strip()
        m = re.match(r"^\s*level\s*(\d+)\s*$", v, flags=re.IGNORECASE)
        if m:
            return f"Level{m.group(1)}"
        return v

    def _get_sheet(self, all_sheets, candidate_names):
        if not CASE_INSENSITIVE:
            for name in candidate_names:
                if name in all_sheets:
                    return all_sheets[name], name
            return None, None
        lower_map = {self._normalize_key(k): k for k in all_sheets.keys()}
        for name in candidate_names:
            key = self._normalize_key(name)
            if key in lower_map:
                resolved = lower_map[key]
                return all_sheets[resolved], resolved
        return None, None

    def _get_row_value(self, row, possible_names, default=None):
        index_map = {self._normalize_key(col): col for col in row.index}
        for name in possible_names:
            norm = self._normalize_key(name)
            if norm in index_map:
                val = row[index_map[norm]]
                return val if pd.notna(val) else default
        return default

    def _to_int(self, v, default=None):
        try:
            if v is None or (isinstance(v, float) and pd.isna(v)):
                return default
            return int(float(str(v).strip()))
        except Exception:
            return default

    def _to_float(self, v, default=None):
        try:
            if v is None or (isinstance(v, float) and pd.isna(v)):
                return default
            return float(str(v).strip())
        except Exception:
            return default

    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_aliases = ['关卡基础配置', '关卡配置', 'Level Config', 'levels', 'level config']
        sheet, resolved = self._get_sheet(all_sheets, sheet_aliases)
        if sheet is not None:
            self.basic_config = sheet
            print(f"找到关卡基础配置工作表: {resolved}")
            return
        print("警告: 未找到关卡基础配置工作表")

    def _parse_weapon_config(self, all_sheets):
        """
        解析关卡武器配置工作表
        """
        sheet_aliases = ['关卡武器配置', '武器配置', 'Level Weapon Config', 'level_weapons', 'level weapons']
        sheet, resolved = self._get_sheet(all_sheets, sheet_aliases)
        if sheet is not None:
            self.weapon_config = sheet
            print(f"找到关卡武器配置工作表: {resolved}")
            return
        print("警告: 未找到关卡武器配置工作表")

    def _parse_wave_config(self, all_sheets):
        """
        解析关卡波次配置工作表
        """
        sheet_aliases = ['关卡波次配置', '波次配置', 'Wave Config', 'waves', 'wave config']
        sheet, resolved = self._get_sheet(all_sheets, sheet_aliases)
        if sheet is not None:
            self.wave_config = sheet
            print(f"找到关卡波次配置工作表: {resolved}")
            return
        print("警告: 未找到关卡波次配置工作表")

    def _parse_enemy_config(self, all_sheets):
        """
        解析敌人详细配置工作表
        """
        sheet_aliases = ['敌人详细配置', '敌人配置', 'Enemy Detail Config', 'enemy_details', 'enemy details']
        sheet, resolved = self._get_sheet(all_sheets, sheet_aliases)
        if sheet is not None:
            self.enemy_config = sheet
            print(f"找到敌人详细配置工作表: {resolved}")
            return
        print("警告: 未找到敌人详细配置工作表")

    def _parse_energy_exp_config(self, all_sheets):
        """
        解析能量条经验值配置工作表
        """
        sheet_aliases = ['能量经验配置', '能量经验', 'Energy Exp Config', 'energy_exp', 'energy exp']
        sheet, resolved = self._get_sheet(all_sheets, sheet_aliases)
        if sheet is not None:
            self.energy_exp_config = sheet
            print(f"找到能量条经验值配置工作表: {resolved}")
            return
        print("警告: 未找到能量条经验值配置工作表")

    def _parse_energy_upgrade_config(self, all_sheets):
        """
        解析能量条升级配置工作表
        """
        sheet_aliases = ['能量最大值升级', '能量升级', 'Energy Upgrade Config', 'energy_upgrade', 'energy upgrade']
        sheet, resolved = self._get_sheet(all_sheets, sheet_aliases)
        if sheet is not None:
            self.energy_upgrade_config = sheet
            print(f"找到能量条升级配置工作表: {resolved}")
            return
        print("警告: 未找到能量条升级配置工作表")

    def get_all_level_ids(self):
        """
        获取所有关卡ID（支持大小写不敏感与英文别名）
        
        Returns:
            set: 所有关卡ID的集合（规范化形式，如 Level10）
        """
        level_ids = set()
        id_aliases = ['关卡ID', '关卡Id', '关卡id', 'LevelID', 'Level ID', 'level_id', 'level id', 'id']
        
        def add_id(val):
            if val is not None:
                level_ids.add(self._canonical_level_id(val))
        
        # 从基础配置中获取关卡ID
        if self.basic_config is not None:
            for _, row in self.basic_config.iterrows():
                add_id(self._get_row_value(row, id_aliases))
        
        # 从武器配置中获取关卡ID
        if self.weapon_config is not None:
            for _, row in self.weapon_config.iterrows():
                add_id(self._get_row_value(row, id_aliases))
        
        # 从波次配置中获取关卡ID
        if self.wave_config is not None:
            for _, row in self.wave_config.iterrows():
                add_id(self._get_row_value(row, id_aliases))
        
        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):
        """
        更新单个关卡的配置
        """
        canonical_id = self._canonical_level_id(level_id)
        cmp_id_lower = canonical_id.strip().lower()
        
        # 选择规范化的文件路径，并修正历史上小写文件名
        json_path = os.path.join(self.levels_dir, f"{canonical_id}.json")
        lower_path = os.path.join(self.levels_dir, f"{canonical_id.lower()}.json")
        if not os.path.exists(json_path) and os.path.exists(lower_path):
            try:
                os.replace(lower_path, json_path)
                print(f"已规范化文件名: {lower_path} -> {json_path}")
            except Exception as e:
                print(f"文件名规范化失败: {e}")
        
        # 读取现有配置或创建新配置
        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": canonical_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
        
        # 保证levelId写入规范化形式
        existing_config['levelId'] = canonical_id
        
        # 备份现有文件
        if not created:
            self._backup_level_file(json_path, canonical_id)
        
        # 公共列别名
        id_aliases = ['关卡ID', '关卡Id', '关卡id', 'LevelID', 'Level ID', 'level_id', 'level id', 'id']
        
        # 更新基础配置
        if self.basic_config is not None:
            basic_row = None
            for _, row in self.basic_config.iterrows():
                rid = self._get_row_value(row, id_aliases)
                if rid is not None and str(rid).strip().lower() == cmp_id_lower:
                    basic_row = row
                    break
            if basic_row is not None:
                name = self._get_row_value(basic_row, ['关卡名称','关卡名','Level Name','name'])
                scene = self._get_row_value(basic_row, ['场景','Scene','scene'])
                desc = self._get_row_value(basic_row, ['描述','Description','desc'])
                bg = self._get_row_value(basic_row, ['关卡背景图路径','背景图','背景','Background Image','backgroundImage'])
                coins = self._get_row_value(basic_row, ['钞票奖励','Coin Reward','coins','coinReward'])
                diamonds = self._get_row_value(basic_row, ['钻石奖励','Diamond Reward','diamonds','diamondReward'])
                health_mul = self._get_row_value(basic_row, ['生命倍数','Health Multiplier','healthMultiplier'])
                difficulty = self._get_row_value(basic_row, ['难度','Difficulty','difficulty'])
                initial = self._get_row_value(basic_row, ['初始金币','Initial Coins','initialCoins'])
                energy_max = self._get_row_value(basic_row, ['能量最大值','Energy Max','energyMax'])
                
                if name is not None:
                    existing_config['name'] = str(name)
                if scene is not None:
                    existing_config['scene'] = str(scene)
                if desc is not None:
                    existing_config['description'] = str(desc)
                if bg is not None:
                    existing_config['backgroundImage'] = str(bg)
                if coins is not None:
                    existing_config['coinReward'] = self._to_int(coins, existing_config.get('coinReward', 100))
                if diamonds is not None:
                    existing_config['diamondReward'] = self._to_int(diamonds, existing_config.get('diamondReward', 0))
                if health_mul is not None:
                    existing_config['healthMultiplier'] = self._to_float(health_mul, existing_config.get('healthMultiplier', 1.0))
                if difficulty is not None:
                    existing_config['difficulty'] = str(difficulty)
                if initial is not None:
                    existing_config['initialCoins'] = self._to_int(initial, existing_config.get('initialCoins', 50))
                if energy_max is not None:
                    existing_config.setdefault('levelSettings', {})['energyMax'] = self._to_int(energy_max, existing_config.get('levelSettings', {}).get('energyMax', 5))
                
                if 'levelSettings' not in existing_config:
                    existing_config['levelSettings'] = {}
        
        # 能量条升级配置
        if self.energy_upgrade_config is not None:
            energy_upgrade_row = None
            for _, row in self.energy_upgrade_config.iterrows():
                rid = self._get_row_value(row, id_aliases)
                if rid is not None and str(rid).strip().lower() == cmp_id_lower:
                    energy_upgrade_row = row
                    break
            if energy_upgrade_row is not None:
                if 'levelSettings' not in existing_config:
                    existing_config['levelSettings'] = {}
                energy_max_values = []
                for i in range(1, 21):
                    cn = f'第{i}次升级后最大值'
                    en1 = f'Energy Max After Upgrade {i}'
                    en2 = f'EnergyMaxUpgrade{i}'
                    en3 = f'EnergyMax{i}'
                    val = self._get_row_value(energy_upgrade_row, [cn, en1, en2, en3])
                    if val is not None:
                        energy_max_values.append(self._to_int(val, None))
                    else:
                        base_max = 6
                        energy_max_values.append(base_max + (i - 1))
                existing_config['levelSettings']['energyMaxUpgrades'] = energy_max_values
        
        # 更新武器配置
        if self.weapon_config is not None:
            available_weapons = []
            for _, weapon_row in self.weapon_config.iterrows():
                rid = self._get_row_value(weapon_row, id_aliases)
                if rid is None or str(rid).strip().lower() != cmp_id_lower:
                    continue
                weapons_str = self._get_row_value(weapon_row, ['可用武器','Available Weapons','weapons'])
                if weapons_str:
                    weapons = re.split(r'[、，,;；]', str(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:
            waves_data = []
            # collect rows for this level
            level_waves = []
            for _, wave_row in self.wave_config.iterrows():
                rid = self._get_row_value(wave_row, id_aliases)
                if rid is not None and str(rid).strip().lower() == cmp_id_lower:
                    level_waves.append(wave_row)
            for wave_row in level_waves:
                wave_id_val = self._get_row_value(wave_row, ['波次ID','波次','Wave ID','waveId','wave id'])
                wave_id = self._to_int(wave_id_val, 1)
                
                wave_enemies = []
                wave_enemy_data = []
                for _, enemy_row in self.enemy_config.iterrows():
                    rid = self._get_row_value(enemy_row, id_aliases)
                    wid = self._get_row_value(enemy_row, ['波次ID','波次','Wave ID','waveId','wave id'])
                    if rid is not None and str(rid).strip().lower() == cmp_id_lower and self._to_int(wid, None) == wave_id:
                        wave_enemy_data.append(enemy_row)
                
                wave_health_multiplier = 1.0
                for enemy_row in wave_enemy_data:
                    enemy_health_multiplier = self._to_float(
                        self._get_row_value(enemy_row, ['血量系数','Health Multiplier','healthMultiplier']), 1.0
                    )
                    wave_health_multiplier = max(wave_health_multiplier, enemy_health_multiplier)
                    
                    enemy_type = self._get_row_value(enemy_row, ['敌人类型','Enemy Type','enemyType'])
                    count = self._to_int(self._get_row_value(enemy_row, ['数量','Count','count']), 1)
                    spawn_interval = self._to_float(self._get_row_value(enemy_row, ['生成间隔','Spawn Interval','spawnInterval']), 2.0)
                    spawn_delay = self._to_float(self._get_row_value(enemy_row, ['生成延迟','Spawn Delay','spawnDelay']), 0.0)
                    characteristics = self._get_row_value(enemy_row, ['特征描述','Characteristics','characteristics'])
                    
                    enemy_data = {
                        'enemyType': str(enemy_type) if enemy_type is not None else 'normal_zombie',
                        'count': count,
                        'spawnInterval': spawn_interval,
                        'spawnDelay': spawn_delay,
                        'characteristics': str(characteristics) if characteristics is not None else '',
                        'healthMultiplier': enemy_health_multiplier
                    }
                    wave_enemies.append(enemy_data)
                
                wave_data = {
                    'waveId': wave_id,
                    'healthMultiplier': wave_health_multiplier,
                    'enemies': wave_enemies
                }
                waves_data.append(wave_data)
            
            if waves_data:
                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 '更新'}关卡配置: {canonical_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/data/excel/关卡配置/关卡配置表.xlsx",
        levels_dir="d:/CocosGame/Pong/assets/data/levels"
    )
    
    # 导入配置
    result = manager.import_from_excel()
    
    if result["success"]:
        print("关卡配置导入成功!")
    else:
        print(f"关卡配置导入失败: {result['message']}")