import logging import re from datetime import datetime from typing import Dict, Any, List, Optional from wcferry import WxMsg from plugin_common.message_plugin_interface import MessagePluginInterface from plugin_common.plugin_interface import PluginStatus from robot_cmd.robot_command import Feature, PermissionStatus, GroupBotManager class GroupAddPlugin(MessagePluginInterface): """群成员加入欢迎插件""" def __init__(self): super().__init__() self.name = "群成员加入欢迎" self.description = "当有新成员加入群聊时,自动发送欢迎消息" self.version = "1.0.0" self.author = "Trae AI" self.wcf = None self.gbm = None self.LOG = logging.getLogger(f"Plugin.{self.name}") self.enable = True def initialize(self, context: Dict[str, Any]) -> bool: """初始化插件""" self.LOG.info(f"正在初始化 {self.name} 插件...") # 保存上下文对象 self.wcf = context.get("wcf") if not self.wcf: self.LOG.error("无法获取wcf对象,插件初始化失败") return False # 获取群管理器 self.gbm = context.get("gbm") if not self.gbm: self.LOG.error("无法获取群管理器对象,插件初始化失败") return False # 从配置中获取启用状态 plugin_config = self._config.get("GroupAdd", {}) self.enable = plugin_config.get("enable", True) self.LOG.info(f"{self.name} 插件初始化完成,启用状态: {self.enable}") return True def on_receive_message(self, msg: WxMsg) -> bool: """处理接收到的消息""" if not self.enable: return False # 只处理群系统消息 if msg.type != 10000 or not msg.from_group(): return False # 检查群权限 if self.gbm.get_group_permission(msg.roomid, Feature.GROUP_ADD) == PermissionStatus.DISABLED: return False # 使用正则表达式提取双引号中的内容,判断是否为加入群聊消息 match = re.search(r'"(.*?)"', msg.content) if not match: return False # 获取当前时间 now_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') # 提取昵称并发送欢迎消息 nickname = match.group(1) welcome_message = f"🎉 欢迎 【{nickname}】 加入群聊👋 \n 🕒 {now_time} 🕒 !" self.wcf.send_text(welcome_message, msg.roomid) self.LOG.info(f"已发送欢迎消息给新成员 {nickname} 在群 {msg.roomid}") return True @property def commands(self) -> List[str]: """插件支持的命令列表""" return [] def get_help(self) -> str: """获取插件帮助信息""" return "群成员加入欢迎插件:当有新成员加入群聊时,自动发送欢迎消息。" def start(self) -> bool: """启动插件""" self.enable = True self.LOG.info(f"{self.name} 插件已启动") return True def stop(self) -> bool: """停止插件""" self.enable = False self.LOG.info(f"{self.name} 插件已停止") return True