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 log_ = Log() config_ = set_config() def video_rank(data, size): """ 视频分发排序 :param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []} :param size: 请求数 :return: rank_result """ if not data['rov_pool_recall'] and not data['flow_pool_recall']: return None # 将各路召回的视频按照score从大到小排序 # ROV召回池 rov_recall_rank = sorted(data['rov_pool_recall'], key=lambda k: (k.get('rovScore'), 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) # log_.info('remove_duplicate finished! rov_recall_rank = {}, flow_recall_rank = {}'.format( # rov_recall_rank, flow_recall_rank)) # 从ROV召回池中获取top k if len(rov_recall_rank) > 0: rank_result = rov_recall_rank[:config_.K] rov_recall_rank = rov_recall_rank[config_.K:] else: rank_result = flow_recall_rank[:config_.K] flow_recall_rank = flow_recall_rank[config_.K:] # 按概率 p 及score排序获取 size - k 个视频 i = 0 while i < size - config_.K: # 随机生成[0, 1)浮点数 rand = random.random() # log_.info('rand: {}'.format(rand)) if rand < config_.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 - config_.K - i]) return rank_result 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 - config_.K - i]) return rank_result i += 1 return rank_result def remove_duplicate(rov_recall, flow_recall): """ 对多路召回的视频去重 去重原则： 如果视频在ROV召回池topK，则保留ROV召回池，否则保留流量池 :param rov_recall: ROV召回池-已排序 :param flow_recall: 流量池-已排序 :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[:config_.K]: if item['videoId'] in flow_recall_video_ids: flow_recall_video_ids.remove(item['videoId']) # other for item in rov_recall[config_.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(size, app_type, ab_code): """ 兜底策略: 从ROV召回池中获取top1000，进行状态过滤后的视频 :param size: 需要获取的视频数 :param app_type: 产品标识 type-int :param ab_code: abCode :return: """ pool_recall = PoolRecall(app_type=app_type, ab_code=ab_code) key_name, _ = pool_recall.get_pool_redis_key(pool_type='rov') redis_helper = RedisHelper() data = redis_helper.get_data_zset_with_index(key_name=key_name, start=0, end=1000) if not data: log_.info('预发布环境 —— ROV推荐进入了二次兜底, data = {}'.format(data)) send_msg_to_feishu('预发布环境 —— ROV推荐进入了二次兜底，请查看是否有数据更新失败问题。') # 二次兜底 bottom_data = bottom_strategy_last(size=size, app_type=app_type, ab_code=ab_code) 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): """ 兜底策略: 从兜底视频中随机获取视频，进行状态过滤后的视频 :param size: 需要获取的视频数 :param app_type: 产品标识 type-int :param ab_code: abCode :return: """ redis_helper = RedisHelper() 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 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 == 1: return videos elif i == 2: top2_index.append(i) # 重排 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(2, 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 if __name__ == '__main__': d_test = [{'videoId': 1, 'rovScore': 10, 'pushFrom': 'op_manual', 'abCode': 10000}, {'videoId': 1919925, 'rovScore': 9, 'pushFrom': 'recall_pool', 'abCode': 10000}, {'videoId': 3, 'rovScore': 8, 'pushFrom': 'recall_pool', 'abCode': 10000}, {'videoId': 4, 'rovScore': 7, 'pushFrom': 'flow_pool', 'abCode': 10000}, {'videoId': 5, 'rovScore': 6, 'pushFrom': 'flow_pool', 'abCode': 10000}, {'videoId': 6, 'rovScore': 5, 'pushFrom': 'recall_pool', 'abCode': 10000}, {'videoId': 3674236, 'rovScore': 4, 'pushFrom': 'recall_pool', 'abCode': 10000}, {'videoId': 3247671, 'rovScore': 3, 'pushFrom': 'recall_pool', 'abCode': 10000}, {'videoId': 7837973, 'rovScore': 2, 'pushFrom': 'recall_pool', 'abCode': 10000}] res = video_rank_by_w_h_rate(videos=d_test) for tmp in res: print(tmp)