测试auth动态化

This commit is contained in:
liuwei
2025-06-09 14:56:07 +08:00
parent 9d15bf965b
commit 1fcabfd365
2 changed files with 36 additions and 46 deletions

View File

@@ -19,10 +19,6 @@ from wechat_ipad.models.appmsg_xml import MUSIC_XML
class MusicPlugin(MessagePluginInterface): class MusicPlugin(MessagePluginInterface):
"""音乐点播插件""" """音乐点播插件"""
# 功能权限常量
FEATURE_KEY = "MUSIC"
FEATURE_DESCRIPTION = "🎵 点歌功能 [点歌, 音乐, 音乐点播, 点播音乐, 音乐点歌]"
@property @property
def name(self) -> str: def name(self) -> str:
return "音乐点播" return "音乐点播"
@@ -47,18 +43,8 @@ class MusicPlugin(MessagePluginInterface):
def commands(self) -> List[str]: def commands(self) -> List[str]:
return self._commands 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): def __init__(self):
super().__init__() super().__init__()
# 注册功能权限
self.feature = self.register_feature()
def initialize(self, context: Dict[str, Any]) -> bool: def initialize(self, context: Dict[str, Any]) -> bool:
"""初始化插件""" """初始化插件"""
@@ -97,27 +83,28 @@ class MusicPlugin(MessagePluginInterface):
return command in self._commands return command in self._commands
@plugin_stats_decorator(plugin_name="点歌") @plugin_stats_decorator(plugin_name="音乐点播")
@plugin_points_cost(2, "点歌消耗积分", FEATURE_KEY) @plugin_points_cost(2, "音乐点播消耗积分", Feature.MUSIC)
async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]: async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
"""处理消息""" """处理消息"""
content = str(message.get("content", "")).strip() content = str(message.get("content", "")).strip()
self.LOG.debug(f"插件执行: {self.name}{content}") self.LOG.debug(f"插件执行: {self.name}{content}")
command = content.split(" ")[0]
sender = message.get("sender") sender = message.get("sender")
roomid = message.get("roomid", "") roomid = message.get("roomid", "")
gbm: GroupBotManager = message.get("gbm") gbm: GroupBotManager = message.get("gbm")
bot: WechatAPIClient = message.get("bot") bot: WechatAPIClient = message.get("bot")
# 检查权限
if roomid and gbm.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED:
return False, "没有权限"
# 检查命令格式 # 检查命令格式
if len(content.split(" ")) == 1: if len(content.split(" ")) == 1:
await bot.send_text_message((roomid if roomid else sender), f"❌命令格式错误!\n{self.command_format}" await bot.send_text_message((roomid if roomid else sender), f"❌命令格式错误!\n{self.command_format}"
, sender) , sender)
return False, "命令格式错误" return False, "命令格式错误"
# 检查权限
if roomid and gbm.get_group_permission(roomid, Feature.MUSIC) == PermissionStatus.DISABLED:
return False, "没有权限"
# 提取歌曲名 # 提取歌曲名
user_song_name = content[len(command):].strip() user_song_name = content[len(command):].strip()

View File

@@ -7,10 +7,8 @@
# 4.群总结能力 #启用群总结 #关闭群总结 # 4.群总结能力 #启用群总结 #关闭群总结
# 5.sehuatang PDF能力 #启用pdf #关闭pdf # 5.sehuatang PDF能力 #启用pdf #关闭pdf
import os import os
from typing import List
import redis
from enum import Enum from enum import Enum
from typing import List
import yaml import yaml
from loguru import logger from loguru import logger
@@ -50,25 +48,30 @@ class Feature(Enum):
Args: Args:
key: 功能键名 key: 功能键名
description: 功能描述 value: 权限值
description: 权限描述
Returns: Returns:
Feature: 新注册的功能枚举 Feature: 新注册的功能权限
""" """
# 检查是否已经注册过 # 检查value是否已存在
if key in cls._member_map_: for feature in cls:
return cls._member_map_[key] if feature.name == key:
logger.warning(f"功能权限值 {key} 已存在,将被覆盖")
feature.description = description
return feature
# 获取当前最大的枚举值并加1 # 创建新的功能权限
new_value = cls.get_max_value() + 1 feature = cls._member_map_.get(key)
if feature is None:
feature = object.__new__(cls)
feature._value_ = cls.get_max_value() + 1
feature._name_ = key # 设置枚举成员名称
feature.description = description
cls._member_map_[key] = feature
cls._member_names_.append(key)
# 创建新的枚举成员 return feature
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 @classmethod
def get_feature(cls, key: str) -> 'Feature': def get_feature(cls, key: str) -> 'Feature':
@@ -336,10 +339,10 @@ if __name__ == '__main__':
print("当前功能列表:") print("当前功能列表:")
for feature in Feature.get_all_features(): for feature in Feature.get_all_features():
print(f"{feature.name}: {feature.value} - {feature.description}") print(f"{feature.name}: {feature.value} - {feature.description}")
print(f"最大VALUE{Feature.get_max_value()}")
# 测试注册新功能 # 测试注册新功能
print("\n注册新功能...") print("\n注册新功能...")
new_feature = Feature.register_feature("TEST_FEATURE", "🧪 测试功能 [测试]") new_feature = Feature.register_feature("TEST_FEATURE", "🧪 测试功能 [测试]")
print(f"新功能: {new_feature.name} = {new_feature.value} - {new_feature.description}") print(f"新功能: {new_feature.name} = {new_feature.value} - {new_feature.description}")
# 验证新功能是否添加成功 # 验证新功能是否添加成功
@@ -358,11 +361,11 @@ if __name__ == '__main__':
print(f"获取到的功能: {retrieved_feature.name} = {retrieved_feature.value} - {retrieved_feature.description}") print(f"获取到的功能: {retrieved_feature.name} = {retrieved_feature.value} - {retrieved_feature.description}")
# 测试通过值查找功能 # 测试通过值查找功能
print("\n测试通过值查找功能...") # print("\n测试通过值查找功能...")
feature_by_value = Feature(1, "") # 查找 ROBOT # feature_by_value = Feature(1, "") # 查找 ROBOT
print(f"通过值1查找: {feature_by_value.name} = {feature_by_value.value} - {feature_by_value.description}") # print(f"通过值1查找: {feature_by_value.name} = {feature_by_value.value} - {feature_by_value.description}")
feature_by_value = Feature(2, "") # 查找 TEST_FEATURE # feature_by_value = Feature(2, "") # 查找 TEST_FEATURE
print(f"通过值2查找: {feature_by_value.name} = {feature_by_value.value} - {feature_by_value.description}") # print(f"通过值2查找: {feature_by_value.name} = {feature_by_value.value} - {feature_by_value.description}")
# 测试枚举属性访问 # 测试枚举属性访问
print("\n测试枚举属性访问...") print("\n测试枚举属性访问...")