feat: AsyncRequest & 日志逻辑改进

common/Httpx.py

@@ -7,7 +7,7 @@
 # ----------------------------------------
 # This file is part of the "lx-music-api-server" project.
 
-# import aiohttp
+import aiohttp
 # import asyncio
 import requests
 import random
@@ -193,7 +193,7 @@ def checkcn():
         logger.warning('检查服务器位置失败，已忽略')
         logger.warning(traceback.format_exc())
 
-async def asyncrequest(url, options = {}):
+async def AsyncRequest(url, options = {}):
     '''
     Http请求主函数, 用于发送网络请求
     - url: 需要请求的URL地址(必填)
@@ -209,6 +209,8 @@ async def asyncrequest(url, options = {}):
     
     @ return: requests.Response类型的响应数据
     '''
+    if (not variable.aioSession):
+        variable.aioSession = aiohttp.ClientSession()
     # 缓存读取
     cache_key = f'{url}{options}'
     if (isinstance(options.get('cache-ignore'), list)):
@@ -217,7 +219,7 @@ async def asyncrequest(url, options = {}):
         options.pop('cache-ignore')
     cache_key = utils.createMD5(cache_key)
     if options.get("cache") and options["cache"] != "no-cache":
-        cache = config.getCache("httpx", cache_key)
+        cache = config.getCache("httpx_async", cache_key)
         if cache:
             logger.debug(f"请求 {url} 有可用缓存")
             return pickle.loads(utils.createBase64Decode(cache["data"]))
@@ -247,3 +249,58 @@ async def asyncrequest(url, options = {}):
     # 检查是否在国内
     if ((not variable.iscn) and (not options["headers"].get("X-Forwarded-For"))):
         options["headers"]["X-Forwarded-For"] = variable.fakeip
+    # 获取请求主函数
+    try:
+        reqattr = getattr(variable.aioSession, method.lower())
+    except AttributeError:
+        raise AttributeError('Unsupported method: '+method)
+    # 请求前记录
+    logger.debug(f'HTTP Request: {url}\noptions: {options}')
+    # 转换body/form参数为原生的data参数，并为form请求追加Content-Type头
+    if (method == 'POST') or (method == 'PUT'):
+        if options.get('body'):
+            options['data'] = options['body']
+            options.pop('body')
+        if options.get('form'):
+            options['data'] = convert_dict_to_form_string(options['form'])
+            options.pop('form')
+            options['headers']['Content-Type'] = 'application/x-www-form-urlencoded'
+        if (isinstance(options['data'], dict)):
+            options['data'] = json.dumps(options['data'])
+    # 进行请求
+    try:
+        logger.info("-----start----- " + url)
+        req = await reqattr(url, **options)
+    except Exception as e:
+        logger.error(f'HTTP Request runs into an Error: {log.highlight_error(traceback.format_exc())}')
+        raise e
+    # 请求后记录
+    logger.debug(f'Request to {url} succeed with code {req.status}')
+    # 为懒人提供的不用改代码移植的方法
+    # 才不是梓澄呢
+    setattr(req, "status_code", req.status)
+    if (req.content.startswith(b'\x78\x9c') or req.content.startswith(b'\x78\x01')): # zlib headers
+        try:
+            decompressed = zlib.decompress(req.content)
+            if (is_valid_utf8(decompressed)):
+                logger.debug(log_plaintext(decompressed.decode("utf-8")))
+            else:
+                logger.debug('response is not text binary, ignore logging it')
+        except:
+            logger.debug('response is not text binary, ignore logging it')
+    else:
+        if (is_valid_utf8(req.content)):
+            logger.debug(log_plaintext(req.content.decode("utf-8")))
+        else:
+            logger.debug('response is not text binary, ignore logging it')
+    # 缓存写入
+    if (cache_info and cache_info != "no-cache"):
+        cache_data = pickle.dumps(req)
+        expire_time = (cache_info if isinstance(cache_info, int) else 3600) + int(time.time())
+        config.updateCache("httpx_async", cache_key, {"expire": True, "time": expire_time, "data": utils.createBase64Encode(cache_data)})
+        logger.debug("缓存已更新: " + url)
+    async def _json():
+        return json.loads(req.content)
+    setattr(req, 'json', _json)
+    # 返回请求
+    return req

common/variable.py

@@ -45,3 +45,4 @@ workdir = os.getcwd()
 banList_suggest = 0
 iscn = True
 fake_ip = None
+aioSession = None

main.py

@@ -9,16 +9,19 @@
 # ----------------------------------------
 # This file is part of the "lx-music-api-server" project.
 
-from aiohttp import web
 from common import config
 from common import lxsecurity
 from common import log
 from common import Httpx
+from common import variable
 from aiohttp.web import Response
 import ujson as json
 import threading
 import traceback
 import modules
+import asyncio
+import aiohttp
+import signal
 import time
 
 def handleResult(dic, status = 200):
@@ -27,7 +30,8 @@ def handleResult(dic, status = 200):
 logger = log.log("main")
 aiologger = log.log('aiohttp_web')
 
-threading.Thread(target=Httpx.checkcn).start()
+def start_checkcn_thread():
+    threading.Thread(target=Httpx.checkcn).start()
 
 # check request info before start
 async def handle_before_request(app, handler):
@@ -99,7 +103,7 @@ async def handle(request):
 async def handle_404(request):
     return handleResult({'code': 6, 'msg': '未找到您所请求的资源', 'data': None}, 404)
 
-app = web.Application(middlewares=[handle_before_request])
+app = aiohttp.web.Application(middlewares=[handle_before_request])
 # mainpage
 app.router.add_get('/', main)
 
@@ -110,10 +114,37 @@ app.router.add_get('/{method}/{source}/{songId}', handle)
 # 404
 app.router.add_route('*', '/{tail:.*}', handle_404)
 
-if (__name__ == "__main__"):
-    try:
-        web.run_app(app, host=config.read_config('common.host'), port=int(config.read_config('common.port')))
-    except Exception as e:
-        logger.error("服务器启动失败, 请查看下方日志")
-        logger.error(traceback.format_exc())
 
+async def run_app():
+    host = config.read_config('common.host')
+    port = int(config.read_config('common.port'))
+    runner = aiohttp.web.AppRunner(app)
+    await runner.setup()
+    site = aiohttp.web.TCPSite(runner, host, port)
+    await site.start()
+    logger.info(f"监听 -> http://{host}:{port}")
+
+async def initMain():
+    variable.aioSession = aiohttp.ClientSession()
+    try:
+        await run_app()
+        logger.info("服务器启动成功，请按下Ctrl + C停止")
+        await asyncio.Event().wait()  # 等待停止事件
+    except (KeyboardInterrupt, asyncio.exceptions.CancelledError):
+        pass
+    except OSError as e:
+        if str(e).startswith("[Errno 98]"):
+            logger.error("端口已被占用，请检查\n" + str(e))
+        else:
+            logger.error("遇到未知错误，请查看日志")
+            logger.error(traceback.format_exc())
+    except:
+        logger.error("遇到未知错误，请查看日志")
+        logger.error(traceback.format_exc())
+    finally:
+        await variable.aioSession.close()
+        logger.info("Server stopped")
+
+if __name__ == "__main__":
+    start_checkcn_thread()
+    asyncio.run(initMain())