diff --git a/service/xhs/logic/common.py b/service/xhs/logic/common.py index 0b71569..d887298 100644 --- a/service/xhs/logic/common.py +++ b/service/xhs/logic/common.py @@ -1,3 +1,4 @@ +from typing import Optional from lib.logger import logger from lib import requests import execjs @@ -25,8 +26,13 @@ "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36", } +# 在程序启动时编译 JavaScript 代码 +with open('lib/js/xhs.js', encoding='utf-8') as f: + xhs_sign_obj = execjs.compile(f.read()) -async def common_request(uri: str, params: dict, headers: dict, need_sign: bool = True, post: bool = True) -> tuple[dict, bool]: + +async def common_request(uri: str, params: dict, headers: dict, need_sign: bool = True, post: bool = True) -> tuple[ + dict, bool]: """ 请求 xhs :param uri: 请求路径 @@ -36,13 +42,8 @@ async def common_request(uri: str, params: dict, headers: dict, need_sign: bool """ url = f'{API_HOST}{uri}' headers.update(COMMON_HEADERS) - if post: - if need_sign: - xhs_sign_obj = execjs.compile(open('lib/js/xhs.js', encoding='utf-8').read()) - sign_header = xhs_sign_obj.call('sign', uri, params, headers.get('cookie', '')) - headers.update(sign_header) - + sign_request(uri, params, headers, need_sign) logger.info(f'url: {url}, request {url}, params={params}, headers={headers}') body = json.dumps(params, separators=(',', ':'), ensure_ascii=False) response = await requests.post(url, data=body, headers=headers) @@ -53,11 +54,7 @@ async def common_request(uri: str, params: dict, headers: dict, need_sign: bool uri = f'{uri}?{params_str}' url = f'{url}?{params_str}' - if need_sign: - xhs_sign_obj = execjs.compile(open('lib/js/xhs.js', encoding='utf-8').read()) - sign_header = xhs_sign_obj.call('sign', uri, None, headers.get('cookie', '')) - headers.update(sign_header) - + sign_request(uri, None, headers, need_sign) logger.info(f'url: {url}, request {url}, params={params}, headers={headers}') response = await requests.get(url, headers=headers) @@ -68,9 +65,24 @@ async def common_request(uri: str, params: dict, headers: dict, need_sign: bool logger.error( f'url: {url}, params: {params}, request error, code: {response.status_code}, body: {response.text}') return {}, False + if response.json().get('code', 0) != 0: logger.error( f'url: {url}, params: {params}, request error, code: {response.status_code}, body: {response.text}') return response.json(), False - return response.json(), True \ No newline at end of file + return response.json(), True + + +def sign_request(uri: str, params: Optional[dict], headers: dict, need_sign: bool) -> None: + """ + 为请求添加签名 + :param uri: + :param params: + :param headers: + :param need_sign: + :return: + """ + if need_sign: + sign_header = xhs_sign_obj.call('sign', uri, params, headers.get('cookie', '')) + headers.update(sign_header)