Module genshin.client.components.base

Base ABC Client.

Classes

class BaseClient (cookies: ForwardRef('http.cookies.BaseCookie[Any]') | Mapping[Any, Any] | str | Sequence[ForwardRef('http.cookies.BaseCookie[Any]') | Mapping[Any, Any] | str] | None = None,
*,
authkey: str | None = None,
lang: Literal['zh-cn', 'zh-tw', 'de-de', 'en-us', 'es-es', 'fr-fr', 'id-id', 'it-it', 'ja-jp', 'ko-kr', 'pt-pt', 'ru-ru', 'th-th', 'vi-vn', 'tr-tr'] = 'en-us',
region: Region = Region.OVERSEAS,
proxy: str | None = None,
game: Game | None = None,
uid: int | None = None,
hoyolab_id: int | None = None,
device_id: str | None = None,
device_fp: str | None = None,
headers: Mapping[str, str] | Mapping[multidict._multidict.istr, str] | multidict._multidict.CIMultiDict | multidict._multidict.CIMultiDictProxy | Iterable[Tuple[str | multidict._multidict.istr, str]] | None = None,
cache: BaseCache | None = None,
debug: bool = False,
on_cookie_update: Callable[[Mapping[str, str]], Awaitable[None] | None] | None = None)

Expand source code

class BaseClient(abc.ABC):
    """Base ABC Client."""

    __slots__ = (
        "cookie_manager",
        "cache",
        "_lang",
        "_region",
        "_default_game",
        "uids",
        "authkeys",
        "_hoyolab_id",
        "_accounts",
        "custom_headers",
    )

    USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36"  # noqa: E501

    logger: logging.Logger = logging.getLogger(__name__)

    cookie_manager: managers.BaseCookieManager
    cache: client_cache.BaseCache
    _lang: str
    _region: types.Region
    _default_game: typing.Optional[types.Game]

    uids: dict[types.Game, int]
    authkeys: dict[types.Game, str]
    _hoyolab_id: typing.Optional[int]
    _accounts: dict[types.Game, hoyolab_models.GenshinAccount]
    custom_headers: multidict.CIMultiDict[str]

    def __init__(
        self,
        cookies: typing.Optional[managers.AnyCookieOrHeader] = None,
        *,
        authkey: typing.Optional[str] = None,
        lang: types.Lang = "en-us",
        region: types.Region = types.Region.OVERSEAS,
        proxy: typing.Optional[str] = None,
        game: typing.Optional[types.Game] = None,
        uid: typing.Optional[int] = None,
        hoyolab_id: typing.Optional[int] = None,
        device_id: typing.Optional[str] = None,
        device_fp: typing.Optional[str] = None,
        headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
        cache: typing.Optional[client_cache.BaseCache] = None,
        debug: bool = False,
        on_cookie_update: typing.Optional[managers.CookieUpdateHook] = None,
    ) -> None:
        self.cookie_manager = managers.BaseCookieManager.from_cookies(cookies)
        self.on_cookie_update = on_cookie_update
        self.cache = cache or client_cache.StaticCache()

        self.uids = {}
        self.authkeys = {}
        self._accounts = {}

        self.default_game = game
        self.lang = lang
        self.region = region
        self.authkey = authkey
        self.debug = debug
        self.proxy = proxy
        self.uid = uid
        self.hoyolab_id = hoyolab_id

        self.custom_headers = parse_loose_headers(headers)
        self.custom_headers.update({"x-rpc-device_id": device_id} if device_id else {})
        self.custom_headers.update({"x-rpc-device_fp": device_fp} if device_fp else {})

    def __repr__(self) -> str:
        kwargs = dict(
            lang=self.lang,
            region=self.region.value,
            default_game=self.default_game and self.default_game.value,
            hoyolab_id=self.hoyolab_id,
            uid=self.default_game and self.uid,
            authkey=self.authkey and self.authkey[:12] + "...",
            proxy=self.proxy,
            debug=self.debug,
        )
        return f"<{type(self).__name__} {', '.join(f'{k}={v!r}' for k, v in kwargs.items() if v)}>"

    @property
    def device_id(self) -> typing.Optional[str]:
        """The device id used in headers."""
        return self.custom_headers.get("x-rpc-device_id")

    @device_id.setter
    def device_id(self, device_id: str) -> None:
        self.custom_headers["x-rpc-device_id"] = device_id

    @property
    def device_fp(self) -> typing.Optional[str]:
        """The device fingerprint used in headers."""
        return self.custom_headers.get("x-rpc-device_fp")

    @device_fp.setter
    def device_fp(self, device_fp: str) -> None:
        self.custom_headers["x-rpc-device_fp"] = device_fp

    @property
    def hoyolab_id(self) -> typing.Optional[int]:
        """The logged-in user's hoyolab uid.

        Returns None if not found or not applicable.
        """
        return self._hoyolab_id or self.cookie_manager.user_id

    @hoyolab_id.setter
    def hoyolab_id(self, hoyolab_id: typing.Optional[int]) -> None:
        if hoyolab_id is None:
            self._hoyolab_id = None
            return

        if self.cookie_manager.multi:
            raise RuntimeError("Cannot specify a hoyolab uid when using multiple cookies.")

        if self.cookie_manager.user_id and hoyolab_id and self.cookie_manager.user_id != hoyolab_id:
            raise ValueError("The provided hoyolab uid does not match the cookie id.")

        self._hoyolab_id = hoyolab_id

    @property
    def lang(self) -> str:
        """The default language, defaults to "en-us" """
        return self._lang

    @lang.setter
    def lang(self, lang: str) -> None:
        if lang not in constants.LANGS:
            raise ValueError(f"{lang} is not a valid language, must be one of: " + ", ".join(constants.LANGS))

        self._lang = lang

    @property
    def region(self) -> types.Region:
        """The default region."""
        return self._region

    @region.setter
    def region(self, region: str) -> None:
        self._region = types.Region(region)

        if region == types.Region.CHINESE:
            self.lang = "zh-cn"

    @property
    def default_game(self) -> typing.Optional[types.Game]:
        """The default game."""
        return self._default_game

    @default_game.setter
    def default_game(self, game: typing.Optional[str]) -> None:
        self._default_game = types.Game(game) if game else None

    game = default_game

    @property
    def uid(self) -> typing.Optional[int]:
        """UID of the default game."""
        if self.default_game is None:
            if len(self.uids) != 1:
                return None

            (self.default_game,) = self.uids.keys()

        return self.uids.get(self.default_game)

    @uid.setter
    def uid(self, uid: typing.Optional[int]) -> None:
        if uid is None:
            self.uids.clear()
            return

        self._default_game = self._default_game or utility.recognize_game(uid, region=self.region)
        if self.default_game is None:
            raise RuntimeError("No default game set. Cannot set uid.")

        self.uids[self.default_game] = uid

    @property
    def authkey(self) -> typing.Optional[str]:
        """The default genshin authkey used for paginators."""
        if self.default_game is None:
            if self.authkeys:
                warnings.warn("Tried to get an authkey without a default game set.")

            return None

        return self.authkeys.get(self.default_game)

    @authkey.setter
    def authkey(self, authkey: typing.Optional[str]) -> None:
        if authkey is None:
            self.authkeys.clear()
            return

        authkey = urllib.parse.unquote(authkey)

        try:
            base64.b64decode(authkey, validate=True)
        except Exception as e:
            raise ValueError("authkey is not a valid base64 encoded string") from e

        if not self.default_game:
            raise RuntimeError("No default game set. Cannot set authkey with property.")

        self.authkeys[self.default_game] = authkey

    @property
    def debug(self) -> bool:
        """Whether the debug logs are being shown in stdout"""
        return logging.getLogger("genshin").level == logging.DEBUG

    @debug.setter
    def debug(self, debug: bool) -> None:
        logging.basicConfig()
        level = logging.DEBUG if debug else logging.NOTSET
        logging.getLogger("genshin").setLevel(level)

    @property
    def on_cookie_update(self) -> typing.Optional[managers.CookieUpdateHook]:
        """Callback invoked with a copy of the cookies whenever the client updates them itself.

        This happens for example when a new cookie_token is minted from an stoken or when
        a response sets new cookies. May be a sync or async callable. Use it to persist
        refreshed cookies so the next client created from storage does not need to refresh again.
        """
        return self.cookie_manager.on_cookie_update

    @on_cookie_update.setter
    def on_cookie_update(self, hook: typing.Optional[managers.CookieUpdateHook]) -> None:
        self.cookie_manager.on_cookie_update = hook

    def set_cookies(self, cookies: typing.Optional[managers.AnyCookieOrHeader] = None, **kwargs: typing.Any) -> None:
        """Parse and set cookies."""
        if not bool(cookies) ^ bool(kwargs):
            raise TypeError("Cannot use both positional and keyword arguments at once")

        hook = self.on_cookie_update
        self.cookie_manager = managers.BaseCookieManager.from_cookies(cookies or kwargs)
        self.on_cookie_update = hook

    def set_browser_cookies(self, browser: typing.Optional[str] = None) -> None:
        """Extract cookies from your browser and set them as client cookies.

        Available browsers: chrome, chromium, opera, edge, firefox.
        """
        hook = self.on_cookie_update
        self.cookie_manager = managers.BaseCookieManager.from_browser_cookies(browser)
        self.on_cookie_update = hook

    def set_authkey(self, authkey: typing.Optional[str] = None, *, game: typing.Optional[types.Game] = None) -> None:
        """Set an authkey for wish & transaction logs.

        Accepts an authkey, a url containing an authkey or a path towards a logfile.
        """
        if authkey is None or os.path.isfile(authkey):
            authkey = utility.get_authkey(authkey)
        else:
            authkey = utility.extract_authkey(authkey) or authkey

        game = game or self.default_game
        if game is None:
            raise RuntimeError("No default game set.")

        self.authkeys[game] = authkey

    def set_cache(
        self, maxsize: int = 1024, *, ttl: int = client_cache.HOUR, static_ttl: int = client_cache.DAY
    ) -> None:
        """Create and set a new cache."""
        self.cache = client_cache.Cache(maxsize, ttl=ttl, static_ttl=static_ttl)

    def set_redis_cache(
        self, url: str, *, ttl: int = client_cache.HOUR, static_ttl: int = client_cache.DAY, **redis_kwargs: typing.Any
    ) -> None:
        """Create and set a new redis cache."""
        import aioredis

        redis = aioredis.Redis.from_url(url, **redis_kwargs)  # pyright: ignore[reportUnknownMemberType]
        self.cache = client_cache.RedisCache(redis, ttl=ttl, static_ttl=static_ttl)

    @property
    def proxy(self) -> typing.Optional[str]:
        """Proxy for http requests."""
        if self.cookie_manager.proxy is None:
            return None

        return str(self.cookie_manager.proxy)

    @proxy.setter
    def proxy(self, proxy: typing.Optional[aiohttp.typedefs.StrOrURL]) -> None:
        self.cookie_manager.proxy = yarl.URL(proxy) if proxy else None

    async def _request_hook(
        self,
        method: str,
        url: aiohttp.typedefs.StrOrURL,
        *,
        params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        data: typing.Any = None,
        **kwargs: typing.Any,
    ) -> None:
        """Perform an action before a request.

        Debug logging by default.
        """
        url = yarl.URL(url)
        if params:
            params = {k: v for k, v in params.items() if k != "authkey"}
            url = url.update_query(params)

        if data:
            self.logger.debug("%s %s\n%s", method, url, json.dumps(data, separators=(",", ":")))
        else:
            self.logger.debug("%s %s", method, url)

    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 = parse_loose_headers(headers)
        headers["User-Agent"] = self.USER_AGENT
        headers.update(self.custom_headers)

        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

    async def request_webstatic(
        self,
        url: aiohttp.typedefs.StrOrURL,
        *,
        headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
        cache: typing.Any = None,
        region: types.Region = types.Region.OVERSEAS,
        **kwargs: typing.Any,
    ) -> typing.Any:
        """Request a static json file."""
        if cache is not None:
            value = await self.cache.get_static(cache)
            if value is not None:
                return value

        url = routes.WEBSTATIC_URL.get_url(region).join(yarl.URL(url))

        headers = parse_loose_headers(headers)
        headers["User-Agent"] = self.USER_AGENT
        headers.update(self.custom_headers)

        await self._request_hook("GET", url, headers=headers, **kwargs)

        async with self.cookie_manager.create_session() as session:
            async with session.get(url, headers=headers, proxy=self.proxy, **kwargs) as r:
                r.raise_for_status()
                data = await r.json()

        if cache is not None:
            await self.cache.set_static(cache, data)

        return data

    async def request_bbs(
        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 bbs 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.BBS_URL.get_url(region).join(yarl.URL(url))

        headers = parse_loose_headers(headers)
        headers.update(ds.get_ds_headers(data=data, params=params, region=region, lang=lang or self.lang))
        headers["Referer"] = str(routes.BBS_REFERER_URL.get_url(self.region))

        data = await self.request(url, method=method, params=params, data=data, headers=headers, **kwargs)
        return data

    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))

        headers = parse_loose_headers(headers)
        headers.update(ds.get_ds_headers(data=data, params=params, region=region, lang=lang or self.lang))

        data = await self.request(url, method=method, params=params, data=data, headers=headers, **kwargs)
        return data

    @managers.no_multi
    async def get_game_accounts(
        self, *, lang: typing.Optional[str] = None
    ) -> typing.Sequence[hoyolab_models.GenshinAccount]:
        """Get the game accounts of the currently logged-in user."""
        if self.hoyolab_id is None:
            warnings.warn("No hoyolab id set, caching may be unreliable.")

        data = await self.request_hoyolab(
            "binding/api/getUserGameRolesByCookie",
            lang=lang,
            cache=client_cache.cache_key("accounts", hoyolab_id=self.hoyolab_id),
        )
        return [hoyolab_models.GenshinAccount(**i) for i in data["list"]]

    @deprecation.deprecated("get_game_accounts")
    async def genshin_accounts(
        self, *, lang: typing.Optional[str] = None
    ) -> typing.Sequence[hoyolab_models.GenshinAccount]:
        """Get the genshin accounts of the currently logged-in user."""
        accounts = await self.get_game_accounts(lang=lang)
        return [account for account in accounts if account.game == types.Game.GENSHIN]

    async def _update_cached_uids(self) -> None:
        """Update cached fallback uids."""
        mixed_accounts = await self.get_game_accounts()

        game_accounts: dict[types.Game, list[hoyolab_models.GenshinAccount]] = {}
        for account in mixed_accounts:
            if not isinstance(account.game, types.Game):  # pyright: ignore[reportUnnecessaryIsInstance]
                continue

            game_accounts.setdefault(account.game, []).append(account)

        self.uids = {game: max(accounts, key=lambda a: a.level).uid for game, accounts in game_accounts.items()}

        if len(self.uids) == 1 and self.default_game is None:
            (self.default_game,) = self.uids.keys()

    @concurrency.prevent_concurrency
    async def _get_uid(self, game: types.Game) -> int:
        """Get a cached fallback uid."""
        # TODO: use lock
        if uid := self.uids.get(game):
            return uid

        if self.cookie_manager.multi:
            raise RuntimeError("UID must be provided when using multi-cookie managers.")

        await self._update_cached_uids()

        if uid := self.uids.get(game):
            return uid

        raise errors.AccountNotFound(msg="No UID provided and account has no game account bound to it.")

    async def _update_cached_accounts(self) -> None:
        """Update cached fallback accounts."""
        mixed_accounts = await self.get_game_accounts()

        game_accounts: dict[types.Game, list[hoyolab_models.GenshinAccount]] = {}
        for account in mixed_accounts:
            if not isinstance(account.game, types.Game):  # pyright: ignore[reportUnnecessaryIsInstance]
                continue

            game_accounts.setdefault(account.game, []).append(account)

        self._accounts = {}
        for game, accounts in game_accounts.items():
            self._accounts[game] = next(
                (acc for acc in accounts if acc.uid == self.uids.get(game)), max(accounts, key=lambda a: a.level)
            )

    @concurrency.prevent_concurrency
    async def _get_account(self, game: types.Game) -> hoyolab_models.GenshinAccount:
        """Get a cached fallback account."""
        if (account := self._accounts.get(game)) and (uid := self.uids.get(game)) and account.uid == uid:
            return account

        await self._update_cached_accounts()

        if account := self._accounts.get(game):
            if (uid := self.uids.get(game)) and account.uid != uid:
                raise errors.AccountNotFound(msg="There is no game account with such UID.")

            return account

        raise errors.AccountNotFound(msg="Account has no game account bound to it.")

    def _get_hoyolab_id(self) -> int:
        """Get a cached fallback hoyolab ID."""
        if self.hoyolab_id is not None:
            return self.hoyolab_id

        if self.cookie_manager.multi:
            raise RuntimeError("Hoyolab ID must be provided when using multi-cookie managers.")

        raise RuntimeError("No default hoyolab ID provided.")

    def get_account_timezone(
        self, *, game: typing.Optional[types.Game] = None, uid: typing.Optional[int] = None
    ) -> typing.Optional[int]:
        """Get the UTC timezone of the default game account."""
        game = game or self.game
        if game is None:
            return None

        uid = uid or self.uids.get(game)
        if uid is None:
            return None

        server = recognize_server(uid, game=game)  # type: ignore[arg-type]

        for tz, servers in constants.SERVER_TIMEZONES.items():
            if server in servers:
                return tz

        return None

    def _add_timezone_to_data(
        self,
        data: typing.Mapping[str, typing.Any],
        keys: typing.Sequence[str],
        *,
        game: typing.Optional[types.Game] = None,
        uid: typing.Optional[int] = None,
    ) -> typing.Mapping[str, typing.Any]:
        """Recursively add timezone info to a data dict based on the default game account.

        Datetime dicts found under the given keys get a "tzinfo" entry added, unix timestamp
        values get wrapped into {"timestamp": ..., "tzinfo": ...} dicts.
        """
        tz = self.get_account_timezone(game=game, uid=uid)
        if tz is None:
            return data

        def add_tz(value: typing.Any) -> None:
            if isinstance(value, dict):
                value = typing.cast("typing.Dict[str, typing.Any]", value)
                for key, item in value.items():
                    if key in keys and isinstance(item, dict) and item:
                        item["tzinfo"] = tz
                    elif key in keys and (isinstance(item, int) or (isinstance(item, str) and item.isdigit())):
                        value[key] = {"timestamp": item, "tzinfo": tz}
                    else:
                        add_tz(item)
            elif isinstance(value, list):
                for item in typing.cast("typing.List[typing.Any]", value):
                    add_tz(item)

        data = dict(data)
        add_tz(data)
        return data

