Merge branch dev into master (#199)

* Support Imaginarium Theater (#198) * Add models and method * Add test * Continue to use Aliased field cuz it magically works again * Add medal_num field Co-authored-by: omg-xtao <[email protected]> * Rename is_enhanced field * Improve implementation * Add need_detail param --------- Co-authored-by: omg-xtao <[email protected]> * Add Partial ZZZ Support (#200) * Add zzz code redeem route * Modify game check logic * Add zzz server recognition function * Add daily reward URLs * Remove unreachable code * Add real-time notes support * Apply reformat and export models to dunder all * Add fetching user stats * Fix code redemption not working * Sort dunder all * Add buddy_list * Return Game.ZZZ for GenshinAccount.game * Prevent enum crashes (#203) * Add Support for Passing in More Device Info (#202) * Update game auth headers with client custom headers * Add device_id and device_fp setter * Allow passing in custom payload * Allow passing in custom device_name, device_model, and client_type * Refactor * Use GameRoute for battle chronicle route * Fix honkai -> honkai3rd * Forgot to add kwargs to overload * Export ImgTheaterData to dunder all * Add unknown img theater difficulty * Use game-specific game_biz header for game auth * Handle special card wapi endpoints * feat: Add ZZZ gacha support + fix for Chronicled Banner (#206) * Add More ZZZ Features (#207) * Fix get_zz_user method doctsring * Add 2 new icon props and rename 1 icon prop * Add get_zzz_characters method * Add get_bangboos method * Add full agent info * Run nox * Clarify docstrings and add list conversion * Rename methods for better consistency * Add icon field to WEngine Co-authored-by: omg-xtao <[email protected]> * Rename star field to refinement * Fix requesting the wrong endpoint --------- Co-authored-by: omg-xtao <[email protected]> * Add property type * Fix lang arg has no effect on ZZZ endpoitns * Add missing disc prop types * Merge DISC_IMPACT and ENGINE_IMPACT * Add full_name field to ZZZBaseAgent * Fix typo speciality -> specialty * Fix stuff related to AgentSkill * Add game_name and game_logo fields to RecordCard * Add ZZZRecordCard * Support TOT daily reward claiming * Fix Game.TOT not being recognized * Change Game.TOT enum value to 'tot' * Implement recognize_region for Game.TOT * Add banner_art prop to BattleSuit model * Support recognizing game_biz for Game.TOT * Support TOT code redemption * Add caching to get_server_region method * Add region param to redeem_code method * Add boss kills and sub-area explorations to the Exploration model and Long-Term Encounter points (#209) * Add boss kills and sub-area explorations to Exploration * Add stored encounter points and refresh countdown * Ran nox reformatting * Change returning None to returning an empty list * Remove validators as they are no longer required * Add "explored" property to AreaExploration * Support ZZZ Shiyu defense * Fix dunder all not formatted * Remove wiki tests * Remove model reserialization test * Fix raising error for no game even the request doesnt need it * Fix calcualtor test not passing * Update fixture genshin UID and hoyolab ID * Remove unused import * Add type ignore for challenge_time field * Fix type error * Fix invalid import of ModelField * Fix type error * Fix missing type annotation on headers attr * Fix invalid attr access * Update user nickname accordingly * Remove event_loop fixture * Fix test failing caused by get_gacha_items --------- Co-authored-by: omg-xtao <[email protected]> Co-authored-by: Furia <[email protected]>
thesadru · Jul 24, 2024 · abed5b9 · abed5b9
1 parent d35d283
commit abed5b9
Show file tree

Hide file tree

Showing 40 changed files with 1,230 additions and 106 deletions.
diff --git a/genshin/__main__.py b/genshin/__main__.py
@@ -276,7 +276,7 @@ async def wishes(client: genshin.Client, limit: typing.Optional[int] = None) ->
     longest = max(len(v) for v in banner_names.values())
 
     async for wish in client.wish_history(limit=limit):
-        banner = click.style(wish.banner_name.ljust(longest), bold=True)
+        banner = click.style(wish.name.ljust(longest), bold=True)
         click.echo(f"{banner} | {wish.time.astimezone()} - {wish.name} ({'★' * wish.rarity} {wish.type})")
 
 

diff --git a/genshin/client/components/auth/client.py b/genshin/client/components/auth/client.py
@@ -322,7 +322,6 @@ async def verify_mmt(self, mmt_result: MMTResult) -> None:
         if not data["data"]:
             errors.raise_for_retcode(data)
 
-    @base.region_specific(types.Region.OVERSEAS)
     async def os_game_login(
         self,
         account: str,
@@ -339,13 +338,15 @@ async def os_game_login(
         - IncorrectGameAccount: Invalid account provided.
         - IncorrectGamePassword: Invalid password provided.
         """
+        api_server = "api.geetest.com" if self.region is types.Region.CHINESE else "api-na.geetest.com"
+
         result = await self._shield_login(account, password, encrypted=encrypted)
 
         if isinstance(result, RiskyCheckMMT):
             if geetest_solver:
                 mmt_result = await geetest_solver(result)
             else:
-                mmt_result = await server.solve_geetest(result, port=port)
+                mmt_result = await server.solve_geetest(result, port=port, api_server=api_server)
 
             result = await self._shield_login(account, password, encrypted=encrypted, mmt_result=mmt_result)
 
@@ -357,7 +358,7 @@ async def os_game_login(
             if geetest_solver:
                 mmt_result = await geetest_solver(mmt)
             else:
-                mmt_result = await server.solve_geetest(mmt, port=port)
+                mmt_result = await server.solve_geetest(mmt, port=port, api_server=api_server)
 
             await self._send_game_verification_email(result.account.device_grant_ticket, mmt_result=mmt_result)
 

diff --git a/genshin/client/components/auth/subclients/game.py b/genshin/client/components/auth/subclients/game.py
@@ -8,7 +8,7 @@
 
 import aiohttp
 
-from genshin import constants, errors, types
+from genshin import constants, errors
 from genshin.client import routes
 from genshin.client.components import base
 from genshin.models.auth.cookie import DeviceGrantResult, GameLoginResult
@@ -26,13 +26,20 @@ async def _risky_check(
         self, action_type: str, api_name: str, *, username: typing.Optional[str] = None
     ) -> RiskyCheckResult:
         """Check if the given action (endpoint) is risky (whether captcha verification is required)."""
+        if self.default_game is None:
+            raise ValueError("No default game set.")
+
         payload = {"action_type": action_type, "api_name": api_name}
         if username:
             payload["username"] = username
 
+        headers = auth_utility.RISKY_CHECK_HEADERS.copy()
+        headers["x-rpc-game_biz"] = constants.GAME_BIZS[self.region][self.default_game]
+        headers.update(self.custom_headers)
+
         async with aiohttp.ClientSession() as session:
             async with session.post(
-                routes.GAME_RISKY_CHECK_URL.get_url(self.region), json=payload, headers=auth_utility.RISKY_CHECK_HEADERS
+                routes.GAME_RISKY_CHECK_URL.get_url(self.region), json=payload, headers=headers
             ) as r:
                 data = await r.json()
 
@@ -77,6 +84,9 @@ async def _shield_login(
             raise ValueError("No default game set.")
 
         headers = auth_utility.SHIELD_LOGIN_HEADERS.copy()
+        headers["x-rpc-game_biz"] = constants.GAME_BIZS[self.region][self.default_game]
+        headers.update(self.custom_headers)
+
         if mmt_result:
             headers["x-rpc-risky"] = mmt_result.to_rpc_risky()
         else:
@@ -108,6 +118,9 @@ async def _send_game_verification_email(  # noqa: D102 missing docstring in over
         self,
         action_ticket: str,
         *,
+        device_model: typing.Optional[str] = None,
+        device_name: typing.Optional[str] = None,
+        client_type: typing.Optional[int] = None,
         mmt_result: RiskyCheckMMTResult,
     ) -> None: ...
 
@@ -116,17 +129,32 @@ async def _send_game_verification_email(  # noqa: D102 missing docstring in over
         self,
         action_ticket: str,
         *,
+        device_model: typing.Optional[str] = None,
+        device_name: typing.Optional[str] = None,
+        client_type: typing.Optional[int] = None,
         mmt_result: None = ...,
     ) -> typing.Union[None, RiskyCheckMMT]: ...
 
     async def _send_game_verification_email(
-        self, action_ticket: str, *, mmt_result: typing.Optional[RiskyCheckMMTResult] = None
+        self,
+        action_ticket: str,
+        *,
+        device_model: typing.Optional[str] = None,
+        device_name: typing.Optional[str] = None,
+        client_type: typing.Optional[int] = None,
+        mmt_result: typing.Optional[RiskyCheckMMTResult] = None,
     ) -> typing.Union[None, RiskyCheckMMT]:
         """Send email verification code.
 
         Returns `None` if success, `RiskyCheckMMT` if geetest verification is required.
         """
+        if self.default_game is None:
+            raise ValueError("No default game set.")
+
         headers = auth_utility.GRANT_TICKET_HEADERS.copy()
+        headers["x-rpc-game_biz"] = constants.GAME_BIZS[self.region][self.default_game]
+        headers.update(self.custom_headers)
+
         if mmt_result:
             headers["x-rpc-risky"] = mmt_result.to_rpc_risky()
         else:
@@ -141,10 +169,10 @@ async def _send_game_verification_email(
             "way": "Way_Email",
             "action_ticket": action_ticket,
             "device": {
-                "device_model": "iPhone15,4",
-                "device_id": auth_utility.DEVICE_ID,
-                "client": 1,
-                "device_name": "iPhone",
+                "device_model": device_model or "iPhone15,4",
+                "device_id": self.device_id or auth_utility.DEVICE_ID,
+                "client": client_type or 1,
+                "device_name": device_name or "iPhone",
             },
         }
         async with aiohttp.ClientSession() as session:
@@ -160,34 +188,42 @@ async def _send_game_verification_email(
 
     async def _verify_game_email(self, code: str, action_ticket: str) -> DeviceGrantResult:
         """Verify the email code."""
+        if self.default_game is None:
+            raise ValueError("No default game set.")
+
         payload = {"code": code, "ticket": action_ticket}
+        headers = auth_utility.GRANT_TICKET_HEADERS.copy()
+        headers["x-rpc-game_biz"] = constants.GAME_BIZS[self.region][self.default_game]
+        headers.update(self.custom_headers)
+
         async with aiohttp.ClientSession() as session:
-            async with session.post(
-                routes.DEVICE_GRANT_URL.get_url(self.region), json=payload, headers=auth_utility.GRANT_TICKET_HEADERS
-            ) as r:
+            async with session.post(routes.DEVICE_GRANT_URL.get_url(self.region), json=payload, headers=headers) as r:
                 data = await r.json()
 
         return DeviceGrantResult(**data["data"])
 
-    @base.region_specific(types.Region.OVERSEAS)
     async def _os_game_login(self, uid: str, game_token: str) -> GameLoginResult:
         """Log in to the game."""
         if self.default_game is None:
             raise ValueError("No default game set.")
 
         payload = {
             "channel_id": 1,
-            "device": auth_utility.DEVICE_ID,
+            "device": self.device_id or auth_utility.DEVICE_ID,
             "app_id": constants.APP_IDS[self.default_game][self.region],
         }
         payload["data"] = json.dumps({"uid": uid, "token": game_token, "guest": False})
         payload["sign"] = auth_utility.generate_sign(payload, constants.APP_KEYS[self.default_game][self.region])
 
+        headers = auth_utility.GAME_LOGIN_HEADERS.copy()
+        headers["x-rpc-game_biz"] = constants.GAME_BIZS[self.region][self.default_game]
+        headers.update(self.custom_headers)
+
         async with aiohttp.ClientSession() as session:
             async with session.post(
                 routes.GAME_LOGIN_URL.get_url(self.region, self.default_game),
                 json=payload,
-                headers=auth_utility.GAME_LOGIN_HEADERS,
+                headers=headers,
             ) as r:
                 data = await r.json()
 

diff --git a/genshin/client/components/base.py b/genshin/client/components/base.py
@@ -94,7 +94,7 @@ def __init__(
         self.uid = uid
         self.hoyolab_id = hoyolab_id
 
-        self.custom_headers = dict(headers or {})
+        self.custom_headers: typing.Dict[str, str] = dict(headers or {})
         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 {})
 
@@ -111,6 +111,24 @@ def __repr__(self) -> str:
         )
         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.

diff --git a/genshin/client/components/chronicle/base.py b/genshin/client/components/chronicle/base.py
@@ -48,15 +48,14 @@ async def request_game_record(
         **kwargs: typing.Any,
     ) -> typing.Mapping[str, typing.Any]:
         """Make a request towards the game record endpoint."""
-        game = game or self.default_game
-        if game is None:
-            raise RuntimeError("No default game set.")
+        if is_card_wapi:
+            base_url = routes.CARD_WAPI_URL.get_url(region or self.region)
+        else:
+            game = game or self.default_game
+            if game is None:
+                raise RuntimeError("No default game set.")
+            base_url = routes.RECORD_URL.get_url(region or self.region, game)
 
-        base_url = (
-            routes.RECORD_URL.get_url(region or self.region, game)
-            if not is_card_wapi
-            else routes.CARD_WAPI_URL.get_url(region or self.region)
-        )
         url = base_url / endpoint
 
         mi18n_task = asyncio.create_task(self._fetch_mi18n("bbs", lang=lang or self.lang))

diff --git a/genshin/client/components/chronicle/client.py b/genshin/client/components/chronicle/client.py
@@ -1,6 +1,6 @@
 """Battle chronicle component."""
 
-from . import genshin, honkai, starrail
+from . import genshin, honkai, starrail, zzz
 
 __all__ = ["BattleChronicleClient"]
 
@@ -9,5 +9,6 @@ class BattleChronicleClient(
     genshin.GenshinBattleChronicleClient,
     honkai.HonkaiBattleChronicleClient,
     starrail.StarRailBattleChronicleClient,
+    zzz.ZZZBattleChronicleClient,
 ):
     """Battle chronicle component."""
diff --git a/genshin/client/components/chronicle/genshin.py b/genshin/client/components/chronicle/genshin.py
@@ -107,6 +107,23 @@ async def get_genshin_spiral_abyss(
 
         return models.SpiralAbyss(**data)
 
+    async def get_imaginarium_theater(
+        self,
+        uid: int,
+        *,
+        previous: bool = False,
+        need_detail: bool = True,
+        lang: typing.Optional[str] = None,
+    ) -> models.ImgTheater:
+        """Get Genshin Impact imaginarium theater runs."""
+        payload = {
+            "schedule_type": 2 if previous else 1,  # There's 1 season for now but I assume it works like this
+            "need_detail": str(need_detail).lower(),
+        }
+        data = await self._request_genshin_record("role_combat", uid, lang=lang, payload=payload)
+
+        return models.ImgTheater(**data)
+
     async def get_genshin_notes(
         self,
         uid: typing.Optional[int] = None,