diff --git a/modules/apihelper/client/components/authclient.py b/modules/apihelper/client/components/authclient.py deleted file mode 100644 index f3933f85..00000000 --- a/modules/apihelper/client/components/authclient.py +++ /dev/null @@ -1,144 +0,0 @@ -import asyncio -import json -import random -import qrcode - -from io import BytesIO -from string import ascii_letters, digits -from typing import Dict, Union, Optional -from httpx import AsyncClient -from qrcode.image.pure import PyPNGImage - -from ...logger import logger -from ...models.genshin.cookies import CookiesModel -from ...utility.devices import devices_methods -from ...utility.helpers import get_device_id, get_ds - -__all__ = ("AuthClient",) - - -class AuthClient: - player_id: Optional[int] = None - user_id: Optional[int] = None - cookies: Optional[CookiesModel] = None - device_id: Optional[str] = None - - USER_AGENT = ( - "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " - "AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.0 Safari/605.1.15" - ) - PASSPORT_HOST = "passport-api.mihoyo.com" - HK4E_SDK_HOST = "hk4e-sdk.mihoyo.com" - TAKUMI_HOST = "api-takumi.mihoyo.com" - QRCODE_GEN_API = f"https://{HK4E_SDK_HOST}/hk4e_cn/combo/panda/qrcode/fetch" - QRCODE_GET_API = f"https://{HK4E_SDK_HOST}/hk4e_cn/combo/panda/qrcode/query" - GET_COOKIE_ACCOUNT_BY_GAME_TOKEN_API = f"https://{TAKUMI_HOST}/auth/api/getCookieAccountInfoByGameToken" - GET_TOKEN_BY_GAME_LTOKEN_API = f"https://{PASSPORT_HOST}/account/ma-cn-session/app/getTokenByGameToken" - - def __init__( - self, - player_id: Optional[int] = None, - user_id: Optional[int] = None, - cookies: Optional[Union[CookiesModel, dict]] = None, - ): - self.client = AsyncClient() - self.player_id = player_id - if cookies is None: - self.cookies = CookiesModel() - else: - if isinstance(cookies, dict): - self.cookies = CookiesModel(**cookies) - elif isinstance(cookies, CookiesModel): - self.cookies = cookies - else: - raise RuntimeError - if user_id: - self.user_id = user_id - else: - self.user_id = self.cookies.user_id - - async def get_ltoken_by_game_token(self, game_token: str) -> bool: - if self.user_id is None: - return False - data = {"account_id": self.user_id, "game_token": game_token} - headers = { - "x-rpc-aigis": "", - "Content-Type": "application/json", - "Accept": "application/json", - "x-rpc-game_biz": "bbs_cn", - "x-rpc-sys_version": "11", - "x-rpc-device_name": "Chrome 108.0.0.0", - "x-rpc-device_model": "Windows 10 64-bit", - "x-rpc-app_id": "bll8iq97cem8", - "User-Agent": "okhttp/4.8.0", - } - await devices_methods.update_device_headers(self.user_id, headers) - app_version, client_type, ds_sign = get_ds(new_ds=True, data=data) - headers["x-rpc-app_version"] = app_version - headers["x-rpc-client_type"] = client_type - headers["DS"] = ds_sign - res = await self.client.post( - self.GET_TOKEN_BY_GAME_LTOKEN_API, - headers=headers, - json={"account_id": self.user_id, "game_token": game_token}, - ) - ltoken_data = res.json() - self.cookies.ltmid_v2 = ltoken_data["data"]["user_info"]["mid"] - self.cookies.ltoken_v2 = ltoken_data["data"]["token"]["token"] - return True - - async def create_qrcode_login(self) -> tuple[str, str]: - self.device_id = get_device_id("".join(random.choices((ascii_letters + digits), k=64))) - 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", "") - if not url: - return "", "" - ticket = url.split("ticket=")[1] - return url, ticket - - async def _get_cookie_token_data(self, game_token: str, account_id: int) -> Dict: - res = await self.client.get( - self.GET_COOKIE_ACCOUNT_BY_GAME_TOKEN_API, - params={"game_token": game_token, "account_id": account_id}, - ) - return res.json() - - async def _set_cookie_by_game_token(self, data: Dict) -> bool: - game_token = json.loads(data.get("payload", {}).get("raw", "{}")) - if not game_token: - return False - uid = game_token["uid"] - self.user_id = int(uid) - cookie_token_data = await self._get_cookie_token_data(game_token["token"], self.user_id) - await self.get_ltoken_by_game_token(game_token["token"]) - cookie_token = cookie_token_data["data"]["cookie_token"] - self.cookies.cookie_token = cookie_token - self.cookies.account_id = game_token["uid"] - return True - - async def check_qrcode_login(self, ticket: str): - data = {"app_id": "8", "ticket": ticket, "device": self.device_id} - for _ in range(20): - await asyncio.sleep(10) - res = await self.client.post(self.QRCODE_GET_API, json=data) - res_json = res.json() - ret_code = res_json.get("retcode", 1) - if ret_code != 0: - logger.debug("QRCODE_GET_API: [%s]%s", res_json.get("retcode"), res_json.get("message")) - return False - logger.debug("QRCODE_GET_API: %s", res_json.get("data")) - res_data = res_json.get("data", {}) - if res_data.get("stat", "") == "Confirmed": - return await self._set_cookie_by_game_token(res_json.get("data", {})) - - @staticmethod - def generate_qrcode(url: str) -> bytes: - qr = qrcode.QRCode(version=1, error_correction=qrcode.constants.ERROR_CORRECT_L, box_size=10, border=4) - qr.add_data(url) - qr.make(fit=True) - img = qr.make_image(image_factory=PyPNGImage, fill_color="black", back_color="white") - bio = BytesIO() - img.save(bio) - return bio.getvalue() diff --git a/modules/apihelper/models/genshin/cookies.py b/modules/apihelper/models/genshin/cookies.py index 87e6561c..7d6845d3 100644 --- a/modules/apihelper/models/genshin/cookies.py +++ b/modules/apihelper/models/genshin/cookies.py @@ -13,6 +13,7 @@ class CookiesModel(BaseModel): stoken: Optional[str] = None stuid: Optional[IntStr] = None + mid: Optional[str] = None account_id: Optional[IntStr] = None cookie_token: Optional[str] = None diff --git a/plugins/account/cookies.py b/plugins/account/cookies.py index e280ceab..ca7b3734 100644 --- a/plugins/account/cookies.py +++ b/plugins/account/cookies.py @@ -1,7 +1,11 @@ +import asyncio from datetime import datetime +from io import BytesIO from typing import Dict, Optional +import qrcode from arkowrapper import ArkoWrapper +from qrcode.image.pure import PyPNGImage from simnet import GenshinClient, Region from simnet.errors import DataNotPublic, InvalidCookies, BadRequest as SimnetBadRequest from simnet.models.lab.record import Account @@ -16,7 +20,6 @@ from core.services.cookies.models import CookiesDataBase as Cookies, CookiesStat from core.services.cookies.services import CookiesService from core.services.players.models import PlayersDataBase as Player, PlayerInfoSQLModel from core.services.players.services import PlayersService, PlayerInfoService -from modules.apihelper.client.components.authclient import AuthClient from modules.apihelper.models.genshin.cookies import CookiesModel from utils.log import logger @@ -93,6 +96,16 @@ class AccountCookiesPlugin(Plugin.Conversation): await message.reply_markdown_v2(text, reply_markup=ReplyKeyboardMarkup(reply_keyboard, one_time_keyboard=True)) return CHECK_SERVER + @staticmethod + def generate_qrcode(url: str) -> bytes: + qr = qrcode.QRCode(version=1, error_correction=qrcode.constants.ERROR_CORRECT_L, box_size=10, border=4) + qr.add_data(url) + qr.make(fit=True) + img = qr.make_image(image_factory=PyPNGImage, fill_color="black", back_color="white") + bio = BytesIO() + img.save(bio) + return bio.getvalue() + @conversation.entry_point @handler.command("qlogin", filters=filters.ChatType.PRIVATE, block=False) async def qrcode_login(self, update: Update, context: CallbackContext): @@ -106,14 +119,25 @@ class AccountCookiesPlugin(Plugin.Conversation): else: account_cookies_plugin_data.reset() account_cookies_plugin_data.region = RegionEnum.HYPERION - auth_client = AuthClient() - url, ticket = await auth_client.create_qrcode_login() - data = auth_client.generate_qrcode(url) - text = f"你好 {user.mention_html()} ！该绑定方法仅支持国服，请在3分钟内使用米游社扫码并确认进行绑定。" - await message.reply_photo(data, caption=text, parse_mode=ParseMode.HTML) - if await auth_client.check_qrcode_login(ticket): - account_cookies_plugin_data.cookies = auth_client.cookies.to_dict() - return await self.check_cookies(update, context) + async with GenshinClient(region=Region.CHINESE) as client: + url, ticket = await client.gen_login_qrcode() + data = self.generate_qrcode(url) + text = f"你好 {user.mention_html()} ！该绑定方法仅支持国服，请在3分钟内使用米游社扫码并确认进行绑定。" + await message.reply_photo(data, caption=text, parse_mode=ParseMode.HTML) + for _ in range(20): + await asyncio.sleep(10) + try: + if game_token := await client.check_login_qrcode(ticket): + cookies = {"stuid": str(client.account_id), "ltuid": str(client.account_id)} + cookies["stoken"], cookies["mid"] = await client.get_stoken_v2_and_mid_by_game_token(game_token) + cookies["cookie_token"] = await client.get_cookie_token_by_stoken() + cookies["ltoken"] = await client.get_ltoken_by_stoken() + account_cookies_plugin_data.cookies = cookies + return await self.check_cookies(update, context) + except SimnetBadRequest as e: + if e.ret_code == -106: + break + raise e await message.reply_markdown_v2("可能是验证码已过期或者你没有同意授权，请重新发送命令进行绑定。") return ConversationHandler.END