diff --git a/app/adapters/system/plugin/package.py b/app/adapters/system/plugin/package.py index 06dbfb163..fad53ffa3 100644 --- a/app/adapters/system/plugin/package.py +++ b/app/adapters/system/plugin/package.py @@ -1,4 +1,4 @@ -"""插件包文件安装、快照恢复和分身处理适配器。""" +"""插件包文件安装和快照恢复适配器。""" from __future__ import annotations @@ -6,7 +6,6 @@ import asyncio import errno import hashlib import io -import re import shutil import stat import uuid @@ -2358,217 +2357,3 @@ class PluginPackageManager: exc_info=True, ) return False - - def clone( - self, - *, - plugin_id: str, - clone_id: str, - original_class_name: str, - suffix: str, - name: str, - description: str, - version: Optional[str] = None, - icon: Optional[str] = None, - ) -> tuple[bool, str]: - """复制并改写插件分身文件，任一步失败都删除不完整目标。""" - original_dir = self.__plugin_dir(plugin_id) - clone_dir = self.__plugin_dir(clone_id) - if not original_dir.is_dir(): - return False, f"原插件目录 {original_dir} 不存在" - if clone_dir.exists(): - return False, f"分身插件 {clone_id} 已存在" - - checkpoint = self.checkpoint(clone_id) - try: - shutil.copytree(original_dir, clone_dir) - success, message = self._modify_plugin_files( - plugin_dir=clone_dir, - original_class_name=original_class_name, - suffix=suffix, - name=name, - description=description, - version=version, - icon=icon, - ) - if not success: - self.rollback(checkpoint) - return False, message - self.commit(checkpoint) - logger.info(f"已复制插件目录：{original_dir} -> {clone_dir}") - return True, "文件修改成功" - except Exception as err: - try: - self.rollback(checkpoint) - except Exception as rollback_err: - logger.error( - f"清理插件分身 {clone_id} 失败：{rollback_err}", - exc_info=True, - ) - return False, f"创建插件分身文件失败：{err}" - - def _modify_plugin_files( - self, - *, - plugin_dir: Path, - original_class_name: str, - suffix: str, - name: str, - description: str, - version: Optional[str], - icon: Optional[str], - ) -> tuple[bool, str]: - """改写分身的 Python 元数据和联邦前端资源。""" - clone_class_name = f"{original_class_name}{suffix}" - init_file = plugin_dir / "__init__.py" - if init_file.exists(): - success, message = self._modify_python_file( - file_path=init_file, - original_class_name=original_class_name, - clone_class_name=clone_class_name, - name=name, - description=description, - version=version, - icon=icon, - ) - if not success: - return False, message - - dist_dir = plugin_dir / "dist" - if dist_dir.exists(): - success, message = self._modify_federation_files( - dist_dir=dist_dir, - original_class_name=original_class_name, - clone_class_name=clone_class_name, - ) - if not success: - return False, message - return True, "文件修改成功" - - @staticmethod - def _modify_python_file( - *, - file_path: Path, - original_class_name: str, - clone_class_name: str, - name: str, - description: str, - version: Optional[str], - icon: Optional[str], - ) -> tuple[bool, str]: - """改写插件主类名称、展示元数据和独立配置前缀。""" - try: - content = file_path.read_text(encoding="utf-8", errors="replace") - content = content.replace( - f"class {original_class_name}", - f"class {clone_class_name}", - ) - if name: - content = re.sub( - r'plugin_name\s*=\s*["\'][^"\']*["\']', - f'plugin_name = "{name}"', - content, - ) - if description: - content = re.sub( - r'plugin_desc\s*=\s*["\'][^"\']*["\']', - f'plugin_desc = "{description}"', - content, - ) - content = re.sub( - r'plugin_config_prefix\s*=\s*["\'][^"\']*["\']', - f'plugin_config_prefix = "{clone_class_name.lower()}_"', - content, - ) - if version: - content = re.sub( - r'plugin_version\s*=\s*["\'][^"\']*["\']', - f'plugin_version = "{version}"', - content, - ) - if icon and icon.strip(): - content = re.sub( - r'plugin_icon\s*=\s*["\'][^"\']*["\']', - f'plugin_icon = "{icon}"', - content, - ) - if "def init_plugin(self" in content: - init_index = content.index("def init_plugin(self") - content = ( - content[:init_index] - + "is_clone = True\n\n " - + content[init_index:] - ) - file_path.write_text(content, encoding="utf-8") - return True, "Python文件修改成功" - except Exception as err: - logger.error(f"修改Python文件失败：{err}") - return False, f"修改Python文件失败：{err}" - - def _modify_federation_files( - self, - *, - dist_dir: Path, - original_class_name: str, - clone_class_name: str, - ) -> tuple[bool, str]: - """改写联邦构建产物中的插件类名和样式命名空间。""" - try: - for file_path in dist_dir.rglob("*"): - if not file_path.is_file() or file_path.suffix not in {".js", ".css"}: - continue - try: - content = file_path.read_text(encoding="utf-8", errors="replace") - if file_path.suffix == ".js": - content = content.replace(original_class_name, clone_class_name) - content = content.replace( - f'"{original_class_name}"', - f'"{clone_class_name}"', - ) - content = content.replace( - f"'{original_class_name}'", - f"'{clone_class_name}'", - ) - content = content.replace( - f"css__{original_class_name}__", - f"css__{clone_class_name}__", - ) - content = content.replace( - original_class_name.lower(), - clone_class_name.lower(), - ) - file_path.write_text(content, encoding="utf-8") - except Exception as err: - logger.warning(f"修改联邦插件文件 {file_path} 失败：{err}") - self._rename_federation_assets( - dist_dir, - original_class_name, - clone_class_name, - ) - return True, "联邦插件文件修改完成" - except Exception as err: - logger.error(f"修改联邦插件文件失败：{err}") - return False, f"修改联邦插件文件失败：{err}" - - @staticmethod - def _rename_federation_assets( - dist_dir: Path, - original_class_name: str, - clone_class_name: str, - ) -> None: - """重命名包含原类名的顶层联邦资源，避免分身资源冲突。""" - try: - for file_path in dist_dir.glob("*"): - if not file_path.is_file(): - continue - if original_class_name.lower() not in file_path.name.lower(): - continue - new_name = file_path.name.replace( - original_class_name.lower(), - clone_class_name.lower(), - ) - new_path = file_path.parent / new_name - if not new_path.exists(): - file_path.rename(new_path) - except Exception as err: - logger.warning(f"重命名联邦插件资源文件失败：{err}") diff --git a/app/runtime/deprecation/notices.py b/app/runtime/deprecation/notices.py index d737c520d..ee78264bc 100644 --- a/app/runtime/deprecation/notices.py +++ b/app/runtime/deprecation/notices.py @@ -68,38 +68,6 @@ class DeprecationNotice: NOTICES: Dict[str, DeprecationNotice] = { notice.key: notice for notice in ( - DeprecationNotice( - key="PluginManager._modify_plugin_files", - subject="PluginManager._modify_plugin_files()", - stage=DeprecationStage.SILENT, - since="v3.0.0", - replacement="get_plugin_system().package._modify_plugin_files()", - reason="分身文件改写已由插件包适配器实现，此处只为旧内部调用保留转发", - ), - DeprecationNotice( - key="PluginManager._modify_python_file", - subject="PluginManager._modify_python_file()", - stage=DeprecationStage.SILENT, - since="v3.0.0", - replacement="get_plugin_system().package._modify_python_file()", - reason="Python 文件改写已由插件包适配器实现，此处只为旧内部调用保留转发", - ), - DeprecationNotice( - key="PluginManager._modify_federation_files", - subject="PluginManager._modify_federation_files()", - stage=DeprecationStage.SILENT, - since="v3.0.0", - replacement="get_plugin_system().package._modify_federation_files()", - reason="联邦文件改写已由插件包适配器实现，此处只为旧内部调用保留转发", - ), - DeprecationNotice( - key="PluginManager._rename_federation_assets", - subject="PluginManager._rename_federation_assets()", - stage=DeprecationStage.SILENT, - since="v3.0.0", - replacement="get_plugin_system().package._rename_federation_assets()", - reason="联邦资源重命名已由插件包适配器实现，此处只为旧内部调用保留转发", - ), DeprecationNotice( key="PluginHelper.find_missing_dependencies", subject="PluginHelper.find_missing_dependencies()", diff --git a/app/runtime/extensions/plugin/manager.py b/app/runtime/extensions/plugin/manager.py index 13ef138cd..6814fc201 100644 --- a/app/runtime/extensions/plugin/manager.py +++ b/app/runtime/extensions/plugin/manager.py @@ -1460,69 +1460,3 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): :raise LookupError: 插件不存在，或 plugin_id 实为某个分身自身的实例ID """ self._plugin_default_target.clear_target(plugin_id, instance_id) - - def _modify_plugin_files(self, plugin_dir: Path, original_id: str, suffix: str, - name: str, description: str, version: str = None, - icon: str = None) -> Tuple[bool, str]: - """ - 兼容旧内部调用，将分身文件改写委托给包适配器。 - :param plugin_dir: 插件目录 - :param original_id: 原插件ID - :param suffix: 分身后缀 - :param name: 分身名称 - :param description: 分身描述 - :param version: 自定义版本号 - :param icon: 自定义图标URL - :return: (是否成功, 错误信息) - """ - original_plugin_class = self._plugins.get(original_id) - if not original_plugin_class: - return False, f"无法获取原插件类 {original_id}" - return self._plugin_runtime.system().modify_plugin_files( - plugin_dir=plugin_dir, - original_class_name=original_plugin_class.__name__, - suffix=suffix, - name=name, - description=description, - version=version, - icon=icon, - ) - - @staticmethod - def _modify_python_file(file_path: Path, original_class_name: str, - clone_class_name: str, name: str, description: str, - version: str = None, icon: str = None) -> Tuple[bool, str]: - """ - 兼容旧内部调用，将 Python 文件改写委托给包适配器。 - """ - return PluginManager()._plugin_runtime.system().modify_python_file( - file_path=file_path, - original_class_name=original_class_name, - clone_class_name=clone_class_name, - name=name, - description=description, - version=version, - icon=icon, - ) - - def _modify_federation_files(self, dist_dir: Path, original_class_name: str, - clone_class_name: str) -> Tuple[bool, str]: - """ - 兼容旧内部调用，将联邦文件改写委托给包适配器。 - """ - return self._plugin_runtime.system().modify_federation_files( - dist_dir=dist_dir, - original_class_name=original_class_name, - clone_class_name=clone_class_name, - ) - - @staticmethod - def _rename_federation_assets(dist_dir: Path, original_class_name: str, clone_class_name: str): - """ - 兼容旧内部调用，将资源重命名委托给包适配器。 - """ - PluginManager()._plugin_runtime.system().rename_federation_assets( - dist_dir, - original_class_name, - clone_class_name, - ) diff --git a/app/runtime/extensions/plugin/system.py b/app/runtime/extensions/plugin/system.py index 9e6eab44a..4d47fa880 100644 --- a/app/runtime/extensions/plugin/system.py +++ b/app/runtime/extensions/plugin/system.py @@ -4,7 +4,7 @@ from __future__ import annotations from collections.abc import Callable from pathlib import Path -from typing import Any, Optional, cast +from typing import Any, Optional class PluginSystemServices: @@ -95,34 +95,6 @@ class PluginSystemServices: """通过唯一包事务 owner 删除插件物理目录。""" return bool(self.package.remove_plugin(plugin_id)) - def modify_plugin_files(self, **kwargs: Any) -> tuple[bool, str]: - """把旧分身源码改写调用收口到唯一包适配器。""" - return cast(tuple[bool, str], self.package._modify_plugin_files(**kwargs)) - - def modify_python_file(self, **kwargs: Any) -> tuple[bool, str]: - """把旧 Python 类改写调用收口到唯一包适配器。""" - return cast(tuple[bool, str], self.package._modify_python_file(**kwargs)) - - def modify_federation_files(self, **kwargs: Any) -> tuple[bool, str]: - """把旧联邦构建产物改写调用收口到唯一包适配器。""" - return cast( - tuple[bool, str], - self.package._modify_federation_files(**kwargs), - ) - - def rename_federation_assets( - self, - dist_dir: Path, - original_class_name: str, - clone_class_name: str, - ) -> None: - """把旧联邦资源重命名调用收口到唯一包适配器。""" - self.package._rename_federation_assets( - dist_dir, - original_class_name, - clone_class_name, - ) - _services: Optional[PluginSystemServices] = None diff --git a/tests/fixtures/architecture/mypy-baseline.json b/tests/fixtures/architecture/mypy-baseline.json index ca50740ff..30e08a6fa 100644 --- a/tests/fixtures/architecture/mypy-baseline.json +++ b/tests/fixtures/architecture/mypy-baseline.json @@ -2533,10 +2533,10 @@ "type-arg": 2 }, "app/runtime/extensions/plugin/manager.py": { - "assignment": 8, + "assignment": 4, "attr-defined": 1, "no-untyped-call": 1, - "no-untyped-def": 12, + "no-untyped-def": 11, "type-arg": 10 }, "app/runtime/extensions/plugin/metadata.py": { diff --git a/tests/test_plugin_package_manager.py b/tests/test_plugin_package_manager.py index c66cc6019..2a22f47e1 100644 --- a/tests/test_plugin_package_manager.py +++ b/tests/test_plugin_package_manager.py @@ -447,47 +447,3 @@ def test_local_sync_failure_restores_previous_runtime_copy(monkeypatch, tmp_path assert manager.sync_local("DemoPlugin", missing_source) is False assert source_file.read_text(encoding="utf-8") == "stable" - - -def test_clone_rewrites_python_and_federation_assets(monkeypatch, tmp_path): - """插件分身文件处理应由包适配器完成并隔离配置命名空间。""" - manager = _manager(monkeypatch, tmp_path) - plugin_dir = tmp_path / "app" / "plugins" / "demoplugin" - dist_dir = plugin_dir / "dist" - dist_dir.mkdir(parents=True) - (plugin_dir / "__init__.py").write_text( - "class DemoPlugin:\n" - " plugin_name = 'Demo'\n" - " plugin_desc = 'Description'\n" - " plugin_config_prefix = 'demo_'\n" - " plugin_version = '1.0.0'\n" - " plugin_icon = 'old.png'\n" - " def init_plugin(self, config=None):\n" - " pass\n", - encoding="utf-8", - ) - (dist_dir / "demoplugin.js").write_text( - "const name = 'DemoPlugin'; const css = 'css__DemoPlugin__root';", - encoding="utf-8", - ) - - success, message = manager.clone( - plugin_id="DemoPlugin", - clone_id="DemoPluginBlue", - original_class_name="DemoPlugin", - suffix="blue", - name="Demo Blue", - description="Blue clone", - version="2.0.0", - icon="blue.png", - ) - - clone_dir = tmp_path / "app" / "plugins" / "demopluginblue" - clone_source = (clone_dir / "__init__.py").read_text(encoding="utf-8") - assert success is True - assert message == "文件修改成功" - assert "class DemoPluginblue" in clone_source - assert 'plugin_name = "Demo Blue"' in clone_source - assert 'plugin_config_prefix = "demopluginblue_"' in clone_source - assert "is_clone = True" in clone_source - assert (clone_dir / "dist" / "demopluginblue.js").is_file()