mirror of
https://github.com/PaiGramTeam/PamGram.git
synced 2024-11-16 12:02:16 +00:00
166 lines
8.0 KiB
Python
166 lines
8.0 KiB
Python
from typing import Optional, Tuple
|
|
|
|
from telegram import Update
|
|
from telegram.constants import ChatAction
|
|
from telegram.ext import CallbackContext
|
|
from telegram.ext import filters
|
|
from telegram.helpers import create_deep_linked_url
|
|
|
|
from core.config import config
|
|
from core.handler.callbackqueryhandler import CallbackQueryHandler
|
|
from core.plugin import Plugin, handler
|
|
from core.services.task.models import Task as SignUser, TaskStatusEnum
|
|
from core.services.task.services import SignServices
|
|
from core.services.users.services import UserAdminService
|
|
from plugins.tools.genshin import GenshinHelper, CookiesNotFoundError, PlayerNotFoundError
|
|
from plugins.tools.sign import SignSystem, NeedChallenge
|
|
from utils.log import logger
|
|
|
|
|
|
class Sign(Plugin):
|
|
"""每日签到"""
|
|
|
|
CHECK_SERVER, COMMAND_RESULT = range(10400, 10402)
|
|
|
|
def __init__(
|
|
self,
|
|
genshin_helper: GenshinHelper,
|
|
sign_service: SignServices,
|
|
user_admin_service: UserAdminService,
|
|
sign_system: SignSystem,
|
|
):
|
|
self.user_admin_service = user_admin_service
|
|
self.sign_service = sign_service
|
|
self.sign_system = sign_system
|
|
self.genshin_helper = genshin_helper
|
|
|
|
async def _process_auto_sign(self, user_id: int, chat_id: int, method: str) -> str:
|
|
try:
|
|
await self.genshin_helper.get_genshin_client(user_id)
|
|
except (PlayerNotFoundError, CookiesNotFoundError):
|
|
return "未查询到账号信息，请先私聊彦卿绑定账号"
|
|
user: SignUser = await self.sign_service.get_by_user_id(user_id)
|
|
if user:
|
|
if method == "关闭":
|
|
await self.sign_service.remove(user)
|
|
return "关闭自动签到成功"
|
|
if method == "开启":
|
|
if user.chat_id == chat_id:
|
|
return "自动签到已经开启过了"
|
|
user.chat_id = chat_id
|
|
user.status = TaskStatusEnum.STATUS_SUCCESS
|
|
await self.sign_service.update(user)
|
|
return "修改自动签到通知对话成功"
|
|
elif method == "关闭":
|
|
return "您还没有开启自动签到"
|
|
elif method == "开启":
|
|
user = self.sign_service.create(user_id, chat_id, TaskStatusEnum.STATUS_SUCCESS)
|
|
await self.sign_service.add(user)
|
|
return "开启自动签到成功"
|
|
|
|
@handler.command(command="sign", cookie=True, block=False)
|
|
@handler.message(filters=filters.Regex("^每日签到(.*)"), cookie=True, block=False)
|
|
@handler.command(command="start", filters=filters.Regex("sign$"), block=False)
|
|
async def command_start(self, update: Update, context: CallbackContext) -> None:
|
|
user_id = await self.get_real_user_id(update)
|
|
message = update.effective_message
|
|
args = self.get_args(context)
|
|
validate: Optional[str] = None
|
|
if len(args) >= 1:
|
|
msg = None
|
|
if args[0] == "开启自动签到":
|
|
if await self.user_admin_service.is_admin(user_id):
|
|
msg = await self._process_auto_sign(user_id, message.chat_id, "开启")
|
|
else:
|
|
msg = await self._process_auto_sign(user_id, user_id, "开启")
|
|
elif args[0] == "关闭自动签到":
|
|
msg = await self._process_auto_sign(user_id, message.chat_id, "关闭")
|
|
else:
|
|
validate = args[0]
|
|
if msg:
|
|
self.log_user(update, logger.info, "自动签到命令请求 || 参数 %s", args[0])
|
|
reply_message = await message.reply_text(msg)
|
|
if filters.ChatType.GROUPS.filter(message):
|
|
self.add_delete_message_job(reply_message, delay=30)
|
|
self.add_delete_message_job(message, delay=30)
|
|
return
|
|
self.log_user(update, logger.info, "每日签到命令请求")
|
|
if filters.ChatType.GROUPS.filter(message):
|
|
self.add_delete_message_job(message)
|
|
try:
|
|
client = await self.genshin_helper.get_genshin_client(user_id)
|
|
await message.reply_chat_action(ChatAction.TYPING)
|
|
_, challenge = await self.sign_system.get_challenge(client.player_id)
|
|
if validate:
|
|
_, challenge = await self.sign_system.get_challenge(client.player_id)
|
|
if challenge:
|
|
sign_text = await self.sign_system.start_sign(client, challenge=challenge, validate=validate)
|
|
else:
|
|
reply_message = await message.reply_text("请求已经过期", allow_sending_without_reply=True)
|
|
if filters.ChatType.GROUPS.filter(reply_message):
|
|
self.add_delete_message_job(reply_message)
|
|
return
|
|
else:
|
|
sign_text = await self.sign_system.start_sign(client)
|
|
reply_message = await message.reply_text(sign_text, allow_sending_without_reply=True)
|
|
if filters.ChatType.GROUPS.filter(reply_message):
|
|
self.add_delete_message_job(reply_message)
|
|
except NeedChallenge as exc:
|
|
button = await self.sign_system.get_challenge_button(
|
|
context.bot.username,
|
|
exc.uid,
|
|
user_id,
|
|
exc.gt,
|
|
exc.challenge,
|
|
not filters.ChatType.PRIVATE.filter(message),
|
|
)
|
|
reply_message = await message.reply_text(
|
|
"签到失败，触发验证码风控，请尝试点击下方按钮重新签到",
|
|
allow_sending_without_reply=True,
|
|
reply_markup=button,
|
|
)
|
|
if filters.ChatType.GROUPS.filter(reply_message):
|
|
self.add_delete_message_job(reply_message)
|
|
|
|
@handler.command(command="start", filters=filters.Regex(r" challenge_sign_(.*)"), block=False)
|
|
async def command_challenge(self, update: Update, context: CallbackContext) -> None:
|
|
user = update.effective_user
|
|
message = update.effective_message
|
|
args = context.args
|
|
_data = args[0].split("_")
|
|
validate = _data[2]
|
|
logger.info("用户 %s[%s] 通过start命令 进入签到流程", user.full_name, user.id)
|
|
try:
|
|
async with self.genshin_helper.genshin(user.id) as client:
|
|
await message.reply_chat_action(ChatAction.TYPING)
|
|
_, challenge = await self.sign_system.get_challenge(client.player_id)
|
|
if not challenge:
|
|
await message.reply_text("验证请求已过期。", allow_sending_without_reply=True)
|
|
return
|
|
sign_text = await self.sign_system.start_sign(client, challenge=challenge, validate=validate)
|
|
await message.reply_text(sign_text, allow_sending_without_reply=True)
|
|
except NeedChallenge:
|
|
await message.reply_text("回调错误，请重新签到", allow_sending_without_reply=True)
|
|
|
|
@handler(CallbackQueryHandler, pattern=r"^sign\|", block=False)
|
|
async def sign_gen_link(self, update: Update, context: CallbackContext) -> None:
|
|
callback_query = update.callback_query
|
|
user = callback_query.from_user
|
|
|
|
async def get_sign_callback(callback_query_data: str) -> Tuple[int, int]:
|
|
_data = callback_query_data.split("|")
|
|
_user_id = int(_data[1])
|
|
_uid = int(_data[2])
|
|
logger.debug("get_sign_callback 函数返回 user_id[%s] uid[%s]", _user_id, _uid)
|
|
return _user_id, _uid
|
|
|
|
user_id, uid = await get_sign_callback(callback_query.data)
|
|
if user.id != user_id:
|
|
await callback_query.answer(text="这不是你的按钮！\n" + config.notice.user_mismatch, show_alert=True)
|
|
return
|
|
_, challenge = await self.sign_system.get_challenge(uid)
|
|
if not challenge:
|
|
await callback_query.answer(text="验证请求已经过期，请重新发起签到！", show_alert=True)
|
|
return
|
|
await callback_query.answer(url=create_deep_linked_url(context.bot.username, "sign"))
|