diff --git a/src/plugin_system/apis/__init__.py b/src/plugin_system/apis/__init__.py index 0e03f6700..49e3e3b25 100644 --- a/src/plugin_system/apis/__init__.py +++ b/src/plugin_system/apis/__init__.py @@ -8,6 +8,7 @@ from src.plugin_system.apis import ( chat_api, component_manage_api, + component_state_api, config_api, database_api, emoji_api, @@ -17,6 +18,7 @@ from src.plugin_system.apis import ( mood_api, permission_api, person_api, + plugin_info_api, plugin_manage_api, schedule_api, send_api, @@ -31,6 +33,7 @@ from .plugin_register_api import register_plugin __all__ = [ "chat_api", "component_manage_api", + "component_state_api", "config_api", "context_api", "database_api", @@ -42,6 +45,7 @@ __all__ = [ "mood_api", "permission_api", "person_api", + "plugin_info_api", "plugin_manage_api", "register_plugin", "schedule_api", diff --git a/src/plugin_system/apis/component_state_api.py b/src/plugin_system/apis/component_state_api.py new file mode 100644 index 000000000..93b4eaa33 --- /dev/null +++ b/src/plugin_system/apis/component_state_api.py @@ -0,0 +1,357 @@ +""" +Component State API +=================== + +该模块提供了用于管理组件启用/禁用状态的核心API。 +支持全局和局部（临时）范围的组件状态控制，以及批量操作。 + +主要功能包括： +- 组件的全局和局部启用/禁用 +- 批量组件状态管理 +- 组件状态查询 +""" + +from typing import Any + +from src.common.logger import get_logger +from src.plugin_system.base.component_types import ComponentType +from src.plugin_system.core.component_registry import ComponentInfo, component_registry + +# 初始化日志记录器 +logger = get_logger("component_state_api") + + +# -------------------------------------------------------------------------------- +# Section 1: 组件状态管理 (Component State Management) +# -------------------------------------------------------------------------------- +# 这部分 API 负责控制单个组件的启用和禁用状态，支持全局和局部（临时）范围。 + + +async def set_component_enabled(name: str, component_type: ComponentType, enabled: bool) -> bool: + """ + 在全局范围内启用或禁用一个组件。 + + 此更改会直接修改组件在注册表中的状态，但此状态是临时的，不会持久化到配置文件中。 + 包含一个保护机制，防止禁用最后一个已启用的 Chatter 组件。 + + Args: + name (str): 要操作的组件的名称。 + component_type (ComponentType): 组件的类型。 + enabled (bool): True 表示启用, False 表示禁用。 + + Returns: + bool: 如果操作成功，则为 True。 + """ + # 特殊保护：确保系统中至少有一个 Chatter 组件处于启用状态 + if component_type == ComponentType.CHATTER and not enabled: + enabled_chatters = component_registry.get_enabled_components_by_type(ComponentType.CHATTER) + if len(enabled_chatters) <= 1 and name in enabled_chatters: + logger.warning(f"操作被阻止：不能禁用最后一个启用的 Chatter 组件 ('{name}')。") + return False + + # 根据 enabled 参数调用相应的注册表方法 + if enabled: + return component_registry.enable_component(name, component_type) + else: + return await component_registry.disable_component(name, component_type) + + +def set_component_enabled_local(stream_id: str, name: str, component_type: ComponentType, enabled: bool) -> bool: + """ + 在一个特定的 stream_id 上下文中临时启用或禁用组件。 + + 此状态仅存在于内存中，并且只对指定的 stream_id 有效，不影响全局组件状态。 + 同样包含对 Chatter 组件的保护机制。 + + Args: + stream_id (str): 唯一的上下文标识符，例如一个会话ID。 + name (str): 组件名称。 + component_type (ComponentType): 组件类型。 + enabled (bool): True 为启用, False 为禁用。 + + Returns: + bool: 如果操作成功，则为 True。 + """ + # 首先，验证组件是否存在 + component_info = component_registry.get_component_info(name, component_type) + if not component_info: + logger.error(f"尝试设置局部状态失败：未找到组件 {name} ({component_type.value})。") + return False + + # Chatter 唯一性保护（在 stream_id 上下文中） + if component_type == ComponentType.CHATTER and not enabled: + enabled_chatters = component_registry.get_enabled_components_by_type(ComponentType.CHATTER, stream_id=stream_id) + if len(enabled_chatters) <= 1 and name in enabled_chatters: + logger.warning(f"操作被阻止：在 stream '{stream_id}' 中，不能禁用最后一个启用的 Chatter 组件 ('{name}')。") + return False + + # 设置局部状态 + component_registry.set_local_component_state(stream_id, name, component_type, enabled) + logger.info(f"在 stream '{stream_id}' 中，组件 {name} ({component_type.value}) 的局部状态已设置为: {enabled}") + return True + + +def clear_local_component_states(stream_id: str) -> None: + """ + 清除指定会话的所有局部组件状态。 + + 当会话结束时应调用此方法来清理资源。 + + Args: + stream_id (str): 要清除状态的会话ID。 + """ + component_registry.clear_local_component_states(stream_id) + logger.debug(f"已清除 stream '{stream_id}' 的所有局部组件状态。") + + +def is_component_enabled(name: str, component_type: ComponentType, stream_id: str | None = None) -> bool: + """ + 检查组件是否在指定上下文中启用。 + + Args: + name (str): 组件名称。 + component_type (ComponentType): 组件类型。 + stream_id (str | None): 可选的会话ID，用于检查局部状态。 + + Returns: + bool: 如果组件启用，则为 True。 + """ + return component_registry.is_component_available(name, component_type, stream_id) + + +def get_component_state(name: str, component_type: ComponentType) -> dict[str, Any] | None: + """ + 获取组件的详细状态信息。 + + Args: + name (str): 组件名称。 + component_type (ComponentType): 组件类型。 + + Returns: + dict | None: 包含组件状态信息的字典，如果组件不存在则返回 None。 + """ + component_info = component_registry.get_component_info(name, component_type) + if not component_info: + return None + + return { + "name": component_info.name, + "component_type": component_info.component_type.value, + "plugin_name": component_info.plugin_name, + "enabled": component_info.enabled, + "description": component_info.description, + } + + +# -------------------------------------------------------------------------------- +# Section 2: 批量组件状态管理 (Batch Component State Management) +# -------------------------------------------------------------------------------- +# 这部分 API 提供批量操作组件状态的功能。 + + +async def enable_all_plugin_components(plugin_name: str) -> dict[str, bool]: + """ + 启用指定插件下的所有组件。 + + Args: + plugin_name (str): 插件名称。 + + Returns: + dict[str, bool]: 每个组件名称及其启用操作是否成功的字典。 + """ + plugin_info = component_registry.get_plugin_info(plugin_name) + if not plugin_info: + logger.error(f"未找到插件 '{plugin_name}'，无法启用其组件。") + return {} + + results = {} + for component_info in plugin_info.components: + success = component_registry.enable_component(component_info.name, component_info.component_type) + results[component_info.name] = success + if success: + logger.debug(f"已启用组件: {component_info.name} ({component_info.component_type.value})") + else: + logger.warning(f"启用组件失败: {component_info.name} ({component_info.component_type.value})") + + logger.info(f"已完成启用插件 '{plugin_name}' 的所有组件，成功: {sum(results.values())}/{len(results)}") + return results + + +async def disable_all_plugin_components(plugin_name: str) -> dict[str, bool]: + """ + 禁用指定插件下的所有组件。 + + 包含对 Chatter 组件的保护机制。 + + Args: + plugin_name (str): 插件名称。 + + Returns: + dict[str, bool]: 每个组件名称及其禁用操作是否成功的字典。 + """ + plugin_info = component_registry.get_plugin_info(plugin_name) + if not plugin_info: + logger.error(f"未找到插件 '{plugin_name}'，无法禁用其组件。") + return {} + + results = {} + for component_info in plugin_info.components: + # Chatter 保护检查 + if component_info.component_type == ComponentType.CHATTER: + enabled_chatters = component_registry.get_enabled_components_by_type(ComponentType.CHATTER) + if len(enabled_chatters) <= 1 and component_info.name in enabled_chatters: + logger.warning( + f"跳过禁用最后一个 Chatter 组件 '{component_info.name}'，系统至少需要一个启用的 Chatter。" + ) + results[component_info.name] = False + continue + + success = await component_registry.disable_component(component_info.name, component_info.component_type) + results[component_info.name] = success + if success: + logger.debug(f"已禁用组件: {component_info.name} ({component_info.component_type.value})") + else: + logger.warning(f"禁用组件失败: {component_info.name} ({component_info.component_type.value})") + + logger.info(f"已完成禁用插件 '{plugin_name}' 的所有组件，成功: {sum(results.values())}/{len(results)}") + return results + + +async def set_components_enabled_by_type( + plugin_name: str, component_type: ComponentType, enabled: bool +) -> dict[str, bool]: + """ + 启用或禁用指定插件下特定类型的所有组件。 + + Args: + plugin_name (str): 插件名称。 + component_type (ComponentType): 要操作的组件类型。 + enabled (bool): True 为启用，False 为禁用。 + + Returns: + dict[str, bool]: 每个组件名称及其操作是否成功的字典。 + """ + plugin_info = component_registry.get_plugin_info(plugin_name) + if not plugin_info: + logger.error(f"未找到插件 '{plugin_name}'。") + return {} + + results = {} + for component_info in plugin_info.components: + if component_info.component_type != component_type: + continue + + success = await set_component_enabled(component_info.name, component_type, enabled) + results[component_info.name] = success + + action = "启用" if enabled else "禁用" + logger.info( + f"已完成{action}插件 '{plugin_name}' 的所有 {component_type.value} 组件，" + f"成功: {sum(results.values())}/{len(results)}" + ) + return results + + +async def batch_set_components_enabled(components: list[tuple[str, ComponentType]], enabled: bool) -> dict[str, bool]: + """ + 批量启用或禁用多个组件。 + + Args: + components (list[tuple[str, ComponentType]]): 要操作的组件列表， + 每个元素为 (组件名称, 组件类型) 元组。 + enabled (bool): True 为启用，False 为禁用。 + + Returns: + dict[str, bool]: 每个组件名称及其操作是否成功的字典。 + """ + results = {} + for name, component_type in components: + success = await set_component_enabled(name, component_type, enabled) + results[name] = success + + action = "启用" if enabled else "禁用" + logger.info(f"批量{action}操作完成，成功: {sum(results.values())}/{len(results)}") + return results + + +# -------------------------------------------------------------------------------- +# Section 3: 组件状态查询与筛选 (Component State Query & Filter) +# -------------------------------------------------------------------------------- +# 这部分 API 提供组件状态的查询和筛选功能。 + + +def get_components_by_state( + component_type: ComponentType | None = None, + enabled: bool | None = None, + plugin_name: str | None = None, +) -> list[ComponentInfo]: + """ + 根据条件筛选组件。 + + Args: + component_type (ComponentType | None): 按组件类型筛选。 + enabled (bool | None): 按启用状态筛选。 + plugin_name (str | None): 按插件名称筛选。 + + Returns: + list[ComponentInfo]: 符合条件的组件信息列表。 + """ + results = [] + + # 确定要搜索的组件类型 + types_to_search = [component_type] if component_type else list(ComponentType) + + for comp_type in types_to_search: + components = component_registry.get_components_by_type(comp_type) + for info in components.values(): + # 按启用状态筛选 + if enabled is not None and info.enabled != enabled: + continue + # 按插件名称筛选 + if plugin_name is not None and info.plugin_name != plugin_name: + continue + results.append(info) + + return results + + +def get_disabled_components(plugin_name: str | None = None) -> list[ComponentInfo]: + """ + 获取所有被禁用的组件。 + + Args: + plugin_name (str | None): 可选，仅获取指定插件的禁用组件。 + + Returns: + list[ComponentInfo]: 禁用组件的信息列表。 + """ + return get_components_by_state(enabled=False, plugin_name=plugin_name) + + +def get_enabled_components(plugin_name: str | None = None) -> list[ComponentInfo]: + """ + 获取所有已启用的组件。 + + Args: + plugin_name (str | None): 可选，仅获取指定插件的启用组件。 + + Returns: + list[ComponentInfo]: 启用组件的信息列表。 + """ + return get_components_by_state(enabled=True, plugin_name=plugin_name) + + +def get_component_count(component_type: ComponentType, stream_id: str | None = None) -> int: + """ + 获取指定类型的已加载并启用的组件的总数。 + + 可以根据 `stream_id` 考虑局部状态，从而获得特定上下文中的组件数量。 + + Args: + component_type (ComponentType): 要查询的组件类型。 + stream_id (str | None): 可选的上下文ID。如果提供，将计入局部状态。 + + Returns: + int: 该类型下已启用的组件的数量。 + """ + return len(component_registry.get_enabled_components_by_type(component_type, stream_id=stream_id)) diff --git a/src/plugin_system/apis/plugin_info_api.py b/src/plugin_system/apis/plugin_info_api.py new file mode 100644 index 000000000..b01df5402 --- /dev/null +++ b/src/plugin_system/apis/plugin_info_api.py @@ -0,0 +1,454 @@ +""" +Plugin Info API +=============== + +该模块提供了用于查询插件和组件信息、生成报告和统计数据的API。 + +主要功能包括： +- 系统状态报告生成 +- 插件详情查询 +- 组件列表和搜索 +- 状态统计 +- 工具函数 +""" + +from typing import Any, Literal + +from src.common.logger import get_logger +from src.plugin_system.base.component_types import ComponentType +from src.plugin_system.core.component_registry import ComponentInfo, component_registry +from src.plugin_system.core.plugin_manager import plugin_manager + +# 初始化日志记录器 +logger = get_logger("plugin_info_api") + + +# -------------------------------------------------------------------------------- +# Section 1: 信息查询与报告 (Information Querying & Reporting) +# -------------------------------------------------------------------------------- +# 这部分 API 用于获取关于插件和组件的详细信息、列表和统计数据。 + + +def get_system_report() -> dict[str, Any]: + """ + 生成一份详细的系统状态报告。 + + 报告包含已加载插件、失败插件和组件的全面信息，是调试和监控系统状态的核心工具。 + + Returns: + dict[str, Any]: 包含系统、插件和组件状态的详细报告字典。 + """ + loaded_plugins_info = {} + # 遍历所有已加载的插件实例 + for name, instance in plugin_manager.loaded_plugins.items(): + plugin_info = component_registry.get_plugin_info(name) + if not plugin_info: + continue + + # 收集该插件下所有组件的详细信息 + components_details = [ + { + "name": comp_info.name, + "component_type": comp_info.component_type.value, + "description": comp_info.description, + "enabled": comp_info.enabled, + } + for comp_info in plugin_info.components + ] + + # 构建单个插件的信息字典 + # 元数据从 PluginInfo 获取，而启用状态(enable_plugin)从插件实例获取 + loaded_plugins_info[name] = { + "display_name": plugin_info.display_name or name, + "version": plugin_info.version, + "author": plugin_info.author, + "enabled": instance.enable_plugin, + "components": components_details, + } + + # 构建最终的完整报告 + report = { + "system_info": { + "loaded_plugins_count": len(plugin_manager.loaded_plugins), + "total_components_count": component_registry.get_registry_stats().get("total_components", 0), + }, + "plugins": loaded_plugins_info, + "failed_plugins": plugin_manager.failed_plugins, + } + return report + + +def get_plugin_details(plugin_name: str) -> dict[str, Any] | None: + """ + 获取单个插件的详细报告。 + + 报告内容包括插件的元数据、所有组件的详细信息及其当前状态。 + 这是 `get_system_report` 的单插件聚焦版本。 + + Args: + plugin_name (str): 要查询的插件名称。 + + Returns: + dict | None: 包含插件详细信息的字典，如果插件未注册则返回 None。 + """ + plugin_info = component_registry.get_plugin_info(plugin_name) + if not plugin_info: + logger.warning(f"尝试获取插件详情失败：未找到名为 '{plugin_name}' 的插件。") + return None + + # 收集该插件下所有组件的信息 + components_details = [ + { + "name": comp_info.name, + "component_type": comp_info.component_type.value, + "description": comp_info.description, + "enabled": comp_info.enabled, + } + for comp_info in plugin_info.components + ] + + # 获取插件实例以检查其启用状态 + plugin_instance = plugin_manager.get_plugin_instance(plugin_name) + is_enabled = plugin_instance.enable_plugin if plugin_instance else False + + # 组装详细信息字典 + return { + "name": plugin_info.name, + "display_name": plugin_info.display_name or plugin_info.name, + "version": plugin_info.version, + "author": plugin_info.author, + "license": plugin_info.license, + "description": plugin_info.description, + "enabled": is_enabled, + "status": "loaded" if is_plugin_loaded(plugin_name) else "registered", + "components": components_details, + } + + +def list_plugins(status: Literal["loaded", "registered", "failed"]) -> list[str]: + """ + 根据指定的状态列出插件名称列表。 + + 提供了一种快速、便捷的方式来监控和调试插件系统，而无需解析完整的系统报告。 + + Args: + status (str): 插件状态，可选值为 'loaded', 'registered', 'failed'。 + + Returns: + list[str]: 对应状态的插件名称列表。 + + Raises: + ValueError: 如果传入了无效的状态字符串。 + """ + if status == "loaded": + # 返回所有当前已成功加载的插件 + return plugin_manager.list_loaded_plugins() + if status == "registered": + # 返回所有已注册（但不一定已加载）的插件 + return plugin_manager.list_registered_plugins() + if status == "failed": + # 返回所有加载失败的插件的名称 + return list(plugin_manager.failed_plugins.keys()) + # 如果状态无效，则引发错误 + raise ValueError(f"无效的插件状态: '{status}'。有效选项为 'loaded', 'registered', 'failed'。") + + +def list_components(component_type: ComponentType, enabled_only: bool = True) -> list[dict[str, Any]]: + """ + 列出指定类型的所有组件的详细信息。 + + 这是查找和管理组件的核心功能，例如，获取所有可用的工具或所有注册的聊天器。 + + Args: + component_type (ComponentType): 要查询的组件类型。 + enabled_only (bool, optional): 是否只返回已启用的组件。默认为 True。 + + Returns: + list[dict[str, Any]]: 一个包含组件信息字典的列表。 + """ + # 根据 enabled_only 参数决定是获取所有组件还是仅获取已启用的组件 + if enabled_only: + components = component_registry.get_enabled_components_by_type(component_type) + else: + components = component_registry.get_components_by_type(component_type) + + # 将组件信息格式化为字典列表 + return [ + { + "name": info.name, + "plugin_name": info.plugin_name, + "description": info.description, + "enabled": info.enabled, + } + for info in components.values() + ] + + +def search_components_by_name( + name_keyword: str, + component_type: ComponentType | None = None, + case_sensitive: bool = False, + exact_match: bool = False, +) -> list[dict[str, Any]]: + """ + 根据名称关键字搜索组件，支持模糊匹配和精确匹配。 + + 极大地增强了组件的可发现性，用户无需知道完整名称即可找到所需组件。 + + Args: + name_keyword (str): 用于搜索的名称关键字。 + component_type (ComponentType | None, optional): 如果提供，则只在该类型中搜索。默认为 None (搜索所有类型)。 + case_sensitive (bool, optional): 是否进行大小写敏感的搜索。默认为 False。 + exact_match (bool, optional): 是否进行精确匹配。默认为 False (模糊匹配)。 + + Returns: + list[dict[str, Any]]: 匹配的组件信息字典的列表。 + """ + results = [] + # 如果未指定组件类型，则搜索所有类型 + types_to_search = [component_type] if component_type else list(ComponentType) + + # 根据是否大小写敏感，预处理搜索关键字 + compare_str = name_keyword if case_sensitive else name_keyword.lower() + + # 遍历要搜索的组件类型 + for comp_type in types_to_search: + all_components = component_registry.get_components_by_type(comp_type) + for name, info in all_components.items(): + # 同样地，预处理组件名称 + target_name = name if case_sensitive else name.lower() + + # 根据 exact_match 参数决定使用精确比较还是模糊包含检查 + is_match = (compare_str == target_name) if exact_match else (compare_str in target_name) + + # 如果匹配，则将组件信息添加到结果列表 + if is_match: + results.append( + { + "name": info.name, + "component_type": info.component_type.value, + "plugin_name": info.plugin_name, + "description": info.description, + "enabled": info.enabled, + } + ) + return results + + +def get_component_info(name: str, component_type: ComponentType) -> ComponentInfo | None: + """ + 获取任何一个已注册组件的详细信息对象。 + + Args: + name (str): 组件的唯一名称。 + component_type (ComponentType): 组件的类型。 + + Returns: + ComponentInfo | None: 包含组件完整信息的 ComponentInfo 对象，如果找不到则返回 None。 + """ + return component_registry.get_component_info(name, component_type) + + +# -------------------------------------------------------------------------------- +# Section 2: 状态查询与统计 (State Querying & Statistics) +# -------------------------------------------------------------------------------- +# 这部分 API 提供状态查询和统计功能。 + + +def get_plugin_state(plugin_name: str) -> dict[str, Any] | None: + """ + 获取插件的详细状态信息。 + + Args: + plugin_name (str): 要查询的插件名称。 + + Returns: + dict | None: 包含插件状态信息的字典，如果插件不存在则返回 None。 + """ + plugin_instance = plugin_manager.get_plugin_instance(plugin_name) + plugin_info = component_registry.get_plugin_info(plugin_name) + + if not plugin_info: + return None + + is_loaded = plugin_name in plugin_manager.list_loaded_plugins() + is_enabled = plugin_instance.enable_plugin if plugin_instance else False + + # 统计组件状态 + total_components = len(plugin_info.components) + enabled_components = sum(1 for c in plugin_info.components if c.enabled) + + return { + "name": plugin_name, + "is_loaded": is_loaded, + "is_enabled": is_enabled, + "total_components": total_components, + "enabled_components": enabled_components, + "disabled_components": total_components - enabled_components, + } + + +def get_all_plugin_states() -> dict[str, dict[str, Any]]: + """ + 获取所有已加载插件的状态信息。 + + Returns: + dict: 插件名称到状态信息的映射。 + """ + result = {} + for plugin_name in plugin_manager.list_loaded_plugins(): + state = get_plugin_state(plugin_name) + if state: + result[plugin_name] = state + return result + + +def get_state_statistics() -> dict[str, Any]: + """ + 获取整体状态统计信息。 + + Returns: + dict: 包含插件和组件状态统计的字典。 + """ + loaded_plugins = plugin_manager.list_loaded_plugins() + registered_plugins = plugin_manager.list_registered_plugins() + failed_plugins = list(plugin_manager.failed_plugins.keys()) + + # 统计启用/禁用的插件数量 + enabled_plugins = sum(1 for name in loaded_plugins if is_plugin_enabled(name)) + + # 统计各类型组件数量 + component_stats = {} + total_enabled = 0 + total_disabled = 0 + + for comp_type in ComponentType: + all_components = component_registry.get_components_by_type(comp_type) + enabled_count = sum(1 for info in all_components.values() if info.enabled) + disabled_count = len(all_components) - enabled_count + + component_stats[comp_type.value] = { + "total": len(all_components), + "enabled": enabled_count, + "disabled": disabled_count, + } + total_enabled += enabled_count + total_disabled += disabled_count + + return { + "plugins": { + "loaded": len(loaded_plugins), + "registered": len(registered_plugins), + "failed": len(failed_plugins), + "enabled": enabled_plugins, + "disabled": len(loaded_plugins) - enabled_plugins, + }, + "components": { + "total": total_enabled + total_disabled, + "enabled": total_enabled, + "disabled": total_disabled, + "by_type": component_stats, + }, + } + + +# -------------------------------------------------------------------------------- +# Section 3: 工具函数 (Utility Functions) +# -------------------------------------------------------------------------------- +# 这部分提供了一些轻量级的辅助函数，用于快速检查状态。 + + +def is_plugin_loaded(plugin_name: str) -> bool: + """ + 快速检查一个插件当前是否已成功加载。 + + 这是一个比 `get_plugin_details` 更轻量级的检查方法，适用于需要快速布尔值判断的场景。 + + Args: + plugin_name (str): 要检查的插件名称。 + + Returns: + bool: 如果插件已加载，则为 True，否则为 False。 + """ + return plugin_name in plugin_manager.list_loaded_plugins() + + +def is_plugin_enabled(plugin_name: str) -> bool: + """ + 检查插件是否处于启用状态。 + + Args: + plugin_name (str): 要检查的插件名称。 + + Returns: + bool: 如果插件已启用，则为 True；如果未加载或已禁用，则为 False。 + """ + plugin_instance = plugin_manager.get_plugin_instance(plugin_name) + if not plugin_instance: + return False + return plugin_instance.enable_plugin + + +def get_component_plugin(component_name: str, component_type: ComponentType) -> str | None: + """ + 查找一个特定组件属于哪个插件。 + + 在调试或管理组件时，此函数能够方便地追溯其定义的源头。 + + Args: + component_name (str): 组件的名称。 + component_type (ComponentType): 组件的类型。 + + Returns: + str | None: 组件所属的插件名称，如果找不到组件则返回 None。 + """ + component_info = component_registry.get_component_info(component_name, component_type) + return component_info.plugin_name if component_info else None + + +def validate_component_exists(name: str, component_type: ComponentType) -> bool: + """ + 验证组件是否存在于注册表中。 + + Args: + name (str): 组件名称。 + component_type (ComponentType): 组件类型。 + + Returns: + bool: 如果组件存在，则为 True。 + """ + return component_registry.get_component_info(name, component_type) is not None + + +def get_plugin_component_summary(plugin_name: str) -> dict[str, Any] | None: + """ + 获取插件的组件摘要信息。 + + Args: + plugin_name (str): 插件名称。 + + Returns: + dict | None: 包含组件摘要的字典，如果插件不存在则返回 None。 + """ + plugin_info = component_registry.get_plugin_info(plugin_name) + if not plugin_info: + return None + + # 按类型统计组件 + by_type = {} + for comp_type in ComponentType: + components = [c for c in plugin_info.components if c.component_type == comp_type] + if components: + enabled = sum(1 for c in components if c.enabled) + by_type[comp_type.value] = { + "total": len(components), + "enabled": enabled, + "disabled": len(components) - enabled, + "names": [c.name for c in components], + } + + return { + "plugin_name": plugin_name, + "total_components": len(plugin_info.components), + "by_type": by_type, + } diff --git a/src/plugin_system/apis/plugin_manage_api.py b/src/plugin_system/apis/plugin_manage_api.py index 30099fca7..ad50ad029 100644 --- a/src/plugin_system/apis/plugin_manage_api.py +++ b/src/plugin_system/apis/plugin_manage_api.py @@ -2,25 +2,21 @@ Plugin Manage API ================= -该模块提供了用于管理插件和组件生命周期、状态和信息查询的核心API。 -功能包括插件的加载、重载、注册、扫描，组件的启用/禁用，以及系统状态报告的生成。 +该模块提供了用于管理插件生命周期的核心API。 +功能包括插件的加载、重载、注册、扫描，以及插件的启用/禁用和卸载。 主要功能包括： - 插件生命周期管理（加载、重载、注册、发现） - 插件的启用/禁用 -- 组件的全局和局部启用/禁用 -- 批量组件状态管理 - 插件卸载 -- 状态查询与验证 -- 信息查询与报告 + +组件状态管理相关功能请使用 component_state_api +信息查询和报告相关功能请使用 plugin_info_api """ import os -from typing import Any, Literal from src.common.logger import get_logger -from src.plugin_system.base.component_types import ComponentType -from src.plugin_system.core.component_registry import ComponentInfo, component_registry from src.plugin_system.core.plugin_manager import plugin_manager # 初始化日志记录器 @@ -234,16 +230,14 @@ async def enable_plugin(plugin_name: str) -> bool: return True -async def disable_plugin(plugin_name: str, disable_components: bool = True) -> bool: +async def disable_plugin(plugin_name: str,) -> bool: """ 禁用一个插件。 禁用插件不会卸载它，只会标记为禁用状态。 - 可选择是否同时禁用该插件下的所有组件。 Args: plugin_name (str): 要禁用的插件名称。 - disable_components (bool): 是否同时禁用该插件下的所有组件。默认为 True。 Returns: bool: 如果插件成功禁用，则为 True。 @@ -254,10 +248,6 @@ async def disable_plugin(plugin_name: str, disable_components: bool = True) -> b logger.warning(f"插件 '{plugin_name}' 未加载，无需禁用。") return True - # 如果需要禁用组件 - if disable_components: - await disable_all_plugin_components(plugin_name) - # 设置插件为禁用状态 plugin_instance.enable_plugin = False logger.info(f"插件 '{plugin_name}' 已禁用。") @@ -299,677 +289,10 @@ def is_plugin_enabled(plugin_name: str) -> bool: return plugin_instance.enable_plugin -def get_plugin_state(plugin_name: str) -> dict[str, Any] | None: - """ - 获取插件的详细状态信息。 - - Args: - plugin_name (str): 要查询的插件名称。 - - Returns: - dict | None: 包含插件状态信息的字典，如果插件不存在则返回 None。 - """ - plugin_instance = plugin_manager.get_plugin_instance(plugin_name) - plugin_info = component_registry.get_plugin_info(plugin_name) - - if not plugin_info: - return None - - is_loaded = plugin_name in plugin_manager.list_loaded_plugins() - is_enabled = plugin_instance.enable_plugin if plugin_instance else False - - # 统计组件状态 - total_components = len(plugin_info.components) - enabled_components = sum(1 for c in plugin_info.components if c.enabled) - - return { - "name": plugin_name, - "is_loaded": is_loaded, - "is_enabled": is_enabled, - "total_components": total_components, - "enabled_components": enabled_components, - "disabled_components": total_components - enabled_components, - } - - -# -------------------------------------------------------------------------------- -# Section 3: 组件状态管理 (Component State Management) -# -------------------------------------------------------------------------------- -# 这部分 API 负责控制单个组件的启用和禁用状态，支持全局和局部（临时）范围。 - - -async def set_component_enabled(name: str, component_type: ComponentType, enabled: bool) -> bool: - """ - 在全局范围内启用或禁用一个组件。 - - 此更改会直接修改组件在注册表中的状态，但此状态是临时的，不会持久化到配置文件中。 - 包含一个保护机制，防止禁用最后一个已启用的 Chatter 组件。 - - Args: - name (str): 要操作的组件的名称。 - component_type (ComponentType): 组件的类型。 - enabled (bool): True 表示启用, False 表示禁用。 - - Returns: - bool: 如果操作成功，则为 True。 - """ - # 特殊保护：确保系统中至少有一个 Chatter 组件处于启用状态 - if component_type == ComponentType.CHATTER and not enabled: - enabled_chatters = component_registry.get_enabled_components_by_type(ComponentType.CHATTER) - if len(enabled_chatters) <= 1 and name in enabled_chatters: - logger.warning(f"操作被阻止：不能禁用最后一个启用的 Chatter 组件 ('{name}')。") - return False - - # 根据 enabled 参数调用相应的注册表方法 - if enabled: - return component_registry.enable_component(name, component_type) - else: - return await component_registry.disable_component(name, component_type) - - -def set_component_enabled_local(stream_id: str, name: str, component_type: ComponentType, enabled: bool) -> bool: - """ - 在一个特定的 stream_id 上下文中临时启用或禁用组件。 - - 此状态仅存在于内存中，并且只对指定的 stream_id 有效，不影响全局组件状态。 - 同样包含对 Chatter 组件的保护机制。 - - Args: - stream_id (str): 唯一的上下文标识符，例如一个会话ID。 - name (str): 组件名称。 - component_type (ComponentType): 组件类型。 - enabled (bool): True 为启用, False 为禁用。 - - Returns: - bool: 如果操作成功，则为 True。 - """ - # 首先，验证组件是否存在 - component_info = component_registry.get_component_info(name, component_type) - if not component_info: - logger.error(f"尝试设置局部状态失败：未找到组件 {name} ({component_type.value})。") - return False - - # Chatter 唯一性保护（在 stream_id 上下文中） - if component_type == ComponentType.CHATTER and not enabled: - enabled_chatters = component_registry.get_enabled_components_by_type(ComponentType.CHATTER, stream_id=stream_id) - if len(enabled_chatters) <= 1 and name in enabled_chatters: - logger.warning(f"操作被阻止：在 stream '{stream_id}' 中，不能禁用最后一个启用的 Chatter 组件 ('{name}')。") - return False - - # 设置局部状态 - component_registry.set_local_component_state(stream_id, name, component_type, enabled) - logger.info(f"在 stream '{stream_id}' 中，组件 {name} ({component_type.value}) 的局部状态已设置为: {enabled}") - return True - - -def clear_local_component_states(stream_id: str) -> None: - """ - 清除指定会话的所有局部组件状态。 - - 当会话结束时应调用此方法来清理资源。 - - Args: - stream_id (str): 要清除状态的会话ID。 - """ - component_registry.clear_local_component_states(stream_id) - logger.debug(f"已清除 stream '{stream_id}' 的所有局部组件状态。") - - -def is_component_enabled(name: str, component_type: ComponentType, stream_id: str | None = None) -> bool: - """ - 检查组件是否在指定上下文中启用。 - - Args: - name (str): 组件名称。 - component_type (ComponentType): 组件类型。 - stream_id (str | None): 可选的会话ID，用于检查局部状态。 - - Returns: - bool: 如果组件启用，则为 True。 - """ - return component_registry.is_component_available(name, component_type, stream_id) - - -def get_component_state(name: str, component_type: ComponentType) -> dict[str, Any] | None: - """ - 获取组件的详细状态信息。 - - Args: - name (str): 组件名称。 - component_type (ComponentType): 组件类型。 - - Returns: - dict | None: 包含组件状态信息的字典，如果组件不存在则返回 None。 - """ - component_info = component_registry.get_component_info(name, component_type) - if not component_info: - return None - - return { - "name": component_info.name, - "component_type": component_info.component_type.value, - "plugin_name": component_info.plugin_name, - "enabled": component_info.enabled, - "description": component_info.description, - } - - -# -------------------------------------------------------------------------------- -# Section 4: 批量组件状态管理 (Batch Component State Management) -# -------------------------------------------------------------------------------- -# 这部分 API 提供批量操作组件状态的功能。 - - -async def enable_all_plugin_components(plugin_name: str) -> dict[str, bool]: - """ - 启用指定插件下的所有组件。 - - Args: - plugin_name (str): 插件名称。 - - Returns: - dict[str, bool]: 每个组件名称及其启用操作是否成功的字典。 - """ - plugin_info = component_registry.get_plugin_info(plugin_name) - if not plugin_info: - logger.error(f"未找到插件 '{plugin_name}'，无法启用其组件。") - return {} - - results = {} - for component_info in plugin_info.components: - success = component_registry.enable_component(component_info.name, component_info.component_type) - results[component_info.name] = success - if success: - logger.debug(f"已启用组件: {component_info.name} ({component_info.component_type.value})") - else: - logger.warning(f"启用组件失败: {component_info.name} ({component_info.component_type.value})") - - logger.info(f"已完成启用插件 '{plugin_name}' 的所有组件，成功: {sum(results.values())}/{len(results)}") - return results - - -async def disable_all_plugin_components(plugin_name: str) -> dict[str, bool]: - """ - 禁用指定插件下的所有组件。 - - 包含对 Chatter 组件的保护机制。 - - Args: - plugin_name (str): 插件名称。 - - Returns: - dict[str, bool]: 每个组件名称及其禁用操作是否成功的字典。 - """ - plugin_info = component_registry.get_plugin_info(plugin_name) - if not plugin_info: - logger.error(f"未找到插件 '{plugin_name}'，无法禁用其组件。") - return {} - - results = {} - for component_info in plugin_info.components: - # Chatter 保护检查 - if component_info.component_type == ComponentType.CHATTER: - enabled_chatters = component_registry.get_enabled_components_by_type(ComponentType.CHATTER) - if len(enabled_chatters) <= 1 and component_info.name in enabled_chatters: - logger.warning( - f"跳过禁用最后一个 Chatter 组件 '{component_info.name}'，系统至少需要一个启用的 Chatter。" - ) - results[component_info.name] = False - continue - - success = await component_registry.disable_component(component_info.name, component_info.component_type) - results[component_info.name] = success - if success: - logger.debug(f"已禁用组件: {component_info.name} ({component_info.component_type.value})") - else: - logger.warning(f"禁用组件失败: {component_info.name} ({component_info.component_type.value})") - - logger.info(f"已完成禁用插件 '{plugin_name}' 的所有组件，成功: {sum(results.values())}/{len(results)}") - return results - - -async def set_components_enabled_by_type( - plugin_name: str, component_type: ComponentType, enabled: bool -) -> dict[str, bool]: - """ - 启用或禁用指定插件下特定类型的所有组件。 - - Args: - plugin_name (str): 插件名称。 - component_type (ComponentType): 要操作的组件类型。 - enabled (bool): True 为启用，False 为禁用。 - - Returns: - dict[str, bool]: 每个组件名称及其操作是否成功的字典。 - """ - plugin_info = component_registry.get_plugin_info(plugin_name) - if not plugin_info: - logger.error(f"未找到插件 '{plugin_name}'。") - return {} - - results = {} - for component_info in plugin_info.components: - if component_info.component_type != component_type: - continue - - success = await set_component_enabled(component_info.name, component_type, enabled) - results[component_info.name] = success - - action = "启用" if enabled else "禁用" - logger.info( - f"已完成{action}插件 '{plugin_name}' 的所有 {component_type.value} 组件，" - f"成功: {sum(results.values())}/{len(results)}" - ) - return results - - -async def batch_set_components_enabled(components: list[tuple[str, ComponentType]], enabled: bool) -> dict[str, bool]: - """ - 批量启用或禁用多个组件。 - - Args: - components (list[tuple[str, ComponentType]]): 要操作的组件列表， - 每个元素为 (组件名称, 组件类型) 元组。 - enabled (bool): True 为启用，False 为禁用。 - - Returns: - dict[str, bool]: 每个组件名称及其操作是否成功的字典。 - """ - results = {} - for name, component_type in components: - success = await set_component_enabled(name, component_type, enabled) - results[name] = success - - action = "启用" if enabled else "禁用" - logger.info(f"批量{action}操作完成，成功: {sum(results.values())}/{len(results)}") - return results - - -# -------------------------------------------------------------------------------- -# Section 5: 信息查询与报告 (Information Querying & Reporting) -# -------------------------------------------------------------------------------- -# 这部分 API 用于获取关于插件和组件的详细信息、列表和统计数据。 - - -def get_system_report() -> dict[str, Any]: - """ - 生成一份详细的系统状态报告。 - - 报告包含已加载插件、失败插件和组件的全面信息，是调试和监控系统状态的核心工具。 - - Returns: - dict[str, Any]: 包含系统、插件和组件状态的详细报告字典。 - """ - loaded_plugins_info = {} - # 遍历所有已加载的插件实例 - for name, instance in plugin_manager.loaded_plugins.items(): - plugin_info = component_registry.get_plugin_info(name) - if not plugin_info: - continue - - # 收集该插件下所有组件的详细信息 - components_details = [ - { - "name": comp_info.name, - "component_type": comp_info.component_type.value, - "description": comp_info.description, - "enabled": comp_info.enabled, - } - for comp_info in plugin_info.components - ] - - # 构建单个插件的信息字典 - # 元数据从 PluginInfo 获取，而启用状态(enable_plugin)从插件实例获取 - loaded_plugins_info[name] = { - "display_name": plugin_info.display_name or name, - "version": plugin_info.version, - "author": plugin_info.author, - "enabled": instance.enable_plugin, - "components": components_details, - } - - # 构建最终的完整报告 - report = { - "system_info": { - "loaded_plugins_count": len(plugin_manager.loaded_plugins), - "total_components_count": component_registry.get_registry_stats().get("total_components", 0), - }, - "plugins": loaded_plugins_info, - "failed_plugins": plugin_manager.failed_plugins, - } - return report - - -def get_plugin_details(plugin_name: str) -> dict[str, Any] | None: - """ - 获取单个插件的详细报告。 - - 报告内容包括插件的元数据、所有组件的详细信息及其当前状态。 - 这是 `get_system_report` 的单插件聚焦版本。 - - Args: - plugin_name (str): 要查询的插件名称。 - - Returns: - dict | None: 包含插件详细信息的字典，如果插件未注册则返回 None。 - """ - plugin_info = component_registry.get_plugin_info(plugin_name) - if not plugin_info: - logger.warning(f"尝试获取插件详情失败：未找到名为 '{plugin_name}' 的插件。") - return None - - # 收集该插件下所有组件的信息 - components_details = [ - { - "name": comp_info.name, - "component_type": comp_info.component_type.value, - "description": comp_info.description, - "enabled": comp_info.enabled, - } - for comp_info in plugin_info.components - ] - - # 获取插件实例以检查其启用状态 - plugin_instance = plugin_manager.get_plugin_instance(plugin_name) - is_enabled = plugin_instance.enable_plugin if plugin_instance else False - - # 组装详细信息字典 - return { - "name": plugin_info.name, - "display_name": plugin_info.display_name or plugin_info.name, - "version": plugin_info.version, - "author": plugin_info.author, - "license": plugin_info.license, - "description": plugin_info.description, - "enabled": is_enabled, - "status": "loaded" if is_plugin_loaded(plugin_name) else "registered", - "components": components_details, - } - - -def list_plugins(status: Literal["loaded", "registered", "failed"]) -> list[str]: - """ - 根据指定的状态列出插件名称列表。 - - 提供了一种快速、便捷的方式来监控和调试插件系统，而无需解析完整的系统报告。 - - Args: - status (str): 插件状态，可选值为 'loaded', 'registered', 'failed'。 - - Returns: - list[str]: 对应状态的插件名称列表。 - - Raises: - ValueError: 如果传入了无效的状态字符串。 - """ - if status == "loaded": - # 返回所有当前已成功加载的插件 - return plugin_manager.list_loaded_plugins() - if status == "registered": - # 返回所有已注册（但不一定已加载）的插件 - return plugin_manager.list_registered_plugins() - if status == "failed": - # 返回所有加载失败的插件的名称 - return list(plugin_manager.failed_plugins.keys()) - # 如果状态无效，则引发错误 - raise ValueError(f"无效的插件状态: '{status}'。有效选项为 'loaded', 'registered', 'failed'。") - - -def list_components(component_type: ComponentType, enabled_only: bool = True) -> list[dict[str, Any]]: - """ - 列出指定类型的所有组件的详细信息。 - - 这是查找和管理组件的核心功能，例如，获取所有可用的工具或所有注册的聊天器。 - - Args: - component_type (ComponentType): 要查询的组件类型。 - enabled_only (bool, optional): 是否只返回已启用的组件。默认为 True。 - - Returns: - list[dict[str, Any]]: 一个包含组件信息字典的列表。 - """ - # 根据 enabled_only 参数决定是获取所有组件还是仅获取已启用的组件 - if enabled_only: - components = component_registry.get_enabled_components_by_type(component_type) - else: - components = component_registry.get_components_by_type(component_type) - - # 将组件信息格式化为字典列表 - return [ - { - "name": info.name, - "plugin_name": info.plugin_name, - "description": info.description, - "enabled": info.enabled, - } - for info in components.values() - ] - - -def search_components_by_name( - name_keyword: str, - component_type: ComponentType | None = None, - case_sensitive: bool = False, - exact_match: bool = False, -) -> list[dict[str, Any]]: - """ - 根据名称关键字搜索组件，支持模糊匹配和精确匹配。 - - 极大地增强了组件的可发现性，用户无需知道完整名称即可找到所需组件。 - - Args: - name_keyword (str): 用于搜索的名称关键字。 - component_type (ComponentType | None, optional): 如果提供，则只在该类型中搜索。默认为 None (搜索所有类型)。 - case_sensitive (bool, optional): 是否进行大小写敏感的搜索。默认为 False。 - exact_match (bool, optional): 是否进行精确匹配。默认为 False (模糊匹配)。 - - Returns: - list[dict[str, Any]]: 匹配的组件信息字典的列表。 - """ - results = [] - # 如果未指定组件类型，则搜索所有类型 - types_to_search = [component_type] if component_type else list(ComponentType) - - # 根据是否大小写敏感，预处理搜索关键字 - compare_str = name_keyword if case_sensitive else name_keyword.lower() - - # 遍历要搜索的组件类型 - for comp_type in types_to_search: - all_components = component_registry.get_components_by_type(comp_type) - for name, info in all_components.items(): - # 同样地，预处理组件名称 - target_name = name if case_sensitive else name.lower() - - # 根据 exact_match 参数决定使用精确比较还是模糊包含检查 - is_match = (compare_str == target_name) if exact_match else (compare_str in target_name) - - # 如果匹配，则将组件信息添加到结果列表 - if is_match: - results.append( - { - "name": info.name, - "component_type": info.component_type.value, - "plugin_name": info.plugin_name, - "description": info.description, - "enabled": info.enabled, - } - ) - return results - - -def get_component_info(name: str, component_type: ComponentType) -> ComponentInfo | None: - """ - 获取任何一个已注册组件的详细信息对象。 - - Args: - name (str): 组件的唯一名称。 - component_type (ComponentType): 组件的类型。 - - Returns: - ComponentInfo | None: 包含组件完整信息的 ComponentInfo 对象，如果找不到则返回 None。 - """ - return component_registry.get_component_info(name, component_type) - - -def get_component_count(component_type: ComponentType, stream_id: str | None = None) -> int: - """ - 获取指定类型的已加载并启用的组件的总数。 - - 可以根据 `stream_id` 考虑局部状态，从而获得特定上下文中的组件数量。 - - Args: - component_type (ComponentType): 要查询的组件类型。 - stream_id (str | None): 可选的上下文ID。如果提供，将计入局部状态。 - - Returns: - int: 该类型下已启用的组件的数量。 - """ - return len(component_registry.get_enabled_components_by_type(component_type, stream_id=stream_id)) - - -# -------------------------------------------------------------------------------- -# Section 6: 状态查询与统计 (State Querying & Statistics) -# -------------------------------------------------------------------------------- -# 这部分 API 提供状态查询和统计功能。 - - -def get_all_plugin_states() -> dict[str, dict[str, Any]]: - """ - 获取所有已加载插件的状态信息。 - - Returns: - dict: 插件名称到状态信息的映射。 - """ - result = {} - for plugin_name in plugin_manager.list_loaded_plugins(): - state = get_plugin_state(plugin_name) - if state: - result[plugin_name] = state - return result - - -def get_components_by_state( - component_type: ComponentType | None = None, - enabled: bool | None = None, - plugin_name: str | None = None, -) -> list[ComponentInfo]: - """ - 根据条件筛选组件。 - - Args: - component_type (ComponentType | None): 按组件类型筛选。 - enabled (bool | None): 按启用状态筛选。 - plugin_name (str | None): 按插件名称筛选。 - - Returns: - list[ComponentInfo]: 符合条件的组件信息列表。 - """ - results = [] - - # 确定要搜索的组件类型 - types_to_search = [component_type] if component_type else list(ComponentType) - - for comp_type in types_to_search: - components = component_registry.get_components_by_type(comp_type) - for info in components.values(): - # 按启用状态筛选 - if enabled is not None and info.enabled != enabled: - continue - # 按插件名称筛选 - if plugin_name is not None and info.plugin_name != plugin_name: - continue - results.append(info) - - return results - - -def get_disabled_components(plugin_name: str | None = None) -> list[ComponentInfo]: - """ - 获取所有被禁用的组件。 - - Args: - plugin_name (str | None): 可选，仅获取指定插件的禁用组件。 - - Returns: - list[ComponentInfo]: 禁用组件的信息列表。 - """ - return get_components_by_state(enabled=False, plugin_name=plugin_name) - - -def get_enabled_components(plugin_name: str | None = None) -> list[ComponentInfo]: - """ - 获取所有已启用的组件。 - - Args: - plugin_name (str | None): 可选，仅获取指定插件的启用组件。 - - Returns: - list[ComponentInfo]: 启用组件的信息列表。 - """ - return get_components_by_state(enabled=True, plugin_name=plugin_name) - - -def get_state_statistics() -> dict[str, Any]: - """ - 获取整体状态统计信息。 - - Returns: - dict: 包含插件和组件状态统计的字典。 - """ - loaded_plugins = plugin_manager.list_loaded_plugins() - registered_plugins = plugin_manager.list_registered_plugins() - failed_plugins = list(plugin_manager.failed_plugins.keys()) - - # 统计启用/禁用的插件数量 - enabled_plugins = sum(1 for name in loaded_plugins if is_plugin_enabled(name)) - - # 统计各类型组件数量 - component_stats = {} - total_enabled = 0 - total_disabled = 0 - - for comp_type in ComponentType: - all_components = component_registry.get_components_by_type(comp_type) - enabled_count = sum(1 for info in all_components.values() if info.enabled) - disabled_count = len(all_components) - enabled_count - - component_stats[comp_type.value] = { - "total": len(all_components), - "enabled": enabled_count, - "disabled": disabled_count, - } - total_enabled += enabled_count - total_disabled += disabled_count - - return { - "plugins": { - "loaded": len(loaded_plugins), - "registered": len(registered_plugins), - "failed": len(failed_plugins), - "enabled": enabled_plugins, - "disabled": len(loaded_plugins) - enabled_plugins, - }, - "components": { - "total": total_enabled + total_disabled, - "enabled": total_enabled, - "disabled": total_disabled, - "by_type": component_stats, - }, - } - - -# -------------------------------------------------------------------------------- -# Section 7: 工具函数 (Utility Functions) -# -------------------------------------------------------------------------------- -# 这部分提供了一些轻量级的辅助函数，用于快速检查状态。 - - def is_plugin_loaded(plugin_name: str) -> bool: """ 快速检查一个插件当前是否已成功加载。 - 这是一个比 `get_plugin_details` 更轻量级的检查方法，适用于需要快速布尔值判断的场景。 - Args: plugin_name (str): 要检查的插件名称。 @@ -979,66 +302,45 @@ def is_plugin_loaded(plugin_name: str) -> bool: return plugin_name in plugin_manager.list_loaded_plugins() -def get_component_plugin(component_name: str, component_type: ComponentType) -> str | None: +def list_loaded_plugins() -> list[str]: """ - 查找一个特定组件属于哪个插件。 - - 在调试或管理组件时，此函数能够方便地追溯其定义的源头。 - - Args: - component_name (str): 组件的名称。 - component_type (ComponentType): 组件的类型。 + 列出所有已加载的插件名称。 Returns: - str | None: 组件所属的插件名称，如果找不到组件则返回 None。 + list[str]: 已加载插件的名称列表。 """ - component_info = component_registry.get_component_info(component_name, component_type) - return component_info.plugin_name if component_info else None + return plugin_manager.list_loaded_plugins() -def validate_component_exists(name: str, component_type: ComponentType) -> bool: +def list_registered_plugins() -> list[str]: """ - 验证组件是否存在于注册表中。 - - Args: - name (str): 组件名称。 - component_type (ComponentType): 组件类型。 + 列出所有已注册的插件名称。 Returns: - bool: 如果组件存在，则为 True。 + list[str]: 已注册插件的名称列表。 """ - return component_registry.get_component_info(name, component_type) is not None + return plugin_manager.list_registered_plugins() -def get_plugin_component_summary(plugin_name: str) -> dict[str, Any] | None: +def list_failed_plugins() -> dict[str, str]: """ - 获取插件的组件摘要信息。 + 获取所有加载失败的插件及其错误信息。 + + Returns: + dict[str, str]: 插件名称到错误信息的映射。 + """ + return plugin_manager.failed_plugins.copy() + + +def get_plugin_instance(plugin_name: str): + """ + 获取插件实例。 Args: plugin_name (str): 插件名称。 Returns: - dict | None: 包含组件摘要的字典，如果插件不存在则返回 None。 + BasePlugin | None: 插件实例，如果不存在则返回 None。 """ - plugin_info = component_registry.get_plugin_info(plugin_name) - if not plugin_info: - return None + return plugin_manager.get_plugin_instance(plugin_name) - # 按类型统计组件 - by_type = {} - for comp_type in ComponentType: - components = [c for c in plugin_info.components if c.component_type == comp_type] - if components: - enabled = sum(1 for c in components if c.enabled) - by_type[comp_type.value] = { - "total": len(components), - "enabled": enabled, - "disabled": len(components) - enabled, - "names": [c.name for c in components], - } - - return { - "plugin_name": plugin_name, - "total_components": len(plugin_info.components), - "by_type": by_type, - } diff --git a/src/plugin_system/core/component_registry.py b/src/plugin_system/core/component_registry.py index 4a6e3f8f3..2218a4fb1 100644 --- a/src/plugin_system/core/component_registry.py +++ b/src/plugin_system/core/component_registry.py @@ -32,7 +32,6 @@ from src.plugin_system.base.component_types import ( PluginInfo, PlusCommandInfo, PromptInfo, - RouterInfo, ToolInfo, ) from src.plugin_system.base.plus_command import PlusCommand, create_legacy_command_adapter @@ -161,16 +160,10 @@ class ComponentRegistry: self._mcp_tools: list[Any] = [] # 存储 MCP 工具适配器实例 self._mcp_tools_loaded = False # 标记 MCP 工具是否已加载 - # --- 局部状态管理 --- - # 局部组件状态管理器，用于在特定会话中临时覆盖全局状态 - # 结构: {stream_id: {(component_name, component_type): enabled}} - self._local_component_states: dict[str, dict[tuple[str, ComponentType], bool]] = {} - # 定义不支持局部状态管理的组件类型集合 - self._no_local_state_types: set[ComponentType] = { - ComponentType.ROUTER, # 路由组件需要全局一致性 - ComponentType.EVENT_HANDLER, # 事件处理器需要全局一致性 - ComponentType.PROMPT, # 提示词组件需要全局一致性 - } + # --- 状态管理器 --- + # 延迟导入以避免循环依赖 + from src.plugin_system.core.component_state_manager import ComponentStateManager + self._state_manager = ComponentStateManager(self) logger.info("组件注册中心初始化完成") @@ -598,6 +591,7 @@ class ComponentRegistry: # ================================================================= # == 组件状态管理 (Component State Management) + # == 委托给 ComponentStateManager 处理 # ================================================================= def enable_component(self, component_name: str, component_type: ComponentType) -> bool: @@ -611,43 +605,7 @@ class ComponentRegistry: Returns: 启用成功返回 True，失败返回 False """ - target_class = self.get_component_class(component_name, component_type) - target_info = self.get_component_info(component_name, component_type) - if not target_class or not target_info: - logger.warning(f"组件 {component_name} 未注册，无法启用") - return False - - # 更新通用注册表中的状态 - target_info.enabled = True - namespaced_name = f"{component_type.value}.{component_name}" - self._components[namespaced_name].enabled = True - self._components_by_type[component_type][component_name].enabled = True - - # 更新特定类型的启用列表 - match component_type: - case ComponentType.ACTION: - self._default_actions[component_name] = cast(ActionInfo, target_info) - case ComponentType.TOOL: - self._llm_available_tools[component_name] = cast(type[BaseTool], target_class) - case ComponentType.EVENT_HANDLER: - self._enabled_event_handlers[component_name] = cast(type[BaseEventHandler], target_class) - # 重新注册事件处理器 - from .event_manager import event_manager - event_manager.register_event_handler( - cast(type[BaseEventHandler], target_class), - self.get_plugin_config(target_info.plugin_name) or {} - ) - case ComponentType.CHATTER: - self._enabled_chatter_registry[component_name] = cast(type[BaseChatter], target_class) - case ComponentType.INTEREST_CALCULATOR: - self._enabled_interest_calculator_registry[component_name] = cast( - type[BaseInterestCalculator], target_class - ) - case ComponentType.PROMPT: - self._enabled_prompt_registry[component_name] = cast(type[BasePrompt], target_class) - - logger.info(f"组件 {component_name} ({component_type.value}) 已全局启用") - return True + return self._state_manager.enable_component(component_name, component_type) async def disable_component(self, component_name: str, component_type: ComponentType) -> bool: """ @@ -660,46 +618,10 @@ class ComponentRegistry: Returns: 禁用成功返回 True，失败返回 False """ - target_info = self.get_component_info(component_name, component_type) - if not target_info: - logger.warning(f"组件 {component_name} 未注册，无法禁用") - return False - - # 更新通用注册表中的状态 - target_info.enabled = False - namespaced_name = f"{component_type.value}.{component_name}" - if namespaced_name in self._components: - self._components[namespaced_name].enabled = False - if component_name in self._components_by_type[component_type]: - self._components_by_type[component_type][component_name].enabled = False - - try: - # 从特定类型的启用列表中移除 - match component_type: - case ComponentType.ACTION: - self._default_actions.pop(component_name, None) - case ComponentType.TOOL: - self._llm_available_tools.pop(component_name, None) - case ComponentType.EVENT_HANDLER: - self._enabled_event_handlers.pop(component_name, None) - # 从事件管理器中取消订阅 - from .event_manager import event_manager - event_manager.remove_event_handler(component_name) - case ComponentType.CHATTER: - self._enabled_chatter_registry.pop(component_name, None) - case ComponentType.INTEREST_CALCULATOR: - self._enabled_interest_calculator_registry.pop(component_name, None) - case ComponentType.PROMPT: - self._enabled_prompt_registry.pop(component_name, None) - - logger.info(f"组件 {component_name} ({component_type.value}) 已全局禁用") - return True - except Exception as e: - logger.error(f"禁用组件时发生错误: {e}", exc_info=True) - return False + return await self._state_manager.disable_component(component_name, component_type) # ================================================================= - # == 局部状态管理 (Local State Management) - 用于会话级别控制 + # == 局部状态管理 (Local State Management) - 委托给 ComponentStateManager # ================================================================= def set_local_component_state( @@ -719,22 +641,7 @@ class ComponentRegistry: Returns: 设置成功返回 True，如果组件类型不支持局部状态则返回 False """ - # 检查组件类型是否支持局部状态 - if component_type in self._no_local_state_types: - logger.warning(f"组件类型 {component_type.value} 不支持局部状态管理") - return False - - # 初始化该会话的状态字典（如果不存在） - if stream_id not in self._local_component_states: - self._local_component_states[stream_id] = {} - - # 设置局部状态 - self._local_component_states[stream_id][(component_name, component_type)] = enabled - logger.debug( - f"已为 stream '{stream_id}' 设置局部状态: " - f"{component_name} ({component_type.value}) -> {'启用' if enabled else '禁用'}" - ) - return True + return self._state_manager.set_local_component_state(stream_id, component_name, component_type, enabled) def clear_local_component_states(self, stream_id: str) -> None: """ @@ -745,7 +652,7 @@ class ComponentRegistry: Args: stream_id: 要清除状态的会话ID """ - self._local_component_states.pop(stream_id, None) + self._state_manager.clear_local_component_states(stream_id) def is_component_available( self, component_name: str, component_type: ComponentType, stream_id: str | None = None @@ -766,24 +673,7 @@ class ComponentRegistry: Returns: 如果组件可用则返回 True """ - component_info = self.get_component_info(component_name, component_type) - - # 1. 检查组件是否存在 - if not component_info: - return False - - # 2. 不支持局部状态的类型，直接返回全局状态 - if component_type in self._no_local_state_types: - return component_info.enabled - - # 3. 如果提供了 stream_id，检查是否存在局部状态覆盖 - if stream_id and stream_id in self._local_component_states: - local_state = self._local_component_states[stream_id].get((component_name, component_type)) - if local_state is not None: - return local_state # 局部状态存在，直接返回 - - # 4. 如果没有局部状态覆盖，返回全局状态 - return component_info.enabled + return self._state_manager.is_component_available(component_name, component_type, stream_id) # ================================================================= # == 组件查询方法 (Component Query Methods) @@ -999,8 +889,19 @@ class ComponentRegistry: return self._event_handler_registry.copy() def get_enabled_event_handlers(self) -> dict[str, type[BaseEventHandler]]: - """获取所有已启用的 EventHandler 类。""" - return self._enabled_event_handlers.copy() + """ + 获取所有已启用的 EventHandler 类。 + + 会检查组件的全局启用状态。 + + Returns: + 可用的 EventHandler 名称到类的字典 + """ + return { + name: cls + for name, cls in self._event_handler_registry.items() + if self.is_component_available(name, ComponentType.EVENT_HANDLER) + } def get_registered_event_handler_info(self, handler_name: str) -> EventHandlerInfo | None: """ @@ -1055,8 +956,19 @@ class ComponentRegistry: return self._interest_calculator_registry.copy() def get_enabled_interest_calculator_registry(self) -> dict[str, type[BaseInterestCalculator]]: - """获取所有已启用的 InterestCalculator 类。""" - return self._enabled_interest_calculator_registry.copy() + """ + 获取所有已启用的 InterestCalculator 类。 + + 会检查组件的全局启用状态。 + + Returns: + 可用的 InterestCalculator 名称到类的字典 + """ + return { + name: cls + for name, cls in self._interest_calculator_registry.items() + if self.is_component_available(name, ComponentType.INTEREST_CALCULATOR) + } # --- Prompt --- def get_prompt_registry(self) -> dict[str, type[BasePrompt]]: @@ -1064,8 +976,19 @@ class ComponentRegistry: return self._prompt_registry.copy() def get_enabled_prompt_registry(self) -> dict[str, type[BasePrompt]]: - """获取所有已启用的 Prompt 类。""" - return self._enabled_prompt_registry.copy() + """ + 获取所有已启用的 Prompt 类。 + + 会检查组件的全局启用状态。 + + Returns: + 可用的 Prompt 名称到类的字典 + """ + return { + name: cls + for name, cls in self._prompt_registry.items() + if self.is_component_available(name, ComponentType.PROMPT) + } # --- Adapter --- def get_adapter_registry(self) -> dict[str, type[BaseAdapter]]: diff --git a/src/plugin_system/core/component_state_manager.py b/src/plugin_system/core/component_state_manager.py new file mode 100644 index 000000000..300764d79 --- /dev/null +++ b/src/plugin_system/core/component_state_manager.py @@ -0,0 +1,341 @@ +""" +组件状态管理器模块。 + +该模块负责管理组件的全局启用/禁用状态以及会话级别的局部（临时）状态。 +将状态管理逻辑从 ComponentRegistry 分离出来，实现职责分离。 + +主要功能: + - 全局启用/禁用组件 + - 会话级别的局部状态管理 + - 组件可用性检查（综合全局和局部状态） +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, cast + +from src.common.logger import get_logger +from src.plugin_system.base.component_types import ( + ActionInfo, + ComponentInfo, + ComponentType, +) + +if TYPE_CHECKING: + from src.plugin_system.base.base_chatter import BaseChatter + from src.plugin_system.base.base_events_handler import BaseEventHandler + from src.plugin_system.base.base_interest_calculator import BaseInterestCalculator + from src.plugin_system.base.base_prompt import BasePrompt + from src.plugin_system.base.base_tool import BaseTool + from src.plugin_system.core.component_registry import ComponentRegistry + +logger = get_logger("component_state_manager") + + +class ComponentStateManager: + """ + 组件状态管理器。 + + 该类负责管理所有组件的启用/禁用状态，包括: + - 全局状态: 影响所有会话的组件启用状态 + - 局部状态: 仅影响特定会话（stream_id）的临时状态覆盖 + + Attributes: + _registry: 组件注册中心的引用 + _local_component_states: 局部组件状态管理器 + _no_local_state_types: 不支持局部状态管理的组件类型集合 + """ + + def __init__(self, registry: ComponentRegistry): + """ + 初始化组件状态管理器。 + + Args: + registry: 组件注册中心实例的引用 + """ + self._registry = registry + + # 局部组件状态管理器 + # 结构: {stream_id: {(component_name, component_type): enabled}} + self._local_component_states: dict[str, dict[tuple[str, ComponentType], bool]] = {} + + # 定义不支持局部状态管理的组件类型集合 + # 这些组件类型需要保持全局一致性 + self._no_local_state_types: set[ComponentType] = { + ComponentType.ROUTER, # 路由组件需要全局一致性 + ComponentType.EVENT_HANDLER, # 事件处理器需要全局一致性 + ComponentType.PROMPT, # 提示词组件需要全局一致性 + ComponentType.ADAPTER, # ADAPTER组件需要全局一致性 + } + + logger.debug("组件状态管理器初始化完成") + + # ================================================================= + # == 全局状态管理 (Global State Management) + # ================================================================= + + def enable_component(self, component_name: str, component_type: ComponentType) -> bool: + """ + 全局启用一个组件。 + + Args: + component_name: 组件名称 + component_type: 组件类型 + + Returns: + 启用成功返回 True，失败返回 False + """ + if component_type is ComponentType.ADAPTER: + logger.error(f"组件 {component_name} 类型是适配器，无法启用或者禁用") + return False + target_class = self._registry.get_component_class(component_name, component_type) + target_info = self._registry.get_component_info(component_name, component_type) + if not target_class or not target_info: + logger.warning(f"组件 {component_name} 未注册，无法启用") + return False + + # 更新通用注册表中的状态 + target_info.enabled = True + namespaced_name = f"{component_type.value}.{component_name}" + self._registry._components[namespaced_name].enabled = True + self._registry._components_by_type[component_type][component_name].enabled = True + + # 更新特定类型的启用列表 + match component_type: + case ComponentType.ACTION: + self._registry._default_actions[component_name] = cast(ActionInfo, target_info) + case ComponentType.TOOL: + self._registry._llm_available_tools[component_name] = cast(type[BaseTool], target_class) + case ComponentType.EVENT_HANDLER: + self._registry._enabled_event_handlers[component_name] = cast(type[BaseEventHandler], target_class) + # 重新注册事件处理器 + from .event_manager import event_manager + event_manager.register_event_handler( + cast(type[BaseEventHandler], target_class), + self._registry.get_plugin_config(target_info.plugin_name) or {} + ) + case ComponentType.CHATTER: + self._registry._enabled_chatter_registry[component_name] = cast(type[BaseChatter], target_class) + case ComponentType.INTEREST_CALCULATOR: + self._registry._enabled_interest_calculator_registry[component_name] = cast( + type[BaseInterestCalculator], target_class + ) + case ComponentType.PROMPT: + self._registry._enabled_prompt_registry[component_name] = cast(type[BasePrompt], target_class) + case ComponentType.ADAPTER: + self._registry._enabled_adapter_registry[component_name] = cast(Any, target_class) + + logger.info(f"组件 {component_name} ({component_type.value}) 已全局启用") + return True + + async def disable_component(self, component_name: str, component_type: ComponentType) -> bool: + """ + 全局禁用一个组件。 + + Args: + component_name: 组件名称 + component_type: 组件类型 + + Returns: + 禁用成功返回 True，失败返回 False + """ + if component_type is ComponentType.ADAPTER: + logger.error(f"组件 {component_name} 类型是适配器，无法启用或者禁用") + return False + target_info = self._registry.get_component_info(component_name, component_type) + if not target_info: + logger.warning(f"组件 {component_name} 未注册，无法禁用") + return False + + # 更新通用注册表中的状态 + target_info.enabled = False + namespaced_name = f"{component_type.value}.{component_name}" + if namespaced_name in self._registry._components: + self._registry._components[namespaced_name].enabled = False + if component_name in self._registry._components_by_type[component_type]: + self._registry._components_by_type[component_type][component_name].enabled = False + + try: + # 从特定类型的启用列表中移除 + match component_type: + case ComponentType.ACTION: + self._registry._default_actions.pop(component_name, None) + case ComponentType.TOOL: + self._registry._llm_available_tools.pop(component_name, None) + case ComponentType.EVENT_HANDLER: + self._registry._enabled_event_handlers.pop(component_name, None) + # 从事件管理器中取消订阅 + from .event_manager import event_manager + event_manager.remove_event_handler(component_name) + case ComponentType.CHATTER: + self._registry._enabled_chatter_registry.pop(component_name, None) + case ComponentType.INTEREST_CALCULATOR: + self._registry._enabled_interest_calculator_registry.pop(component_name, None) + case ComponentType.PROMPT: + self._registry._enabled_prompt_registry.pop(component_name, None) + case ComponentType.ADAPTER: + self._registry._enabled_adapter_registry.pop(component_name, None) + + logger.info(f"组件 {component_name} ({component_type.value}) 已全局禁用") + return True + except Exception as e: + logger.error(f"禁用组件时发生错误: {e}", exc_info=True) + return False + + # ================================================================= + # == 局部状态管理 (Local State Management) - 用于会话级别控制 + # ================================================================= + + def set_local_component_state( + self, stream_id: str, component_name: str, component_type: ComponentType, enabled: bool + ) -> bool: + """ + 为指定的会话（stream_id）设置组件的局部（临时）状态。 + + 这允许在单个对话流中动态启用或禁用组件，而不影响全局设置。 + + Args: + stream_id: 唯一的会话ID + component_name: 组件名称 + component_type: 组件类型 + enabled: True 表示启用，False 表示禁用 + + Returns: + 设置成功返回 True，如果组件类型不支持局部状态则返回 False + """ + # 检查组件类型是否支持局部状态 + if component_type in self._no_local_state_types: + logger.warning(f"组件类型 {component_type.value} 不支持局部状态管理") + return False + + # 初始化该会话的状态字典（如果不存在） + if stream_id not in self._local_component_states: + self._local_component_states[stream_id] = {} + + # 设置局部状态 + self._local_component_states[stream_id][(component_name, component_type)] = enabled + logger.debug( + f"已为 stream '{stream_id}' 设置局部状态: " + f"{component_name} ({component_type.value}) -> {'启用' if enabled else '禁用'}" + ) + return True + + def clear_local_component_states(self, stream_id: str) -> None: + """ + 清除指定会话的所有局部状态。 + + 当会话结束时应调用此方法来清理资源。 + + Args: + stream_id: 要清除状态的会话ID + """ + self._local_component_states.pop(stream_id, None) + + def get_local_state( + self, stream_id: str, component_name: str, component_type: ComponentType + ) -> bool | None: + """ + 获取指定会话中组件的局部状态。 + + Args: + stream_id: 会话ID + component_name: 组件名称 + component_type: 组件类型 + + Returns: + 局部状态值（True/False），如果没有设置则返回 None + """ + if stream_id not in self._local_component_states: + return None + return self._local_component_states[stream_id].get((component_name, component_type)) + + # ================================================================= + # == 组件可用性检查 (Component Availability Check) + # ================================================================= + + def is_component_available( + self, component_name: str, component_type: ComponentType, stream_id: str | None = None + ) -> bool: + """ + 检查一个组件在给定上下文中是否可用。 + + 检查顺序: + 1. 组件是否存在 + 2. (如果提供了 stream_id 且组件类型支持局部状态) 是否有局部状态覆盖 + 3. 全局启用状态 + + Args: + component_name: 组件名称 + component_type: 组件类型 + stream_id: 会话ID（可选） + + Returns: + 如果组件可用则返回 True + """ + component_info = self._registry.get_component_info(component_name, component_type) + + # 1. 检查组件是否存在 + if not component_info: + return False + + # 2. 不支持局部状态的类型，直接返回全局状态 + if component_type in self._no_local_state_types: + return component_info.enabled + + # 3. 如果提供了 stream_id，检查是否存在局部状态覆盖 + if stream_id: + local_state = self.get_local_state(stream_id, component_name, component_type) + if local_state is not None: + return local_state # 局部状态存在，直接返回 + + # 4. 如果没有局部状态覆盖，返回全局状态 + return component_info.enabled + + def get_enabled_components_by_type( + self, component_type: ComponentType, stream_id: str | None = None + ) -> dict[str, ComponentInfo]: + """ + 获取指定类型的所有可用组件。 + + 这会同时考虑全局启用状态和 stream_id 对应的局部状态。 + + Args: + component_type: 要查询的组件类型 + stream_id: 会话ID，用于检查局部状态覆盖（可选） + + Returns: + 一个包含可用组件名称和信息的字典 + """ + return { + name: info + for name, info in self._registry.get_components_by_type(component_type).items() + if self.is_component_available(name, component_type, stream_id) + } + + # ================================================================= + # == 辅助方法 (Helper Methods) + # ================================================================= + + def supports_local_state(self, component_type: ComponentType) -> bool: + """ + 检查指定的组件类型是否支持局部状态管理。 + + Args: + component_type: 组件类型 + + Returns: + 如果支持局部状态则返回 True + """ + return component_type not in self._no_local_state_types + + def get_all_local_states(self) -> dict[str, dict[tuple[str, ComponentType], bool]]: + """ + 获取所有会话的局部状态（用于调试）。 + + Returns: + 所有局部状态的字典（副本） + """ + return { + stream_id: states.copy() + for stream_id, states in self._local_component_states.items() + } diff --git a/src/plugins/built_in/system_management/plugin.py b/src/plugins/built_in/system_management/plugin.py index 72bc826c6..247c459b7 100644 --- a/src/plugins/built_in/system_management/plugin.py +++ b/src/plugins/built_in/system_management/plugin.py @@ -12,6 +12,8 @@ from src.chat.utils.prompt_component_manager import prompt_component_manager from src.chat.utils.prompt_params import PromptParameters from src.plugin_system.apis import ( chat_api, + component_state_api, + plugin_info_api, plugin_manage_api, ) from src.plugin_system.apis.logging_api import get_logger @@ -434,7 +436,7 @@ class SystemCommand(PlusCommand): @require_permission("plugin.manage", deny_message="❌ 你没有权限查看插件报告") async def _show_system_report(self): """显示系统插件报告""" - report = plugin_manage_api.get_system_report() + report = plugin_info_api.get_system_report() response_parts = [ "📊 **系统插件报告**", @@ -509,7 +511,7 @@ class SystemCommand(PlusCommand): stream_id = self.message.chat_info.stream_id # 默认作用于当前会话 # 1. 搜索组件 - found_components = plugin_manage_api.search_components_by_name(comp_name, exact_match=True) + found_components = plugin_info_api.search_components_by_name(comp_name, exact_match=True) if not found_components: await self.send_text(f"❌ 未找到名为 '{comp_name}' 的组件。") @@ -563,7 +565,7 @@ class SystemCommand(PlusCommand): stream_id = target_stream.stream_id # 4. 执行操作 - success = plugin_manage_api.set_component_enabled_local( + success = component_state_api.set_component_enabled_local( stream_id=stream_id, name=comp_name, component_type=component_type,