Module genshin.client.components.daily

Daily reward component.

Classes

class DailyRewardClient (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: str = '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)

Expand source code

class DailyRewardClient(base.BaseClient):
    """Daily reward component."""

    @managers.no_multi
    async def request_daily_reward(
        self,
        endpoint: str,
        *,
        game: typing.Optional[types.Game] = None,
        method: str = "GET",
        lang: typing.Optional[str] = None,
        params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
        **kwargs: typing.Any,
    ) -> typing.Mapping[str, typing.Any]:
        """Make a request towards the daily reward endpoint."""
        params = dict(params or {})
        headers = base.parse_loose_headers(headers)

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

            game = self.default_game

        base_url = routes.REWARD_URL.get_url(self.region, game)
        url = (base_url / endpoint).update_query(**base_url.query)

        if game is types.Game.GENSHIN:
            headers["x-rpc-signgame"] = "hk4e"
        elif game is types.Game.STARRAIL:
            headers["x-rpc-signgame"] = "hkrpg"
        elif game is types.Game.ZZZ:
            headers["x-rpc-signgame"] = "zzz"

        if self.region == types.Region.OVERSEAS:
            params["lang"] = lang or self.lang
            headers["referer"] = "https://act.hoyolab.com/"

        elif self.region == types.Region.CHINESE:
            account = await self._get_account(game)

            params["uid"] = account.uid
            params["region"] = account.server

            # These headers are optional but left here because they might affect geetest trigger rate
            headers["x-rpc-app_version"] = "2.70.1"
            headers["x-rpc-client_type"] = "5"
            headers["x-rpc-device_id"] = str(uuid.uuid4())
            headers["x-rpc-sys_version"] = "12"
            headers["x-rpc-platform"] = "android"
            headers["x-rpc-channel"] = "miyousheluodi"
            headers["x-rpc-device_model"] = str(self.hoyolab_id) or ""

            headers["ds"] = ds_utility.generate_dynamic_secret(constants.DS_SALT["cn_signin"])

        if challenge:
            headers["x-rpc-challenge"] = challenge["challenge"]
            headers["x-rpc-seccode"] = challenge["seccode"]
            headers["x-rpc-validate"] = challenge["validate"]

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

    async def get_reward_info(
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> models.DailyRewardInfo:
        """Get the daily reward info for the current user."""
        data = await self.request_daily_reward("info", game=game, lang=lang)
        return models.DailyRewardInfo(data["is_sign"], data["total_sign_day"])

    async def get_monthly_rewards(
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> typing.Sequence[models.DailyReward]:
        """Get a list of all availible rewards for the current month."""
        data = await self.request_daily_reward(
            "home",
            game=game,
            static_cache=cache.cache_key(
                "rewards",
                month=datetime.datetime.now(constants.CN_TIMEZONE).month,
                region=self.region,
                game=typing.cast("types.Game", game or self.default_game),  # (resolved later)
                lang=lang or self.lang,
            ),
        )
        return [models.DailyReward(**i) for i in data["awards"]]

    async def _get_claimed_rewards_page(
        self,
        page: int,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> typing.Sequence[models.ClaimedDailyReward]:
        """Get a single page of claimed rewards for the current user."""
        data = await self.request_daily_reward("award", params=dict(current_page=page), game=game, lang=lang)
        return [models.ClaimedDailyReward(**i) for i in data["list"]]

    def claimed_rewards(
        self,
        *,
        limit: typing.Optional[int] = None,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> paginators.Paginator[models.ClaimedDailyReward]:
        """Get all claimed rewards for the current user."""
        return paginators.PagedPaginator(
            functools.partial(
                self._get_claimed_rewards_page,
                game=game,
                lang=lang,
            ),
            limit=limit,
            page_size=10,
        )

    @typing.overload
    async def claim_daily_reward(  # noqa: D102 missing docstring in overload?
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
        reward: typing.Literal[True] = ...,
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
    ) -> models.DailyReward: ...

    @typing.overload
    async def claim_daily_reward(  # noqa: D102 missing docstring in overload?
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
        reward: typing.Literal[False],
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
    ) -> None: ...

    async def claim_daily_reward(
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
        reward: bool = True,
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
    ) -> typing.Optional[models.DailyReward]:
        """Signs into hoyolab and claims the daily reward."""
        await self.request_daily_reward("sign", method="POST", game=game, lang=lang, challenge=challenge)

        if not reward:
            return None

        info, rewards = await asyncio.gather(
            self.get_reward_info(game=game, lang=lang),
            self.get_monthly_rewards(game=game, lang=lang),
        )
        return rewards[info.claimed_rewards - 1]

Daily reward component.

Ancestors

• BaseClient
• abc.ABC

Subclasses

• Client

Class variables

var logger : logging.Logger

Instance variables

var authkeys : dict[Game, str]

Expand source code

class DailyRewardClient(base.BaseClient):
    """Daily reward component."""

    @managers.no_multi
    async def request_daily_reward(
        self,
        endpoint: str,
        *,
        game: typing.Optional[types.Game] = None,
        method: str = "GET",
        lang: typing.Optional[str] = None,
        params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
        **kwargs: typing.Any,
    ) -> typing.Mapping[str, typing.Any]:
        """Make a request towards the daily reward endpoint."""
        params = dict(params or {})
        headers = base.parse_loose_headers(headers)

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

            game = self.default_game

        base_url = routes.REWARD_URL.get_url(self.region, game)
        url = (base_url / endpoint).update_query(**base_url.query)

        if game is types.Game.GENSHIN:
            headers["x-rpc-signgame"] = "hk4e"
        elif game is types.Game.STARRAIL:
            headers["x-rpc-signgame"] = "hkrpg"
        elif game is types.Game.ZZZ:
            headers["x-rpc-signgame"] = "zzz"

        if self.region == types.Region.OVERSEAS:
            params["lang"] = lang or self.lang
            headers["referer"] = "https://act.hoyolab.com/"

        elif self.region == types.Region.CHINESE:
            account = await self._get_account(game)

            params["uid"] = account.uid
            params["region"] = account.server

            # These headers are optional but left here because they might affect geetest trigger rate
            headers["x-rpc-app_version"] = "2.70.1"
            headers["x-rpc-client_type"] = "5"
            headers["x-rpc-device_id"] = str(uuid.uuid4())
            headers["x-rpc-sys_version"] = "12"
            headers["x-rpc-platform"] = "android"
            headers["x-rpc-channel"] = "miyousheluodi"
            headers["x-rpc-device_model"] = str(self.hoyolab_id) or ""

            headers["ds"] = ds_utility.generate_dynamic_secret(constants.DS_SALT["cn_signin"])

        if challenge:
            headers["x-rpc-challenge"] = challenge["challenge"]
            headers["x-rpc-seccode"] = challenge["seccode"]
            headers["x-rpc-validate"] = challenge["validate"]

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

    async def get_reward_info(
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> models.DailyRewardInfo:
        """Get the daily reward info for the current user."""
        data = await self.request_daily_reward("info", game=game, lang=lang)
        return models.DailyRewardInfo(data["is_sign"], data["total_sign_day"])

    async def get_monthly_rewards(
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> typing.Sequence[models.DailyReward]:
        """Get a list of all availible rewards for the current month."""
        data = await self.request_daily_reward(
            "home",
            game=game,
            static_cache=cache.cache_key(
                "rewards",
                month=datetime.datetime.now(constants.CN_TIMEZONE).month,
                region=self.region,
                game=typing.cast("types.Game", game or self.default_game),  # (resolved later)
                lang=lang or self.lang,
            ),
        )
        return [models.DailyReward(**i) for i in data["awards"]]

    async def _get_claimed_rewards_page(
        self,
        page: int,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> typing.Sequence[models.ClaimedDailyReward]:
        """Get a single page of claimed rewards for the current user."""
        data = await self.request_daily_reward("award", params=dict(current_page=page), game=game, lang=lang)
        return [models.ClaimedDailyReward(**i) for i in data["list"]]

    def claimed_rewards(
        self,
        *,
        limit: typing.Optional[int] = None,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> paginators.Paginator[models.ClaimedDailyReward]:
        """Get all claimed rewards for the current user."""
        return paginators.PagedPaginator(
            functools.partial(
                self._get_claimed_rewards_page,
                game=game,
                lang=lang,
            ),
            limit=limit,
            page_size=10,
        )

    @typing.overload
    async def claim_daily_reward(  # noqa: D102 missing docstring in overload?
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
        reward: typing.Literal[True] = ...,
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
    ) -> models.DailyReward: ...

    @typing.overload
    async def claim_daily_reward(  # noqa: D102 missing docstring in overload?
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
        reward: typing.Literal[False],
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
    ) -> None: ...

    async def claim_daily_reward(
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
        reward: bool = True,
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
    ) -> typing.Optional[models.DailyReward]:
        """Signs into hoyolab and claims the daily reward."""
        await self.request_daily_reward("sign", method="POST", game=game, lang=lang, challenge=challenge)

        if not reward:
            return None

        info, rewards = await asyncio.gather(
            self.get_reward_info(game=game, lang=lang),
            self.get_monthly_rewards(game=game, lang=lang),
        )
        return rewards[info.claimed_rewards - 1]

var cache : BaseCache

Expand source code

class DailyRewardClient(base.BaseClient):
    """Daily reward component."""

    @managers.no_multi
    async def request_daily_reward(
        self,
        endpoint: str,
        *,
        game: typing.Optional[types.Game] = None,
        method: str = "GET",
        lang: typing.Optional[str] = None,
        params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
        **kwargs: typing.Any,
    ) -> typing.Mapping[str, typing.Any]:
        """Make a request towards the daily reward endpoint."""
        params = dict(params or {})
        headers = base.parse_loose_headers(headers)

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

            game = self.default_game

        base_url = routes.REWARD_URL.get_url(self.region, game)
        url = (base_url / endpoint).update_query(**base_url.query)

        if game is types.Game.GENSHIN:
            headers["x-rpc-signgame"] = "hk4e"
        elif game is types.Game.STARRAIL:
            headers["x-rpc-signgame"] = "hkrpg"
        elif game is types.Game.ZZZ:
            headers["x-rpc-signgame"] = "zzz"

        if self.region == types.Region.OVERSEAS:
            params["lang"] = lang or self.lang
            headers["referer"] = "https://act.hoyolab.com/"

        elif self.region == types.Region.CHINESE:
            account = await self._get_account(game)

            params["uid"] = account.uid
            params["region"] = account.server

            # These headers are optional but left here because they might affect geetest trigger rate
            headers["x-rpc-app_version"] = "2.70.1"
            headers["x-rpc-client_type"] = "5"
            headers["x-rpc-device_id"] = str(uuid.uuid4())
            headers["x-rpc-sys_version"] = "12"
            headers["x-rpc-platform"] = "android"
            headers["x-rpc-channel"] = "miyousheluodi"
            headers["x-rpc-device_model"] = str(self.hoyolab_id) or ""

            headers["ds"] = ds_utility.generate_dynamic_secret(constants.DS_SALT["cn_signin"])

        if challenge:
            headers["x-rpc-challenge"] = challenge["challenge"]
            headers["x-rpc-seccode"] = challenge["seccode"]
            headers["x-rpc-validate"] = challenge["validate"]

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

    async def get_reward_info(
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> models.DailyRewardInfo:
        """Get the daily reward info for the current user."""
        data = await self.request_daily_reward("info", game=game, lang=lang)
        return models.DailyRewardInfo(data["is_sign"], data["total_sign_day"])

    async def get_monthly_rewards(
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> typing.Sequence[models.DailyReward]:
        """Get a list of all availible rewards for the current month."""
        data = await self.request_daily_reward(
            "home",
            game=game,
            static_cache=cache.cache_key(
                "rewards",
                month=datetime.datetime.now(constants.CN_TIMEZONE).month,
                region=self.region,
                game=typing.cast("types.Game", game or self.default_game),  # (resolved later)
                lang=lang or self.lang,
            ),
        )
        return [models.DailyReward(**i) for i in data["awards"]]

    async def _get_claimed_rewards_page(
        self,
        page: int,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> typing.Sequence[models.ClaimedDailyReward]:
        """Get a single page of claimed rewards for the current user."""
        data = await self.request_daily_reward("award", params=dict(current_page=page), game=game, lang=lang)
        return [models.ClaimedDailyReward(**i) for i in data["list"]]

    def claimed_rewards(
        self,
        *,
        limit: typing.Optional[int] = None,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> paginators.Paginator[models.ClaimedDailyReward]:
        """Get all claimed rewards for the current user."""
        return paginators.PagedPaginator(
            functools.partial(
                self._get_claimed_rewards_page,
                game=game,
                lang=lang,
            ),
            limit=limit,
            page_size=10,
        )

    @typing.overload
    async def claim_daily_reward(  # noqa: D102 missing docstring in overload?
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
        reward: typing.Literal[True] = ...,
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
    ) -> models.DailyReward: ...

    @typing.overload
    async def claim_daily_reward(  # noqa: D102 missing docstring in overload?
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
        reward: typing.Literal[False],
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
    ) -> None: ...

    async def claim_daily_reward(
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
        reward: bool = True,
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
    ) -> typing.Optional[models.DailyReward]:
        """Signs into hoyolab and claims the daily reward."""
        await self.request_daily_reward("sign", method="POST", game=game, lang=lang, challenge=challenge)

        if not reward:
            return None

        info, rewards = await asyncio.gather(
            self.get_reward_info(game=game, lang=lang),
            self.get_monthly_rewards(game=game, lang=lang),
        )
        return rewards[info.claimed_rewards - 1]

var cookie_manager : BaseCookieManager

Expand source code

class DailyRewardClient(base.BaseClient):
    """Daily reward component."""

    @managers.no_multi
    async def request_daily_reward(
        self,
        endpoint: str,
        *,
        game: typing.Optional[types.Game] = None,
        method: str = "GET",
        lang: typing.Optional[str] = None,
        params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
        **kwargs: typing.Any,
    ) -> typing.Mapping[str, typing.Any]:
        """Make a request towards the daily reward endpoint."""
        params = dict(params or {})
        headers = base.parse_loose_headers(headers)

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

            game = self.default_game

        base_url = routes.REWARD_URL.get_url(self.region, game)
        url = (base_url / endpoint).update_query(**base_url.query)

        if game is types.Game.GENSHIN:
            headers["x-rpc-signgame"] = "hk4e"
        elif game is types.Game.STARRAIL:
            headers["x-rpc-signgame"] = "hkrpg"
        elif game is types.Game.ZZZ:
            headers["x-rpc-signgame"] = "zzz"

        if self.region == types.Region.OVERSEAS:
            params["lang"] = lang or self.lang
            headers["referer"] = "https://act.hoyolab.com/"

        elif self.region == types.Region.CHINESE:
            account = await self._get_account(game)

            params["uid"] = account.uid
            params["region"] = account.server

            # These headers are optional but left here because they might affect geetest trigger rate
            headers["x-rpc-app_version"] = "2.70.1"
            headers["x-rpc-client_type"] = "5"
            headers["x-rpc-device_id"] = str(uuid.uuid4())
            headers["x-rpc-sys_version"] = "12"
            headers["x-rpc-platform"] = "android"
            headers["x-rpc-channel"] = "miyousheluodi"
            headers["x-rpc-device_model"] = str(self.hoyolab_id) or ""

            headers["ds"] = ds_utility.generate_dynamic_secret(constants.DS_SALT["cn_signin"])

        if challenge:
            headers["x-rpc-challenge"] = challenge["challenge"]
            headers["x-rpc-seccode"] = challenge["seccode"]
            headers["x-rpc-validate"] = challenge["validate"]

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

    async def get_reward_info(
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> models.DailyRewardInfo:
        """Get the daily reward info for the current user."""
        data = await self.request_daily_reward("info", game=game, lang=lang)
        return models.DailyRewardInfo(data["is_sign"], data["total_sign_day"])

    async def get_monthly_rewards(
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> typing.Sequence[models.DailyReward]:
        """Get a list of all availible rewards for the current month."""
        data = await self.request_daily_reward(
            "home",
            game=game,
            static_cache=cache.cache_key(
                "rewards",
                month=datetime.datetime.now(constants.CN_TIMEZONE).month,
                region=self.region,
                game=typing.cast("types.Game", game or self.default_game),  # (resolved later)
                lang=lang or self.lang,
            ),
        )
        return [models.DailyReward(**i) for i in data["awards"]]

    async def _get_claimed_rewards_page(
        self,
        page: int,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> typing.Sequence[models.ClaimedDailyReward]:
        """Get a single page of claimed rewards for the current user."""
        data = await self.request_daily_reward("award", params=dict(current_page=page), game=game, lang=lang)
        return [models.ClaimedDailyReward(**i) for i in data["list"]]

    def claimed_rewards(
        self,
        *,
        limit: typing.Optional[int] = None,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> paginators.Paginator[models.ClaimedDailyReward]:
        """Get all claimed rewards for the current user."""
        return paginators.PagedPaginator(
            functools.partial(
                self._get_claimed_rewards_page,
                game=game,
                lang=lang,
            ),
            limit=limit,
            page_size=10,
        )

    @typing.overload
    async def claim_daily_reward(  # noqa: D102 missing docstring in overload?
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
        reward: typing.Literal[True] = ...,
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
    ) -> models.DailyReward: ...

    @typing.overload
    async def claim_daily_reward(  # noqa: D102 missing docstring in overload?
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
        reward: typing.Literal[False],
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
    ) -> None: ...

    async def claim_daily_reward(
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
        reward: bool = True,
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
    ) -> typing.Optional[models.DailyReward]:
        """Signs into hoyolab and claims the daily reward."""
        await self.request_daily_reward("sign", method="POST", game=game, lang=lang, challenge=challenge)

        if not reward:
            return None

        info, rewards = await asyncio.gather(
            self.get_reward_info(game=game, lang=lang),
            self.get_monthly_rewards(game=game, lang=lang),
        )
        return rewards[info.claimed_rewards - 1]

var custom_headers : multidict._multidict.CIMultiDict[str]

Expand source code

class DailyRewardClient(base.BaseClient):
    """Daily reward component."""

    @managers.no_multi
    async def request_daily_reward(
        self,
        endpoint: str,
        *,
        game: typing.Optional[types.Game] = None,
        method: str = "GET",
        lang: typing.Optional[str] = None,
        params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
        **kwargs: typing.Any,
    ) -> typing.Mapping[str, typing.Any]:
        """Make a request towards the daily reward endpoint."""
        params = dict(params or {})
        headers = base.parse_loose_headers(headers)

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

            game = self.default_game

        base_url = routes.REWARD_URL.get_url(self.region, game)
        url = (base_url / endpoint).update_query(**base_url.query)

        if game is types.Game.GENSHIN:
            headers["x-rpc-signgame"] = "hk4e"
        elif game is types.Game.STARRAIL:
            headers["x-rpc-signgame"] = "hkrpg"
        elif game is types.Game.ZZZ:
            headers["x-rpc-signgame"] = "zzz"

        if self.region == types.Region.OVERSEAS:
            params["lang"] = lang or self.lang
            headers["referer"] = "https://act.hoyolab.com/"

        elif self.region == types.Region.CHINESE:
            account = await self._get_account(game)

            params["uid"] = account.uid
            params["region"] = account.server

            # These headers are optional but left here because they might affect geetest trigger rate
            headers["x-rpc-app_version"] = "2.70.1"
            headers["x-rpc-client_type"] = "5"
            headers["x-rpc-device_id"] = str(uuid.uuid4())
            headers["x-rpc-sys_version"] = "12"
            headers["x-rpc-platform"] = "android"
            headers["x-rpc-channel"] = "miyousheluodi"
            headers["x-rpc-device_model"] = str(self.hoyolab_id) or ""

            headers["ds"] = ds_utility.generate_dynamic_secret(constants.DS_SALT["cn_signin"])

        if challenge:
            headers["x-rpc-challenge"] = challenge["challenge"]
            headers["x-rpc-seccode"] = challenge["seccode"]
            headers["x-rpc-validate"] = challenge["validate"]

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

    async def get_reward_info(
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> models.DailyRewardInfo:
        """Get the daily reward info for the current user."""
        data = await self.request_daily_reward("info", game=game, lang=lang)
        return models.DailyRewardInfo(data["is_sign"], data["total_sign_day"])

    async def get_monthly_rewards(
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> typing.Sequence[models.DailyReward]:
        """Get a list of all availible rewards for the current month."""
        data = await self.request_daily_reward(
            "home",
            game=game,
            static_cache=cache.cache_key(
                "rewards",
                month=datetime.datetime.now(constants.CN_TIMEZONE).month,
                region=self.region,
                game=typing.cast("types.Game", game or self.default_game),  # (resolved later)
                lang=lang or self.lang,
            ),
        )
        return [models.DailyReward(**i) for i in data["awards"]]

    async def _get_claimed_rewards_page(
        self,
        page: int,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> typing.Sequence[models.ClaimedDailyReward]:
        """Get a single page of claimed rewards for the current user."""
        data = await self.request_daily_reward("award", params=dict(current_page=page), game=game, lang=lang)
        return [models.ClaimedDailyReward(**i) for i in data["list"]]

    def claimed_rewards(
        self,
        *,
        limit: typing.Optional[int] = None,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> paginators.Paginator[models.ClaimedDailyReward]:
        """Get all claimed rewards for the current user."""
        return paginators.PagedPaginator(
            functools.partial(
                self._get_claimed_rewards_page,
                game=game,
                lang=lang,
            ),
            limit=limit,
            page_size=10,
        )

    @typing.overload
    async def claim_daily_reward(  # noqa: D102 missing docstring in overload?
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
        reward: typing.Literal[True] = ...,
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
    ) -> models.DailyReward: ...

    @typing.overload
    async def claim_daily_reward(  # noqa: D102 missing docstring in overload?
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
        reward: typing.Literal[False],
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
    ) -> None: ...

    async def claim_daily_reward(
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
        reward: bool = True,
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
    ) -> typing.Optional[models.DailyReward]:
        """Signs into hoyolab and claims the daily reward."""
        await self.request_daily_reward("sign", method="POST", game=game, lang=lang, challenge=challenge)

        if not reward:
            return None

        info, rewards = await asyncio.gather(
            self.get_reward_info(game=game, lang=lang),
            self.get_monthly_rewards(game=game, lang=lang),
        )
        return rewards[info.claimed_rewards - 1]

var uids : dict[Game, int]

Expand source code

class DailyRewardClient(base.BaseClient):
    """Daily reward component."""

    @managers.no_multi
    async def request_daily_reward(
        self,
        endpoint: str,
        *,
        game: typing.Optional[types.Game] = None,
        method: str = "GET",
        lang: typing.Optional[str] = None,
        params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
        **kwargs: typing.Any,
    ) -> typing.Mapping[str, typing.Any]:
        """Make a request towards the daily reward endpoint."""
        params = dict(params or {})
        headers = base.parse_loose_headers(headers)

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

            game = self.default_game

        base_url = routes.REWARD_URL.get_url(self.region, game)
        url = (base_url / endpoint).update_query(**base_url.query)

        if game is types.Game.GENSHIN:
            headers["x-rpc-signgame"] = "hk4e"
        elif game is types.Game.STARRAIL:
            headers["x-rpc-signgame"] = "hkrpg"
        elif game is types.Game.ZZZ:
            headers["x-rpc-signgame"] = "zzz"

        if self.region == types.Region.OVERSEAS:
            params["lang"] = lang or self.lang
            headers["referer"] = "https://act.hoyolab.com/"

        elif self.region == types.Region.CHINESE:
            account = await self._get_account(game)

            params["uid"] = account.uid
            params["region"] = account.server

            # These headers are optional but left here because they might affect geetest trigger rate
            headers["x-rpc-app_version"] = "2.70.1"
            headers["x-rpc-client_type"] = "5"
            headers["x-rpc-device_id"] = str(uuid.uuid4())
            headers["x-rpc-sys_version"] = "12"
            headers["x-rpc-platform"] = "android"
            headers["x-rpc-channel"] = "miyousheluodi"
            headers["x-rpc-device_model"] = str(self.hoyolab_id) or ""

            headers["ds"] = ds_utility.generate_dynamic_secret(constants.DS_SALT["cn_signin"])

        if challenge:
            headers["x-rpc-challenge"] = challenge["challenge"]
            headers["x-rpc-seccode"] = challenge["seccode"]
            headers["x-rpc-validate"] = challenge["validate"]

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

    async def get_reward_info(
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> models.DailyRewardInfo:
        """Get the daily reward info for the current user."""
        data = await self.request_daily_reward("info", game=game, lang=lang)
        return models.DailyRewardInfo(data["is_sign"], data["total_sign_day"])

    async def get_monthly_rewards(
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> typing.Sequence[models.DailyReward]:
        """Get a list of all availible rewards for the current month."""
        data = await self.request_daily_reward(
            "home",
            game=game,
            static_cache=cache.cache_key(
                "rewards",
                month=datetime.datetime.now(constants.CN_TIMEZONE).month,
                region=self.region,
                game=typing.cast("types.Game", game or self.default_game),  # (resolved later)
                lang=lang or self.lang,
            ),
        )
        return [models.DailyReward(**i) for i in data["awards"]]

    async def _get_claimed_rewards_page(
        self,
        page: int,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> typing.Sequence[models.ClaimedDailyReward]:
        """Get a single page of claimed rewards for the current user."""
        data = await self.request_daily_reward("award", params=dict(current_page=page), game=game, lang=lang)
        return [models.ClaimedDailyReward(**i) for i in data["list"]]

    def claimed_rewards(
        self,
        *,
        limit: typing.Optional[int] = None,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
    ) -> paginators.Paginator[models.ClaimedDailyReward]:
        """Get all claimed rewards for the current user."""
        return paginators.PagedPaginator(
            functools.partial(
                self._get_claimed_rewards_page,
                game=game,
                lang=lang,
            ),
            limit=limit,
            page_size=10,
        )

    @typing.overload
    async def claim_daily_reward(  # noqa: D102 missing docstring in overload?
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
        reward: typing.Literal[True] = ...,
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
    ) -> models.DailyReward: ...

    @typing.overload
    async def claim_daily_reward(  # noqa: D102 missing docstring in overload?
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
        reward: typing.Literal[False],
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
    ) -> None: ...

    async def claim_daily_reward(
        self,
        *,
        game: typing.Optional[types.Game] = None,
        lang: typing.Optional[str] = None,
        reward: bool = True,
        challenge: typing.Optional[typing.Mapping[str, str]] = None,
    ) -> typing.Optional[models.DailyReward]:
        """Signs into hoyolab and claims the daily reward."""
        await self.request_daily_reward("sign", method="POST", game=game, lang=lang, challenge=challenge)

        if not reward:
            return None

        info, rewards = await asyncio.gather(
            self.get_reward_info(game=game, lang=lang),
            self.get_monthly_rewards(game=game, lang=lang),
        )
        return rewards[info.claimed_rewards - 1]

Methods

async def claim_daily_reward(self,
*,
game: Game | None = None,
lang: str | None = None,
reward: bool = True,
challenge: Mapping[str, str] | None = None) ‑> DailyReward | None

Expand source code

async def claim_daily_reward(
    self,
    *,
    game: typing.Optional[types.Game] = None,
    lang: typing.Optional[str] = None,
    reward: bool = True,
    challenge: typing.Optional[typing.Mapping[str, str]] = None,
) -> typing.Optional[models.DailyReward]:
    """Signs into hoyolab and claims the daily reward."""
    await self.request_daily_reward("sign", method="POST", game=game, lang=lang, challenge=challenge)

    if not reward:
        return None

    info, rewards = await asyncio.gather(
        self.get_reward_info(game=game, lang=lang),
        self.get_monthly_rewards(game=game, lang=lang),
    )
    return rewards[info.claimed_rewards - 1]

Signs into hoyolab and claims the daily reward.

def claimed_rewards(self,
*,
limit: int | None = None,
game: Game | None = None,
lang: str | None = None) ‑> Paginator[ClaimedDailyReward]

Expand source code

def claimed_rewards(
    self,
    *,
    limit: typing.Optional[int] = None,
    game: typing.Optional[types.Game] = None,
    lang: typing.Optional[str] = None,
) -> paginators.Paginator[models.ClaimedDailyReward]:
    """Get all claimed rewards for the current user."""
    return paginators.PagedPaginator(
        functools.partial(
            self._get_claimed_rewards_page,
            game=game,
            lang=lang,
        ),
        limit=limit,
        page_size=10,
    )

Get all claimed rewards for the current user.

async def get_monthly_rewards(self,
*,
game: Game | None = None,
lang: str | None = None) ‑> Sequence[DailyReward]

Expand source code

async def get_monthly_rewards(
    self,
    *,
    game: typing.Optional[types.Game] = None,
    lang: typing.Optional[str] = None,
) -> typing.Sequence[models.DailyReward]:
    """Get a list of all availible rewards for the current month."""
    data = await self.request_daily_reward(
        "home",
        game=game,
        static_cache=cache.cache_key(
            "rewards",
            month=datetime.datetime.now(constants.CN_TIMEZONE).month,
            region=self.region,
            game=typing.cast("types.Game", game or self.default_game),  # (resolved later)
            lang=lang or self.lang,
        ),
    )
    return [models.DailyReward(**i) for i in data["awards"]]

Get a list of all availible rewards for the current month.

async def get_reward_info(self,
*,
game: Game | None = None,
lang: str | None = None) ‑> DailyRewardInfo

Expand source code

async def get_reward_info(
    self,
    *,
    game: typing.Optional[types.Game] = None,
    lang: typing.Optional[str] = None,
) -> models.DailyRewardInfo:
    """Get the daily reward info for the current user."""
    data = await self.request_daily_reward("info", game=game, lang=lang)
    return models.DailyRewardInfo(data["is_sign"], data["total_sign_day"])

Get the daily reward info for the current user.

async def request_daily_reward(self,
endpoint: str,
*,
game: Game | None = None,
method: str = 'GET',
lang: str | None = None,
params: Mapping[str, Any] | 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,
challenge: Mapping[str, str] | None = None,
**kwargs: Any) ‑> Mapping[str, Any]

Expand source code

@managers.no_multi
async def request_daily_reward(
    self,
    endpoint: str,
    *,
    game: typing.Optional[types.Game] = None,
    method: str = "GET",
    lang: typing.Optional[str] = None,
    params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
    headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
    challenge: typing.Optional[typing.Mapping[str, str]] = None,
    **kwargs: typing.Any,
) -> typing.Mapping[str, typing.Any]:
    """Make a request towards the daily reward endpoint."""
    params = dict(params or {})
    headers = base.parse_loose_headers(headers)

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

        game = self.default_game

    base_url = routes.REWARD_URL.get_url(self.region, game)
    url = (base_url / endpoint).update_query(**base_url.query)

    if game is types.Game.GENSHIN:
        headers["x-rpc-signgame"] = "hk4e"
    elif game is types.Game.STARRAIL:
        headers["x-rpc-signgame"] = "hkrpg"
    elif game is types.Game.ZZZ:
        headers["x-rpc-signgame"] = "zzz"

    if self.region == types.Region.OVERSEAS:
        params["lang"] = lang or self.lang
        headers["referer"] = "https://act.hoyolab.com/"

    elif self.region == types.Region.CHINESE:
        account = await self._get_account(game)

        params["uid"] = account.uid
        params["region"] = account.server

        # These headers are optional but left here because they might affect geetest trigger rate
        headers["x-rpc-app_version"] = "2.70.1"
        headers["x-rpc-client_type"] = "5"
        headers["x-rpc-device_id"] = str(uuid.uuid4())
        headers["x-rpc-sys_version"] = "12"
        headers["x-rpc-platform"] = "android"
        headers["x-rpc-channel"] = "miyousheluodi"
        headers["x-rpc-device_model"] = str(self.hoyolab_id) or ""

        headers["ds"] = ds_utility.generate_dynamic_secret(constants.DS_SALT["cn_signin"])

    if challenge:
        headers["x-rpc-challenge"] = challenge["challenge"]
        headers["x-rpc-seccode"] = challenge["seccode"]
        headers["x-rpc-validate"] = challenge["validate"]

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

Make a request towards the daily reward endpoint.

Inherited members

• BaseClient:
  • authkey
  • debug
  • default_game
  • device_fp
  • device_id
  • game
  • genshin_accounts
  • get_game_accounts
  • hoyolab_id
  • lang
  • proxy
  • region
  • request
  • request_bbs
  • request_hoyolab
  • request_webstatic
  • set_authkey
  • set_browser_cookies
  • set_cache
  • set_cookies
  • set_redis_cache
  • uid