From 9d15bf965b44d7fda55cfd2a9feb150b31d659e7 Mon Sep 17 00:00:00 2001 From: liuwei Date: Mon, 9 Jun 2025 14:12:31 +0800 Subject: [PATCH] =?UTF-8?q?=E4=B8=B4=E6=97=B6=E8=B0=83=E6=95=B4=E6=9D=83?= =?UTF-8?q?=E9=99=90=E6=A8=A1=E5=9D=97=EF=BC=8C=E5=A4=87=E4=BB=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- base/auth/permission.py | 118 +++++++++++++++ base/auth/test_permission.py | 98 ++++++++++++ .../plugin_common/message_plugin_interface.py | 21 +++ base/plugin_common/plugin_manager.py | 2 +- plugins/ai_auto_response/main.py | 18 ++- plugins/beautyleg/main.py | 18 ++- plugins/dify/main.py | 22 ++- plugins/douyin_parser/main.py | 16 +- plugins/game_task/main.py | 16 +- plugins/global_news/main.py | 28 +++- plugins/group_auto_invite/main.py | 19 ++- plugins/group_member_change/main.py | 26 +++- plugins/group_virtual/main.py | 22 ++- plugins/guess_song/main.py | 36 ++--- plugins/jd_sign_token/main.py | 21 ++- plugins/message_sign/main.py | 10 ++ plugins/message_summary/main.py | 33 +++- plugins/music/main.py | 27 +++- plugins/plugin_manager/main.py | 16 +- plugins/point_trade/main.py | 14 ++ plugins/robot_menu/main.py | 25 ++- plugins/stats_collector/main.py | 17 ++- plugins/system_updater/main.py | 65 ++++---- plugins/video/main.py | 12 ++ plugins/video_man/main.py | 24 ++- plugins/weather/main.py | 19 ++- plugins/xiuren_image/main.py | 20 ++- utils/decorator/points_decorator.py | 26 ++-- utils/robot_cmd/Feature.py | 88 +++++++++++ utils/robot_cmd/robot_command.py | 143 ++++++++++++++---- 30 files changed, 882 insertions(+), 138 deletions(-) create mode 100644 base/auth/permission.py create mode 100644 base/auth/test_permission.py create mode 100644 utils/robot_cmd/Feature.py diff --git a/base/auth/permission.py b/base/auth/permission.py new file mode 100644 index 0000000..50a96d7 --- /dev/null +++ b/base/auth/permission.py @@ -0,0 +1,118 @@ +from typing import Dict, List, Optional +from dataclasses import dataclass +from enum import Enum +import json +import os +from utils.logger import logger + + +@dataclass +class Permission: + """权限类""" + key: str + name: str + description: str + value: int + plugin_id: Optional[str] = None + + +class PermissionStatus(Enum): + """权限状态枚举""" + ENABLED = 1 + DISABLED = 0 + + +class PermissionManager: + """权限管理器（单例模式）""" + _instance = None + _initialized = False + + def __new__(cls): + if cls._instance is None: + cls._instance = super().__new__(cls) + return cls._instance + + def __init__(self): + if not self._initialized: + self._permissions: Dict[str, Permission] = {} + self._config_file = "config/permissions.json" + self._load_permissions() + self._initialized = True + + def _load_permissions(self): + """从配置文件加载权限""" + try: + if os.path.exists(self._config_file): + with open(self._config_file, 'r', encoding='utf-8') as f: + data = json.load(f) + for key, perm_data in data.items(): + self._permissions[key] = Permission(**perm_data) + except Exception as e: + logger.error(f"加载权限配置失败: {e}") + + def _save_permissions(self): + """保存权限到配置文件""" + try: + os.makedirs(os.path.dirname(self._config_file), exist_ok=True) + with open(self._config_file, 'w', encoding='utf-8') as f: + json.dump( + {k: v.__dict__ for k, v in self._permissions.items()}, + f, + ensure_ascii=False, + indent=2 + ) + except Exception as e: + logger.error(f"保存权限配置失败: {e}") + + def register_permission( + self, + key: str, + name: str, + description: str, + plugin_id: Optional[str] = None + ) -> Permission: + """注册新权限""" + if key in self._permissions: + return self._permissions[key] + + value = len(self._permissions) + 1 + permission = Permission( + key=key, + name=name, + description=description, + value=value, + plugin_id=plugin_id + ) + self._permissions[key] = permission + self._save_permissions() + return permission + + def get_permission(self, key: str) -> Optional[Permission]: + """获取权限""" + return self._permissions.get(key) + + def get_permission_by_value(self, value: int) -> Optional[Permission]: + """通过值获取权限""" + for perm in self._permissions.values(): + if perm.value == value: + return perm + return None + + def get_all_permissions(self) -> List[Permission]: + """获取所有权限""" + return list(self._permissions.values()) + + def get_plugin_permissions(self, plugin_id: str) -> List[Permission]: + """获取插件的所有权限""" + return [ + perm for perm in self._permissions.values() + if perm.plugin_id == plugin_id + ] + + def remove_permission(self, key: str) -> bool: + """移除权限""" + if key in self._permissions: + del self._permissions[key] + self._save_permissions() + return True + return False \ No newline at end of file diff --git a/base/auth/test_permission.py b/base/auth/test_permission.py new file mode 100644 index 0000000..d017388 --- /dev/null +++ b/base/auth/test_permission.py @@ -0,0 +1,98 @@ +from .permission import PermissionManager, PermissionStatus +from .group_permission import GroupPermissionManager + + +def test_permission_system(): + # 获取权限管理器实例 + perm_manager = PermissionManager() + group_manager = GroupPermissionManager() + + # 测试注册权限 + print("测试注册权限...") + robot_perm = perm_manager.register_permission( + "ROBOT", + "群机器人", + "🔧 群机器人 [总开关]" + ) + print(f"注册权限: {robot_perm.key} = {robot_perm.value} - {robot_perm.description}") + + # 测试重复注册 + print("\n测试重复注册...") + same_perm = perm_manager.register_permission( + "ROBOT", + "群机器人", + "🔧 群机器人 [总开关]" + ) + print(f"重复注册结果: {same_perm.key} = {same_perm.value} - {same_perm.description}") + + # 测试注册新权限 + print("\n测试注册新权限...") + news_perm = perm_manager.register_permission( + "NEWS", + "新闻播报", + "📰 每日新闻播报", + plugin_id="news_plugin" + ) + print(f"新权限: {news_perm.key} = {news_perm.value} - {news_perm.description}") + + # 测试获取权限 + print("\n测试获取权限...") + retrieved_perm = perm_manager.get_permission("ROBOT") + print(f"获取权限: {retrieved_perm.key} = {retrieved_perm.value} - {retrieved_perm.description}") + + # 测试通过值获取权限 + print("\n测试通过值获取权限...") + perm_by_value = perm_manager.get_permission_by_value(1) + print(f"通过值获取: {perm_by_value.key} = {perm_by_value.value} - {perm_by_value.description}") + + # 测试获取所有权限 + print("\n测试获取所有权限...") + all_perms = perm_manager.get_all_permissions() + print("所有权限:") + for perm in all_perms: + print(f" {perm.key} = {perm.value} - {perm.description}") + + # 测试获取插件权限 + print("\n测试获取插件权限...") + plugin_perms = perm_manager.get_plugin_permissions("news_plugin") + print("插件权限:") + for perm in plugin_perms: + print(f" {perm.key} = {perm.value} - {perm.description}") + + # 测试群组权限 + print("\n测试群组权限...") + group_id = "test_group" + + # 添加群组 + group_manager.add_group(group_id) + print(f"添加群组: {group_id}") + + # 设置群组权限 + group_manager.set_group_permission(group_id, robot_perm, PermissionStatus.ENABLED) + print(f"设置权限: {group_id} - {robot_perm.key} = {PermissionStatus.ENABLED.value}") + + # 获取群组权限 + status = group_manager.get_group_permission(group_id, robot_perm) + print(f"获取权限状态: {group_id} - {robot_perm.key} = {status.value}") + + # 检查权限 + result = group_manager.check_permission(group_id, robot_perm) + print(f"检查权限: {group_id} - {robot_perm.key} = {result}") + + # 列出群组权限 + print("\n列出群组权限...") + perms = group_manager.list_group_permissions(group_id) + print(f"群组 {group_id} 的权限:") + for perm, status in perms.items(): + print(f" {perm.key} = {status.value}") + + # 移除群组 + print("\n测试移除群组...") + if group_manager.remove_group(group_id): + print(f"成功移除群组: {group_id}") + else: + print(f"移除群组失败: {group_id}") + + +if __name__ == '__main__': + test_permission_system() \ No newline at end of file diff --git a/base/plugin_common/message_plugin_interface.py b/base/plugin_common/message_plugin_interface.py index 42bcde6..cadc2b8 100644 --- a/base/plugin_common/message_plugin_interface.py +++ b/base/plugin_common/message_plugin_interface.py @@ -1,6 +1,7 @@ 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): @@ -16,6 +17,26 @@ class MessagePluginInterface(PluginInterface): """支持的命令列表""" 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 diff --git a/base/plugin_common/plugin_manager.py b/base/plugin_common/plugin_manager.py index 497728c..16ad3dc 100644 --- a/base/plugin_common/plugin_manager.py +++ b/base/plugin_common/plugin_manager.py @@ -331,7 +331,7 @@ class PluginManager: return plugin except Exception as e: - self.LOG.error(f"PluginManager：加载插件模块 {module_name} 失败: {e}", exc_info=True) + self.LOG.exception(f"PluginManager：加载插件模块 {module_name} 失败: {e}", exc_info=True) return None def unload_plugin(self, name: str) -> bool: diff --git a/plugins/ai_auto_response/main.py b/plugins/ai_auto_response/main.py index e3f830a..ea8d0c3 100644 --- a/plugins/ai_auto_response/main.py +++ b/plugins/ai_auto_response/main.py @@ -14,6 +14,10 @@ from .bot_ai import InterventionBot class AIAutoResponsePlugin(MessagePluginInterface): """AI自动对话插件""" + # 功能权限常量 + FEATURE_KEY = "AI_AUTO_RESPONSE" + FEATURE_DESCRIPTION = "🤖 AI自动对话功能 [自动对话]" + @property def name(self) -> str: return "AI自动对话" @@ -38,11 +42,21 @@ class AIAutoResponsePlugin(MessagePluginInterface): def commands(self) -> List[str]: return self._commands + @property + def feature_key(self) -> Optional[str]: + return self.FEATURE_KEY + + @property + def feature_description(self) -> Optional[str]: + return self.FEATURE_DESCRIPTION + def __init__(self): super().__init__() self.intervention_bot = None self.group_messages = {} # 存储每个群的最近消息 self.max_messages = 20 # 每个群最多存储的消息数量 + # 注册功能权限 + self.feature = self.register_feature() # DIFY API配置 self.dify_api_url = "" @@ -89,7 +103,7 @@ class AIAutoResponsePlugin(MessagePluginInterface): content = str(message.get("content", "")).strip() roomid = message.get("roomid", "") - if GroupBotManager.get_group_permission(roomid, Feature.AI_AUTO) == PermissionStatus.DISABLED: + if GroupBotManager.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED: return False # 如果是群消息，且该群启用了自动回复，则处理 if roomid: @@ -133,7 +147,7 @@ class AIAutoResponsePlugin(MessagePluginInterface): roomid = message.get("roomid", "") bot: WechatAPIClient = message.get("bot") # 检查权限 - if roomid and GroupBotManager.get_group_permission(roomid, Feature.AI_AUTO) == PermissionStatus.DISABLED: + if roomid and GroupBotManager.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED: return False, "没有权限" # 处理自动回复 try: diff --git a/plugins/beautyleg/main.py b/plugins/beautyleg/main.py index d86d638..474b436 100644 --- a/plugins/beautyleg/main.py +++ b/plugins/beautyleg/main.py @@ -15,6 +15,10 @@ from wechat_ipad import WechatAPIClient class BeautyLegPlugin(MessagePluginInterface): """美腿图片插件""" + # 功能权限常量 + FEATURE_KEY = "BEAUTY_LEG" + FEATURE_DESCRIPTION = "👠 美腿图片功能 [美腿, 腿来]" + @property def name(self) -> str: return "美腿图片" @@ -39,8 +43,18 @@ class BeautyLegPlugin(MessagePluginInterface): def commands(self) -> List[str]: return self._commands + @property + def feature_key(self) -> Optional[str]: + return self.FEATURE_KEY + + @property + def feature_description(self) -> Optional[str]: + return self.FEATURE_DESCRIPTION + def __init__(self): super().__init__() + # 注册功能权限 + self.feature = self.register_feature() # 修改图片目录路径指向 resource/beauty_leg self.image_folder = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), "resource", "beauty_leg") @@ -90,7 +104,7 @@ class BeautyLegPlugin(MessagePluginInterface): return command in self._commands @plugin_stats_decorator(plugin_name="美腿图片") - @plugin_points_cost(2, "美腿图片消耗积分", Feature.BEAUTY_LEG) + @plugin_points_cost(2, "美腿图片消耗积分", FEATURE_KEY) async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]: """处理消息""" content = str(message.get("content", "")).strip() @@ -101,7 +115,7 @@ class BeautyLegPlugin(MessagePluginInterface): bot: WechatAPIClient = message.get("bot") # 检查权限 - if roomid and gbm.get_group_permission(roomid, Feature.BEAUTY_LEG) == PermissionStatus.DISABLED: + if roomid and gbm.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED: return False, "没有权限" try: diff --git a/plugins/dify/main.py b/plugins/dify/main.py index 4fe25dc..f65e256 100644 --- a/plugins/dify/main.py +++ b/plugins/dify/main.py @@ -29,6 +29,10 @@ VIDEO_EXTENSIONS = {'.mp4', '.avi', '.mov', '.wmv', '.flv', '.mkv', '.webm'} class DifyPlugin(MessagePluginInterface): """Dify AI聊天插件""" + # 功能权限常量 + FEATURE_KEY = "AI_CAPABILITY" + FEATURE_DESCRIPTION = "🤖 AI聊天功能 [ai, dify, 聊天, AI]" + @property def name(self) -> str: return "Dify聊天" @@ -53,6 +57,14 @@ class DifyPlugin(MessagePluginInterface): def commands(self) -> List[str]: return self._commands + @property + def feature_key(self) -> Optional[str]: + return self.FEATURE_KEY + + @property + def feature_description(self) -> Optional[str]: + return self.FEATURE_DESCRIPTION + def __init__(self): super().__init__() # 会话上下文管理，格式: {group_id/wxid: [conversation_history]} @@ -64,6 +76,8 @@ class DifyPlugin(MessagePluginInterface): # 会话过期时间（秒） self.conversation_timeout = 3600 # 1小时 self.last_activity: Dict[str, float] = {} + # 注册功能权限 + self.feature = self.register_feature() def initialize(self, context: Dict[str, Any]) -> bool: """初始化插件""" @@ -120,7 +134,7 @@ class DifyPlugin(MessagePluginInterface): return False @plugin_stats_decorator(plugin_name="Dify聊天") - @plugin_points_cost(2, "AI聊天消耗积分", Feature.AI_CAPABILITY) + @plugin_points_cost(2, "AI聊天消耗积分", FEATURE_KEY) async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]: """处理消息""" content = str(message.get("content", "")).strip() @@ -135,7 +149,7 @@ class DifyPlugin(MessagePluginInterface): target = roomid if roomid else sender # 检查权限 - if roomid and gbm.get_group_permission(target, Feature.AI_CAPABILITY) == PermissionStatus.DISABLED: + if roomid and gbm.get_group_permission(target, self.feature) == PermissionStatus.DISABLED: return False, "没有权限" # 处理被@的消息 @@ -184,8 +198,8 @@ class DifyPlugin(MessagePluginInterface): revoke.add_message_to_revoke(target, client_msg_id, create_time, new_msg_id, 5) return False, f"处理出错: {e}" - async def _send_response(self, bot: WechatAPIClient, target: str, sender: str, - response: str, roomid: str) -> Tuple[bool, str]: + async def _send_response(self, bot: WechatAPIClient, target: str, sender: str, + response: str, roomid: str) -> Tuple[bool, str]: """发送响应消息的辅助方法""" try: # 判断是否为本地文件路径 diff --git a/plugins/douyin_parser/main.py b/plugins/douyin_parser/main.py index add753c..5f068f3 100644 --- a/plugins/douyin_parser/main.py +++ b/plugins/douyin_parser/main.py @@ -22,6 +22,10 @@ class DouyinParserError(Exception): class DouyinParserPlugin(MessagePluginInterface): """抖音无水印解析插件""" + # 功能权限常量 + FEATURE_KEY = "DOUYIN_PARSER" + FEATURE_DESCRIPTION = "🎵 抖音解析功能 [自动解析抖音链接]" + @property def name(self) -> str: return "抖音解析" @@ -46,10 +50,20 @@ class DouyinParserPlugin(MessagePluginInterface): def commands(self) -> List[str]: return [] # 不使用命令触发，而是通过消息内容匹配 + @property + def feature_key(self) -> Optional[str]: + return self.FEATURE_KEY + + @property + def feature_description(self) -> Optional[str]: + return self.FEATURE_DESCRIPTION + def __init__(self): super().__init__() self.LOG = logger self.url_pattern = re.compile(r'https?://v\.douyin\.com/\w+/?') + # 注册功能权限 + self.feature = self.register_feature() # 修改为使用插件目录下的down_load_dir文件夹 self.download_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "down_load_dir") # 确保下载目录存在 @@ -105,7 +119,7 @@ class DouyinParserPlugin(MessagePluginInterface): bot: WechatAPIClient = message.get("bot") # 检查权限 - if roomid and gbm.get_group_permission(roomid, Feature.DOUYIN_PARSER) == PermissionStatus.DISABLED: + if roomid and gbm.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED: return False, "没有权限" try: diff --git a/plugins/game_task/main.py b/plugins/game_task/main.py index c4cead1..e3a4eb5 100644 --- a/plugins/game_task/main.py +++ b/plugins/game_task/main.py @@ -19,6 +19,18 @@ import json class GameTaskPlugin(MessagePluginInterface): """游戏任务插件""" + # 功能权限常量 + FEATURE_KEY = "TASK_GAME" + FEATURE_DESCRIPTION = "🎮 百科问答功能 [/t, /a, /s, /r, /l, /h]" + + @property + def feature_key(self) -> Optional[str]: + return self.FEATURE_KEY + + @property + def feature_description(self) -> Optional[str]: + return self.FEATURE_DESCRIPTION + @property def name(self) -> str: return "百科问答" @@ -46,6 +58,8 @@ class GameTaskPlugin(MessagePluginInterface): def __init__(self): super().__init__() self.LOG = logger + # 注册功能权限 + self.feature = self.register_feature() async_job.at_times(["17:58"])(self.run_random_task_assignment) def initialize(self, context: Dict[str, Any]) -> bool: @@ -123,7 +137,7 @@ class GameTaskPlugin(MessagePluginInterface): self.LOG.debug(f"插件执行： {self.name}：{content}") # 检查权限 - if roomid and gbm.get_group_permission(roomid, Feature.TASK_GAME) == PermissionStatus.DISABLED: + if roomid and gbm.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED: return False, "没有权限" try: diff --git a/plugins/global_news/main.py b/plugins/global_news/main.py index d3381cb..346e16e 100644 --- a/plugins/global_news/main.py +++ b/plugins/global_news/main.py @@ -18,11 +18,15 @@ from .news_crawler import nbc, cnn, abc, fox, bbc class GlobalNewsPlugin(MessagePluginInterface): - """全球政治经济新闻插件""" + """全球新闻插件""" + + # 功能权限常量 + FEATURE_KEY = "GLOBAL_NEWS" + FEATURE_DESCRIPTION = "🌍 全球新闻功能 [全球新闻, 国际新闻, 环球新闻, 政经新闻]" @property def name(self) -> str: - return "全球政治经济新闻" + return "全球新闻" @property def version(self) -> str: @@ -30,7 +34,7 @@ class GlobalNewsPlugin(MessagePluginInterface): @property def description(self) -> str: - return "提供全球政治经济新闻，支持多个国际新闻源" + return "提供全球政治经济新闻查询功能" @property def author(self) -> str: @@ -44,11 +48,21 @@ class GlobalNewsPlugin(MessagePluginInterface): def commands(self) -> List[str]: return self._commands + @property + def feature_key(self) -> Optional[str]: + return self.FEATURE_KEY + + @property + def feature_description(self) -> Optional[str]: + return self.FEATURE_DESCRIPTION + def __init__(self): super().__init__() self.bot: WechatAPIClient = None self._news_tasks = {} # 存储正在进行的新闻抓取任务 + # 注册功能权限 + self.feature = self.register_feature() def initialize(self, context: Dict[str, Any]) -> bool: """初始化插件""" @@ -89,8 +103,8 @@ class GlobalNewsPlugin(MessagePluginInterface): return command in self._commands - @plugin_stats_decorator(plugin_name="全球政治经济新闻") - @plugin_points_cost(5, "全球新闻消耗积分", Feature.NEWS) + @plugin_stats_decorator(plugin_name="全球新闻") + @plugin_points_cost(1, "全球新闻消耗积分", FEATURE_KEY) async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]: """处理消息""" content = str(message.get("content", "")).strip() @@ -98,10 +112,10 @@ class GlobalNewsPlugin(MessagePluginInterface): sender = message.get("sender") roomid = message.get("roomid", "") gbm: GroupBotManager = message.get("gbm") - self.bot: WechatAPIClient = message.get("bot") + bot: WechatAPIClient = message.get("bot") # 检查权限 - if roomid and gbm.get_group_permission(roomid, Feature.NEWS) == PermissionStatus.DISABLED: + if roomid and gbm.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED: return False, "没有权限" # 生成唯一任务ID diff --git a/plugins/group_auto_invite/main.py b/plugins/group_auto_invite/main.py index 41c500a..02de094 100644 --- a/plugins/group_auto_invite/main.py +++ b/plugins/group_auto_invite/main.py @@ -5,7 +5,7 @@ from typing import Dict, Any, List, Optional, Tuple from base.plugin_common.message_plugin_interface import MessagePluginInterface from base.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.robot_cmd.robot_command import GroupBotManager, PermissionStatus from utils.wechat.contact_manager import ContactManager from wechat_ipad import WechatAPIClient @@ -13,6 +13,10 @@ from wechat_ipad import WechatAPIClient class GroupAutoInvitePlugin(MessagePluginInterface): """自动加群功能插件""" + # 功能权限常量 + FEATURE_KEY = "GROUP_AUTO_INVITE" + FEATURE_DESCRIPTION = "🤝 自动加群功能 [#加群配置, #加群]" + @property def name(self) -> str: return "自动加群功能" @@ -37,12 +41,22 @@ class GroupAutoInvitePlugin(MessagePluginInterface): def commands(self) -> List[str]: return self._commands + @property + def feature_key(self) -> Optional[str]: + return self.FEATURE_KEY + + @property + def feature_description(self) -> Optional[str]: + return self.FEATURE_DESCRIPTION + def __init__(self): super().__init__() # Redis 中存储群组映射的前缀 self.mapping_prefix = "group:group_mapping:" self._commands = [] self.bot: WechatAPIClient = None + # 注册功能权限 + self.feature = self.register_feature() def initialize(self, context: Dict[str, Any]) -> bool: """初始化插件""" @@ -86,6 +100,9 @@ class GroupAutoInvitePlugin(MessagePluginInterface): content = str(message.get("content", "")).strip() roomid = message.get("roomid", "") + if GroupBotManager.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED: + return False + # 处理加群配置命令 if content.startswith("#加群配置|"): return True diff --git a/plugins/group_member_change/main.py b/plugins/group_member_change/main.py index 7b6da90..57bf41f 100644 --- a/plugins/group_member_change/main.py +++ b/plugins/group_member_change/main.py @@ -16,6 +16,10 @@ from wechat_ipad.models.appmsg_xml import LINK_XML class GroupMemberChangePlugin(MessagePluginInterface): """群成员变更监控插件""" + # 功能权限常量 + FEATURE_KEY = "GROUP_MEMBER_CHANGE" + FEATURE_DESCRIPTION = "👥 群成员变更监控 [自动监控群成员变动并发送通知]" + @property def name(self) -> str: return "群成员变更监控" @@ -46,6 +50,8 @@ class GroupMemberChangePlugin(MessagePluginInterface): def __init__(self): super().__init__() + # 注册功能权限 + self.feature = self.register_feature() def initialize(self, context: Dict[str, Any]) -> bool: """初始化插件""" @@ -56,9 +62,15 @@ class GroupMemberChangePlugin(MessagePluginInterface): def can_process(self, message: Dict[str, Any]) -> bool: """检查是否可以处理该消息""" - content = message.get("content") - if not content or " Tuple[bool, Optional[str]]: @@ -74,7 +86,7 @@ class GroupMemberChangePlugin(MessagePluginInterface): bot: WechatAPIClient = message.get("bot") # 检查权限 - if roomid and gbm.get_group_permission(roomid, Feature.GROUP_MEMBER_CHANGE) == PermissionStatus.DISABLED: + if roomid and gbm.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED: return False, "没有权限" xml_content = str(content).strip().replace("\n", "").replace("\t", "") @@ -146,6 +158,14 @@ class GroupMemberChangePlugin(MessagePluginInterface): """插件支持的命令列表""" return [] + @property + def feature_key(self) -> Optional[str]: + return self.FEATURE_KEY + + @property + def feature_description(self) -> Optional[str]: + return self.FEATURE_DESCRIPTION + def get_help(self) -> str: """获取插件帮助信息""" return "群成员变更监控插件：自动监控群成员变动并发送通知。" diff --git a/plugins/group_virtual/main.py b/plugins/group_virtual/main.py index 624a424..7b7af0b 100644 --- a/plugins/group_virtual/main.py +++ b/plugins/group_virtual/main.py @@ -2,6 +2,7 @@ from typing import Dict, Any, List, Optional, Tuple from base.plugin_common.message_plugin_interface import MessagePluginInterface from base.plugin_common.plugin_interface import PluginStatus +from utils.robot_cmd.robot_command import GroupBotManager, PermissionStatus from utils.wechat.contact_manager import ContactManager from db.connection import DBConnectionManager from db.group_virtual_redis import GroupVirtualRedisDB @@ -12,6 +13,10 @@ from wechat_ipad.models.message import WxMessage class GroupVirtualPlugin(MessagePluginInterface): """跨群聊天插件""" + # 功能权限常量 + FEATURE_KEY = "GROUP_VIRTUAL" + FEATURE_DESCRIPTION = "🔄 跨群聊天功能 [自动转发群组间消息]" + @property def name(self) -> str: return "跨群聊天" @@ -36,11 +41,21 @@ class GroupVirtualPlugin(MessagePluginInterface): def commands(self) -> List[str]: return [] # 不处理任何命令 + @property + def feature_key(self) -> Optional[str]: + return self.FEATURE_KEY + + @property + def feature_description(self) -> Optional[str]: + return self.FEATURE_DESCRIPTION + def __init__(self): super().__init__() self.data = None self.message_cache = set() # 用于防止消息循环转发 self.group_virtual_redis = None + # 注册功能权限 + self.feature = self.register_feature() def initialize(self, context: Dict[str, Any]) -> bool: """初始化插件""" @@ -79,8 +94,13 @@ class GroupVirtualPlugin(MessagePluginInterface): if not self.enable: return False - # 只处理群消息 + content = str(message.get("content", "")).strip() roomid = message.get("roomid", "") + + if GroupBotManager.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED: + return False + + # 只处理群消息 if not roomid: return False diff --git a/plugins/guess_song/main.py b/plugins/guess_song/main.py index bf3211e..73bb4fa 100644 --- a/plugins/guess_song/main.py +++ b/plugins/guess_song/main.py @@ -76,6 +76,10 @@ class GuessSongRedisDB: class GuessSongPlugin(MessagePluginInterface): """猜歌名游戏插件""" + # 功能权限常量 + FEATURE_KEY = "GUESS_MUSIC" + FEATURE_DESCRIPTION = "🎵 猜歌名游戏 [猜歌名]" + @property def name(self) -> str: return "猜歌名游戏" @@ -100,8 +104,17 @@ class GuessSongPlugin(MessagePluginInterface): def commands(self) -> List[str]: return self._commands + @property + def feature_key(self) -> Optional[str]: + return self.FEATURE_KEY + + @property + def feature_description(self) -> Optional[str]: + return self.FEATURE_DESCRIPTION + def __init__(self): super().__init__() + self.feature = self.register_feature() self.redis_db = None def initialize(self, context: Dict[str, Any]) -> bool: @@ -166,35 +179,24 @@ class GuessSongPlugin(MessagePluginInterface): return command in self._commands - @plugin_stats_decorator(plugin_name="猜歌名游戏") + @plugin_stats_decorator(plugin_name="猜歌名") + @plugin_points_cost(2, "猜歌名消耗积分", FEATURE_KEY) async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]: """处理消息""" content = str(message.get("content", "")).strip() self.LOG.debug(f"插件执行： {self.name}：{content}") - - # 简化命令处理，只检查开头是否包含命令前缀 - command_found = False - for cmd in self._commands: - if content.startswith(cmd): - command_found = True - content = content[len(cmd):].strip() - break - - if not command_found: - return False, "不匹配的命令" - sender = message.get("sender") roomid = message.get("roomid", "") gbm: GroupBotManager = message.get("gbm") bot: WechatAPIClient = message.get("bot") + # 检查权限 + if roomid and gbm.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED: + return False, "没有权限" + # 使用roomid或sender作为游戏会话ID session_id = roomid if roomid else sender - # 检查权限 - if roomid and gbm.get_group_permission(roomid, Feature.GUESS_MUSIC) == PermissionStatus.DISABLED: - return False, "没有权限" - # 获取当前游戏会话 current_game = None if self.redis_db: diff --git a/plugins/jd_sign_token/main.py b/plugins/jd_sign_token/main.py index 53e9519..f0b45f5 100644 --- a/plugins/jd_sign_token/main.py +++ b/plugins/jd_sign_token/main.py @@ -151,6 +151,10 @@ class QL: class JDTokenPlugin(MessagePluginInterface): """京东签到Token设置插件""" + + # 功能权限常量 + FEATURE_KEY = "JD_TOKEN" + FEATURE_DESCRIPTION = "🔑 京东签到Token设置 [设置京东]" @property def name(self) -> str: @@ -176,8 +180,18 @@ class JDTokenPlugin(MessagePluginInterface): def commands(self) -> List[str]: return self._commands + @property + def feature_key(self) -> Optional[str]: + return self.FEATURE_KEY + + @property + def feature_description(self) -> Optional[str]: + return self.FEATURE_DESCRIPTION + def __init__(self): super().__init__() + # 注册功能权限 + self.feature = self.register_feature() def initialize(self, context: Dict[str, Any]) -> bool: """初始化插件""" @@ -221,9 +235,12 @@ class JDTokenPlugin(MessagePluginInterface): return False content = str(message.get("content", "")).strip() - command = content.split(" ")[0] + roomid = message.get("roomid", "") - return command in self._commands + if GroupBotManager.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED: + return False + + return True @plugin_stats_decorator(plugin_name="京东签到Token设置") async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]: diff --git a/plugins/message_sign/main.py b/plugins/message_sign/main.py index 59d9a0a..9a17b4b 100644 --- a/plugins/message_sign/main.py +++ b/plugins/message_sign/main.py @@ -45,6 +45,14 @@ class MessageSignPlugin(MessagePluginInterface): def commands(self) -> List[str]: return self._commands + @property + def feature_key(self) -> Optional[str]: + return "SIGN_IN" + + @property + def feature_description(self) -> Optional[str]: + return "📝 签到功能 [签到]" + def __init__(self): super().__init__() self.today_signin_count = {} @@ -57,6 +65,8 @@ class MessageSignPlugin(MessagePluginInterface): "resource", "6 托福-乱序.txt") self.vocab_list = [] self.bot: WechatAPIClient = None + # 注册功能权限 + self.feature = self.register_feature() def initialize(self, context: Dict[str, Any]) -> bool: """初始化插件""" diff --git a/plugins/message_summary/main.py b/plugins/message_summary/main.py index a9d5018..9dd2d1b 100644 --- a/plugins/message_summary/main.py +++ b/plugins/message_summary/main.py @@ -24,6 +24,10 @@ from wechat_ipad import WechatAPIClient class MessageSummaryPlugin(MessagePluginInterface): """消息总结插件，用于生成群聊消息总结""" + # 功能权限常量 + FEATURE_KEY = "SUMMARY_CAPABILITY" + FEATURE_DESCRIPTION = "📝 群聊总结功能 [总结]" + @property def name(self) -> str: return "群聊总结" @@ -46,12 +50,24 @@ class MessageSummaryPlugin(MessagePluginInterface): @property def commands(self) -> List[str]: - return ["总结", "summary"] + return self._commands + + @property + def feature_key(self) -> Optional[str]: + return self.FEATURE_KEY + + @property + def feature_description(self) -> Optional[str]: + return self.FEATURE_DESCRIPTION def __init__(self): super().__init__() - self.bot: WechatAPIClient = None - self.revoke: MessageAutoRevoke = None + self._commands = [] + self.message_storage = None + self.revoke = None + self.bot = None + # 注册功能权限 + self.feature = self.register_feature() def initialize(self, context: Dict[str, Any]) -> bool: """初始化插件""" @@ -242,3 +258,14 @@ class MessageSummaryPlugin(MessagePluginInterface): except Exception as e: self.LOG.error(f"处理总结时出现未知错误: {e}") return f"生成总结时出现未知错误: {str(e)}", None + + 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 GroupBotManager.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED: + return False diff --git a/plugins/music/main.py b/plugins/music/main.py index 4ecb3dd..7c3ffdf 100644 --- a/plugins/music/main.py +++ b/plugins/music/main.py @@ -19,6 +19,10 @@ from wechat_ipad.models.appmsg_xml import MUSIC_XML class MusicPlugin(MessagePluginInterface): """音乐点播插件""" + # 功能权限常量 + FEATURE_KEY = "MUSIC" + FEATURE_DESCRIPTION = "🎵 点歌功能 [点歌, 音乐, 音乐点播, 点播音乐, 音乐点歌]" + @property def name(self) -> str: return "音乐点播" @@ -43,8 +47,18 @@ class MusicPlugin(MessagePluginInterface): def commands(self) -> List[str]: return self._commands + @property + def feature_key(self) -> Optional[str]: + return self.FEATURE_KEY + + @property + def feature_description(self) -> Optional[str]: + return self.FEATURE_DESCRIPTION + def __init__(self): super().__init__() + # 注册功能权限 + self.feature = self.register_feature() def initialize(self, context: Dict[str, Any]) -> bool: """初始化插件""" @@ -83,28 +97,27 @@ class MusicPlugin(MessagePluginInterface): return command in self._commands - @plugin_stats_decorator(plugin_name="音乐点播") - @plugin_points_cost(2, "音乐点播消耗积分", Feature.MUSIC) + @plugin_stats_decorator(plugin_name="点歌") + @plugin_points_cost(2, "点歌消耗积分", FEATURE_KEY) async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]: """处理消息""" content = str(message.get("content", "")).strip() self.LOG.debug(f"插件执行： {self.name}：{content}") - command = content.split(" ")[0] sender = message.get("sender") roomid = message.get("roomid", "") gbm: GroupBotManager = message.get("gbm") bot: WechatAPIClient = message.get("bot") + # 检查权限 + if roomid and gbm.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED: + return False, "没有权限" + # 检查命令格式 if len(content.split(" ")) == 1: await bot.send_text_message((roomid if roomid else sender), f"❌命令格式错误！\n{self.command_format}" , sender) return False, "命令格式错误" - # 检查权限 - if roomid and gbm.get_group_permission(roomid, Feature.MUSIC) == PermissionStatus.DISABLED: - return False, "没有权限" - # 提取歌曲名 user_song_name = content[len(command):].strip() diff --git a/plugins/plugin_manager/main.py b/plugins/plugin_manager/main.py index 0060a99..b13aa6e 100644 --- a/plugins/plugin_manager/main.py +++ b/plugins/plugin_manager/main.py @@ -37,6 +37,14 @@ class PluginManagerPlugin(MessagePluginInterface): def commands(self) -> List[str]: return self._commands + @property + def feature_key(self) -> Optional[str]: + return "PLUGIN_MANAGER" + + @property + def feature_description(self) -> Optional[str]: + return "🔧 插件管理功能 [插件管理]" + def start(self) -> bool: """启动插件""" self.LOG.info(f"[{self.name}] 插件已启动") @@ -52,6 +60,8 @@ class PluginManagerPlugin(MessagePluginInterface): def __init__(self): super().__init__() self.bot: WechatAPIClient = None + # 注册功能权限 + self.feature = self.register_feature() def initialize(self, context: Dict[str, Any]) -> bool: """初始化插件""" @@ -93,6 +103,10 @@ class PluginManagerPlugin(MessagePluginInterface): self.LOG.error("WechatAPIClient 未初始化") return False, "Bot 未初始化" + # 检查功能权限 + if roomid and gbm.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED: + return False, "没有权限" + # 检查命令格式 parts = content.split(" ") if len(parts) == 1: @@ -103,7 +117,7 @@ class PluginManagerPlugin(MessagePluginInterface): self.plugin_registry = PluginRegistry() self.plugin_manager = PluginManager().get_instance() - # 检查权限 (只允许管理员操作) + # 检查管理员权限 if not self._is_admin(sender, gbm): await self.bot.send_at_message(target, f"❌权限不足，只有管理员可以管理插件", [sender]) return True, "权限不足" diff --git a/plugins/point_trade/main.py b/plugins/point_trade/main.py index ba9240b..23d19ff 100644 --- a/plugins/point_trade/main.py +++ b/plugins/point_trade/main.py @@ -21,6 +21,10 @@ from wechat_ipad.models.message import WxMessage class PointTradePlugin(MessagePluginInterface): """积分交易插件""" + # 功能权限常量 + FEATURE_KEY = "POINT_TRADE" + FEATURE_DESCRIPTION = "💱 积分交易功能 [积分交易, 积分转账, 我的积分, 积分排行, 打劫, 保释]" + @property def name(self) -> str: return "积分交易" @@ -45,11 +49,21 @@ class PointTradePlugin(MessagePluginInterface): def commands(self) -> List[str]: return self._commands + @property + def feature_key(self) -> Optional[str]: + return self.FEATURE_KEY + + @property + def feature_description(self) -> Optional[str]: + return self.FEATURE_DESCRIPTION + def __init__(self): super().__init__() self.db_pool = None self.bot: WechatAPIClient = None self.revoke: MessageAutoRevoke = None + # 注册功能权限 + self.feature = self.register_feature() def initialize(self, context: Dict[str, Any]) -> bool: """初始化插件""" diff --git a/plugins/robot_menu/main.py b/plugins/robot_menu/main.py index c0ce59d..84740de 100644 --- a/plugins/robot_menu/main.py +++ b/plugins/robot_menu/main.py @@ -4,7 +4,7 @@ from typing import Dict, Any, List, Optional, Tuple from base.plugin_common.message_plugin_interface import MessagePluginInterface from base.plugin_common.plugin_interface import PluginStatus -from utils.decorator.plugin_decorators import plugin_stats_decorator +from utils.decorator.plugin_decorators import plugin_stats_decorator, plugin_points_cost from utils.revoke.message_auto_revoke import MessageAutoRevoke from utils.robot_cmd.robot_command import Feature, PermissionStatus, GroupBotManager from wechat_ipad import WechatAPIClient @@ -12,6 +12,10 @@ from wechat_ipad import WechatAPIClient class RobotMenuPlugin(MessagePluginInterface): """功能菜单插件""" + + # 功能权限常量 + FEATURE_KEY = "ROBOT_MENU" + FEATURE_DESCRIPTION = "📋 功能菜单 [菜单, 功能]" @property def name(self) -> str: @@ -37,8 +41,18 @@ class RobotMenuPlugin(MessagePluginInterface): def commands(self) -> List[str]: return self._commands + @property + def feature_key(self) -> Optional[str]: + return self.FEATURE_KEY + + @property + def feature_description(self) -> Optional[str]: + return self.FEATURE_DESCRIPTION + def __init__(self): super().__init__() + # 注册功能权限 + self.feature = self.register_feature() def initialize(self, context: Dict[str, Any]) -> bool: """初始化插件""" @@ -133,17 +147,22 @@ class RobotMenuPlugin(MessagePluginInterface): return "当前没有启用机器人的群组" return "\n".join(group_list) - @plugin_stats_decorator(plugin_name="功能菜单") + @plugin_stats_decorator(plugin_name="机器人菜单") + @plugin_points_cost(1, "机器人菜单消耗积分", FEATURE_KEY) async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]: """处理消息""" content = str(message.get("content", "")).strip() self.LOG.debug(f"插件执行： {self.name}：{content}") - command = content.split(" ")[0] sender = message.get("sender") roomid = message.get("roomid", "") + gbm: GroupBotManager = message.get("gbm") bot: WechatAPIClient = message.get("bot") revoke: MessageAutoRevoke = message.get("revoke") + # 检查权限 + if roomid and gbm.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED: + return False, "没有权限" + # 检查命令格式 if len(content.split(" ")) == 1: # 显示功能菜单 diff --git a/plugins/stats_collector/main.py b/plugins/stats_collector/main.py index bd29057..9c6fc4d 100644 --- a/plugins/stats_collector/main.py +++ b/plugins/stats_collector/main.py @@ -13,6 +13,18 @@ from db.connection import DBConnectionManager class StatsCollectorPlugin(PluginInterface): """统计收集插件""" + # 功能权限常量 + FEATURE_KEY = "STATS_COLLECTOR" + FEATURE_DESCRIPTION = "📊 指令记录功能 [自动记录指令使用情况]" + + @property + def feature_key(self) -> Optional[str]: + return self.FEATURE_KEY + + @property + def feature_description(self) -> Optional[str]: + return self.FEATURE_DESCRIPTION + @property def name(self) -> str: return "指令记录" @@ -38,10 +50,8 @@ class StatsCollectorPlugin(PluginInterface): return [] def __init__(self): - super().__init__() self.LOG = logger - self.LOG.info(f"正在初始化 {self.name} 插件...") # 默认配置 self.config = { @@ -49,10 +59,11 @@ class StatsCollectorPlugin(PluginInterface): "record_all_plugins": True, # 是否记录所有插件的调用 "excluded_plugins": [], # 排除的插件列表 } - self.event_manager = EventManager.get_instance() self.db_manager = DBConnectionManager.get_instance() self.stats_db = StatsDBOperator(self.db_manager) + # 注册功能权限 + self.feature = self.register_feature() def initialize(self, config: Dict[str, Any]) -> bool: """初始化插件""" diff --git a/plugins/system_updater/main.py b/plugins/system_updater/main.py index f8dd9ed..3c3f483 100644 --- a/plugins/system_updater/main.py +++ b/plugins/system_updater/main.py @@ -5,13 +5,20 @@ from typing import Dict, Any, List, Optional, Tuple from base.plugin_common.message_plugin_interface import MessagePluginInterface from base.plugin_common.plugin_interface import PluginStatus -from utils.robot_cmd.robot_command import Feature, PermissionStatus +from utils.decorator.plugin_decorators import plugin_stats_decorator +from utils.decorator.points_decorator import plugin_points_cost +from utils.robot_cmd.robot_command import Feature, PermissionStatus, GroupBotManager from wechat_ipad import WechatAPIClient + class SystemUpdaterPlugin(MessagePluginInterface): """系统更新插件""" + # 功能权限常量 + FEATURE_KEY = "SYSTEM_UPDATER" + FEATURE_DESCRIPTION = "🔄 系统更新功能 [更新系统, 系统更新]" + @property def name(self) -> str: return "系统更新" @@ -36,11 +43,21 @@ class SystemUpdaterPlugin(MessagePluginInterface): def commands(self) -> List[str]: return self._commands + @property + def feature_key(self) -> Optional[str]: + return self.FEATURE_KEY + + @property + def feature_description(self) -> Optional[str]: + return self.FEATURE_DESCRIPTION + def __init__(self): super().__init__() self.admin_wxids = [] self.wait_time = 5 # 默认等待15秒 self.bot: WechatAPIClient = None + # 注册功能权限 + self.feature = self.register_feature() def initialize(self, context: Dict[str, Any]) -> bool: """初始化插件""" @@ -87,47 +104,31 @@ class SystemUpdaterPlugin(MessagePluginInterface): return False + @plugin_stats_decorator(plugin_name="系统更新") + @plugin_points_cost(1, "系统更新消耗积分", FEATURE_KEY) async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]: """处理消息""" content = str(message.get("content", "")).strip() + self.LOG.debug(f"插件执行： {self.name}：{content}") sender = message.get("sender") roomid = message.get("roomid", "") - gbm = message.get("gbm", None) + gbm: GroupBotManager = message.get("gbm") + bot: WechatAPIClient = message.get("bot") - self.bot: WechatAPIClient = message.get("bot") # 检查权限 - if self.admin_wxids and sender not in self.admin_wxids: - await self.bot.send_text_message((roomid if roomid else sender), "⚠️ 您没有执行此操作的权限", - sender) - return True, "无权限" + if roomid and gbm.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED: + return False, "没有权限" - # 如果是群消息，检查群权限 - if roomid and gbm and hasattr(gbm, 'get_group_permission'): - if gbm.get_group_permission(roomid, Feature.ROBOT) == PermissionStatus.DISABLED: - return False, "机器人功能已禁用" + # 检查是否是命令 + if content in self._commands: + return True, "命令已接收" - # 提取等待时间参数 - wait_time = self.wait_time - command = content.split(" ")[0] + # 检查是否是带参数的更新命令 + for cmd in self._commands: + if content.startswith(f"{cmd} "): + return True, "带参数的命令已接收" - if len(content.split(" ")) > 1: - try: - param = content.split(" ")[1].strip() - if param.isdigit(): - wait_time = int(param) - self.LOG.info(f"使用自定义等待时间: {wait_time}秒") - except: - pass - - # 发送更新通知 - await self.bot.send_text_message((roomid if roomid else sender), - f"🔄 系统即将更新并重启，等待时间设置为{wait_time}秒...", - sender) - - # 执行系统更新脚本 - self._execute_system_update() - - return True, "系统更新中" + return False, "无法处理的消息" def _execute_system_update(self): """执行系统更新操作""" diff --git a/plugins/video/main.py b/plugins/video/main.py index 8aedf99..083918c 100644 --- a/plugins/video/main.py +++ b/plugins/video/main.py @@ -18,6 +18,18 @@ from wechat_ipad import WechatAPIClient class VideoPlugin(MessagePluginInterface): """视频插件""" + # 功能权限常量 + FEATURE_KEY = "VIDEO" + FEATURE_DESCRIPTION = "🎥 黑丝视频功能 [黑丝视频, 黑丝, 来个黑丝, 搞个黑丝]" + + @property + def feature_key(self) -> Optional[str]: + return self.FEATURE_KEY + + @property + def feature_description(self) -> Optional[str]: + return self.FEATURE_DESCRIPTION + @property def name(self) -> str: return "黑丝视频" diff --git a/plugins/video_man/main.py b/plugins/video_man/main.py index c0299f2..b58bf6f 100644 --- a/plugins/video_man/main.py +++ b/plugins/video_man/main.py @@ -13,11 +13,16 @@ from base.plugin_common.plugin_interface import PluginStatus from utils.decorator.plugin_decorators import plugin_stats_decorator from utils.decorator.points_decorator import plugin_points_cost from utils.robot_cmd.robot_command import Feature, PermissionStatus, GroupBotManager +from base.wechat_api.wechat_api_client import WechatAPIClient class VideoManPlugin(MessagePluginInterface): """猛男视频插件""" + # 功能权限常量 + FEATURE_KEY = "VIDEO_MAN" + FEATURE_DESCRIPTION = "💪 猛男视频功能 [猛男视频, 猛男, 来个猛男, 搞个猛男]" + @property def name(self) -> str: return "猛男视频" @@ -42,10 +47,20 @@ class VideoManPlugin(MessagePluginInterface): def commands(self) -> List[str]: return self._commands + @property + def feature_key(self) -> Optional[str]: + return self.FEATURE_KEY + + @property + def feature_description(self) -> Optional[str]: + return self.FEATURE_DESCRIPTION + def __init__(self): super().__init__() # 使用插件目录下的down_load_dir文件夹 self.download_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "down_load_dir") + # 注册功能权限 + self.feature = self.register_feature() def initialize(self, context: Dict[str, Any]) -> bool: """初始化插件""" @@ -90,7 +105,7 @@ class VideoManPlugin(MessagePluginInterface): return command in self._commands @plugin_stats_decorator(plugin_name="猛男视频") - @plugin_points_cost(2, "猛男视频消耗积分", Feature.VIDEO_MAN) + @plugin_points_cost(2, "猛男视频消耗积分", FEATURE_KEY) async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]: """处理消息""" content = str(message.get("content", "")).strip() @@ -98,9 +113,10 @@ class VideoManPlugin(MessagePluginInterface): sender = message.get("sender") roomid = message.get("roomid", "") gbm: GroupBotManager = message.get("gbm") + bot: WechatAPIClient = message.get("bot") # 检查权限 - if roomid and gbm.get_group_permission(roomid, Feature.VIDEO_MAN) == PermissionStatus.DISABLED: + if roomid and gbm.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED: return False, "没有权限" try: @@ -108,12 +124,12 @@ class VideoManPlugin(MessagePluginInterface): file_abspath, first_frame = await self._download_video("https://api.52vmy.cn/api/video/boy?type=json") # FIXME 需要换成web容器地址。否则无法获取。 if not file_abspath: - await self.bot.send_text_message((roomid if roomid else sender), f"\n❌视频下载失败，请稍后再试", + await bot.send_text_message((roomid if roomid else sender), f"\n❌视频下载失败，请稍后再试", sender) return False, "视频下载失败" # 发送视频 - result = await self.bot.send_video_message((roomid if roomid else sender), Path(file_abspath), + result = await bot.send_video_message((roomid if roomid else sender), Path(file_abspath), Path(first_frame)) self.LOG.info(f"发送视频结果: {result}") return True, "发送成功" diff --git a/plugins/weather/main.py b/plugins/weather/main.py index cefa1e3..a4bd5f0 100644 --- a/plugins/weather/main.py +++ b/plugins/weather/main.py @@ -14,6 +14,10 @@ from wechat_ipad import WechatAPIClient class WeatherPlugin(MessagePluginInterface): """天气查询插件""" + # 功能权限常量 + FEATURE_KEY = "WEATHER" + FEATURE_DESCRIPTION = "🌤️ 天气查询功能 [天气]" + @property def name(self) -> str: return "天气查询" @@ -38,9 +42,19 @@ class WeatherPlugin(MessagePluginInterface): def commands(self) -> List[str]: return self._commands + @property + def feature_key(self) -> Optional[str]: + return self.FEATURE_KEY + + @property + def feature_description(self) -> Optional[str]: + return self.FEATURE_DESCRIPTION + def __init__(self): super().__init__() self.plugin_dir = os.path.dirname(os.path.abspath(__file__)) + # 注册功能权限 + self.feature = self.register_feature() def initialize(self, context: Dict[str, Any]) -> bool: """初始化插件""" @@ -88,17 +102,18 @@ class WeatherPlugin(MessagePluginInterface): return command in self._commands @plugin_stats_decorator(plugin_name="天气查询") - @plugin_points_cost(2, "天气查询消耗积分", Feature.WEATHER) + @plugin_points_cost(1, "天气查询消耗积分", FEATURE_KEY) async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]: """处理消息""" content = str(message.get("content", "")).strip() self.LOG.debug(f"插件执行： {self.name}：{content}") sender = message.get("sender") roomid = message.get("roomid", "") + gbm: GroupBotManager = message.get("gbm") bot: WechatAPIClient = message.get("bot") # 检查权限 - if roomid and GroupBotManager.get_group_permission(roomid, Feature.WEATHER) == PermissionStatus.DISABLED: + if roomid and gbm.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED: return False, "没有权限" # 处理消息内容 - 不再使用jieba分词 diff --git a/plugins/xiuren_image/main.py b/plugins/xiuren_image/main.py index ca2e485..3b7f3ab 100644 --- a/plugins/xiuren_image/main.py +++ b/plugins/xiuren_image/main.py @@ -18,6 +18,10 @@ from wechat_ipad import WechatAPIClient class XiurenImagePlugin(MessagePluginInterface): """秀人图片插件""" + # 功能权限常量 + FEATURE_KEY = "XIUREN_IMAGE" + FEATURE_DESCRIPTION = "🖼️ 秀人图片功能 [秀人]" + @property def name(self) -> str: return "秀人图片" @@ -42,10 +46,20 @@ class XiurenImagePlugin(MessagePluginInterface): def commands(self) -> List[str]: return self._commands + @property + def feature_key(self) -> Optional[str]: + return self.FEATURE_KEY + + @property + def feature_description(self) -> Optional[str]: + return self.FEATURE_DESCRIPTION + def __init__(self): super().__init__() # 使用Path对象处理路径，自动适应不同操作系统 self.image_folder = str(Path(Path(__file__).parent.parent.parent, "xiuren")) + # 注册功能权限 + self.feature = self.register_feature() def initialize(self, context: Dict[str, Any]) -> bool: """初始化插件""" @@ -97,8 +111,8 @@ class XiurenImagePlugin(MessagePluginInterface): return command in self._commands - @plugin_stats_decorator(plugin_name="秀人图片") - @plugin_points_cost(9, "秀人图片消耗积分", Feature.PIC) + @plugin_stats_decorator(plugin_name="秀人网图片") + @plugin_points_cost(2, "秀人网图片消耗积分", FEATURE_KEY) async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]: """处理消息""" content = str(message.get("content", "")).strip() @@ -109,7 +123,7 @@ class XiurenImagePlugin(MessagePluginInterface): bot: WechatAPIClient = message.get("bot") revoke: MessageAutoRevoke = message.get("revoke") # 检查权限 - if roomid and gbm.get_group_permission(roomid, Feature.PIC) == PermissionStatus.DISABLED: + if roomid and gbm.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED: return False, "没有权限" try: diff --git a/utils/decorator/points_decorator.py b/utils/decorator/points_decorator.py index ac1b573..3023d79 100644 --- a/utils/decorator/points_decorator.py +++ b/utils/decorator/points_decorator.py @@ -177,13 +177,13 @@ def points_reward_decorator(points_calculator: Union[int, Callable], source_type return decorator -def plugin_points_cost(points: int, description: str = None, feature: Feature = None): +def plugin_points_cost(points: int, description: str = None, feature_key: str = None): """插件积分消费装饰器 Args: points: 消费积分数量 description: 积分消费描述 - feature: 功能权限枚举 + feature_key: 功能权限键名 Returns: 装饰器函数 @@ -195,8 +195,10 @@ def plugin_points_cost(points: int, description: str = None, feature: Feature = try: # 检查权限 roomid = message.get("roomid", "") - if feature and roomid: - if GroupBotManager.get_group_permission(roomid, feature) == PermissionStatus.DISABLED: + if feature_key and roomid: + # 获取功能权限枚举 + feature = getattr(Feature, feature_key, None) + if feature and GroupBotManager.get_group_permission(roomid, feature) == PermissionStatus.DISABLED: return False, "没有权限" # 检查是否开启了积分获取功能 @@ -271,16 +273,18 @@ def plugin_points_cost(points: int, description: str = None, feature: Feature = try: # 检查权限 roomid = message.get("roomid", "") - if feature and roomid: - if GroupBotManager.get_group_permission(roomid, feature) == PermissionStatus.DISABLED: + if feature_key and roomid: + # 获取功能权限枚举 + feature = getattr(Feature, feature_key, None) + if feature and GroupBotManager.get_group_permission(roomid, feature) == PermissionStatus.DISABLED: return False, "没有权限" # 检查是否开启了积分获取功能 - if (GroupBotManager.get_group_permission(roomid, - Feature.SIGNIN) == PermissionStatus.DISABLED - and GroupBotManager.get_group_permission( - roomid, Feature.TASK_GAME) == PermissionStatus.DISABLED): - return func(self, message) + # if (GroupBotManager.get_group_permission(roomid, + # Feature.SIGNIN) == PermissionStatus.DISABLED + # and GroupBotManager.get_group_permission( + # roomid, Feature.TASK_GAME) == PermissionStatus.DISABLED): + # return func(self, message) # 获取消息信息 sender = message.get("sender", "") diff --git a/utils/robot_cmd/Feature.py b/utils/robot_cmd/Feature.py new file mode 100644 index 0000000..4153c8b --- /dev/null +++ b/utils/robot_cmd/Feature.py @@ -0,0 +1,88 @@ +from enum import Enum +from typing import List + + +class Feature(Enum): + """功能权限枚举，带序号和描述""" + ROBOT = "🔧 群机器人 [总开关]" + + def __new__(cls, *args): + if len(args) == 2: + value, description = args + elif len(args) == 1 and isinstance(args[0], tuple): + value, description = args[0] + else: + raise TypeError(f"Invalid arguments for Feature.__new__: {args}") + + obj = object.__new__(cls) + obj._value_ = value + obj.description = description + return obj + + def __str__(self): + return self.description + + @classmethod + def get_max_value(cls) -> int: + return max(member.value for member in cls) + + _dynamic_features = {} + + @classmethod + def register_feature(cls, key: str, description: str) -> 'Feature': + if key in cls._dynamic_features: + return cls._dynamic_features[key] + + new_value = cls.get_max_value() + 1 + new_feature = object.__new__(cls) + new_feature._value_ = new_value + new_feature.description = description + cls._dynamic_features[key] = new_feature + return new_feature + + @classmethod + def get_feature(cls, key: str) -> 'Feature': + return cls._dynamic_features.get(key) + + @classmethod + def get_all_features(cls) -> List['Feature']: + return list(cls) + list(cls._dynamic_features.values()) + + @classmethod + def _missing_(cls, value): + if isinstance(value, int): + for member in cls: + if member.value == value: + return member + for feature in cls._dynamic_features.values(): + if feature.value == value: + return feature + return None + + + +if __name__ == '__main__': + # 测试新增功能 + print("当前功能列表：") + for feature in Feature.get_all_features(): + print(f"{feature.value} - {feature.description}") + + # 测试注册新功能 + print("\n注册新功能...") + new_feature = Feature.register_feature("TEST_FEATURE", "🧪 测试功能 [测试]") + print(f"新功能: {new_feature.value} - {new_feature.description}") + + # 验证新功能是否添加成功 + print("\n更新后的功能列表：") + for feature in Feature.get_all_features(): + print(f"{feature.value} - {feature.description}") + + # 测试重复注册 + print("\n测试重复注册...") + same_feature = Feature.register_feature("TEST_FEATURE", "🧪 测试功能 [测试]") + print(f"重复注册结果: {same_feature.value} - {same_feature.description}") + + # 测试获取功能 + print("\n测试获取功能...") + retrieved_feature = Feature.get_feature("TEST_FEATURE") + print(f"获取到的功能: {retrieved_feature.value} - {retrieved_feature.description}") \ No newline at end of file diff --git a/utils/robot_cmd/robot_command.py b/utils/robot_cmd/robot_command.py index 4ea2152..6e9b3ca 100644 --- a/utils/robot_cmd/robot_command.py +++ b/utils/robot_cmd/robot_command.py @@ -29,39 +29,77 @@ class PermissionStatus(Enum): 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 # 添加描述 + obj.description = description return obj def __str__(self): return self.description + @classmethod + def get_max_value(cls) -> int: + """获取当前最大的枚举值""" + return max([member.value for member in cls]) + + @classmethod + def register_feature(cls, key: str, description: str) -> 'Feature': + """注册新的功能权限 + + Args: + key: 功能键名 + description: 功能描述 + + Returns: + Feature: 新注册的功能枚举 + """ + # 检查是否已经注册过 + if key in cls._member_map_: + return cls._member_map_[key] + + # 获取当前最大的枚举值并加1 + new_value = cls.get_max_value() + 1 + + # 创建新的枚举成员 + new_feature = cls(new_value, description) + cls._member_map_[key] = new_feature + cls._member_names_.append(key) + new_feature._name_ = key + + return new_feature + + @classmethod + def get_feature(cls, key: str) -> 'Feature': + """获取已注册的功能 + + Args: + key: 功能键名 + + Returns: + Feature: 功能枚举实例 + """ + return cls._member_map_.get(key) + + @classmethod + def get_all_features(cls) -> List['Feature']: + """获取所有功能 + + Returns: + List[Feature]: 所有功能列表 + """ + return list(cls) + + @classmethod + def _missing_(cls, value): + """处理未找到的枚举值""" + if isinstance(value, int): + for member in cls: + if member.value == value: + return member + return None + def get_redis_connection(): # 初始化时加载本地缓存 @@ -181,7 +219,7 @@ class GroupBotManager: if feature_str.isdigit(): feature_num = int(feature_str) try: - feature = Feature(feature_num) # 使用枚举序号查找功能 + feature = Feature(feature_num, "") # 使用枚举序号查找功能 except ValueError: return "无效的功能序号" else: @@ -294,5 +332,56 @@ def simulate_commands(): if __name__ == '__main__': - # 执行模拟命令 - simulate_commands() + # 测试新增功能 + print("当前功能列表：") + for feature in Feature.get_all_features(): + print(f"{feature.name}: {feature.value} - {feature.description}") + + # 测试注册新功能 + print("\n注册新功能...") + new_feature = Feature.register_feature("TEST_FEATURE", "🧪 测试功能 [测试]") + print(f"新功能: {new_feature.name} = {new_feature.value} - {new_feature.description}") + + # 验证新功能是否添加成功 + print("\n更新后的功能列表：") + for feature in Feature.get_all_features(): + print(f"{feature.name}: {feature.value} - {feature.description}") + + # 测试重复注册 + print("\n测试重复注册...") + same_feature = Feature.register_feature("TEST_FEATURE", "🧪 测试功能 [测试]") + print(f"重复注册结果: {same_feature.name} = {same_feature.value} - {same_feature.description}") + + # 测试获取功能 + print("\n测试获取功能...") + retrieved_feature = Feature.get_feature("TEST_FEATURE") + print(f"获取到的功能: {retrieved_feature.name} = {retrieved_feature.value} - {retrieved_feature.description}") + + # 测试通过值查找功能 + print("\n测试通过值查找功能...") + feature_by_value = Feature(1, "") # 查找 ROBOT + print(f"通过值1查找: {feature_by_value.name} = {feature_by_value.value} - {feature_by_value.description}") + feature_by_value = Feature(2, "") # 查找 TEST_FEATURE + print(f"通过值2查找: {feature_by_value.name} = {feature_by_value.value} - {feature_by_value.description}") + + # 测试枚举属性访问 + print("\n测试枚举属性访问...") + print(f"ROBOT.name: {Feature.ROBOT.name}") + print(f"ROBOT.value: {Feature.ROBOT.value}") + print(f"ROBOT.description: {Feature.ROBOT.description}") + print(f"TEST_FEATURE.name: {Feature.TEST_FEATURE.name}") + print(f"TEST_FEATURE.value: {Feature.TEST_FEATURE.value}") + print(f"TEST_FEATURE.description: {Feature.TEST_FEATURE.description}") + + # 测试枚举比较 + print("\n测试枚举比较...") + print(f"ROBOT == TEST_FEATURE: {Feature.ROBOT == Feature.TEST_FEATURE}") + print(f"ROBOT != TEST_FEATURE: {Feature.ROBOT != Feature.TEST_FEATURE}") + print(f"ROBOT == ROBOT: {Feature.ROBOT == Feature.ROBOT}") + print(f"TEST_FEATURE == TEST_FEATURE: {Feature.TEST_FEATURE == Feature.TEST_FEATURE}") + + # 测试枚举迭代 + print("\n测试枚举迭代...") + print("所有功能:") + for feature in Feature: + print(f" {feature.name}: {feature.value} - {feature.description}")