Base ABC Client.

Ancestors

• abc.ABC

Subclasses

• AppAuthClient
• GameAuthClient
• WebAuthClient
• BadgeLoginClient
• CalculatorClient
• CardPlazaClient
• BaseBattleChronicleClient
• DailyRewardClient
• DiaryClient
• WishClient
• HoyolabClient
• HSRLineupClient
• LineupClient
• TeapotClient
• TransactionClient
• WikiClient

Class variables

var USER_AGENT

var logger : logging.Logger

Instance variables

prop authkey : str | None

Expand source code

@property
def authkey(self) -> typing.Optional[str]:
    """The default genshin authkey used for paginators."""
    if self.default_game is None:
        if self.authkeys:
            warnings.warn("Tried to get an authkey without a default game set.")

        return None

    return self.authkeys.get(self.default_game)

The default genshin authkey used for paginators.

var authkeys : dict[Game, str]

Expand source code

class BaseClient(abc.ABC):
    """Base ABC Client."""

    __slots__ = (
        "cookie_manager",
        "cache",
        "_lang",
        "_region",
        "_default_game",
        "uids",
        "authkeys",
        "_hoyolab_id",
        "_accounts",
        "custom_headers",
    )

    USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36"  # noqa: E501

    logger: logging.Logger = logging.getLogger(__name__)

    cookie_manager: managers.BaseCookieManager
    cache: client_cache.BaseCache
    _lang: str
    _region: types.Region
    _default_game: typing.Optional[types.Game]

    uids: dict[types.Game, int]
    authkeys: dict[types.Game, str]
    _hoyolab_id: typing.Optional[int]
    _accounts: dict[types.Game, hoyolab_models.GenshinAccount]
    custom_headers: multidict.CIMultiDict[str]

    def __init__(
        self,
        cookies: typing.Optional[managers.AnyCookieOrHeader] = None,
        *,
        authkey: typing.Optional[str] = None,
        lang: types.Lang = "en-us",
        region: types.Region = types.Region.OVERSEAS,
        proxy: typing.Optional[str] = None,
        game: typing.Optional[types.Game] = None,
        uid: typing.Optional[int] = None,
        hoyolab_id: typing.Optional[int] = None,
        device_id: typing.Optional[str] = None,
        device_fp: typing.Optional[str] = None,
        headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
        cache: typing.Optional[client_cache.BaseCache] = None,
        debug: bool = False,
        on_cookie_update: typing.Optional[managers.CookieUpdateHook] = None,
    ) -> None:
        self.cookie_manager = managers.BaseCookieManager.from_cookies(cookies)
        self.on_cookie_update = on_cookie_update
        self.cache = cache or client_cache.StaticCache()

        self.uids = {}
        self.authkeys = {}
        self._accounts = {}

        self.default_game = game
        self.lang = lang
        self.region = region
        self.authkey = authkey
        self.debug = debug
        self.proxy = proxy
        self.uid = uid
        self.hoyolab_id = hoyolab_id

        self.custom_headers = parse_loose_headers(headers)
        self.custom_headers.update({"x-rpc-device_id": device_id} if device_id else {})
        self.custom_headers.update({"x-rpc-device_fp": device_fp} if device_fp else {})

    def __repr__(self) -> str:
        kwargs = dict(
            lang=self.lang,
            region=self.region.value,
            default_game=self.default_game and self.default_game.value,
            hoyolab_id=self.hoyolab_id,
            uid=self.default_game and self.uid,
            authkey=self.authkey and self.authkey[:12] + "...",
            proxy=self.proxy,
            debug=self.debug,
        )
        return f"<{type(self).__name__} {', '.join(f'{k}={v!r}' for k, v in kwargs.items() if v)}>"

    @property
    def device_id(self) -> typing.Optional[str]:
        """The device id used in headers."""
        return self.custom_headers.get("x-rpc-device_id")

    @device_id.setter
    def device_id(self, device_id: str) -> None:
        self.custom_headers["x-rpc-device_id"] = device_id

    @property
    def device_fp(self) -> typing.Optional[str]:
        """The device fingerprint used in headers."""
        return self.custom_headers.get("x-rpc-device_fp")

    @device_fp.setter
    def device_fp(self, device_fp: str) -> None:
        self.custom_headers["x-rpc-device_fp"] = device_fp

    @property
    def hoyolab_id(self) -> typing.Optional[int]:
        """The logged-in user's hoyolab uid.

        Returns None if not found or not applicable.
        """
        return self._hoyolab_id or self.cookie_manager.user_id

    @hoyolab_id.setter
    def hoyolab_id(self, hoyolab_id: typing.Optional[int]) -> None:
        if hoyolab_id is None:
            self._hoyolab_id = None
            return

        if self.cookie_manager.multi:
            raise RuntimeError("Cannot specify a hoyolab uid when using multiple cookies.")

        if self.cookie_manager.user_id and hoyolab_id and self.cookie_manager.user_id != hoyolab_id:
            raise ValueError("The provided hoyolab uid does not match the cookie id.")

        self._hoyolab_id = hoyolab_id

    @property
    def lang(self) -> str:
        """The default language, defaults to "en-us" """
        return self._lang

    @lang.setter
    def lang(self, lang: str) -> None:
        if lang not in constants.LANGS:
            raise ValueError(f"{lang} is not a valid language, must be one of: " + ", ".join(constants.LANGS))

        self._lang = lang

    @property
    def region(self) -> types.Region:
        """The default region."""
        return self._region

    @region.setter
    def region(self, region: str) -> None:
        self._region = types.Region(region)

        if region == types.Region.CHINESE:
            self.lang = "zh-cn"

    @property
    def default_game(self) -> typing.Optional[types.Game]:
        """The default game."""
        return self._default_game

    @default_game.setter
    def default_game(self, game: typing.Optional[str]) -> None:
        self._default_game = types.Game(game) if game else None

    game = default_game

    @property
    def uid(self) -> typing.Optional[int]:
        """UID of the default game."""
        if self.default_game is None:
            if len(self.uids) != 1:
                return None

            (self.default_game,) = self.uids.keys()

        return self.uids.get(self.default_game)

    @uid.setter
    def uid(self, uid: typing.Optional[int]) -> None:
        if uid is None:
            self.uids.clear()
            return

        self._default_game = self._default_game or utility.recognize_game(uid, region=self.region)
        if self.default_game is None:
            raise RuntimeError("No default game set. Cannot set uid.")

        self.uids[self.default_game] = uid

    @property
    def authkey(self) -> typing.Optional[str]:
        """The default genshin authkey used for paginators."""
        if self.default_game is None:
            if self.authkeys:
                warnings.warn("Tried to get an authkey without a default game set.")

            return None

        return self.authkeys.get(self.default_game)

    @authkey.setter
    def authkey(self, authkey: typing.Optional[str]) -> None:
        if authkey is None:
            self.authkeys.clear()
            return

        authkey = urllib.parse.unquote(authkey)

        try:
            base64.b64decode(authkey, validate=True)
        except Exception as e:
            raise ValueError("authkey is not a valid base64 encoded string") from e

        if not self.default_game:
            raise RuntimeError("No default game set. Cannot set authkey with property.")

        self.authkeys[self.default_game] = authkey

    @property
    def debug(self) -> bool:
        """Whether the debug logs are being shown in stdout"""
        return logging.getLogger("genshin").level == logging.DEBUG

    @debug.setter
    def debug(self, debug: bool) -> None:
        logging.basicConfig()
        level = logging.DEBUG if debug else logging.NOTSET
        logging.getLogger("genshin").setLevel(level)

    @property
    def on_cookie_update(self) -> typing.Optional[managers.CookieUpdateHook]:
        """Callback invoked with a copy of the cookies whenever the client updates them itself.

        This happens for example when a new cookie_token is minted from an stoken or when
        a response sets new cookies. May be a sync or async callable. Use it to persist
        refreshed cookies so the next client created from storage does not need to refresh again.
        """
        return self.cookie_manager.on_cookie_update

    @on_cookie_update.setter
    def on_cookie_update(self, hook: typing.Optional[managers.CookieUpdateHook]) -> None:
        self.cookie_manager.on_cookie_update = hook

    def set_cookies(self, cookies: typing.Optional[managers.AnyCookieOrHeader] = None, **kwargs: typing.Any) -> None:
        """Parse and set cookies."""
        if not bool(cookies) ^ bool(kwargs):
            raise TypeError("Cannot use both positional and keyword arguments at once")

        hook = self.on_cookie_update
        self.cookie_manager = managers.BaseCookieManager.from_cookies(cookies or kwargs)
        self.on_cookie_update = hook

    def set_browser_cookies(self, browser: typing.Optional[str] = None) -> None:
        """Extract cookies from your browser and set them as client cookies.

        Available browsers: chrome, chromium, opera, edge, firefox.
        """
        hook = self.on_cookie_update
        self.cookie_manager = managers.BaseCookieManager.from_browser_cookies(browser)
        self.on_cookie_update = hook

    def set_authkey(self, authkey: typing.Optional[str] = None, *, game: typing.Optional[types.Game] = None) -> None:
        """Set an authkey for wish & transaction logs.

        Accepts an authkey, a url containing an authkey or a path towards a logfile.
        """
        if authkey is None or os.path.isfile(authkey):
            authkey = utility.get_authkey(authkey)
        else:
            authkey = utility.extract_authkey(authkey) or authkey

        game = game or self.default_game
        if game is None:
            raise RuntimeError("No default game set.")

        self.authkeys[game] = authkey

    def set_cache(
        self, maxsize: int = 1024, *, ttl: int = client_cache.HOUR, static_ttl: int = client_cache.DAY
    ) -> None:
        """Create and set a new cache."""
        self.cache = client_cache.Cache(maxsize, ttl=ttl, static_ttl=static_ttl)

    def set_redis_cache(
        self, url: str, *, ttl: int = client_cache.HOUR, static_ttl: int = client_cache.DAY, **redis_kwargs: typing.Any
    ) -> None:
        """Create and set a new redis cache."""
        import aioredis

        redis = aioredis.Redis.from_url(url, **redis_kwargs)  # pyright: ignore[reportUnknownMemberType]
        self.cache = client_cache.RedisCache(redis, ttl=ttl, static_ttl=static_ttl)

    @property
    def proxy(self) -> typing.Optional[str]:
        """Proxy for http requests."""
        if self.cookie_manager.proxy is None:
            return None

        return str(self.cookie_manager.proxy)

    @proxy.setter
    def proxy(self, proxy: typing.Optional[aiohttp.typedefs.StrOrURL]) -> None:
        self.cookie_manager.proxy = yarl.URL(proxy) if proxy else None

    async def _request_hook(
        self,
        method: str,
        url: aiohttp.typedefs.StrOrURL,
        *,
        params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        data: typing.Any = None,
        **kwargs: typing.Any,
    ) -> None:
        """Perform an action before a request.

        Debug logging by default.
        """
        url = yarl.URL(url)
        if params:
            params = {k: v for k, v in params.items() if k != "authkey"}
            url = url.update_query(params)

        if data:
            self.logger.debug("%s %s\n%s", method, url, json.dumps(data, separators=(",", ":")))
        else:
            self.logger.debug("%s %s", method, url)

    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 = parse_loose_headers(headers)
        headers["User-Agent"] = self.USER_AGENT
        headers.update(self.custom_headers)

        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

    async def request_webstatic(
        self,
        url: aiohttp.typedefs.StrOrURL,
        *,
        headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
        cache: typing.Any = None,
        region: types.Region = types.Region.OVERSEAS,
        **kwargs: typing.Any,
    ) -> typing.Any:
        """Request a static json file."""
        if cache is not None:
            value = await self.cache.get_static(cache)
            if value is not None:
                return value

        url = routes.WEBSTATIC_URL.get_url(region).join(yarl.URL(url))

        headers = parse_loose_headers(headers)
        headers["User-Agent"] = self.USER_AGENT
        headers.update(self.custom_headers)

        await self._request_hook("GET", url, headers=headers, **kwargs)

        async with self.cookie_manager.create_session() as session:
            async with session.get(url, headers=headers, proxy=self.proxy, **kwargs) as r:
                r.raise_for_status()
                data = await r.json()

        if cache is not None:
            await self.cache.set_static(cache, data)

        return data

    async def request_bbs(
        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 bbs 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.BBS_URL.get_url(region).join(yarl.URL(url))

        headers = parse_loose_headers(headers)
        headers.update(ds.get_ds_headers(data=data, params=params, region=region, lang=lang or self.lang))
        headers["Referer"] = str(routes.BBS_REFERER_URL.get_url(self.region))

        data = await self.request(url, method=method, params=params, data=data, headers=headers, **kwargs)
        return data

    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))

        headers = parse_loose_headers(headers)
        headers.update(ds.get_ds_headers(data=data, params=params, region=region, lang=lang or self.lang))

        data = await self.request(url, method=method, params=params, data=data, headers=headers, **kwargs)
        return data

    @managers.no_multi
    async def get_game_accounts(
        self, *, lang: typing.Optional[str] = None
    ) -> typing.Sequence[hoyolab_models.GenshinAccount]:
        """Get the game accounts of the currently logged-in user."""
        if self.hoyolab_id is None:
            warnings.warn("No hoyolab id set, caching may be unreliable.")

        data = await self.request_hoyolab(
            "binding/api/getUserGameRolesByCookie",
            lang=lang,
            cache=client_cache.cache_key("accounts", hoyolab_id=self.hoyolab_id),
        )
        return [hoyolab_models.GenshinAccount(**i) for i in data["list"]]

    @deprecation.deprecated("get_game_accounts")
    async def genshin_accounts(
        self, *, lang: typing.Optional[str] = None
    ) -> typing.Sequence[hoyolab_models.GenshinAccount]:
        """Get the genshin accounts of the currently logged-in user."""
        accounts = await self.get_game_accounts(lang=lang)
        return [account for account in accounts if account.game == types.Game.GENSHIN]

    async def _update_cached_uids(self) -> None:
        """Update cached fallback uids."""
        mixed_accounts = await self.get_game_accounts()

        game_accounts: dict[types.Game, list[hoyolab_models.GenshinAccount]] = {}
        for account in mixed_accounts:
            if not isinstance(account.game, types.Game):  # pyright: ignore[reportUnnecessaryIsInstance]
                continue

            game_accounts.setdefault(account.game, []).append(account)

        self.uids = {game: max(accounts, key=lambda a: a.level).uid for game, accounts in game_accounts.items()}

        if len(self.uids) == 1 and self.default_game is None:
            (self.default_game,) = self.uids.keys()

    @concurrency.prevent_concurrency
    async def _get_uid(self, game: types.Game) -> int:
        """Get a cached fallback uid."""
        # TODO: use lock
        if uid := self.uids.get(game):
            return uid

        if self.cookie_manager.multi:
            raise RuntimeError("UID must be provided when using multi-cookie managers.")

        await self._update_cached_uids()

        if uid := self.uids.get(game):
            return uid

        raise errors.AccountNotFound(msg="No UID provided and account has no game account bound to it.")

    async def _update_cached_accounts(self) -> None:
        """Update cached fallback accounts."""
        mixed_accounts = await self.get_game_accounts()

        game_accounts: dict[types.Game, list[hoyolab_models.GenshinAccount]] = {}
        for account in mixed_accounts:
            if not isinstance(account.game, types.Game):  # pyright: ignore[reportUnnecessaryIsInstance]
                continue

            game_accounts.setdefault(account.game, []).append(account)

        self._accounts = {}
        for game, accounts in game_accounts.items():
            self._accounts[game] = next(
                (acc for acc in accounts if acc.uid == self.uids.get(game)), max(accounts, key=lambda a: a.level)
            )

    @concurrency.prevent_concurrency
    async def _get_account(self, game: types.Game) -> hoyolab_models.GenshinAccount:
        """Get a cached fallback account."""
        if (account := self._accounts.get(game)) and (uid := self.uids.get(game)) and account.uid == uid:
            return account

        await self._update_cached_accounts()

        if account := self._accounts.get(game):
            if (uid := self.uids.get(game)) and account.uid != uid:
                raise errors.AccountNotFound(msg="There is no game account with such UID.")

            return account

        raise errors.AccountNotFound(msg="Account has no game account bound to it.")

    def _get_hoyolab_id(self) -> int:
        """Get a cached fallback hoyolab ID."""
        if self.hoyolab_id is not None:
            return self.hoyolab_id

        if self.cookie_manager.multi:
            raise RuntimeError("Hoyolab ID must be provided when using multi-cookie managers.")

        raise RuntimeError("No default hoyolab ID provided.")

    def get_account_timezone(
        self, *, game: typing.Optional[types.Game] = None, uid: typing.Optional[int] = None
    ) -> typing.Optional[int]:
        """Get the UTC timezone of the default game account."""
        game = game or self.game
        if game is None:
            return None

        uid = uid or self.uids.get(game)
        if uid is None:
            return None

        server = recognize_server(uid, game=game)  # type: ignore[arg-type]

        for tz, servers in constants.SERVER_TIMEZONES.items():
            if server in servers:
                return tz

        return None

    def _add_timezone_to_data(
        self,
        data: typing.Mapping[str, typing.Any],
        keys: typing.Sequence[str],
        *,
        game: typing.Optional[types.Game] = None,
        uid: typing.Optional[int] = None,
    ) -> typing.Mapping[str, typing.Any]:
        """Recursively add timezone info to a data dict based on the default game account.

        Datetime dicts found under the given keys get a "tzinfo" entry added, unix timestamp
        values get wrapped into {"timestamp": ..., "tzinfo": ...} dicts.
        """
        tz = self.get_account_timezone(game=game, uid=uid)
        if tz is None:
            return data

        def add_tz(value: typing.Any) -> None:
            if isinstance(value, dict):
                value = typing.cast("typing.Dict[str, typing.Any]", value)
                for key, item in value.items():
                    if key in keys and isinstance(item, dict) and item:
                        item["tzinfo"] = tz
                    elif key in keys and (isinstance(item, int) or (isinstance(item, str) and item.isdigit())):
                        value[key] = {"timestamp": item, "tzinfo": tz}
                    else:
                        add_tz(item)
            elif isinstance(value, list):
                for item in typing.cast("typing.List[typing.Any]", value):
                    add_tz(item)

        data = dict(data)
        add_tz(data)
        return data

