add test rule_rank1_50

config.py

@@ -72,12 +72,13 @@ class BaseConfig(object):
 
     # 小程序离线ROV模型结果存放 redis key前缀，完整格式：com.weiqu.video.recall.hot.item.score.{date}
     RECALL_KEY_NAME_PREFIX = 'com.weiqu.video.recall.hot.item.score.'
-    # 小程序小时级更新结果存放 redis key前缀，完整格式：com.weiqu.video.recall.item.score.h.{date}.{h}
+    # 小程序小时级更新结果存放 redis key前缀，完整格式：com.weiqu.video.recall.item.score.h.{return_count}.{date}.{h}
     RECALL_KEY_NAME_PREFIX_BY_H = 'com.weiqu.video.recall.item.score.h.'
-    # 小程序离线ROV模型结果与小程序小时级更新结果去重后 存放 redis key前缀，完整格式：com.weiqu.video.recall.hot.item.score.dup.h.{date}.{h}
+    # 小程序离线ROV模型结果与小程序小时级更新结果去重后 存放 redis key前缀，
+    # 完整格式：com.weiqu.video.recall.hot.item.score.dup.h.{return_count}{date}.{h}
     RECALL_KEY_NAME_PREFIX_DUP_H = 'com.weiqu.video.recall.hot.item.score.dup.h.'
-    # 小时级视频状态不符合推荐要求的列表 redis key，完整格式：com.weiqu.video.filter.h.item
-    H_VIDEO_FILER = 'com.weiqu.video.filter.h.item'
+    # 小时级视频状态不符合推荐要求的列表 redis key，完整格式：com.weiqu.video.filter.h.item.{return_count}
+    H_VIDEO_FILER = 'com.weiqu.video.filter.h.item.'
 
     # app应用 小程序离线ROV模型结果存放 redis key前缀，完整格式：com.weiqu.video.recall.hot.item.score.app.{date}
     RECALL_KEY_NAME_PREFIX_APP = 'com.weiqu.video.recall.hot.item.score.app.'

rule_rank_h.py

@@ -94,12 +94,13 @@ def cal_score(df):
     return df
 
 
-def video_rank(df, now_date, now_h):
+def video_rank(df, now_date, now_h, return_count):
     """
     获取符合进入召回源条件的视频，与每日更新的rov模型结果视频列表进行合并
     :param df:
     :param now_date:
     :param now_h:
+    :param return_count: 小时级数据回流限制数
     :return:
     """
     # 获取rov模型结果
@@ -109,7 +110,7 @@ def video_rank(df, now_date, now_h):
     log_.info(f'initial data count = {len(initial_data)}')
 
     # 获取符合进入召回源条件的视频，进入条件：小时级回流>=20 && score>=0.005
-    h_recall_df = df[(df['lastonehour_return'] >= 20) & (df['score'] >= 0.005)]
+    h_recall_df = df[(df['lastonehour_return'] >= return_count) & (df['score'] >= 0.005)]
     h_recall_videos = h_recall_df['videoid'].to_list()
     log_.info(f'h_recall videos count = {len(h_recall_videos)}')
     # 写入对应的redis
@@ -119,10 +120,11 @@ def video_rank(df, now_date, now_h):
         score = h_recall_df[h_recall_df['videoid'] == video_id]['score']
         h_recall_result[int(video_id)] = float(score)
         h_video_ids.append(int(video_id))
