Merge branch 'dev' of https://github.com/MaiM-with-u/MaiBot into dev
This commit is contained in:
@@ -23,6 +23,14 @@ from src.plugin_system.core.plugin_manager import plugin_manager
|
||||
from src.plugin_system.core.component_registry import component_registry
|
||||
from src.plugin_system.core.dependency_manager import dependency_manager
|
||||
|
||||
# 导入工具模块
|
||||
from src.plugin_system.utils import (
|
||||
ManifestValidator,
|
||||
ManifestGenerator,
|
||||
validate_plugin_manifest,
|
||||
generate_plugin_manifest,
|
||||
)
|
||||
|
||||
|
||||
__version__ = "1.0.0"
|
||||
|
||||
@@ -47,4 +55,9 @@ __all__ = [
|
||||
# 装饰器
|
||||
"register_plugin",
|
||||
"ConfigField",
|
||||
# 工具函数
|
||||
"ManifestValidator",
|
||||
"ManifestGenerator",
|
||||
"validate_plugin_manifest",
|
||||
"generate_plugin_manifest",
|
||||
]
|
||||
|
||||
@@ -3,6 +3,7 @@ from typing import Dict, List, Type, Optional, Any, Union
|
||||
import os
|
||||
import inspect
|
||||
import toml
|
||||
import json
|
||||
from src.common.logger import get_logger
|
||||
from src.plugin_system.base.component_types import (
|
||||
PluginInfo,
|
||||
@@ -37,6 +38,10 @@ class BasePlugin(ABC):
|
||||
python_dependencies: List[PythonDependency] = [] # Python包依赖
|
||||
config_file_name: Optional[str] = None # 配置文件名
|
||||
|
||||
# 新增:manifest文件相关
|
||||
manifest_file_name: str = "_manifest.json" # manifest文件名
|
||||
manifest_data: Dict[str, Any] = {} # manifest数据
|
||||
|
||||
# 新增:配置定义
|
||||
config_schema: Dict[str, Union[Dict[str, ConfigField], str]] = {}
|
||||
config_section_descriptions: Dict[str, str] = {}
|
||||
@@ -51,13 +56,14 @@ class BasePlugin(ABC):
|
||||
self.plugin_dir = plugin_dir # 插件目录路径
|
||||
self.log_prefix = f"[Plugin:{self.plugin_name}]"
|
||||
|
||||
# 加载manifest文件
|
||||
self._load_manifest()
|
||||
|
||||
# 验证插件信息
|
||||
self._validate_plugin_info()
|
||||
|
||||
# 加载插件配置
|
||||
self._load_plugin_config()
|
||||
|
||||
# 创建插件信息对象
|
||||
self._load_plugin_config() # 创建插件信息对象
|
||||
self.plugin_info = PluginInfo(
|
||||
name=self.plugin_name,
|
||||
description=self.plugin_description,
|
||||
@@ -68,6 +74,15 @@ class BasePlugin(ABC):
|
||||
config_file=self.config_file_name or "",
|
||||
dependencies=self.dependencies.copy(),
|
||||
python_dependencies=self.python_dependencies.copy(),
|
||||
# 新增:manifest相关信息
|
||||
manifest_data=self.manifest_data.copy(),
|
||||
license=self.get_manifest_info("license", ""),
|
||||
homepage_url=self.get_manifest_info("homepage_url", ""),
|
||||
repository_url=self.get_manifest_info("repository_url", ""),
|
||||
keywords=self.get_manifest_info("keywords", []).copy() if self.get_manifest_info("keywords") else [],
|
||||
categories=self.get_manifest_info("categories", []).copy() if self.get_manifest_info("categories") else [],
|
||||
min_host_version=self.get_manifest_info("host_application.min_version", ""),
|
||||
max_host_version=self.get_manifest_info("host_application.max_version", ""),
|
||||
)
|
||||
|
||||
logger.debug(f"{self.log_prefix} 插件基类初始化完成")
|
||||
@@ -79,6 +94,135 @@ class BasePlugin(ABC):
|
||||
if not self.plugin_description:
|
||||
raise ValueError(f"插件 {self.plugin_name} 必须定义 plugin_description")
|
||||
|
||||
def _load_manifest(self):
|
||||
"""加载manifest文件(强制要求)"""
|
||||
if not self.plugin_dir:
|
||||
raise ValueError(f"{self.log_prefix} 没有插件目录路径,无法加载manifest")
|
||||
|
||||
manifest_path = os.path.join(self.plugin_dir, self.manifest_file_name)
|
||||
|
||||
if not os.path.exists(manifest_path):
|
||||
error_msg = f"{self.log_prefix} 缺少必需的manifest文件: {manifest_path}"
|
||||
logger.error(error_msg)
|
||||
raise FileNotFoundError(error_msg)
|
||||
|
||||
try:
|
||||
with open(manifest_path, "r", encoding="utf-8") as f:
|
||||
self.manifest_data = json.load(f)
|
||||
|
||||
logger.debug(f"{self.log_prefix} 成功加载manifest文件: {manifest_path}")
|
||||
|
||||
# 验证manifest格式
|
||||
self._validate_manifest()
|
||||
|
||||
# 从manifest覆盖插件基本信息(如果插件类中未定义)
|
||||
self._apply_manifest_overrides()
|
||||
|
||||
except json.JSONDecodeError as e:
|
||||
error_msg = f"{self.log_prefix} manifest文件格式错误: {e}"
|
||||
logger.error(error_msg)
|
||||
raise ValueError(error_msg) # noqa
|
||||
except IOError as e:
|
||||
error_msg = f"{self.log_prefix} 读取manifest文件失败: {e}"
|
||||
logger.error(error_msg)
|
||||
raise IOError(error_msg) # noqa
|
||||
|
||||
def _apply_manifest_overrides(self):
|
||||
"""从manifest文件覆盖插件信息"""
|
||||
if not self.manifest_data:
|
||||
return
|
||||
|
||||
# 如果插件类中的信息为空,则从manifest中获取
|
||||
if not self.plugin_name:
|
||||
self.plugin_name = self.manifest_data.get("name", "")
|
||||
|
||||
if not self.plugin_description:
|
||||
self.plugin_description = self.manifest_data.get("description", "")
|
||||
|
||||
if self.plugin_version == "1.0.0": # 默认版本
|
||||
self.plugin_version = self.manifest_data.get("version", "1.0.0")
|
||||
|
||||
if not self.plugin_author:
|
||||
author_info = self.manifest_data.get("author", {})
|
||||
if isinstance(author_info, dict):
|
||||
self.plugin_author = author_info.get("name", "")
|
||||
else:
|
||||
self.plugin_author = str(author_info)
|
||||
|
||||
def _validate_manifest(self):
|
||||
"""验证manifest文件格式(使用强化的验证器)"""
|
||||
if not self.manifest_data:
|
||||
return
|
||||
|
||||
# 导入验证器
|
||||
from src.plugin_system.utils.manifest_utils import ManifestValidator
|
||||
|
||||
validator = ManifestValidator()
|
||||
is_valid = validator.validate_manifest(self.manifest_data)
|
||||
|
||||
# 记录验证结果
|
||||
if validator.validation_errors or validator.validation_warnings:
|
||||
report = validator.get_validation_report()
|
||||
logger.info(f"{self.log_prefix} Manifest验证结果:\n{report}")
|
||||
|
||||
# 如果有验证错误,抛出异常
|
||||
if not is_valid:
|
||||
error_msg = f"{self.log_prefix} Manifest文件验证失败"
|
||||
if validator.validation_errors:
|
||||
error_msg += f": {'; '.join(validator.validation_errors)}"
|
||||
raise ValueError(error_msg)
|
||||
|
||||
def _generate_default_manifest(self, manifest_path: str):
|
||||
"""生成默认的manifest文件"""
|
||||
if not self.plugin_name:
|
||||
logger.debug(f"{self.log_prefix} 插件名称未定义,无法生成默认manifest")
|
||||
return
|
||||
|
||||
default_manifest = {
|
||||
"manifest_version": 1,
|
||||
"name": self.plugin_name,
|
||||
"version": self.plugin_version,
|
||||
"description": self.plugin_description or "插件描述",
|
||||
"author": {"name": self.plugin_author or "Unknown", "url": ""},
|
||||
"license": "MIT",
|
||||
"host_application": {"min_version": "1.0.0", "max_version": "4.0.0"},
|
||||
"keywords": [],
|
||||
"categories": [],
|
||||
"default_locale": "zh-CN",
|
||||
"locales_path": "_locales",
|
||||
}
|
||||
|
||||
try:
|
||||
with open(manifest_path, "w", encoding="utf-8") as f:
|
||||
json.dump(default_manifest, f, ensure_ascii=False, indent=2)
|
||||
logger.info(f"{self.log_prefix} 已生成默认manifest文件: {manifest_path}")
|
||||
except IOError as e:
|
||||
logger.error(f"{self.log_prefix} 保存默认manifest文件失败: {e}")
|
||||
|
||||
def get_manifest_info(self, key: str, default: Any = None) -> Any:
|
||||
"""获取manifest信息
|
||||
|
||||
Args:
|
||||
key: 信息键,支持点分割的嵌套键(如 "author.name")
|
||||
default: 默认值
|
||||
|
||||
Returns:
|
||||
Any: 对应的值
|
||||
"""
|
||||
if not self.manifest_data:
|
||||
return default
|
||||
|
||||
keys = key.split(".")
|
||||
value = self.manifest_data
|
||||
|
||||
for k in keys:
|
||||
if isinstance(value, dict) and k in value:
|
||||
value = value[k]
|
||||
else:
|
||||
return default
|
||||
|
||||
return value
|
||||
|
||||
def _generate_and_save_default_config(self, config_file_path: str):
|
||||
"""根据插件的Schema生成并保存默认配置文件"""
|
||||
if not self.config_schema:
|
||||
|
||||
@@ -131,6 +131,15 @@ class PluginInfo:
|
||||
python_dependencies: List[PythonDependency] = None # Python包依赖
|
||||
config_file: str = "" # 配置文件路径
|
||||
metadata: Dict[str, Any] = None # 额外元数据
|
||||
# 新增:manifest相关信息
|
||||
manifest_data: Dict[str, Any] = None # manifest文件数据
|
||||
license: str = "" # 插件许可证
|
||||
homepage_url: str = "" # 插件主页
|
||||
repository_url: str = "" # 插件仓库地址
|
||||
keywords: List[str] = None # 插件关键词
|
||||
categories: List[str] = None # 插件分类
|
||||
min_host_version: str = "" # 最低主机版本要求
|
||||
max_host_version: str = "" # 最高主机版本要求
|
||||
|
||||
def __post_init__(self):
|
||||
if self.components is None:
|
||||
@@ -141,6 +150,12 @@ class PluginInfo:
|
||||
self.python_dependencies = []
|
||||
if self.metadata is None:
|
||||
self.metadata = {}
|
||||
if self.manifest_data is None:
|
||||
self.manifest_data = {}
|
||||
if self.keywords is None:
|
||||
self.keywords = []
|
||||
if self.categories is None:
|
||||
self.categories = []
|
||||
|
||||
def get_missing_packages(self) -> List[PythonDependency]:
|
||||
"""检查缺失的Python包"""
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
from typing import Dict, List, Optional, Any, TYPE_CHECKING
|
||||
from typing import Dict, List, Optional, Any, TYPE_CHECKING, Tuple
|
||||
import os
|
||||
import importlib
|
||||
import importlib.util
|
||||
@@ -83,44 +83,86 @@ class PluginManager:
|
||||
total_failed_registration = 0
|
||||
|
||||
for plugin_name, plugin_class in plugin_classes.items():
|
||||
# 使用记录的插件目录路径
|
||||
plugin_dir = self.plugin_paths.get(plugin_name)
|
||||
try:
|
||||
# 使用记录的插件目录路径
|
||||
plugin_dir = self.plugin_paths.get(plugin_name)
|
||||
|
||||
# 如果没有记录,则尝试查找(fallback)
|
||||
if not plugin_dir:
|
||||
plugin_dir = self._find_plugin_directory(plugin_class)
|
||||
if plugin_dir:
|
||||
self.plugin_paths[plugin_name] = plugin_dir
|
||||
# 如果没有记录,则尝试查找(fallback)
|
||||
if not plugin_dir:
|
||||
plugin_dir = self._find_plugin_directory(plugin_class)
|
||||
if plugin_dir:
|
||||
self.plugin_paths[plugin_name] = plugin_dir # 实例化插件(可能因为缺少manifest而失败)
|
||||
plugin_instance = plugin_class(plugin_dir=plugin_dir)
|
||||
|
||||
plugin_instance = plugin_class(plugin_dir=plugin_dir)
|
||||
# 检查插件是否启用
|
||||
if not plugin_instance.enable_plugin:
|
||||
logger.info(f"插件 {plugin_name} 已禁用,跳过加载")
|
||||
continue
|
||||
|
||||
# 检查插件是否启用
|
||||
if not plugin_instance.enable_plugin:
|
||||
logger.info(f"插件 {plugin_name} 已禁用,跳过加载")
|
||||
continue
|
||||
# 检查版本兼容性
|
||||
is_compatible, compatibility_error = self.check_plugin_version_compatibility(
|
||||
plugin_name, plugin_instance.manifest_data
|
||||
)
|
||||
if not is_compatible:
|
||||
total_failed_registration += 1
|
||||
self.failed_plugins[plugin_name] = compatibility_error
|
||||
logger.error(f"❌ 插件加载失败: {plugin_name} - {compatibility_error}")
|
||||
continue
|
||||
|
||||
if plugin_instance.register_plugin():
|
||||
total_registered += 1
|
||||
self.loaded_plugins[plugin_name] = plugin_instance
|
||||
if plugin_instance.register_plugin():
|
||||
total_registered += 1
|
||||
self.loaded_plugins[plugin_name] = plugin_instance
|
||||
|
||||
# 📊 显示插件详细信息
|
||||
plugin_info = component_registry.get_plugin_info(plugin_name)
|
||||
if plugin_info:
|
||||
component_types = {}
|
||||
for comp in plugin_info.components:
|
||||
comp_type = comp.component_type.name
|
||||
component_types[comp_type] = component_types.get(comp_type, 0) + 1
|
||||
# 📊 显示插件详细信息
|
||||
plugin_info = component_registry.get_plugin_info(plugin_name)
|
||||
if plugin_info:
|
||||
component_types = {}
|
||||
for comp in plugin_info.components:
|
||||
comp_type = comp.component_type.name
|
||||
component_types[comp_type] = component_types.get(comp_type, 0) + 1
|
||||
|
||||
components_str = ", ".join([f"{count}个{ctype}" for ctype, count in component_types.items()])
|
||||
logger.info(
|
||||
f"✅ 插件加载成功: {plugin_name} v{plugin_info.version} ({components_str}) - {plugin_info.description}"
|
||||
)
|
||||
components_str = ", ".join([f"{count}个{ctype}" for ctype, count in component_types.items()])
|
||||
|
||||
# 显示manifest信息
|
||||
manifest_info = ""
|
||||
if plugin_info.license:
|
||||
manifest_info += f" [{plugin_info.license}]"
|
||||
if plugin_info.keywords:
|
||||
manifest_info += f" 关键词: {', '.join(plugin_info.keywords[:3])}" # 只显示前3个关键词
|
||||
if len(plugin_info.keywords) > 3:
|
||||
manifest_info += "..."
|
||||
|
||||
logger.info(
|
||||
f"✅ 插件加载成功: {plugin_name} v{plugin_info.version} ({components_str}){manifest_info} - {plugin_info.description}"
|
||||
)
|
||||
else:
|
||||
logger.info(f"✅ 插件加载成功: {plugin_name}")
|
||||
else:
|
||||
logger.info(f"✅ 插件加载成功: {plugin_name}")
|
||||
else:
|
||||
total_failed_registration += 1
|
||||
self.failed_plugins[plugin_name] = "插件注册失败"
|
||||
logger.error(f"❌ 插件注册失败: {plugin_name}")
|
||||
|
||||
except FileNotFoundError as e:
|
||||
# manifest文件缺失
|
||||
total_failed_registration += 1
|
||||
self.failed_plugins[plugin_name] = "插件注册失败"
|
||||
logger.error(f"❌ 插件加载失败: {plugin_name}")
|
||||
error_msg = f"缺少manifest文件: {str(e)}"
|
||||
self.failed_plugins[plugin_name] = error_msg
|
||||
logger.error(f"❌ 插件加载失败: {plugin_name} - {error_msg}")
|
||||
|
||||
except ValueError as e:
|
||||
# manifest文件格式错误或验证失败
|
||||
total_failed_registration += 1
|
||||
error_msg = f"manifest验证失败: {str(e)}"
|
||||
self.failed_plugins[plugin_name] = error_msg
|
||||
logger.error(f"❌ 插件加载失败: {plugin_name} - {error_msg}")
|
||||
|
||||
except Exception as e:
|
||||
# 其他错误
|
||||
total_failed_registration += 1
|
||||
error_msg = f"未知错误: {str(e)}"
|
||||
self.failed_plugins[plugin_name] = error_msg
|
||||
logger.error(f"❌ 插件加载失败: {plugin_name} - {error_msg}")
|
||||
logger.debug("详细错误信息: ", exc_info=True)
|
||||
|
||||
# 获取组件统计信息
|
||||
stats = component_registry.get_registry_stats()
|
||||
@@ -136,19 +178,28 @@ class PluginManager:
|
||||
f"📊 总览: {total_registered}个插件, {total_components}个组件 (Action: {action_count}, Command: {command_count})"
|
||||
)
|
||||
|
||||
# 显示详细的插件列表
|
||||
logger.info("📋 已加载插件详情:")
|
||||
# 显示详细的插件列表 logger.info("📋 已加载插件详情:")
|
||||
for plugin_name, _plugin_class in self.loaded_plugins.items():
|
||||
plugin_info = component_registry.get_plugin_info(plugin_name)
|
||||
if plugin_info:
|
||||
# 插件基本信息
|
||||
version_info = f"v{plugin_info.version}" if plugin_info.version else ""
|
||||
author_info = f"by {plugin_info.author}" if plugin_info.author else "unknown"
|
||||
info_parts = [part for part in [version_info, author_info] if part]
|
||||
license_info = f"[{plugin_info.license}]" if plugin_info.license else ""
|
||||
info_parts = [part for part in [version_info, author_info, license_info] if part]
|
||||
extra_info = f" ({', '.join(info_parts)})" if info_parts else ""
|
||||
|
||||
logger.info(f" 📦 {plugin_name}{extra_info}")
|
||||
|
||||
# Manifest信息
|
||||
if plugin_info.manifest_data:
|
||||
if plugin_info.keywords:
|
||||
logger.info(f" 🏷️ 关键词: {', '.join(plugin_info.keywords)}")
|
||||
if plugin_info.categories:
|
||||
logger.info(f" 📁 分类: {', '.join(plugin_info.categories)}")
|
||||
if plugin_info.homepage_url:
|
||||
logger.info(f" 🌐 主页: {plugin_info.homepage_url}")
|
||||
|
||||
# 组件列表
|
||||
if plugin_info.components:
|
||||
action_components = [c for c in plugin_info.components if c.component_type.name == "ACTION"]
|
||||
@@ -162,6 +213,18 @@ class PluginManager:
|
||||
command_names = [c.name for c in command_components]
|
||||
logger.info(f" ⚡ Command组件: {', '.join(command_names)}")
|
||||
|
||||
# 版本兼容性信息
|
||||
if plugin_info.min_host_version or plugin_info.max_host_version:
|
||||
version_range = ""
|
||||
if plugin_info.min_host_version:
|
||||
version_range += f">={plugin_info.min_host_version}"
|
||||
if plugin_info.max_host_version:
|
||||
if version_range:
|
||||
version_range += f", <={plugin_info.max_host_version}"
|
||||
else:
|
||||
version_range += f"<={plugin_info.max_host_version}"
|
||||
logger.info(f" 📋 兼容版本: {version_range}")
|
||||
|
||||
# 依赖信息
|
||||
if plugin_info.dependencies:
|
||||
logger.info(f" 🔗 依赖: {', '.join(plugin_info.dependencies)}")
|
||||
@@ -453,6 +516,46 @@ class PluginManager:
|
||||
|
||||
return dependency_manager.generate_requirements_file(all_dependencies, output_path)
|
||||
|
||||
def check_plugin_version_compatibility(self, plugin_name: str, manifest_data: Dict[str, Any]) -> Tuple[bool, str]:
|
||||
"""检查插件版本兼容性
|
||||
|
||||
Args:
|
||||
plugin_name: 插件名称
|
||||
manifest_data: manifest数据
|
||||
|
||||
Returns:
|
||||
Tuple[bool, str]: (是否兼容, 错误信息)
|
||||
"""
|
||||
if "host_application" not in manifest_data:
|
||||
# 没有版本要求,默认兼容
|
||||
return True, ""
|
||||
|
||||
host_app = manifest_data["host_application"]
|
||||
if not isinstance(host_app, dict):
|
||||
return True, ""
|
||||
|
||||
min_version = host_app.get("min_version", "")
|
||||
max_version = host_app.get("max_version", "")
|
||||
|
||||
if not min_version and not max_version:
|
||||
return True, ""
|
||||
|
||||
try:
|
||||
from src.plugin_system.utils.manifest_utils import VersionComparator
|
||||
|
||||
current_version = VersionComparator.get_current_host_version()
|
||||
is_compatible, error_msg = VersionComparator.is_version_in_range(current_version, min_version, max_version)
|
||||
|
||||
if not is_compatible:
|
||||
return False, f"版本不兼容: {error_msg}"
|
||||
else:
|
||||
logger.debug(f"插件 {plugin_name} 版本兼容性检查通过")
|
||||
return True, ""
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"插件 {plugin_name} 版本兼容性检查失败: {e}")
|
||||
return True, "" # 检查失败时默认允许加载
|
||||
|
||||
|
||||
# 全局插件管理器实例
|
||||
plugin_manager = PluginManager()
|
||||
|
||||
14
src/plugin_system/utils/__init__.py
Normal file
14
src/plugin_system/utils/__init__.py
Normal file
@@ -0,0 +1,14 @@
|
||||
"""
|
||||
插件系统工具模块
|
||||
|
||||
提供插件开发和管理的实用工具
|
||||
"""
|
||||
|
||||
from src.plugin_system.utils.manifest_utils import (
|
||||
ManifestValidator,
|
||||
ManifestGenerator,
|
||||
validate_plugin_manifest,
|
||||
generate_plugin_manifest,
|
||||
)
|
||||
|
||||
__all__ = ["ManifestValidator", "ManifestGenerator", "validate_plugin_manifest", "generate_plugin_manifest"]
|
||||
445
src/plugin_system/utils/manifest_utils.py
Normal file
445
src/plugin_system/utils/manifest_utils.py
Normal file
@@ -0,0 +1,445 @@
|
||||
"""
|
||||
插件Manifest工具模块
|
||||
|
||||
提供manifest文件的验证、生成和管理功能
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
from typing import Dict, Any, Optional, Tuple
|
||||
from src.common.logger import get_logger
|
||||
from src.config.config import MMC_VERSION
|
||||
|
||||
logger = get_logger("manifest_utils")
|
||||
|
||||
|
||||
class VersionComparator:
|
||||
"""版本号比较器
|
||||
|
||||
支持语义化版本号比较,自动处理snapshot版本
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def normalize_version(version: str) -> str:
|
||||
"""标准化版本号,移除snapshot标识
|
||||
|
||||
Args:
|
||||
version: 原始版本号,如 "0.8.0-snapshot.1"
|
||||
|
||||
Returns:
|
||||
str: 标准化后的版本号,如 "0.8.0"
|
||||
"""
|
||||
if not version:
|
||||
return "0.0.0"
|
||||
|
||||
# 移除snapshot部分
|
||||
normalized = re.sub(r"-snapshot\.\d+", "", version.strip())
|
||||
|
||||
# 确保版本号格式正确
|
||||
if not re.match(r"^\d+(\.\d+){0,2}$", normalized):
|
||||
# 如果不是有效的版本号格式,返回默认版本
|
||||
return "0.0.0"
|
||||
|
||||
# 尝试补全版本号
|
||||
parts = normalized.split(".")
|
||||
while len(parts) < 3:
|
||||
parts.append("0")
|
||||
normalized = ".".join(parts[:3])
|
||||
|
||||
return normalized
|
||||
|
||||
@staticmethod
|
||||
def parse_version(version: str) -> Tuple[int, int, int]:
|
||||
"""解析版本号为元组
|
||||
|
||||
Args:
|
||||
version: 版本号字符串
|
||||
|
||||
Returns:
|
||||
Tuple[int, int, int]: (major, minor, patch)
|
||||
"""
|
||||
normalized = VersionComparator.normalize_version(version)
|
||||
try:
|
||||
parts = normalized.split(".")
|
||||
return (int(parts[0]), int(parts[1]), int(parts[2]))
|
||||
except (ValueError, IndexError):
|
||||
logger.warning(f"无法解析版本号: {version},使用默认版本 0.0.0")
|
||||
return (0, 0, 0)
|
||||
|
||||
@staticmethod
|
||||
def compare_versions(version1: str, version2: str) -> int:
|
||||
"""比较两个版本号
|
||||
|
||||
Args:
|
||||
version1: 第一个版本号
|
||||
version2: 第二个版本号
|
||||
|
||||
Returns:
|
||||
int: -1 if version1 < version2, 0 if equal, 1 if version1 > version2
|
||||
"""
|
||||
v1_tuple = VersionComparator.parse_version(version1)
|
||||
v2_tuple = VersionComparator.parse_version(version2)
|
||||
|
||||
if v1_tuple < v2_tuple:
|
||||
return -1
|
||||
elif v1_tuple > v2_tuple:
|
||||
return 1
|
||||
else:
|
||||
return 0
|
||||
|
||||
@staticmethod
|
||||
def is_version_in_range(version: str, min_version: str = "", max_version: str = "") -> Tuple[bool, str]:
|
||||
"""检查版本是否在指定范围内
|
||||
|
||||
Args:
|
||||
version: 要检查的版本号
|
||||
min_version: 最小版本号(可选)
|
||||
max_version: 最大版本号(可选)
|
||||
|
||||
Returns:
|
||||
Tuple[bool, str]: (是否兼容, 错误信息)
|
||||
"""
|
||||
if not min_version and not max_version:
|
||||
return True, ""
|
||||
|
||||
version_normalized = VersionComparator.normalize_version(version)
|
||||
|
||||
# 检查最小版本
|
||||
if min_version:
|
||||
min_normalized = VersionComparator.normalize_version(min_version)
|
||||
if VersionComparator.compare_versions(version_normalized, min_normalized) < 0:
|
||||
return False, f"版本 {version_normalized} 低于最小要求版本 {min_normalized}"
|
||||
|
||||
# 检查最大版本
|
||||
if max_version:
|
||||
max_normalized = VersionComparator.normalize_version(max_version)
|
||||
if VersionComparator.compare_versions(version_normalized, max_normalized) > 0:
|
||||
return False, f"版本 {version_normalized} 高于最大支持版本 {max_normalized}"
|
||||
|
||||
return True, ""
|
||||
|
||||
@staticmethod
|
||||
def get_current_host_version() -> str:
|
||||
"""获取当前主机应用版本
|
||||
|
||||
Returns:
|
||||
str: 当前版本号
|
||||
"""
|
||||
return VersionComparator.normalize_version(MMC_VERSION)
|
||||
|
||||
|
||||
class ManifestValidator:
|
||||
"""Manifest文件验证器"""
|
||||
|
||||
# 必需字段(必须存在且不能为空)
|
||||
REQUIRED_FIELDS = ["manifest_version", "name", "version", "description", "author"]
|
||||
|
||||
# 可选字段(可以不存在或为空)
|
||||
OPTIONAL_FIELDS = [
|
||||
"license",
|
||||
"host_application",
|
||||
"homepage_url",
|
||||
"repository_url",
|
||||
"keywords",
|
||||
"categories",
|
||||
"default_locale",
|
||||
"locales_path",
|
||||
"plugin_info",
|
||||
]
|
||||
|
||||
# 建议填写的字段(会给出警告但不会导致验证失败)
|
||||
RECOMMENDED_FIELDS = ["license", "keywords", "categories"]
|
||||
|
||||
SUPPORTED_MANIFEST_VERSIONS = [3]
|
||||
|
||||
def __init__(self):
|
||||
self.validation_errors = []
|
||||
self.validation_warnings = []
|
||||
|
||||
def validate_manifest(self, manifest_data: Dict[str, Any]) -> bool:
|
||||
"""验证manifest数据
|
||||
|
||||
Args:
|
||||
manifest_data: manifest数据字典
|
||||
|
||||
Returns:
|
||||
bool: 是否验证通过(只有错误会导致验证失败,警告不会)
|
||||
"""
|
||||
self.validation_errors.clear()
|
||||
self.validation_warnings.clear()
|
||||
|
||||
# 检查必需字段
|
||||
for field in self.REQUIRED_FIELDS:
|
||||
if field not in manifest_data:
|
||||
self.validation_errors.append(f"缺少必需字段: {field}")
|
||||
elif not manifest_data[field]:
|
||||
self.validation_errors.append(f"必需字段不能为空: {field}")
|
||||
|
||||
# 检查manifest版本
|
||||
if "manifest_version" in manifest_data:
|
||||
version = manifest_data["manifest_version"]
|
||||
if version not in self.SUPPORTED_MANIFEST_VERSIONS:
|
||||
self.validation_errors.append(
|
||||
f"不支持的manifest版本: {version},支持的版本: {self.SUPPORTED_MANIFEST_VERSIONS}"
|
||||
)
|
||||
|
||||
# 检查作者信息格式
|
||||
if "author" in manifest_data:
|
||||
author = manifest_data["author"]
|
||||
if isinstance(author, dict):
|
||||
if "name" not in author or not author["name"]:
|
||||
self.validation_errors.append("作者信息缺少name字段或为空")
|
||||
# url字段是可选的
|
||||
if "url" in author and author["url"]:
|
||||
url = author["url"]
|
||||
if not (url.startswith("http://") or url.startswith("https://")):
|
||||
self.validation_warnings.append("作者URL建议使用完整的URL格式")
|
||||
elif isinstance(author, str):
|
||||
if not author.strip():
|
||||
self.validation_errors.append("作者信息不能为空")
|
||||
else:
|
||||
self.validation_errors.append("作者信息格式错误,应为字符串或包含name字段的对象")
|
||||
# 检查主机应用版本要求(可选)
|
||||
if "host_application" in manifest_data:
|
||||
host_app = manifest_data["host_application"]
|
||||
if isinstance(host_app, dict):
|
||||
min_version = host_app.get("min_version", "")
|
||||
max_version = host_app.get("max_version", "")
|
||||
|
||||
# 验证版本字段格式
|
||||
for version_field in ["min_version", "max_version"]:
|
||||
if version_field in host_app and not host_app[version_field]:
|
||||
self.validation_warnings.append(f"host_application.{version_field}为空")
|
||||
|
||||
# 检查当前主机版本兼容性
|
||||
if min_version or max_version:
|
||||
current_version = VersionComparator.get_current_host_version()
|
||||
is_compatible, error_msg = VersionComparator.is_version_in_range(
|
||||
current_version, min_version, max_version
|
||||
)
|
||||
|
||||
if not is_compatible:
|
||||
self.validation_errors.append(f"版本兼容性检查失败: {error_msg} (当前版本: {current_version})")
|
||||
else:
|
||||
logger.debug(
|
||||
f"版本兼容性检查通过: 当前版本 {current_version} 符合要求 [{min_version}, {max_version}]"
|
||||
)
|
||||
else:
|
||||
self.validation_errors.append("host_application格式错误,应为对象")
|
||||
|
||||
# 检查URL格式(可选字段)
|
||||
for url_field in ["homepage_url", "repository_url"]:
|
||||
if url_field in manifest_data and manifest_data[url_field]:
|
||||
url = manifest_data[url_field]
|
||||
if not (url.startswith("http://") or url.startswith("https://")):
|
||||
self.validation_warnings.append(f"{url_field}建议使用完整的URL格式")
|
||||
|
||||
# 检查数组字段格式(可选字段)
|
||||
for list_field in ["keywords", "categories"]:
|
||||
if list_field in manifest_data:
|
||||
field_value = manifest_data[list_field]
|
||||
if field_value is not None and not isinstance(field_value, list):
|
||||
self.validation_errors.append(f"{list_field}应为数组格式")
|
||||
elif isinstance(field_value, list):
|
||||
# 检查数组元素是否为字符串
|
||||
for i, item in enumerate(field_value):
|
||||
if not isinstance(item, str):
|
||||
self.validation_warnings.append(f"{list_field}[{i}]应为字符串")
|
||||
|
||||
# 检查建议字段(给出警告)
|
||||
for field in self.RECOMMENDED_FIELDS:
|
||||
if field not in manifest_data or not manifest_data[field]:
|
||||
self.validation_warnings.append(f"建议填写字段: {field}")
|
||||
|
||||
# 检查plugin_info结构(可选)
|
||||
if "plugin_info" in manifest_data:
|
||||
plugin_info = manifest_data["plugin_info"]
|
||||
if isinstance(plugin_info, dict):
|
||||
# 检查components数组
|
||||
if "components" in plugin_info:
|
||||
components = plugin_info["components"]
|
||||
if not isinstance(components, list):
|
||||
self.validation_errors.append("plugin_info.components应为数组格式")
|
||||
else:
|
||||
for i, component in enumerate(components):
|
||||
if not isinstance(component, dict):
|
||||
self.validation_errors.append(f"plugin_info.components[{i}]应为对象")
|
||||
else:
|
||||
# 检查组件必需字段
|
||||
for comp_field in ["type", "name", "description"]:
|
||||
if comp_field not in component or not component[comp_field]:
|
||||
self.validation_errors.append(
|
||||
f"plugin_info.components[{i}]缺少必需字段: {comp_field}"
|
||||
)
|
||||
else:
|
||||
self.validation_errors.append("plugin_info应为对象格式")
|
||||
|
||||
return len(self.validation_errors) == 0
|
||||
|
||||
def get_validation_report(self) -> str:
|
||||
"""获取验证报告"""
|
||||
report = []
|
||||
|
||||
if self.validation_errors:
|
||||
report.append("❌ 验证错误:")
|
||||
for error in self.validation_errors:
|
||||
report.append(f" - {error}")
|
||||
|
||||
if self.validation_warnings:
|
||||
report.append("⚠️ 验证警告:")
|
||||
for warning in self.validation_warnings:
|
||||
report.append(f" - {warning}")
|
||||
|
||||
if not self.validation_errors and not self.validation_warnings:
|
||||
report.append("✅ Manifest文件验证通过")
|
||||
|
||||
return "\n".join(report)
|
||||
|
||||
|
||||
class ManifestGenerator:
|
||||
"""Manifest文件生成器"""
|
||||
|
||||
def __init__(self):
|
||||
self.template = {
|
||||
"manifest_version": 1,
|
||||
"name": "",
|
||||
"version": "1.0.0",
|
||||
"description": "",
|
||||
"author": {"name": "", "url": ""},
|
||||
"license": "MIT",
|
||||
"host_application": {"min_version": "1.0.0", "max_version": "4.0.0"},
|
||||
"homepage_url": "",
|
||||
"repository_url": "",
|
||||
"keywords": [],
|
||||
"categories": [],
|
||||
"default_locale": "zh-CN",
|
||||
"locales_path": "_locales",
|
||||
}
|
||||
|
||||
def generate_from_plugin(self, plugin_instance) -> Dict[str, Any]:
|
||||
"""从插件实例生成manifest
|
||||
|
||||
Args:
|
||||
plugin_instance: BasePlugin实例
|
||||
|
||||
Returns:
|
||||
Dict[str, Any]: 生成的manifest数据
|
||||
"""
|
||||
manifest = self.template.copy()
|
||||
|
||||
# 基本信息
|
||||
manifest["name"] = plugin_instance.plugin_name
|
||||
manifest["version"] = plugin_instance.plugin_version
|
||||
manifest["description"] = plugin_instance.plugin_description
|
||||
|
||||
# 作者信息
|
||||
if plugin_instance.plugin_author:
|
||||
manifest["author"]["name"] = plugin_instance.plugin_author
|
||||
|
||||
# 组件信息
|
||||
components = []
|
||||
plugin_components = plugin_instance.get_plugin_components()
|
||||
|
||||
for component_info, component_class in plugin_components:
|
||||
component_data = {
|
||||
"type": component_info.component_type.value,
|
||||
"name": component_info.name,
|
||||
"description": component_info.description,
|
||||
}
|
||||
|
||||
# 添加激活模式信息(对于Action组件)
|
||||
if hasattr(component_class, "focus_activation_type"):
|
||||
activation_modes = []
|
||||
if hasattr(component_class, "focus_activation_type"):
|
||||
activation_modes.append(component_class.focus_activation_type.value)
|
||||
if hasattr(component_class, "normal_activation_type"):
|
||||
activation_modes.append(component_class.normal_activation_type.value)
|
||||
component_data["activation_modes"] = list(set(activation_modes))
|
||||
|
||||
# 添加关键词信息
|
||||
if hasattr(component_class, "activation_keywords"):
|
||||
keywords = getattr(component_class, "activation_keywords", [])
|
||||
if keywords:
|
||||
component_data["keywords"] = keywords
|
||||
|
||||
components.append(component_data)
|
||||
|
||||
manifest["plugin_info"] = {"is_built_in": True, "plugin_type": "general", "components": components}
|
||||
|
||||
return manifest
|
||||
|
||||
def save_manifest(self, manifest_data: Dict[str, Any], plugin_dir: str) -> bool:
|
||||
"""保存manifest文件
|
||||
|
||||
Args:
|
||||
manifest_data: manifest数据
|
||||
plugin_dir: 插件目录
|
||||
|
||||
Returns:
|
||||
bool: 是否保存成功
|
||||
"""
|
||||
try:
|
||||
manifest_path = os.path.join(plugin_dir, "_manifest.json")
|
||||
with open(manifest_path, "w", encoding="utf-8") as f:
|
||||
json.dump(manifest_data, f, ensure_ascii=False, indent=2)
|
||||
logger.info(f"Manifest文件已保存: {manifest_path}")
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"保存manifest文件失败: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def validate_plugin_manifest(plugin_dir: str) -> bool:
|
||||
"""验证插件目录中的manifest文件
|
||||
|
||||
Args:
|
||||
plugin_dir: 插件目录路径
|
||||
|
||||
Returns:
|
||||
bool: 是否验证通过
|
||||
"""
|
||||
manifest_path = os.path.join(plugin_dir, "_manifest.json")
|
||||
|
||||
if not os.path.exists(manifest_path):
|
||||
logger.warning(f"未找到manifest文件: {manifest_path}")
|
||||
return False
|
||||
|
||||
try:
|
||||
with open(manifest_path, "r", encoding="utf-8") as f:
|
||||
manifest_data = json.load(f)
|
||||
|
||||
validator = ManifestValidator()
|
||||
is_valid = validator.validate_manifest(manifest_data)
|
||||
|
||||
logger.info(f"Manifest验证结果:\n{validator.get_validation_report()}")
|
||||
|
||||
return is_valid
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"读取或验证manifest文件失败: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def generate_plugin_manifest(plugin_instance, save_to_file: bool = True) -> Optional[Dict[str, Any]]:
|
||||
"""为插件生成manifest文件
|
||||
|
||||
Args:
|
||||
plugin_instance: BasePlugin实例
|
||||
save_to_file: 是否保存到文件
|
||||
|
||||
Returns:
|
||||
Optional[Dict[str, Any]]: 生成的manifest数据
|
||||
"""
|
||||
try:
|
||||
generator = ManifestGenerator()
|
||||
manifest_data = generator.generate_from_plugin(plugin_instance)
|
||||
|
||||
if save_to_file and plugin_instance.plugin_dir:
|
||||
generator.save_manifest(manifest_data, plugin_instance.plugin_dir)
|
||||
|
||||
return manifest_data
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"生成manifest文件失败: {e}")
|
||||
return None
|
||||
Reference in New Issue
Block a user