feat(commands): support sending CommandRegister events for clients

app/chain/command.py

@@ -20,6 +20,7 @@ from app.schemas.event import CommandRegisterEventData
 from app.schemas.types import EventType, MessageChannel, ChainEventType
 from app.utils.object import ObjectUtils
 from app.utils.singleton import Singleton
+from app.utils.structures import DictUtils
 
 
 class CommandChain(ChainBase, metaclass=Singleton):
@@ -173,29 +174,40 @@ class CommandChain(ChainBase, metaclass=Singleton):
                     **self._other_commands
                 }
 
+                # 强制触发注册
+                force_register = False
                 # 触发事件允许可以拦截和调整命令
                 event, initial_commands = self.__trigger_register_commands_event()
 
-                # 如果事件返回有效的 event_data，使用事件中调整后的命令
                 if event and event.event_data:
+                    # 如果事件返回有效的 event_data，使用事件中调整后的命令
                     event_data: CommandRegisterEventData = event.event_data
+                    # 如果事件被取消，跳过命令注册
+                    if event_data.cancel:
+                        logger.debug(f"Command initialization canceled by event: {event_data.source}")
+                        return
+                    # 如果拦截源与插件标识一致时，这里认为需要强制触发注册
+                    if pid is not None and pid == event_data.source:
+                        force_register = True
                     initial_commands = event_data.commands or {}
                     logger.debug(f"Registering command count from event: {len(initial_commands)}")
                 else:
                     logger.debug(f"Registering initial command count: {len(initial_commands)}")
 
                 # initial_commands 必须是 self._commands 的子集
-                filtered_initial_commands = {
-                    cmd: details for cmd, details in initial_commands.items() if cmd in self._commands
-                }
+                filtered_initial_commands = DictUtils.filter_keys_to_subset(initial_commands, self._commands)
+                # 如果 filtered_initial_commands 为空，则跳过注册
+                if not filtered_initial_commands and not force_register:
+                    logger.debug("Filtered commands are empty, skipping registration.")
+                    return
 
                 # 对比调整后的命令与当前命令
-                if filtered_initial_commands == self._registered_commands:
-                    logger.debug("Command set unchanged, skipping broadcast registration.")
-                else:
-                    logger.debug("Command set has changed, Updating and broadcasting new commands.")
+                if filtered_initial_commands != self._registered_commands or force_register:
+                    logger.debug("Command set has changed or force registration is enabled.")
                     self._registered_commands = filtered_initial_commands
                     super().register_commands(commands=filtered_initial_commands)
+                else:
+                    logger.debug("Command set unchanged, skipping broadcast registration.")
         except Exception as e:
             logger.error(f"Error occurred during command initialization in background: {e}", exc_info=True)
 
@@ -220,13 +232,16 @@ class CommandChain(ChainBase, metaclass=Singleton):
                     command_data["pid"] = plugin_id
                 commands[cmd] = command_data
 
-        # 触发事件允许可以拦截和调整命令
-        commands = {}
+        # 初始化命令字典
+        commands: Dict[str, dict] = {}
         add_commands(self._preset_commands, "preset")
         add_commands(self._plugin_commands, "plugin")
         add_commands(self._other_commands, "other")
+
+        # 触发事件允许可以拦截和调整命令
         event_data = CommandRegisterEventData(commands=commands, origin="CommandChain", service=None)
-        return eventmanager.send_event(ChainEventType.CommandRegister, event_data), commands
+        event = eventmanager.send_event(ChainEventType.CommandRegister, event_data)
+        return event, commands
 
     def __build_plugin_commands(self, pid: Optional[str] = None) -> Dict[str, dict]:
         """

app/modules/telegram/__init__.py

@@ -1,12 +1,16 @@
+import copy
 import json
 from typing import Optional, Union, List, Tuple, Any, Dict
 
 from app.core.context import MediaInfo, Context
+from app.core.event import eventmanager
 from app.log import logger
 from app.modules import _ModuleBase, _MessageBase
 from app.modules.telegram.telegram import Telegram
 from app.schemas import MessageChannel, CommingMessage, Notification
-from app.schemas.types import ModuleType
+from app.schemas.event import CommandRegisterEventData
+from app.schemas.types import ModuleType, ChainEventType
+from app.utils.structures import DictUtils
 
 
 class TelegramModule(_ModuleBase, _MessageBase[Telegram]):
@@ -189,5 +193,41 @@ class TelegramModule(_ModuleBase, _MessageBase[Telegram]):
         注册命令，实现这个函数接收系统可用的命令菜单
         :param commands: 命令字典
         """
