rechore: login & request_sync

bilibili_api/__init__.py

@@ -20,6 +20,8 @@
     set_session,
     get_aiohttp_session,
     set_aiohttp_session,
+    get_httpx_sync_session,
+    set_httpx_sync_session
 )
 from .errors import (
     LoginError,

bilibili_api/login.py

@@ -31,7 +31,7 @@
 from .utils.credential import Credential
 from .exceptions.LoginError import LoginError
 from .utils.network import to_form_urlencoded, Api
-from .utils.network import HEADERS, get_session, get_spi_buvid_sync
+from .utils.network import HEADERS, get_session, get_spi_buvid_sync, get_httpx_sync_session
 from .utils.captcha import get_result, close_server, start_server
 from .utils.safecenter_captcha import get_result as safecenter_get_result
 from .utils.safecenter_captcha import close_server as safecenter_close_server
@@ -261,7 +261,7 @@ def parse_tv_resp(events: dict) -> Credential:
             cookies["bili_jct"] = cookie["value"]
         elif cookie["name"] == "DedeUserID":
             cookies["dedeuserid"] = cookie["value"]
-    
+
     return Credential(**cookies)
 
 
@@ -348,7 +348,7 @@ def login_with_password(username: str, password: str) -> Union[Credential, "Chec
     """
     api_token = API["password"]["get_token"]
     geetest_data = get_geetest()
-    sess = httpx.Client()
+    sess = get_httpx_sync_session()
     token_data = json.loads(sess.get(api_token["url"]).text)
     hash_ = token_data["data"]["hash"]
     key = token_data["data"]["key"]

bilibili_api/login_func.py

@@ -24,11 +24,13 @@ class QrCodeLoginEvents(enum.Enum):
 
     + SCAN: 未扫描二维码
     + CONF: 未确认登录
+    + TIMEOUT: 二维码过期
     + DONE: 成功
     """
 
     SCAN = "scan"
     CONF = "confirm"
+    TIMEOUT = "timeout"
     DONE = "done"
 
 
@@ -37,15 +39,15 @@ def get_qrcode() -> Tuple[Picture, str]:
     获取二维码及登录密钥（后面有用）
 
     Returns:
-        Tuple[dir, str]: 第一项是二维码图片地址（本地缓存）和登录密钥。登录密钥需要保存。
+        Tuple[Picture, str]: 第一项是二维码图片地址（本地缓存）和登录密钥。登录密钥需要保存。
     """
     login_data = login.update_qrcode_data()
     login_key = login_data["qrcode_key"]
     img = login.make_qrcode(login_data["url"])
     return (Picture.from_file(img), login_key)
 
 
-def check_qrcode_events(login_key) -> Tuple[QrCodeLoginEvents, Union[str, Credential]]:
+def check_qrcode_events(login_key: str) -> Tuple[QrCodeLoginEvents, Union[str, Credential]]:
     """
     检查登录状态。（建议频率 1s，这个 API 也有风控！）
 
@@ -61,6 +63,8 @@ def check_qrcode_events(login_key) -> Tuple[QrCodeLoginEvents, Union[str, Creden
         return QrCodeLoginEvents.SCAN, events["message"]
     elif events["code"] == 86090:
         return QrCodeLoginEvents.CONF, events["message"]
+    elif events["code"] == 86038:
+        return QrCodeLoginEvents.TIMEOUT, events["message"]
     elif events["code"] == 0:
         url: str = events["data"]["url"]
         cookies_list = url.split("?")[1].split("&")
@@ -80,6 +84,43 @@ def check_qrcode_events(login_key) -> Tuple[QrCodeLoginEvents, Union[str, Creden
         raise LoginError(events["message"])
 
 
+def get_tv_qrcode() -> Tuple[Picture, str]:
+    """
+    获取 TV 端登录二维码及登录密钥（后面有用）
+
+    Returns:
+        Tuple[Picture, str]: 第一项是二维码图片地址（本地缓存）和登录密钥。登录密钥需要保存。
+    """
+    qrcode_data = login.update_tv_qrcode_data()
+    qrcode_url = qrcode_data["url"]
+    auth_code = qrcode_data["auth_code"]
+    img = login.make_qrcode(qrcode_url)
+    return (Picture.from_file(img), auth_code)
+
+
+def check_tv_qrcode_events(auth_code: str) -> Tuple[QrCodeLoginEvents, Union[str, Credential]]:
+    """
+    检查登录状态。
+
+    Args:
+        auth_code (str): 登录密钥
+
+    Returns:
+        Tuple[QrCodeLoginEvents, str|Credential]: 状态(第一项）和信息（第二项）（如果成功登录信息为凭据类）
+    """
+    events = login.verify_tv_login_status(auth_code=auth_code)
+
+    if events["code"] == 86039:
+        return QrCodeLoginEvents.SCAN, events["message"]
+    elif events["code"] == 86038:
+        return QrCodeLoginEvents.TIMEOUT, events["message"]
+    elif events["code"] == 0:
+        c = login.parse_tv_resp(events["data"])
+        return QrCodeLoginEvents.DONE, c
+    else:
+        raise LoginError(events["message"])
+
+
 def start_geetest_server() -> "ServerThreadModel":
     """
     验证码服务打开服务器

bilibili_api/utils/network.py

@@ -29,6 +29,7 @@
 
 __httpx_session_pool: Dict[asyncio.AbstractEventLoop, httpx.AsyncClient] = {}
 __aiohttp_session_pool: Dict[asyncio.AbstractEventLoop, aiohttp.ClientSession] = {}
+__httpx_sync_session: httpx.Client = None
 last_proxy = ""
 wbi_mixin_key = ""
 buvid3 = ""
@@ -513,7 +514,7 @@ def request_sync(self, raw: bool = False, **kwargs) -> Union[int, str, dict]:
         """
         self._prepare_params_data()
         config = self._prepare_request_sync(**kwargs)
-        session = httpx.Client()
+        session = get_httpx_sync_session()
         resp = session.request(**config)
         try:
             resp.raise_for_status()
@@ -792,6 +793,39 @@ async def get_bili_ticket() -> str:
     )["ticket"]
 
 
+def get_httpx_sync_session() -> httpx.Client:
+    """
+    获取当前模块的 httpx.Client 对象，用于自定义请求
+
+    Returns:
+        httpx.Client
+    """
+    global __httpx_sync_session
+
+    if __httpx_sync_session is None or last_proxy != settings.proxy:
+        if settings.proxy != "":
+            last_proxy = settings.proxy
+            proxies = {"all://": settings.proxy}
+            session = httpx.AsyncClient(proxies=proxies)  # type: ignore
+        else:
+            last_proxy = ""
+            session = httpx.AsyncClient()
+        __httpx_sync_session = session
+
+    return __httpx_sync_session
+
+
+def set_httpx_sync_session(session: httpx.Client) -> None:
+    """
+    用户手动设置 Session
+
+    Args:
+        session (httpx.Client):  httpx.Client 实例。
+    """
+    global __httpx_sync_session
+    __httpx_sync_session = session
+
+
 def get_session() -> httpx.AsyncClient:
     """
     获取当前模块的 httpx.AsyncClient 对象，用于自定义请求

docs/modules/bilibili_api.md

@@ -18,7 +18,7 @@ import bilibili_api
 
 | name    | type                  | description                |
 | ------- | --------------------- | -------------------------- |
-| session | httpx.AsyncSession | httpx.AsyncSession 实例 |
+| session | httpx.AsyncClient | httpx.AsyncClient 实例 |
 
 用户手动设置 Session
 
@@ -28,9 +28,9 @@ import bilibili_api
 
 ## def get_session()
 
-获取当前模块的 httpx.AsyncSession 对象，用于自定义请求
+获取当前模块的 httpx.AsyncClient 对象，用于自定义请求
 
-**Returns:** httpx.AsyncSession
+**Returns:** httpx.AsyncClient
 
 ---
 
@@ -54,6 +54,26 @@ import bilibili_api
 
 ---
 
+## def set_httpx_sync_session()
+
+| name    | type                  | description                |
+| ------- | --------------------- | -------------------------- |
+| session | httpx.Client | httpx.Client 实例 |
+
+用户手动设置 Session
+
+**Returns:** None
+
+---
+
+## def get_httpx_sync_session()
+
+获取当前模块的 httpx.Client 对象，用于自定义请求
+
+**Returns:** httpx.Client
+
+---
+
 ## class Credential
 
 凭据类，用于各种请求操作的验证。

docs/modules/login_func.md

@@ -36,6 +36,7 @@ from bilibili_api import login_func
 
 + SCAN: 未扫描二维码
 + CONF: 未确认登录
++ TIMEOUT: 二维码过期
 + DONE: 成功
 
 ### def get_qrcode()
@@ -54,6 +55,22 @@ from bilibili_api import login_func
 
 **returns:** Tuple[QrCodeLoginEvents, str|Credential]: 状态(第一项）和信息（第二项）（如果成功登录信息为凭据类）
 
+### def get_tv_qrcode()
+
+获取二维码及登录密钥（后面有用）
+
+**returns:** Tuple[Picture, str]: 第一项是二维码图片地址（本地缓存）和登录密钥。登录密钥需要保存。
+
+### def check_tv_qrcode_events()
+
+| name | type | description |
+| - | - | - |
+| auth_code | str | 登录密钥（get_qrcode 的返回值第二项) |
+
+检查登录状态。
+
+**returns:** Tuple[QrCodeLoginEvents, str|Credential]: 状态(第一项）和信息（第二项）（如果成功登录信息为凭据类）
+
 ---
 
 ## 2. 极验验证码