refactor: 优化xhs签名js编译一次

Author: NanmiCoder

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
+    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)