异步化

applications/match_alg/rank.py

@@ -29,12 +29,12 @@ def jac_score(d1, d2):
     return score_1 * 0.4 + score_2 * 0.4 + score_3 * 0.2, d2['video_id']
 
 
-def best_choice(params_obj, request_param, trace_id):
+async def best_choice(params_obj, request_param, trace_id):
     """
     计算，返回出最合适的 video_id
     :return: video_id
     """
-    pq_list, search_list = recall_videos(params=request_param, trace_id=trace_id)
+    pq_list, search_list = await recall_videos(params=request_param, trace_id=trace_id)
 
     def best_video_id(target_list):
         """

applications/match_alg/recall.py

@@ -4,8 +4,10 @@
 import os
 import json
 import time
+import asyncio
 from concurrent.futures import ThreadPoolExecutor
 
+import httpx
 import requests
 
 from applications.log import logging
@@ -343,7 +345,7 @@ def ask_kimi_and_save_to_local(info_tuple):
             f.write(json.dumps(result, ensure_ascii=False))
 
 
-def recall_videos(params, trace_id):
+async def recall_videos(params, trace_id):
     """
     通过请求的数据来召回视频
     :param trace_id:
@@ -365,7 +367,7 @@ def recall_videos(params, trace_id):
     # url = "http://61.48.133.26:8111/search_videos"
     # requests.post(url, json=payload)
     # # print("请求完成")
-    time.sleep(15)
+    await asyncio.sleep(15)
     # select_sql = "select video_id, video_title from crawler_video where platform='weixin_search' and user_id = '{}' order by update_time DESC limit 10".format(
     #     user_id)
     # out_video_list = select(sql=select_sql)

applications/process.py

@@ -68,7 +68,7 @@ class ProcessParams(object):
             with open(save_path, "w", encoding="utf-8") as f:
                 f.write(json.dumps(result, ensure_ascii=False))
 
-    def deal(self, data):
+    async def deal(self, data):
         """执行代码"""
         params = self.get_params(data)
         title = params['title']
@@ -89,7 +89,7 @@ class ProcessParams(object):
         with open(title_p, encoding="utf-8") as f:
             params_obj = json.loads(f.read())
 
-        best_video_id = best_choice(
+        best_video_id = await best_choice(
             params_obj=params_obj,
             trace_id=self.trace_id,
             request_param=params

applications/routes.py

@@ -42,7 +42,7 @@ async def post_data():
     )
     p = ProcessParams(t_id=trace_id)
     data = await request.get_json()
-    processed_data = p.deal(data)
+    processed_data = await p.deal(data)
     return jsonify(processed_data)