diff --git a/modules/apihelper/client/components/signin.py b/modules/apihelper/client/components/signin.py index 2756d113..beb769e2 100644 --- a/modules/apihelper/client/components/signin.py +++ b/modules/apihelper/client/components/signin.py @@ -91,7 +91,7 @@ class SignIn: async def create_login_data(self) -> str: self.device_id = get_device_id("".join(random.choices((ascii_letters + digits), k=64))) - data = {"app_id": "4", "device": self.device_id} + data = {"app_id": "8", "device": self.device_id} res = await self.client.post(self.QRCODE_GEN_API, json=data) res_json = res.json() url = res_json.get("data", {}).get("url", "") @@ -124,7 +124,7 @@ class SignIn: return True async def check_login(self): - data = {"app_id": "4", "ticket": self.ticket, "device": self.device_id} + data = {"app_id": "8", "ticket": self.ticket, "device": self.device_id} for _ in range(20): await asyncio.sleep(10) res = await self.client.post(self.QRCODE_GET_API, json=data) diff --git a/plugins/genshin/reg_time.py b/plugins/genshin/reg_time.py new file mode 100644 index 00000000..821a2d37 --- /dev/null +++ b/plugins/genshin/reg_time.py @@ -0,0 +1,114 @@ +from datetime import datetime + + +from genshin import Client +from genshin.client.routes import InternationalRoute # noqa F401 +from genshin.utility import recognize_genshin_server, get_ds_headers +from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup +from telegram.constants import ParseMode +from telegram.ext import CommandHandler, CallbackContext, MessageHandler +from telegram.ext import filters +from telegram.helpers import create_deep_linked_url + +from core.base.redisdb import RedisDB +from core.baseplugin import BasePlugin +from core.cookies import CookiesService +from core.cookies.error import CookiesNotFoundError +from core.plugin import Plugin, handler +from core.user import UserService +from core.user.error import UserNotFoundError +from utils.decorators.error import error_callable +from utils.decorators.restricts import restricts +from utils.genshin import fetch_hk4e_token_by_cookie, recognize_genshin_game_biz +from utils.helpers import get_genshin_client +from utils.log import logger + +try: + import ujson as jsonlib + +except ImportError: + import json as jsonlib + +REG_TIME_URL = InternationalRoute( + overseas="https://sg-hk4e-api.hoyoverse.com/event/e20220928anniversary/game_data", + chinese="https://hk4e-api.mihoyo.com/event/e20220928anniversary/game_data", +) + + +class RegTimePlugin(Plugin, BasePlugin): + """查询原神注册时间""" + + def __init__( + self, + user_service: UserService = None, + cookie_service: CookiesService = None, + redis: RedisDB = None, + ): + self.cache = redis.client + self.cache_key = "plugin:reg_time:" + self.user_service = user_service + self.cookie_service = cookie_service + + @staticmethod + async def get_reg_time(client: Client) -> str: + """获取原神注册时间""" + await fetch_hk4e_token_by_cookie(client) + url = REG_TIME_URL.get_url(client.region) + params = { + "game_biz": recognize_genshin_game_biz(client.uid), + "lang": "zh-cn", + "badge_uid": client.uid, + "badge_region": recognize_genshin_server(client.uid), + } + headers = get_ds_headers( + client.region, + params=params, + lang="zh-cn", + ) + data = await client.cookie_manager.request(url, method="GET", params=params, headers=headers) + if time := jsonlib.loads(data.get("data", "{}")).get("1", 0): + return datetime.fromtimestamp(time).strftime("%Y-%m-%d %H:%M:%S") + raise RegTimePlugin.NotFoundRegTimeError + + async def get_reg_time_from_cache(self, client: Client) -> str: + """从缓存中获取原神注册时间""" + if reg_time := await self.cache.get(f"{self.cache_key}{client.uid}"): + return reg_time.decode("utf-8") + reg_time = await self.get_reg_time(client) + await self.cache.set(f"{self.cache_key}{client.uid}", reg_time) + return reg_time + + @handler(CommandHandler, command="reg_time", block=False) + @handler(MessageHandler, filters=filters.Regex("^原神账号注册时间$"), block=False) + @restricts() + @error_callable + async def command_start(self, update: Update, context: CallbackContext) -> None: + message = update.effective_message + user = update.effective_user + logger.info("用户 %s[%s] 原神注册时间命令请求", user.full_name, user.id) + try: + client = await get_genshin_client(user.id) + game_uid = client.uid + reg_time = await self.get_reg_time_from_cache(client) + await message.reply_text(f"你的原神账号 [{game_uid}] 注册时间为:{reg_time}") + except (UserNotFoundError, CookiesNotFoundError): + buttons = [[InlineKeyboardButton("点我绑定账号", url=create_deep_linked_url(context.bot.username, "set_cookie"))]] + if filters.ChatType.GROUPS.filter(message): + reply_msg = await message.reply_text( + "此功能需要绑定cookie后使用,请先私聊派蒙绑定账号", + reply_markup=InlineKeyboardMarkup(buttons), + parse_mode=ParseMode.HTML, + ) + self._add_delete_message_job(context, reply_msg.chat_id, reply_msg.message_id, 30) + self._add_delete_message_job(context, message.chat_id, message.message_id, 30) + else: + await message.reply_text( + "此功能需要绑定cookie后使用,请先私聊派蒙进行绑定", + parse_mode=ParseMode.HTML, + reply_markup=InlineKeyboardMarkup(buttons), + ) + except RegTimePlugin.NotFoundRegTimeError: + await message.reply_text("未找到你的原神账号 [{game_uid}] 注册时间,仅限 2022 年 10 月 之前注册的账号") + + class NotFoundRegTimeError(Exception): + """未找到注册时间""" diff --git a/utils/genshin.py b/utils/genshin.py index 65d135c4..2610d65f 100644 --- a/utils/genshin.py +++ b/utils/genshin.py @@ -1,11 +1,16 @@ from typing import Optional from genshin import Client +from genshin.client.routes import InternationalRoute # noqa F401 from genshin.utility import recognize_genshin_server from modules.apihelper.utility.helpers import hex_digest, get_ds AUTHKEY_API = "https://api-takumi.mihoyo.com/binding/api/genAuthKey" +HK4E_LOGIN_URL = InternationalRoute( + overseas="https://sg-public-api.hoyoverse.com/common/badge/v1/login/account", + chinese="https://api-takumi.mihoyo.com/common/badge/v1/login/account", +) GACHA_HEADERS = { "User-Agent": "okhttp/4.8.0", "x-rpc-sys_version": "12", @@ -18,18 +23,22 @@ GACHA_HEADERS = { } +def recognize_genshin_game_biz(game_uid: int) -> str: + return "hk4e_cn" if game_uid < 600000000 else "hk4e_global" + + async def get_authkey_by_stoken(client: Client) -> Optional[str]: """通过 stoken 获取 authkey""" headers = GACHA_HEADERS.copy() json = { "auth_appid": "webview_gacha", - "game_biz": "hk4e_cn", + "game_biz": recognize_genshin_game_biz(client.uid), "game_uid": client.uid, "region": recognize_genshin_server(client.uid), } device_id = hex_digest(str(client.uid)) headers["x-rpc-device_id"] = device_id - device = "Paimon Build " + device_id[0:5] + device = f"Paimon Build {device_id[:5]}" headers["x-rpc-device_name"] = device headers["x-rpc-device_model"] = device app_version, client_type, ds_sign = get_ds() @@ -38,3 +47,18 @@ async def get_authkey_by_stoken(client: Client) -> Optional[str]: headers["ds"] = ds_sign data = await client.cookie_manager.request(AUTHKEY_API, method="POST", json=json, headers=headers) return data.get("authkey") + + +async def fetch_hk4e_token_by_cookie(client: Client) -> None: + """通过 cookie_token 获取 hk4e_token 保存到 client""" + url = HK4E_LOGIN_URL.get_url(client.region) + headers = { + "Content-Type": "application/json;charset=UTF-8", + } + json = { + "game_biz": recognize_genshin_game_biz(client.uid), + "lang": "zh-cn", + "uid": str(client.uid), + "region": recognize_genshin_server(client.uid), + } + await client.cookie_manager.request(url, method="POST", json=json, headers=headers) diff --git a/utils/patch/genshin.py b/utils/patch/genshin.py index 885b96ea..2d317518 100644 --- a/utils/patch/genshin.py +++ b/utils/patch/genshin.py @@ -1,4 +1,6 @@ +import asyncio import typing +import warnings import aiohttp.typedefs import genshin # pylint: disable=W0406 @@ -11,18 +13,63 @@ from modules.apihelper.utility.helpers import get_ds, get_ua, get_device_id, hex from utils.patch.methods import patch, patchable DEVICE_ID = get_device_id() +UPDATE_CHARACTERS = False def get_account_mid_v2(cookies: typing.Dict[str, str]) -> typing.Optional[str]: - for name, value in cookies.items(): - if name == "account_mid_v2": - return value - - return None + return next( + (value for name, value in cookies.items() if name == "account_mid_v2"), + None, + ) @patch(genshin.client.components.calculator.CalculatorClient) # noqa class CalculatorClient: + @patchable + async def request_calculator( + self, + endpoint: str, + *, + method: str = "POST", + lang: typing.Optional[str] = None, + params: typing.Optional[typing.Mapping[str, typing.Any]] = None, + data: typing.Optional[typing.Mapping[str, typing.Any]] = None, + headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None, + **kwargs: typing.Any, + ) -> typing.Mapping[str, typing.Any]: + global UPDATE_CHARACTERS + params = dict(params or {}) + headers = dict(headers or {}) + + base_url = routes.CALCULATOR_URL.get_url(self.region) + url = base_url / endpoint + + if method == "GET": + params["lang"] = lang or self.lang + data = None + else: + data = dict(data or {}) + data["lang"] = lang or self.lang + + if self.region == types.Region.CHINESE: + headers["referer"] = str(routes.CALCULATOR_REFERER_URL.get_url()) + + update_task = ( + None + if UPDATE_CHARACTERS + else asyncio.create_task(utility.update_characters_any(lang or self.lang, lenient=True)) + ) + data = await self.request(url, method=method, params=params, data=data, headers=headers, **kwargs) + + if update_task: + try: + await update_task + UPDATE_CHARACTERS = True + except Exception as e: + warnings.warn(f"Failed to update characters: {e!r}") + + return data + @patchable async def get_character_details( self,