2024-10-23 基于裂变 0 层数据的重排序

applications/match_algorithm/rank.py

@@ -55,28 +55,37 @@ def title_similarity_rank(content_title, recall_list):
     return sorted_list
 
 
-def rank_by_fission_info(db_client, config, content_id):
+async def get_content_oss_fission_dict(db_client, config, content_id) -> dict[str: float]:
     """
     通过 content_id 对应的 oss 路径对应的裂变表现进行排序
     oss 数据每天凌晨 2 点更新
     :return:
     """
     FISSION_DETAIL_TABLE = config.fission_detail_table
-    yesterday_str = (datetime.now() - timedelta(days=1)).strftime('%Y%m%d')
+    two_days_ago_dt = (datetime.now() - timedelta(days=2)).strftime('%Y%m%d')
     sql = f"""
         SELECT 
-            oss_name, fission_0_on_read
+            oss_name, fission_rate_0, fission_0_on_read
         FROM
             {FISSION_DETAIL_TABLE}
-        WHERE content_id = '{content_id}' and dt = '{yesterday_str}'
-        ORDER BY fission_0_on_read DESC;
+        WHERE content_id = '{content_id}' and dt >= '{two_days_ago_dt}'
+        ORDER BY dt DESC;
     """
-    result = db_client.select(sql)
+    result = await db_client.select(sql)
+    fission_info_dict = {}
     if result:
-        oss_name_list = [i[0] for i in result]
-        return oss_name_list
+        for item in result:
+            key = item[0]
+            value = {
+                "fission_rate_0": item[1],
+                "fission_0_on_read": item[2]
+            }
+            if fission_info_dict.get(key):
+                continue
+            else:
+                fission_info_dict[key] = value
     else:
-        return
+        return {}
 
 
 

tasks/history_task.py

@@ -9,6 +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_content_oss_fission_dict
 
 
 class historyContentIdTask(object):
@@ -83,6 +84,11 @@ 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 [
                 {
@@ -91,7 +97,9 @@ class historyContentIdTask(object):
                     "like_count": i[2],
                     "video_oss_path": i[3],
                     "cover_oss_path": i[4],
-                    "uid": i[5]
+                    "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
             ]
@@ -160,8 +168,9 @@ class historyContentIdTask(object):
             case "autoArticlePoolLevel2":
                 video_list = []
             case "autoArticlePoolLevel1":
-                # 头条，先不做
-                video_list = download_videos[:3]
+                # 头条内容，使用重排后结果
+                sorted_videos = sorted(download_videos, key=lambda x: x['fission_0_rate'], reverse=True)
+                video_list = sorted_videos[:3]
             case _:
                 print("未传流量池信息")
                 video_list = download_videos[:3]