var cache : BaseCache

Expand source code

class BaseClient(abc.ABC):
    """Base ABC Client."""

    __slots__ = (
        "cookie_manager",
        "cache",
        "_lang",
        "_region",
        "_default_game",
        "uids",
        "authkeys",
        "_hoyolab_id",
        "_accounts",
        "custom_headers",
    )

    USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36"  # noqa: E501

    logger: logging.Logger = logging.getLogger(__name__)

    cookie_manager: managers.BaseCookieManager
    cache: client_cache.BaseCache
    _lang: str
    _region: types.Region
    _default_game: typing.Optional[types.Game]

    uids: dict[types.Game, int]
    authkeys: dict[types.Game, str]
    _hoyolab_id: typing.Optional[int]
    _accounts: dict[types.Game, hoyolab_models.GenshinAccount]
    custom_headers: multidict.CIMultiDict[str]

    def __init__(
        self,
        cookies: typing.Optional[managers.AnyCookieOrHeader] = None,
        *,
        authkey: typing.Optional[str] = None,
        lang: types.Lang = "en-us",
        region: types.Region = types.Region.OVERSEAS,
        proxy: typing.Optional[str] = None,
        game: typing.Optional[types.Game] = None,
        uid: typing.Optional[int] = None,
        hoyolab_id: typing.Optional[int] = None,
        device_id: typing.Optional[str] = None,
        device_fp: typing.Optional[str] = None,
        headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
        cache: typing.Optional[client_cache.BaseCache] = None,
        debug: bool = False,
        on_cookie_update: typing.Optional[managers.CookieUpdateHook] = None,
    ) -> None:
        self.cookie_manager = managers.BaseCookieManager.from_cookies(cookies)
        self.on_cookie_update = on_cookie_update
        self.cache = cache or client_cache.StaticCache()

        self.uids = {}
        self.authkeys = {}
        self._accounts = {}

        self.default_game = game
        self.lang = lang
        self.region = region
        self.authkey = authkey
        self.debug = debug
        self.proxy = proxy
        self.uid = uid
        self.hoyolab_id = hoyolab_id

        self.custom_headers = parse_loose_headers(headers)
        self.custom_headers.update({"x-rpc-device_id": device_id} if device_id else {})
        self.custom_headers.update({"x-rpc-device_fp": device_fp} if device_fp else {})

    def __repr__(self) -> str:
        kwargs = dict(
            lang=self.lang,
            region=self.region.value,
            default_game=self.default_game and self.default_game.value,
            hoyolab_id=self.hoyolab_id,
            uid=self.default_game and self.uid,
            authkey=self.authkey and self.authkey[:12] + "...",
            proxy=self.proxy,
            debug=self.debug,
        )
        return f"<{type(self).__name__} {', '.join(f'{k}={v!r}' for k, v in kwargs.items() if v)}>"

    @property
    def device_id(self) -> typing.Optional[str]:
        """The device id used in headers."""
        return self.custom_headers.get("x-rpc-device_id")

    @device_id.setter
    def device_id(self, device_id: str) -> None:
        self.custom_headers["x-rpc-device_id"] = device_id

    @property
    def device_fp(self) -> typing.Optional[str]:
        """The device fingerprint used in headers."""
        return self.custom_headers.get("x-rpc-device_fp")

    @device_fp.setter
    def device_fp(self, device_fp: str) -> None:
        self.custom_headers["x-rpc-device_fp"] = device_fp

    @property
    def hoyolab_id(self) -> typing.Optional[int]:
        """The logged-in user's hoyolab uid.

        Returns None if not found or not applicable.
        """
        return self._hoyolab_id or self.cookie_manager.user_id

    @hoyolab_id.setter
    def hoyolab_id(self, hoyolab_id: typing.Optional[int]) -> None:
        if hoyolab_id is None:
            self._hoyolab_id = None
            return

        if self.cookie_manager.multi:
            raise RuntimeError("Cannot specify a hoyolab uid when using multiple cookies.")

        if self.cookie_manager.user_id and hoyolab_id and self.cookie_manager.user_id != hoyolab_id:
            raise ValueError("The provided hoyolab uid does not match the cookie id.")

        self._hoyolab_id = hoyolab_id

    @property
    def lang(self) -> str:
        """The default language, defaults to "en-us" """
        return self._lang

    @lang.setter
    def lang(self, lang: str) -> None:
        if lang not in constants.LANGS:
            raise ValueError(f"{lang} is not a valid language, must be one of: " + ", ".join(constants.LANGS))

        self._lang = lang

    @property
    def region(self) -> types.Region:
        """The default region."""
        return self._region

    @region.setter
    def region(self, region: str) -> None:
        self._region = types.Region(region)

        if region == types.Region.CHINESE:
            self.lang = "zh-cn"

    @property
    def default_game(self) -> typing.Optional[types.Game]:
        """The default game."""
        return self._default_game

    @default_game.setter
    def default_game(self, game: typing.Optional[str]) -> None:
        self._default_game = types.Game(game) if game else None

    game = default_game

    @property
    def uid(self) -> typing.Optional[int]:
        """UID of the default game."""
        if self.default_game is None:
            if len(self.uids) != 1:
                return None

            (self.default_game,) = self.uids.keys()

        return self.uids.get(self.default_game)

    @uid.setter
    def uid(self, uid: typing.Optional[int]) -> None:
        if uid is None:
            self.uids.clear()
            return

        self._default_game = self._default_game or utility.recognize_game(uid, region=self.region)
        if self.default_game is None:
            raise RuntimeError("No default game set. Cannot set uid.")

        self.uids[self.default_game] = uid

    @property
    def authkey(self) -> typing.Optional[str]:
        """The default genshin authkey used for paginators."""
        if self.default_game is None:
            if self.authkeys:
                warnings.warn("Tried to get an authkey without a default game set.")

            return None

        return self.authkeys.get(self.default_game)

    @authkey.setter
    def authkey(self, authkey: typing.Optional[str]) -> None:
        if authkey is None:
            self.authkeys.clear()
            return

        authkey = urllib.parse.unquote(authkey)

        try:
            base64.b64decode(authkey, validate=True)
        except Exception as e:
            raise ValueError("authkey is not a valid base64 encoded string") from e

        if not self.default_game:
            raise RuntimeError("No default game set. Cannot set authkey with property.")

        self.authkeys[self.default_game] = authkey

    @property
    def debug(self) -> bool:
        """Whether the debug logs are being shown in stdout"""
        return logging.getLogger("genshin").level == logging.DEBUG

    @debug.setter
    def debug(self, debug: bool) -> None:
        logging.basicConfig()
        level = logging.DEBUG if debug else logging.NOTSET
        logging.getLogger("genshin").setLevel(level)

    @property
    def on_cookie_update(self) -> typing.Optional[managers.CookieUpdateHook]:
        """Callback invoked with a copy of the cookies whenever the client updates them itself.

        This happens for example when a new cookie_token is minted from an stoken or when
        a response sets new cookies. May be a sync or async callable. Use it to persist
        refreshed cookies so the next client created from storage does not need to refresh again.
        """
        return self.cookie_manager.on_cookie_update

    @on_cookie_update.setter
    def on_cookie_update(self, hook: typing.Optional[managers.CookieUpdateHook]) -> None:
        self.cookie_manager.on_cookie_update = hook

    def set_cookies(self, cookies: typing.Optional[managers.AnyCookieOrHeader] = None, **kwargs: typing.Any) -> None:
        """Parse and set cookies."""
        if not bool(cookies) ^ bool(kwargs):
            raise TypeError("Cannot use both positional and keyword arguments at once")

        hook = self.on_cookie_update
        self.cookie_manager = managers.BaseCookieManager.from_cookies(cookies or kwargs)
        self.on_cookie_update = hook

    def set_browser_cookies(self, browser: typing.Optional[str] = None) -> None:
        """Extract cookies from your browser and set them as client cookies.

        Available browsers: chrome, chromium, opera, edge, firefox.
        """
        hook = self.on_cookie_update
        self.cookie_manager = managers.BaseCookieManager.from_browser_cookies(browser)
        self.on_cookie_update = hook

    def set_authkey(self, authkey: typing.Optional[str] = None, *, game: typing.Optional[types.Game] = None) -> None:
        """Set an authkey for wish & transaction logs.

        Accepts an authkey, a url containing an authkey or a path towards a logfile.
        """
        if authkey is None or os.path.isfile(authkey):
            authkey = utility.get_authkey(authkey)
        else:
            authkey = utility.extract_authkey(authkey) or authkey

        game = game or self.default_game
        if game is None:
            raise RuntimeError("No default game set.")

        self.authkeys[game] = authkey

    def set_cache(
        self, maxsize: int = 1024, *, ttl: int = client_cache.HOUR, static_ttl: int = client_cache.DAY
    ) -> None:
        """Create and set a new cache."""
        self.cache = client_cache.Cache(maxsize, ttl=ttl, static_ttl=static_ttl)

    def set_redis_cache(
        self, url: str, *, ttl: int = client_cache.HOUR, static_ttl: int = client_cache.DAY, **redis_kwargs: typing.Any
    ) -> None:
        """Create and set a new redis cache."""
        import aioredis

        redis = aioredis.Redis.from_url(url, **redis_kwargs)  # pyright: ignore[reportUnknownMemberType]
        self.cache = client_cache.RedisCache(redis, ttl=ttl, static_ttl=static_ttl)

    @property
    def proxy(self) -> typing.Optional[str]:
        """Proxy for http requests."""
        if self.cookie_manager.proxy is None:
            return None

        return str(self.cookie_manager.proxy)

    @proxy.setter
    def proxy(self, proxy: typing.Optional[aiohttp.typedefs.StrOrURL]) -> None:
        self.cookie_manager.proxy = yarl.URL(proxy) if proxy else None

    async def _request_hook(
        self,
        method: str,
        url: aiohttp.typedefs.StrOrURL,
        *,
        params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        data: typing.Any = None,
        **kwargs: typing.Any,
    ) -> None:
        """Perform an action before a request.

        Debug logging by default.
        """
        url = yarl.URL(url)
        if params:
            params = {k: v for k, v in params.items() if k != "authkey"}
            url = url.update_query(params)

        if data:
            self.logger.debug("%s %s\n%s", method, url, json.dumps(data, separators=(",", ":")))
        else:
            self.logger.debug("%s %s", method, url)

    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 = parse_loose_headers(headers)
        headers["User-Agent"] = self.USER_AGENT
        headers.update(self.custom_headers)

        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

    async def request_webstatic(
        self,
        url: aiohttp.typedefs.StrOrURL,
        *,
        headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
        cache: typing.Any = None,
        region: types.Region = types.Region.OVERSEAS,
        **kwargs: typing.Any,
    ) -> typing.Any:
        """Request a static json file."""
        if cache is not None:
            value = await self.cache.get_static(cache)
            if value is not None:
                return value

        url = routes.WEBSTATIC_URL.get_url(region).join(yarl.URL(url))

        headers = parse_loose_headers(headers)
        headers["User-Agent"] = self.USER_AGENT
        headers.update(self.custom_headers)

        await self._request_hook("GET", url, headers=headers, **kwargs)

        async with self.cookie_manager.create_session() as session:
            async with session.get(url, headers=headers, proxy=self.proxy, **kwargs) as r:
                r.raise_for_status()
                data = await r.json()

        if cache is not None:
            await self.cache.set_static(cache, data)

        return data

    async def request_bbs(
        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 bbs 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.BBS_URL.get_url(region).join(yarl.URL(url))

        headers = parse_loose_headers(headers)
        headers.update(ds.get_ds_headers(data=data, params=params, region=region, lang=lang or self.lang))
        headers["Referer"] = str(routes.BBS_REFERER_URL.get_url(self.region))

        data = await self.request(url, method=method, params=params, data=data, headers=headers, **kwargs)
        return data

    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))

        headers = parse_loose_headers(headers)
        headers.update(ds.get_ds_headers(data=data, params=params, region=region, lang=lang or self.lang))

        data = await self.request(url, method=method, params=params, data=data, headers=headers, **kwargs)
        return data

    @managers.no_multi
    async def get_game_accounts(
        self, *, lang: typing.Optional[str] = None
    ) -> typing.Sequence[hoyolab_models.GenshinAccount]:
        """Get the game accounts of the currently logged-in user."""
        if self.hoyolab_id is None:
            warnings.warn("No hoyolab id set, caching may be unreliable.")

        data = await self.request_hoyolab(
            "binding/api/getUserGameRolesByCookie",
            lang=lang,
            cache=client_cache.cache_key("accounts", hoyolab_id=self.hoyolab_id),
        )
        return [hoyolab_models.GenshinAccount(**i) for i in data["list"]]

    @deprecation.deprecated("get_game_accounts")
    async def genshin_accounts(
        self, *, lang: typing.Optional[str] = None
    ) -> typing.Sequence[hoyolab_models.GenshinAccount]:
        """Get the genshin accounts of the currently logged-in user."""
        accounts = await self.get_game_accounts(lang=lang)
        return [account for account in accounts if account.game == types.Game.GENSHIN]

    async def _update_cached_uids(self) -> None:
        """Update cached fallback uids."""
        mixed_accounts = await self.get_game_accounts()

        game_accounts: dict[types.Game, list[hoyolab_models.GenshinAccount]] = {}
        for account in mixed_accounts:
            if not isinstance(account.game, types.Game):  # pyright: ignore[reportUnnecessaryIsInstance]
                continue

            game_accounts.setdefault(account.game, []).append(account)

        self.uids = {game: max(accounts, key=lambda a: a.level).uid for game, accounts in game_accounts.items()}

        if len(self.uids) == 1 and self.default_game is None:
            (self.default_game,) = self.uids.keys()

    @concurrency.prevent_concurrency
    async def _get_uid(self, game: types.Game) -> int:
        """Get a cached fallback uid."""
        # TODO: use lock
        if uid := self.uids.get(game):
            return uid

        if self.cookie_manager.multi:
            raise RuntimeError("UID must be provided when using multi-cookie managers.")

        await self._update_cached_uids()

        if uid := self.uids.get(game):
            return uid

        raise errors.AccountNotFound(msg="No UID provided and account has no game account bound to it.")

    async def _update_cached_accounts(self) -> None:
        """Update cached fallback accounts."""
        mixed_accounts = await self.get_game_accounts()

        game_accounts: dict[types.Game, list[hoyolab_models.GenshinAccount]] = {}
        for account in mixed_accounts:
            if not isinstance(account.game, types.Game):  # pyright: ignore[reportUnnecessaryIsInstance]
                continue

            game_accounts.setdefault(account.game, []).append(account)

        self._accounts = {}
        for game, accounts in game_accounts.items():
            self._accounts[game] = next(
                (acc for acc in accounts if acc.uid == self.uids.get(game)), max(accounts, key=lambda a: a.level)
            )

    @concurrency.prevent_concurrency
    async def _get_account(self, game: types.Game) -> hoyolab_models.GenshinAccount:
        """Get a cached fallback account."""
        if (account := self._accounts.get(game)) and (uid := self.uids.get(game)) and account.uid == uid:
            return account

        await self._update_cached_accounts()

        if account := self._accounts.get(game):
            if (uid := self.uids.get(game)) and account.uid != uid:
                raise errors.AccountNotFound(msg="There is no game account with such UID.")

            return account

        raise errors.AccountNotFound(msg="Account has no game account bound to it.")

    def _get_hoyolab_id(self) -> int:
        """Get a cached fallback hoyolab ID."""
        if self.hoyolab_id is not None:
            return self.hoyolab_id

        if self.cookie_manager.multi:
            raise RuntimeError("Hoyolab ID must be provided when using multi-cookie managers.")

        raise RuntimeError("No default hoyolab ID provided.")

    def get_account_timezone(
        self, *, game: typing.Optional[types.Game] = None, uid: typing.Optional[int] = None
    ) -> typing.Optional[int]:
        """Get the UTC timezone of the default game account."""
        game = game or self.game
        if game is None:
            return None

        uid = uid or self.uids.get(game)
        if uid is None:
            return None

        server = recognize_server(uid, game=game)  # type: ignore[arg-type]

        for tz, servers in constants.SERVER_TIMEZONES.items():
            if server in servers:
                return tz

        return None

    def _add_timezone_to_data(
        self,
        data: typing.Mapping[str, typing.Any],
        keys: typing.Sequence[str],
        *,
        game: typing.Optional[types.Game] = None,
        uid: typing.Optional[int] = None,
    ) -> typing.Mapping[str, typing.Any]:
        """Recursively add timezone info to a data dict based on the default game account.

        Datetime dicts found under the given keys get a "tzinfo" entry added, unix timestamp
        values get wrapped into {"timestamp": ..., "tzinfo": ...} dicts.
        """
        tz = self.get_account_timezone(game=game, uid=uid)
        if tz is None:
            return data

        def add_tz(value: typing.Any) -> None:
            if isinstance(value, dict):
                value = typing.cast("typing.Dict[str, typing.Any]", value)
                for key, item in value.items():
                    if key in keys and isinstance(item, dict) and item:
                        item["tzinfo"] = tz
                    elif key in keys and (isinstance(item, int) or (isinstance(item, str) and item.isdigit())):
                        value[key] = {"timestamp": item, "tzinfo": tz}
                    else:
                        add_tz(item)
            elif isinstance(value, list):
                for item in typing.cast("typing.List[typing.Any]", value):
                    add_tz(item)

        data = dict(data)
        add_tz(data)
        return data

