rov-server/video_rank.py

import random
import numpy

from log import Log
from config import set_config
from video_recall import PoolRecall
from utils import FilterVideos

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


async def bottom_strategy(size, app_type, ab_code):
    """
    兜底策略: 从ROV召回池中获取top1000，进行状态过滤后的视频
    :param size: 需要获取的视频数
    :param app_type: 产品标识 type-int
    :param ab_code: abCode
    :param mid:
    :param uid:
    :return:
    """
    pool_recall = PoolRecall(app_type=app_type, ab_code=ab_code)
    key_name, _ = await pool_recall.get_pool_redis_key(pool_type='rov')
    if not key_name:
        log_.info('bottom strategy no data!')
        return []
    data = await aiocache.get_data_zset_with_index(key_name=key_name, start=0, end=1000)
    if not data:
        log_.info('bottom strategy no data!')
        return []
    # 状态过滤
    filter_videos = FilterVideos(app_type=app_type, video_ids=data)
    filtered_data = await filter_videos.filter_video_status(video_ids=data)
    if len(filtered_data) > size:
        random_data = numpy.random.choice(filtered_data, size, False)
    else:
        random_data = filtered_data
    bottom_data = [{'videoId': item, 'pushFrom': 'bottom_strategy', 'abCode': ab_code} for item in random_data]
    return bottom_data


if __name__ == '__main__':
    d_test = [[{'videoId': 3674236, 'rovScore': 99.24105262298141, 'pushFrom': 'recall_pool', 'abCode': 10000}, {'videoId': 1915009, 'rovScore': 99.248872388032, 'pushFrom': 'recall_pool', 'abCode': 10000}, {'videoId': 9033859, 'rovScore': 99.21956695197761, 'pushFrom': 'recall_pool', 'abCode': 10000}, {'videoId': 4258137, 'rovScore': 99.24737622823497, 'pushFrom': 'recall_pool', 'abCode': 10000}, {'videoId': 9034962, 'rovScore': 99.18993382219318, 'pushFrom': 'recall_pool', 'abCode': 10000}, {'videoId': 1922051, 'rovScore': 99.2351969813565, 'pushFrom': 'recall_pool', 'abCode': 10000}, {'videoId': 7829308, 'rovScore': 99.25465474490638, 'pushFrom': 'recall_pool', 'abCode': 10000}, {'videoId': 3247671, 'rovScore': 99.24601245746983, 'pushFrom': 'recall_pool', 'abCode': 10000}, {'videoId': 5831941, 'rovScore': 99.16776814766304, 'pushFrom': 'recall_pool', 'abCode': 10000}, {'videoId': 7837973, 'rovScore': 99.253749334822, 'pushFrom': 'recall_pool', 'abCode': 10000}], [{'videoId': 9035245, 'flowPool': '1#1#1#1636085384424', 'rovScore': 1.0, 'pushFrom': 'flow_pool', 'abCode': 10000}, {'videoId': 9034828, 'flowPool': '1#1#1#1636090368461', 'rovScore': 1.0, 'pushFrom': 'flow_pool', 'abCode': 10000}, {'videoId': 9035244, 'flowPool': '1#1#1#1636085467105', 'rovScore': 1.0, 'pushFrom': 'flow_pool', 'abCode': 10000}, {'videoId': 9035237, 'flowPool': '1#1#1#1636086478074', 'rovScore': 1.0, 'pushFrom': 'flow_pool', 'abCode': 10000}]]
    data = {
        'rov_pool_recall': d_test[0],
        'flow_pool_recall': d_test[1]
    }
    res = video_rank(data, size=10)
    for item in res:
        print(item)