diff --git a/base/auth/permission.py b/base/auth/permission.py deleted file mode 100644 index 50a96d7..0000000 --- a/base/auth/permission.py +++ /dev/null @@ -1,118 +0,0 @@ -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 deleted file mode 100644 index d017388..0000000 --- a/base/auth/test_permission.py +++ /dev/null @@ -1,98 +0,0 @@ -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/plugins/game_task/main.py b/plugins/game_task/main.py index b377bd2..3cc2f90 100644 --- a/plugins/game_task/main.py +++ b/plugins/game_task/main.py @@ -533,7 +533,7 @@ class GameTaskPlugin(MessagePluginInterface): groups = self.encyclopedia_db.get_all_groups() for group in groups: # 检查权限 - if GroupBotManager.get_group_permission(group, Feature.TASK_GAME) == PermissionStatus.DISABLED: + if GroupBotManager.get_group_permission(group,self.feature) == PermissionStatus.DISABLED: continue # 获取群内所有玩家 diff --git a/plugins/jd_sign_token/main.py b/plugins/jd_sign_token/main.py index 4c50157..b94c71f 100644 --- a/plugins/jd_sign_token/main.py +++ b/plugins/jd_sign_token/main.py @@ -263,7 +263,7 @@ class JDTokenPlugin(MessagePluginInterface): return False, "命令格式错误" # 检查权限 - if roomid and gbm.get_group_permission(roomid, Feature.JD_TOKEN) == PermissionStatus.DISABLED: + if roomid and gbm.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED: return False, "没有权限" # 提取token和备注 diff --git a/plugins/message_summary/main.py b/plugins/message_summary/main.py index 9dd2d1b..671ee03 100644 --- a/plugins/message_summary/main.py +++ b/plugins/message_summary/main.py @@ -100,7 +100,7 @@ class MessageSummaryPlugin(MessagePluginInterface): return True @plugin_stats_decorator(plugin_name="群聊总结") - @plugin_points_cost(10, "群聊总结消耗积分", Feature.SUMMARY_CAPABILITY) + @plugin_points_cost(10, "群聊总结消耗积分", FEATURE_KEY) async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]: """处理消息""" try: @@ -122,7 +122,7 @@ class MessageSummaryPlugin(MessagePluginInterface): return False, None # 权限判断 gbm: GroupBotManager = message.get("gbm") - if gbm and gbm.get_group_permission(group_id, Feature.SUMMARY_CAPABILITY) == PermissionStatus.DISABLED: + if gbm and gbm.get_group_permission(group_id, self.feature) == PermissionStatus.DISABLED: return False, None # 从消息历史中获取群聊记录 all_contacts: dict = message.get("all_contacts") diff --git a/plugins/point_trade/main.py b/plugins/point_trade/main.py index 23d19ff..4dbb30f 100644 --- a/plugins/point_trade/main.py +++ b/plugins/point_trade/main.py @@ -149,7 +149,7 @@ class PointTradePlugin(MessagePluginInterface): self.revoke: MessageAutoRevoke = message.get("revoke") # 检查权限 - if roomid and gbm.get_group_permission(roomid, Feature.POINT_TRADE) == PermissionStatus.DISABLED: + if roomid and gbm.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED: return False, "没有权限" # 处理不同的命令 diff --git a/plugins/robot_menu/main.py b/plugins/robot_menu/main.py index 84740de..8a34310 100644 --- a/plugins/robot_menu/main.py +++ b/plugins/robot_menu/main.py @@ -155,6 +155,7 @@ class RobotMenuPlugin(MessagePluginInterface): self.LOG.debug(f"插件执行： {self.name}：{content}") sender = message.get("sender") roomid = message.get("roomid", "") + command = content.split(" ")[0] gbm: GroupBotManager = message.get("gbm") bot: WechatAPIClient = message.get("bot") revoke: MessageAutoRevoke = message.get("revoke")