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, 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 None # 将各路召回的视频按照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) # 地域分组小时级规则更新数据 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 + \ 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 # 流量池 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 # 从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 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 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] 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)