import asyncio from typing import Optional, List, Dict, Any from simnet.client.base import BaseClient from simnet.client.headers import Headers from simnet.client.routes import TAKUMI_URL, HK4E_URL, CODE_URL from simnet.models.lab.announcement import Announcement from simnet.models.lab.record import PartialUser, FullUser from simnet.utils.enum_ import Region from simnet.utils.lang import create_short_lang_code from simnet.utils.player import recognize_genshin_server from simnet.utils.types import HeaderTypes __all__ = ("LabClient",) class LabClient(BaseClient): """LabClient component.""" async def request_bbs( self, endpoint: str, *, lang: Optional[str] = None, region: Optional[Region] = None, method: Optional[str] = None, params: Optional[Dict[str, Any]] = None, data: Any = None, headers: Optional[HeaderTypes] = None, ) -> Dict[str, Any]: """Makes a request to a bbs endpoint. Args: endpoint (str): The URL of the endpoint to make the request to. lang (str, optional): The language code used for the request. Defaults to None. region (Region, optional): The server region used for the request. Defaults to None. method (str, optional): The HTTP method used for the request. Defaults to None. params (dict, optional): The parameters to include in the request. Defaults to None. data (any, optional): The data to include in the request. Defaults to None. headers (dict, optional): The headers to include in the request. Defaults to None. Returns: dict: The response data from the request. """ headers = Headers(headers) lang = lang or self.lang region = region or self.region url = TAKUMI_URL.get_url(region) / endpoint if self.region == Region.CHINESE: headers["Referer"] = "https://www.miyoushe.com/" data = await self.request_lab( url, method=method, params=params, data=data, headers=headers, lang=lang, new_ds=self.region == Region.CHINESE, ) return data async def search_users( self, keyword: str, *, lang: Optional[str] = None, ) -> List[PartialUser]: """Searches for users by keyword. Args: keyword (str): The keyword to search for. lang (str, optional): The language code used for the request. Defaults to None. Returns: list of PartialUser: A list of partial user objects that match the search criteria. """ data = await self.request_bbs( "community/search/wapi/search/user", lang=lang, params=dict(keyword=keyword, page_size=20), ) return [PartialUser(**i["user"]) for i in data["list"]] async def get_user_info( self, accident: Optional[int] = None, *, lang: Optional[str] = None, ) -> FullUser: """Gets user information for the current or specified user. Args: accident (int, optional): The user ID to get information for. Defaults to None. lang (str, optional): The language code used for the request. Defaults to None. Returns: FullUser: A full user object for the specified user. """ if self.region == Region.OVERSEAS: url = "/community/painter/wapi/user/full" elif self.region == Region.CHINESE: url = "/user/wapi/getUserFullInfo" else: raise TypeError(f"{self.region!r} is not a valid region.") data = await self.request_bbs( endpoint=url, lang=lang, params=dict(uid=accident) if accident else None, ) return FullUser(**data["user_info"]) async def get_recommended_users(self, *, limit: int = 200) -> List[PartialUser]: """Gets a list of recommended active users. Args: limit (int, optional): The maximum number of users to retrieve. Defaults to 200. Returns: list of PartialUser: A list of partial user objects for recommended active users. """ data = await self.request_bbs( "community/user/wapi/recommendActive", params=dict(page_size=limit), ) return [PartialUser(**i["user"]) for i in data["list"]] async def get_genshin_announcements( self, player_id: Optional[str] = None, *, lang: Optional[str] = None, ) -> List[Announcement]: """Gets a list of Genshin Impact game announcements. Args: player_id (str, optional): The player ID to get announcements for. Defaults to None. lang (str, optional): The language code used for the request. Defaults to None. Returns: list of Announcement: A list of announcement objects for the specified player. """ player_id = self.player_id or player_id if player_id is None: player_id = 900000005 params = dict( game="hk4e", game_biz="hk4e_global", bundle_id="hk4e_global", platform="pc", region=recognize_genshin_server(player_id), uid=player_id, level=8, lang=lang or self.lang, ) info, details = await asyncio.gather( self.request_bbs( HK4E_URL.get_url() / "announcement/api/getAnnList", lang=lang, params=params, ), self.request_bbs( HK4E_URL.get_url() / "announcement/api/getAnnContent", lang=lang, params=params, ), ) announcements: List[Dict[str, Any]] = [] for sublist in info["list"]: for info in sublist["list"]: detail = next( (i for i in details["list"] if i["ann_id"] == info["ann_id"]), None ) announcements.append({**info, **(detail or {})}) return [Announcement(**i) for i in announcements] async def redeem_code( self, code: str, player_id: Optional[int] = None, *, lang: Optional[str] = None, ) -> None: """Redeems a gift code for the current or specified user. Args: code (str): The gift code to redeem. player_id (int, optional): The player ID to redeem the code for. Defaults to None. lang (str, optional): The language code used for the request. Defaults to None. """ player_id = self.player_id or player_id await self.request_bbs( CODE_URL.get_url(), params=dict( uid=player_id, region=recognize_genshin_server(player_id), cdkey=code, game_biz="hk4e_global", lang=create_short_lang_code(lang or self.lang), ), )