From 52ab902be52feff4495f0f1013c5c253c2d2a6d2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=98=9F=E3=83=8E=E8=B0=B7=E7=BB=AB=E6=B2=AB?= <30362789+yokinanya@users.noreply.github.com> Date: Tue, 27 Jun 2023 20:42:29 +0800 Subject: [PATCH] =?UTF-8?q?=E6=9B=B4=E6=8D=A2B=E7=AB=99=E8=8E=B7=E5=8F=96?= =?UTF-8?q?=E7=94=A8=E6=88=B7=E4=BF=A1=E6=81=AF=E7=9A=84Api=EF=BC=8C?= =?UTF-8?q?=E9=99=8D=E4=BD=8E=E9=A3=8E=E6=8E=A7=E6=A6=82=E7=8E=87=20(#125)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 更换获取用户信息的Api，避免风控 * Fix: 调整 wbi 参数获取为异步方法 - 调整 wbi 参数获取为异步方法 --------- Co-authored-by: Ailitonia <41713304+Ailitonia@users.noreply.github.com> --- omega_miya/web_resource/bilibili/bilibili.py | 6 +- omega_miya/web_resource/bilibili/wbi.py | 68 ++++++++++++++++++++ 2 files changed, 72 insertions(+), 2 deletions(-) create mode 100644 omega_miya/web_resource/bilibili/wbi.py diff --git a/omega_miya/web_resource/bilibili/bilibili.py b/omega_miya/web_resource/bilibili/bilibili.py index 936d6c88..0909dd7a 100644 --- a/omega_miya/web_resource/bilibili/bilibili.py +++ b/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) diff --git a/omega_miya/web_resource/bilibili/wbi.py b/omega_miya/web_resource/bilibili/wbi.py new file mode 100644 index 00000000..ce3cd928 --- /dev/null +++ b/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' +]