Support mi creator reward query

This commit is contained in:
xtaodada 2024-09-21 23:59:47 +08:00
parent c07234d94e
commit 3e96d77771
Signed by: xtaodada
GPG Key ID: 4CBB3F4FA8C85659
9 changed files with 224 additions and 13 deletions

View File

@ -0,0 +1,119 @@
from typing import Optional, Any, List
from simnet.client.base import BaseClient
from simnet.client.routes import MI_CREATOR_URL
from simnet.models.lab.mi_creator import RewardHistory, RewardHistoryAwardItem
from simnet.utils.enums import Region, Game, SocialPlatform
from simnet.utils.types import QueryParamTypes
__all__ = ("BaseMiCreatorClient",)
class BaseMiCreatorClient(BaseClient):
"""The base class for the MiCreator API client.
This class provides the basic functionality for making requests to the
BaseClient API endpoints. It is meant to be subclassed by other clients
that provide a more specific interface to the BaseClient API.
Attributes:
region (Region): The region associated with the API client.
"""
async def request_mi_creator(
self,
endpoint: str,
data: Optional[Any] = None,
params: Optional[QueryParamTypes] = None,
lang: Optional[str] = None,
platform: Optional[SocialPlatform] = None,
game: Optional[Game] = None,
):
"""Make a request towards the mi creator endpoint.
Args:
endpoint (str): The endpoint to send the request to.
data (Optional[Any], optional): The request payload.
params (Optional[QueryParamTypes], optional): The query parameters for the request.
lang (Optional[str], optional): The language for the response.
platform (Optional[SocialPlatform], optional): The platform associated with the request.
game (Optional[Game], optional): The game associated with the request.
Returns:
The response from the server.
Raises:
NetworkError: If an HTTP error occurs while making the request.
TimedOut: If the request times out.
BadRequest: If the response contains an error.
"""
base_url = MI_CREATOR_URL
if not params:
params = {}
game = game or self.game
params["game"] = {Game.GENSHIN: "hk4e"}.get(game, "")
params["platform"] = platform.value if platform else ""
url = base_url / endpoint
return await self.request_lab(url, data=data, params=params, lang=lang)
async def get_mi_creator_reward_history(
self,
year: int,
month: int,
page: int = 1,
page_size: int = 10,
*,
platform: Optional[SocialPlatform] = None,
game: Optional[Game] = None,
) -> RewardHistory:
"""Get a player reward history.
Args:
year: (int), the year of the request
month: (int), the month of the request
page: (int, optional), the page number, defaults to 1
page_size: (int, optional), the number of items per page, defaults to 10
platform: (SocialPlatform, optional), the platform associated with the request, defaults to None
game: (Game, optional), the game associated with the request, defaults to None
Returns:
RewardHistory: The reward history for the player
"""
params = {
"page": page,
"page_size": page_size,
"no_empty_past_year": True,
"time": f"{year}-{month:02d}",
}
data = await self.request_mi_creator(
"reward/history",
params=params,
platform=platform,
game=game,
)
return RewardHistory(**data)
async def get_mi_creator_reward_count(
self,
platform: Optional[SocialPlatform] = None,
game: Optional[Game] = None,
) -> List[RewardHistoryAwardItem]:
"""Get the reward count for the player.
Args:
platform: (SocialPlatform, optional), the platform associated with the request, defaults to None
game: (Game, optional), the game associated with the request, defaults to None
Returns:
int: The reward count for the player
"""
data = await self.request_mi_creator(
"reward/count/all",
platform=platform,
game=game,
)
item_data = data.get("list", [])
return [RewardHistoryAwardItem(**item) for item in item_data]

View File

@ -6,6 +6,7 @@ from simnet.client.components.chronicle.genshin import GenshinBattleChronicleCli
from simnet.client.components.daily import DailyRewardClient
from simnet.client.components.diary.genshin import GenshinDiaryClient
from simnet.client.components.lab import LabClient
from simnet.client.components.micreator.base import BaseMiCreatorClient
from simnet.client.components.transaction import TransactionClient
from simnet.client.components.verify import VerifyClient
from simnet.client.components.wish.genshin import GenshinWishClient
@ -24,6 +25,7 @@ class GenshinClient(
LabClient,
TransactionClient,
VerifyClient,
BaseMiCreatorClient,
):
"""A simple http client for Genshin endpoints."""

View File

@ -6,6 +6,7 @@ from simnet.client.components.chronicle.genshin import GenshinBattleChronicleCli
from simnet.client.components.daily import DailyRewardClient
from simnet.client.components.diary.genshin import GenshinDiaryClient
from simnet.client.components.lab import LabClient
from simnet.client.components.micreator.base import BaseMiCreatorClient
from simnet.client.components.transaction import TransactionClient
from simnet.client.components.verify import VerifyClient
from simnet.client.components.wish.genshin import GenshinWishClient
@ -22,6 +23,7 @@ class GenshinClient(
LabClient,
TransactionClient,
VerifyClient,
BaseMiCreatorClient,
):
def __init__(
self,

View File

@ -34,6 +34,7 @@ __all__ = (
"GET_FP_URL",
"BBS_URL",
"SELF_HELP_URL",
"MI_CREATOR_URL",
)
@ -362,3 +363,5 @@ SELF_HELP_URL = GameRoute(
nap="https://public-operation-nap.mihoyo.com/common/nap_self_help_query",
),
)
MI_CREATOR_URL = Route("https://api-micreator.mihoyo.com/kolcms_hchcn/v1/hch")

