Q103: 如何实现脚本热更新？

# Python 热更新实现

import sys
import importlib
import os
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class ScriptReloader(FileSystemEventHandler):
    """脚本文件变化处理器"""

    def __init__(self, script_dir="scripts"):
        self.script_dir = script_dir
        self.loaded_modules = {}
        self.reload_callbacks = []

    def on_modified(self, event):
        """文件修改时调用"""
        if event.is_directory:
            return

        file_path = event.src_path
        if not file_path.endswith('.py'):
            return

        # 转换为模块名
        module_name = self.path_to_module(file_path)

        if module_name in sys.modules:
            print(f"[HotReload] Reloading: {module_name}")

            # 保存旧模块状态
            old_module = sys.modules[module_name]

            # 调用清理函数
            if hasattr(old_module, 'on_unload'):
                old_module.on_unload()

            # 重新加载
            try:
                new_module = importlib.reload(old_module)

                # 调用加载函数
                if hasattr(new_module, 'on_load'):
                    new_module.on_load()

                # 通知回调
                for callback in self.reload_callbacks:
                    callback(module_name)

                print(f"[HotReload] Success: {module_name}")
            except Exception as e:
                print(f"[HotReload] Error reloading {module_name}: {e}")

    def path_to_module(self, file_path):
        """文件路径转模块名"""
        # 移除 .py 后缀
        module_path = file_path.replace('.py', '')

        # 转换路径分隔符
        module_path = module_path.replace('/', '.').replace('\\', '.')

        # 移除 scripts 前缀
        if module_path.startswith('scripts.'):
            module_path = module_path[8:]

        return module_path


class HotReloadManager:
    """热更新管理器"""

    def __init__(self, script_dir="scripts"):
        self.script_dir = script_dir
        self.observer = None
        self.handlers = {}

    def start(self):
        """启动热更新监控"""
        event_handler = ScriptReloader(self.script_dir)
        self.observer = Observer()
        self.observer.schedule(event_handler, self.script_dir, recursive=True)
        self.observer.start()

        print(f"[HotReload] Watching: {self.script_dir}")

    def stop(self):
        """停止监控"""
        if self.observer:
            self.observer.stop()
            self.observer.join()


# 使用示例
if __name__ == "__main__":
    reloader = HotReloadManager("scripts")
    reloader.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        reloader.stop()

# 状态保持的热更新

import types
import copy

class StatefulReloader:
    """带状态保持的重新加载器"""

    def __init__(self):
        self.module_states = {}

    def reload_with_state(self, module_name):
        """重新加载模块并保持状态"""
        if module_name not in sys.modules:
            return None

        old_module = sys.modules[module_name]

        # 保存状态
        state = self.save_state(old_module)
        self.module_states[module_name] = state

        # 重新加载
        new_module = importlib.reload(old_module)

        # 恢复状态
        self.restore_state(new_module, state)

        return new_module

    def save_state(self, module):
        """保存模块状态"""
        state = {}

        for name in dir(module):
            if name.startswith('_'):
                continue

            value = getattr(module, name)

            # 保存数据属性，跳过函数/类
            if not isinstance(value, (types.FunctionType, types.ModuleType, type)):
                try:
                    # 尝试深拷贝
                    state[name] = copy.deepcopy(value)
                except:
                    # 无法拷贝的跳过
                    pass

        return state

    def restore_state(self, module, state):
        """恢复模块状态"""
        for name, value in state.items():
            if hasattr(module, name):
                try:
                    setattr(module, name, value)
                except:
                    pass


# 脚本中定义保存/恢复函数
"""
# game_logic.py

class GameState:
    def __init__(self):
        self.players = {}
        self.counter = 0

# 全局实例
game_state = GameState()

def on_load():
    print("Module loaded")

def on_unload():
    print("Module unloading")
    # 保存必要数据
    return save_game_state()

def save_game_state():
    return {
        'players': game_state.players.copy(),
        'counter': game_state.counter
    }

def load_game_state(state):
    game_state.players.update(state.get('players', {}))
    game_state.counter = state.get('counter', 0)
"""