-        for client in self.get_instances().values():
-            client.register_commands(commands)
+        for client_config in self.get_configs().values():
+            client = self.get_instance(client_config.name)
+            if not client:
+                continue
+
+            # 触发事件，允许调整命令数据，这里需要进行深复制，避免实例共享
+            scoped_commands = copy.deepcopy(commands)
+            event = eventmanager.send_event(
+                ChainEventType.CommandRegister,
+                CommandRegisterEventData(commands=scoped_commands, origin="Telegram", service=client_config.name)
+            )
+
+            # 如果事件返回有效的 event_data，使用事件中调整后的命令
+            if event and event.event_data:
+                event_data: CommandRegisterEventData = event.event_data
+                # 如果事件被取消，跳过命令注册，并清理菜单
+                if event_data.cancel:
+                    client.delete_commands()
+                    logger.debug(
+                        f"Command registration for {client_config.name} canceled by event: {event_data.source}"
+                    )
+                    continue
+                scoped_commands = event_data.commands or {}
+                if not scoped_commands:
+                    logger.debug("Filtered commands are empty, skipping registration.")
+                    client.delete_commands()
+
+            # scoped_commands 必须是 commands 的子集
+            filtered_scoped_commands = DictUtils.filter_keys_to_subset(scoped_commands, commands)
+            # 如果 filtered_scoped_commands 为空，则跳过注册
+            if not filtered_scoped_commands:
+                logger.debug("Filtered commands are empty, skipping registration.")
+                client.delete_commands()
+                continue
+            # 对比调整后的命令与当前命令
+            if filtered_scoped_commands != commands:
+                logger.debug(f"Command set has changed, Updating new commands: {filtered_scoped_commands}")
+            client.register_commands(filtered_scoped_commands)

app/modules/wechat/__init__.py

@@ -1,14 +1,18 @@
+import copy
 import xml.dom.minidom
 from typing import Optional, Union, List, Tuple, Any, Dict
 
 from app.core.context import Context, MediaInfo
+from app.core.event import eventmanager
 from app.log import logger
 from app.modules import _ModuleBase, _MessageBase
 from app.modules.wechat.WXBizMsgCrypt3 import WXBizMsgCrypt
 from app.modules.wechat.wechat import WeChat
 from app.schemas import MessageChannel, CommingMessage, Notification
-from app.schemas.types import ModuleType
+from app.schemas.event import CommandRegisterEventData
+from app.schemas.types import ModuleType, ChainEventType
 from app.utils.dom import DomUtils
+from app.utils.structures import DictUtils
 
 
 class WechatModule(_ModuleBase, _MessageBase[WeChat]):
@@ -222,7 +226,42 @@ class WechatModule(_ModuleBase, _MessageBase[WeChat]):
             # 如果没有配置消息解密相关参数，则也没有必要进行菜单初始化
             if not client_config.config.get("WECHAT_ENCODING_AESKEY") or not client_config.config.get("WECHAT_TOKEN"):
                 logger.debug(f"{client_config.name} 缺少消息解密参数，跳过后续菜单初始化")