-    h_recall_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_H}{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
+    h_recall_key_name = \
+        f"{config_.RECALL_KEY_NAME_PREFIX_BY_H}{return_count}.{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
     redis_helper.add_data_with_zset(key_name=h_recall_key_name, data=h_recall_result, expire_time=24 * 3600)
     # 清空线上过滤应用列表
-    redis_helper.del_keys(key_name=config_.H_VIDEO_FILER)
+    redis_helper.del_keys(key_name=f"{config_.H_VIDEO_FILER}{return_count}")
 
     # 去重更新rov模型结果，并另存为redis中
     initial_data_dup = {}
@@ -130,7 +132,7 @@ def video_rank(df, now_date, now_h):
         if int(video_id) not in h_video_ids:
             initial_data_dup[int(video_id)] = score
     log_.info(f"initial data dup count = {len(initial_data_dup)}")
-    initial_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP_H}{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
+    initial_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP_H}{return_count}.{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
     redis_helper.add_data_with_zset(key_name=initial_key_name, data=initial_data_dup, expire_time=24 * 3600)
 
 
@@ -151,20 +153,23 @@ def video_rank(df, now_date, now_h):
     # redis_helper.add_data_with_zset(key_name=final_key_name, data=final_data, expire_time=24 * 3600)
 
 
-def rank_by_h(now_date, now_h):
+def rank_by_h(now_date, now_h, return_count_list):
     # 获取特征数据
     feature_df = get_feature_data(now_date=now_date)
     # 计算score
     score_df = cal_score(df=feature_df)
     # rank
-    video_rank(df=score_df, now_date=now_date, now_h=now_h)
+    for cnt in return_count_list:
+        log_.info(f"return_count = {cnt}")
+        video_rank(df=score_df, now_date=now_date, now_h=now_h, return_count=cnt)
     # to-csv
     score_filename = f"score_{datetime.datetime.strftime(now_date, '%Y%m%d%H')}.csv"
     score_df.to_csv(f'./data/{score_filename}')
 
 
-def h_rank_bottom(now_date, now_h):
+def h_rank_bottom(now_date, now_h, return_count):
     """未按时更新数据，用上一小时结果作为当前小时的数据"""
+    log_.info(f"return_count = {return_count}")
     # 获取rov模型结果
     redis_helper = RedisHelper()
     if now_h == 0:
@@ -173,35 +178,39 @@ def h_rank_bottom(now_date, now_h):
     else:
         redis_dt = datetime.datetime.strftime(now_date, '%Y%m%d')
         redis_h = now_h - 1
-    key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_H}{redis_dt}.{redis_h}"
+    key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_H}{return_count}.{redis_dt}.{redis_h}"
     initial_data = redis_helper.get_data_zset_with_index(key_name=key_name, start=0, end=-1, with_scores=True)
     final_data = dict()
     for video_id, score in initial_data:
         final_data[video_id] = score
     # 存入对应的redis