var cookie_manager : BaseCookieManager

Expand source code

class BaseClient(abc.ABC):
    """Base ABC Client."""

    __slots__ = (
        "cookie_manager",
        "cache",
        "_lang",
        "_region",
        "_default_game",
        "uids",
        "authkeys",
        "_hoyolab_id",
        "_accounts",
        "custom_headers",
    )

    USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36"  # noqa: E501

    logger: logging.Logger = logging.getLogger(__name__)

    cookie_manager: managers.BaseCookieManager
    cache: client_cache.BaseCache
    _lang: str
    _region: types.Region
    _default_game: typing.Optional[types.Game]

    uids: dict[types.Game, int]
    authkeys: dict[types.Game, str]
    _hoyolab_id: typing.Optional[int]
    _accounts: dict[types.Game, hoyolab_models.GenshinAccount]
    custom_headers: multidict.CIMultiDict[str]

    def __init__(
        self,
        cookies: typing.Optional[managers.AnyCookieOrHeader] = None,
        *,
        authkey: typing.Optional[str] = None,
        lang: types.Lang = "en-us",
        region: types.Region = types.Region.OVERSEAS,
        proxy: typing.Optional[str] = None,
        game: typing.Optional[types.Game] = None,
        uid: typing.Optional[int] = None,
        hoyolab_id: typing.Optional[int] = None,
        device_id: typing.Optional[str] = None,
        device_fp: typing.Optional[str] = None,
        headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
        cache: typing.Optional[client_cache.BaseCache] = None,
        debug: bool = False,
        on_cookie_update: typing.Optional[managers.CookieUpdateHook] = None,
    ) -> None:
        self.cookie_manager = managers.BaseCookieManager.from_cookies(cookies)
        self.on_cookie_update = on_cookie_update
        self.cache = cache or client_cache.StaticCache()

        self.uids = {}
        self.authkeys = {}
        self._accounts = {}

        self.default_game = game
        self.lang = lang
        self.region = region
        self.authkey = authkey
        self.debug = debug
        self.proxy = proxy
        self.uid = uid
        self.hoyolab_id = hoyolab_id

        self.custom_headers = parse_loose_headers(headers)
        self.custom_headers.update({"x-rpc-device_id": device_id} if device_id else {})
        self.custom_headers.update({"x-rpc-device_fp": device_fp} if device_fp else {})

    def __repr__(self) -> str:
        kwargs = dict(
            lang=self.lang,
            region=self.region.value,
            default_game=self.default_game and self.default_game.value,
            hoyolab_id=self.hoyolab_id,
            uid=self.default_game and self.uid,
            authkey=self.authkey and self.authkey[:12] + "...",
            proxy=self.proxy,
            debug=self.debug,
        )
        return f"<{type(self).__name__} {', '.join(f'{k}={v!r}' for k, v in kwargs.items() if v)}>"

    @property
    def device_id(self) -> typing.Optional[str]:
        """The device id used in headers."""
        return self.custom_headers.get("x-rpc-device_id")

    @device_id.setter
    def device_id(self, device_id: str) -> None:
        self.custom_headers["x-rpc-device_id"] = device_id

    @property
    def device_fp(self) -> typing.Optional[str]:
        """The device fingerprint used in headers."""
        return self.custom_headers.get("x-rpc-device_fp")

    @device_fp.setter
    def device_fp(self, device_fp: str) -> None:
        self.custom_headers["x-rpc-device_fp"] = device_fp

    @property
    def hoyolab_id(self) -> typing.Optional[int]:
        """The logged-in user's hoyolab uid.

        Returns None if not found or not applicable.
        """
        return self._hoyolab_id or self.cookie_manager.user_id

    @hoyolab_id.setter
    def hoyolab_id(self, hoyolab_id: typing.Optional[int]) -> None:
        if hoyolab_id is None:
            self._hoyolab_id = None
            return

        if self.cookie_manager.multi:
            raise RuntimeError("Cannot specify a hoyolab uid when using multiple cookies.")

        if self.cookie_manager.user_id and hoyolab_id and self.cookie_manager.user_id != hoyolab_id:
            raise ValueError("The provided hoyolab uid does not match the cookie id.")

        self._hoyolab_id = hoyolab_id

    @property
    def lang(self) -> str:
        """The default language, defaults to "en-us" """
        return self._lang

    @lang.setter
    def lang(self, lang: str) -> None:
        if lang not in constants.LANGS:
            raise ValueError(f"{lang} is not a valid language, must be one of: " + ", ".join(constants.LANGS))

        self._lang = lang

    @property
    def region(self) -> types.Region:
        """The default region."""
        return self._region

    @region.setter
    def region(self, region: str) -> None:
        self._region = types.Region(region)

        if region == types.Region.CHINESE:
            self.lang = "zh-cn"

    @property
    def default_game(self) -> typing.Optional[types.Game]:
        """The default game."""
        return self._default_game

    @default_game.setter
    def default_game(self, game: typing.Optional[str]) -> None:
        self._default_game = types.Game(game) if game else None

    game = default_game

    @property
    def uid(self) -> typing.Optional[int]:
        """UID of the default game."""
        if self.default_game is None:
            if len(self.uids) != 1:
                return None

            (self.default_game,) = self.uids.keys()

        return self.uids.get(self.default_game)

    @uid.setter
    def uid(self, uid: typing.Optional[int]) -> None:
        if uid is None:
            self.uids.clear()
            return

        self._default_game = self._default_game or utility.recognize_game(uid, region=self.region)
        if self.default_game is None:
            raise RuntimeError("No default game set. Cannot set uid.")

        self.uids[self.default_game] = uid

    @property
    def authkey(self) -> typing.Optional[str]:
        """The default genshin authkey used for paginators."""
        if self.default_game is None:
            if self.authkeys:
                warnings.warn("Tried to get an authkey without a default game set.")

            return None

        return self.authkeys.get(self.default_game)

    @authkey.setter
    def authkey(self, authkey: typing.Optional[str]) -> None:
        if authkey is None:
            self.authkeys.clear()
            return

        authkey = urllib.parse.unquote(authkey)

        try:
            base64.b64decode(authkey, validate=True)
        except Exception as e:
            raise ValueError("authkey is not a valid base64 encoded string") from e

        if not self.default_game:
            raise RuntimeError("No default game set. Cannot set authkey with property.")

        self.authkeys[self.default_game] = authkey

    @property
    def debug(self) -> bool:
        """Whether the debug logs are being shown in stdout"""
        return logging.getLogger("genshin").level == logging.DEBUG

    @debug.setter
    def debug(self, debug: bool) -> None:
        logging.basicConfig()
        level = logging.DEBUG if debug else logging.NOTSET
        logging.getLogger("genshin").setLevel(level)

    @property
    def on_cookie_update(self) -> typing.Optional[managers.CookieUpdateHook]:
        """Callback invoked with a copy of the cookies whenever the client updates them itself.

        This happens for example when a new cookie_token is minted from an stoken or when
        a response sets new cookies. May be a sync or async callable. Use it to persist
        refreshed cookies so the next client created from storage does not need to refresh again.
        """
        return self.cookie_manager.on_cookie_update

    @on_cookie_update.setter
    def on_cookie_update(self, hook: typing.Optional[managers.CookieUpdateHook]) -> None:
        self.cookie_manager.on_cookie_update = hook

    def set_cookies(self, cookies: typing.Optional[managers.AnyCookieOrHeader] = None, **kwargs: typing.Any) -> None:
        """Parse and set cookies."""
        if not bool(cookies) ^ bool(kwargs):
            raise TypeError("Cannot use both positional and keyword arguments at once")

        hook = self.on_cookie_update
        self.cookie_manager = managers.BaseCookieManager.from_cookies(cookies or kwargs)
        self.on_cookie_update = hook

    def set_browser_cookies(self, browser: typing.Optional[str] = None) -> None:
        """Extract cookies from your browser and set them as client cookies.

        Available browsers: chrome, chromium, opera, edge, firefox.
        """
        hook = self.on_cookie_update
        self.cookie_manager = managers.BaseCookieManager.from_browser_cookies(browser)
        self.on_cookie_update = hook

    def set_authkey(self, authkey: typing.Optional[str] = None, *, game: typing.Optional[types.Game] = None) -> None:
        """Set an authkey for wish & transaction logs.

        Accepts an authkey, a url containing an authkey or a path towards a logfile.
        """
        if authkey is None or os.path.isfile(authkey):
            authkey = utility.get_authkey(authkey)
        else:
            authkey = utility.extract_authkey(authkey) or authkey

        game = game or self.default_game
        if game is None:
            raise RuntimeError("No default game set.")

        self.authkeys[game] = authkey

    def set_cache(
        self, maxsize: int = 1024, *, ttl: int = client_cache.HOUR, static_ttl: int = client_cache.DAY
    ) -> None:
        """Create and set a new cache."""
        self.cache = client_cache.Cache(maxsize, ttl=ttl, static_ttl=static_ttl)

    def set_redis_cache(
        self, url: str, *, ttl: int = client_cache.HOUR, static_ttl: int = client_cache.DAY, **redis_kwargs: typing.Any
    ) -> None:
        """Create and set a new redis cache."""
        import aioredis

        redis = aioredis.Redis.from_url(url, **redis_kwargs)  # pyright: ignore[reportUnknownMemberType]
        self.cache = client_cache.RedisCache(redis, ttl=ttl, static_ttl=static_ttl)

    @property
    def proxy(self) -> typing.Optional[str]:
        """Proxy for http requests."""
        if self.cookie_manager.proxy is None:
            return None

        return str(self.cookie_manager.proxy)

    @proxy.setter
    def proxy(self, proxy: typing.Optional[aiohttp.typedefs.StrOrURL]) -> None:
        self.cookie_manager.proxy = yarl.URL(proxy) if proxy else None

    async def _request_hook(
        self,
        method: str,
        url: aiohttp.typedefs.StrOrURL,
        *,
        params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        data: typing.Any = None,
        **kwargs: typing.Any,
    ) -> None:
        """Perform an action before a request.

        Debug logging by default.
        """
        url = yarl.URL(url)
        if params:
            params = {k: v for k, v in params.items() if k != "authkey"}
            url = url.update_query(params)

        if data:
            self.logger.debug("%s %s\n%s", method, url, json.dumps(data, separators=(",", ":")))
        else:
            self.logger.debug("%s %s", method, url)

    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 = parse_loose_headers(headers)
        headers["User-Agent"] = self.USER_AGENT
        headers.update(self.custom_headers)

        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

    async def request_webstatic(
        self,
        url: aiohttp.typedefs.StrOrURL,
        *,
        headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
        cache: typing.Any = None,
        region: types.Region = types.Region.OVERSEAS,
        **kwargs: typing.Any,
    ) -> typing.Any:
        """Request a static json file."""
        if cache is not None:
            value = await self.cache.get_static(cache)
            if value is not None:
                return value

        url = routes.WEBSTATIC_URL.get_url(region).join(yarl.URL(url))

        headers = parse_loose_headers(headers)
        headers["User-Agent"] = self.USER_AGENT
        headers.update(self.custom_headers)

        await self._request_hook("GET", url, headers=headers, **kwargs)

        async with self.cookie_manager.create_session() as session:
            async with session.get(url, headers=headers, proxy=self.proxy, **kwargs) as r:
                r.raise_for_status()
                data = await r.json()

        if cache is not None:
            await self.cache.set_static(cache, data)

        return data

    async def request_bbs(
        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 bbs 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.BBS_URL.get_url(region).join(yarl.URL(url))

        headers = parse_loose_headers(headers)
        headers.update(ds.get_ds_headers(data=data, params=params, region=region, lang=lang or self.lang))
        headers["Referer"] = str(routes.BBS_REFERER_URL.get_url(self.region))

        data = await self.request(url, method=method, params=params, data=data, headers=headers, **kwargs)
        return data

    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))

        headers = parse_loose_headers(headers)
        headers.update(ds.get_ds_headers(data=data, params=params, region=region, lang=lang or self.lang))

        data = await self.request(url, method=method, params=params, data=data, headers=headers, **kwargs)
        return data

    @managers.no_multi
    async def get_game_accounts(
        self, *, lang: typing.Optional[str] = None
    ) -> typing.Sequence[hoyolab_models.GenshinAccount]:
        """Get the game accounts of the currently logged-in user."""
        if self.hoyolab_id is None:
            warnings.warn("No hoyolab id set, caching may be unreliable.")

        data = await self.request_hoyolab(
            "binding/api/getUserGameRolesByCookie",
            lang=lang,
            cache=client_cache.cache_key("accounts", hoyolab_id=self.hoyolab_id),
        )
        return [hoyolab_models.GenshinAccount(**i) for i in data["list"]]

    @deprecation.deprecated("get_game_accounts")
    async def genshin_accounts(
        self, *, lang: typing.Optional[str] = None
    ) -> typing.Sequence[hoyolab_models.GenshinAccount]:
        """Get the genshin accounts of the currently logged-in user."""
        accounts = await self.get_game_accounts(lang=lang)
        return [account for account in accounts if account.game == types.Game.GENSHIN]

    async def _update_cached_uids(self) -> None:
        """Update cached fallback uids."""
        mixed_accounts = await self.get_game_accounts()

        game_accounts: dict[types.Game, list[hoyolab_models.GenshinAccount]] = {}
        for account in mixed_accounts:
            if not isinstance(account.game, types.Game):  # pyright: ignore[reportUnnecessaryIsInstance]
                continue

            game_accounts.setdefault(account.game, []).append(account)

        self.uids = {game: max(accounts, key=lambda a: a.level).uid for game, accounts in game_accounts.items()}

        if len(self.uids) == 1 and self.default_game is None:
            (self.default_game,) = self.uids.keys()

    @concurrency.prevent_concurrency
    async def _get_uid(self, game: types.Game) -> int:
        """Get a cached fallback uid."""
        # TODO: use lock
        if uid := self.uids.get(game):
            return uid

        if self.cookie_manager.multi:
            raise RuntimeError("UID must be provided when using multi-cookie managers.")

        await self._update_cached_uids()

        if uid := self.uids.get(game):
            return uid

        raise errors.AccountNotFound(msg="No UID provided and account has no game account bound to it.")

    async def _update_cached_accounts(self) -> None:
        """Update cached fallback accounts."""
        mixed_accounts = await self.get_game_accounts()

        game_accounts: dict[types.Game, list[hoyolab_models.GenshinAccount]] = {}
        for account in mixed_accounts:
            if not isinstance(account.game, types.Game):  # pyright: ignore[reportUnnecessaryIsInstance]
                continue

            game_accounts.setdefault(account.game, []).append(account)

        self._accounts = {}
        for game, accounts in game_accounts.items():
            self._accounts[game] = next(
                (acc for acc in accounts if acc.uid == self.uids.get(game)), max(accounts, key=lambda a: a.level)
            )

    @concurrency.prevent_concurrency
    async def _get_account(self, game: types.Game) -> hoyolab_models.GenshinAccount:
        """Get a cached fallback account."""
        if (account := self._accounts.get(game)) and (uid := self.uids.get(game)) and account.uid == uid:
            return account

        await self._update_cached_accounts()

        if account := self._accounts.get(game):
            if (uid := self.uids.get(game)) and account.uid != uid:
                raise errors.AccountNotFound(msg="There is no game account with such UID.")

            return account

        raise errors.AccountNotFound(msg="Account has no game account bound to it.")

    def _get_hoyolab_id(self) -> int:
        """Get a cached fallback hoyolab ID."""
        if self.hoyolab_id is not None:
            return self.hoyolab_id

        if self.cookie_manager.multi:
            raise RuntimeError("Hoyolab ID must be provided when using multi-cookie managers.")

        raise RuntimeError("No default hoyolab ID provided.")

    def get_account_timezone(
        self, *, game: typing.Optional[types.Game] = None, uid: typing.Optional[int] = None
    ) -> typing.Optional[int]:
        """Get the UTC timezone of the default game account."""
        game = game or self.game
        if game is None:
            return None

        uid = uid or self.uids.get(game)
        if uid is None:
            return None

        server = recognize_server(uid, game=game)  # type: ignore[arg-type]

        for tz, servers in constants.SERVER_TIMEZONES.items():
            if server in servers:
                return tz

        return None

    def _add_timezone_to_data(
        self,
        data: typing.Mapping[str, typing.Any],
        keys: typing.Sequence[str],
        *,
        game: typing.Optional[types.Game] = None,
        uid: typing.Optional[int] = None,
    ) -> typing.Mapping[str, typing.Any]:
        """Recursively add timezone info to a data dict based on the default game account.

        Datetime dicts found under the given keys get a "tzinfo" entry added, unix timestamp
        values get wrapped into {"timestamp": ..., "tzinfo": ...} dicts.
        """
        tz = self.get_account_timezone(game=game, uid=uid)
        if tz is None:
            return data

        def add_tz(value: typing.Any) -> None:
            if isinstance(value, dict):
                value = typing.cast("typing.Dict[str, typing.Any]", value)
                for key, item in value.items():
                    if key in keys and isinstance(item, dict) and item:
                        item["tzinfo"] = tz
                    elif key in keys and (isinstance(item, int) or (isinstance(item, str) and item.isdigit())):
                        value[key] = {"timestamp": item, "tzinfo": tz}
                    else:
                        add_tz(item)
            elif isinstance(value, list):
                for item in typing.cast("typing.List[typing.Any]", value):
                    add_tz(item)

        data = dict(data)
        add_tz(data)
        return data

