只针对level1的通过裂变率重拍

tasks/history_task.py

@@ -84,11 +84,6 @@ class historyContentIdTask(object):
         ORDER BY score DESC;
         """
         res_tuple = await self.mysql_client.async_select(sql)
-        fission_dict = await get_content_oss_fission_dict(
-            db_client=self.mysql_client,
-            config=self.config,
-            content_id=content_id
-        )
         if len(res_tuple) >= 3:
             return [
                 {
@@ -97,9 +92,9 @@ class historyContentIdTask(object):
                     "like_count": i[2],
                     "video_oss_path": i[3],
                     "cover_oss_path": i[4],
-                    "uid": i[5],
-                    "fission_0_rate": fission_dict.get(i[3], {}).get("fission_0_rate", 0),
-                    "fission_0_on_read": fission_dict.get(i[3], {}).get("fission_0_on_read", 0)
+                    "uid": i[5]
+                    # "fission_0_rate": fission_dict.get(i[3], {}).get("fission_0_rate", 0),
+                    # "fission_0_on_read": fission_dict.get(i[3], {}).get("fission_0_on_read", 0)
                 }
                 for i in res_tuple
             ]
@@ -144,9 +139,10 @@ class historyContentIdTask(object):
         )
         return row_counts
 
-    async def publish_videos_to_pq(self, trace_id, flow_pool_level, kimi_title, gh_id, download_videos, process_times):
+    async def publish_videos_to_pq(self, trace_id, flow_pool_level, kimi_title, gh_id, download_videos, process_times, content_id):
         """
         发布至 pq
+        :param content_id:
         :param process_times:
         :param trace_id:
         :param download_videos: 已下载的视频---> list [{}, {}, {}.... ]
@@ -169,7 +165,17 @@ class historyContentIdTask(object):
                 video_list = []
             case "autoArticlePoolLevel1":
                 # 头条内容，使用重排后结果
-                sorted_videos = sorted(download_videos, key=lambda x: x['fission_0_rate'], reverse=True)
+                fission_dict = await get_content_oss_fission_dict(
+                    db_client=self.mysql_client,
+                    config=self.config,
+                    content_id=content_id
+                )
+                download_videos_with_fission_info = []
+                for video in download_videos:
+                    video["fission_0_rate"] = fission_dict.get(video['video_oss_path'], {}).get("fission_0_rate", 0)
+                    video["fission_0_on_read"] = fission_dict.get(video['video_oss_path'], {}).get("fission_0_on_read", 0)
+                    download_videos_with_fission_info.append(video)
+                sorted_videos = sorted(download_videos_with_fission_info, key=lambda x: x['fission_0_rate'], reverse=True)
                 video_list = sorted_videos[:3]
             case _:
                 print("未传流量池信息")
@@ -273,7 +279,8 @@ class historyContentIdTask(object):
                     gh_id=gh_id,
                     trace_id=trace_id,
                     download_videos=download_videos,
-                    process_times=process_times
+                    process_times=process_times,
+                    content_id=content_id
                 )
             except Exception as e:
                 logging(