Version 0.1.3

apis: Implement `LoginViaCookie`

Author: mos9527

pyncm_async/__init__.py

@@ -2,9 +2,9 @@
 """PyNCM-Async 网易云音乐 Python 异步 API / 下载工具"""
 __VERSION_MAJOR__ = 0
 __VERSION_MINOR__ = 1
-__VERSION_PATCH__ = 2
+__VERSION_PATCH__ = 3
 
-__version__ = '%s.%s.%s' % (__VERSION_MAJOR__,__VERSION_MINOR__,__VERSION_PATCH__)
+__version__ = "%s.%s.%s" % (__VERSION_MAJOR__, __VERSION_MINOR__, __VERSION_PATCH__)
 
 
 from threading import current_thread
@@ -17,12 +17,13 @@
 import httpx
 
 logger = logging.getLogger("pyncm.api")
-if 'PYNCM_DEBUG' in os.environ:
-    debug_level = os.environ['PYNCM_DEBUG'].upper()
-    if not debug_level in {'CRITICAL', 'DEBUG', 'ERROR', 'FATAL', 'INFO', 'WARNING'}:
-        debug_level = 'DEBUG'
-    logging.basicConfig(level=debug_level,
-                        format="[%(levelname).4s] %(name)s %(message)s")
+if "PYNCM_DEBUG" in os.environ:
+    debug_level = os.environ["PYNCM_DEBUG"].upper()
+    if not debug_level in {"CRITICAL", "DEBUG", "ERROR", "FATAL", "INFO", "WARNING"}:
+        debug_level = "DEBUG"
+    logging.basicConfig(
+        level=debug_level, format="[%(levelname).4s] %(name)s %(message)s"
+    )
 
 DEVICE_ID_DEFAULT = "pyncm!"
 # This sometimes fails with some strings, for no particular reason. Though `pyncm!` seem to work everytime..?