var custom_headers : multidict._multidict.CIMultiDict[str]

Expand source code

class BaseClient(abc.ABC):
    """Base ABC Client."""

    __slots__ = (
        "cookie_manager",
        "cache",
        "_lang",
        "_region",
        "_default_game",
        "uids",
        "authkeys",
        "_hoyolab_id",
        "_accounts",
        "custom_headers",
    )

    USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36"  # noqa: E501

    logger: logging.Logger = logging.getLogger(__name__)

    cookie_manager: managers.BaseCookieManager
    cache: client_cache.BaseCache
    _lang: str
    _region: types.Region
    _default_game: typing.Optional[types.Game]

    uids: dict[types.Game, int]
    authkeys: dict[types.Game, str]
    _hoyolab_id: typing.Optional[int]
    _accounts: dict[types.Game, hoyolab_models.GenshinAccount]
    custom_headers: multidict.CIMultiDict[str]

    def __init__(
        self,
        cookies: typing.Optional[managers.AnyCookieOrHeader] = None,
        *,
        authkey: typing.Optional[str] = None,
        lang: types.Lang = "en-us",
        region: types.Region = types.Region.OVERSEAS,
        proxy: typing.Optional[str] = None,
        game: typing.Optional[types.Game] = None,
        uid: typing.Optional[int] = None,
        hoyolab_id: typing.Optional[int] = None,
        device_id: typing.Optional[str] = None,
        device_fp: typing.Optional[str] = None,
        headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
        cache: typing.Optional[client_cache.BaseCache] = None,
        debug: bool = False,
        on_cookie_update: typing.Optional[managers.CookieUpdateHook] = None,
    ) -> None:
        self.cookie_manager = managers.BaseCookieManager.from_cookies(cookies)
        self.on_cookie_update = on_cookie_update
        self.cache = cache or client_cache.StaticCache()

        self.uids = {}
        self.authkeys = {}
        self._accounts = {}

        self.default_game = game
        self.lang = lang
        self.region = region
        self.authkey = authkey
        self.debug = debug
        self.proxy = proxy
        self.uid = uid
        self.hoyolab_id = hoyolab_id

        self.custom_headers = parse_loose_headers(headers)
        self.custom_headers.update({"x-rpc-device_id": device_id} if device_id else {})
        self.custom_headers.update({"x-rpc-device_fp": device_fp} if device_fp else {})

    def __repr__(self) -> str:
        kwargs = dict(
            lang=self.lang,
            region=self.region.value,
            default_game=self.default_game and self.default_game.value,
            hoyolab_id=self.hoyolab_id,
            uid=self.default_game and self.uid,
            authkey=self.authkey and self.authkey[:12] + "...",
            proxy=self.proxy,
            debug=self.debug,
        )
        return f"<{type(self).__name__} {', '.join(f'{k}={v!r}' for k, v in kwargs.items() if v)}>"

    @property
    def device_id(self) -> typing.Optional[str]:
        """The device id used in headers."""
        return self.custom_headers.get("x-rpc-device_id")

    @device_id.setter
    def device_id(self, device_id: str) -> None:
        self.custom_headers["x-rpc-device_id"] = device_id

    @property
    def device_fp(self) -> typing.Optional[str]:
        """The device fingerprint used in headers."""
        return self.custom_headers.get("x-rpc-device_fp")

    @device_fp.setter
    def device_fp(self, device_fp: str) -> None:
        self.custom_headers["x-rpc-device_fp"] = device_fp

    @property
    def hoyolab_id(self) -> typing.Optional[int]:
        """The logged-in user's hoyolab uid.

        Returns None if not found or not applicable.
        """
        return self._hoyolab_id or self.cookie_manager.user_id

    @hoyolab_id.setter
    def hoyolab_id(self, hoyolab_id: typing.Optional[int]) -> None:
        if hoyolab_id is None:
            self._hoyolab_id = None
            return

        if self.cookie_manager.multi:
            raise RuntimeError("Cannot specify a hoyolab uid when using multiple cookies.")

        if self.cookie_manager.user_id and hoyolab_id and self.cookie_manager.user_id != hoyolab_id:
            raise ValueError("The provided hoyolab uid does not match the cookie id.")

        self._hoyolab_id = hoyolab_id

    @property
    def lang(self) -> str:
        """The default language, defaults to "en-us" """
        return self._lang

    @lang.setter
    def lang(self, lang: str) -> None:
        if lang not in constants.LANGS:
            raise ValueError(f"{lang} is not a valid language, must be one of: " + ", ".join(constants.LANGS))

        self._lang = lang

    @property
    def region(self) -> types.Region:
        """The default region."""
        return self._region

    @region.setter
    def region(self, region: str) -> None:
        self._region = types.Region(region)

        if region == types.Region.CHINESE:
            self.lang = "zh-cn"

    @property
    def default_game(self) -> typing.Optional[types.Game]:
        """The default game."""
        return self._default_game

    @default_game.setter
    def default_game(self, game: typing.Optional[str]) -> None:
        self._default_game = types.Game(game) if game else None

    game = default_game

    @property
    def uid(self) -> typing.Optional[int]:
        """UID of the default game."""
        if self.default_game is None:
            if len(self.uids) != 1:
                return None

            (self.default_game,) = self.uids.keys()

        return self.uids.get(self.default_game)

    @uid.setter
    def uid(self, uid: typing.Optional[int]) -> None:
        if uid is None:
            self.uids.clear()
            return

        self._default_game = self._default_game or utility.recognize_game(uid, region=self.region)
        if self.default_game is None:
            raise RuntimeError("No default game set. Cannot set uid.")

        self.uids[self.default_game] = uid

    @property
    def authkey(self) -> typing.Optional[str]:
        """The default genshin authkey used for paginators."""
        if self.default_game is None:
            if self.authkeys:
                warnings.warn("Tried to get an authkey without a default game set.")

            return None

        return self.authkeys.get(self.default_game)

    @authkey.setter
    def authkey(self, authkey: typing.Optional[str]) -> None:
        if authkey is None:
            self.authkeys.clear()
            return

        authkey = urllib.parse.unquote(authkey)

        try:
            base64.b64decode(authkey, validate=True)
        except Exception as e:
            raise ValueError("authkey is not a valid base64 encoded string") from e

        if not self.default_game:
            raise RuntimeError("No default game set. Cannot set authkey with property.")

        self.authkeys[self.default_game] = authkey

    @property
    def debug(self) -> bool:
        """Whether the debug logs are being shown in stdout"""
        return logging.getLogger("genshin").level == logging.DEBUG

    @debug.setter
    def debug(self, debug: bool) -> None:
        logging.basicConfig()
        level = logging.DEBUG if debug else logging.NOTSET
        logging.getLogger("genshin").setLevel(level)

    @property
    def on_cookie_update(self) -> typing.Optional[managers.CookieUpdateHook]:
        """Callback invoked with a copy of the cookies whenever the client updates them itself.

        This happens for example when a new cookie_token is minted from an stoken or when
        a response sets new cookies. May be a sync or async callable. Use it to persist
        refreshed cookies so the next client created from storage does not need to refresh again.
        """
        return self.cookie_manager.on_cookie_update

    @on_cookie_update.setter
    def on_cookie_update(self, hook: typing.Optional[managers.CookieUpdateHook]) -> None:
        self.cookie_manager.on_cookie_update = hook

    def set_cookies(self, cookies: typing.Optional[managers.AnyCookieOrHeader] = None, **kwargs: typing.Any) -> None:
        """Parse and set cookies."""
        if not bool(cookies) ^ bool(kwargs):
            raise TypeError("Cannot use both positional and keyword arguments at once")

        hook = self.on_cookie_update
        self.cookie_manager = managers.BaseCookieManager.from_cookies(cookies or kwargs)
        self.on_cookie_update = hook

    def set_browser_cookies(self, browser: typing.Optional[str] = None) -> None:
        """Extract cookies from your browser and set them as client cookies.

        Available browsers: chrome, chromium, opera, edge, firefox.
        """
        hook = self.on_cookie_update
        self.cookie_manager = managers.BaseCookieManager.from_browser_cookies(browser)
        self.on_cookie_update = hook

    def set_authkey(self, authkey: typing.Optional[str] = None, *, game: typing.Optional[types.Game] = None) -> None:
        """Set an authkey for wish & transaction logs.

        Accepts an authkey, a url containing an authkey or a path towards a logfile.
        """
        if authkey is None or os.path.isfile(authkey):
            authkey = utility.get_authkey(authkey)
        else:
            authkey = utility.extract_authkey(authkey) or authkey

        game = game or self.default_game
        if game is None:
            raise RuntimeError("No default game set.")

        self.authkeys[game] = authkey

    def set_cache(
        self, maxsize: int = 1024, *, ttl: int = client_cache.HOUR, static_ttl: int = client_cache.DAY
    ) -> None:
        """Create and set a new cache."""
        self.cache = client_cache.Cache(maxsize, ttl=ttl, static_ttl=static_ttl)

    def set_redis_cache(
        self, url: str, *, ttl: int = client_cache.HOUR, static_ttl: int = client_cache.DAY, **redis_kwargs: typing.Any
    ) -> None:
        """Create and set a new redis cache."""
        import aioredis

        redis = aioredis.Redis.from_url(url, **redis_kwargs)  # pyright: ignore[reportUnknownMemberType]
        self.cache = client_cache.RedisCache(redis, ttl=ttl, static_ttl=static_ttl)

    @property
    def proxy(self) -> typing.Optional[str]:
        """Proxy for http requests."""
        if self.cookie_manager.proxy is None:
            return None

        return str(self.cookie_manager.proxy)

    @proxy.setter
    def proxy(self, proxy: typing.Optional[aiohttp.typedefs.StrOrURL]) -> None:
        self.cookie_manager.proxy = yarl.URL(proxy) if proxy else None

    async def _request_hook(
        self,
        method: str,
        url: aiohttp.typedefs.StrOrURL,
        *,
        params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        data: typing.Any = None,
        **kwargs: typing.Any,
    ) -> None:
        """Perform an action before a request.

        Debug logging by default.
        """
        url = yarl.URL(url)
        if params:
            params = {k: v for k, v in params.items() if k != "authkey"}
            url = url.update_query(params)

        if data:
            self.logger.debug("%s %s\n%s", method, url, json.dumps(data, separators=(",", ":")))
        else:
            self.logger.debug("%s %s", method, url)

    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 = parse_loose_headers(headers)
        headers["User-Agent"] = self.USER_AGENT
        headers.update(self.custom_headers)

        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

    async def request_webstatic(
        self,
        url: aiohttp.typedefs.StrOrURL,
        *,
        headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
        cache: typing.Any = None,
        region: types.Region = types.Region.OVERSEAS,
        **kwargs: typing.Any,
    ) -> typing.Any:
        """Request a static json file."""
        if cache is not None:
            value = await self.cache.get_static(cache)
            if value is not None:
                return value

        url = routes.WEBSTATIC_URL.get_url(region).join(yarl.URL(url))

        headers = parse_loose_headers(headers)
        headers["User-Agent"] = self.USER_AGENT
        headers.update(self.custom_headers)

        await self._request_hook("GET", url, headers=headers, **kwargs)

        async with self.cookie_manager.create_session() as session:
            async with session.get(url, headers=headers, proxy=self.proxy, **kwargs) as r:
                r.raise_for_status()
                data = await r.json()

        if cache is not None:
            await self.cache.set_static(cache, data)

        return data

    async def request_bbs(
        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 bbs 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.BBS_URL.get_url(region).join(yarl.URL(url))

        headers = parse_loose_headers(headers)
        headers.update(ds.get_ds_headers(data=data, params=params, region=region, lang=lang or self.lang))
        headers["Referer"] = str(routes.BBS_REFERER_URL.get_url(self.region))

        data = await self.request(url, method=method, params=params, data=data, headers=headers, **kwargs)
        return data

    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))

        headers = parse_loose_headers(headers)
        headers.update(ds.get_ds_headers(data=data, params=params, region=region, lang=lang or self.lang))

        data = await self.request(url, method=method, params=params, data=data, headers=headers, **kwargs)
        return data

    @managers.no_multi
    async def get_game_accounts(
        self, *, lang: typing.Optional[str] = None
    ) -> typing.Sequence[hoyolab_models.GenshinAccount]:
        """Get the game accounts of the currently logged-in user."""
        if self.hoyolab_id is None:
            warnings.warn("No hoyolab id set, caching may be unreliable.")

        data = await self.request_hoyolab(
            "binding/api/getUserGameRolesByCookie",
            lang=lang,
            cache=client_cache.cache_key("accounts", hoyolab_id=self.hoyolab_id),
        )
        return [hoyolab_models.GenshinAccount(**i) for i in data["list"]]

    @deprecation.deprecated("get_game_accounts")
    async def genshin_accounts(
        self, *, lang: typing.Optional[str] = None
    ) -> typing.Sequence[hoyolab_models.GenshinAccount]:
        """Get the genshin accounts of the currently logged-in user."""
        accounts = await self.get_game_accounts(lang=lang)
        return [account for account in accounts if account.game == types.Game.GENSHIN]

    async def _update_cached_uids(self) -> None:
        """Update cached fallback uids."""
        mixed_accounts = await self.get_game_accounts()

        game_accounts: dict[types.Game, list[hoyolab_models.GenshinAccount]] = {}
        for account in mixed_accounts:
            if not isinstance(account.game, types.Game):  # pyright: ignore[reportUnnecessaryIsInstance]
                continue

            game_accounts.setdefault(account.game, []).append(account)

        self.uids = {game: max(accounts, key=lambda a: a.level).uid for game, accounts in game_accounts.items()}

        if len(self.uids) == 1 and self.default_game is None:
            (self.default_game,) = self.uids.keys()

    @concurrency.prevent_concurrency
    async def _get_uid(self, game: types.Game) -> int:
        """Get a cached fallback uid."""
        # TODO: use lock
        if uid := self.uids.get(game):
            return uid

        if self.cookie_manager.multi:
            raise RuntimeError("UID must be provided when using multi-cookie managers.")

        await self._update_cached_uids()

        if uid := self.uids.get(game):
            return uid

        raise errors.AccountNotFound(msg="No UID provided and account has no game account bound to it.")

    async def _update_cached_accounts(self) -> None:
        """Update cached fallback accounts."""
        mixed_accounts = await self.get_game_accounts()

        game_accounts: dict[types.Game, list[hoyolab_models.GenshinAccount]] = {}
        for account in mixed_accounts:
            if not isinstance(account.game, types.Game):  # pyright: ignore[reportUnnecessaryIsInstance]
                continue

            game_accounts.setdefault(account.game, []).append(account)

        self._accounts = {}
        for game, accounts in game_accounts.items():
            self._accounts[game] = next(
                (acc for acc in accounts if acc.uid == self.uids.get(game)), max(accounts, key=lambda a: a.level)
            )

    @concurrency.prevent_concurrency
    async def _get_account(self, game: types.Game) -> hoyolab_models.GenshinAccount:
        """Get a cached fallback account."""
        if (account := self._accounts.get(game)) and (uid := self.uids.get(game)) and account.uid == uid:
            return account

        await self._update_cached_accounts()

        if account := self._accounts.get(game):
            if (uid := self.uids.get(game)) and account.uid != uid:
                raise errors.AccountNotFound(msg="There is no game account with such UID.")

            return account

        raise errors.AccountNotFound(msg="Account has no game account bound to it.")

    def _get_hoyolab_id(self) -> int:
        """Get a cached fallback hoyolab ID."""
        if self.hoyolab_id is not None:
            return self.hoyolab_id

        if self.cookie_manager.multi:
            raise RuntimeError("Hoyolab ID must be provided when using multi-cookie managers.")

        raise RuntimeError("No default hoyolab ID provided.")

    def get_account_timezone(
        self, *, game: typing.Optional[types.Game] = None, uid: typing.Optional[int] = None
    ) -> typing.Optional[int]:
        """Get the UTC timezone of the default game account."""
        game = game or self.game
        if game is None:
            return None

        uid = uid or self.uids.get(game)
        if uid is None:
            return None

        server = recognize_server(uid, game=game)  # type: ignore[arg-type]

        for tz, servers in constants.SERVER_TIMEZONES.items():
            if server in servers:
                return tz

        return None

    def _add_timezone_to_data(
        self,
        data: typing.Mapping[str, typing.Any],
        keys: typing.Sequence[str],
        *,
        game: typing.Optional[types.Game] = None,
        uid: typing.Optional[int] = None,
    ) -> typing.Mapping[str, typing.Any]:
        """Recursively add timezone info to a data dict based on the default game account.

        Datetime dicts found under the given keys get a "tzinfo" entry added, unix timestamp
        values get wrapped into {"timestamp": ..., "tzinfo": ...} dicts.
        """
        tz = self.get_account_timezone(game=game, uid=uid)
        if tz is None:
            return data

        def add_tz(value: typing.Any) -> None:
            if isinstance(value, dict):
                value = typing.cast("typing.Dict[str, typing.Any]", value)
                for key, item in value.items():
                    if key in keys and isinstance(item, dict) and item:
                        item["tzinfo"] = tz
                    elif key in keys and (isinstance(item, int) or (isinstance(item, str) and item.isdigit())):
                        value[key] = {"timestamp": item, "tzinfo": tz}
                    else:
                        add_tz(item)
            elif isinstance(value, list):
                for item in typing.cast("typing.List[typing.Any]", value):
                    add_tz(item)

        data = dict(data)
        add_tz(data)
        return data

