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

applications/match_algorithm/rank.py

@@ -89,41 +89,46 @@ async def get_content_oss_fission_dict(db_client, config, content_id) -> dict[st
         return {}
 
 
-async def get_title_oss_fission_dict(db_client, config, content_id) -> dict[str: float]:
+async def get_title_oss_fission_list(db_client, config, content_id) -> list[dict]:
     """
     通过 content_id 对应的 oss 路径对应的裂变表现进行排序
     oss 数据每天凌晨 2 点更新
+    todo: 获取有数据的最新dt
     :return:
     """
     FISSION_DETAIL_TABLE = config.fission_detail_table
     LONG_ARTICLES_TEXT_TABLE = config.article_text_table
+    LONG_ARTICLES_CRAWLER_TABLE = config.article_crawler_video_table
     two_days_ago_dt = (datetime.now() - timedelta(days=2)).strftime('%Y%m%d')
     sql = f"""
         SELECT
-            oss_name, fission_rate_0, fission_0_on_read, dt
+            lavfi.oss_name, lavfi.fission_0_on_read, lacv.platform, lacv.cover_oss_path, lacv.user_id, max(lavfi.dt) as max_dt
         FROM
-            {FISSION_DETAIL_TABLE}
+            {FISSION_DETAIL_TABLE} lavfi
+        JOIN {LONG_ARTICLES_CRAWLER_TABLE} lacv on lavfi.oss_name = lacv.video_oss_path
         WHERE title = (
-            SELECT article_title 
+            SELECT article_title
             FROM {LONG_ARTICLES_TEXT_TABLE}
             WHERE content_id = '{content_id}'
-            ) 
-        AND dt >= {two_days_ago_dt}
-        ORDER BY dt DESC;
+            );
+        AND lavfi.dt = (
+            SELECT MAX(dt)
+            FROM long_articles_videos_fission_info
+            WHERE oss_name = lavfi.oss_name
+            )
     """
     result = await db_client.async_select(sql)
-    fission_info_dict = {}
+    fission_info_list = []
     if result:
         for item in result:
-            key = item[0]
-            value = {
-                "fission_rate_0": item[1],
-                "fission_0_on_read": item[2]
+            obj = {
+                "platform": item[2],
+                "video_oss_path": item[0],
+                "cover_oss_path": item[3],
+                "uid": item[4],
+                "fission_0_on_read": item[1],
             }
-            if fission_info_dict.get(key):
-                continue
-            else:
-                fission_info_dict[key] = value
-        return fission_info_dict
+            fission_info_list.append(obj)
+        return sorted(fission_info_list, key=lambda x: x['fission_0_on_read'], reverse=True)
     else:
-        return {}
+        return []

tasks/history_task.py

@@ -9,7 +9,7 @@ from applications.config import Config
 from applications.log import logging
 from applications.functions.pqFunctions import publish_to_pq, get_pq_video_detail
 from applications.functions.common import shuffle_list
-from applications.match_algorithm.rank import get_title_oss_fission_dict
+from applications.match_algorithm.rank import get_title_oss_fission_list
 
 
 class historyContentIdTask(object):
@@ -163,24 +163,22 @@ class historyContentIdTask(object):
                 video_list = []
             case "autoArticlePoolLevel1":
                 # 头条内容，使用重排后结果
-                fission_dict = await get_title_oss_fission_dict(
+                fission_resort_list = await get_title_oss_fission_list(
                     db_client=self.mysql_client,
                     config=self.config,
                     content_id=content_id
                 )
-                if fission_dict:
-                    fission_list = [[i] + [fission_dict[i]['fission_0_on_read']] for i in fission_dict.keys()]
-                    sorted_fission_list = sorted(fission_list, key=lambda x: x[1], reverse=True)
-                    download_videos_with_fission_info = []
-                    for index, video in enumerate(download_videos[:3]):
-                        # 使用fission_list中的oss路径替换， 若替换失败则使用原来的视频
-                        try:
-                            video['video_oss_path'] = sorted_fission_list[index][0]
-                            video["fission_0_on_read"] = sorted_fission_list[index][1]
-                        except IndexError:
-                            continue
-                        download_videos_with_fission_info.append(video)
-                    video_list = download_videos_with_fission_info
+                if fission_resort_list:
+                    total_video_list = fission_resort_list + download_videos
+                    # for index, video in enumerate(download_videos[:3]):
+                    #     # 使用fission_list中的oss路径替换， 若替换失败则使用原来的视频
+                    #     try:
+                    #         video['video_oss_path'] = sorted_fission_list[index][0]
+                    #         video["fission_0_on_read"] = sorted_fission_list[index][1]
+                    #     except IndexError:
+                    #         continue
+                    #     download_videos_with_fission_info.append(video)
+                    video_list = total_video_list[:3]
                 else:
                     # 未找到裂变信息，采用原来的顺序
                     video_list = download_videos[:3]