更换B站获取用户信息的Api，降低风控概率

omega_miya/web_resource/bilibili/bilibili.py

@@ -16,6 +16,7 @@
 from omega_miya.web_resource import HttpFetcher
 
 from .config import bilibili_config, bilibili_resource_config
+from .wbi import transform_params
 from .exception import BilibiliApiError, BilibiliNetworkError
 from .model.search import BaseBilibiliSearchingModel, UserSearchingModel
 from .model import (BilibiliUserModel, BilibiliUserDynamicModel, BilibiliDynamicModel,
@@ -126,7 +127,7 @@ async def _global_search(
 
 
 class BilibiliUser(Bilibili):
-    _data_api_url = 'https://api.bilibili.com/x/space/acc/info'
+    _data_api_url = 'https://api.bilibili.com/x/space/wbi/acc/info'
     _dynamic_api_url = 'https://api.vc.bilibili.com/dynamic_svr/v1/dynamic_svr/space_history'
 
     def __init__(self, uid: int):
@@ -164,7 +165,8 @@ async def get_user_model(self) -> BilibiliUserModel:
         """获取并初始化用户对应 BilibiliUserModel"""
         if not isinstance(self.user_model, BilibiliUserModel):
             params = {'mid': self.mid}
-            user_result = await self._fetcher.get_json_dict(url=self._data_api_url, params=params)
+            signed_params = await transform_params(params)
+            user_result = await self._fetcher.get_json_dict(url=self._data_api_url, params=signed_params)
             if user_result.status != 200:
                 raise BilibiliApiError(f'BilibiliApiError, {user_result.result}')
             self.user_model = BilibiliUserModel.parse_obj(user_result.result)

omega_miya/web_resource/bilibili/wbi.py

@@ -0,0 +1,68 @@
+from functools import reduce
+from hashlib import md5
+import urllib.parse
+import time
+
+from omega_miya.web_resource import HttpFetcher
+
+from .exception import BilibiliApiError
+
+
+mixinKeyEncTab = [
+    46, 47, 18, 2, 53, 8, 23, 32, 15, 50, 10, 31, 58, 3, 45, 35, 27, 43, 5, 49,
+    33, 9, 42, 19, 29, 28, 14, 39, 12, 38, 41, 13, 37, 48, 7, 16, 24, 55, 40,
+    61, 26, 17, 0, 1, 60, 51, 30, 4, 22, 25, 54, 21, 56, 59, 6, 63, 57, 62, 11,
+    36, 20, 34, 44, 52
+]
+
+
+def get_mixin_key(orig: str):
+    """对 imgKey 和 subKey 进行字符顺序打乱编码"""
+    return reduce(lambda s, i: s + orig[i], mixinKeyEncTab, '')[:32]
+
+
+def enc_wbi(params: dict, img_key: str, sub_key: str):
+    """为请求参数进行 wbi 签名"""
+    mixin_key = get_mixin_key(img_key + sub_key)
+    curr_time = round(time.time())
+    params['wts'] = curr_time                                   # 添加 wts 字段
+    params = dict(sorted(params.items()))                       # 按照 key 重排参数
+    # 过滤 value 中的 "!'()*" 字符
+    params = {
+        k: ''.join(filter(lambda x: x not in "!'()*", str(v)))
+        for k, v
+        in params.items()
+    }
+    query = urllib.parse.urlencode(params)                      # 序列化参数
+    wbi_sign = md5((query + mixin_key).encode()).hexdigest()    # 计算 w_rid
+    params['w_rid'] = wbi_sign
+    return params
+
+
+async def get_wbi_keys() -> tuple[str, str]:
+    """获取最新的 img_key 和 sub_key"""
+    resp = await HttpFetcher().get_json_dict('https://api.bilibili.com/x/web-interface/nav')
+    if resp.status != 200:
+        raise BilibiliApiError(f'BilibiliApiError, {resp.result}')
+
+    json_content = resp.result
+    img_url: str = json_content['data']['wbi_img']['img_url']
+    sub_url: str = json_content['data']['wbi_img']['sub_url']
+    img_key = img_url.rsplit('/', 1)[1].split('.')[0]
+    sub_key = sub_url.rsplit('/', 1)[1].split('.')[0]
+    return img_key, sub_key
+
+
+async def transform_params(params: dict):
+    img_key, sub_key = await get_wbi_keys()
+    signed_params = enc_wbi(
+        params=params,
+        img_key=img_key,
+        sub_key=sub_key
+    )
+    return signed_params
+
+
+__all__ = [
+    'transform_params'
+]