测试auth动态化

This commit is contained in:
liuwei
2025-06-09 15:10:54 +08:00
parent 1fcabfd365
commit aaaf555156
10 changed files with 62 additions and 39 deletions

View File

@@ -1,26 +1,29 @@
import os
import random
from datetime import datetime, timedelta
from loguru import logger
import pytz
from typing import Dict, Any, List, Optional, Tuple
from db.connection import DBConnectionManager
import pytz
from loguru import logger
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.revoke.message_auto_revoke import MessageAutoRevoke
from utils.robot_cmd.robot_command import Feature, PermissionStatus, GroupBotManager
from db.connection import DBConnectionManager
from db.sign_in import SignInDB
from db.sign_in_redis import SignInRedisDB
import random
import os
from utils.decorator.plugin_decorators import plugin_stats_decorator
from utils.decorator.points_decorator import points_reward_decorator
from utils.revoke.message_auto_revoke import MessageAutoRevoke
from utils.robot_cmd.robot_command import PermissionStatus, GroupBotManager
from wechat_ipad import WechatAPIClient
class MessageSignPlugin(MessagePluginInterface):
"""签到插件"""
# 功能权限常量
FEATURE_KEY = "SIGN_IN"
FEATURE_DESCRIPTION = "✅ 签到功能 [签到, 每日签到, qd, Qd, QD, 上班, 牛马]"
@property
def name(self) -> str:
return "签到系统"
@@ -207,7 +210,7 @@ class MessageSignPlugin(MessagePluginInterface):
self.bot = message.get("bot")
# 检查权限
if roomid and gbm.get_group_permission(roomid, Feature.SIGNIN) == PermissionStatus.DISABLED:
if roomid and gbm.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED:
return False, "没有权限"
# 处理补签命令
@@ -221,7 +224,7 @@ class MessageSignPlugin(MessagePluginInterface):
return False, "不支持的命令"
# 添加签到处理方法,应用积分奖励装饰器
@points_reward_decorator(calculate_sign_in_points, "checkin", "每日签到奖励", Feature.SIGNIN)
@points_reward_decorator(calculate_sign_in_points, "checkin", "每日签到奖励", FEATURE_KEY)
async def _handle_sign_in(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
"""处理签到请求"""
sender = message.get("sender")