perf: 提取模块清理逻辑到 _purge_modules 方法

astrbot/core/star/star_manager.py

@@ -169,12 +169,16 @@ class PluginManager:
     def _get_plugin_related_modules(
         self, plugin_root_dir: str, is_reserved: bool = False
     ) -> list[str]:
-        """获取插件相关的所有模块名
+        """获取与指定插件相关的所有已加载模块名
+
+        根据插件根目录名和是否为保留插件，从 sys.modules 中筛选出相关的模块名
+
         Args:
             plugin_root_dir: 插件根目录名
-            is_reserved: 是否是保留插件
+            is_reserved: 是否是保留插件，影响模块路径前缀
+
         Returns:
-            模块名列表
+            list[str]: 与该插件相关的模块名列表
         """
         prefix = "packages." if is_reserved else "data.plugins."
         return [
@@ -183,6 +187,38 @@ class PluginManager:
             if key.startswith(f"{prefix}{plugin_root_dir}")
         ]
 
+    def _purge_modules(
+        self,
+        module_patterns: list[str] = None,
+        root_dir_name: str = None,
+        is_reserved: bool = False,
+    ):
+        """从 sys.modules 中移除指定的模块
+
+        可以基于模块名模式或插件目录名移除模块，用于清理插件相关的模块缓存
+
+        Args:
+            module_patterns: 要移除的模块名模式列表（例如 ["data.plugins", "packages"]）
+            root_dir_name: 插件根目录名，用于移除与该插件相关的所有模块
+            is_reserved: 插件是否为保留插件（影响模块路径前缀）
+        """
+        if module_patterns:
+            for pattern in module_patterns:
+                for key in list(sys.modules.keys()):
+                    if key.startswith(pattern):
+                        del sys.modules[key]
+                        logger.debug(f"删除模块 {key}")
+
+        if root_dir_name:
+            for module_name in self._get_plugin_related_modules(
+                root_dir_name, is_reserved
+            ):
+                try:
+                    del sys.modules[module_name]
+                    logger.debug(f"删除模块 {module_name}")
+                except KeyError:
+                    logger.warning(f"模块 {module_name} 未载入")
+
     async def reload(self, specified_plugin_name=None):
         """重新加载插件
 
@@ -194,13 +230,6 @@ class PluginManager:
             tuple: 返回 load() 方法的结果，包含 (success, error_message)
                 - success (bool): 重载是否成功
                 - error_message (str|None): 错误信息，成功时为 None
-
-        流程:
-            1. 如果指定了插件名，查找对应的模块路径
-            2. 终止现有插件实例
-            3. 解绑插件相关的处理器和注册信息
-            4. 清理系统模块缓存
-            5. 重新加载插件
         """
         specified_module_path = None
         if specified_plugin_name:
@@ -226,9 +255,6 @@ class PluginManager:
             star_handlers_registry.clear()
             star_map.clear()
             star_registry.clear()
-            for key in list(sys.modules.keys()):
-                if key.startswith("data.plugins") or key.startswith("packages"):
-                    del sys.modules[key]
         else:
             # 只重载指定插件
             smd = star_map.get(specified_module_path)
@@ -243,14 +269,6 @@ class PluginManager:
 
                 await self._unbind_plugin(smd.name, specified_module_path)
 
-                for module_name in self._get_plugin_related_modules(
-                    smd.root_dir_name, smd.reserved
-                ):
-                    try:
-                        del sys.modules[module_name]
-                    except KeyError:
-                        pass
-
         return await self.load(specified_module_path)
 
     async def load(self, specified_module_path=None, specified_dir_name=None):
@@ -265,23 +283,6 @@ class PluginManager:
             tuple: (success, error_message)
                 - success (bool): 是否全部加载成功
                 - error_message (str|None): 错误信息，成功时为 None
-
-        执行步骤:
-        1. 获取已禁用的插件和 LLM 工具列表
-        2. 获取所有插件模块信息
-        3. 对每个插件模块:
-            - 导入模块并实例化插件类
-            - 加载插件配置(如果存在 _conf_schema.json)
-            - 处理插件元数据(metadata)
-            - 绑定事件处理器和 LLM 工具
-            - 执行插件的 initialize() 方法
-        4. 记录加载失败的插件信息
-        5. 清理 pip.main 产生的日志处理器
-
-        注意:
-        - 插件可以通过装饰器方式注册(v3.4.0+)或旧版本方式注册
-        - 禁用的插件不会被实例化
-        - 支持自定义命令权限过滤
         """
         inactivated_plugins: list = sp.get("inactivated_plugins", [])
         inactivated_llm_tools: list = sp.get("inactivated_llm_tools", [])
@@ -515,24 +516,19 @@ class PluginManager:
             return False, fail_rec
 
     async def install_plugin(self, repo_url: str, proxy=""):
-        """从仓库 URL 安装插件。
+        """从仓库 URL 安装插件
 
-        参数:
+        从指定的仓库 URL 下载并安装插件，然后加载该插件到系统中
+
+        Args:
             repo_url (str): 要安装的插件仓库 URL
             proxy (str, optional): 用于下载的代理服务器。默认为空字符串。
 
-        返回:
+        Returns:
             dict | None: 安装成功时返回包含插件信息的字典:
                 - repo: 插件的仓库 URL
                 - readme: README.md 文件的内容(如果存在)
                 如果找不到插件元数据则返回 None。
-
-        函数执行步骤:
-        1. 从仓库下载并安装插件
-        2. 将插件重新加载到系统中
-        3. 尝试使用目录名称查找插件元数据
-        4. 提取 README.md 内容(如果可用)
-        5. 返回仓库和说明文档信息
         """
         plugin_path = await self.updator.install(repo_url, proxy)
         # reload the plugin
@@ -575,12 +571,6 @@ class PluginManager:
 
         Raises:
             Exception: 当插件不存在、是保留插件时，或删除插件文件夹失败时抛出异常
-
-        执行以下步骤:
-        1. 检查插件是否存在且不是保留插件
-        2. 调用插件的终止方法
-        3. 从注册表中解绑插件
-        4. 删除插件文件夹
         """
         plugin = self.context.get_registered_star(plugin_name)
         if not plugin:
@@ -615,12 +605,6 @@ class PluginManager:
         Args:
             plugin_name: 要解绑的插件名称
             plugin_module_path: 插件的完整模块路径
-
-        该方法会执行以下操作：
-        1. 从 star_map 和 star_registry 中移除插件
-        2. 移除该插件注册的所有处理函数
-        3. 清理处理函数映射
-        4. 从 sys.modules 中卸载所有相关的插件模块
         """
         plugin = None
         del star_map[plugin_module_path]
@@ -644,14 +628,9 @@ class PluginManager:
         ]:
             del star_handlers_registry.star_handlers_map[k]
 
-        for module_name in self._get_plugin_related_modules(
-            plugin.root_dir_name, plugin.reserved
-        ):
-            try:
-                del sys.modules[module_name]
-                logger.debug(f"删除模块 {module_name}")
-            except KeyError:
-                logger.warning(f"模块 {module_name} 未载入")
+        self._purge_modules(
+            root_dir_name=plugin.root_dir_name, is_reserved=plugin.reserved
+        )
 
     async def update_plugin(self, plugin_name: str, proxy=""):
         """升级一个插件"""