✨ Support Query Game Registration Time

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)

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(
+                    "此功能需要绑定<code>cookie</code>后使用，请先私聊派蒙绑定账号",
+                    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(
+                    "此功能需要绑定<code>cookie</code>后使用，请先私聊派蒙进行绑定",
+                    parse_mode=ParseMode.HTML,
+                    reply_markup=InlineKeyboardMarkup(buttons),
+                )
+        except RegTimePlugin.NotFoundRegTimeError:
+            await message.reply_text("未找到你的原神账号 [{game_uid}] 注册时间，仅限 2022 年 10 月 之前注册的账号")
+
+    class NotFoundRegTimeError(Exception):
+        """未找到注册时间"""

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)

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,