# 群清单管理 # 群功能管理 # 0.加入或者关闭群机器人 #启用群机器人 #关闭群机器人 # 1.每日新闻自动播报 #启用每日新闻播报 #关闭每日新闻播报 # 2.每日群发言总结 #启用群发言 #关闭群发言 # 3.群AI能力 #启用群AI #关闭群AI # 4.群总结能力 #启用群总结 #关闭群总结 # 5.sehuatang PDF能力 #启用pdf #关闭pdf from typing import List import redis from enum import Enum from loguru import logger from db.connection import DBConnectionManager class PermissionStatus(Enum): """权限状态枚举""" ENABLED = "enabled" DISABLED = "disabled" # 在Feature枚举类中添加新的功能权限 class Feature(Enum): """功能权限枚举，带序号""" ROBOT = 1, "群机器人-总开关" DAILY_NEWS = 2, "每日新闻自动播报[每日8点30分定时发送]" DAILY_SUMMARY = 3, "每日群发言总结[每日9点30分定时发送]" AI_CAPABILITY = 4, "AI对话 [ai, 聊天, AI] 用法：ai 今天是什么天气？" SUMMARY_CAPABILITY = 5, "群总结能力 [#总结]" PDF_CAPABILITY = 6, "sehuatang PDF能力[无]" EPIC = 7, "EPIC自动播报[每周五自动发送]" # 新增的功能 PIC = 8, "图来能力 [图来, 秀人]" TASK_GAME = 9, "百科答题 [/t , /s, /a 任务ID 答案]" MUSIC = 10, "点歌功能 [点歌, 音乐, 音乐点播, 点播音乐, 音乐点歌]" SIGNIN = 11, "签到功能 [签到, 每日签到, qd, Qd, QD, 上班, 牛马]" POINT_TRADE = 12, "积分赠送 [积分赠送 1 @XX, 积分排行, 打劫 @XX, 保释 @XX]" BEAUTY_LEG = 13, "腿来能力 [美腿, 腿来]" VIDEO = 14, "黑丝视频 [黑丝视频, 黑丝, 来个黑丝,搞个黑丝]" VIDEO_MAN = 15, "肌肉视频 [猛男, 肌肉, 帅哥]" # GROUP_ADD = 16, "加群提醒" DOUYIN_PARSER = 17, "抖音链接转视频" GROUP_MEMBER_CHANGE = 18, "群成员变更提醒功能[自动触发]" # KID_PHOTO_EXTRACT = 19, "儿童照片提取转发功能" # 小朋友照片提取功能 NEWS = 20, "全球政治经济新闻" WEATHER = 21, "天气查询[上海天气 天气上海]" JD_TOKEN = 22, "JD_京豆token设置[设置京东 pt_key=xxx;pt_pin=xxx; 备注名称]" AI_AUTO = 23, "仿真对话" GUESS_MUSIC = 24, "猜歌名游戏 [猜歌名 - 开始猜歌游戏 |猜歌名 歌手名 - 开始指定歌手的猜歌游戏| 猜歌名 歌名 - 提交你的答案]" def __new__(cls, value, description): obj = object.__new__(cls) obj._value_ = value obj.description = description # 添加描述 return obj def __str__(self): return self.description def get_redis_connection(): # 初始化时加载本地缓存 db_manager: DBConnectionManager = DBConnectionManager.get_instance() return db_manager.get_redis_connection() class GroupBotManager: """群机器人管理，支持本地缓存""" # 本地缓存作为类级别静态属性 local_cache = { "group_permissions": {}, # 用于缓存群组功能权限 "group_list": set() # 用于缓存 group:list } @staticmethod def display_menu_status(group_id): """显示所有功能列表及其在指定群组中的当前状态，带emoji""" menu = [] for feature in Feature: status = GroupBotManager.get_group_permission(group_id, feature) status_emoji = "✅" if status == PermissionStatus.ENABLED else "❌" status_str = "启用" if status == PermissionStatus.ENABLED else "关闭" menu.append(f"{status_emoji} {status_str}-{feature.value}-{feature.description}") return "\n".join(menu) @staticmethod def load_local_cache(): r = get_redis_connection() """从 Redis 加载数据到本地缓存""" group_list = r.smembers("group:list") # 输出group_list 列表 logger.info(f"group_list: {group_list}") GroupBotManager.local_cache["group_list"] = set(group_list) # 加载群组权限 for group_id in GroupBotManager.local_cache["group_list"]: key = f'group:{group_id}:permissions' GroupBotManager.local_cache["group_permissions"][group_id] = {} for feature in Feature: status_value = r.hget(key, feature.name) if status_value: GroupBotManager.local_cache["group_permissions"][group_id][feature] = PermissionStatus(status_value) else: GroupBotManager.local_cache["group_permissions"][group_id][feature] = PermissionStatus.DISABLED @staticmethod def save_to_redis(): r = get_redis_connection() """将本地缓存保存回 Redis""" # 保存 group:list 到 Redis r.sadd("group:list", *GroupBotManager.local_cache["group_list"]) # 保存每个群组的权限到 Redis for group_id, permissions in GroupBotManager.local_cache["group_permissions"].items(): key = f'group:{group_id}:permissions' for feature, status in permissions.items(): r.hset(key, feature.name, status.value) @staticmethod def set_group_permission(group_id, feature: Feature, status: PermissionStatus): r = get_redis_connection() """设置群组功能权限并更新本地缓存""" # 更新本地缓存 if group_id not in GroupBotManager.local_cache["group_permissions"]: GroupBotManager.local_cache["group_permissions"][group_id] = {} GroupBotManager.local_cache["group_permissions"][group_id][feature] = status # 同步到 Redis key = f'group:{group_id}:permissions' r.hset(key, feature.name, status.value) # 输出保存到redis成功 logger.info(f"set_group_permission({group_id}, {feature.name}, {status.value})") @staticmethod def get_group_permission(group_id, feature: Feature): """获取群组某个功能的权限状态""" # 先从本地缓存获取 if group_id in GroupBotManager.local_cache["group_permissions"]: return GroupBotManager.local_cache["group_permissions"][group_id].get(feature, PermissionStatus.DISABLED) else: return PermissionStatus.DISABLED @staticmethod def check_permission(group_id, feature: Feature): """检查某个功能是否启用，若未启用则返回提示信息""" status = GroupBotManager.get_group_permission(group_id, feature) if status == PermissionStatus.DISABLED: return f"该功能未启用，请开启 {feature.description}。" return None # 如果已启用，则返回 None，不做处理 @staticmethod def handle_command(group_id, command_str): """统一处理群功能指令""" # print(f"PermissionStatus handle_command command_str: {command_str}") # 命令解析 command_parts = command_str.strip().split("-") # 如果是MENU指令，返回功能列表 if command_str.strip().upper() == "菜单": return f"群ID：{group_id} \n {GroupBotManager.get_enabled_features(group_id)}" # 如果是MENU-STATUS指令，返回功能列表及其状态 if command_str.strip().upper() == "菜单状态": return f"群ID：{group_id} \n {GroupBotManager.display_menu_status(group_id)}" # 如果是GROUP_LIST指令，返回 group:list 清单 if command_str.strip().upper() == "群列表": return GroupBotManager.get_group_list() # 如果是清除群指令 if command_str.strip().startswith("清除群-"): target_group_id = command_str.strip().split("-")[1] return GroupBotManager.remove_group(target_group_id) if len(command_parts) < 2: return None feature_str = command_parts[0] action = command_parts[1] # 如果第一个参数是序号，则转化为对应的功能 if feature_str.isdigit(): feature_num = int(feature_str) try: feature = Feature(feature_num) # 使用枚举序号查找功能 except ValueError: return "无效的功能序号" else: try: feature = Feature[feature_str] # 通过枚举名称获取功能枚举 except KeyError: return "无效功能名称" r = get_redis_connection() # 处理群机器人的启用和关闭（特别操作：更新 group:list） if feature == Feature.ROBOT: if action == "启用": GroupBotManager.set_group_permission(group_id, feature, PermissionStatus.ENABLED) # 启用群机器人时，将 group_id 加入 group:list GroupBotManager.local_cache["group_list"].add(group_id) # 同步到 Redis r.sadd("group:list", group_id) return f"BOT已启用，群组 {group_id} 已加入" elif action == "关闭": GroupBotManager.set_group_permission(group_id, feature, PermissionStatus.DISABLED) # 关闭群机器人时，从 group:list 中删除 group_id GroupBotManager.local_cache["group_list"].remove(group_id) # 同步到 Redis r.srem("group:list", group_id) return f"BOT已关闭，群组 {group_id} 已移除" else: return "无效操作，仅支持启用或关闭" # 先检查群机器人权限 robot_status = GroupBotManager.get_group_permission(group_id, Feature.ROBOT) if robot_status != PermissionStatus.ENABLED and feature != Feature.ROBOT: return "群机器人未启用，无法执行其他功能操作" # 根据不同的操作启用或禁用功能 if action == "启用": GroupBotManager.set_group_permission(group_id, feature, PermissionStatus.ENABLED) return f"群ID：{group_id} \n {feature.description} 已启用" elif action == "关闭": GroupBotManager.set_group_permission(group_id, feature, PermissionStatus.DISABLED) return f"群ID：{group_id} \n {feature.description} 已关闭" else: return "无效操作，仅支持启用或关闭" @staticmethod def list_group_permissions(group_id): """列出群组所有功能及其状态""" permissions = {} for feature in Feature: status = GroupBotManager.get_group_permission(group_id, feature) permissions[feature] = status if status else PermissionStatus.DISABLED # 默认为禁用状态 return permissions @staticmethod def display_menu(): """显示所有功能列表及其当前状态""" menu = [] for feature in Feature: menu.append(f"{feature.value}. {feature.description}") return "\n".join(menu) @staticmethod def get_enabled_features(group_id): """获取某个群已启用的功能列表及其描述，并返回格式化的字符串 只返回描述中包含指令(方括号[])的功能 Args: group_id: 群ID Returns: str: 格式化的已启用功能列表字符串 """ enabled_features = [] # 检查群是否在列表中 if group_id not in GroupBotManager.local_cache["group_list"]: return "该群未启用机器人功能" # 遍历所有功能，检查哪些已启用且包含指令 for feature in Feature: status = GroupBotManager.get_group_permission(group_id, feature) # 只添加已启用且描述中包含方括号的功能 if status == PermissionStatus.ENABLED and "[" in feature.description and "]" in feature.description: enabled_features.append({ "id": feature.value, "name": feature.name, "description": feature.description }) # 如果没有启用任何带指令的功能 if not enabled_features: return "该群未启用任何带指令的功能" # 构建格式化的字符串 result = f"不支持@交互，请通过指令触发\n 群功能菜单：\n" for feature in enabled_features: result += f"{feature['id']}.{feature['description']}\n" return result @staticmethod def get_group_list(): """返回所有启用了群机器人的群组清单，格式为集合""" return list(GroupBotManager.local_cache["group_list"]) @staticmethod def remove_group(group_id): """一键清除某个群的所有设置，用于退群或关闭群时处理 Args: group_id: 群ID Returns: str: 操作结果信息 """ # 检查群是否在列表中 if group_id not in GroupBotManager.local_cache["group_list"]: return f"群 {group_id} 不在机器人管理列表中" r = get_redis_connection() # 从本地缓存中移除群组 GroupBotManager.local_cache["group_list"].remove(group_id) if group_id in GroupBotManager.local_cache["group_permissions"]: del GroupBotManager.local_cache["group_permissions"][group_id] # 从Redis中移除群组 r.srem("group:list", group_id) r.delete(f'group:{group_id}:permissions') return f"已成功清除群 {group_id} 的所有设置" @staticmethod def get_admin_list() -> List[str]: """获取管理员列表 返回系统管理员的微信ID列表 """ # 从配置文件中获取管理员列表 config_admin_list = [] # self.config.get("admin_list", []) # 手动添加的管理员ID列表 manual_admin_list = [ "Jyunere", # 示例ID，请替换为实际的微信ID "wxid_abcdef", # 示例ID，请替换为实际的微信ID "filehelper" # 文件传输助手，方便自己测试 ] # 合并所有管理员列表并去重 all_admin_list = list(set(config_admin_list + manual_admin_list)) return all_admin_list # 示例命令 def simulate_commands(): # 加载本地缓存 GroupBotManager.load_local_cache() group_id = "49571962306@chatroom" print(GroupBotManager.get_group_permission(group_id, Feature.AI_CAPABILITY) == PermissionStatus.DISABLED) print(GroupBotManager.get_group_permission(group_id, Feature.SUMMARY_CAPABILITY) == PermissionStatus.ENABLED) if __name__ == '__main__': # 执行模拟命令 simulate_commands()