rov-server/video_rank.py

import copy
import json
import random
import numpy
from typing import Dict
from typing import Set

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
from parameter_update import param_update_rule

log_ = Log()
config_ = set_config()


def video_rank(data, size, top_K, flow_pool_P, flow_pool_recall_process=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
    """
    # add_flow_pool_recall_log
    if flow_pool_recall_process is None:
        flow_pool_recall_process = {}

    if not data['rov_pool_recall'] and not data['flow_pool_recall']:
        # add_flow_pool_recall_log
        return [], flow_pool_recall_process
        # 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)

    # 不分地域小时级规则更新数据
    rule_h_recall = [item for item in data['rov_pool_recall']
                     if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_h_h']]
    rule_h_recall_rank = sorted(rule_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 + rule_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 = []

    # add_flow_pool_recall_log
    flow_pool_recall_process['recall_duplicate_res'] = {'rov_recall_rank': rov_recall_rank,
                                                        'flow_recall_rank': copy.deepcopy(flow_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()

        # add_flow_pool_recall_log
        flow_pool_recall_process['flow_pool_P'] = flow_pool_P
        flow_pool_recall_process[f'{i}_rand'] = rand

        # 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_pool_recall_process
        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_pool_recall_process
        i += 1
    return rank_result[:size], flow_pool_recall_process


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<len(all_flow_recall_rank) or jj<len(rov_recall_rank)):
        # 随机生成[0, 1)浮点数
        rand = random.random()
        # log_.info('rand: {}'.format(rand))
        if rand < flow_pool_P:
            for flow_item in all_flow_recall_rank:
                j+=1
                if flow_item[0] in rank_set:
                    continue
                else:
                    rank_result.append(flow_item)
                    rank_set.add(flow_item[0])
                    add_flow_set.add(flow_item[0])
                i += 1
                if i>= 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:', flow_pool_recall_process=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
        :param rank_key_prefix:
        :return: rank_result
    """
    redis_helper = RedisHelper()

    if flow_pool_recall_process is None:
        flow_pool_recall_process = {}

    if not data['rov_pool_recall'] and not data['flow_pool_recall']:
        return [], 0, flow_pool_recall_process

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

    flow_pool_recall_process['recall_duplicate_res'] = {'rov_recall_rank': rov_recall_rank,
                                                        'flow_recall_rank': copy.deepcopy(flow_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 个视频
    flow_num = 0
    i = 0
    while i < size - top_K:
        # 随机生成[0, 1)浮点数
        rand = random.random()
        flow_pool_recall_process['flow_pool_P'] = flow_pool_P
        flow_pool_recall_process[f'{i}_rand'] = 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, flow_pool_recall_process
        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, flow_pool_recall_process
        i += 1
    return rank_result[:size], flow_num, flow_pool_recall_process

def video_new_rank3_4density(data, size, top_K, flow_pool_P, rank_key_prefix='rank:score1:',
                             flow_pool_recall_process=None,
                             app_type=None, rule_key_str = 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
        :param rank_key_prefix:
        :return: rank_result
    """
    # 1 读取多样性密度控制规则 todo 没有判断非字典数据，有报错风险，先用try兜底。
    redis_helper = RedisHelper()
    density_rules = {}
    rules_all = param_update_rule(redis_helper, rule_key_str)
    try:
        if len(rules_all) != 0:
            for k, v in rules_all.items():
                if str(app_type) in v.keys():
                    app_type_rule = v[str(app_type)]
                    if "density" in app_type_rule.keys():
                        density_rules[k] = app_type_rule["density"]
    except Exception as e:
        log_.error("something is wrong in parsing density_rules:{}".format(e))
        density_rules = {}

    if flow_pool_recall_process is None:
        flow_pool_recall_process = {}

    if not data['rov_pool_recall'] and not data['flow_pool_recall']:
        return [], 0, flow_pool_recall_process

    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
    )
    # 2 多样性需求，给video添加tag todo
    video_ids = []
    video_ids.extend([v["videoId"] for v in rov_recall_rank])
    video_ids.extend([v["videoId"] for v in flow_recall_rank])
    video_ids = list(set(video_ids))
    video_tag_dict = get_video_tags(redis_helper, video_ids)
    for v in rov_recall_rank:
        v["tags"] = video_tag_dict.get(v["videoId"], [])
    for v in flow_recall_rank:
        v["tags"] = video_tag_dict.get(v["videoId"], [])

    rank_result = []
    flow_pool_recall_process['recall_duplicate_res'] = {'rov_recall_rank': rov_recall_rank,
                                                        'flow_recall_rank': copy.deepcopy(flow_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 个视频
    flow_num = 0
    i = 0
    # print("zb-flow_pool_recall_process:" + str(flow_pool_recall_process))
    # print("zb-density_rules:" + str(density_rules))
    # print("zb-2:" + str([i for i in rov_recall_rank if i["videoId"] not in [j["videoId"] for j in rank_result[:size]]]))
    # print("zb-3:" + str([i for i in flow_recall_rank if i["videoId"] not in [j["videoId"] for j in rank_result[:size]]]))

    while i < size - top_K:
        # 随机生成[0, 1)浮点数
        rand = random.random()
        flow_pool_recall_process['flow_pool_P'] = flow_pool_P
        flow_pool_recall_process[f'{i}_rand'] = 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])
                # todo zhangbo rank
                result = merge_density_control(
                    rank_result[:size],
                    [i for i in rov_recall_rank if i["videoId"] not in [j["videoId"] for j in rank_result[:size]]],
                    [i for i in flow_recall_rank if i["videoId"] not in [j["videoId"] for j in rank_result[:size]]],
                    density_rules
                )
                return result, flow_num, flow_pool_recall_process
        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])
                # todo zhangbo rank
                result = merge_density_control(
                    rank_result[:size],
                    [i for i in rov_recall_rank if i["videoId"] not in [j["videoId"] for j in rank_result[:size]]],
                    [i for i in flow_recall_rank if i["videoId"] not in [j["videoId"] for j in rank_result[:size]]],
                    density_rules
                )
                return result, flow_num, flow_pool_recall_process
        i += 1
    # todo zhangbo rank
    result = merge_density_control(
        rank_result[:size],
        [i for i in rov_recall_rank if i["videoId"] not in [j["videoId"] for j in rank_result[:size]]],
        [i for i in flow_recall_rank if i["videoId"] not in [j["videoId"] for j in rank_result[:size]]],
        density_rules
    )
    return result, flow_num, flow_pool_recall_process
# 排序服务兜底
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_recall_freq and per_recall_name in recall_dict:
                per_recall = recall_dict[per_recall_name]
                #print("per_recall_item:", per_recall_item)
                cur_recall_num = 0
                for recall_item in per_recall:
                    vid = recall_item['videoId']
                    if vid in select_ids:
                        continue
                    rov_recall_rank.append(recall_item)
                    select_ids.add(vid)
                    cur_recall_num+=1
                    if cur_recall_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

def merge_density_control(
        data: list, rov: list, flow: list, rule: dict
) -> list:
    # 1 判断是否满足规则
    status_cur: Dict[str, int] = {}
    for d in data:
        if "tags" in d.keys() and len(d["tags"]) != 0:
            for t in d["tags"]:
                if t in rule.keys():
                    status_cur[t] = 1 + status_cur[t] if t in status_cur.keys() else 1
    status_cur_illegal: Dict[str, int] = {}
    for k, v in status_cur.items():
        if k in rule.keys() and rule[k] < v:
            status_cur_illegal[k] = v - rule[k]
    if len(status_cur_illegal) == 0:
        return data
    # 2 反向遍历，直到status_cur_illegal满足，记录要替换的index和召回池标记。
    indexes = []
    pushes = []
    var1 = len(data)
    for i in range(var1-1, -1, -1):
        d = data[i]
        tags = d["tags"] if "tags" in d.keys() else []
        inters = set(tags) & set(status_cur_illegal.keys())
        if len(inters) == 0:
            continue
        indexes.append(i)
        pushes.append(d["flowPool"] if "flowPool" in d.keys() and len(d["flowPool"]) != 0 else "")
        for inter in inters:
            status_cur_illegal[inter] = status_cur_illegal[inter] - 1
            if status_cur_illegal[inter] == 0:
                status_cur_illegal.pop(inter)
            status_cur[inter] = status_cur[inter] - 1
            if status_cur[inter] == 0:
                status_cur.pop(inter)
    # 3 反向遍历index，再正向遍历增补列表，取可替换的video
    for index, push in zip(reversed(indexes), reversed(pushes)):
        if len(push) > 0:
            # 5 如果是flow的video 取不到 不做替换
            candidate = flow
        else:
            # 5 如果是rov的video  取不到 不做替换
            candidate = rov
        for i, d in enumerate(candidate):
            judge_rule_set = judge_rule(rule=rule, status=status_cur)
            tags = set(d["tags"] if "tags" in d.keys() else [])
            if len(judge_rule_set & tags) != 0:
                continue
            # 开始插入
            tmp = copy.deepcopy(data[index])
            data[index] = d
            candidate[i] = tmp
            # 更新状态
            for tag in tags:
                status_cur[tag] = status_cur[tag] + 1 if tag in status_cur.keys() else 1
            break
        if len(push) > 0:
            flow = candidate
        else:
            rov = candidate
    return data
def judge_rule(rule: dict, status: dict) -> Set:
    result = set()
    for k, v in status.items():
        if k in rule.keys() and rule[k] <= v:
            result.add(k)
    return result
def get_video_tags(redis_helper, video_ids) -> dict:
    REDIS_PREFIX = "alg_recsys_video_tags_"
    redis_keys = [REDIS_PREFIX + str(i) for i in video_ids]
    video_tags = redis_helper.get_batch_key(redis_keys)
    video_tag_dict = {}
    if video_tags is not None:
        for i, tags_str in enumerate(video_tags):
            tags = []
            if tags_str is not None and len(tags_str) != 0:
                tags = str(tags_str).split(",")
            video_tag_dict[video_ids[i]] = tags
    return video_tag_dict

if __name__ == '__main__':
    data: list = [
        {'videoId': 1, 'flowPool': '', 'tags': ['下午好','元旦','祝福']},
        {'videoId': 2, 'flowPool': '', 'tags': ['下午好','祝福']},
        {'videoId': 3, 'flowPool': '', 'tags': ['早上好']},
        {'videoId': 4, 'flowPool': 'flow', 'tags': ['下午好','元旦','祝福']},
    ]
    rov = [
        {'videoId': 10, 'flowPool': '', 'tags': ['下午好']},
        {'videoId': 11, 'flowPool': '', 'tags': ['祝福']},
        {'videoId': 12, 'flowPool': '', 'tags': ['下午好','元旦','祝福']},
        {'videoId': 13, 'flowPool': '', 'tags': ['元旦', "下午好"]},
        # {'videoId': 14, 'flowPool': '', 'tags': []},
        # {'videoId': 15, 'flowPool': '', 'tags': []}
    ]
    flow = [
        {'videoId': 20, 'flowPool': 'flow', 'tags': ['下午好']},
        {'videoId': 21, 'flowPool': 'flow', 'tags': ['下午好']},
        {'videoId': 22, 'flowPool': 'flow', 'tags': []},
        {'videoId': 23, 'flowPool': 'flow', 'tags': []}
    ]
    rule = {
        '下午好': 2,
        '早上好': 1,
        '祝福': 1
    }
    result = merge_density_control(data, rov, flow, rule)
    print(result)

    # 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)
    pass