-            else:
-                client = self.get_instance(client_config.name)
-                if client:
-                    client.create_menus(commands)
+                continue
+
+            client = self.get_instance(client_config.name)
+            if not client:
+                continue
+
+            # 触发事件，允许调整命令数据，这里需要进行深复制，避免实例共享
+            scoped_commands = copy.deepcopy(commands)
+            event = eventmanager.send_event(
+                ChainEventType.CommandRegister,
+                CommandRegisterEventData(commands=scoped_commands, origin="WeChat", service=client_config.name)
+            )
+
+            # 如果事件返回有效的 event_data，使用事件中调整后的命令
+            if event and event.event_data:
+                event_data: CommandRegisterEventData = event.event_data
+                # 如果事件被取消，跳过命令注册，并清理菜单
+                if event_data.cancel:
+                    client.delete_menus()
+                    logger.debug(
+                        f"Command registration for {client_config.name} canceled by event: {event_data.source}"
+                    )
+                    continue
+                scoped_commands = event_data.commands or {}
+                if not scoped_commands:
+                    logger.debug("Filtered commands are empty, skipping registration.")
+                    client.delete_menus()
+
+            # scoped_commands 必须是 commands 的子集
+            filtered_scoped_commands = DictUtils.filter_keys_to_subset(scoped_commands, commands)
+            # 如果 filtered_scoped_commands 为空，则跳过注册
+            if not filtered_scoped_commands:
+                logger.debug("Filtered commands are empty, skipping registration.")
+                client.delete_menus()
+                continue
+            # 对比调整后的命令与当前命令
+            if filtered_scoped_commands != commands:
+                logger.debug(f"Command set has changed, Updating new commands: {filtered_scoped_commands}")
+            client.create_menus(filtered_scoped_commands)

app/utils/structures.py

@@ -0,0 +1,72 @@
+from typing import Dict, List, Set, TypeVar, Any, Union
+
+K = TypeVar("K")
+V = TypeVar("V")
+
+
+class DictUtils:
+    @staticmethod
+    def filter_keys_to_subset(source: Dict[K, V], reference: Dict[K, V]) -> Dict[K, V]:
+        """
+        过滤 source 字典，使其键成为 reference 字典键的子集
+
+        :param source: 要被过滤的字典
+        :param reference: 参考字典，定义允许的键
+        :return: 过滤后的字典，只包含在 reference 中存在的键
+        """
+        if not isinstance(source, dict) or not isinstance(reference, dict):
+            return {}
+
+        return {key: value for key, value in source.items() if key in reference}
+
+    @staticmethod
+    def is_keys_subset(source: Dict[K, V], reference: Dict[K, V]) -> bool:
+        """
+        判断 source 字典的键是否为 reference 字典键的子集
+
+        :param source: 要检查的字典
+        :param reference: 参考字典
+        :return: 如果 source 的键是 reference 的键子集，则返回 True，否则返回 False
+        """
+        if not isinstance(source, dict) or not isinstance(reference, dict):
+            return False
+
+        return all(key in reference for key in source)
+
+
+class ListUtils:
+    @staticmethod
+    def flatten(nested_list: Union[List[List[Any]], List[Any]]) -> List[Any]:
+        """
+        将嵌套的列表展平成单个列表
+
+        :param nested_list: 嵌套的列表
+        :return: 展平后的列表
+        """
+        if not isinstance(nested_list, list):
+            return []
+
+        # 检查是否嵌套，若不嵌套直接返回
+        if not any(isinstance(sublist, list) for sublist in nested_list):
+            return nested_list
+
+        return [item for sublist in nested_list if isinstance(sublist, list) for item in sublist]
+
+
+class SetUtils:
+    @staticmethod
+    def flatten(nested_sets: Union[Set[Set[Any]], Set[Any]]) -> Set[Any]:
+        """
+        将嵌套的集合展开为单个集合
+
+        :param nested_sets: 嵌套的集合
+        :return: 展开的集合
+        """
+        if not isinstance(nested_sets, set):
+            return set()
+
+        # 检查是否嵌套，若不嵌套直接返回
+        if not any(isinstance(subset, set) for subset in nested_sets):
+            return nested_sets
+
+        return {item for subset in nested_sets if isinstance(subset, set) for item in subset}