✨ Add Daily Reward Client

simnet/client/base.py

@@ -9,7 +9,7 @@ from simnet.client.cookies import Cookies
 from simnet.client.headers import Headers
 from simnet.errors import TimedOut, NetworkError, BadRequest, raise_for_ret_code
 from simnet.utils.ds import generate_dynamic_secret
-from simnet.utils.enum_ import Region
+from simnet.utils.enum_ import Region, Game
 from simnet.utils.types import (
     RT,
     HeaderTypes,
@@ -44,8 +44,10 @@ class BaseClient(AsyncContextManager["BaseClient"]):
         player_id (Optional[int]): The player id used for the client.
         region (Region): The region used for the client.
         lang (str): The language used for the client.
+        game (Optional[Game]): The game used for the client.
     """
 
+    game: Optional[Game] = None
     _device_id = str(uuid.uuid3(uuid.NAMESPACE_URL, "SIMNet"))
 
     def __init__(

simnet/client/components/daily.py

@@ -0,0 +1,222 @@
+"""Daily reward component."""
+import asyncio
+from typing import Optional, Dict, Any, List
+
+from simnet.client.base import BaseClient
+from simnet.utils.ds import hex_digest
+from simnet.client.routes import REWARD_URL
+from simnet.models.lab.daily import DailyRewardInfo, DailyReward, ClaimedDailyReward
+from simnet.utils.enum_ import Game, Region
+
+__all__ = ("DailyRewardClient",)
+
+
+class DailyRewardClient(BaseClient):
+    """A client for interacting with the daily reward system."""
+
+    async def request_daily_reward(
+        self,
+        endpoint: str,
+        *,
+        method: str = "GET",
+        challenge: Optional[str] = None,
+        validate: Optional[str] = None,
+        game: Optional[Game] = None,
+        lang: Optional[str] = None,
+        params: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        """
+        Makes a request to the daily reward endpoint.
+
+        Args:
+            endpoint (str): The endpoint to request.
+            method (str): The HTTP method to use. Defaults to "GET".
+            challenge (str): A challenge string for validating the request. Defaults to None.
+            validate (str): A validation string for validating the request. Defaults to None.
+            game (Game): The game to request data for. Defaults to None.
+            lang (str): The language to use. Defaults to None.
+            params (Dict[str, Any]): Any parameters to include in the request.
+
+        Returns:
+            A dictionary containing the response data.
+        """
+        new_ds: bool = False
+        headers: Optional[Dict[str, str]] = None
+        if self.region == Region.CHINESE:
+            headers = {}
+            if challenge is not None and validate is not None:
+                headers["x-rpc-challenge"] = challenge
+                headers["x-rpc-validate"] = validate
+                headers["x-rpc-seccode"] = f"{validate}|jordan"
+            headers["Referer"] = (
+                "https://webstatic.mihoyo.com/bbs/event/signin-ys/index.html?"
+                "bbs_auth_required=true&act_id=e202009291139501&utm_source=bbs&utm_medium=mys&utm_campaign=icon"
+            )
+
+            headers["x-rpc-device_name"] = "Chrome 20 2023"
+            headers["x-rpc-channel"] = "chrome"
+            headers["x-rpc-device_model"] = "Chrome 2023"
+            headers["x-rpc-sys_version"] = "13"
+            headers["x-rpc-platform"] = "android"
+            device_id = self.device_id
+            hash_value = hex_digest(device_id)
+            headers["x-rpc-device_fp"] = hash_value[:13]
+            new_ds = endpoint == "sign"
+
+        base_url = REWARD_URL.get_url(self.region, self.game or game)
+        url = (base_url / endpoint).update_query(**base_url.query)
+
+        return await self.request_lab(
+            method, url, params=params, headers=headers, lang=lang, new_ds=new_ds
+        )
+
+    async def get_reward_info(
+        self,
+        *,
+        game: Optional[Game] = None,
+        lang: Optional[str] = None,
+    ) -> DailyRewardInfo:
+        """Gets the daily reward info for the current user.
+
+        Args:
+            game (Game): The game to request data for. Defaults to None.
+            lang (str): The language to use. Defaults to None.
+
+        Returns:
+            A DailyRewardInfo object containing information about the user's daily reward status.
+        """
+        data = await self.request_daily_reward("info", game=game, lang=lang)
+        return DailyRewardInfo(data["is_sign"], data["total_sign_day"])
+
+    async def get_monthly_rewards(
+        self,
+        *,
+        game: Optional[Game] = None,
+        lang: Optional[str] = None,
+    ) -> List[DailyReward]:
+        """Gets a list of all available rewards for the current month.
+
+        Args:
+            game (Game): The game to request data for. Defaults to None.
+            lang (str): The language to use. Defaults to None.
+
+        Returns:
+            A list of DailyReward objects representing the available rewards for the current month.
+        """
+        data = await self.request_daily_reward(
+            "home",
+            game=game,
+            lang=lang,
+        )
+        return [DailyReward(**i) for i in data["awards"]]
+
+    async def _get_claimed_rewards_page(
+        self,
+        page: int,
+        *,
+        game: Optional[Game] = None,
+        lang: Optional[str] = None,
+    ) -> List[ClaimedDailyReward]:
+        """Gets a single page of claimed rewards for the current user.
+
+        Args:
+            page (int): The page number to retrieve.
+            game (Game): The game to request data for. Defaults to None.
+            lang (str): The language to use. Defaults to None.
+
+        Returns:
+            A list of ClaimedDailyReward objects representing the claimed rewards for the current user on the specified
+                page.
+        """
+        data = await self.request_daily_reward(
+            "award", params=dict(current_page=page), game=game, lang=lang
+        )
+        return [ClaimedDailyReward(**i) for i in data["list"]]
+
+    async def claimed_rewards(
+        self,
+        *,
+        limit: Optional[int] = None,
+        game: Optional[Game] = None,
+        lang: Optional[str] = None,
+    ) -> List[ClaimedDailyReward]:
+        """Gets all claimed rewards for the current user.
+
+        Args:
+            limit (int): The maximum number of rewards to return. Defaults to None.
+            game (Game): The game to request data for. Defaults to None.
+            lang (str): The language to use. Defaults to None.
+
+        Returns:
+            A list of ClaimedDailyReward objects representing the claimed rewards for the current user.
+        """
+        result = []
+        index = 0
+        page = 1
+
+        while True:
+            if page >= 10:
+                break
+
+            fetched_items = await self._get_claimed_rewards_page(
+                page, game=game, lang=lang
+            )
+            if not fetched_items:
+                break
+
+            # Calculate how many items should be added
+            items_to_add = (
+                limit - index
+                if limit is not None and limit - index < len(fetched_items)
+                else len(fetched_items)
+            )
+
+            result.extend(fetched_items[:items_to_add])
+            index += items_to_add
+
+            if limit is not None and index >= limit:
+                break
+
+            page += 1
+
+        return result
+
+    async def claim_daily_reward(
+        self,
+        *,
+        challenge: Optional[str] = None,
+        validate: Optional[str] = None,
+        game: Optional[Game] = None,
+        lang: Optional[str] = None,
+        reward: bool = True,
+    ) -> Optional[DailyReward]:
+        """
+        Signs into lab and claims the daily reward.
+
+        Args:
+            challenge (str): A challenge string for validating the request. Defaults to None.
+            validate (str): A validation string for validating the request. Defaults to None.
+            game (Game): The game to claim the reward for. Defaults to None.
+            lang (str): The language to use. Defaults to None.
+            reward (bool): Whether to return the claimed reward. Defaults to True.
+
+        Returns:
+            If `reward` is True, a DailyReward object representing the claimed reward. Otherwise, None.
+        """
+        await self.request_daily_reward(
+            "sign",
+            method="POST",
+            game=game,
+            lang=lang,
+            challenge=challenge,
+            validate=validate,
+        )
+
+        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]

simnet/client/genshin.py

@@ -1,9 +1,14 @@
+from typing import Optional
+
 from simnet.client.components.auth import AuthClient
 from simnet.client.components.chronicle.genshin import GenshinBattleChronicleClient
 from simnet.client.components.wish.genshin import GenshinWishClient
+from simnet.utils.enum_ import Game
 
 __all__ = ("GenshinClient",)
 
 
 class GenshinClient(GenshinBattleChronicleClient, GenshinWishClient, AuthClient):
     """A simple http client for StarRail endpoints."""
+
+    game: Optional[Game] = Game.GENSHIN

simnet/client/routes.py

@@ -20,6 +20,7 @@ __all__ = (
     "GET_LTOKEN_BY_STOKEN_URL",
     "AUTH_KEY_URL",
     "HK4E_LOGIN_URL",
+    "REWARD_URL",
 )
 
 
@@ -213,3 +214,16 @@ HK4E_LOGIN_URL = InternationalRoute(
     overseas="https://sg-public-api.hoyoverse.com/common/badge/v1/login/account",
     chinese="https://api-takumi.mihoyo.com/common/badge/v1/login/account",
 )
+
+REWARD_URL = GameRoute(
+    overseas=dict(
+        genshin="https://sg-hk4e-api.hoyolab.com/event/sol?act_id=e202102251931481",
+        honkai3rd="https://sg-public-api.hoyolab.com/event/mani?act_id=e202110291205111",
+        hkrpg="https://sg-public-api.hoyolab.com/event/luna/os/?act_id=e202303301540311",
+    ),
+    chinese=dict(
+        genshin="https://api-takumi.mihoyo.com/event/bbs_sign_reward/?act_id=e202009291139501",
+        honkai3rd="https://api-takumi.mihoyo.com/event/luna/?act_id=e202207181446311",
+        hkrpg="https://api-takumi.mihoyo.com/event/luna/?act_id=e202304121516551",
+    ),
+)

simnet/client/starrail.py

@@ -1,9 +1,14 @@
+from typing import Optional
+
 from simnet.client.components.auth import AuthClient
 from simnet.client.components.chronicle.starrail import StarRailBattleChronicleClient
 from simnet.client.components.wish.starrail import StarRailWishClient
+from simnet.utils.enum_ import Game
 
 __all__ = ("StarRailClient",)
 
 
 class StarRailClient(StarRailBattleChronicleClient, StarRailWishClient, AuthClient):
     """A simple http client for StarRail endpoints."""
+
+    game: Optional[Game] = Game.STARRAIL

simnet/models/lab/daily.py

@@ -0,0 +1,61 @@
+import calendar
+from datetime import timezone, timedelta, datetime
+from typing import NamedTuple
+
+from pydantic import Field
+
+from simnet.models.base import APIModel
+
+__all__ = ("ClaimedDailyReward", "DailyReward", "DailyRewardInfo")
+
+
+class DailyRewardInfo(NamedTuple):
+    """Information about the current daily reward status.
+
+    Attributes:
+        signed_in (bool): Whether the user has signed in today.
+        claimed_rewards (int): The number of rewards claimed today.
+    """
+
+    signed_in: bool
+    claimed_rewards: int
+
+    @property
+    def missed_rewards(self) -> int:
+        """The number of rewards that the user has missed since the start of the month."""
+        cn_timezone = timezone(timedelta(hours=8))
+        now = datetime.now(cn_timezone)
+        month_days = calendar.monthrange(now.year, now.month)[1]
+        return month_days - self.claimed_rewards
+
+
+class DailyReward(APIModel):
+    """Claimable daily reward.
+
+    Attributes:
+        name (str): The name of the reward.
+        amount (int): The amount of the reward.
+        icon (str): The URL of the icon for the reward.
+    """
+
+    name: str
+    amount: int = Field(alias="cnt")
+    icon: str
+
+
+class ClaimedDailyReward(APIModel):
+    """Claimed daily reward.
+
+    Attributes:
+        id (int): The ID of the claimed reward.
+        name (str): The name of the reward.
+        amount (int): The amount of the reward.
+        icon (str): The URL of the icon for the reward.
+        time (datetime): The time at which the reward was claimed.
+    """
+
+    id: int
+    name: str
+    amount: int = Field(alias="cnt")
+    icon: str = Field(alias="img")
+    time: datetime = Field(alias="created_at", tzinfo=timezone(timedelta(hours=8)))