liuwei_x_x/abot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

动态auth完成。

Browse Source
This commit is contained in:
liuwei
2025-06-09 15:20:48 +08:00
parent 1f262d0399
commit f994f5cbbb
7 changed files with 6 additions and 221 deletions
Download Patch File Download Diff File Expand all files Collapse all files

118
base/auth/permission.py
View File

@@ -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

98
base/auth/test_permission.py
View File

@@ -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()

2
plugins/game_task/main.py
View File

@@ -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
# 获取群内所有玩家

2
plugins/jd_sign_token/main.py
View File

@@ -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和备注

4
plugins/message_summary/main.py
View File

@@ -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")

2
plugins/point_trade/main.py
View File

@@ -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, "没有权限"
# 处理不同的命令

1
plugins/robot_menu/main.py
View File

@@ -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")