prop debug : bool

Expand source code

@property
def debug(self) -> bool:
    """Whether the debug logs are being shown in stdout"""
    return logging.getLogger("genshin").level == logging.DEBUG

Whether the debug logs are being shown in stdout

prop default_game : Game | None

Expand source code

@property
def default_game(self) -> typing.Optional[types.Game]:
    """The default game."""
    return self._default_game

The default game.

prop device_fp : str | None

Expand source code

@property
def device_fp(self) -> typing.Optional[str]:
    """The device fingerprint used in headers."""
    return self.custom_headers.get("x-rpc-device_fp")

The device fingerprint used in headers.

prop device_id : str | None

Expand source code

@property
def device_id(self) -> typing.Optional[str]:
    """The device id used in headers."""
    return self.custom_headers.get("x-rpc-device_id")

The device id used in headers.

prop game : Game | None

Expand source code

@property
def default_game(self) -> typing.Optional[types.Game]:
    """The default game."""
    return self._default_game

The default game.

prop hoyolab_id : int | None

Expand source code

@property
def hoyolab_id(self) -> typing.Optional[int]:
    """The logged-in user's hoyolab uid.

    Returns None if not found or not applicable.
    """
    return self._hoyolab_id or self.cookie_manager.user_id

The logged-in user's hoyolab uid.

