PaiGramTeam/PaiGram
3
0
Fork 0
mirror of https://github.com/PaiGramTeam/PaiGram.git synced 2024-11-21 22:58:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

✨ 支持通过 Stoken 自动刷新抽卡记录

Browse Source
This commit is contained in:
omg-xtao 2022-10-11 12:00:55 +08:00 committed by GitHub
parent be8bca8011
commit 4660b34db4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 125 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
core/config.py
View File

@ -22,21 +22,21 @@ dotenv.load_dotenv()
class BotConfig(BaseSettings):
debug: bool = False
db_host: str
db_port: int
db_username: str
db_password: str
db_database: str
db_host: str = ""
db_port: int = 0
db_username: str = ""
db_password: str = ""
db_database: str = ""
redis_host: str
redis_port: int
redis_db: int
redis_host: str = ""
redis_port: int = 0
redis_db: int = 0
bot_token: str
error_notification_chat_id: str
bot_token: str = ""
error_notification_chat_id: str = ""
api_id: Optional[int]
api_hash: Optional[str]
api_id: Optional[int] = None
api_hash: Optional[str] = None
channels: List["ConfigChannel"] = []
admins: List["ConfigUser"] = []

2
metadata/shortname.py
View File

@ -9,7 +9,7 @@ __all__ = ["roles", "weapons", "roleToId", "roleToName", "weaponToName", "weapon
# noinspection SpellCheckingInspection
roles = {
20000000: ["主角", "旅行者", "卑鄙的外乡人", "荣誉骑士", "爷", "风主", "岩主", "雷主", "草主", "履刑者", "抽卡不歪真君"],
10000002: ["神里绫华", "Ayaka", "ayaka", "Kamisato Ayaka", "神里", "绫华", "神里凌华", "凌华", "白鹭公主", "神里大小 姐"],
10000002: ["神里绫华", "Ayaka", "ayaka", "Kamisato Ayaka", "神里", "绫华", "神里凌华", "凌华", "白鹭公主", "神里大小姐"],
10000003: ["琴", "Jean", "jean", "团长", "代理团长", "琴团长", "蒲公英骑士"],
10000005: ["空", "Aether", "aether", "男主", "男主角", "龙哥", "空哥"],
10000006: ["丽莎", "Lisa", "lisa", "图书管理员", "图书馆管理员", "蔷薇魔女"],

48
modules/apihelper/hyperion.py
View File

@ -1,13 +1,18 @@
import asyncio
import re
import time
from typing import List
from json import JSONDecodeError
from typing import List, Optional
from genshin import Client, InvalidCookies
from genshin.utility.uid import recognize_genshin_server
from genshin.utility.ds import generate_dynamic_secret
from httpx import AsyncClient
from modules.apihelper.base import ArtworkImage, PostInfo
from modules.apihelper.helpers import get_device_id
from modules.apihelper.request.hoyorequest import HOYORequest
from utils.log import logger
from utils.typedefs import JSONDict
@ -169,7 +174,7 @@ class GachaInfo:
class SignIn:
LOGIN_URL = "https://webapi.account.mihoyo.com/Api/login_by_mobilecaptcha"
S_TOKEN_URL = (
"https://api-takumi.mihoyo.com/auth/api/getMultiTokenByLoginTicket?" "login_ticket={0}&token_types=3&uid={1}"
"https://api-takumi.mihoyo.com/auth/api/getMultiTokenByLoginTicket?login_ticket={0}&token_types=3&uid={1}"
)
BBS_URL = "https://api-takumi.mihoyo.com/account/auth/api/webLoginByMobile"
USER_AGENT = (
@ -209,8 +214,21 @@ class SignIn:
"Referer": "https://bbs.mihoyo.com/",
"Accept-Language": "zh-CN,zh-Hans;q=0.9",
}
AUTHKEY_API = "https://api-takumi.mihoyo.com/binding/api/genAuthKey"
GACHA_HEADERS = {
"User-Agent": "okhttp/4.8.0",
"x-rpc-app_version": "2.28.1",
"x-rpc-sys_version": "12",
"x-rpc-client_type": "5",
"x-rpc-channel": "mihoyo",
"x-rpc-device_id": get_device_id(USER_AGENT),
"x-rpc-device_name": "Mi 10",
"x-rpc-device_model": "Mi 10",
"Referer": "https://app.mihoyo.com",
"Host": "api-takumi.mihoyo.com",
}
def __init__(self, phone: int):
def __init__(self, phone: int = 0):
self.phone = phone
self.client = AsyncClient()
self.uid = 0
@ -287,3 +305,27 @@ class SignIn:
self.cookie[k] = v
return "cookie_token" in self.cookie
@staticmethod
async def get_authkey_by_stoken(client: Client) -> Optional[str]:
"""通过 stoken 获取 authkey"""
try:
headers = SignIn.GACHA_HEADERS.copy()
headers["DS"] = generate_dynamic_secret("ulInCDohgEs557j0VsPDYnQaaz6KJcv5")
data = await client.cookie_manager.request(
SignIn.AUTHKEY_API,
method="POST",
json={
"auth_appid": "webview_gacha",
"game_biz": "hk4e_cn",
"game_uid": client.uid,
"region": recognize_genshin_server(client.uid),
},
headers=headers,
)
return data.get("authkey")
except JSONDecodeError:
logger.warning("Stoken 获取 Authkey JSON解析失败")
except InvalidCookies:
logger.warning("Stoken 获取 Authkey 失败 | 用户 Stoken 失效")
return None

59
plugins/genshin/gacha/gacha_log.py
View File

@ -1,4 +1,6 @@
import json
import genshin
from io import BytesIO
from genshin.models import BannerType
@ -9,17 +11,20 @@ from telegram.ext import CallbackContext, CommandHandler, MessageHandler, filter
from core.base.assets import AssetsService
from core.baseplugin import BasePlugin
from core.cookies.error import CookiesNotFoundError
from core.cookies import CookiesService
from core.plugin import Plugin, handler, conversation
from core.template import TemplateService
from core.user import UserService
from core.user.error import UserNotFoundError
from modules.apihelper.gacha_log import GachaLog as GachaLogService
from modules.apihelper.hyperion import SignIn
from utils.bot import get_all_args
from utils.decorators.admins import bot_admins_rights_check
from utils.decorators.error import error_callable
from utils.decorators.restricts import restricts
from utils.helpers import get_genshin_client
from utils.log import logger
from utils.models.base import RegionEnum
INPUT_URL, INPUT_FILE, CONFIRM_DELETE = range(10100, 10103)
@ -28,11 +33,16 @@ class GachaLog(Plugin.Conversation, BasePlugin.Conversation):
"""抽卡记录导入/导出/分析"""
def __init__(
self, template_service: TemplateService = None, user_service: UserService = None, assets: AssetsService = None
self,
template_service: TemplateService = None,
user_service: UserService = None,
assets: AssetsService = None,
cookie_service: CookiesService = None,
):
self.template_service = template_service
self.user_service = user_service
self.assets_service = assets
self.cookie_service = cookie_service
@staticmethod
def from_url_get_authkey(url: str) -> str:
@ -78,7 +88,8 @@ class GachaLog(Plugin.Conversation, BasePlugin.Conversation):
data = data.getvalue().decode("utf-8")
data = json.loads(data)
except Exception as exc:
logger.error(f"文件解析失败：{repr(exc)}")
if not isinstance(exc, UnicodeDecodeError):
logger.error(f"文件解析失败：{repr(exc)}")
await message.reply_text("文件解析失败，请检查文件是否符合 UIGF 标准")
return
await message.reply_chat_action(ChatAction.TYPING)
@ -101,6 +112,7 @@ class GachaLog(Plugin.Conversation, BasePlugin.Conversation):
user = update.effective_user
args = get_all_args(context)
logger.info(f"用户 {user.full_name}[{user.id}] 导入抽卡记录命令请求")
authkey = self.from_url_get_authkey(args[0] if args else "")
if not args:
if message.document:
await self.import_from_file(user, message)
@ -108,18 +120,46 @@ class GachaLog(Plugin.Conversation, BasePlugin.Conversation):
elif message.reply_to_message and message.reply_to_message.document:
await self.import_from_file(user, message, document=message.reply_to_message.document)
return ConversationHandler.END
try:
user_info = await self.user_service.get_user_by_id(user.id)
except UserNotFoundError:
user_info = None
if user_info and user_info.region == RegionEnum.HYPERION:
try:
cookies = await self.cookie_service.get_cookies(user_info.user_id, user_info.region)
except CookiesNotFoundError:
cookies = None
if cookies and cookies.cookies and "stoken" in cookies.cookies:
if stuid := next(
(value for key, value in cookies.cookies.items() if key in ["ltuid", "login_uid"]), None
):
cookies.cookies["stuid"] = stuid
client = genshin.Client(
cookies=cookies.cookies,
game=genshin.types.Game.GENSHIN,
region=genshin.Region.CHINESE,
lang="zh-cn",
uid=user_info.yuanshen_uid,
)
authkey = await SignIn.get_authkey_by_stoken(client)
if not authkey:
await message.reply_text(
"<b>导入祈愿历史记录</b>\n\n"
"1.请发送从其他工具导出的 UIGF JSON 标准的记录文件\n"
"2.你还可以向派蒙发送从游戏中获取到的抽卡记录链接\n\n"
"<b>注意：导入的数据将会与旧数据进行合并。</b>\n"
"获取抽卡记录链接可以参考：https://paimon.moe/wish/import",
"<b>开始导入祈愿历史记录：请通过 https://paimon.moe/wish/import 获取抽卡记录链接后发送给我"
"（非 paimon.moe 导出的文件数据）</b>\n\n"
"> 你还可以向派蒙发送从其他工具导出的 UIGF JSON 标准的记录文件\n"
"> 在绑定 Cookie 时添加 stoken 可能有特殊效果哦（仅限国服）\n"
"<b>注意：导入的数据将会与旧数据进行合并。</b>",
parse_mode="html",
)
return INPUT_URL
authkey = self.from_url_get_authkey(args[0])
text = "小派蒙正在从米哈游服务器获取数据，请稍后"
if not args:
text += "\n\n> 由于你绑定的 Cookie 中存在 stoken ，本次通过 stoken 自动刷新数据"
reply = await message.reply_text(text)
await message.reply_chat_action(ChatAction.TYPING)
data = await self._refresh_user_data(user, authkey=authkey)
await message.reply_text(data)
await reply.edit_text(data)
return ConversationHandler.END
@conversation.state(state=INPUT_URL)
@handler.message(filters=~filters.COMMAND, block=False)
@ -133,6 +173,7 @@ class GachaLog(Plugin.Conversation, BasePlugin.Conversation):
return ConversationHandler.END
authkey = self.from_url_get_authkey(message.text)
reply = await message.reply_text("小派蒙正在从米哈游服务器获取数据，请稍后")
await message.reply_chat_action(ChatAction.TYPING)
text = await self._refresh_user_data(user, authkey=authkey)
await reply.edit_text(text)
return ConversationHandler.END

28
plugins/genshin/sign.py
View File

@ -100,14 +100,14 @@ class Sign(Plugin, BasePlugin):
"x-rpc-seccode": f'{data["data"]["validate"]}|jordan',
}
except JSONDecodeError:
logger.warning("签到ajax自动通过JSON解析失败")
logger.warning("签到 ajax 请求 JSON 解析失败")
except TimeoutException:
logger.warning("签到ajax自动通过请求超时")
logger.warning("签到 ajax 请求超时")
except (KeyError, IndexError):
logger.warning("签到ajax自动通过数据错误")
logger.warning("签到 ajax 请求数据错误")
except RuntimeError:
logger.warning("签到ajax自动通过请求错误")
logger.warning("ajax自动通过失败")
logger.warning("签到 ajax 请求错误")
logger.warning("签到 ajax 请求失败")
if not config.pass_challenge_api:
return None
pass_challenge_params = {
@ -125,12 +125,11 @@ class Sign(Plugin, BasePlugin):
params=pass_challenge_params,
timeout=45,
)
logger.info(f"签到自定义打码平台返回：{resp.text}")
logger.info(f"签到请求返回：{resp.text}")
data = resp.json()
status = data.get("status")
if status is not None:
if status != 0:
logger.error(f"签到自定义打码平台解析错误：{data.get('msg')}")
if status is not None and status != 0:
logger.error(f"签到请求解析错误：{data.get('msg')}")
if data.get("code", 0) != 0:
raise RuntimeError
return {
@ -139,13 +138,13 @@ class Sign(Plugin, BasePlugin):
"x-rpc-seccode": f'{data["data"]["validate"]}|jordan',
}
except JSONDecodeError:
logger.warning("签到自定义打码平台JSON解析失败")
logger.warning("签到请求 JSON 解析失败")
except TimeoutException:
logger.warning("签到自定义打码平台请求超时")
logger.warning("签到请求超时")
except KeyError:
logger.warning("签到自定义打码平台数据错误")
logger.warning("签到请求数据错误")
except RuntimeError:
logger.warning("签到自定义打码平台自动通过失败")
logger.warning("签到请求失败")
return None
@staticmethod
@ -166,7 +165,6 @@ class Sign(Plugin, BasePlugin):
"sign", method="POST", game=Game.GENSHIN, lang="zh-cn"
)
if request_daily_reward and request_daily_reward.get("success", 0) == 1:
# 米游社国内签到自动打码
headers = await Sign.pass_challenge(
request_daily_reward.get("gt", ""),
request_daily_reward.get("challenge", ""),
@ -184,7 +182,7 @@ class Sign(Plugin, BasePlugin):
if request_daily_reward and request_daily_reward.get("success", 0) == 1:
logger.warning(f"UID {client.uid} 签到失败，触发验证码风控")
return f"UID {client.uid} 签到失败，触发验证码风控，请尝试重新签到。"
logger.info(f"UID {client.uid} 通过自动打码签到成功")
logger.info(f"UID {client.uid} 签到成功")
except AlreadyClaimed:
logger.info(f"UID {client.uid} 已经签到")
result = "今天旅行者已经签到过了~"

7
plugins/jobs/sign.py
View File

@ -41,13 +41,12 @@ class SignJob(Plugin):
if not daily_reward_info.signed_in:
request_daily_reward = await client.request_daily_reward("sign", method="POST", game=Game.GENSHIN)
if request_daily_reward and request_daily_reward.get("success", 0) == 1:
# 米游社国内签到自动打码
headers = await Sign.pass_challenge(
request_daily_reward.get("gt", ""),
request_daily_reward.get("challenge", ""),
)
if not headers:
logger.warning(f"UID {client.uid} 签到失败，触发验证码风控 | 打码平台打码失败，请检查")
logger.warning(f"UID {client.uid} 签到失败，触发验证码风控")
raise NeedChallenge
request_daily_reward = await client.request_daily_reward(
"sign",
@ -57,9 +56,9 @@ class SignJob(Plugin):
headers=headers,
)
if request_daily_reward and request_daily_reward.get("success", 0) == 1:
logger.warning(f"UID {client.uid} 签到失败，触发验证码风控 | 打码平台打码失败，请检查")
logger.warning(f"UID {client.uid} 签到失败，触发验证码风控")
raise NeedChallenge
logger.info(f"UID {client.uid} 签到请求 {request_daily_reward} | 通过自动打码签到成功")
logger.info(f"UID {client.uid} 签到请求 {request_daily_reward} | 签到成功")
else:
logger.info(f"UID {client.uid} 签到请求 {request_daily_reward}")
result = "OK"

3
utils/log/_logger.py
View File

@ -35,7 +35,7 @@ from rich.traceback import (
)
from ujson import JSONDecodeError
from core.config import BotConfig
from core.config import config
from utils.const import NOT_SET, PROJECT_ROOT
from utils.log._file import FileIO
from utils.log._style import (
@ -64,7 +64,6 @@ __initialized__ = False
FormatTimeCallable = Callable[[datetime], Text]
config = BotConfig()
logging.addLevelName(5, "TRACE")
logging.addLevelName(25, "SUCCESS")
color_system: Literal["windows", "truecolor"]