from __future__ import annotations import asyncio import hashlib import json import os import ssl from collections.abc import Awaitable, Callable from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any from urllib.parse import urlparse import aiohttp import certifi import yaml from astrbot.api import sp from astrbot.core import DEMO_MODE, file_token_service, logger from astrbot.core.computer.computer_client import sync_skills_to_active_sandboxes from astrbot.core.skills.skill_manager import SkillManager from astrbot.core.star.filter.command import CommandFilter from astrbot.core.star.filter.command_group import CommandGroupFilter from astrbot.core.star.filter.permission import PermissionTypeFilter from astrbot.core.star.filter.regex import RegexFilter from astrbot.core.star.star import StarMetadata from astrbot.core.star.star_handler import EventType, star_handlers_registry from astrbot.core.star.star_manager import ( PluginManager, PluginVersionUnsupportedError, ) from astrbot.core.star.updator import PLUGIN_METADATA_FILENAMES from astrbot.core.utils.astrbot_path import get_astrbot_data_path, get_astrbot_temp_path PLUGIN_UPDATE_CONCURRENCY = 3 PLUGIN_OPERATION_FAILED_MESSAGE = "插件操作失败，请查看服务端日志。" PLUGIN_UPDATE_FAILED_MESSAGE = "更新失败，请查看服务端日志。" PLUGIN_INSTALL_SOURCES_KEY = "plugin_install_sources" PLUGIN_DEFAULT_REGISTRY_NAME = "Default" PLUGIN_UPDATE_DISABLED_MESSAGE = "该插件不是通过插件市场安装，无法检测或执行更新。" PLUGIN_UPDATE_SOURCE_REQUIRED_MESSAGE = "请先选择插件安装源后再更新。" PLUGIN_REPO_VALIDATE_TIMEOUT_SECONDS = 15 PLUGIN_METADATA_MAX_BYTES = 1024 * 1024 PLUGIN_COMPONENT_TYPE_ORDER = { "page": 0, "skill": 1, "command": 2, "llm_tool": 3, "listener": 4, "hook": 5, } LogoTokenResolver = Callable[[str], Awaitable[str | None]] InstalledAtResolver = Callable[[StarMetadata], str | None] PluginPagesResolver = Callable[[StarMetadata], Awaitable[list]] PluginPagesSerializer = Callable[[StarMetadata], Awaitable[list[dict]]] @dataclass class RegistrySource: urls: list[str] cache_file: str md5_url: str | None class PluginServiceError(Exception): def __init__( self, message: str, *, public_message: str | None = None, ) -> None: super().__init__(message) self.public_message = public_message or message class PluginServiceWarning(Exception): def __init__( self, message: str, data: dict[str, Any], *, public_message: str | None = None, ) -> None: super().__init__(message) self.public_message = public_message or message self.data = data class PluginService: def __init__( self, core_lifecycle, plugin_manager: PluginManager, ) -> None: self.core_lifecycle = core_lifecycle self.plugin_manager = plugin_manager self.translated_event_type = { EventType.AdapterMessageEvent: "平台消息下发时", EventType.OnLLMRequestEvent: "LLM 请求时", EventType.OnLLMResponseEvent: "LLM 响应后", EventType.OnAgentBeginEvent: "Agent 开始运行时", EventType.OnAgentDoneEvent: "Agent 运行完成后", EventType.OnDecoratingResultEvent: "回复消息前", EventType.OnCallingFuncToolEvent: "函数工具", EventType.OnAfterMessageSentEvent: "发送消息后", EventType.OnPluginErrorEvent: "插件报错时", } self._logo_cache: dict[str, str] = {} @staticmethod def _payload(data: object) -> dict[str, Any]: return data if isinstance(data, dict) else {} @staticmethod def _ensure_not_demo() -> None: if DEMO_MODE: raise PluginServiceError( "You are not permitted to do this operation in demo mode" ) async def sync_skills_after_plugin_change(self) -> None: try: await sync_skills_to_active_sandboxes() except Exception: logger.warning("Failed to sync plugin-provided skills to active sandboxes.") async def check_plugin_version_support(self, data: object) -> dict: payload = self._payload(data) version_spec = payload.get("astrbot_version", "") is_valid, message = self.plugin_manager._validate_astrbot_version_specifier( version_spec ) return { "supported": is_valid, "message": message, "astrbot_version": version_spec, } async def reload_failed_plugin(self, data: object) -> tuple[None, str]: self._ensure_not_demo() payload = data if isinstance(data, dict) else {} dir_name = payload.get("dir_name") if not dir_name: raise PluginServiceError("缺少插件目录名") success, err = await self.plugin_manager.reload_failed_plugin(dir_name) if not success: raise PluginServiceError(f"重载失败: {err}", public_message="重载失败") await self.sync_skills_after_plugin_change() return None, f"插件 {dir_name} 重载成功。" async def reload_plugin(self, data: object) -> tuple[None, str]: self._ensure_not_demo() payload = data if isinstance(data, dict) else {} plugin_name = payload.get("name", None) success, message = await self.plugin_manager.reload(plugin_name) if not success: raise PluginServiceError( message or "插件重载失败", public_message="插件重载失败", ) await self.sync_skills_after_plugin_change() return None, "重载成功。" async def list_plugins( self, *, plugin_name: str | None, logo_token_resolver: LogoTokenResolver, installed_at_resolver: InstalledAtResolver, discover_pages: PluginPagesResolver, ) -> tuple[list[dict], str | None]: plugins = [ plugin for plugin in self.plugin_manager.context.get_all_stars() if not (plugin_name and plugin.name != plugin_name) ] install_sources = await self.get_plugin_install_sources() async def process_plugin(plugin: StarMetadata): logo_url = await self.resolve_plugin_logo_url(plugin, logo_token_resolver) pages = await discover_pages(plugin) return plugin, logo_url, pages results = await asyncio.gather(*(process_plugin(plugin) for plugin in plugins)) payload = [] for plugin, logo_url, pages in results: if self.is_ghost_plugin(plugin): continue payload.append( { **self.serialize_plugin_base( plugin, logo_url=logo_url, installed_at=installed_at_resolver(plugin), install_source=self.resolve_effective_plugin_install_source( plugin, install_sources, ), ), "pages": [page.name for page in pages], } ) return payload, getattr(self.plugin_manager, "failed_plugin_info", None) async def list_plugins_from_dashboard_query( self, *, plugin_name: str | None, logo_token_resolver: LogoTokenResolver, installed_at_resolver: InstalledAtResolver, discover_pages: PluginPagesResolver, ) -> tuple[list[dict], str | None]: return await self.list_plugins( plugin_name=plugin_name, logo_token_resolver=logo_token_resolver, installed_at_resolver=installed_at_resolver, discover_pages=discover_pages, ) async def get_plugin_detail( self, *, plugin_name: str | None, logo_token_resolver: LogoTokenResolver, installed_at_resolver: InstalledAtResolver, serialize_pages: PluginPagesSerializer, ) -> dict: if not plugin_name: raise PluginServiceError("缺少插件名") install_sources = await self.get_plugin_install_sources() for plugin in self.plugin_manager.context.get_all_stars(): if plugin.name != plugin_name: continue logo_url = await self.resolve_plugin_logo_url(plugin, logo_token_resolver) return { **self.serialize_plugin_base( plugin, logo_url=logo_url, installed_at=installed_at_resolver(plugin), install_source=self.resolve_effective_plugin_install_source( plugin, install_sources, ), ), "components": await self.get_plugin_components_info( plugin, serialize_pages, ), } raise PluginServiceError("插件不存在") async def get_plugin_detail_from_dashboard_query( self, *, plugin_name: str | None, logo_token_resolver: LogoTokenResolver, installed_at_resolver: InstalledAtResolver, serialize_pages: PluginPagesSerializer, ) -> dict: return await self.get_plugin_detail( plugin_name=plugin_name, logo_token_resolver=logo_token_resolver, installed_at_resolver=installed_at_resolver, serialize_pages=serialize_pages, ) async def resolve_plugin_logo_url( self, plugin: StarMetadata, logo_token_resolver: LogoTokenResolver, ) -> str | None: if not plugin.logo_path: return None logo_token = await logo_token_resolver(plugin.logo_path) return f"/api/file/{logo_token}" if logo_token else None async def get_plugin_logo_token(self, logo_path: str) -> str | None: try: if token := self._logo_cache.get(logo_path): if not await file_token_service.check_token_expired(token): return token token = await file_token_service.register_file(logo_path, timeout=300) self._logo_cache[logo_path] = token return token except Exception as exc: logger.warning(f"获取插件 Logo 失败: {exc}") return None def resolve_plugin_metadata_dir(self, plugin: StarMetadata) -> Path | None: if not plugin.root_dir_name: return None base_dir = Path( self.plugin_manager.reserved_plugin_path if plugin.reserved else self.plugin_manager.plugin_store_path ) plugin_dir = base_dir / plugin.root_dir_name if not plugin_dir.is_dir(): return None return plugin_dir def get_plugin_installed_at(self, plugin: StarMetadata) -> str | None: plugin_dir = self.resolve_plugin_metadata_dir(plugin) if plugin_dir is None: return None try: return datetime.fromtimestamp( plugin_dir.stat().st_mtime, timezone.utc, ).isoformat() except OSError as exc: logger.warning(f"获取插件安装时间失败 {plugin.name}: {exc!s}") return None @staticmethod def get_market_plugin_id( data: dict[str, Any] | None, fallback_name: str | None = None, ) -> str: """Return a marketplace plugin ID from explicit fields or author/name. Args: data: Marketplace entry or persisted source record. fallback_name: Marketplace root key to use as name for legacy entries. Returns: The marketplace plugin ID, or an empty string. """ if not isinstance(data, dict): return "" market_plugin_id = str(data.get("market_plugin_id") or "").strip() if market_plugin_id: return market_plugin_id author_name = str(data.get("author") or "").strip() plugin_name = str(data.get("name") or "").strip() fallback_name = str(fallback_name or "").strip() if not plugin_name and fallback_name and "/" not in fallback_name: plugin_name = fallback_name if author_name and plugin_name: return f"{author_name}/{plugin_name}" return "" @staticmethod async def get_plugin_install_sources() -> dict[str, dict[str, Any]]: """Return persisted plugin installation source records. Returns: A mapping keyed by local plugin root directory name. """ records = await sp.global_get(PLUGIN_INSTALL_SOURCES_KEY, {}) if not isinstance(records, dict): return {} return { str(key): value for key, value in records.items() if isinstance(value, dict) } @staticmethod async def save_plugin_install_sources( records: dict[str, dict[str, Any]], ) -> None: """Persist plugin installation source records. Args: records: Mapping keyed by local plugin root directory name. """ await sp.global_put(PLUGIN_INSTALL_SOURCES_KEY, records) @staticmethod def resolve_plugin_install_source( plugin: StarMetadata, records: dict[str, dict[str, Any]], ) -> dict[str, Any] | None: """Resolve a source record for a loaded plugin. Args: plugin: Loaded plugin metadata. records: Persisted source records. Returns: The source record when one is available. """ for key in (plugin.root_dir_name, plugin.name): if key and isinstance(records.get(key), dict): return records[key] return None def resolve_effective_plugin_install_source( self, plugin: StarMetadata, records: dict[str, dict[str, Any]], ) -> dict[str, Any] | None: """Resolve a plugin source for display. Args: plugin: Loaded plugin metadata. records: Persisted source records. Returns: The persisted source record, an implicit display-only source record, or None. """ record = self.resolve_plugin_install_source(plugin, records) if isinstance(record, dict): return record if plugin.reserved: return None plugin_name = str(plugin.name or "").strip() implicit_record = { "schema_version": 1, "root_dir_name": plugin.root_dir_name, "install_method": "market", "registry_url": None, "registry_name": PLUGIN_DEFAULT_REGISTRY_NAME, "repo": str(plugin.repo or "").strip(), "download_url": "", "implicit": True, } if plugin_name: implicit_record["name"] = plugin_name implicit_record["marketplace_name"] = plugin_name.replace("_", "-") return implicit_record def find_plugin_by_name(self, plugin_name: str | None) -> StarMetadata | None: """Find a loaded plugin by metadata name. Args: plugin_name: Metadata plugin name. Returns: The matching plugin metadata, if loaded. """ if not plugin_name: return None plugin = self.plugin_manager.context.get_registered_star(plugin_name) if plugin: return plugin for item in self.plugin_manager.context.get_all_stars(): if item.name == plugin_name or item.root_dir_name == plugin_name: return item return None @staticmethod def normalize_registry_url(value: object) -> str | None: """Normalize a plugin registry URL for persistence. Args: value: Raw registry URL value. Returns: The normalized URL, or None for the default registry. """ text = str(value or "").strip().rstrip("/") return text or None async def resolve_registry_name(self, registry_url: object) -> str: """Resolve a registry display name from configured plugin sources. Args: registry_url: Registry URL, or None for the default source. Returns: A display name for the registry. """ normalized_url = self.normalize_registry_url(registry_url) if not normalized_url: return PLUGIN_DEFAULT_REGISTRY_NAME for source in await self.get_custom_sources(): if not isinstance(source, dict): continue source_url = self.normalize_registry_url(source.get("url")) if source_url != normalized_url: continue source_name = str(source.get("name") or "").strip() return source_name or "Custom" return "Custom" @staticmethod def repo_identifier_from_url(repo_url: object) -> str: """Build an owner/repo identifier from a GitHub repository URL. Args: repo_url: Repository URL. Returns: The owner/repo identifier, or an empty string when unavailable. """ text = str(repo_url or "").strip().rstrip("/") if not text: return "" if not text.startswith(("http://", "https://")): text = f"https://{text}" parsed = urlparse(text) if parsed.netloc.lower() not in {"github.com", "www.github.com"}: return "" parts = [part for part in parsed.path.strip("/").split("/") if part] if len(parts) < 2: return "" repo_name = parts[1].removesuffix(".git") return f"{parts[0]}/{repo_name}" if repo_name else "" def build_install_source_record( self, plugin: StarMetadata, payload: dict[str, Any], *, fallback_method: str, repo_url: str, download_url: str, registry_name: str, ) -> dict[str, Any]: """Build a persisted installation source record. Args: plugin: Loaded plugin metadata after installation. payload: Installation request payload. fallback_method: Method to use when the request did not declare one. repo_url: Repository URL used for installation. download_url: Download URL used for installation. registry_name: Resolved registry display name. Returns: A serializable installation source record. """ requested_method = str(payload.get("install_method") or "").strip().lower() install_method = requested_method or fallback_method if install_method != "market": install_method = fallback_method record = { "schema_version": 1, "root_dir_name": plugin.root_dir_name, "install_method": install_method, "registry_url": self.normalize_registry_url(payload.get("registry_url")), "registry_name": registry_name, "repo": str(repo_url or plugin.repo or "").strip(), "download_url": str( download_url or payload.get("download_url") or "" ).strip(), "installed_at": datetime.now(timezone.utc).isoformat(), } market_plugin_id = self.get_market_plugin_id(payload) if market_plugin_id: record["market_plugin_id"] = market_plugin_id return record async def persist_plugin_install_source( self, plugin_info: dict[str, Any] | None, payload: dict[str, Any], *, fallback_method: str, repo_url: str, download_url: str, ) -> None: """Persist an installation source record for a newly installed plugin. Args: plugin_info: Plugin information returned by PluginManager. payload: Installation request payload. fallback_method: Method to use for non-market installs. repo_url: Repository URL used for installation. download_url: Download URL used for installation. """ plugin_name = None if isinstance(plugin_info, dict): plugin_name = str(plugin_info.get("name") or "").strip() plugin = self.find_plugin_by_name(plugin_name) if not plugin or not plugin.root_dir_name: logger.warning("插件安装成功，但无法记录安装来源：缺少插件元数据。") return registry_name = await self.resolve_registry_name(payload.get("registry_url")) records = await self.get_plugin_install_sources() records[plugin.root_dir_name] = self.build_install_source_record( plugin, payload, fallback_method=fallback_method, repo_url=repo_url, download_url=download_url, registry_name=registry_name, ) await self.save_plugin_install_sources(records) async def remove_plugin_install_source(self, root_dir_name: str | None) -> None: """Remove an installation source record. Args: root_dir_name: Local plugin root directory name. """ if not root_dir_name: return records = await self.get_plugin_install_sources() if root_dir_name not in records: return del records[root_dir_name] await self.save_plugin_install_sources(records) @staticmethod def is_ghost_plugin(plugin: StarMetadata) -> bool: return not any( [ plugin.name, plugin.author, plugin.desc, plugin.version, plugin.display_name, ] ) @staticmethod def serialize_plugin_base( plugin: StarMetadata, *, logo_url: str | None, installed_at: str | None, install_source: dict[str, Any] | None, ) -> dict: install_method = ( str(install_source.get("install_method") or "").strip().lower() if isinstance(install_source, dict) else "" ) updates_enabled = install_method in {"market", "github"} and not plugin.reserved return { "name": plugin.name, "marketplace_name": (plugin.name or "").replace("_", "-"), "repo": "" if plugin.repo is None else str(plugin.repo), "author": plugin.author, "desc": plugin.desc, "version": plugin.version, "reserved": plugin.reserved, "activated": plugin.activated, "online_vesion": "", "display_name": plugin.display_name, "logo": logo_url, "support_platforms": plugin.support_platforms, "astrbot_version": plugin.astrbot_version, "installed_at": installed_at, "i18n": plugin.i18n, "root_dir_name": plugin.root_dir_name, "install_source": install_source, "updates_enabled": updates_enabled, "update_disabled_reason": "" if updates_enabled else PLUGIN_UPDATE_DISABLED_MESSAGE, } def get_failed_plugins(self) -> dict: return self.plugin_manager.failed_plugin_dict async def get_plugin_components_info( self, plugin: StarMetadata, serialize_pages: PluginPagesSerializer, ) -> list[dict]: components = [ *await self.get_plugin_page_components(plugin, serialize_pages), *self.get_plugin_skill_components(plugin), *await self.get_plugin_handler_components(plugin.star_handler_full_names), ] return sorted( components, key=lambda item: PLUGIN_COMPONENT_TYPE_ORDER.get(item["type"], 99), ) async def get_plugin_page_components( self, plugin: StarMetadata, serialize_pages: PluginPagesSerializer, ) -> list[dict]: pages = await serialize_pages(plugin) return [ { "type": "page", "name": page["title"], "title": page["title"], "page_name": page["name"], "i18n_key": page["i18n_key"], "description": "Plugin Page entry", "plugin_name": plugin.name, "plugin_marketplace_name": (plugin.name or "").replace("_", "-"), } for page in pages ] async def get_plugin_handler_components( self, handler_full_names: list[str], ) -> list[dict]: components = [] for handler_full_name in handler_full_names: info = {} handler = star_handlers_registry.star_handlers_map.get( handler_full_name, None, ) if handler is None: continue info["event_type"] = handler.event_type.name info["event_type_h"] = self.translated_event_type.get( handler.event_type, handler.event_type.name, ) info["handler_full_name"] = handler.handler_full_name info["description"] = handler.desc or "无描述" info["handler_name"] = handler.handler_name component_type = "hook" component: dict[str, Any] | None = None if handler.event_type == EventType.AdapterMessageEvent: has_admin = False for event_filter in handler.event_filters: if isinstance(event_filter, CommandFilter): component_type = "command" info["display_type"] = "指令" info["cmd"] = self._get_command_filter_display_name( event_filter ) component = self._build_command_filter_component( event_filter, handler.desc, ) elif isinstance(event_filter, CommandGroupFilter): component_type = "command" info["display_type"] = "指令组" info["cmd"] = event_filter.get_complete_command_names()[ 0 ].strip() component = self._build_command_group_component( event_filter, handler.desc, ) elif isinstance(event_filter, RegexFilter): component_type = "command" info["display_type"] = "正则匹配" info["cmd"] = event_filter.regex_str component = { "type": "command", "name": event_filter.regex_str, "description": handler.desc or "无描述", "match": "regex", } elif isinstance(event_filter, PermissionTypeFilter): has_admin = True info["has_admin"] = has_admin if "cmd" not in info: info["cmd"] = "未知" if "display_type" not in info: info["display_type"] = "事件监听器" component_type = "listener" else: info["cmd"] = "自动触发" info["display_type"] = "无" if handler.event_type == EventType.OnCallingFuncToolEvent: component_type = "llm_tool" if component is None: component = { "type": component_type, "name": handler.handler_name or handler.event_type.name, "description": handler.desc or "无描述", } else: component["type"] = component_type if component_type == "command": component["event_type"] = info["event_type"] component["event_type_h"] = info["event_type_h"] component["handler_name"] = info["handler_name"] component["has_admin"] = info.get("has_admin", False) if "display_type" in info: component["display_type"] = info["display_type"] if "cmd" in info: component["command"] = info["cmd"] else: component.update(info) components.append(component) return self._merge_command_components(components) def get_plugin_skill_components(self, plugin: StarMetadata) -> list[dict]: plugin_names = { str(name) for name in (plugin.root_dir_name, plugin.name) if str(name or "").strip() } if not plugin_names: return [] try: skills = SkillManager().list_skills( active_only=False, runtime="local", show_sandbox_path=False, ) except Exception as exc: logger.warning(f"获取插件 Skills 失败 {plugin.name}: {exc!s}") return [] components = [] for skill in skills: if skill.source_type != "plugin" or skill.plugin_name not in plugin_names: continue components.append( { "type": "skill", "name": skill.name, "description": skill.description or "无描述", "path": skill.path, } ) return components @staticmethod def _get_command_filter_display_name(command_filter: CommandFilter) -> str: return command_filter.get_complete_command_names()[0].strip() @staticmethod def _get_command_description( command_filter: CommandFilter | CommandGroupFilter, fallback: str = "", ) -> str: handler_md = getattr(command_filter, "handler_md", None) desc = getattr(handler_md, "desc", "") if handler_md else "" return desc or fallback or "无描述" def _build_command_filter_component( self, command_filter: CommandFilter, fallback_desc: str = "", ) -> dict: parts = self._get_command_filter_display_name(command_filter).split() if not parts: parts = [command_filter.command_name] component: dict[str, Any] = { "type": "command", "name": parts[-1], "description": self._get_command_description( command_filter, fallback_desc, ), } return self._wrap_command_component(parts[:-1], component) def _build_command_group_component( self, command_group_filter: CommandGroupFilter, fallback_desc: str = "", ) -> dict: parts = command_group_filter.get_complete_command_names()[0].strip().split() if not parts: parts = [command_group_filter.group_name] subcommands = [ self._build_command_group_child(sub_filter) for sub_filter in command_group_filter.sub_command_filters ] component: dict[str, Any] = { "type": "command", "name": parts[-1], "description": self._get_command_description( command_group_filter, fallback_desc, ), } if subcommands: component["subcommands"] = subcommands return self._wrap_command_component(parts[:-1], component) def _build_command_group_child( self, command_filter: CommandFilter | CommandGroupFilter, ) -> dict: if isinstance(command_filter, CommandGroupFilter): component: dict[str, Any] = { "name": command_filter.group_name, "description": self._get_command_description(command_filter), } subcommands = [ self._build_command_group_child(sub_filter) for sub_filter in command_filter.sub_command_filters ] if subcommands: component["subcommands"] = subcommands return component return { "name": command_filter.command_name, "description": self._get_command_description(command_filter), } @staticmethod def _wrap_command_component(parent_names: list[str], component: dict) -> dict: for parent_name in reversed(parent_names): component = { "type": "command", "name": parent_name, "description": "无描述", "subcommands": [component], } return component def _merge_command_components(self, components: list[dict]) -> list[dict]: merged: list[dict] = [] for component in components: if component.get("type") != "command": merged.append(component) continue existing = next( ( item for item in merged if item.get("type") == "command" and item.get("name") == component.get("name") and item.get("match") == component.get("match") ), None, ) if existing is None: merged.append(component) continue self._merge_command_component(existing, component) return merged def _merge_command_component(self, target: dict, source: dict) -> None: if target.get("description") == "无描述" and source.get("description"): target["description"] = source["description"] for key, value in source.items(): if key in {"subcommands", "description"}: continue target.setdefault(key, value) source_subcommands = source.get("subcommands") if not isinstance(source_subcommands, list): return target_subcommands = target.setdefault("subcommands", []) for source_subcommand in source_subcommands: if not isinstance(source_subcommand, dict): continue existing = next( ( item for item in target_subcommands if isinstance(item, dict) and item.get("name") == source_subcommand.get("name") ), None, ) if existing is None: target_subcommands.append(source_subcommand) continue self._merge_command_component(existing, source_subcommand) async def get_online_plugins( self, *, custom_registry: str | None, force_refresh: bool, ) -> tuple[Any, str | None]: source = self.build_registry_source(custom_registry) cached_data = None if not force_refresh and await self.is_cache_valid(source): cached_data = self.load_plugin_cache(source.cache_file) if cached_data: logger.debug("缓存MD5匹配，使用缓存的插件市场数据") return cached_data, None remote_data = None ssl_context = ssl.create_default_context(cafile=certifi.where()) connector = aiohttp.TCPConnector(ssl=ssl_context) for url in source.urls: try: async with ( aiohttp.ClientSession( trust_env=True, connector=connector, ) as session, session.get(url) as response, ): if response.status == 200: try: remote_data = await response.json() except aiohttp.ContentTypeError: remote_text = await response.text() remote_data = json.loads(remote_text) if not remote_data or ( isinstance(remote_data, dict) and len(remote_data) == 0 ): logger.warning(f"远程插件市场数据为空: {url}") continue logger.info( f"成功获取远程插件市场数据，包含 {len(remote_data)} 个插件" ) current_md5 = await self.fetch_remote_md5(source.md5_url) self.save_plugin_cache( source.cache_file, remote_data, current_md5, ) return remote_data, None logger.error(f"请求 {url} 失败，状态码：{response.status}") except Exception as exc: logger.error(f"请求 {url} 失败，错误：{exc}") if not cached_data: cached_data = self.load_plugin_cache(source.cache_file) if cached_data: logger.warning("远程插件市场数据获取失败，使用缓存数据") return cached_data, "使用缓存数据，可能不是最新版本" raise PluginServiceError("获取插件列表失败，且没有可用的缓存数据") async def get_online_plugins_from_dashboard_query( self, *, custom_registry: str | None, force_refresh, ) -> tuple[Any, str | None]: return await self.get_online_plugins( custom_registry=custom_registry, force_refresh=self._to_bool(force_refresh), ) @staticmethod def build_registry_source(custom_url: str | None) -> RegistrySource: data_dir = get_astrbot_data_path() if custom_url: url_hash = hashlib.md5(custom_url.encode()).hexdigest()[:8] cache_file = os.path.join(data_dir, f"plugins_custom_{url_hash}.json") md5_url = ( custom_url[:-5] + "-md5.json" if custom_url.endswith(".json") else custom_url + "-md5.json" ) urls = [custom_url] else: cache_file = os.path.join(data_dir, "plugins.json") md5_url = "https://api.soulter.top/astrbot/plugins-md5" urls = [ "https://api.soulter.top/astrbot/plugins", "https://github.com/AstrBotDevs/AstrBot_Plugins_Collection/raw/refs/heads/main/plugin_cache_original.json", ] return RegistrySource(urls=urls, cache_file=cache_file, md5_url=md5_url) @staticmethod def load_cached_md5(cache_file: str) -> str | None: if not os.path.exists(cache_file): return None try: with open(cache_file, encoding="utf-8") as file: cache_data = json.load(file) return cache_data.get("md5") except Exception as exc: logger.warning(f"Failed to load cached MD5: {exc}") return None @staticmethod async def fetch_remote_md5(md5_url: str | None) -> str | None: if not md5_url: return None try: ssl_context = ssl.create_default_context(cafile=certifi.where()) connector = aiohttp.TCPConnector(ssl=ssl_context) async with ( aiohttp.ClientSession( trust_env=True, connector=connector, ) as session, session.get(md5_url) as response, ): if response.status == 200: data = await response.json() return data.get("md5", "") except Exception as exc: logger.debug(f"Failed to fetch remote MD5: {exc}") return None async def is_cache_valid(self, source: RegistrySource) -> bool: try: cached_md5 = self.load_cached_md5(source.cache_file) if not cached_md5: logger.debug("MD5 not found in cache, treating cache as invalid") return False remote_md5 = await self.fetch_remote_md5(source.md5_url) if remote_md5 is None: logger.warning( "Cannot fetch remote MD5, using cache without validation" ) return True is_valid = cached_md5 == remote_md5 logger.debug( f"Plugin cache: local={cached_md5}, remote={remote_md5}, effective={is_valid}", ) return is_valid except Exception as exc: logger.warning(f"检查缓存有效性失败: {exc}") return False @staticmethod def load_plugin_cache(cache_file: str): try: if os.path.exists(cache_file): with open(cache_file, encoding="utf-8") as file: cache_data = json.load(file) if "data" in cache_data and "timestamp" in cache_data: logger.debug( f"Loading cached file: {cache_file}, Cache time: {cache_data['timestamp']}", ) return cache_data["data"] except Exception as exc: logger.warning(f"Failed to load plugin market cache: {exc}") return None @staticmethod def save_plugin_cache(cache_file: str, data, md5: str | None = None) -> None: try: os.makedirs(os.path.dirname(cache_file), exist_ok=True) cache_data = { "timestamp": datetime.now().isoformat(), "data": data, "md5": md5 or "", } with open(cache_file, "w", encoding="utf-8") as file: json.dump(cache_data, file, ensure_ascii=False, indent=2) logger.debug(f"Cached plugin market data: {cache_file}, MD5: {md5}") except Exception as exc: logger.warning(f"Failed to save plugin market cache: {exc}") @staticmethod def iter_market_plugin_entries(market_data: object): """Iterate market entries with their market identifiers. Args: market_data: Raw marketplace data from a registry. Yields: Tuples of market identifier and plugin entry. """ if isinstance(market_data, dict): for key, value in market_data.items(): if key == "$meta": continue if isinstance(value, dict): entry = value if "/" not in str(key) and not str(value.get("name") or "").strip(): entry = {**value, "name": str(key)} yield PluginService.get_market_plugin_id(entry, str(key)), entry return if isinstance(market_data, list): for value in market_data: if not isinstance(value, dict): continue identifier = PluginService.get_market_plugin_id(value) yield identifier, value def resolve_market_plugin_entry( self, market_data: object, record: dict[str, Any], ) -> dict[str, Any] | None: """Resolve a source registry plugin entry for a source record. Args: market_data: Raw marketplace data from a registry. record: Persisted install source record. Returns: The matching marketplace entry, if found. """ record_identifier = self.get_market_plugin_id(record) record_repo_identifier = self.repo_identifier_from_url(record.get("repo")) record_repo = str(record.get("repo") or "").strip().rstrip("/").lower() record_names = { str(record.get("name") or "").strip(), str(record.get("marketplace_name") or "").strip(), } record_names = {name for name in record_names if name} for plugin_identifier, plugin in self.iter_market_plugin_entries(market_data): if record_identifier and plugin_identifier == record_identifier: return plugin if record_repo_identifier and plugin_identifier == record_repo_identifier: return plugin plugin_repo_identifier = self.repo_identifier_from_url(plugin.get("repo")) if ( record_repo_identifier and plugin_repo_identifier and plugin_repo_identifier == record_repo_identifier ): return plugin plugin_repo = str(plugin.get("repo") or "").strip().rstrip("/").lower() if record_repo and plugin_repo == record_repo: return plugin plugin_name = str(plugin.get("name") or "").strip() if plugin_name and plugin_name in record_names: return plugin return None def resolve_market_plugin_entry_by_id( self, market_data: object, market_plugin_id: str, ) -> dict[str, Any] | None: """Resolve a marketplace entry by its exact market plugin ID. Args: market_data: Raw marketplace data from a registry. market_plugin_id: Market plugin ID, usually owner/repo. Returns: The matching marketplace entry, if found. """ normalized_plugin_id = str(market_plugin_id or "").strip() if not normalized_plugin_id: return None for plugin_id, plugin in self.iter_market_plugin_entries(market_data): if plugin_id == normalized_plugin_id: return plugin return None async def resolve_market_install_info( self, payload: dict[str, Any], ) -> dict[str, str] | None: """Validate a marketplace install request and resolve install URLs. Args: payload: Installation request payload. Returns: Marketplace install information, or None for non-market installs. Raises: PluginServiceError: If the marketplace entry cannot be resolved. """ install_method = str(payload.get("install_method") or "").strip().lower() if install_method != "market": return None market_plugin_id = self.get_market_plugin_id(payload) if not market_plugin_id: raise PluginServiceError( "缺少插件市场标识，无法安装。", public_message="缺少插件市场标识，无法安装。", ) registry_url = self.normalize_registry_url(payload.get("registry_url")) market_data, _ = await self.get_online_plugins( custom_registry=registry_url, force_refresh=False, ) market_plugin = self.resolve_market_plugin_entry_by_id( market_data, market_plugin_id, ) if not market_plugin: raise PluginServiceError( "插件源中未找到该插件，无法安装。", public_message="插件源中未找到该插件，无法安装。", ) repo_url = str(market_plugin.get("repo") or "").strip() if not repo_url: raise PluginServiceError( "插件市场记录缺少仓库地址，无法安装。", public_message="插件市场记录缺少仓库地址，无法安装。", ) return { "registry_url": registry_url or "", "market_plugin_id": market_plugin_id, "repo": repo_url, "download_url": str(market_plugin.get("download_url") or "").strip(), } async def resolve_market_update_info(self, plugin_name: str) -> dict[str, Any]: """Resolve update information from the original marketplace source. Args: plugin_name: Installed plugin metadata name. Returns: Update information from the original registry entry. Raises: PluginServiceError: If the plugin cannot be updated from a marketplace. """ plugin = self.find_plugin_by_name(plugin_name) if not plugin: raise PluginServiceError("插件不存在") records = await self.get_plugin_install_sources() record = self.resolve_plugin_install_source(plugin, records) if not isinstance(record, dict) or record.get("implicit"): raise PluginServiceError( PLUGIN_UPDATE_SOURCE_REQUIRED_MESSAGE, public_message=PLUGIN_UPDATE_SOURCE_REQUIRED_MESSAGE, ) install_method = str(record.get("install_method") or "").strip().lower() if install_method == "github": repo_url = str(record.get("repo") or plugin.repo or "").strip() if not repo_url: raise PluginServiceError( PLUGIN_UPDATE_DISABLED_MESSAGE, public_message=PLUGIN_UPDATE_DISABLED_MESSAGE, ) return { "record": record, "market_plugin": None, "download_url": "", "repo": repo_url, } if install_method != "market": raise PluginServiceError( PLUGIN_UPDATE_DISABLED_MESSAGE, public_message=PLUGIN_UPDATE_DISABLED_MESSAGE, ) registry_url = self.normalize_registry_url(record.get("registry_url")) market_data, _ = await self.get_online_plugins( custom_registry=registry_url, force_refresh=False, ) market_plugin = self.resolve_market_plugin_entry(market_data, record) if not market_plugin: raise PluginServiceError( "原插件源中未找到该插件，无法更新。", public_message="原插件源中未找到该插件，无法更新。", ) return { "record": record, "market_plugin": market_plugin, "download_url": str( market_plugin.get("download_url") or record.get("download_url") or "" ).strip(), "repo": str(market_plugin.get("repo") or record.get("repo") or "").strip(), } async def bind_plugin_market_source(self, data: object) -> tuple[dict, str]: """Bind an installed plugin to a marketplace source for future updates. Args: data: Binding request payload. Returns: The persisted install source record and a success message. Raises: PluginServiceError: If the plugin or marketplace entry cannot be matched. """ self._ensure_not_demo() payload = data if isinstance(data, dict) else {} plugin_name = str(payload.get("name") or "").strip() plugin = self.find_plugin_by_name(plugin_name) if not plugin or not plugin.root_dir_name: raise PluginServiceError( "插件不存在", public_message="插件不存在", ) plugin_repo = str(plugin.repo or "").strip() if not plugin_repo: raise PluginServiceError( "当前插件缺少仓库地址，无法更换插件源。", public_message="当前插件缺少仓库地址，无法更换插件源。", ) install_method = str(payload.get("install_method") or "market").strip().lower() if install_method in {"github", "repo"}: records = await self.get_plugin_install_sources() old_record = self.resolve_plugin_install_source(plugin, records) installed_at = ( old_record.get("installed_at") if isinstance(old_record, dict) and old_record.get("installed_at") else self.get_plugin_installed_at(plugin) or datetime.now(timezone.utc).isoformat() ) record = { "schema_version": 1, "root_dir_name": plugin.root_dir_name, "install_method": "github", "registry_url": None, "registry_name": "Repository", "repo": plugin_repo, "download_url": "", "installed_at": installed_at, "updated_at": datetime.now(timezone.utc).isoformat(), } plugin_name_value = str(plugin.name or "").strip() if plugin_name_value: record["name"] = plugin_name_value record["marketplace_name"] = plugin_name_value.replace("_", "-") for key in (plugin.root_dir_name, plugin.name): if key: records.pop(key, None) records[plugin.root_dir_name] = record await self.save_plugin_install_sources(records) return record, "插件源已更新。" if install_method != "market": raise PluginServiceError( "不支持的插件源类型。", public_message="不支持的插件源类型。", ) market_plugin_id = self.get_market_plugin_id(payload) if not market_plugin_id: raise PluginServiceError( "缺少插件市场标识，无法更换插件源。", public_message="缺少插件市场标识，无法更换插件源。", ) registry_url = self.normalize_registry_url(payload.get("registry_url")) market_data, _ = await self.get_online_plugins( custom_registry=registry_url, force_refresh=False, ) market_plugin = self.resolve_market_plugin_entry_by_id( market_data, market_plugin_id, ) if not market_plugin: raise PluginServiceError( "插件源中未找到该插件，无法更换插件源。", public_message="插件源中未找到该插件，无法更换插件源。", ) repo_url = str(market_plugin.get("repo") or "").strip() if not repo_url: raise PluginServiceError( "插件市场记录缺少仓库地址，无法更换插件源。", public_message="插件市场记录缺少仓库地址，无法更换插件源。", ) plugin_repo_identifier = self.repo_identifier_from_url(plugin_repo) market_repo_identifier = self.repo_identifier_from_url(repo_url) repo_matches = False if plugin_repo_identifier or market_repo_identifier: repo_matches = ( bool(plugin_repo_identifier) and bool(market_repo_identifier) and plugin_repo_identifier == market_repo_identifier ) else: repo_matches = ( plugin_repo.strip().rstrip("/").lower() == repo_url.strip().rstrip("/").lower() ) if not repo_matches: raise PluginServiceError( "插件仓库地址与所选插件源不一致，无法更换插件源。", public_message="插件仓库地址与所选插件源不一致，无法更换插件源。", ) registry_name = await self.resolve_registry_name(registry_url) records = await self.get_plugin_install_sources() old_record = self.resolve_plugin_install_source(plugin, records) installed_at = ( old_record.get("installed_at") if isinstance(old_record, dict) and old_record.get("installed_at") else self.get_plugin_installed_at(plugin) or datetime.now(timezone.utc).isoformat() ) record = { "schema_version": 1, "root_dir_name": plugin.root_dir_name, "install_method": "market", "registry_url": registry_url, "registry_name": registry_name, "market_plugin_id": market_plugin_id, "repo": repo_url, "download_url": str(market_plugin.get("download_url") or "").strip(), "installed_at": installed_at, "updated_at": datetime.now(timezone.utc).isoformat(), } for key in (plugin.root_dir_name, plugin.name): if key: records.pop(key, None) records[plugin.root_dir_name] = record await self.save_plugin_install_sources(records) return record, "插件源已更新。" async def refresh_plugin_install_source_after_update( self, plugin_name: str, update_info: dict[str, Any], ) -> None: """Refresh stored source fields after a successful plugin update. Args: plugin_name: Installed plugin metadata name. update_info: Marketplace update information. """ plugin = self.find_plugin_by_name(plugin_name) if not plugin or not plugin.root_dir_name: return records = await self.get_plugin_install_sources() record = records.get(plugin.root_dir_name) if not isinstance(record, dict): update_record = update_info.get("record") if not isinstance(update_record, dict): return record = dict(update_record) market_plugin = update_info.get("market_plugin") if isinstance(market_plugin, dict): market_plugin_id = self.get_market_plugin_id(market_plugin) if market_plugin_id: record["market_plugin_id"] = market_plugin_id record["repo"] = str( market_plugin.get("repo") or record.get("repo") or "" ).strip() record["download_url"] = str( market_plugin.get("download_url") or update_info.get("download_url") or record.get("download_url") or "" ).strip() record["root_dir_name"] = plugin.root_dir_name record["updated_at"] = datetime.now(timezone.utc).isoformat() records[plugin.root_dir_name] = record await self.save_plugin_install_sources(records) async def install_plugin(self, data: object) -> tuple[dict, str]: self._ensure_not_demo() payload = data if isinstance(data, dict) else {} repo_url = str(payload.get("url") or "").strip() download_url = str(payload.get("download_url") or "").strip() ignore_version_check = bool(payload.get("ignore_version_check", False)) market_install_info = await self.resolve_market_install_info(payload) if market_install_info: repo_url = market_install_info["repo"] download_url = market_install_info["download_url"] payload = { **payload, "registry_url": market_install_info["registry_url"] or None, "market_plugin_id": market_install_info["market_plugin_id"], } if not repo_url: raise PluginServiceError("缺少插件仓库地址") proxy: str | None = payload.get("proxy", None) if proxy: proxy = proxy.removesuffix("/") try: logger.info(f"正在安装插件 {repo_url}") plugin_info = await self.plugin_manager.install_plugin( repo_url, proxy or "", ignore_version_check=ignore_version_check, download_url=download_url, ) await self.persist_plugin_install_source( plugin_info, payload, fallback_method="github" if self.repo_identifier_from_url(repo_url) else "url", repo_url=repo_url, download_url=download_url, ) await self.sync_skills_after_plugin_change() logger.info(f"安装插件 {repo_url} 成功。") return plugin_info or {}, "安装成功。" except PluginVersionUnsupportedError as exc: raise PluginServiceWarning( str(exc), { "warning_type": "astrbot_version_unsupported", "can_ignore": True, }, public_message="当前 AstrBot 版本不满足插件要求", ) from exc async def validate_plugin_repo(self, data: object) -> tuple[dict[str, Any], str]: """Validate whether a GitHub repository contains AstrBot plugin metadata. Args: data: Dashboard request payload containing repository or url. Returns: Plugin metadata fetched from the GitHub repository and a success message. Raises: PluginServiceError: If the repository is not a valid AstrBot plugin. """ payload = self._payload(data) repo_url = str(payload.get("url") or payload.get("repository") or "").strip() if not repo_url: raise PluginServiceError("缺少插件仓库地址") if not repo_url.startswith(("http://", "https://")): repo_url = f"https://github.com/{repo_url}" proxy = str(payload.get("proxy") or "").strip().removesuffix("/") try: ( author, repo, branch, ) = await self.plugin_manager.updator.resolve_github_source_branch(repo_url) except ValueError as exc: raise PluginServiceError( "请输入有效的 GitHub 仓库地址。", public_message="请输入有效的 GitHub 仓库地址。", ) from exc ssl_context = ssl.create_default_context(cafile=certifi.where()) connector = aiohttp.TCPConnector(ssl=ssl_context) try: async with aiohttp.ClientSession( trust_env=True, connector=connector, timeout=aiohttp.ClientTimeout( total=PLUGIN_REPO_VALIDATE_TIMEOUT_SECONDS ), ) as session: for filename in PLUGIN_METADATA_FILENAMES: raw_url = ( f"https://raw.githubusercontent.com/" f"{author}/{repo}/{branch}/{filename}" ) request_url = f"{proxy}/{raw_url}" if proxy else raw_url async with session.get(request_url) as response: if response.status != 200: continue content_length = response.headers.get("Content-Length") if content_length: try: if int(content_length) > PLUGIN_METADATA_MAX_BYTES: raise PluginServiceError( f"{filename} 超过 1MB。", public_message=f"{filename} 超过 1MB。", ) except ValueError: pass metadata_bytes = await response.content.read( PLUGIN_METADATA_MAX_BYTES + 1 ) if len(metadata_bytes) > PLUGIN_METADATA_MAX_BYTES: raise PluginServiceError( f"{filename} 超过 1MB。", public_message=f"{filename} 超过 1MB。", ) try: metadata_text = metadata_bytes.decode("utf-8") except UnicodeDecodeError as exc: raise PluginServiceError( f"{filename} 必须使用 UTF-8 编码。", public_message=f"{filename} 必须使用 UTF-8 编码。", ) from exc try: metadata = yaml.safe_load(metadata_text) except yaml.YAMLError as exc: raise PluginServiceError( f"{filename} 格式错误。", public_message=f"{filename} 格式错误。", ) from exc try: self.plugin_manager.updator.validate_plugin_metadata( metadata, filename, ) except ValueError as exc: raise PluginServiceError( str(exc), public_message=f"插件校验失败：{exc!s}", ) from exc metadata = dict(metadata) if isinstance(metadata, dict) else {} if "desc" not in metadata and "description" in metadata: metadata["desc"] = metadata["description"] return { "valid": True, "metadata_entry": filename, "metadata_branch": branch, "name": str(metadata.get("name") or ""), "display_name": metadata.get("display_name"), "desc": str(metadata.get("desc") or ""), "version": str(metadata.get("version") or ""), "author": metadata.get("author"), "repo": str(metadata.get("repo") or repo_url), }, "插件校验通过。" raise PluginServiceError( "未在 GitHub 仓库根目录找到 metadata.yaml 或 metadata.yml。", public_message="未在 GitHub 仓库根目录找到 metadata.yaml 或 metadata.yml。", ) except Exception as exc: if isinstance(exc, PluginServiceError): raise logger.warning("插件仓库校验失败 %s: %s", repo_url, exc) raise PluginServiceError( "插件校验失败", public_message="插件校验失败，请查看服务端日志。", ) from exc async def install_plugin_upload( self, *, upload_file, ignore_version_check: bool, ) -> tuple[dict, str]: self._ensure_not_demo() logger.info(f"正在安装用户上传的插件 {upload_file.filename}") filename = str(upload_file.filename or "plugin.zip").replace("\\", "/") file_path = os.path.join( get_astrbot_temp_path(), f"plugin_upload_{os.path.basename(filename) or 'plugin.zip'}", ) await upload_file.save(file_path) try: plugin_info = await self.plugin_manager.install_plugin_from_file( file_path, ignore_version_check=ignore_version_check, ) await self.persist_plugin_install_source( plugin_info, {}, fallback_method="upload", repo_url="", download_url="", ) await self.sync_skills_after_plugin_change() logger.info(f"安装插件 {upload_file.filename} 成功") return plugin_info or {}, "安装成功。" except PluginVersionUnsupportedError as exc: raise PluginServiceWarning( str(exc), { "warning_type": "astrbot_version_unsupported", "can_ignore": True, }, public_message="当前 AstrBot 版本不满足插件要求", ) from exc async def install_plugin_upload_from_dashboard_form( self, *, upload_file, ignore_version_check, ) -> tuple[dict, str]: return await self.install_plugin_upload( upload_file=upload_file, ignore_version_check=self._to_bool(ignore_version_check), ) async def uninstall_plugin(self, data: object) -> tuple[None, str]: self._ensure_not_demo() payload = data if isinstance(data, dict) else {} plugin_name = payload["name"] delete_config = payload.get("delete_config", False) delete_data = payload.get("delete_data", False) logger.info(f"正在卸载插件 {plugin_name}") plugin = self.find_plugin_by_name(plugin_name) root_dir_name = plugin.root_dir_name if plugin else None await self.plugin_manager.uninstall_plugin( plugin_name, delete_config=delete_config, delete_data=delete_data, ) await self.remove_plugin_install_source(root_dir_name) await self.sync_skills_after_plugin_change() logger.info(f"卸载插件 {plugin_name} 成功") return None, "卸载成功" async def uninstall_failed_plugin(self, data: object) -> tuple[None, str]: self._ensure_not_demo() payload = data if isinstance(data, dict) else {} dir_name = payload.get("dir_name", "") delete_config = payload.get("delete_config", False) delete_data = payload.get("delete_data", False) if not dir_name: raise PluginServiceError("缺少失败插件目录名") logger.info(f"正在卸载失败插件 {dir_name}") await self.plugin_manager.uninstall_failed_plugin( dir_name, delete_config=delete_config, delete_data=delete_data, ) await self.sync_skills_after_plugin_change() logger.info(f"卸载失败插件 {dir_name} 成功") return None, "卸载成功" async def update_plugin(self, data: object) -> tuple[None, str]: self._ensure_not_demo() payload = data if isinstance(data, dict) else {} plugin_name = payload["name"] proxy: str | None = payload.get("proxy", None) update_info = await self.resolve_market_update_info(plugin_name) download_url = str(update_info.get("download_url") or "").strip() logger.info(f"正在更新插件 {plugin_name}") await self.plugin_manager.update_plugin( plugin_name, proxy or "", download_url=download_url ) await self.refresh_plugin_install_source_after_update(plugin_name, update_info) await self.plugin_manager.reload(plugin_name) await self.sync_skills_after_plugin_change() logger.info(f"更新插件 {plugin_name} 成功。") return None, "更新成功。" async def update_all_plugins(self, data: object) -> tuple[dict, str]: self._ensure_not_demo() payload = data if isinstance(data, dict) else {} plugin_names: list[str] = payload.get("names") or [] proxy: str = payload.get("proxy", "") if not isinstance(plugin_names, list) or not plugin_names: raise PluginServiceError("插件列表不能为空") results = [] sem = asyncio.Semaphore(PLUGIN_UPDATE_CONCURRENCY) async def _update_one(name: str): async with sem: try: logger.info(f"批量更新插件 {name}") update_info = await self.resolve_market_update_info(name) download_url = str(update_info.get("download_url") or "").strip() await self.plugin_manager.update_plugin( name, proxy, download_url=download_url ) await self.refresh_plugin_install_source_after_update( name, update_info, ) return {"name": name, "status": "ok", "message": "更新成功"} except PluginServiceError as exc: logger.error( f"/api/plugin/update-all: 更新插件 {name} 失败: {exc}", ) return { "name": name, "status": "error", "message": exc.public_message, } except Exception: logger.error( f"/api/plugin/update-all: 更新插件 {name} 失败", exc_info=True, ) return { "name": name, "status": "error", "message": PLUGIN_UPDATE_FAILED_MESSAGE, } raw_results = await asyncio.gather( *(_update_one(name) for name in plugin_names), return_exceptions=True, ) for name, result in zip(plugin_names, raw_results): if isinstance(result, asyncio.CancelledError): raise result if isinstance(result, BaseException): logger.error( f"/api/plugin/update-all: 更新插件 {name} 任务失败: {result!r}" ) results.append( { "name": name, "status": "error", "message": PLUGIN_UPDATE_FAILED_MESSAGE, } ) else: results.append(result) failed = [result for result in results if result["status"] == "error"] if len(failed) < len(results): await self.sync_skills_after_plugin_change() message = ( "批量更新完成，全部成功。" if not failed else f"批量更新完成，其中 {len(failed)}/{len(results)} 个插件失败。" ) return {"results": results}, message async def set_plugin_enabled( self, data: object, *, enabled: bool ) -> tuple[None, str]: self._ensure_not_demo() payload = data if isinstance(data, dict) else {} plugin_name = payload["name"] if enabled: await self.plugin_manager.turn_on_plugin(plugin_name) message = "启用成功。" log_action = "启用" else: await self.plugin_manager.turn_off_plugin(plugin_name) message = "停用成功。" log_action = "停用" await self.sync_skills_after_plugin_change() logger.info(f"{log_action}插件 {plugin_name} 。") return None, message def resolve_plugin_dir(self, plugin_name: str) -> Path: if not plugin_name: raise PluginServiceError("插件名称不能为空") plugin_obj = None for plugin in self.plugin_manager.context.get_all_stars(): if plugin.name == plugin_name: plugin_obj = plugin break if not plugin_obj: raise PluginServiceError(f"插件 {plugin_name} 不存在") if not plugin_obj.root_dir_name: raise PluginServiceError(f"插件 {plugin_name} 目录不存在") plugin_dir = ( Path( self.plugin_manager.reserved_plugin_path if plugin_obj.reserved else self.plugin_manager.plugin_store_path ) / plugin_obj.root_dir_name ) if not plugin_dir.is_dir(): raise PluginServiceError(f"无法找到插件 {plugin_name} 的目录") return plugin_dir def get_plugin_readme(self, plugin_name: str | None) -> tuple[dict, str]: if not plugin_name: logger.warning("插件名称为空") raise PluginServiceError("插件名称不能为空") plugin_dir = self.resolve_plugin_dir(plugin_name) readme_path = plugin_dir / "README.md" if not readme_path.is_file(): logger.warning(f"插件 {plugin_name} 没有README文件") raise PluginServiceError(f"插件 {plugin_name} 没有README文件") try: return { "content": readme_path.read_text(encoding="utf-8") }, "成功获取README内容" except Exception as exc: logger.warning(f"读取插件 {plugin_name} README 文件失败: {exc}") raise PluginServiceError( "读取README文件失败", public_message="读取README文件失败", ) from exc def get_plugin_readme_from_dashboard_query( self, plugin_name: str | None, ) -> tuple[dict, str]: return self.get_plugin_readme(plugin_name) def get_plugin_changelog(self, plugin_name: str | None) -> tuple[dict, str]: logger.debug(f"正在获取插件 {plugin_name} 的更新日志") if not plugin_name: logger.warning("插件名称为空") raise PluginServiceError("插件名称不能为空") plugin_dir = self.resolve_plugin_dir(plugin_name) changelog_names = ["CHANGELOG.md", "changelog.md", "CHANGELOG", "changelog"] for name in changelog_names: changelog_path = plugin_dir / name if not changelog_path.is_file(): continue try: return ( {"content": changelog_path.read_text(encoding="utf-8")}, "成功获取更新日志", ) except Exception as exc: logger.warning(f"读取插件 {plugin_name} 更新日志失败: {exc}") raise PluginServiceError( "读取更新日志失败", public_message="读取更新日志失败", ) from exc logger.warning(f"插件 {plugin_name} 没有更新日志文件") return {"content": None}, "该插件没有更新日志文件" def get_plugin_changelog_from_dashboard_query( self, plugin_name: str | None, ) -> tuple[dict, str]: return self.get_plugin_changelog(plugin_name) @staticmethod async def get_custom_sources() -> list: return PluginService._normalize_custom_sources( await sp.global_get("custom_plugin_sources", []) ) @staticmethod async def save_custom_sources(data: object) -> tuple[None, str]: payload = data if isinstance(data, dict) else {} sources = PluginService._custom_sources_from_payload(payload) await sp.global_put("custom_plugin_sources", sources) return None, "保存成功" @staticmethod def _normalize_custom_sources(raw_sources: object) -> list: return raw_sources if isinstance(raw_sources, list) else [] @staticmethod def _custom_sources_from_payload(payload: dict[str, Any]) -> list: sources = payload.get("sources", []) if not isinstance(sources, list): raise PluginServiceError("sources fields must be a list") return sources @staticmethod async def create_custom_source(data: object) -> list: source = data if isinstance(data, dict) else {} sources = await PluginService.get_custom_sources() sources.append(source) await sp.global_put("custom_plugin_sources", sources) return sources @staticmethod async def replace_custom_sources(data: object) -> list: payload = data if isinstance(data, dict) else {} sources = PluginService._custom_sources_from_payload(payload) await sp.global_put("custom_plugin_sources", sources) return sources @staticmethod async def delete_custom_source(source_id: str) -> list: sources = await PluginService.get_custom_sources() filtered = [ source for source in sources if not ( isinstance(source, dict) and str(source.get("id") or source.get("url") or source.get("name")) == source_id ) and not (isinstance(source, str) and source == source_id) ] await sp.global_put("custom_plugin_sources", filtered) return filtered @staticmethod def _to_bool(value: object, default: bool = False) -> bool: if value is None: return default if isinstance(value, bool): return value if isinstance(value, str): return value.strip().lower() in {"1", "true", "yes", "on"} return bool(value) __all__ = [ "PLUGIN_UPDATE_CONCURRENCY", "PLUGIN_OPERATION_FAILED_MESSAGE", "PLUGIN_UPDATE_DISABLED_MESSAGE", "PLUGIN_UPDATE_FAILED_MESSAGE", "PLUGIN_UPDATE_SOURCE_REQUIRED_MESSAGE", "PluginService", "PluginServiceError", "PluginServiceWarning", "RegistrySource", ]