Returns None if not found or not applicable.

prop lang : str

Expand source code

@property
def lang(self) -> str:
    """The default language, defaults to "en-us" """
    return self._lang

The default language, defaults to "en-us"

prop on_cookie_update : Callable[[Mapping[str, str]], Awaitable[None] | None] | None

Expand source code

@property
def on_cookie_update(self) -> typing.Optional[managers.CookieUpdateHook]:
    """Callback invoked with a copy of the cookies whenever the client updates them itself.

    This happens for example when a new cookie_token is minted from an stoken or when
    a response sets new cookies. May be a sync or async callable. Use it to persist
    refreshed cookies so the next client created from storage does not need to refresh again.
    """
    return self.cookie_manager.on_cookie_update

Callback invoked with a copy of the cookies whenever the client updates them itself.

This happens for example when a new cookie_token is minted from an stoken or when a response sets new cookies. May be a sync or async callable. Use it to persist refreshed cookies so the next client created from storage does not need to refresh again.

prop proxy : str | None

Expand source code

@property
def proxy(self) -> typing.Optional[str]:
    """Proxy for http requests."""
    if self.cookie_manager.proxy is None:
        return None

    return str(self.cookie_manager.proxy)

Proxy for http requests.

prop region : Region

Expand source code

@property
def region(self) -> types.Region:
    """The default region."""
    return self._region

The default region.

prop uid : int | None

Expand source code

@property
def uid(self) -> typing.Optional[int]:
    """UID of the default game."""
    if self.default_game is None:
        if len(self.uids) != 1:
            return None

        (self.default_game,) = self.uids.keys()

    return self.uids.get(self.default_game)

UID of the default game.

var uids : dict[Game, int]

Expand source code

class BaseClient(abc.ABC):
    """Base ABC Client."""

    __slots__ = (
        "cookie_manager",
        "cache",
        "_lang",
        "_region",
        "_default_game",
        "uids",
        "authkeys",
        "_hoyolab_id",
        "_accounts",
        "custom_headers",
    )

    USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36"  # noqa: E501

    logger: logging.Logger = logging.getLogger(__name__)

    cookie_manager: managers.BaseCookieManager
    cache: client_cache.BaseCache
    _lang: str
    _region: types.Region
    _default_game: typing.Optional[types.Game]

    uids: dict[types.Game, int]
    authkeys: dict[types.Game, str]
    _hoyolab_id: typing.Optional[int]
    _accounts: dict[types.Game, hoyolab_models.GenshinAccount]
    custom_headers: multidict.CIMultiDict[str]

    def __init__(
        self,
        cookies: typing.Optional[managers.AnyCookieOrHeader] = None,
        *,
        authkey: typing.Optional[str] = None,
        lang: types.Lang = "en-us",
        region: types.Region = types.Region.OVERSEAS,
        proxy: typing.Optional[str] = None,
        game: typing.Optional[types.Game] = None,
        uid: typing.Optional[int] = None,
        hoyolab_id: typing.Optional[int] = None,
        device_id: typing.Optional[str] = None,
        device_fp: typing.Optional[str] = None,
        headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
        cache: typing.Optional[client_cache.BaseCache] = None,
        debug: bool = False,
        on_cookie_update: typing.Optional[managers.CookieUpdateHook] = None,
    ) -> None:
        self.cookie_manager = managers.BaseCookieManager.from_cookies(cookies)
        self.on_cookie_update = on_cookie_update
        self.cache = cache or client_cache.StaticCache()

        self.uids = {}
        self.authkeys = {}
        self._accounts = {}

        self.default_game = game
        self.lang = lang
        self.region = region
        self.authkey = authkey
        self.debug = debug
        self.proxy = proxy
        self.uid = uid
        self.hoyolab_id = hoyolab_id

        self.custom_headers = parse_loose_headers(headers)
        self.custom_headers.update({"x-rpc-device_id": device_id} if device_id else {})
        self.custom_headers.update({"x-rpc-device_fp": device_fp} if device_fp else {})

    def __repr__(self) -> str:
        kwargs = dict(
            lang=self.lang,
            region=self.region.value,
            default_game=self.default_game and self.default_game.value,
            hoyolab_id=self.hoyolab_id,
            uid=self.default_game and self.uid,
            authkey=self.authkey and self.authkey[:12] + "...",
            proxy=self.proxy,
            debug=self.debug,
        )
        return f"<{type(self).__name__} {', '.join(f'{k}={v!r}' for k, v in kwargs.items() if v)}>"

    @property
    def device_id(self) -> typing.Optional[str]:
        """The device id used in headers."""
        return self.custom_headers.get("x-rpc-device_id")

    @device_id.setter
    def device_id(self, device_id: str) -> None:
        self.custom_headers["x-rpc-device_id"] = device_id

    @property
    def device_fp(self) -> typing.Optional[str]:
        """The device fingerprint used in headers."""
        return self.custom_headers.get("x-rpc-device_fp")

    @device_fp.setter
    def device_fp(self, device_fp: str) -> None:
        self.custom_headers["x-rpc-device_fp"] = device_fp

    @property
    def hoyolab_id(self) -> typing.Optional[int]:
        """The logged-in user's hoyolab uid.

        Returns None if not found or not applicable.
        """
        return self._hoyolab_id or self.cookie_manager.user_id

    @hoyolab_id.setter
    def hoyolab_id(self, hoyolab_id: typing.Optional[int]) -> None:
        if hoyolab_id is None:
            self._hoyolab_id = None
            return

        if self.cookie_manager.multi:
            raise RuntimeError("Cannot specify a hoyolab uid when using multiple cookies.")

        if self.cookie_manager.user_id and hoyolab_id and self.cookie_manager.user_id != hoyolab_id:
            raise ValueError("The provided hoyolab uid does not match the cookie id.")

        self._hoyolab_id = hoyolab_id

    @property
    def lang(self) -> str:
        """The default language, defaults to "en-us" """
        return self._lang

    @lang.setter
    def lang(self, lang: str) -> None:
        if lang not in constants.LANGS:
            raise ValueError(f"{lang} is not a valid language, must be one of: " + ", ".join(constants.LANGS))

        self._lang = lang

    @property
    def region(self) -> types.Region:
        """The default region."""
        return self._region

    @region.setter
    def region(self, region: str) -> None:
        self._region = types.Region(region)

        if region == types.Region.CHINESE:
            self.lang = "zh-cn"

    @property
    def default_game(self) -> typing.Optional[types.Game]:
        """The default game."""
        return self._default_game

    @default_game.setter
    def default_game(self, game: typing.Optional[str]) -> None:
        self._default_game = types.Game(game) if game else None

    game = default_game

    @property
    def uid(self) -> typing.Optional[int]:
        """UID of the default game."""
        if self.default_game is None:
            if len(self.uids) != 1:
                return None

            (self.default_game,) = self.uids.keys()

        return self.uids.get(self.default_game)

    @uid.setter
    def uid(self, uid: typing.Optional[int]) -> None:
        if uid is None:
            self.uids.clear()
            return

        self._default_game = self._default_game or utility.recognize_game(uid, region=self.region)
        if self.default_game is None:
            raise RuntimeError("No default game set. Cannot set uid.")

        self.uids[self.default_game] = uid

    @property
    def authkey(self) -> typing.Optional[str]:
        """The default genshin authkey used for paginators."""
        if self.default_game is None:
            if self.authkeys:
                warnings.warn("Tried to get an authkey without a default game set.")

            return None

        return self.authkeys.get(self.default_game)

    @authkey.setter
    def authkey(self, authkey: typing.Optional[str]) -> None:
        if authkey is None:
            self.authkeys.clear()
            return

        authkey = urllib.parse.unquote(authkey)

        try:
            base64.b64decode(authkey, validate=True)
        except Exception as e:
            raise ValueError("authkey is not a valid base64 encoded string") from e

        if not self.default_game:
            raise RuntimeError("No default game set. Cannot set authkey with property.")

        self.authkeys[self.default_game] = authkey

    @property
    def debug(self) -> bool:
        """Whether the debug logs are being shown in stdout"""
        return logging.getLogger("genshin").level == logging.DEBUG

    @debug.setter
    def debug(self, debug: bool) -> None:
        logging.basicConfig()
        level = logging.DEBUG if debug else logging.NOTSET
        logging.getLogger("genshin").setLevel(level)

    @property
    def on_cookie_update(self) -> typing.Optional[managers.CookieUpdateHook]:
        """Callback invoked with a copy of the cookies whenever the client updates them itself.

        This happens for example when a new cookie_token is minted from an stoken or when
        a response sets new cookies. May be a sync or async callable. Use it to persist
        refreshed cookies so the next client created from storage does not need to refresh again.
        """
        return self.cookie_manager.on_cookie_update

    @on_cookie_update.setter
    def on_cookie_update(self, hook: typing.Optional[managers.CookieUpdateHook]) -> None:
        self.cookie_manager.on_cookie_update = hook

    def set_cookies(self, cookies: typing.Optional[managers.AnyCookieOrHeader] = None, **kwargs: typing.Any) -> None:
        """Parse and set cookies."""
        if not bool(cookies) ^ bool(kwargs):
            raise TypeError("Cannot use both positional and keyword arguments at once")

        hook = self.on_cookie_update
        self.cookie_manager = managers.BaseCookieManager.from_cookies(cookies or kwargs)
        self.on_cookie_update = hook

    def set_browser_cookies(self, browser: typing.Optional[str] = None) -> None:
        """Extract cookies from your browser and set them as client cookies.

        Available browsers: chrome, chromium, opera, edge, firefox.
        """
        hook = self.on_cookie_update
        self.cookie_manager = managers.BaseCookieManager.from_browser_cookies(browser)
        self.on_cookie_update = hook

    def set_authkey(self, authkey: typing.Optional[str] = None, *, game: typing.Optional[types.Game] = None) -> None:
        """Set an authkey for wish & transaction logs.

        Accepts an authkey, a url containing an authkey or a path towards a logfile.
        """
        if authkey is None or os.path.isfile(authkey):
            authkey = utility.get_authkey(authkey)
        else:
            authkey = utility.extract_authkey(authkey) or authkey

        game = game or self.default_game
        if game is None:
            raise RuntimeError("No default game set.")

        self.authkeys[game] = authkey

    def set_cache(
        self, maxsize: int = 1024, *, ttl: int = client_cache.HOUR, static_ttl: int = client_cache.DAY
    ) -> None:
        """Create and set a new cache."""
        self.cache = client_cache.Cache(maxsize, ttl=ttl, static_ttl=static_ttl)

    def set_redis_cache(
        self, url: str, *, ttl: int = client_cache.HOUR, static_ttl: int = client_cache.DAY, **redis_kwargs: typing.Any
    ) -> None:
        """Create and set a new redis cache."""
        import aioredis

        redis = aioredis.Redis.from_url(url, **redis_kwargs)  # pyright: ignore[reportUnknownMemberType]
        self.cache = client_cache.RedisCache(redis, ttl=ttl, static_ttl=static_ttl)

    @property
    def proxy(self) -> typing.Optional[str]:
        """Proxy for http requests."""
        if self.cookie_manager.proxy is None:
            return None

        return str(self.cookie_manager.proxy)

    @proxy.setter
    def proxy(self, proxy: typing.Optional[aiohttp.typedefs.StrOrURL]) -> None:
        self.cookie_manager.proxy = yarl.URL(proxy) if proxy else None

    async def _request_hook(
        self,
        method: str,
        url: aiohttp.typedefs.StrOrURL,
        *,
        params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        data: typing.Any = None,
        **kwargs: typing.Any,
    ) -> None:
        """Perform an action before a request.

        Debug logging by default.
        """
        url = yarl.URL(url)
        if params:
            params = {k: v for k, v in params.items() if k != "authkey"}
            url = url.update_query(params)

        if data:
            self.logger.debug("%s %s\n%s", method, url, json.dumps(data, separators=(",", ":")))
        else:
            self.logger.debug("%s %s", method, url)

    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 = parse_loose_headers(headers)
        headers["User-Agent"] = self.USER_AGENT
        headers.update(self.custom_headers)

        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

    async def request_webstatic(
        self,
        url: aiohttp.typedefs.StrOrURL,
        *,
        headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
        cache: typing.Any = None,
        region: types.Region = types.Region.OVERSEAS,
        **kwargs: typing.Any,
    ) -> typing.Any:
        """Request a static json file."""
        if cache is not None:
            value = await self.cache.get_static(cache)
            if value is not None:
                return value

        url = routes.WEBSTATIC_URL.get_url(region).join(yarl.URL(url))

        headers = parse_loose_headers(headers)
        headers["User-Agent"] = self.USER_AGENT
        headers.update(self.custom_headers)

        await self._request_hook("GET", url, headers=headers, **kwargs)

        async with self.cookie_manager.create_session() as session:
            async with session.get(url, headers=headers, proxy=self.proxy, **kwargs) as r:
                r.raise_for_status()
                data = await r.json()

        if cache is not None:
            await self.cache.set_static(cache, data)

        return data

    async def request_bbs(
        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 bbs 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.BBS_URL.get_url(region).join(yarl.URL(url))

        headers = parse_loose_headers(headers)
        headers.update(ds.get_ds_headers(data=data, params=params, region=region, lang=lang or self.lang))
        headers["Referer"] = str(routes.BBS_REFERER_URL.get_url(self.region))

        data = await self.request(url, method=method, params=params, data=data, headers=headers, **kwargs)
        return data

    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))

        headers = parse_loose_headers(headers)
        headers.update(ds.get_ds_headers(data=data, params=params, region=region, lang=lang or self.lang))

        data = await self.request(url, method=method, params=params, data=data, headers=headers, **kwargs)
        return data

    @managers.no_multi
    async def get_game_accounts(
        self, *, lang: typing.Optional[str] = None
    ) -> typing.Sequence[hoyolab_models.GenshinAccount]:
        """Get the game accounts of the currently logged-in user."""
        if self.hoyolab_id is None:
            warnings.warn("No hoyolab id set, caching may be unreliable.")

        data = await self.request_hoyolab(
            "binding/api/getUserGameRolesByCookie",
            lang=lang,
            cache=client_cache.cache_key("accounts", hoyolab_id=self.hoyolab_id),
        )
        return [hoyolab_models.GenshinAccount(**i) for i in data["list"]]

    @deprecation.deprecated("get_game_accounts")
    async def genshin_accounts(
        self, *, lang: typing.Optional[str] = None
    ) -> typing.Sequence[hoyolab_models.GenshinAccount]:
        """Get the genshin accounts of the currently logged-in user."""
        accounts = await self.get_game_accounts(lang=lang)
        return [account for account in accounts if account.game == types.Game.GENSHIN]

    async def _update_cached_uids(self) -> None:
        """Update cached fallback uids."""
        mixed_accounts = await self.get_game_accounts()

        game_accounts: dict[types.Game, list[hoyolab_models.GenshinAccount]] = {}
        for account in mixed_accounts:
            if not isinstance(account.game, types.Game):  # pyright: ignore[reportUnnecessaryIsInstance]
                continue

            game_accounts.setdefault(account.game, []).append(account)

        self.uids = {game: max(accounts, key=lambda a: a.level).uid for game, accounts in game_accounts.items()}

        if len(self.uids) == 1 and self.default_game is None:
            (self.default_game,) = self.uids.keys()

    @concurrency.prevent_concurrency
    async def _get_uid(self, game: types.Game) -> int:
        """Get a cached fallback uid."""
        # TODO: use lock
        if uid := self.uids.get(game):
            return uid

        if self.cookie_manager.multi:
            raise RuntimeError("UID must be provided when using multi-cookie managers.")

        await self._update_cached_uids()

        if uid := self.uids.get(game):
            return uid

        raise errors.AccountNotFound(msg="No UID provided and account has no game account bound to it.")

    async def _update_cached_accounts(self) -> None:
        """Update cached fallback accounts."""
        mixed_accounts = await self.get_game_accounts()

        game_accounts: dict[types.Game, list[hoyolab_models.GenshinAccount]] = {}
        for account in mixed_accounts:
            if not isinstance(account.game, types.Game):  # pyright: ignore[reportUnnecessaryIsInstance]
                continue

            game_accounts.setdefault(account.game, []).append(account)

        self._accounts = {}
        for game, accounts in game_accounts.items():
            self._accounts[game] = next(
                (acc for acc in accounts if acc.uid == self.uids.get(game)), max(accounts, key=lambda a: a.level)
            )

    @concurrency.prevent_concurrency
    async def _get_account(self, game: types.Game) -> hoyolab_models.GenshinAccount:
        """Get a cached fallback account."""
        if (account := self._accounts.get(game)) and (uid := self.uids.get(game)) and account.uid == uid:
            return account

        await self._update_cached_accounts()

        if account := self._accounts.get(game):
            if (uid := self.uids.get(game)) and account.uid != uid:
                raise errors.AccountNotFound(msg="There is no game account with such UID.")

            return account

        raise errors.AccountNotFound(msg="Account has no game account bound to it.")

    def _get_hoyolab_id(self) -> int:
        """Get a cached fallback hoyolab ID."""
        if self.hoyolab_id is not None:
            return self.hoyolab_id

        if self.cookie_manager.multi:
            raise RuntimeError("Hoyolab ID must be provided when using multi-cookie managers.")

        raise RuntimeError("No default hoyolab ID provided.")

    def get_account_timezone(
        self, *, game: typing.Optional[types.Game] = None, uid: typing.Optional[int] = None
    ) -> typing.Optional[int]:
        """Get the UTC timezone of the default game account."""
        game = game or self.game
        if game is None:
            return None

        uid = uid or self.uids.get(game)
        if uid is None:
            return None

        server = recognize_server(uid, game=game)  # type: ignore[arg-type]

        for tz, servers in constants.SERVER_TIMEZONES.items():
            if server in servers:
                return tz

        return None

    def _add_timezone_to_data(
        self,
        data: typing.Mapping[str, typing.Any],
        keys: typing.Sequence[str],
        *,
        game: typing.Optional[types.Game] = None,
        uid: typing.Optional[int] = None,
    ) -> typing.Mapping[str, typing.Any]:
        """Recursively add timezone info to a data dict based on the default game account.

        Datetime dicts found under the given keys get a "tzinfo" entry added, unix timestamp
        values get wrapped into {"timestamp": ..., "tzinfo": ...} dicts.
        """
        tz = self.get_account_timezone(game=game, uid=uid)
        if tz is None:
            return data

        def add_tz(value: typing.Any) -> None:
            if isinstance(value, dict):
                value = typing.cast("typing.Dict[str, typing.Any]", value)
                for key, item in value.items():
                    if key in keys and isinstance(item, dict) and item:
                        item["tzinfo"] = tz
                    elif key in keys and (isinstance(item, int) or (isinstance(item, str) and item.isdigit())):
                        value[key] = {"timestamp": item, "tzinfo": tz}
                    else:
                        add_tz(item)
            elif isinstance(value, list):
                for item in typing.cast("typing.List[typing.Any]", value):
                    add_tz(item)

        data = dict(data)
        add_tz(data)
        return data

