import json import random import numpy from log import Log from config import set_config from video_recall import PoolRecall from db_helper import RedisHelper from utils import FilterVideos, send_msg_to_feishu from rank_service import get_featurs, get_tf_serving_sores log_ = Log() config_ = set_config() def video_rank(data, size, top_K, flow_pool_P): """ 视频分发排序 :param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []} :param size: 请求数 :param top_K: 保证topK为召回池视频 type-int :param flow_pool_P: size-top_K视频为流量池视频的概率 type-float :return: rank_result """ if not data['rov_pool_recall'] and not data['flow_pool_recall']: return [] # 将各路召回的视频按照score从大到小排序 # 最惊奇相关推荐相似视频 # relevant_recall = [item for item in data['rov_pool_recall'] # if item.get('pushFrom') == config_.PUSH_FROM['top_video_relevant_appType_19']] # relevant_recall_rank = sorted(relevant_recall, key=lambda k: k.get('rovScore', 0), reverse=True) # 最惊奇完整影视视频 # whole_movies_recall = [item for item in data['rov_pool_recall'] # if item.get('pushFrom') == config_.PUSH_FROM['whole_movies']] # whole_movies_recall_rank = sorted(whole_movies_recall, key=lambda k: k.get('rovScore', 0), reverse=True) # 最惊奇影视解说视频 # talk_videos_recall = [item for item in data['rov_pool_recall'] # if item.get('pushFrom') == config_.PUSH_FROM['talk_videos']] # talk_videos_recall_rank = sorted(talk_videos_recall, key=lambda k: k.get('rovScore', 0), reverse=True) # 小时级更新数据 # h_recall = [item for item in data['rov_pool_recall'] if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_h']] # h_recall_rank = sorted(h_recall, key=lambda k: k.get('rovScore', 0), reverse=True) # 相对30天天级规则更新数据 day_30_recall = [item for item in data['rov_pool_recall'] if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_30day']] day_30_recall_rank = sorted(day_30_recall, key=lambda k: k.get('rovScore', 0), reverse=True) # 地域分组小时级规则更新数据 region_h_recall = [item for item in data['rov_pool_recall'] if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_region_h']] region_h_recall_rank = sorted(region_h_recall, key=lambda k: k.get('rovScore', 0), reverse=True) # 地域分组小时级更新24h规则更新数据 region_24h_recall = [item for item in data['rov_pool_recall'] if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_region_24h']] region_24h_recall_rank = sorted(region_24h_recall, key=lambda k: k.get('rovScore', 0), reverse=True) # 地域分组天级规则更新数据 # region_day_recall = [item for item in data['rov_pool_recall'] # if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_region_day']] # region_day_recall_rank = sorted(region_day_recall, key=lambda k: k.get('rovScore', 0), reverse=True) # 相对24h规则更新数据 rule_24h_recall = [item for item in data['rov_pool_recall'] if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_24h']] rule_24h_recall_rank = sorted(rule_24h_recall, key=lambda k: k.get('rovScore', 0), reverse=True) # 相对24h规则筛选后剩余更新数据 rule_24h_dup_recall = [item for item in data['rov_pool_recall'] if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_24h_dup']] rule_24h_dup_recall_rank = sorted(rule_24h_dup_recall, key=lambda k: k.get('rovScore', 0), reverse=True) # 相对48h规则更新数据 rule_48h_recall = [item for item in data['rov_pool_recall'] if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_48h']] rule_48h_recall_rank = sorted(rule_48h_recall, key=lambda k: k.get('rovScore', 0), reverse=True) # 相对48h规则筛选后剩余更新数据 rule_48h_dup_recall = [item for item in data['rov_pool_recall'] if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_48h_dup']] rule_48h_dup_recall_rank = sorted(rule_48h_dup_recall, key=lambda k: k.get('rovScore', 0), reverse=True) # 天级规则更新数据 # day_recall = [item for item in data['rov_pool_recall'] if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_day']] # day_recall_rank = sorted(day_recall, key=lambda k: k.get('rovScore', 0), reverse=True) # ROV召回池 # rov_initial_recall = [ # item for item in data['rov_pool_recall'] # if item.get('pushFrom') not in # [config_.PUSH_FROM['top_video_relevant_appType_19'], # config_.PUSH_FROM['rov_recall_h'], # config_.PUSH_FROM['rov_recall_region_h'], # config_.PUSH_FROM['rov_recall_region_24h'], # config_.PUSH_FROM['rov_recall_region_day'], # config_.PUSH_FROM['rov_recall_24h'], # config_.PUSH_FROM['rov_recall_24h_dup'], # config_.PUSH_FROM['rov_recall_48h'], # config_.PUSH_FROM['rov_recall_48h_dup'], # config_.PUSH_FROM['rov_recall_day'], # config_.PUSH_FROM['whole_movies'], # config_.PUSH_FROM['talk_videos']] # ] # rov_initial_recall_rank = sorted(rov_initial_recall, key=lambda k: k.get('rovScore', 0), reverse=True) # rov_recall_rank = whole_movies_recall_rank + talk_videos_recall_rank + h_recall_rank + \ # day_30_recall_rank + region_h_recall_rank + region_24h_recall_rank + \ # region_day_recall_rank + rule_24h_recall_rank + rule_24h_dup_recall_rank + \ # rule_48h_recall_rank + rule_48h_dup_recall_rank + \ # day_recall_rank + rov_initial_recall_rank rov_recall_rank = day_30_recall_rank + \ region_h_recall_rank + region_24h_recall_rank + \ rule_24h_recall_rank + rule_24h_dup_recall_rank + \ rule_48h_recall_rank + rule_48h_dup_recall_rank # 流量池 flow_recall_rank = sorted(data['flow_pool_recall'], key=lambda k: k.get('rovScore', 0), reverse=True) # 对各路召回的视频进行去重 rov_recall_rank, flow_recall_rank = remove_duplicate(rov_recall=rov_recall_rank, flow_recall=flow_recall_rank, top_K=top_K) # log_.info('remove_duplicate finished! rov_recall_rank = {}, flow_recall_rank = {}'.format( # rov_recall_rank, flow_recall_rank)) # rank_result = relevant_recall_rank rank_result = [] # 从ROV召回池中获取top k if len(rov_recall_rank) > 0: rank_result.extend(rov_recall_rank[:top_K]) rov_recall_rank = rov_recall_rank[top_K:] else: rank_result.extend(flow_recall_rank[:top_K]) flow_recall_rank = flow_recall_rank[top_K:] # 按概率 p 及score排序获取 size - k 个视频 i = 0 while i < size - top_K: # 随机生成[0, 1)浮点数 rand = random.random() # log_.info('rand: {}'.format(rand)) if rand < flow_pool_P: if flow_recall_rank: rank_result.append(flow_recall_rank[0]) flow_recall_rank.remove(flow_recall_rank[0]) else: rank_result.extend(rov_recall_rank[:size - top_K - i]) return rank_result[:size] else: if rov_recall_rank: rank_result.append(rov_recall_rank[0]) rov_recall_rank.remove(rov_recall_rank[0]) else: rank_result.extend(flow_recall_rank[:size - top_K - i]) return rank_result[:size] i += 1 return rank_result[:size] def video_new_rank(videoIds, fast_flow_set, flow_set, size, top_K, flow_pool_P): """ 视频分发排序 :param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []} :param size: 请求数 :param top_K: 保证topK为召回池视频 type-int :param flow_pool_P: size-top_K视频为流量池视频的概率 type-float :return: rank_result """ add_flow_set = set('') if not videoIds or len(videoIds)==0: return [], add_flow_set redisObj = RedisHelper() vidKeys = [] for vid in videoIds: vidKeys.append("k_p:"+str(vid)) #print("vidKeys:", vidKeys) video_scores = redisObj.get_batch_key(vidKeys) #print(video_scores) video_items = [] for i in range(len(video_scores)): try: #print(video_scores[i]) if video_scores[i] is None: video_items.append((videoIds[i], 0.0)) else: video_score_str = json.loads(video_scores[i]) #print("video_score_str:",video_score_str) video_items.append((videoIds[i], video_score_str[0])) except Exception: video_items.append((videoIds[i], 0.0)) sort_items = sorted(video_items, key=lambda k: k[1], reverse=True) #print("sort_items:", sort_items) rov_recall_rank = sort_items fast_flow_recall_rank = [] flow_recall_rank = [] for item in sort_items: if item[0] in fast_flow_set: fast_flow_recall_rank.append(item) elif item[0] in flow_set: flow_recall_rank.append(item) # all flow result all_flow_recall_rank = fast_flow_recall_rank+flow_recall_rank rank_result = [] rank_set = set('') # 从ROV召回池中获取top k if len(rov_recall_rank) > 0: rank_result.extend(rov_recall_rank[:top_K]) rov_recall_rank = rov_recall_rank[top_K:] else: rank_result.extend(all_flow_recall_rank[:top_K]) all_flow_recall_rank = all_flow_recall_rank[top_K:] for rank_item in rank_result: rank_set.add(rank_item[0]) #print("rank_result:", rank_result) # 按概率 p 及score排序获取 size - k 个视频， 第4个位置按概率取流量池 i = 0 left_quato = size - top_K j = 0 jj = 0 while i < left_quato and (j= left_quato: break else: for recall_item in rov_recall_rank: jj+=1 if recall_item[0] in rank_set: continue else: rank_result.append(recall_item) rank_set.add(recall_item[0]) i += 1 if i>= left_quato: break #print("rank_result:", rank_result) #print("add_flow_set:", add_flow_set) return rank_result[:size], add_flow_set def refactor_video_rank(rov_recall_rank, fast_flow_set, flow_set, size, top_K, flow_pool_P): """ 视频分发排序 :param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []} :param size: 请求数 :param top_K: 保证topK为召回池视频 type-int :param flow_pool_P: size-top_K视频为流量池视频的概率 type-float :return: rank_result """ if not rov_recall_rank or len(rov_recall_rank) == 0: return [] fast_flow_recall_rank = [] flow_recall_rank = [] for item in rov_recall_rank: vid = item.get('videoId', 0) #print(item) if vid in fast_flow_set: fast_flow_recall_rank.append(item) elif vid in flow_set: flow_recall_rank.append(item) # all flow result all_flow_recall_rank = fast_flow_recall_rank + flow_recall_rank rank_result = [] rank_set = set('') # 从ROV召回池中获取top k if len(rov_recall_rank) > 0: rank_result.extend(rov_recall_rank[:top_K]) rov_recall_rank = rov_recall_rank[top_K:] else: rank_result.extend(all_flow_recall_rank[:top_K]) all_flow_recall_rank = all_flow_recall_rank[top_K:] #已存放了多少VID for rank_item in rank_result: rank_set.add(rank_item.get('videoId', 0)) # 按概率 p 及score排序获取 size - k 个视频， 第4个位置按概率取流量池 i = 0 while i < size - top_K: # 随机生成[0, 1)浮点数 rand = random.random() # log_.info('rand: {}'.format(rand)) if rand < flow_pool_P: for flow_item in all_flow_recall_rank: flow_vid = flow_item.get('videoId', 0) if flow_vid in rank_set: continue else: rank_result.append(flow_item) rank_set.add(flow_vid) else: for recall_item in rov_recall_rank: flow_vid = recall_item.get('videoId', 0) if flow_vid in rank_set: continue else: rank_result.append(recall_item) rank_set.add(flow_vid) i += 1 return rank_result[:size] def remove_duplicate(rov_recall, flow_recall, top_K): """ 对多路召回的视频去重 去重原则： 如果视频在ROV召回池topK，则保留ROV召回池，否则保留流量池 :param rov_recall: ROV召回池-已排序 :param flow_recall: 流量池-已排序 :param top_K: 保证topK为召回池视频 type-int :return: """ flow_recall_result = [] rov_recall_remove = [] flow_recall_video_ids = [item['videoId'] for item in flow_recall] # rov_recall topK for item in rov_recall[:top_K]: if item['videoId'] in flow_recall_video_ids: flow_recall_video_ids.remove(item['videoId']) # other for item in rov_recall[top_K:]: if item['videoId'] in flow_recall_video_ids: rov_recall_remove.append(item) # rov recall remove for item in rov_recall_remove: rov_recall.remove(item) # flow recall remove for item in flow_recall: if item['videoId'] in flow_recall_video_ids: flow_recall_result.append(item) return rov_recall, flow_recall_result def bottom_strategy(request_id, size, app_type, ab_code, params): """ 兜底策略: 从ROV召回池中获取top1000，进行状态过滤后的视频 :param request_id: request_id :param size: 需要获取的视频数 :param app_type: 产品标识 type-int :param ab_code: abCode :param params: :return: """ pool_recall = PoolRecall(request_id=request_id, app_type=app_type, ab_code=ab_code) key_name, _ = pool_recall.get_pool_redis_key(pool_type='rov') redis_helper = RedisHelper(params=params) data = redis_helper.get_data_zset_with_index(key_name=key_name, start=0, end=1000) if not data: log_.info('{} —— ROV推荐进入了二次兜底, data = {}'.format(config_.ENV_TEXT, data)) send_msg_to_feishu('{} —— ROV推荐进入了二次兜底，请查看是否有数据更新失败问题。'.format(config_.ENV_TEXT)) # 二次兜底 bottom_data = bottom_strategy_last(size=size, app_type=app_type, ab_code=ab_code, params=params) return bottom_data # 视频状态过滤采用离线定时过滤方案 # 状态过滤 # filter_videos = FilterVideos(app_type=app_type, video_ids=data) # filtered_data = filter_videos.filter_video_status(video_ids=data) if len(data) > size: random_data = numpy.random.choice(data, size, False) else: random_data = data bottom_data = [{'videoId': int(item), 'pushFrom': config_.PUSH_FROM['bottom'], 'abCode': ab_code} for item in random_data] return bottom_data def bottom_strategy_last(size, app_type, ab_code, params): """ 兜底策略: 从兜底视频中随机获取视频，进行状态过滤后的视频 :param size: 需要获取的视频数 :param app_type: 产品标识 type-int :param ab_code: abCode :param params: :return: """ redis_helper = RedisHelper(params=params) bottom_data = redis_helper.get_data_zset_with_index(key_name=config_.BOTTOM_KEY_NAME, start=0, end=-1) random_data = numpy.random.choice(bottom_data, size * 30, False) # 视频状态过滤采用离线定时过滤方案 # 状态过滤 # filter_videos = FilterVideos(app_type=app_type, video_ids=random_data) # filtered_data = filter_videos.filter_video_status(video_ids=random_data) bottom_data = [{'videoId': int(video_id), 'pushFrom': config_.PUSH_FROM['bottom_last'], 'abCode': ab_code} for video_id in random_data[:size]] return bottom_data def bottom_strategy2(size, app_type, mid, uid, ab_code, client_info, params): """ 兜底策略: 从兜底视频中随机获取视频，进行过滤后的视频 :param size: 需要获取的视频数 :param app_type: 产品标识 type-int :param mid: mid :param uid: uid :param ab_code: abCode :param client_info: 地域信息 :param params: :return: """ # 获取存在城市分组数据的城市编码列表 city_code_list = [code for _, code in config_.CITY_CODE.items()] # 获取provinceCode province_code = client_info.get('provinceCode', '-1') # 获取cityCode city_code = client_info.get('cityCode', '-1') if city_code in city_code_list: # 分城市数据存在时，获取城市分组数据 region_code = city_code else: region_code = province_code if region_code == '': region_code = '-1' redis_helper = RedisHelper(params=params) bottom_data = redis_helper.get_data_from_set(key_name=config_.BOTTOM2_KEY_NAME) bottom_result = [] if bottom_data is None: return bottom_result if len(bottom_data) > 0: try: random_data = numpy.random.choice(bottom_data, size * 5, False) except Exception as e: random_data = bottom_data video_ids = [int(item) for item in random_data] # 过滤 filter_ = FilterVideos(request_id=params.request_id, app_type=app_type, mid=mid, uid=uid, video_ids=video_ids) filtered_data = filter_.filter_videos(pool_type='flow', region_code=region_code) if filtered_data: bottom_result = [{'videoId': int(video_id), 'pushFrom': config_.PUSH_FROM['bottom2'], 'abCode': ab_code} for video_id in filtered_data[:size]] return bottom_result def video_rank_by_w_h_rate(videos): """ 视频宽高比实验(每组的前两个视频调整为横屏视频)，根据视频宽高比信息对视频进行重排 :param videos: :return: """ redis_helper = RedisHelper() # ##### 判断前两个视频是否是置顶视频 或者 流量池视频 top_2_push_from_flag = [False, False] for i, video in enumerate(videos[:2]): if video['pushFrom'] in [config_.PUSH_FROM['top'], config_.PUSH_FROM['flow_recall']]: top_2_push_from_flag[i] = True if top_2_push_from_flag[0] and top_2_push_from_flag[1]: return videos # ##### 判断前两个视频是否为横屏 top_2_w_h_rate_flag = [False, False] for i, video in enumerate(videos[:2]): if video['pushFrom'] in [config_.PUSH_FROM['top'], config_.PUSH_FROM['flow_recall']]: # 视频来源为置顶 或 流量池时，不做判断 top_2_w_h_rate_flag[i] = True elif video['pushFrom'] in [config_.PUSH_FROM['rov_recall'], config_.PUSH_FROM['bottom']]: # 视频来源为 rov召回池 或 一层兜底时，判断是否是横屏 w_h_rate = redis_helper.get_score_with_value( key_name=config_.W_H_RATE_UP_1_VIDEO_LIST_KEY_NAME['rov_recall'], value=video['videoId']) if w_h_rate is not None: top_2_w_h_rate_flag[i] = True elif video['pushFrom'] == config_.PUSH_FROM['bottom_last']: # 视频来源为 二层兜底时，判断是否是横屏 w_h_rate = redis_helper.get_score_with_value( key_name=config_.W_H_RATE_UP_1_VIDEO_LIST_KEY_NAME['bottom_last'], value=video['videoId']) if w_h_rate is not None: top_2_w_h_rate_flag[i] = True if top_2_w_h_rate_flag[0] and top_2_w_h_rate_flag[1]: return videos # ##### 前两个视频中有不符合前面两者条件的，对视频进行位置调整 # 记录横屏视频位置 horizontal_video_index = [] # 记录流量池视频位置 flow_video_index = [] # 记录置顶视频位置 top_video_index = [] for i, video in enumerate(videos): # 视频来源为置顶 if video['pushFrom'] == config_.PUSH_FROM['top']: top_video_index.append(i) # 视频来源为流量池 elif video['pushFrom'] == config_.PUSH_FROM['flow_recall']: flow_video_index.append(i) # 视频来源为rov召回池 或 一层兜底 elif video['pushFrom'] in [config_.PUSH_FROM['rov_recall'], config_.PUSH_FROM['bottom']]: w_h_rate = redis_helper.get_score_with_value( key_name=config_.W_H_RATE_UP_1_VIDEO_LIST_KEY_NAME['rov_recall'], value=video['videoId']) if w_h_rate is not None: horizontal_video_index.append(i) else: continue # 视频来源为 二层兜底 elif video['pushFrom'] == config_.PUSH_FROM['bottom_last']: w_h_rate = redis_helper.get_score_with_value( key_name=config_.W_H_RATE_UP_1_VIDEO_LIST_KEY_NAME['bottom_last'], value=video['videoId']) if w_h_rate is not None: horizontal_video_index.append(i) else: continue # 重新排序 top2_index = [] for i in range(2): if i in top_video_index: top2_index.append(i) elif i in flow_video_index: top2_index.append(i) flow_video_index.remove(i) elif i in horizontal_video_index: top2_index.append(i) horizontal_video_index.remove(i) elif len(horizontal_video_index) > 0: # 调整横屏视频到第一位 top2_index.append(horizontal_video_index[0]) # 从横屏位置记录中移除 horizontal_video_index.pop(0) elif i == 0: return videos # 重排 flow_result = [videos[i] for i in flow_video_index] other_result = [videos[i] for i in range(len(videos)) if i not in top2_index and i not in flow_video_index] top2_result = [] for i, j in enumerate(top2_index): item = videos[j] if i != j: # 修改abCode item['abCode'] = config_.AB_CODE['w_h_rate'] top2_result.append(item) new_rank_result = top2_result for i in range(len(top2_index), len(videos)): if i in flow_video_index: new_rank_result.append(flow_result[0]) flow_result.pop(0) else: new_rank_result.append(other_result[0]) other_result.pop(0) return new_rank_result def video_rank_with_old_video(rank_result, old_video_recall, size, top_K, old_video_index=2): """ 视频分发排序 - 包含老视频， 老视频插入固定位置 :param rank_result: 排序后的结果 :param size: 请求数 :param old_video_index: 老视频插入的位置索引，默认为2 :return: new_rank_result """ if not old_video_recall: return rank_result if not rank_result: return old_video_recall[:size] # 视频去重 rank_video_ids = [item['videoId'] for item in rank_result] old_video_remove = [] for old_video in old_video_recall: if old_video['videoId'] in rank_video_ids: old_video_remove.append(old_video) for item in old_video_remove: old_video_recall.remove(item) if not old_video_recall: return rank_result # 插入老视频 # 随机获取一个视频 ind = random.randint(0, len(old_video_recall) - 1) old_video = old_video_recall[ind] # 插入 if len(rank_result) < top_K: new_rank_result = rank_result + [old_video] else: new_rank_result = rank_result[:old_video_index] + [old_video] + rank_result[old_video_index:] if len(new_rank_result) > size: # 判断后两位视频来源 push_from_1 = new_rank_result[-1]['pushFrom'] push_from_2 = new_rank_result[-2]['pushFrom'] if push_from_2 == config_.PUSH_FROM['rov_recall'] and push_from_1 == config_.PUSH_FROM['flow_recall']: new_rank_result = new_rank_result[:-2] + new_rank_result[-1:] return new_rank_result[:size] def video_new_rank2(data, size, top_K, flow_pool_P, ab_code, mid, exp_config=None, env_dict=None): """ 视频分发排序 :param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []} :param size: 请求数 :param top_K: 保证topK为召回池视频 type-int :param flow_pool_P: size-top_K视频为流量池视频的概率 type-float :return: rank_result """ if not data['rov_pool_recall'] and not data['flow_pool_recall']: return [], 0 #全量的是vlog,票圈精选， 334，60057, # 60054: simrecall, pre_str = "k_p2:" rov_recall_rank = data['rov_pool_recall'] #print(rov_recall_rank) #call rank service #flag_call_service = 0 sort_index = 0 if exp_config and "sort_flag" in exp_config: sort_index = exp_config["sort_flag"] #print("sort_index:", sort_index) redisObj = RedisHelper() vidKeys = [] rec_recall_item_list = [] rec_recall_vid_list = [] day_vidKeys = [] hour_vidKeys = [] pre_day_str = "v_ctr:" pre_hour_str = "v_hour_ctr:" for recall_item in data['rov_pool_recall']: try: vid = int(recall_item.get("videoId", 0)) rec_recall_vid_list.append(vid) rec_recall_item_list.append(recall_item) vidKeys.append(pre_str + str(vid)) day_vidKeys.append(pre_day_str+str(vid)) hour_vidKeys.append(pre_hour_str+str(vid)) except: continue video_scores = redisObj.get_batch_key(vidKeys) #print("video_scores:", video_scores) if (ab_code == 60066 or ab_code == 60069 or ab_code == 60070 or ab_code == 60071) and len(rec_recall_vid_list)>0: video_static_info = redisObj.get_batch_key(day_vidKeys) video_hour_static_info = redisObj.get_batch_key(hour_vidKeys) #print("env_dict:", env_dict) feature_dict = get_featurs(mid, data, size, top_K, flow_pool_P, rec_recall_vid_list,env_dict, video_static_info, video_hour_static_info) score_result = get_tf_serving_sores(feature_dict) #print("score_result:", score_result) if video_scores and len(video_scores)>0 and rec_recall_item_list and score_result and len(score_result) > 0\ and len(score_result) == len(rec_recall_item_list) and len(video_scores)== len(score_result): for i in range(len(score_result)): try: if video_scores[i] is None and len(score_result[i])>0: return_score = 0.000000001 # sore_index :10 = model score if sort_index == 10: total_score = score_result[i][0] else: total_score = return_score * score_result[i][0] rec_recall_item_list[i]['sort_score'] = total_score rec_recall_item_list[i]['base_rov_score'] = 0.0 rec_recall_item_list[i]['share_score'] = return_score rec_recall_item_list[i]['model_score'] = score_result[i][0] else: video_score_str = json.loads(video_scores[i]) # sore_index :10 = model score return_score = 0.000000001 if sort_index == 10: total_score = score_result[i][0] else: if len(video_score_str)>= sort_index and len(video_score_str)>0: return_score = video_score_str[sort_index] total_score = return_score * score_result[i][0] #print("total_score:", total_score, " model score :", score_result[i][0], "return_score:", # return_score) rec_recall_item_list[i]['sort_score'] = total_score rec_recall_item_list[i]['base_rov_score'] = video_score_str[0] rec_recall_item_list[i]['share_score'] = return_score rec_recall_item_list[i]['model_score'] = score_result[i][0] except Exception as e: #print('exception: {}:', e) return_score = 0.000000001 if sort_index == 10: total_score = 0.00000001 else: total_score = return_score * 0.00000001 rec_recall_item_list[i]['sort_score'] = total_score rec_recall_item_list[i]['base_rov_score'] = 0 rec_recall_item_list[i]['share_score'] = return_score rec_recall_item_list[i]['model_score'] = 0.00000001 rec_recall_item_list[i]['flag_call_service'] = 1 rov_recall_rank = sorted(rec_recall_item_list, key=lambda k: k.get('sort_score', 0), reverse=True) else: rov_recall_rank = sup_rank(video_scores, rec_recall_item_list) else: if video_scores and len(rec_recall_item_list) > 0 and len(video_scores)>0: for i in range(len(video_scores)): try: if video_scores[i] is None: rec_recall_item_list[i]['sort_score'] = 0.0 else: video_score_str = json.loads(video_scores[i]) # print("video_score_str:", video_score_str) rec_recall_item_list[i]['sort_score'] = video_score_str[0] except Exception: rec_recall_item_list[i]['sort_score'] = 0.0 rov_recall_rank = sorted(rec_recall_item_list, key=lambda k: k.get('sort_score', 0), reverse=True) #print(rov_recall_rank) flow_recall_rank = sorted(data['flow_pool_recall'], key=lambda k: k.get('rovScore', 0), reverse=True) rov_recall_rank, flow_recall_rank = remove_duplicate(rov_recall=rov_recall_rank, flow_recall=flow_recall_rank, top_K=top_K) rank_result = [] rank_set = set('') # 从ROV召回池中获取top k if len(rov_recall_rank) > 0: rank_result.extend(rov_recall_rank[:top_K]) rov_recall_rank = rov_recall_rank[top_K:] else: rank_result.extend(flow_recall_rank[:top_K]) flow_recall_rank = flow_recall_rank[top_K:] # 按概率 p 及score排序获取 size - k 个视频 flow_num = 0 flowConfig = 0 # 本段代码控制流量池，通过实验传参，现不动 if flowConfig == 1 and len(rov_recall_rank) > 0: for recall_item in rank_result: flow_recall_name = recall_item.get("flowPool", '') flow_num = flow_num + 1 all_recall_rank = rov_recall_rank + flow_recall_rank if flow_num > 0: rank_result.extend(all_recall_rank[:size - top_K]) return rank_result, flow_num else: i = 0 while i < size - top_K: # 随机生成[0, 1)浮点数 rand = random.random() # log_.info('rand: {}'.format(rand)) if rand < flow_pool_P: if flow_recall_rank: rank_result.append(flow_recall_rank[0]) flow_recall_rank.remove(flow_recall_rank[0]) else: rank_result.extend(rov_recall_rank[:size - top_K - i]) return rank_result[:size], flow_num else: if rov_recall_rank: rank_result.append(rov_recall_rank[0]) rov_recall_rank.remove(rov_recall_rank[0]) else: rank_result.extend(flow_recall_rank[:size - top_K - i]) return rank_result[:size], flow_num i += 1 else: i = 0 while i < size - top_K: # 随机生成[0, 1)浮点数 rand = random.random() # log_.info('rand: {}'.format(rand)) if rand < flow_pool_P: if flow_recall_rank: rank_result.append(flow_recall_rank[0]) flow_recall_rank.remove(flow_recall_rank[0]) else: rank_result.extend(rov_recall_rank[:size - top_K - i]) return rank_result[:size], flow_num else: if rov_recall_rank: rank_result.append(rov_recall_rank[0]) rov_recall_rank.remove(rov_recall_rank[0]) else: rank_result.extend(flow_recall_rank[:size - top_K - i]) return rank_result[:size], flow_num i += 1 return rank_result[:size], flow_num def video_new_rank3(data, size, top_K, flow_pool_P, rank_key_prefix='rank:score1:'): """ 视频分发排序 :param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []} :param size: 请求数 :param top_K: 保证topK为召回池视频 type-int :param flow_pool_P: size-top_K视频为流量池视频的概率 type-float :param rank_key_prefix: :return: rank_result """ redis_helper = RedisHelper() if not data['rov_pool_recall'] and not data['flow_pool_recall']: return [], 0 rov_recall_rank = data['rov_pool_recall'] vid_keys = [] rec_recall_item_list = [] rec_recall_vid_list = [] for recall_item in data['rov_pool_recall']: try: vid = int(recall_item.get("videoId", 0)) rec_recall_vid_list.append(vid) rec_recall_item_list.append(recall_item) vid_keys.append(f"{rank_key_prefix}{vid}") except: continue video_scores = redis_helper.get_batch_key(vid_keys) if video_scores and len(rec_recall_item_list) > 0 and len(rec_recall_item_list) == len(video_scores): for i in range(len(video_scores)): try: if video_scores[i] is None: rec_recall_item_list[i]['sort_score'] = 0.0 else: rec_recall_item_list[i]['sort_score'] = float(video_scores[i]) except Exception: rec_recall_item_list[i]['sort_score'] = 0.0 rov_recall_rank = sorted(rec_recall_item_list, key=lambda k: k.get('sort_score', 0), reverse=True) flow_recall_rank = sorted(data['flow_pool_recall'], key=lambda k: k.get('rovScore', 0), reverse=True) rov_recall_rank, flow_recall_rank = remove_duplicate( rov_recall=rov_recall_rank, flow_recall=flow_recall_rank, top_K=top_K ) rank_result = [] # 从ROV召回池中获取top k if len(rov_recall_rank) > 0: rank_result.extend(rov_recall_rank[:top_K]) rov_recall_rank = rov_recall_rank[top_K:] else: rank_result.extend(flow_recall_rank[:top_K]) flow_recall_rank = flow_recall_rank[top_K:] # 按概率 p 及score排序获取 size - k 个视频 flow_num = 0 i = 0 while i < size - top_K: # 随机生成[0, 1)浮点数 rand = random.random() # log_.info('rand: {}'.format(rand)) if rand < flow_pool_P: if flow_recall_rank: rank_result.append(flow_recall_rank[0]) flow_recall_rank.remove(flow_recall_rank[0]) else: rank_result.extend(rov_recall_rank[:size - top_K - i]) return rank_result[:size], flow_num else: if rov_recall_rank: rank_result.append(rov_recall_rank[0]) rov_recall_rank.remove(rov_recall_rank[0]) else: rank_result.extend(flow_recall_rank[:size - top_K - i]) return rank_result[:size], flow_num i += 1 return rank_result[:size], flow_num # 排序服务兜底 def sup_rank(video_scores, recall_list): if video_scores and len(recall_list) > 0: for i in range(len(video_scores)): try: if video_scores[i] is None: recall_list[i]['sort_score'] = 0.0 else: video_score_str = json.loads(video_scores[i]) recall_list[i]['flag_call_service'] = 0 recall_list[i]['sort_score'] = video_score_str[0] except Exception: recall_list[i]['sort_score'] = 0.0 rov_recall_rank = sorted(recall_list, key=lambda k: k.get('sort_score', 0), reverse=True) #print("rov_recall_rank:", rov_recall_rank) else: rov_recall_rank = recall_list return rov_recall_rank def video_sanke_rank(data, size, top_K, flow_pool_P, ab_Code='', exp_config=None): """ 视频分发排序 :param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []} :param size: 请求数 :param top_K: 保证topK为召回池视频 type-int :param flow_pool_P: size-top_K视频为流量池视频的概率 type-float :return: rank_result """ if not data['rov_pool_recall'] and not data['flow_pool_recall'] \ and len(data['u2i_recall'])==0 and len(data['w2v_recall'])==0 \ and len(data['sim_recall']) == 0 and len(data['u2u2i_recall']) == 0 : return [], 0 # 地域分组小时级规则更新数据 recall_dict = {} region_h_recall = [item for item in data['rov_pool_recall'] if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_region_h']] region_h_recall_rank = sorted(region_h_recall, key=lambda k: k.get('rovScore', 0), reverse=True) recall_dict['rov_recall_region_h'] = region_h_recall_rank # 地域分组小时级更新24h规则更新数据 region_24h_recall = [item for item in data['rov_pool_recall'] if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_region_24h']] region_24h_recall_rank = sorted(region_24h_recall, key=lambda k: k.get('rovScore', 0), reverse=True) recall_dict['rov_recall_region_24h'] = region_24h_recall_rank # 相对24h规则更新数据 rule_24h_recall = [item for item in data['rov_pool_recall'] if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_24h']] rule_24h_recall_rank = sorted(rule_24h_recall, key=lambda k: k.get('rovScore', 0), reverse=True) recall_dict['rov_recall_24h'] = rule_24h_recall_rank # 相对24h规则筛选后剩余更新数据 rule_24h_dup_recall = [item for item in data['rov_pool_recall'] if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_24h_dup']] rule_24h_dup_recall_rank = sorted(rule_24h_dup_recall, key=lambda k: k.get('rovScore', 0), reverse=True) recall_dict['rov_recall_24h_dup'] = rule_24h_dup_recall_rank hot_recall = [] w2v_recall =[] sim_recall = [] u2u2i_recall = [] if ab_Code==60058: if len(data['u2i_recall'])>0: hot_recall = sorted(data['u2i_recall'], key=lambda k: k.get('rovScore', 0), reverse=True) recall_dict['u2i_recall'] = hot_recall elif ab_Code==60059: if len(data['w2v_recall'])>0: recall_dict['w2v_recall'] = data['w2v_recall'] else: recall_dict['w2v_recall'] = w2v_recall elif ab_Code==60061 or ab_Code==60063: if len(data['sim_recall'])>0: recall_dict['sim_recall'] = data['sim_recall'] else: recall_dict['sim_recall'] = sim_recall elif ab_Code==60062: if len(data['u2u2i_recall'])>0: recall_dict['u2u2i_recall'] = data['u2u2i_recall'] else: recall_dict['u2u2i_recall'] = u2u2i_recall recall_list = [('rov_recall_region_h',1, 1),('rov_recall_region_h',0.5, 1),('rov_recall_region_24h',1,1), ('u2i_recall',0.5,1), ('w2v_recall',0.5,1),('rov_recall_24h',1,1), ('rov_recall_24h_dup',0.5,1)] if exp_config and exp_config['recall_list']: recall_list = exp_config['recall_list'] #print("recall_config:", recall_list) rov_recall_rank = [] select_ids = set('') for i in range(3): if len(rov_recall_rank)>8: break for per_recall_item in recall_list: per_recall_name = per_recall_item[0] per_recall_freq = per_recall_item[1] per_limt_num = per_recall_item[2] rand_num = random.random() #print(recall_dict[per_recall_name]) if rand_num=per_limt_num: break # print("rov_recall_rank:") # print(rov_recall_rank) #rov_recall_rank = region_h_recall_rank + region_24h_recall_rank + \ # rule_24h_recall_rank + rule_24h_dup_recall_rank # 流量池 flow_recall_rank = sorted(data['flow_pool_recall'], key=lambda k: k.get('rovScore', 0), reverse=True) # 对各路召回的视频进行去重 rov_recall_rank, flow_recall_rank = remove_duplicate(rov_recall=rov_recall_rank, flow_recall=flow_recall_rank, top_K=top_K) # log_.info('remove_duplicate finished! rov_recall_rank = {}, flow_recall_rank = {}'.format( # rov_recall_rank, flow_recall_rank)) # rank_result = relevant_recall_rank rank_result = [] # 从ROV召回池中获取top k if len(rov_recall_rank) > 0: rank_result.extend(rov_recall_rank[:top_K]) rov_recall_rank = rov_recall_rank[top_K:] else: rank_result.extend(flow_recall_rank[:top_K]) flow_recall_rank = flow_recall_rank[top_K:] flow_num = 0 flowConfig =0 if exp_config and exp_config['flowConfig']: flowConfig = exp_config['flowConfig'] if flowConfig == 1 and len(rov_recall_rank) > 0: rank_result.extend(rov_recall_rank[:top_K]) for recall_item in rank_result: flow_recall_name = recall_item.get("flowPool", '') if flow_recall_name is not None and flow_recall_name.find("#") > -1: flow_num = flow_num + 1 all_recall_rank = rov_recall_rank + flow_recall_rank if flow_num > 0: rank_result.extend(all_recall_rank[:size - top_K]) return rank_result[:size], flow_num else: # 按概率 p 及score排序获取 size - k 个视频 i = 0 while i < size - top_K: # 随机生成[0, 1)浮点数 rand = random.random() # log_.info('rand: {}'.format(rand)) if rand < flow_pool_P: if flow_recall_rank: rank_result.append(flow_recall_rank[0]) flow_recall_rank.remove(flow_recall_rank[0]) else: rank_result.extend(rov_recall_rank[:size - top_K - i]) return rank_result[:size], flow_num else: if rov_recall_rank: rank_result.append(rov_recall_rank[0]) rov_recall_rank.remove(rov_recall_rank[0]) else: rank_result.extend(flow_recall_rank[:size - top_K - i]) return rank_result[:size], flow_num i += 1 else: # 按概率 p 及score排序获取 size - k 个视频 i = 0 while i < size - top_K: # 随机生成[0, 1)浮点数 rand = random.random() # log_.info('rand: {}'.format(rand)) if rand < flow_pool_P: if flow_recall_rank: rank_result.append(flow_recall_rank[0]) flow_recall_rank.remove(flow_recall_rank[0]) else: rank_result.extend(rov_recall_rank[:size - top_K - i]) return rank_result[:size], flow_num else: if rov_recall_rank: rank_result.append(rov_recall_rank[0]) rov_recall_rank.remove(rov_recall_rank[0]) else: rank_result.extend(flow_recall_rank[:size - top_K - i]) return rank_result[:size],flow_num i += 1 return rank_result[:size], flow_num def video_sank_pos_rank(data, size, top_K, flow_pool_P, ab_Code='', exp_config=None): """ 视频分发排序 :param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []} :param size: 请求数 :param top_K: 保证topK为召回池视频 type-int :param flow_pool_P: size-top_K视频为流量池视频的概率 type-float :return: rank_result """ if not data['rov_pool_recall'] and not data['flow_pool_recall'] \ and len(data['u2i_recall'])==0 and len(data['w2v_recall'])==0 \ and len(data['sim_recall']) == 0 and len(data['u2u2i_recall']) == 0 : return [], 0 # 地域分组小时级规则更新数据 recall_dict = {} region_h_recall = [item for item in data['rov_pool_recall'] if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_region_h']] region_h_recall_rank = sorted(region_h_recall, key=lambda k: k.get('rovScore', 0), reverse=True) recall_dict['rov_recall_region_h'] = region_h_recall_rank # 地域分组小时级更新24h规则更新数据 region_24h_recall = [item for item in data['rov_pool_recall'] if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_region_24h']] region_24h_recall_rank = sorted(region_24h_recall, key=lambda k: k.get('rovScore', 0), reverse=True) recall_dict['rov_recall_region_24h'] = region_24h_recall_rank # 相对24h规则更新数据 rule_24h_recall = [item for item in data['rov_pool_recall'] if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_24h']] rule_24h_recall_rank = sorted(rule_24h_recall, key=lambda k: k.get('rovScore', 0), reverse=True) recall_dict['rov_recall_24h'] = rule_24h_recall_rank # 相对24h规则筛选后剩余更新数据 rule_24h_dup_recall = [item for item in data['rov_pool_recall'] if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_24h_dup']] rule_24h_dup_recall_rank = sorted(rule_24h_dup_recall, key=lambda k: k.get('rovScore', 0), reverse=True) recall_dict['rov_recall_24h_dup'] = rule_24h_dup_recall_rank u2i_recall = [] u2i_play_recall = [] w2v_recall =[] sim_recall = [] u2u2i_recall = [] return_video_recall = [] #print("") if ab_Code==60058: if len(data['u2i_recall'])>0: recall_dict['u2i_recall'] = data['u2i_recall'] else: recall_dict['u2i_recall'] = u2i_recall if len(data['u2i_play_recall']) > 0: recall_dict['u2i_play_recall'] = data['u2i_play_recall'] else: recall_dict['u2i_play_recall'] = u2i_play_recall elif ab_Code==60059: if len(data['w2v_recall'])>0: recall_dict['w2v_recall'] = data['w2v_recall'] else: recall_dict['w2v_recall'] = w2v_recall elif ab_Code==60061 or ab_Code==60063: if len(data['sim_recall'])>0: recall_dict['sim_recall'] = data['sim_recall'] else: recall_dict['sim_recall'] = sim_recall elif ab_Code==60062: if len(data['u2u2i_recall'])>0: recall_dict['u2u2i_recall'] = data['u2u2i_recall'] else: recall_dict['u2u2i_recall'] = u2u2i_recall elif ab_Code==60064: if len(data['return_video_recall'])>0: recall_dict['return_video_recall'] = data['return_video_recall'] else: recall_dict['return_video_recall'] = return_video_recall recall_pos1 = [('rov_recall_region_h',0, 0.98),('rov_recall_24h',0.98, 1),('rov_recall_region_24h',0,1), ('rov_recall_24h',0,1), ('rov_recall_24h_dup',0,1)] recall_pos2 = [('rov_recall_region_h',0,0.98),('rov_recall_24h',0.98,1),('rov_recall_region_24h',0,1), ('rov_recall_24h',0,1),('rov_recall_24h_dup',0,1)] recall_pos3 = [('rov_recall_region_h', 0,0.98), ('rov_recall_24h', 0.98,1), ('rov_recall_region_24h', 0,1), ('rov_recall_24h', 0,1), ('rov_recall_24h_dup', 0,1)] recall_pos4 = [('rov_recall_region_h', 0,0.98), ('rov_recall_24h', 0,0.02), ('rov_recall_region_24h', 0,1), ('rov_recall_24h', 0,1), ('rov_recall_24h_dup', 0,1)] if exp_config and 'recall_pos1' in exp_config \ and 'recall_pos2' in exp_config \ and 'recall_pos3' in exp_config \ and 'recall_pos4' in exp_config : recall_pos1 = exp_config['recall_pos1'] recall_pos2 = exp_config['recall_pos2'] recall_pos3 = exp_config['recall_pos3'] recall_pos4 = exp_config['recall_pos4'] #print("recall_config:", recall_pos1) rov_recall_rank = [] recall_list = [] recall_list.append(recall_pos1) recall_list.append(recall_pos2) recall_list.append(recall_pos3) recall_list.append(recall_pos4) select_ids = set('') recall_num_limit_dict = {} if exp_config and 'recall_num_limit' in exp_config: recall_num_limit_dict = exp_config['recall_num_limit'] exp_recall_dict = {} #index_pos = 0 for j in range(3): if len(rov_recall_rank)>12: break # choose pos for recall_pos_config in recall_list: rand_num = random.random() index_pos = 0 # choose pos recall for per_recall_item in recall_pos_config: if index_pos == 1: break if len(per_recall_item)<3: continue per_recall_name = per_recall_item[0] per_recall_min = float(per_recall_item[1]) per_recall_max = float(per_recall_item[2]) per_recall_num = exp_recall_dict.get(per_recall_name, 0) per_recall_total_num = recall_num_limit_dict.get(per_recall_name, 0) # recall set total num if len(recall_num_limit_dict)>0 and per_recall_total_num>0 and per_recall_num>= per_recall_total_num: continue if rand_num >= per_recall_min and rand_num < per_recall_max and per_recall_name in recall_dict: per_recall = recall_dict[per_recall_name] for recall_item in per_recall: vid = recall_item['videoId'] if vid in select_ids: continue recall_item['rand'] = rand_num rov_recall_rank.append(recall_item) select_ids.add(vid) if per_recall_name in exp_recall_dict: exp_recall_dict[per_recall_name] +=1 else: exp_recall_dict[per_recall_name] = 1 index_pos = 1 break #print("rov_recall_rank:", rov_recall_rank) if len(rov_recall_rank)<4: rov_doudi_rank = region_h_recall_rank + sim_recall + u2i_recall + u2u2i_recall + w2v_recall +return_video_recall+u2i_play_recall+ region_24h_recall_rank + rule_24h_recall_rank + rule_24h_dup_recall_rank for recall_item in rov_doudi_rank: vid = recall_item['videoId'] if vid in select_ids: continue rov_recall_rank.append(recall_item) select_ids.add(vid) if len(rov_recall_rank)>12: break # print("rov_recall_rank:") #print(rov_recall_rank) # 流量池 flow_recall_rank = sorted(data['flow_pool_recall'], key=lambda k: k.get('rovScore', 0), reverse=True) # 对各路召回的视频进行去重 rov_recall_rank, flow_recall_rank = remove_duplicate(rov_recall=rov_recall_rank, flow_recall=flow_recall_rank, top_K=top_K) # log_.info('remove_duplicate finished! rov_recall_rank = {}, flow_recall_rank = {}'.format( # rov_recall_rank, flow_recall_rank)) # rank_result = relevant_recall_rank rank_result = [] # 从ROV召回池中获取top k if len(rov_recall_rank) > 0: rank_result.extend(rov_recall_rank[:top_K]) rov_recall_rank = rov_recall_rank[top_K:] else: rank_result.extend(flow_recall_rank[:top_K]) flow_recall_rank = flow_recall_rank[top_K:] flow_num = 0 flowConfig =0 if exp_config and exp_config['flowConfig']: flowConfig = exp_config['flowConfig'] if flowConfig == 1 and len(rov_recall_rank) > 0: rank_result.extend(rov_recall_rank[:top_K]) for recall_item in rank_result: flow_recall_name = recall_item.get("flowPool", '') if flow_recall_name is not None and flow_recall_name.find("#") > -1: flow_num = flow_num + 1 all_recall_rank = rov_recall_rank + flow_recall_rank if flow_num > 0: rank_result.extend(all_recall_rank[:size - top_K]) return rank_result[:size], flow_num else: # 按概率 p 及score排序获取 size - k 个视频 i = 0 while i < size - top_K: # 随机生成[0, 1)浮点数 rand = random.random() # log_.info('rand: {}'.format(rand)) if rand < flow_pool_P: if flow_recall_rank: rank_result.append(flow_recall_rank[0]) flow_recall_rank.remove(flow_recall_rank[0]) else: rank_result.extend(rov_recall_rank[:size - top_K - i]) return rank_result[:size], flow_num else: if rov_recall_rank: rank_result.append(rov_recall_rank[0]) rov_recall_rank.remove(rov_recall_rank[0]) else: rank_result.extend(flow_recall_rank[:size - top_K - i]) return rank_result[:size], flow_num i += 1 else: # 按概率 p 及score排序获取 size - k 个视频 i = 0 while i < size - top_K: # 随机生成[0, 1)浮点数 rand = random.random() # log_.info('rand: {}'.format(rand)) if rand < flow_pool_P: if flow_recall_rank: rank_result.append(flow_recall_rank[0]) flow_recall_rank.remove(flow_recall_rank[0]) else: rank_result.extend(rov_recall_rank[:size - top_K - i]) return rank_result[:size], flow_num else: if rov_recall_rank: rank_result.append(rov_recall_rank[0]) rov_recall_rank.remove(rov_recall_rank[0]) else: rank_result.extend(flow_recall_rank[:size - top_K - i]) return rank_result[:size],flow_num i += 1 return rank_result[:size], flow_num if __name__ == '__main__': d_test = [{'videoId': 10028734, 'rovScore': 99.977, 'pushFrom': 'recall_pool', 'abCode': 10000}, {'videoId': 1919925, 'rovScore': 99.974, 'pushFrom': 'recall_pool', 'abCode': 10000}, {'videoId': 9968118, 'rovScore': 99.972, 'pushFrom': 'recall_pool', 'abCode': 10000}, {'videoId': 9934863, 'rovScore': 99.971, 'pushFrom': 'recall_pool', 'abCode': 10000}, {'videoId': 10219869, 'flowPool': '1#1#1#1640830818883', 'rovScore': 82.21929728934731, 'pushFrom': 'flow_pool', 'abCode': 10000}, {'videoId': 10212814, 'flowPool': '1#1#1#1640759014984', 'rovScore': 81.26694187726412, 'pushFrom': 'flow_pool', 'abCode': 10000}, {'videoId': 10219437, 'flowPool': '1#1#1#1640827620520', 'rovScore': 81.21634156641908, 'pushFrom': 'flow_pool', 'abCode': 10000}, {'videoId': 1994050, 'rovScore': 99.97, 'pushFrom': 'recall_pool', 'abCode': 10000}, {'videoId': 9894474, 'rovScore': 99.969, 'pushFrom': 'recall_pool', 'abCode': 10000}, {'videoId': 10028081, 'rovScore': 99.966, 'pushFrom': 'recall_pool', 'abCode': 10000}] res = video_rank_by_w_h_rate(videos=d_test) for tmp in res: print(tmp)