from loguru import logger import redis import re from typing import Dict, Any, List, Optional, Tuple from plugin_common.message_plugin_interface import MessagePluginInterface from plugin_common.plugin_interface import PluginStatus from utils.decorator.plugin_decorators import plugin_stats_decorator from utils.robot_cmd.robot_command import GroupBotManager from utils.wechat.contact_manager import ContactManager from wechat_ipad import WechatAPIClient class GroupAutoInvitePlugin(MessagePluginInterface): """自动加群功能插件""" @property def name(self) -> str: return "自动加群功能" @property def version(self) -> str: return "1.0.0" @property def description(self) -> str: return "提供自动加群功能，支持通过关键词邀请用户加入指定群聊" @property def author(self) -> str: return "Trae AI" @property def command_prefix(self) -> Optional[str]: return "#加群配置|" # 使用前缀来识别命令 @property def commands(self) -> List[str]: return self._commands def __init__(self): super().__init__() # Redis 中存储群组映射的前缀 self.mapping_prefix = "group:group_mapping:" self._commands = [] self.bot: WechatAPIClient = None def initialize(self, context: Dict[str, Any]) -> bool: """初始化插件""" self.LOG.info(f"正在初始化 {self.name} 插件...") # 获取群管理器 self.gbm = context.get("gbm") # 获取Redis连接池 self.redis_pool = context.get("redis_pool") # 从配置中获取命令和启用状态 plugin_config = self._config.get("GroupAutoInvite", {}) self._commands = plugin_config.get("command", ["#加群配置"]) self.command_format = plugin_config.get("command-format", "#加群配置|help") self.enable = plugin_config.get("enable", True) self.LOG.info(f"[{self.name}] 插件初始化完成，指令：{self._commands}") return True def start(self) -> bool: """启动插件""" self.LOG.info(f"[{self.name}] 插件已启动") self.status = PluginStatus.RUNNING return True def stop(self) -> bool: """停止插件""" self.LOG.info(f"[{self.name}] 插件已停止") self.status = PluginStatus.STOPPED return True def _get_redis_connection(self): """从连接池获取 Redis 连接""" return redis.Redis(connection_pool=self.redis_pool) def can_process(self, message: Dict[str, Any]) -> bool: """检查是否可以处理该消息""" if not self.enable: return False content = str(message.get("content", "")).strip() roomid = message.get("roomid", "") # 处理加群配置命令 if content.startswith("#加群配置|"): return True # 处理加群请求 - 只在私聊中处理 if re.search(r"^#加群\s+(\w+)$", content): # 如果是在群聊中发送的加群请求，不处理 if roomid: self.LOG.info(f"⚠️ 请私聊机器人发送 \"#加群 key\" 命令加入群聊") return False return True return False @plugin_stats_decorator(plugin_name="自动加群功能") async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]: """处理消息""" content = str(message.get("content", "")).strip() self.LOG.info(f"插件执行： {self.name}：{content}") sender = message.get("sender") roomid = message.get("roomid", "") gbm: GroupBotManager = message.get("gbm") self.bot: WechatAPIClient = message.get("bot") # 处理加群配置命令 if content.startswith("#加群配置|"): return await self._handle_config_command(content, sender, roomid, gbm) # 处理加群请求 match = re.search(r"^#加群\s+(\w+)$", content) if match: return await self._handle_join_request(match.group(1), sender, roomid, gbm) return False, "无法处理的消息" async def _handle_config_command(self, content: str, sender: str, roomid: str, gbm: GroupBotManager) -> Tuple[ bool, Optional[str]]: """处理配置命令""" # 检查是否为管理员 admin_list = self.gbm.get_admin_list() if sender not in admin_list: await self.bot.send_text_message((roomid if roomid else sender), "⚠️ 权限不足，只有管理员才能配置群邀请功能", sender) return True, "权限不足" # 解析命令 command = content.replace("#加群配置|", "").strip() result = self.process_command(command) # 发送结果 await self.bot.send_text_message((roomid if roomid else sender), result, sender) return True, "配置命令处理成功" async def _handle_join_request(self, key: str, sender: str, roomid: str, gbm: GroupBotManager) -> Tuple[ bool, Optional[str]]: """处理加群请求""" try: # 获取对应的群ID group_id = self.get_first_group_id(key) # 检查是否找到群ID if isinstance(group_id, str) and "没有关联的群ID" in group_id: await self.bot.send_text_message(sender, f"⚠️ 未找到关键词 '{key}' 对应的群聊") return True, "未找到群聊" # 判断是否在群里面，如果在，则不添加 con = ContactManager.get_instance() members = con.get_group_members(group_id) # 如果在群里面，则不添加 if sender in members: await self.bot.send_text_message(sender, f"⚠️ 你已经在群聊中了，无需重复添加") return True, "你已经在群聊中了" # 发送邀请 self.LOG.info(f"邀请用户 {sender} 加入群 {group_id}") result = await self.bot.invite_chatroom_member(sender, group_id) if result: await self.bot.send_text_message(sender, f"✅ 已发送邀请，请查看群聊邀请通知") return True, "邀请发送成功" else: await self.bot.send_text_message(sender, f"❌ 邀请发送失败，请稍后再试") return False, "邀请发送失败" except Exception as e: self.LOG.error(f"处理加群请求出错: {e}") await self.bot.send_text_message(sender, f"❌ 处理加群请求出错: {e}") return False, f"处理出错: {e}" def add_mapping(self, key, group_id): """添加群组ID到指定key""" try: if self._get_redis_connection().sismember(self.mapping_prefix + key, group_id): return f"群ID {group_id} 已存在于关键词 {key} 中" else: self._get_redis_connection().sadd(self.mapping_prefix + key, group_id) return f"已添加: {key} -> {group_id}" except redis.RedisError as e: return f"操作失败: {e}" def del_mapping(self, key, group_id): """删除指定key下的某个群组ID""" try: if self._get_redis_connection().sismember(self.mapping_prefix + key, group_id): self._get_redis_connection().srem(self.mapping_prefix + key, group_id) return f"已删除: {key} -> {group_id}" else: return f"群ID {group_id} 不存在于关键词 {key} 中" except redis.RedisError as e: return f"操作失败: {e}" def get_group_ids(self, key): """获取指定key下的所有群组ID""" try: group_ids = self._get_redis_connection().smembers(self.mapping_prefix + key) if group_ids: return f"关键词 {key} 的群ID列表: {', '.join(group_ids)}" else: return f"关键词 '{key}' 没有关联的群ID" except redis.RedisError as e: return f"操作失败: {e}" def get_first_group_id(self, key): """获取指定key的第一个群组ID""" try: group_ids = self._get_redis_connection().smembers(self.mapping_prefix + key) if group_ids: first_group_id = next(iter(group_ids)) self.LOG.info(f"关键词 {key} 的第一个群ID: {first_group_id}") return first_group_id else: return f"关键词 '{key}' 没有关联的群ID" except redis.RedisError as e: return f"操作失败: {e}" def process_command(self, command): """处理命令行输入""" command_parts = command.split() if len(command_parts) == 0: return "无效命令" cmd = command_parts[0] if cmd == "add" and len(command_parts) == 3: key = command_parts[1] group_id = command_parts[2] return self.add_mapping(key, group_id) elif cmd == "del" and len(command_parts) == 3: key = command_parts[1] group_id = command_parts[2] return self.del_mapping(key, group_id) elif cmd == "get" and len(command_parts) == 2: key = command_parts[1] return self.get_group_ids(key) elif cmd == "get_first" and len(command_parts) == 2: key = command_parts[1] return self.get_first_group_id(key) elif cmd == "help": commands = ( "群自动邀请配置命令：\n" "'#加群配置|add key group_id' - 添加群组ID\n" "'#加群配置|del key group_id' - 删除群组ID\n" "'#加群配置|get key' - 获取所有群组ID\n" "'#加群配置|get_first key' - 获取第一个群组ID\n" "\n" "用户加群命令：\n" "'#加群 key' - 请求加入关键词对应的群聊\n" ) return commands else: return "未知命令或参数数量错误，请输入: #加群配置|help 获取帮助" def get_help(self) -> str: """获取插件帮助信息""" return """自动加群功能插件： 1. 管理员可以通过 #加群配置 命令管理群聊映射 2. 用户可以通过 #加群 关键词 命令请求加入指定群聊 3. 使用 #加群配置|help 获取详细命令说明"""