@@ -50,7 +51,7 @@ class Session(httpx.AsyncClient):
 
     ```python
     # 利用全局 Session 完成该 API Call
-    await LoginViaEmail(...) 
+    await LoginViaEmail(...)
     async with CreateNewSession(): # 建立新的 Session，并进入该 Session, 在 `with` 内的 API 将由该 Session 完成
         await LoginViaCellPhone(...)
     # 离开 Session. 此后 API 将继续由全局 Session 管理
@@ -61,9 +62,12 @@ class Session(httpx.AsyncClient):
 
     获取其他具体信息请参考该文档注释
     """
+
     HOST = "music.163.com"
     # 网易云音乐 API 服务器域名，可直接改为代理服务器之域名
-    UA_DEFAULT = "Mozilla/5.0 ([email protected]/mos9527/pyncm) Chrome/PyNCM.%s" % __version__
+    UA_DEFAULT = (
+        "Mozilla/5.0 ([email protected]/mos9527/pyncm) Chrome/PyNCM.%s" % __version__
+    )
     # Weapi 使用的 UA
     UA_EAPI = "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Safari/537.36 Chrome/91.0.4472.164 NeteaseMusicDesktop/2.10.2.200154"
     # EAPI 使用的 UA，不推荐更改
@@ -88,11 +92,7 @@ def __init__(self, *args, **kwargs):
             "User-Agent": self.UA_DEFAULT,
             "Referer": self.HOST,
         }
-        self.login_info = {
-            "success": False, 
-            "tick": time(), 
-            "content": None
-        }
+        self.login_info = {"success": False, "tick": time(), "content": None}
         self.eapi_config = {
             "os": "ios",
             "appver": "9.0.0",
@@ -180,9 +180,8 @@ async def request(
         ),
         "cookies": (
             lambda self: [
-                {"name": c.name, "value": c.value,
-                    "domain": c.domain, "path": c.path}
-                for c in getattr(getattr(self, "cookies"),"jar")
+                {"name": c.name, "value": c.value, "domain": c.domain, "path": c.path}
+                for c in getattr(getattr(self, "cookies"), "jar")
             ],
             lambda self, cookies: [
                 getattr(self, "cookies").set(**cookie) for cookie in cookies
@@ -203,6 +202,7 @@ def load(self, dumped):
             self._session_info[k][1](self, v)
         return True
 
+
 class SessionManager:
     """PyNCM Session 单例储存对象"""
 
@@ -217,7 +217,8 @@ def get(self):
     def set(self, session):
         if SESSION_STACK.get(current_thread(), None):
             raise Exception(
-                "Current Session is in `with` block, which cannot be reassigned.")
+                "Current Session is in `with` block, which cannot be reassigned."
+            )
         self.session = session
 
     # Session serialization
@@ -243,23 +244,29 @@ def stringify(session: Session) -> str:
         from json import dumps
         from zlib import compress
         from base64 import b64encode
-        return 'PYNCM' + b64encode(compress(dumps(session.dump()).encode())).decode()
+
+        return "PYNCM" + b64encode(compress(dumps(session.dump()).encode())).decode()
 
     @staticmethod
     def parse(dump: str) -> Session:
         """反序列化 `str` 为 `Session`"""
-        if dump[:5] == 'PYNCM':  # New marshaler (compressed,base64 encoded) has magic header
+        if (
+            dump[:5] == "PYNCM"
+        ):  # New marshaler (compressed,base64 encoded) has magic header
             from json import loads
             from zlib import decompress
             from base64 import b64decode
+
             session = Session()
             session.load(loads(decompress(b64decode(dump[5:])).decode()))
             return session
         else:
             return SessionManager.parse_legacy(dump)
 
+
 sessionManager = SessionManager()
 
+
 def GetCurrentSession() -> Session:
     """获取当前正在被 PyNCM 使用的 Session / 登录态"""
     return sessionManager.get()

pyncm_async/apis/login.py

@@ -27,7 +27,7 @@ def WriteLoginInfo(response, session):
         session.login_info["success"] = False
         raise LoginFailedException(session.login_info["content"])
     session.login_info["success"] = True
-    session.csrf_token = session.cookies.get('__csrf')
+    session.csrf_token = session.cookies.get("__csrf")
 
 
 @WeapiCryptoRequest
@@ -40,14 +40,14 @@ def LoginLogout():
     return "/weapi/logout", {}
 
 
-@WeapiCryptoRequest
+@EapiCryptoRequest
 def LoginRefreshToken():
     """网页端 - 刷新登录令牌
 
     Returns:
         dict
     """
-    return "/weapi/w/login/cellphone", {}
+    return "/eapi/login/token/refresh", {}
 
 
 @WeapiCryptoRequest
@@ -102,7 +102,31 @@ def GetCurrentLoginStatus():
     return "/weapi/w/nuser/account/get", {}
 
 
-async def LoginViaCellphone(phone="", password="",passwordHash="",captcha="", ctcode=86, remeberLogin=True, session=None) -> dict:
+async def LoginViaCookie(MUSIC_U="", **kwargs):
+    """通过 Cookie 登陆
+
+    Args:
+        MUSIC_U (str, optional): Cookie 中的 MUSIC_U. Defaults to ''.
+
+    Returns:
+        dict
+    """
+    session = GetCurrentSession()
+    session.cookies.update({"MUSIC_U": MUSIC_U, **kwargs})
+    resp = await GetCurrentLoginStatus()
+    WriteLoginInfo(resp, session)
+    return {"code": 200, "result": session.login_info}
+
+
+async def LoginViaCellphone(
+    phone="",
+    password="",
+    passwordHash="",
+    captcha="",
+    ctcode=86,
+    remeberLogin=True,
+    session=None,
+) -> dict:
     """PC 端 - 手机号登陆
 
     * 若同时指定 password 和 passwordHash, 优先使用 password
@@ -114,8 +138,8 @@ async def LoginViaCellphone(phone="", password="",passwordHash="",captcha="", ct
         remeberLogin (bool, optional): 是否‘自动登录’，设置 `False` 可能导致权限问题. Defaults to True.
         * 以下验证方式有 1 个含参即可
         password (str, optional): 明文密码. Defaults to ''.
-        passwordHash (str, optional): 密码md5哈希. Defaults to ''.        
-        captcha (str, optional): 手机验证码. 需要已在同一 Session 中发送过 SetSendRegisterVerifcationCodeViaCellphone. Defaults to ''.        
+        passwordHash (str, optional): 密码md5哈希. Defaults to ''.
+        captcha (str, optional): 手机验证码. 需要已在同一 Session 中发送过 SetSendRegisterVerifcationCodeViaCellphone. Defaults to ''.
 
     Raises:
         LoginFailedException: 登陆失败时发生
@@ -126,43 +150,47 @@ async def LoginViaCellphone(phone="", password="",passwordHash="",captcha="", ct
     path = "/eapi/w/login/cellphone"
     session = session or GetCurrentSession()
     if password:
-        passwordHash = HashHexDigest(password)        
-    
+        passwordHash = HashHexDigest(password)
+
     if not (passwordHash or captcha):
         raise LoginFailedException("未提供密码或验证码")
 
-    auth_token = {"password": str(passwordHash)} if not captcha else {"captcha": str(captcha)}
+    auth_token = (
+        {"password": str(passwordHash)} if not captcha else {"captcha": str(captcha)}
+    )
 
     login_status = await EapiCryptoRequest(
         lambda: (
             path,
             {
-                "type": '1',
-                "phone": str(phone),                
+                "type": "1",
+                "phone": str(phone),
                 "remember": str(remeberLogin).lower(),
                 "countrycode": str(ctcode),
-                "checkToken" : "",
-                **auth_token
+                "checkToken": "",
+                **auth_token,
             },
         )
     )(session=session)
-    
+
     WriteLoginInfo(login_status, session)
-    return {'code':200,'result':session.login_info}
+    return {"code": 200, "result": session.login_info}
 
 
-async def LoginViaEmail(email="", password="",passwordHash="", remeberLogin=True, session=None) -> dict:
+async def LoginViaEmail(
+    email="", password="", passwordHash="", remeberLogin=True, session=None
+) -> dict:
     """网页端 - 邮箱登陆
 
     * 若同时指定 password 和 passwordHash, 优先使用 password
-    
+
     Args:
         email (str, optional): 邮箱地址. Defaults to ''.
         remeberLogin (bool, optional): 是否‘自动登录’，设置 `False` 可能导致权限问题. Defaults to True.
         * 以下验证方式有 1 个含参即可
         password (str, optional): 明文密码. Defaults to ''.
-        passwordHash (str, optional): 密码md5哈希. Defaults to ''.        
-        
+        passwordHash (str, optional): 密码md5哈希. Defaults to ''.
+
     Raises:
         LoginFailedException: 登陆失败时发生
 
@@ -172,8 +200,8 @@ async def LoginViaEmail(email="", password="",passwordHash="", remeberLogin=True
     path = "/eapi/login"
     session = session or GetCurrentSession()
     if password:
-        passwordHash = HashHexDigest(password)        
-    
+        passwordHash = HashHexDigest(password)
+
     if not passwordHash:
         raise LoginFailedException("未提供密码")
 
@@ -183,56 +211,55 @@ async def LoginViaEmail(email="", password="",passwordHash="", remeberLogin=True
         lambda: (
             path,
             {
-                "type": '1',
-                "username": str(email),                
-                "remember": str(remeberLogin).lower(),                
-                **auth_token
+                "type": "1",
+                "username": str(email),
+                "remember": str(remeberLogin).lower(),
+                **auth_token,
             },
         )
     )(session=session)
-    
-    WriteLoginInfo(login_status,session)
-    return {'code':200,'result':session.login_info}
+
+    WriteLoginInfo(login_status, session)
+    return {"code": 200, "result": session.login_info}
+
 
 async def LoginViaAnonymousAccount(deviceId=None, session=None):
-    '''PC 端 - 游客登陆
+    """PC 端 - 游客登陆
 
     Args:
         deviceId (str optional): 设备 ID. 设置非 None 将同时改变 Session 的设备 ID. Defaults to None.
-    
+
     Notes:
         Session 默认使用 `pyncm!` 作为设备 ID
 
     Returns:
         dict
-    '''
+    """
     session = session or GetCurrentSession()
     if not deviceId:
-        deviceId = session.deviceId    
+        deviceId = session.deviceId
     login_status = WeapiCryptoRequest(
-        lambda: ("/api/register/anonimous" , {
-        "username" : b64encode(
-            ('%s %s' % (
-                deviceId,
-                cloudmusic_dll_encode_id(deviceId))).encode()
-        ).decode()
-        }
+        lambda: (
+            "/api/register/anonimous",
+            {
+                "username": b64encode(
+                    ("%s %s" % (deviceId, cloudmusic_dll_encode_id(deviceId))).encode()
+                ).decode()
+            },
         )
-    )(session=session)    
-    assert login_status['code'] == 200,"匿名登陆失败"
-    WriteLoginInfo({
-        **login_status,
-        'profile':{
-            'nickname' : 'Anonymous',
-            **login_status
+    )(session=session)
+    assert login_status["code"] == 200, "匿名登陆失败"
+    WriteLoginInfo(
+        {
+            **login_status,
+            "profile": {"nickname": "Anonymous", **login_status},
+            "account": {"id": login_status["userId"], **login_status},
         },
-        'account':{
-            'id' : login_status['userId'],
-            **login_status
-        }
-    }, session)
+        session,
+    )
     return session.login_info
 
+
 @WeapiCryptoRequest
 def SetSendRegisterVerifcationCodeViaCellphone(cell: str, ctcode=86):
     """网页端 - 发送验证码
@@ -294,6 +321,7 @@ def SetRegisterAccountViaCellphone(
         "phone": str(cell),
     }
 
+
 @EapiCryptoRequest
 def CheckIsCellphoneRegistered(cell: str, prefix=86):
     """移动端 - 检查某手机号是否已注册