feat: Add OTP-related methods

genshin/client/components/geetest/client.py

@@ -239,6 +239,81 @@ async def _verify_email(self, code: str, ticket: typing.Dict[str, typing.Any]) -
 
         return None
 
+    async def _send_mobile_otp(
+        self,
+        mobile: str,
+        *,
+        geetest: typing.Optional[typing.Dict[str, typing.Any]] = None,
+    ) -> typing.Dict[str, typing.Any] | None:
+        """Attempt to send OTP to the provided mobile number.
+
+        May return aigis headers if captcha is triggered, None otherwise.
+        """
+        headers = {
+            **geetest_utility.CN_LOGIN_HEADERS,
+            "ds": ds_utility.generate_dynamic_secret(constants.DS_SALT["cn_signin"]),
+        }
+        if geetest:
+            mmt_data = geetest["data"]
+            session_id = geetest["session_id"]
+            headers["x-rpc-aigis"] = geetest_utility.get_aigis_header(session_id, mmt_data)
+
+        payload = {
+            "mobile": geetest_utility.encrypt_geetest_credentials(mobile, self._region),
+            "area_code": geetest_utility.encrypt_geetest_credentials("+86", self._region),
+        }
+
+        async with aiohttp.ClientSession() as session:
+            async with session.post(
+                routes.MOBILE_OTP_URL.get_url(),
+                json=payload,
+                headers=headers,
+            ) as r:
+                data = await r.json()
+
+        if data["retcode"] == -3101:
+            # Captcha triggered
+            aigis = json.loads(r.headers["x-rpc-aigis"])
+            aigis["data"] = json.loads(aigis["data"])
+            return aigis
+
+        if not data["data"]:
+            errors.raise_for_retcode(data)
+
+        return None
+
+    async def _login_with_mobile_otp(self, mobile: str, otp: str) -> typing.Dict[str, typing.Any]:
+        """Login with OTP and mobile number.
+
+        Returns cookies if OTP matches the one sent, raises an error otherwise.
+        """
+        headers = {
+            **geetest_utility.CN_LOGIN_HEADERS,
+            "ds": ds_utility.generate_dynamic_secret(constants.DS_SALT["cn_signin"]),
+        }
+
+        payload = {
+            "mobile": geetest_utility.encrypt_geetest_credentials(mobile, self._region),
+            "area_code": geetest_utility.encrypt_geetest_credentials("+86", self._region),
+            "captcha": otp,
+        }
+
+        async with aiohttp.ClientSession() as session:
+            async with session.post(
+                routes.MOBILE_LOGIN_URL.get_url(),
+                json=payload,
+                headers=headers,
+            ) as r:
+                data = await r.json()
+
+        if not data["data"]:
+            errors.raise_for_retcode(data)
+
+        cookies = {cookie.key: cookie.value for cookie in r.cookies.values()}
+        self.set_cookies(cookies)
+
+        return cookies
+
     async def login_with_password(
         self,
         account: str,
@@ -286,8 +361,7 @@ async def cn_login_by_password(
     ) -> typing.Dict[str, str]:
         """Login with a password via Miyoushe loginByPassword endpoint.
 
-        Note that this will start a webserver if captcha is
-        triggered and `geetest_solver` is not passed.
+        Note that this will start a webserver if captcha is triggered and `geetest_solver` is not passed.
         """
         result = await self._cn_login_by_password(account, password)
 
@@ -302,6 +376,45 @@ async def cn_login_by_password(
 
         return await self._cn_login_by_password(account, password, geetest=geetest)
 
+    async def check_mobile_number_validity(self, mobile: str) -> bool:
+        """Check if a mobile number is valid (it's registered on Miyoushe).
+
+        Returns True if the mobile number is valid, False otherwise.
+        """
+        async with aiohttp.ClientSession() as session:
+            async with session.get(
+                routes.CHECK_MOBILE_VALIDITY_URL.get_url(),
+                params={"mobile": mobile},
+            ) as r:
+                data = await r.json()
+
+        return data["data"]["status"] != data["data"]["is_registable"]
+
+    async def login_with_mobile_number(
+        self,
+        mobile: str,
+    ) -> typing.Dict[str, str]:
+        """Login with mobile number, returns cookies.
+
+        Only works for Chinese region (Miyoushe) users, do not include area code (+86) in the mobile number.
+        Steps:
+        1. Sends OTP to the provided mobile number.
+        1-1. If captcha is triggered, prompts the user to solve it.
+        2. Lets user enter the OTP.
+        3. Logs in with the OTP.
+        4. Returns cookies.
+        """
+        result = await self._send_mobile_otp(mobile)
+
+        if result is not None and "session_id" in result:
+            # Captcha triggered
+            geetest = await server.solve_geetest(result)
+            await self._send_mobile_otp(mobile, geetest=geetest)
+
+        otp = await server.enter_otp()
+        cookies = await self._login_with_mobile_otp(mobile, otp)
+        return cookies
+
     async def login_with_app_password(
         self,
         account: str,

genshin/client/components/geetest/server.py

@@ -14,7 +14,7 @@
 __all__ = ["get_page", "launch_webapp", "solve_geetest", "verify_email"]
 
 
-def get_page(page: typing.Literal["captcha", "verify-email"]) -> str:
+def get_page(page: typing.Literal["captcha", "verify-email", "enter-otp"]) -> str:
     """Get the HTML page."""
     return (
         """
@@ -84,7 +84,7 @@ def get_page(page: typing.Literal["captcha", "verify-email"]) -> str:
 
 
 async def launch_webapp(
-    page: typing.Literal["captcha", "verify-email"],
+    page: typing.Literal["captcha", "verify-email", "enter-otp"],
     *,
     port: int = 5000,
     mmt: typing.Optional[typing.Dict[str, typing.Any]] = None,
@@ -101,6 +101,10 @@ async def captcha(request: web.Request) -> web.StreamResponse:
     async def verify_email(request: web.Request) -> web.StreamResponse:
         return web.Response(body=get_page("verify-email"), content_type="text/html")
 
+    @routes.get("/enter-otp")
+    async def enter_otp(request: web.Request) -> web.StreamResponse:
+        return web.Response(body=get_page("enter-otp"), content_type="text/html")
+
     @routes.get("/gt.js")
     async def gt(request: web.Request) -> web.StreamResponse:
         async with aiohttp.ClientSession() as session:
@@ -162,3 +166,11 @@ async def verify_email(
     code = data["code"]
 
     return await client._verify_email(code, ticket)
+
+
+async def enter_otp(port: int = 5000) -> str:
+    """Lets user enter the OTP."""
+    # The enter-otp page is the same as verify-email page.
+    data = await launch_webapp("enter-otp", port=port)
+    code = data["code"]
+    return code

genshin/client/routes.py

@@ -224,3 +224,7 @@ def get_url(self, region: types.Region, game: types.Game) -> yarl.URL:
     "https://sg-public-api.hoyoverse.com/account/ma-verifier/api/createEmailCaptchaByActionTicket"
 )
 VERIFY_EMAIL_URL = Route("https://sg-public-api.hoyoverse.com/account/ma-verifier/api/verifyActionTicketPartly")
+
+CHECK_MOBILE_VALIDITY_URL = Route("https://webapi.account.mihoyo.com/Api/is_mobile_registrable")
+MOBILE_OTP_URL = Route("https://passport-api.miyoushe.com/account/ma-cn-verifier/verifier/createLoginCaptcha")
+MOBILE_LOGIN_URL = Route("https://passport-api.miyoushe.com/account/ma-cn-passport/web/loginByMobileCaptcha")

genshin/errors.py

@@ -31,7 +31,11 @@ class GenshinException(Exception):
     original: str = ""
     msg: str = ""
 
-    def __init__(self, response: typing.Mapping[str, typing.Any] = {}, msg: typing.Optional[str] = None) -> None:
+    def __init__(
+        self,
+        response: typing.Mapping[str, typing.Any] = {},
+        msg: typing.Optional[str] = None,
+    ) -> None:
         self.retcode = response.get("retcode", self.retcode)
         self.original = response.get("message", "")
         self.msg = msg or self.msg or self.original
@@ -176,6 +180,12 @@ class AccountHasLocked(GenshinException):
     msg = "Account has been locked because exceeded password limit. Please wait 20 minute and try again"
 
 
+class WrongOTP(GenshinException):
+    """Wrong OTP code."""
+
+    msg = "The provided OTP code is wrong."
+
+
 _TGE = typing.Type[GenshinException]
 _errors: typing.Dict[int, typing.Union[_TGE, str, typing.Tuple[_TGE, typing.Optional[str]]]] = {
     # misc hoyolab
@@ -189,7 +199,10 @@ class AccountHasLocked(GenshinException):
     # database game record
     10101: TooManyRequests,
     10102: DataNotPublic,
-    10103: (InvalidCookies, "Cookies are valid but do not have a hoyolab account bound to them."),
+    10103: (
+        InvalidCookies,
+        "Cookies are valid but do not have a hoyolab account bound to them.",
+    ),
     10104: "Cannot view real-time notes of other users.",
     # calculator
     -500001: "Invalid fields in calculation.",
@@ -210,7 +223,10 @@ class AccountHasLocked(GenshinException):
     -2016: RedemptionCooldown,
     -2017: RedemptionClaimed,
     -2018: RedemptionClaimed,
-    -2021: (RedemptionException, "Cannot claim codes for accounts with adventure rank lower than 10."),
+    -2021: (
+        RedemptionException,
+        "Cannot claim codes for accounts with adventure rank lower than 10.",
+    ),
     # rewards
     -5003: AlreadyClaimed,
     # chinese
@@ -219,6 +235,7 @@ class AccountHasLocked(GenshinException):
     # account
     -3208: AccountLoginFail,
     -3202: AccountHasLocked,
+    -3205: WrongOTP,
 }
 
 ERRORS: typing.Dict[int, typing.Tuple[_TGE, typing.Optional[str]]] = {