mirror of
https://github.com/PaiGramTeam/MibooGram.git
synced 2024-11-16 12:51:45 +00:00
32 lines
1.4 KiB
Python
32 lines
1.4 KiB
Python
from telegram import Update
|
|
from telegram.ext import CommandHandler, CallbackContext
|
|
|
|
from core.plugin import Plugin, handler
|
|
from core.sign import SignServices
|
|
from utils.decorators.admins import bot_admins_rights_check
|
|
from utils.log import logger
|
|
|
|
|
|
class SignStatus(Plugin):
|
|
def __init__(self, sign_service: SignServices = None):
|
|
self.sign_service = sign_service
|
|
|
|
@staticmethod
|
|
async def get_sign_status(sign_service: SignServices) -> str:
|
|
sign_db = await sign_service.get_all()
|
|
names = ["签到成功", "Cookie 无效", "提前签到", "触发验证码", "API异常", "请求超时", "请求失败", "通知失败"]
|
|
values = [0, 0, 0, 0, 0, 0, 0, 0]
|
|
for sign in sign_db:
|
|
values[sign.status.value] += 1
|
|
text = f"<b>自动签到统计信息</b>\n\n总人数：<code>{len(sign_db)}</code>\n"
|
|
return text + "\n".join(f"{name}: <code>{value}</code>" for name, value in zip(names, values))
|
|
|
|
@handler(CommandHandler, command="sign_status", block=False)
|
|
@bot_admins_rights_check
|
|
async def sign_status(self, update: Update, _: CallbackContext):
|
|
user = update.effective_user
|
|
logger.info(f"用户 {user.full_name}[{user.id}] sign_status 命令请求")
|
|
message = update.effective_message
|
|
text = await self.get_sign_status(self.sign_service)
|
|
await message.reply_text(text, parse_mode="html", quote=True)
|