PaiGramTeam/MibooGram
3
0
Fork 0
mirror of https://github.com/PaiGramTeam/MibooGram.git synced 2024-11-25 18:04:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

🎨 Optimize auto-sign login

Browse Source
This commit is contained in:
洛水居室 2023-03-16 11:25:10 +08:00
parent c19eb4679a
commit 6d540b9c66
No known key found for this signature in database
GPG Key ID: C9DE87DA724B88FC
2 changed files with 13 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
plugins/jobs/sign.py
View File

@ -1,25 +1,27 @@
import datetime
from telegram.ext import CallbackContext
from typing import TYPE_CHECKING
from core.plugin import Plugin, job
from plugins.genshin.sign import SignSystem
from plugins.tools.sign import SignJobType
from utils.log import logger
if TYPE_CHECKING:
from telegram.ext import ContextTypes
class SignJob(Plugin):
def __init__(self, sign_system: SignSystem):
self.sign_system = sign_system
@job.run_daily(time=datetime.time(hour=0, minute=1, second=0), name="SignJob")
async def sign(self, context: CallbackContext):
async def sign(self, context: "ContextTypes.DEFAULT_TYPE"):
logger.info("正在执行自动签到")
await self.sign_system.do_sign_job(context, job_type=SignJobType.START)
logger.success("执行自动签到完成")
await self.re_sign(context)
async def re_sign(self, context: CallbackContext):
async def re_sign(self, context: "ContextTypes.DEFAULT_TYPE"):
logger.info("正在执行自动重签")
await self.sign_system.do_sign_job(context, job_type=SignJobType.REDO)
logger.success("执行自动重签完成")

11
plugins/tools/sign.py
View File

@ -4,7 +4,7 @@ import random
import time
from enum import Enum
from json import JSONDecodeError
from typing import Optional, Tuple, List
from typing import Optional, Tuple, List, TYPE_CHECKING
from aiohttp import ClientConnectorError
from genshin import Game, GenshinException, AlreadyClaimed, Client, InvalidCookies
@ -13,7 +13,6 @@ from httpx import AsyncClient, TimeoutException
from telegram import InlineKeyboardButton, InlineKeyboardMarkup
from telegram.constants import ParseMode
from telegram.error import Forbidden, BadRequest
from telegram.ext import CallbackContext
from core.config import config
from core.dependence.redisdb import RedisDB
@ -26,6 +25,9 @@ from modules.apihelper.client.components.verify import Verify
from plugins.tools.genshin import GenshinHelper, CookiesNotFoundError, PlayerNotFoundError
from utils.log import logger
if TYPE_CHECKING:
from telegram.ext import ContextTypes
class SignJobType(Enum):
START = 1
@ -311,7 +313,7 @@ class SignSystem(Plugin):
)
return message
async def do_sign_job(self, context: CallbackContext, job_type: SignJobType):
async def do_sign_job(self, context: "ContextTypes.DEFAULT_TYPE", job_type: SignJobType):
include_status: List[SignStatusEnum] = [
SignStatusEnum.STATUS_SUCCESS,
SignStatusEnum.TIMEOUT_ERROR,
@ -321,6 +323,7 @@ class SignSystem(Plugin):
title = "自动签到"
elif job_type == SignJobType.REDO:
title = "自动重新签到"
include_status.remove(SignStatusEnum.STATUS_SUCCESS)
else:
raise ValueError
sign_list = await self.sign_service.get_all()
@ -344,7 +347,7 @@ class SignSystem(Plugin):
text = "签到失败了呜呜呜 ~ 服务器连接超时 服务器熟啦 ~ "
sign_db.status = SignStatusEnum.TIMEOUT_ERROR
except NeedChallenge:
text = "签到失败，触发验证码风控，自动签到自动关闭"
text = "签到失败，触发验证码风控"
sign_db.status = SignStatusEnum.NEED_CHALLENGE
except PlayerNotFoundError:
logger.info("用户 user_id[%s] 玩家不存在 关闭并移除自动签到", user_id)