Merge upstream dev branch (commit 'd36dbb10555a4e29c8150a0bc3d8f42102…

…570f19') into gdb

genshin/client/components/auth/subclients/game.py

@@ -30,9 +30,12 @@ async def _risky_check(
         if username:
             payload["username"] = username
 
+        headers = auth_utility.RISKY_CHECK_HEADERS.copy()
+        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 +80,8 @@ async def _shield_login(
             raise ValueError("No default game set.")
 
         headers = auth_utility.SHIELD_LOGIN_HEADERS.copy()
+        headers.update(self.custom_headers)
+
         if mmt_result:
             headers["x-rpc-risky"] = mmt_result.to_rpc_risky()
         else:
@@ -108,6 +113,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 +124,28 @@ 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.
         """
         headers = auth_utility.GRANT_TICKET_HEADERS.copy()
+        headers.update(self.custom_headers)
+
         if mmt_result:
             headers["x-rpc-risky"] = mmt_result.to_rpc_risky()
         else:
@@ -141,10 +160,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:
@@ -161,10 +180,11 @@ async def _send_game_verification_email(
     async def _verify_game_email(self, code: str, action_ticket: str) -> DeviceGrantResult:
         """Verify the email code."""
         payload = {"code": code, "ticket": action_ticket}
+        headers = auth_utility.GRANT_TICKET_HEADERS.copy()
+        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"])
@@ -177,17 +197,20 @@ async def _os_game_login(self, uid: str, game_token: str) -> GameLoginResult:
 
         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.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()
 

genshin/client/components/base.py

@@ -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.

genshin/client/components/chronicle/base.py

@@ -47,11 +47,11 @@ async def request_game_record(
         **kwargs: typing.Any,
     ) -> typing.Mapping[str, typing.Any]:
         """Make a request towards the game record endpoint."""
-        base_url = routes.RECORD_URL.get_url(region or self.region)
-
-        if game:
-            base_url = base_url / game.value / "api"
+        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)
         url = base_url / endpoint
 
         mi18n_task = asyncio.create_task(self._fetch_mi18n("bbs", lang=lang or self.lang))

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."""

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,

genshin/client/components/chronicle/zzz.py

@@ -0,0 +1,89 @@
+"""StarRail battle chronicle component."""
+
+import typing
+
+from genshin import errors, types, utility
+from genshin.models.zzz import chronicle as models
+
+from . import base
+
+__all__ = ("ZZZBattleChronicleClient",)
+
+
+class ZZZBattleChronicleClient(base.BaseBattleChronicleClient):
+    """ZZZ battle chronicle component."""
+
+    async def _request_zzz_record(
+        self,
+        endpoint: str,
+        uid: typing.Optional[int] = None,
+        *,
+        method: str = "GET",
+        lang: typing.Optional[str] = None,
+        payload: typing.Optional[typing.Mapping[str, typing.Any]] = None,
+        cache: bool = True,
+    ) -> typing.Mapping[str, typing.Any]:
+        """Get an arbitrary ZZZ object."""
+        payload = dict(payload or {})
+        original_payload = payload.copy()
+
+        uid = uid or await self._get_uid(types.Game.ZZZ)
+        payload = dict(role_id=uid, server=utility.recognize_zzz_server(uid), **payload)
+
+        data, params = None, None
+        if method == "POST":
+            data = payload
+        else:
+            params = payload
+
+        cache_key: typing.Optional[base.ChronicleCacheKey] = None
+        if cache:
+            cache_key = base.ChronicleCacheKey(
+                types.Game.ZZZ,
+                endpoint,
+                uid,
+                lang=lang or self.lang,
+                params=tuple(original_payload.values()),
+            )
+
+        return await self.request_game_record(
+            endpoint,
+            lang=lang,
+            game=types.Game.ZZZ,
+            region=utility.recognize_region(uid, game=types.Game.ZZZ),
+            params=params,
+            data=data,
+            cache=cache_key,
+        )
+
+    async def get_zzz_notes(
+        self,
+        uid: typing.Optional[int] = None,
+        *,
+        lang: typing.Optional[str] = None,
+        autoauth: bool = True,
+    ) -> models.ZZZNotes:
+        """Get ZZZ sticky notes (real-time notes)."""
+        try:
+            data = await self._request_zzz_record("note", uid, lang=lang, cache=False)
+        except errors.DataNotPublic as e:
+            # error raised only when real-time notes are not enabled
+            if uid and (await self._get_uid(types.Game.ZZZ)) != uid:
+                raise errors.GenshinException(e.response, "Cannot view real-time notes of other users.") from e
+            if not autoauth:
+                raise errors.GenshinException(e.response, "Real-time notes are not enabled.") from e
+
+            await self.update_settings(3, True, game=types.Game.ZZZ)
+            data = await self._request_zzz_record("note", uid, lang=lang, cache=False)
+
+        return models.ZZZNotes(**data)
+
+    async def get_zzz_user(
+        self,
+        uid: typing.Optional[int] = None,
+        *,
+        lang: typing.Optional[str] = None,
+    ) -> models.ZZZUserStats:
+        """Get starrail user."""
+        data = await self._request_zzz_record("index", uid, lang=lang, cache=False)
+        return models.ZZZUserStats(**data)

genshin/client/components/hoyolab.py

@@ -124,7 +124,7 @@ async def redeem_code(
 
             game = self.default_game
 
-        if not (game == types.Game.GENSHIN or game == types.Game.STARRAIL):
+        if game not in {types.Game.GENSHIN, types.Game.ZZZ, types.Game.STARRAIL}:
             raise ValueError(f"{game} does not support code redemption.")
 
         uid = uid or await self._get_uid(game)

genshin/client/manager/managers.py

@@ -113,6 +113,7 @@ def proxy(self) -> typing.Optional[yarl.URL]:
     def proxy(self, proxy: typing.Optional[aiohttp.typedefs.StrOrURL]) -> None:
         if proxy is None:
             self._proxy = None
+            self._socks_proxy = None
             return
 
         proxy = yarl.URL(proxy)

genshin/client/routes.py

@@ -134,9 +134,19 @@ def get_url(self, region: types.Region, game: types.Game) -> yarl.URL:
     overseas="https://bbs-api-os.hoyolab.com/community/",
     chinese="https://api-takumi-record.mihoyo.com/community/",
 )
-RECORD_URL = InternationalRoute(
-    overseas="https://bbs-api-os.hoyolab.com/game_record/",
-    chinese="https://api-takumi-record.mihoyo.com/game_record/app/",
+RECORD_URL = GameRoute(
+    overseas=dict(
+        genshin="https://bbs-api-os.hoyolab.com/game_record/genshin/api",
+        hkrpg="https://bbs-api-os.hoyolab.com/game_record/hkrpg/api",
+        honkai3rd="https://bbs-api-os.hoyolab.com/game_record/honkai3rd/api",
+        nap="https://sg-act-nap-api.hoyolab.com/event/game_record_zzz/api/zzz",
+    ),
+    chinese=dict(
+        genshin="https://api-takumi-record.mihoyo.com/game_record/app/genshin/api",
+        hkrpg="https://api-takumi-record.mihoyo.com/game_record/app/hkrpg/api",
+        honkai3rd="https://api-takumi-record.mihoyo.com/game_record/app/honkai3rd/api",
+        nap="https://api-takumi-record.mihoyo.com/event/game_record_zzz/api/zzz",
+    ),
 )
 LINEUP_URL = InternationalRoute(
     overseas="https://sg-public-api.hoyoverse.com/event/simulatoros/",
@@ -186,19 +196,22 @@ def get_url(self, region: types.Region, game: types.Game) -> yarl.URL:
         hkrpg="https://sg-public-api.hoyolab.com/event/luna/os?act_id=e202303301540311",
         tot="https://sg-public-api.hoyolab.com/event/luna/os?act_id=e202202281857121",
         tot_tw="https://sg-public-api.hoyolab.com/event/luna/os?act_id=e202308141137581",
+        nap="https://sg-act-nap-api.hoyolab.com/event/luna/zzz/os?act_id=e202406031448091",
     ),
     chinese=dict(
         genshin="https://api-takumi.mihoyo.com/event/luna/?act_id=e202311201442471",
         honkai3rd="https://api-takumi.mihoyo.com/event/luna/?act_id=e202306201626331",
         hkrpg="https://api-takumi.mihoyo.com/event/luna/?act_id=e202304121516551",
         tot="https://api-takumi.mihoyo.com/event/luna/?act_id=e202202251749321",
+        nap="https://act-nap-api.mihoyo.com/event/luna/zzz/?act_id=e202406242138391",
     ),
 )
 
 CODE_URL = GameRoute(
     overseas=dict(
         genshin="https://sg-hk4e-api.hoyoverse.com/common/apicdkey/api/webExchangeCdkey",
         hkrpg="https://sg-hkrpg-api.hoyoverse.com/common/apicdkey/api/webExchangeCdkey",
+        nap="https://public-operation-nap.hoyoverse.com/common/apicdkey/api/webExchangeCdkey",
     ),
     chinese=dict(),
 )

genshin/models/__init__.py

@@ -6,3 +6,4 @@
 from .hoyolab import *
 from .model import *
 from .starrail import *
+from .zzz import *

genshin/models/genshin/chronicle/__init__.py

@@ -3,6 +3,7 @@
 from .abyss import *
 from .activities import *
 from .characters import *
+from .img_theater import *
 from .notes import *
 from .stats import *
 from .tcg import *