🎨 Use the latest DS function for genshin.py requests

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}"
+    )

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