View File

@ -0,0 +1,32 @@
from datetime import datetime
from typing import List
from simnet.models.base import APIModel
class RewardHistoryAwardItem(APIModel):
"""Reward history award item."""
name: str
num: int
pic: str
pic_bg: str
sort: int
is_main: bool
class RewardHistoryItem(APIModel):
"""Reward history item."""
award: List[RewardHistoryAwardItem]
pack_name: str
ts: datetime
task_name: str
class RewardHistory(APIModel):
"""Reward history."""
list: List[RewardHistoryItem]
count: int
time: str

View File

@ -1,6 +1,6 @@
import enum as _enum
__all__ = ("Region", "Game")
__all__ = ("Region", "Game", "SocialPlatform")
class Region(str, _enum.Enum):
@ -31,3 +31,22 @@ class Game(str, _enum.Enum):
HONKAI = "honkai3rd"
STARRAIL = "hkrpg"
ZZZ = "nap"
class SocialPlatform(str, _enum.Enum):
"""
Represents a social platform where a user can be registered.
Attributes:
BILIBILI (SocialPlatform): Represents the social platform "Bilibili".
XIAOHONGSHU (SocialPlatform): Represents the social platform "Xiaohongshu".
DOUYIN (SocialPlatform): Represents the social platform "Douyin".
WECHAT (SocialPlatform): Represents the social platform "WeChat".
QQ (SocialPlatform): Represents the social platform "QQ".
"""
BILIBILI = "bilibili"
XIAOHONGSHU = "xiaohongshu"
DOUYIN = "douyin"
WECHAT = "wechat"
QQ = "qq"

View File

@ -28,9 +28,8 @@ class TestAuthClient:
if genshin_player_id is None:
pytest.skip("Test case test_get_hk4e_token_by_cookie_token skipped: No genshin player id set.")
game_biz = recognize_genshin_game_biz(genshin_player_id)
await auth_client.get_hk4e_token_by_cookie_token(
game_biz, recognize_genshin_server(genshin_player_id), player_id=genshin_player_id
)
region = recognize_genshin_server(genshin_player_id)
await auth_client.get_hk4e_token_by_cookie_token(game_biz, region, player_id=genshin_player_id)
hk4e_token = auth_client.client.cookies.get("e_hk4e_token")
assert hk4e_token is not None
@ -70,18 +69,15 @@ class TestAuthClient:
assert ltoken is not None
@staticmethod
async def test_get_authkey_by_stoken(stoken: str, account_id: int, region: "Region", genshin_player_id: int):
if region != Region.CHINESE:
async def test_get_authkey_by_stoken(auth_client: "AuthClient", stoken: str, genshin_player_id: int):
if auth_client.region != Region.CHINESE:
pytest.skip(
"Test case test_get_authkey_by_stoken skipped:This method is only available for the Chinese region."
)
if stoken is None:
pytest.skip("Test case test_get_authkey_by_stoken skipped: Parameter stoken is None")
async with AuthClient(
cookies={"stoken": stoken},
player_id=genshin_player_id,
account_id=account_id,
region=region,
) as client_instance:
authkey = await client_instance.get_authkey_by_stoken("webview_gacha")
auth_client.player_id = genshin_player_id
game_biz = recognize_genshin_game_biz(genshin_player_id)
region = recognize_genshin_server(genshin_player_id)
authkey = await auth_client.get_authkey_by_stoken("webview_gacha", game_biz, region)
assert authkey is not None

View File

@ -0,0 +1,38 @@
from datetime import datetime
from typing import TYPE_CHECKING
import pytest
import pytest_asyncio
from simnet import GenshinClient
if TYPE_CHECKING:
from simnet.client.cookies import Cookies
from simnet.utils.enums import Region
@pytest_asyncio.fixture
async def genshin_client(genshin_player_id: int, account_id: int, region: "Region", cookies: "Cookies"):
if genshin_player_id is None:
pytest.skip("Test case test_genshin_mi_creator skipped: No genshin player id set.")
async with GenshinClient(
player_id=genshin_player_id,
cookies=cookies,
account_id=account_id,
region=region,
) as client_instance:
yield client_instance
@pytest.mark.asyncio
class TestGenshinMiCreatorClient:
@staticmethod
async def test_get_mi_creator_reward_history(genshin_client: "GenshinClient"):
now = datetime.now()
data = await genshin_client.get_mi_creator_reward_history(now.year, now.month)
assert data.count >= 0
@staticmethod
async def test_get_mi_creator_reward_count(genshin_client: "GenshinClient"):
data = await genshin_client.get_mi_creator_reward_count()
assert len(data) >= 0