from typing import Dict, Any, Tuple, Optional, List from base.plugin_common.plugin_interface import PluginInterface from wechat_ipad import WechatAPIClient from utils.robot_cmd.robot_command import Feature class MessagePluginInterface(PluginInterface): """消息处理插件接口""" @property def command_prefix(self) -> Optional[str]: """命令前缀,如 '/'""" return None @property def commands(self) -> List[str]: """支持的命令列表""" return [] @property def feature_key(self) -> Optional[str]: """插件对应的功能权限键名""" return None @property def feature_description(self) -> Optional[str]: """插件对应的功能权限描述""" return None def register_feature(self) -> Optional[Feature]: """注册插件功能权限 Returns: Feature: 注册的功能权限枚举,如果不需要权限则返回None """ if self.feature_key and self.feature_description: return Feature.register_feature(self.feature_key, self.feature_description) return None # 需要完成jobs 的业务,所以完成bot注入 def set_bot(self, bot: WechatAPIClient) -> None: self.bot: WechatAPIClient = bot def can_process(self, message: Dict[str, Any]) -> bool: """ 检查插件是否可以处理该消息 Args: message: 消息字典,包含消息的各种属性 Returns: 是否可以处理 """ # 默认实现:检查是否是命令 if self.command_prefix and self.commands: content = message.get("content", "") if content.startswith(self.command_prefix): command = content[len(self.command_prefix):].split()[0] return command in self.commands return False def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]: """ 处理消息 Args: message: 消息字典,包含消息的各种属性,以及发送消息所需的对象 - wcf: WcfAPI对象,可用于发送消息 - message_util: 消息工具类,提供更高级的消息处理功能 Returns: (是否已处理, 处理结果) """ raise NotImplementedError("子类必须实现此方法")