# KBEngine 脚本热更新

"""
KBEngine 脚本热更新配置:

在 kbengine_defaults.xml 中配置:

<root>
    <reloadScripts>
        <enable>true</enable>
        <scanInterval>1.0</scanInterval>
        <scriptsPath>scripts/</scriptsPath>
    </reloadScripts>
</root>
"""

import KBEngine
import sys

class HotReloadHandler:
    """KBEngine 热更新处理器"""

    def __init__(self):
        self.entity_types = {}
        self.callbacks = []

    def onInit(self, isReload):
        """初始化回调"""
        if isReload:
            self.handle_reload()
        else:
            print("Scripts initializing...")
            self.register_entities()

    def handle_reload(self):
        """处理热更新"""
        print("Scripts reloading...")

        # 通知所有实体
        for entity_id, entity in KBEngine.entities.items():
            if hasattr(entity, 'onScriptReload'):
                try:
                    entity.onScriptReload()
                except Exception as e:
                    print(f"Error in entity {entity_id} onScriptReload: {e}")

        # 执行注册回调
        for callback in self.callbacks:
            try:
                callback()
            except Exception as e:
                print(f"Error in reload callback: {e}")

        print("Scripts reloaded successfully")

    def register_callback(self, callback):
        """注册重载回调"""
        self.callbacks.append(callback)

    def register_entities(self):
        """注册实体类型"""
        # 注册所有实体类
        KBEngine.registerEntityType("Account", Account)
        KBEngine.registerEntityType("Avatar", Avatar)
        # ...


# 实体中处理热更新
class Account(KBEngine.Account):
    def __init__(self):
        KBEngine.Account.__init__(self)
        self._state_version = 0

    def onScriptReload(self):
        """脚本热更新时调用"""
        print(f"Account {self.id}: Script reloaded")

        # 保存需要保持的状态
        self._save_reload_state()

        self._state_version += 1

    def _save_reload_state(self):
        """保存重载状态"""
        self._reload_state = {
            'playerName': self.playerName,
            'level': getattr(self, 'level', 1),
            'gold': getattr(self, 'gold', 0)
        }

    def _restore_reload_state(self):
        """恢复重载状态"""
        if hasattr(self, '_reload_state'):
            state = self._reload_state
            self.playerName = state.get('playerName', '')
            self.level = state.get('level', 1)
            self.gold = state.get('gold', 0)
            delattr(self, '_reload_state')

# 渐进式热更新策略

class ProgressiveReloader:
    """渐进式热更新器"""

    def __init__(self):
        self.phases = []
        self.current_phase = 0

    def add_phase(self, description, action):
        """添加热更新阶段"""
        self.phases.append({
            'description': description,
            'action': action
        })

    def execute(self):
        """执行渐进式热更新"""
        for i, phase in enumerate(self.phases):
            print(f"[Phase {i+1}/{len(self.phases)}] {phase['description']}")

            try:
                result = phase['action']()

                if result is False:
                    print(f"Phase {i+1} failed, aborting")
                    return False

                print(f"Phase {i+1} completed")

            except Exception as e:
                print(f"Phase {i+1} error: {e}")
                return False

        print("All phases completed successfully")
        return True


# 使用示例
def hot_update_game_logic():
    reloader = ProgressiveReloader()

    # Phase 1: 保存关键状态
    reloader.add_phase("Saving critical state", lambda: save_critical_state())

    # Phase 2: 重新加载核心模块
    reloader.add_phase("Reloading core modules", lambda: reload_core_modules())

    # Phase 3: 更新配置
    reloader.add_phase("Updating configuration", lambda: update_configuration())

    # Phase 4: 通知玩家
    reloader.add_phase("Notifying players", lambda: notify_players_reload())

    # Phase 5: 验证状态
    reloader.add_phase("Validating state", lambda: validate_reload_state())

    return reloader.execute()