-    final_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_H}{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
+    final_key_name = \
+        f"{config_.RECALL_KEY_NAME_PREFIX_BY_H}{return_count}.{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
     redis_helper.add_data_with_zset(key_name=final_key_name, data=final_data, expire_time=24 * 3600)
     # 清空线上过滤应用列表
-    redis_helper.del_keys(key_name=config_.H_VIDEO_FILER)
+    redis_helper.del_keys(key_name=f"{config_.H_VIDEO_FILER}{return_count}")
 
 
 def h_timer_check():
+    return_count_list = [20, 50]
     now_date = datetime.datetime.today()
     log_.info(f"now_date: {datetime.datetime.strftime(now_date, '%Y%m%d%H')}")
     now_h = datetime.datetime.now().hour
     now_min = datetime.datetime.now().minute
     if now_h == 0:
-        h_rank_bottom(now_date=now_date, now_h=now_h)
+        for cnt in return_count_list:
+            h_rank_bottom(now_date=now_date, now_h=now_h, return_count=cnt)
         return
     # 查看当前小时更新的数据是否已准备好
     h_data_count = h_data_check(project=project, table=table, now_date=now_date)
     if h_data_count > 0:
         log_.info(f'h_data_count = {h_data_count}')
         # 数据准备好，进行更新
-        rank_by_h(now_date=now_date, now_h=now_h)
+        rank_by_h(now_date=now_date, now_h=now_h, return_count_list=return_count_list)
     elif now_min > 50:
         log_.info('h_recall data is None, use bottom data!')
-        h_rank_bottom(now_date=now_date, now_h=now_h)
+        for cnt in return_count_list:
+            h_rank_bottom(now_date=now_date, now_h=now_h, return_count=cnt)
     else:
         # 数据没准备好，1分钟后重新检查
         Timer(60, h_timer_check).start()

videos_filter.py

@@ -367,6 +367,7 @@ def filter_app_pool():
 
 def filter_rov_h():
     """过滤小程序小时级数据"""
+    return_count_list = [20, 50]
     log_.info("rov_h pool filter start ...")
     redis_helper = RedisHelper()
     # 获取当前日期
@@ -374,57 +375,60 @@ def filter_rov_h():
     # 获取当前所在小时
     now_h = datetime.now().hour
     log_.info(f'now_date = {now_date}, now_h = {now_h}.')
-    # 需过滤两个视频列表
-    key_prefix_list = [config_.RECALL_KEY_NAME_PREFIX_BY_H, config_.RECALL_KEY_NAME_PREFIX_DUP_H]
-    for i, key_prefix in enumerate(key_prefix_list):
-        # 拼接key
-        key_name = f"{key_prefix}{now_date}.{now_h}"
-        log_.info(f"key_name: {key_name}")
-        # 获取视频
-        data = redis_helper.get_data_zset_with_index(key_name=key_name, start=0, end=-1)
-        if data is None:
-            log_.info("data is None")
-            log_.info("filter end!")
-            continue
-        # 过滤
-        video_ids = [int(video_id) for video_id in data]
-        filtered_result = filter_video_status(video_ids=video_ids)
-        # 求差集，获取需要过滤掉的视频，并从redis中移除
-        filter_videos = set(video_ids) - set(filtered_result)
-        log_.info("video_ids size = {}, filtered size = {}, filter sizer = {}".format(len(video_ids),
-                                                                                      len(filtered_result),
-                                                                                      len(filter_videos)))
-        if len(filter_videos) == 0:
-            log_.info("filter end!")
-            continue
-        redis_helper.remove_value_from_zset(key_name=key_name, value=list(filter_videos))
-        if i == 0:
-            # 将小时级的数据需要过滤的视频加入到线上过滤应用列表中
-            redis_helper.add_data_with_set(key_name=config_.H_VIDEO_FILER, values=filter_videos, expire_time=2*3600)
+    for cnt in return_count_list:
+        log_.info(f"return_count = {cnt}")
+        # 需过滤两个视频列表
+        key_prefix_list = [config_.RECALL_KEY_NAME_PREFIX_BY_H, config_.RECALL_KEY_NAME_PREFIX_DUP_H]
+        for i, key_prefix in enumerate(key_prefix_list):
+            # 拼接key
+            key_name = f"{key_prefix}{cnt}.{now_date}.{now_h}"
+            log_.info(f"key_name: {key_name}")
+            # 获取视频
+            data = redis_helper.get_data_zset_with_index(key_name=key_name, start=0, end=-1)
+            if data is None:
+                log_.info("data is None")
+                log_.info("filter end!")
+                continue
+            # 过滤
+            video_ids = [int(video_id) for video_id in data]
+            filtered_result = filter_video_status(video_ids=video_ids)
+            # 求差集，获取需要过滤掉的视频，并从redis中移除
+            filter_videos = set(video_ids) - set(filtered_result)
+            log_.info("video_ids size = {}, filtered size = {}, filter sizer = {}".format(len(video_ids),
+                                                                                          len(filtered_result),
+                                                                                          len(filter_videos)))
+            if len(filter_videos) == 0:
+                log_.info("filter end!")
+                continue
+            redis_helper.remove_value_from_zset(key_name=key_name, value=list(filter_videos))
+            if i == 0:
+                # 将小时级的数据需要过滤的视频加入到线上过滤应用列表中
+                redis_helper.add_data_with_set(key_name=f"{config_.H_VIDEO_FILER}{cnt}",
+                                               values=filter_videos, expire_time=2*3600)
     log_.info("rov_h pool filter end!")
 
 
 def main():
     try:
-        # ROV召回池视频过滤
-        filter_rov_pool()
-        # appType = 6，ROV召回池视频过滤
-        filter_rov_pool(app_type=config_.APP_TYPE['SHORT_VIDEO'])
-        # appType = 13，票圈视频APP视频过滤
-        filter_rov_pool(app_type=config_.APP_TYPE['APP'])
-        # 流量池视频过滤
-        filter_flow_pool()
-        # 兜底视频过滤
-        filter_bottom()
-        # 修改过ROV的视频过滤
-        filter_rov_updated()
-        filter_rov_updated_app()
-        # 运营强插相关推荐视频过滤
-        filter_relevant_videos()
-        # 按位置排序视频过滤
-        filter_position_videos()
-        # 过滤票圈视频APP小时级数据
-        filter_app_pool()
+        # # ROV召回池视频过滤
+        # filter_rov_pool()
+        # # appType = 6，ROV召回池视频过滤
+        # filter_rov_pool(app_type=config_.APP_TYPE['SHORT_VIDEO'])
+        # # appType = 13，票圈视频APP视频过滤
+        # filter_rov_pool(app_type=config_.APP_TYPE['APP'])
+        # # 流量池视频过滤
+        # filter_flow_pool()
+        # # 兜底视频过滤
+        # filter_bottom()
+        # # 修改过ROV的视频过滤
+        # filter_rov_updated()
+        # filter_rov_updated_app()
+        # # 运营强插相关推荐视频过滤
+        # filter_relevant_videos()
+        # # 按位置排序视频过滤
+        # filter_position_videos()
+        # # 过滤票圈视频APP小时级数据
+        # filter_app_pool()
         # 过滤小程序小时级数据
         filter_rov_h()
     except Exception as e: