✨ 为 bilibili 初步适配Cookie功能

nonebot_bison/platform/bilibili/platforms.py

@@ -16,7 +16,7 @@ from nonebot_bison.compat import model_rebuild
 from nonebot_bison.utils import text_similarity, decode_unicode_escapes
 from nonebot_bison.types import Tag, Target, RawPost, ApiError, Category
 
-from .retry import ApiCode352Error, retry_for_352
+from .retry import ApiCode352Error
 from .scheduler import BilibiliSite, BililiveSite, BiliBangumiSite
 from ..platform import NewMessage, StatusChange, CategoryNotSupport, CategoryNotRecognize
 from .models import (
@@ -68,6 +68,21 @@ class Bilibili(NewMessage):
     has_target = True
     parse_target_promot = "请输入用户主页的链接"
 
+    _HEADERS = {
+        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:132.0) Gecko/20100101 Firefox/132.0",
+        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
+        "Accept-Language": "zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2",
+        "Connection": "keep-alive",
+        "Upgrade-Insecure-Requests": "1",
+        "Sec-Fetch-Dest": "document",
+        "Sec-Fetch-Mode": "navigate",
+        "Sec-Fetch-Site": "none",
+        "Sec-Fetch-User": "?1",
+        "Priority": "u=0, i",
+        "Pragma": "no-cache",
+        "Cache-Control": "no-cache",
+    }
+
     @classmethod
     async def get_target_name(cls, client: AsyncClient, target: Target) -> str | None:
         res = await client.get("https://api.bilibili.com/x/web-interface/card", params={"mid": target})
@@ -90,13 +105,14 @@ class Bilibili(NewMessage):
                 prompt="正确格式:\n1. 用户纯数字id\n2. UID:<用户id>\n3. 用户主页链接: https://space.bilibili.com/xxxx"
             )
 
-    @retry_for_352
+    # @retry_for_352
     async def get_sub_list(self, target: Target) -> list[DynRawPost]:
         client = await self.ctx.get_client(target)
         params = {"host_mid": target, "timezone_offset": -480, "offset": ""}
         res = await client.get(
             "https://api.bilibili.com/x/polymer/web-dynamic/v1/feed/space",
             params=params,
+            headers=self._HEADERS,
             timeout=4.0,
         )
         res.raise_for_status()

nonebot_bison/platform/bilibili/scheduler.py

@@ -1,13 +1,17 @@
+import json
 import random
 from typing_extensions import override
+from datetime import datetime, timedelta
 from typing import TYPE_CHECKING, TypeVar
 
 from httpx import AsyncClient
 from nonebot import logger, require
 from playwright.async_api import Cookie
 
-from nonebot_bison.types import Target
-from nonebot_bison.utils import Site, ClientManager, http_client
+from nonebot_bison.utils import Site, http_client
+
+from ...config.db_model import Cookie as CookieModel
+from ...utils.site import CookieSite, CookieClientManager
 
 if TYPE_CHECKING:
     from .platforms import Bilibili
@@ -18,13 +22,17 @@ from nonebot_plugin_htmlrender import get_browser
 B = TypeVar("B", bound="Bilibili")
 
 
-class BilibiliClientManager(ClientManager):
+class BilibiliClientManager(CookieClientManager):
     _client: AsyncClient
     _inited: bool = False
 
+    _site_name: str = "bilibili.com"
+    _default_cd: int = timedelta(seconds=120)
+
     def __init__(self) -> None:
         self._client = http_client()
 
+    @classmethod
     async def _get_cookies(self) -> list[Cookie]:
         browser = await get_browser()
         async with await browser.new_page() as page:
@@ -43,19 +51,34 @@ class BilibiliClientManager(ClientManager):
                 path=cookie.get("path", "/"),
             )
 
+    @classmethod
+    def _gen_json_cookie(self, cookies: list[Cookie]):
+        cookie_dict = {}
+        for cookie in cookies:
+            cookie_dict[cookie.get("name", "")] = cookie.get("value", "")
+        return cookie_dict
+
+    @classmethod
     @override
-    async def refresh_client(self):
-        cookies = await self._get_cookies()
-        await self._reset_client_cookies(cookies)
-        logger.debug("刷新B站客户端的cookie")
+    async def _generate_anonymous_cookie(cls) -> CookieModel:
+        cookies = await cls._get_cookies()
+        cookie = CookieModel(
+            cookie_name=f"{cls._site_name} anonymous",
+            site_name=cls._site_name,
+            content=json.dumps(cls._gen_json_cookie(cookies)),
+            is_universal=True,
+            is_anonymous=True,
+            last_usage=datetime.now(),
+            cd_milliseconds=0,
+            tags="{}",
+            status="",
+        )
+        return cookie
 
     @override
-    async def get_client(self, target: Target | None) -> AsyncClient:
-        if not self._inited:
-            logger.debug("初始化B站客户端")
-            await self.refresh_client()
-            self._inited = True
-        return self._client
+    async def refresh_client(self):
+        self.refresh_anonymous_cookie()
+        logger.debug("刷新B站客户端的cookie")
 
     @override
     async def get_client_for_static(self) -> AsyncClient:
@@ -66,7 +89,7 @@ class BilibiliClientManager(ClientManager):
         return http_client()
 
 
-class BilibiliSite(Site):
+class BilibiliSite(CookieSite):
     name = "bilibili.com"
     schedule_setting = {"seconds": 60}
     schedule_type = "interval"