update

region_rule_rank_h.py

@@ -166,7 +166,7 @@ def cal_score(df, param):
     return df
 
 
-def add_videos(initial_df, now_date, rule_key, region, data_key):
+def add_videos(initial_df, now_date, rule_key, region, data_key, hour_count, top):
     """
     地域小时级数据列表中增加前6h优质视频
     :param initial_df: 地域小时级筛选结果
@@ -174,20 +174,23 @@ def add_videos(initial_df, now_date, rule_key, region, data_key):
     :param data_key:
     :param region:
     :param rule_key:
+    :param hour_count: 前几个小时, type-int
+    :param top: type-int
     :return: df
     """
     redis_helper = RedisHelper()
     pre_h_data = []
-    for pre_h in range(1, 7):
-        pre_date = now_date - datetime.timedelta(hours=pre_h)
-
-
+    for i in range(1, hour_count+1):
+        pre_date = now_date - datetime.timedelta(hours=i)
+        pre_h = pre_date.hour
         pre_h_recall_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H}{region}:{data_key}:{rule_key}:" \
-                                f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{pre_h}"
-        initial_data = redis_helper.get_all_data_from_zset(key_name=pre_h_recall_key_name, with_scores=True)
-        if initial_data is None:
+                                f"{datetime.datetime.strftime(pre_date, '%Y%m%d')}:{pre_h}"
+        pre_h_top_data = redis_helper.get_data_zset_with_index(key_name=pre_h_recall_key_name,
+                                                               start=0, end=top-1,
+                                                               desc=True, with_scores=True)
+        if pre_h_top_data is None:
             continue
-        pre_h_data.extend(initial_data)
+        pre_h_data.extend(pre_h_top_data)
     pre_h_df = pd.DataFrame(data=pre_h_data, columns=['videoid', 'score'])
     score_list = initial_df['score'].to_list()
     if len(score_list) > 0:
@@ -203,7 +206,7 @@ def add_videos(initial_df, now_date, rule_key, region, data_key):
     return df
 
 
-def video_rank(df, now_date, now_h, rule_key, param, region, data_key, rule_rank_h_flag, add_videos_in_20h=False):
+def video_rank(df, now_date, now_h, rule_key, param, region, data_key, rule_rank_h_flag, add_videos_with_pre6h=False):
     """
     获取符合进入召回源条件的视频，与每日更新的rov模型结果视频列表进行合并
     :param df:
@@ -228,13 +231,10 @@ def video_rank(df, now_date, now_h, rule_key, param, region, data_key, rule_rank
     h_recall_df = h_recall_df.drop_duplicates(subset=['videoid'], keep='first')
     h_recall_df['videoid'] = h_recall_df['videoid'].astype(int)
 
-    # 20点增加打捞的优质视频
-    if now_h == 20 and add_videos_in_20h is True:
-        # print(len(h_recall_df))
+    # 增加打捞的优质视频
+    if add_videos_with_pre6h is True:
         h_recall_df = add_videos(initial_df=h_recall_df, now_date=now_date, rule_key=rule_key,
-                                 region=region, data_key=data_key)
-        # print(len(h_recall_df))
-
+                                 region=region, data_key=data_key, hour_count=6, top=10)
 
     h_recall_videos = h_recall_df['videoid'].to_list()
     # log_.info(f'h_recall videos count = {len(h_recall_videos)}')
@@ -421,7 +421,7 @@ def merge_df_with_score(df_left, df_right):
 
 
 def process_with_region(region, df_merged, data_key, rule_key, rule_param, now_date, now_h,
-                        rule_rank_h_flag, add_videos_in_20h):
+                        rule_rank_h_flag, add_videos_with_pre6h):
     log_.info(f"region = {region} start...")
     # 计算score
     region_df = df_merged[df_merged['code'] == region]
@@ -429,18 +429,18 @@ def process_with_region(region, df_merged, data_key, rule_key, rule_param, now_d
     score_df = cal_score(df=region_df, param=rule_param)
     video_rank(df=score_df, now_date=now_date, now_h=now_h, rule_key=rule_key, param=rule_param,
                region=region, data_key=data_key, rule_rank_h_flag=rule_rank_h_flag,
-               add_videos_in_20h=add_videos_in_20h)
+               add_videos_with_pre6h=add_videos_with_pre6h)
     log_.info(f"region = {region} end!")
 
 
 def process_with_region2(region, df_merged, data_key, rule_key, rule_param, now_date, now_h,
-                         rule_rank_h_flag, add_videos_in_20h):
+                         rule_rank_h_flag, add_videos_with_pre6h):
     log_.info(f"region = {region} start...")
     region_score_df = df_merged[df_merged['code'] == region]
     log_.info(f'region = {region}, region_score_df count = {len(region_score_df)}')
     video_rank(df=region_score_df, now_date=now_date, now_h=now_h, region=region,
                rule_key=rule_key, param=rule_param, data_key=data_key, rule_rank_h_flag=rule_rank_h_flag,
-               add_videos_in_20h=add_videos_in_20h)
+               add_videos_with_pre6h=add_videos_with_pre6h)
     log_.info(f"region = {region} end!")
 
 
@@ -536,8 +536,8 @@ def process_with_param(param, data_params_item, rule_params_item, region_code_li
     rule_param = rule_params_item.get(rule_key)
     log_.info(f"rule_key = {rule_key}, rule_param = {rule_param}")
     merge_func = rule_param.get('merge_func', None)
-    # 是否在20点的数据中增加打捞的优质视频
-    add_videos_in_20h = rule_param.get('add_videos_in_20h', False)
+    # 是否在地域小时级数据中增加打捞的优质视频
+    add_videos_with_pre6h = rule_param.get('add_videos_with_pre6h', False)
 
     if merge_func == 2:
         score_df_list = []
@@ -554,7 +554,7 @@ def process_with_param(param, data_params_item, rule_params_item, region_code_li
         task_list = [
             gevent.spawn(process_with_region2,
                          region, df_merged, data_key, rule_key, rule_param, now_date, now_h, rule_rank_h_flag,
-                         add_videos_in_20h)
+                         add_videos_with_pre6h)
             for region in region_code_list
         ]
     else:
@@ -563,7 +563,7 @@ def process_with_param(param, data_params_item, rule_params_item, region_code_li
         task_list = [
             gevent.spawn(process_with_region,
                          region, df_merged, data_key, rule_key, rule_param, now_date, now_h, rule_rank_h_flag,
-                         add_videos_in_20h)
+                         add_videos_with_pre6h)
             for region in region_code_list
         ]
 
@@ -743,9 +743,9 @@ def h_timer_check():
         project = config_.PROJECT_REGION_APP_TYPE
         table = config_.TABLE_REGION_APP_TYPE
         region_code_list = [code for region, code in region_code.items()]
-        now_date = datetime.datetime.today()
+        now_date = datetime.datetime.today() - datetime.timedelta(hours=1)
         log_.info(f"now_date: {datetime.datetime.strftime(now_date, '%Y%m%d%H')}, rule_rank_h_flag: {rule_rank_h_flag}")
-        now_h = datetime.datetime.now().hour
+        now_h = datetime.datetime.now().hour - 1
         now_min = datetime.datetime.now().minute
         if now_h == 0:
             h_rank_bottom(now_date=now_date, now_h=now_h, rule_params=rule_params, region_code_list=region_code_list,