rov-server/video_rank.py

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

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(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 []

    redisObj = RedisHelper()
    video_id_list = []
    for d in data:
        video_id_list.append(d.get('rovScore', '0'))
    video_scores = redisObj.get_batch_key(video_id_list)
    video_items = []
    for i in range(len(video_id_list)):
        try:
            video_score_str = json.load(video_scores[i])
            video_items.append((video_id_list[i], video_score_str[0]))
        except Exception:
            video_items.append((video_id_list[i], 0.0))
    sort_items = sorted(video_items, key=lambda k: k[1], reverse=True)
    rov_recall_rank = sort_items
    # 流量池
    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 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]


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)