Methods

async def genshin_accounts(self, *, lang: str | None = None) ‑> Sequence[GenshinAccount]

Expand source code

@deprecation.deprecated("get_game_accounts")
async def genshin_accounts(
    self, *, lang: typing.Optional[str] = None
) -> typing.Sequence[hoyolab_models.GenshinAccount]:
    """Get the genshin accounts of the currently logged-in user."""
    accounts = await self.get_game_accounts(lang=lang)
    return [account for account in accounts if account.game == types.Game.GENSHIN]

Get the genshin accounts of the currently logged-in user.

Warning

This function is deprecated and will be removed in the following version. You can use get_game_accounts instead.

def get_account_timezone(self,
*,
game: Game | None = None,
uid: int | None = None) ‑> int | None

Expand source code

def get_account_timezone(
    self, *, game: typing.Optional[types.Game] = None, uid: typing.Optional[int] = None
) -> typing.Optional[int]:
    """Get the UTC timezone of the default game account."""
    game = game or self.game
    if game is None:
        return None

    uid = uid or self.uids.get(game)
    if uid is None:
        return None

    server = recognize_server(uid, game=game)  # type: ignore[arg-type]

    for tz, servers in constants.SERVER_TIMEZONES.items():
        if server in servers:
            return tz

    return None

Get the UTC timezone of the default game account.

async def get_game_accounts(self, *, lang: str | None = None) ‑> Sequence[GenshinAccount]

Expand source code

@managers.no_multi
async def get_game_accounts(
    self, *, lang: typing.Optional[str] = None
) -> typing.Sequence[hoyolab_models.GenshinAccount]:
    """Get the game accounts of the currently logged-in user."""
    if self.hoyolab_id is None:
        warnings.warn("No hoyolab id set, caching may be unreliable.")

    data = await self.request_hoyolab(
        "binding/api/getUserGameRolesByCookie",
        lang=lang,
        cache=client_cache.cache_key("accounts", hoyolab_id=self.hoyolab_id),
    )
    return [hoyolab_models.GenshinAccount(**i) for i in data["list"]]

Get the game accounts of the currently logged-in user.

async def request(self,
url: str | yarl.URL,
*,
method: str | None = None,
params: Mapping[str, Any] | None = None,
data: Any = None,
headers: Mapping[str, str] | Mapping[multidict._multidict.istr, str] | multidict._multidict.CIMultiDict | multidict._multidict.CIMultiDictProxy | Iterable[Tuple[str | multidict._multidict.istr, str]] | None = None,
cache: Any = None,
static_cache: Any = None,
**kwargs: Any) ‑> Mapping[str, Any]

Expand source code

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 = parse_loose_headers(headers)
    headers["User-Agent"] = self.USER_AGENT
    headers.update(self.custom_headers)

    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

Make a request and return a parsed json response.

async def request_bbs(self,
url: str | yarl.URL,
*,
lang: str | None = None,
region: Region | None = None,
method: str | None = None,
params: Mapping[str, Any] | None = None,
data: Any = None,
headers: Mapping[str, str] | Mapping[multidict._multidict.istr, str] | multidict._multidict.CIMultiDict | multidict._multidict.CIMultiDictProxy | Iterable[Tuple[str | multidict._multidict.istr, str]] | None = None,
**kwargs: Any) ‑> Mapping[str, Any]

Expand source code

async def request_bbs(
    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 bbs 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.BBS_URL.get_url(region).join(yarl.URL(url))

    headers = parse_loose_headers(headers)
    headers.update(ds.get_ds_headers(data=data, params=params, region=region, lang=lang or self.lang))
    headers["Referer"] = str(routes.BBS_REFERER_URL.get_url(self.region))

    data = await self.request(url, method=method, params=params, data=data, headers=headers, **kwargs)
    return data

Make a request any bbs endpoint.

async def request_hoyolab(self,
url: str | yarl.URL,
*,
lang: str | None = None,
region: Region | None = None,
method: str | None = None,
params: Mapping[str, Any] | None = None,
data: Any = None,
headers: Mapping[str, str] | Mapping[multidict._multidict.istr, str] | multidict._multidict.CIMultiDict | multidict._multidict.CIMultiDictProxy | Iterable[Tuple[str | multidict._multidict.istr, str]] | None = None,
**kwargs: Any) ‑> Mapping[str, Any]

Expand source code

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))

    headers = parse_loose_headers(headers)
    headers.update(ds.get_ds_headers(data=data, params=params, region=region, lang=lang or self.lang))

    data = await self.request(url, method=method, params=params, data=data, headers=headers, **kwargs)
    return data

Make a request any hoyolab endpoint.

async def request_webstatic(self,
url: str | yarl.URL,
*,
headers: Mapping[str, str] | Mapping[multidict._multidict.istr, str] | multidict._multidict.CIMultiDict | multidict._multidict.CIMultiDictProxy | Iterable[Tuple[str | multidict._multidict.istr, str]] | None = None,
cache: Any = None,
region: Region = Region.OVERSEAS,
**kwargs: Any) ‑> Any

Expand source code

async def request_webstatic(
    self,
    url: aiohttp.typedefs.StrOrURL,
    *,
    headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
    cache: typing.Any = None,
    region: types.Region = types.Region.OVERSEAS,
    **kwargs: typing.Any,
) -> typing.Any:
    """Request a static json file."""
    if cache is not None:
        value = await self.cache.get_static(cache)
        if value is not None:
            return value

    url = routes.WEBSTATIC_URL.get_url(region).join(yarl.URL(url))

    headers = parse_loose_headers(headers)
    headers["User-Agent"] = self.USER_AGENT
    headers.update(self.custom_headers)

    await self._request_hook("GET", url, headers=headers, **kwargs)

    async with self.cookie_manager.create_session() as session:
        async with session.get(url, headers=headers, proxy=self.proxy, **kwargs) as r:
            r.raise_for_status()
            data = await r.json()

    if cache is not None:
        await self.cache.set_static(cache, data)

    return data

Request a static json file.

def set_authkey(self,
authkey: str | None = None,
*,
game: Game | None = None) ‑> None

Expand source code

def set_authkey(self, authkey: typing.Optional[str] = None, *, game: typing.Optional[types.Game] = None) -> None:
    """Set an authkey for wish & transaction logs.

    Accepts an authkey, a url containing an authkey or a path towards a logfile.
    """
    if authkey is None or os.path.isfile(authkey):
        authkey = utility.get_authkey(authkey)
    else:
        authkey = utility.extract_authkey(authkey) or authkey

    game = game or self.default_game
    if game is None:
        raise RuntimeError("No default game set.")

    self.authkeys[game] = authkey

Set an authkey for wish & transaction logs.

Accepts an authkey, a url containing an authkey or a path towards a logfile.

def set_browser_cookies(self, browser: str | None = None) ‑> None

Expand source code

def set_browser_cookies(self, browser: typing.Optional[str] = None) -> None:
    """Extract cookies from your browser and set them as client cookies.

    Available browsers: chrome, chromium, opera, edge, firefox.
    """
    hook = self.on_cookie_update
    self.cookie_manager = managers.BaseCookieManager.from_browser_cookies(browser)
    self.on_cookie_update = hook

Extract cookies from your browser and set them as client cookies.

Available browsers: chrome, chromium, opera, edge, firefox.

def set_cache(self, maxsize: int = 1024, *, ttl: int = 3600, static_ttl: int = 86400) ‑> None

Expand source code

def set_cache(
    self, maxsize: int = 1024, *, ttl: int = client_cache.HOUR, static_ttl: int = client_cache.DAY
) -> None:
    """Create and set a new cache."""
    self.cache = client_cache.Cache(maxsize, ttl=ttl, static_ttl=static_ttl)

Create and set a new cache.

def set_cookies(self,
cookies: ForwardRef('http.cookies.BaseCookie[Any]') | Mapping[Any, Any] | str | Sequence[ForwardRef('http.cookies.BaseCookie[Any]') | Mapping[Any, Any] | str] | None = None,
**kwargs: Any)

Expand source code

def set_cookies(self, cookies: typing.Optional[managers.AnyCookieOrHeader] = None, **kwargs: typing.Any) -> None:
    """Parse and set cookies."""
    if not bool(cookies) ^ bool(kwargs):
        raise TypeError("Cannot use both positional and keyword arguments at once")

    hook = self.on_cookie_update
    self.cookie_manager = managers.BaseCookieManager.from_cookies(cookies or kwargs)
    self.on_cookie_update = hook

Parse and set cookies.

def set_redis_cache(self, url: str, *, ttl: int = 3600, static_ttl: int = 86400, **redis_kwargs: Any) ‑> None

Expand source code

def set_redis_cache(
    self, url: str, *, ttl: int = client_cache.HOUR, static_ttl: int = client_cache.DAY, **redis_kwargs: typing.Any
) -> None:
    """Create and set a new redis cache."""
    import aioredis

    redis = aioredis.Redis.from_url(url, **redis_kwargs)  # pyright: ignore[reportUnknownMemberType]
    self.cache = client_cache.RedisCache(redis, ttl=ttl, static_ttl=static_ttl)

Create and set a new redis cache.