fetch enka data from mirror ysin

ysin_panel.py

import asyncio
import hmac
import json
from datetime import datetime
from pathlib import Path
from typing import Dict, Optional, Tuple, Union

from httpx import AsyncClient

# https://yuanshen.xin/static/js/chunk-vendors.6364f242.js "spaceName"
SPACEID = "79e44238-f896-4f4e-8e1a-2cb2d1958cdb"
CLIENTSECRET = "DPHVqwvSMO/1fsIUaiR36Q=="
ENDPOINT = "https://api.bspapp.com"
# https://yuanshen.xin/static/js/index.93aa0995.js "A.__uniConfig"
APPID = "__UNI__9D54A71"
APPNAME = "ysin"
APPVERSION = "1.0.0"
APPVERSIONCODE = "100"
x_basement_token, expires_time = "", 0


def get_body(
    timestamp: float, uid: str = "", token: Optional[str] = None
) -> Dict[str, Union[str, int]]:
    """获取请求载荷。传入 token 时为获取面板，未传入 token 时为刷新令牌"""

    return (
        {
            "method": "serverless.function.runtime.invoke",
            "params": json.dumps(
                {
                    "functionTarget": "getEnka",
                    "functionArgs": {
                        "uid": uid,
                        "clientInfo": {
                            "PLATFORM": "web",
                            "OS": "windows",
                            "APPID": APPID,
                            # "deviceId": "16726373402827372880",
                            "deviceId": str(int(timestamp * 1e10)),
                            "scene": 1001,
                            "appId": APPID,
                            "appLanguage": "zh-Hans",
                            "appName": APPNAME,
                            "appVersion": APPVERSION,
                            "appVersionCode": APPVERSIONCODE,
                            "browserName": "chrome",
                            "browserVersion": "108.0.0.0",
                            # "deviceId": "16726373402827372880",
                            "deviceId": str(int(timestamp * 1e10)),
                            "deviceModel": "PC",
                            "deviceOrientation": "portrait",
                            "devicePixelRatio": 1.25,
                            "deviceType": "pc",
                            "hostLanguage": "zh-CN",
                            "hostName": "chrome",
                            "hostTheme": "light",
                            "hostVersion": "108.0.0.0",
                            "osName": "windows",
                            "osVersion": "10 x64",
                            "ua": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36",
                            "locale": "zh-Hans",
                            "LOCALE": "zh-Hans",
                        },
                    },
                },
                separators=(",", ":"),
            ),
            "spaceId": SPACEID,
            "timestamp": int(str(timestamp * 1e3)[:13]),
            "token": token,
        }
        if token
        else {
            "method": "serverless.auth.user.anonymousAuthorize",
            "params": "{}",
            "spaceId": SPACEID,
            "timestamp": int(str(timestamp * 1e3)[:13]),
        }
    )


def get_x_serverless_sign(data: Dict[str, Union[str, int]]) -> str:
    """获取 ``x-serverless-sign`` 请求头"""

    data_str = "&".join(f"{k}={data[k]}" for k in sorted(data.keys()) if data[k])
    # print(data_str)
    key = bytes(CLIENTSECRET, encoding="utf-8")
    msg = bytes(data_str, encoding="utf-8")
    return hmac.new(key, msg, digestmod="md5").hexdigest()


async def get_x_basement_token(client: AsyncClient) -> Tuple[str, float]:
    """获取 ``x-basement-token`` 请求头及其过期时间"""

    timestamp = datetime.now().timestamp()
    body = get_body(timestamp)
    res = await client.post(
        "/client",
        headers={
            "content-type": "application/json",
            "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36",
            "x-serverless-sign": get_x_serverless_sign(body),
        },
        json=body,
    )
    res_json = res.json()
    access_token = res_json["data"]["accessToken"]
    expires_time = timestamp + res_json["data"]["expiresInSecond"]
    return access_token, expires_time


async def ysin_panel(uid: str) -> Dict:
    """从 Ysin（原神心）获取角色展柜数据"""

    global x_basement_token
    global expires_time
    now = datetime.now().timestamp()

    async with AsyncClient(base_url=ENDPOINT) as client:

        if now > expires_time:
            x_basement_token, expires_time = await get_x_basement_token(client)

        body = get_body(now, uid=uid, token=x_basement_token)
        x_serverless_sign = get_x_serverless_sign(body)

        raw = await client.post(
            "/client",
            headers={
                "content-type": "application/json",
                "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36",
                "x-basement-token": x_basement_token,
                "x-serverless-sign": x_serverless_sign,
            },
            json=body,
        )
        raw_json = raw.json()
        panel_data = raw_json["data"]["data"]

        Path("ysin.raw.json").write_text(
            json.dumps(raw_json, ensure_ascii=False, indent=2), encoding="UTF-8"
        )
        Path("ysin.panel.json").write_text(
            json.dumps(panel_data, ensure_ascii=False, indent=2), encoding="UTF-8"
        )

    return panel_data


asyncio.run(ysin_panel("134266985"))