From 3b9596c6afbb432a5bb9366acaeb74b20e565e40 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B4=9B=E6=B0=B4=E5=B1=85=E5=AE=A4?= Date: Mon, 14 Nov 2022 14:06:07 +0800 Subject: [PATCH] =?UTF-8?q?=F0=9F=8E=A8=20Use=20the=20latest=20DS=20functi?= =?UTF-8?q?on=20for=20`genshin.py`=20requests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- modules/apihelper/helpers.py | 33 +++++------ utils/patch/genshin.py | 109 +++++++++++++++++++++++++++++++++++ 2 files changed, 124 insertions(+), 18 deletions(-) diff --git a/modules/apihelper/helpers.py b/modules/apihelper/helpers.py index 3045548..72199b4 100644 --- a/modules/apihelper/helpers.py +++ b/modules/apihelper/helpers.py @@ -22,14 +22,6 @@ def get_device_id(name: str = None): return str(uuid.uuid3(uuid.NAMESPACE_URL, name)) -def random_text(num: int) -> str: - return "".join(random.sample(string.ascii_lowercase + string.digits, num)) - - -def timestamp() -> int: - return int(time.time()) - - def _hexdigest(text): _md5 = hashlib.md5() # nosec B303 _md5.update(text.encode()) @@ -37,10 +29,15 @@ def _hexdigest(text): def get_ds(ds_type: str = None, new_ds: bool = False, data: Mapping[str, str] = None, params: Mapping[str, str] = None): - # 1: ios - # 2: android - # 4: pc web - # 5: mobile web + """DS 算法 + 代码来自 https://github.com/y1ndan/genshinhelper + :param ds_type: 1:ios 2:android 4:pc web 5:mobile web + :param new_ds: 是否为DS2算法 + :param data: 需要签名的Data + :param params: 需要签名的Params + :return: + """ + def new(): t = str(int(time.time())) r = str(random.randint(100001, 200000)) # nosec @@ -86,9 +83,9 @@ def get_recognize_server(uid: int) -> str: raise TypeError(f"UID {uid} isn't associated with any recognize server") -def get_headers(): - headers = { - "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " - "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/102.0.5005.63 Safari/537.36", - } - return headers +def get_headers(device: str = "Paimon Build", version: str = "2.36.1"): + return ( + f"Mozilla/5.0 (Linux; Android 12; {device}; wv) " + "AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/103.0.5060.129 Mobile Safari/537.36 " + f"miHoYoBBS/{version}" + ) \ No newline at end of file diff --git a/utils/patch/genshin.py b/utils/patch/genshin.py index ddfb768..3de045b 100644 --- a/utils/patch/genshin.py +++ b/utils/patch/genshin.py @@ -1,7 +1,13 @@ import typing +import aiohttp.typedefs import genshin # pylint: disable=W0406 +import yarl +from genshin import constants, types +from genshin.client import routes +from genshin.utility import ds +from modules.apihelper.helpers import get_ds, get_headers from utils.patch.methods import patch, patchable @@ -39,3 +45,106 @@ class CalculatorClient: } data["weapon"] = weapon return genshin.models.genshin.CalculatorCharacterDetails(**data) + + +@patch(genshin.client.components.base.BaseClient) # noqa +class BaseClient: + @patchable + async def request_hoyolab( + self, + url: aiohttp.typedefs.StrOrURL, + *, + lang: typing.Optional[str] = None, + region: typing.Optional[types.Region] = None, + method: typing.Optional[str] = None, + params: typing.Optional[typing.Mapping[str, typing.Any]] = None, + data: typing.Any = None, + headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None, + **kwargs: typing.Any, + ) -> typing.Mapping[str, typing.Any]: + """Make a request any hoyolab endpoint.""" + if lang is not None and lang not in constants.LANGS: + raise ValueError(f"{lang} is not a valid language, must be one of: " + ", ".join(constants.LANGS)) + + lang = lang or self.lang + region = region or self.region + + url = routes.TAKUMI_URL.get_url(region).join(yarl.URL(url)) + + if region == types.Region.OVERSEAS: + headers = { + "x-rpc-app_version": "1.5.0", + "x-rpc-client_type": "4", + "x-rpc-language": lang, + "ds": ds.generate_dynamic_secret(), + } + elif region == types.Region.CHINESE: + _app_version, _client_type, _ds = get_ds(new_ds=True, data=data, params=params) + ua = get_headers(version=_app_version) + headers = { + "User-Agent": ua, + "X_Requested_With": "com.mihoyo.hoyolab", + "Referer": "https://webstatic-sea.hoyolab.com", + "x-rpc-app_version": _app_version, + "x-rpc-client_type": _client_type, + "ds": _ds, + } + else: + raise TypeError(f"{region!r} is not a valid region.") + + data = await self.request(url, method=method, params=params, data=data, headers=headers, **kwargs) + return data + + @patchable + async def request( + self, + url: aiohttp.typedefs.StrOrURL, + *, + method: typing.Optional[str] = None, + params: typing.Optional[typing.Mapping[str, typing.Any]] = None, + data: typing.Any = None, + headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None, + cache: typing.Any = None, + static_cache: typing.Any = None, + **kwargs: typing.Any, + ) -> typing.Mapping[str, typing.Any]: + """Make a request and return a parsed json response.""" + if cache is not None: + value = await self.cache.get(cache) + if value is not None: + return value + elif static_cache is not None: + value = await self.cache.get_static(static_cache) + if value is not None: + return value + + # actual request + + headers = dict(headers or {}) + headers.setdefault("User-Agent", self.USER_AGENT) + + if method is None: + method = "POST" if data else "GET" + + if "json" in kwargs: + raise TypeError("Use data instead of json in request.") + + await self._request_hook(method, url, params=params, data=data, headers=headers, **kwargs) + + response = await self.cookie_manager.request( + url, + method=method, + params=params, + json=data, + headers=headers, + **kwargs, + ) + + # cache + + if cache is not None: + await self.cache.set(cache, response) + elif static_cache is not None: + await self.cache.set_static(static_cache, response) + + return response