Merge branch 'ab-test-339-update-2023052622' into dev

config.py

@@ -148,6 +148,16 @@ class BaseConfig(object):
             'abtest_320': 60049,
             'abtest_322': 60050,
             'abtest_323': 60051,
+            'abtest_326': 60052,
+            'abtest_327': 60053,
+            'abtest_328': 60054,
+            'abtest_332': 60055,
+            'abtest_333': 60056,
+            'abtest_334': 60057,
+            'abtest_335': 60058,
+            'abtest_336': 60059,
+            'abtest_337': 60060,
+            'abtest_338': 60061,
         },  # 地域分组小时级规则实验
 
         'rank_by_24h': {
@@ -339,6 +349,46 @@ class BaseConfig(object):
         '323': {
             'data_key': 'data10', 'rule_key': 'rule7',
             'ab_code': AB_CODE['region_rank_by_h'].get('abtest_323')
+        },
+        '326': {
+            'data_key': 'data10', 'rule_key': 'rule7',
+            'ab_code': AB_CODE['region_rank_by_h'].get('abtest_326')
+        },
+        '327': {
+            'data_key': 'data10', 'rule_key': 'rule7',
+            'ab_code': AB_CODE['region_rank_by_h'].get('abtest_327')
+        },
+        '328': {
+            'data_key': 'data10', 'rule_key': 'rule7',
+            'ab_code': AB_CODE['region_rank_by_h'].get('abtest_328')
+        },
+        '332': {
+            'data_key': 'data10', 'rule_key': 'rule7',
+            'ab_code': AB_CODE['region_rank_by_h'].get('abtest_332')
+        },
+        '333': {
+            'data_key': 'data10', 'rule_key': 'rule7',
+            'ab_code': AB_CODE['region_rank_by_h'].get('abtest_333')
+        },
+        '334': {
+            'data_key': 'data10', 'rule_key': 'rule7',
+            'ab_code': AB_CODE['region_rank_by_h'].get('abtest_334')
+        },
+        '335': {
+            'data_key': 'data10', 'rule_key': 'rule7',
+            'ab_code': AB_CODE['region_rank_by_h'].get('abtest_335')
+        },
+        '336': {
+            'data_key': 'data10', 'rule_key': 'rule7',
+            'ab_code': AB_CODE['region_rank_by_h'].get('abtest_336')
+        },
+        '337': {
+            'data_key': 'data10', 'rule_key': 'rule7',
+            'ab_code': AB_CODE['region_rank_by_h'].get('abtest_337')
+        },
+        '338': {
+            'data_key': 'data10', 'rule_key': 'rule7',
+            'ab_code': AB_CODE['region_rank_by_h'].get('abtest_338')
         }
     }
 
@@ -417,7 +467,9 @@ class BaseConfig(object):
         'sim_hot_vid_recall': 'sim_hot_vid_recall',  # 相似视频召回
         'fast_flow_recall': 'fast_flow_recall', #快速流量池召回
         'normal_flow_recall': 'normal_flow_recall',  # 普通流量池召回
-
+        'hot_3_day_recall': 'hot_3_day_recall',  # 3天召回
+        'hot_recall': 'hot_recall',  # hot召回
+        'w2v_recall': 'w2v_recall', #word2vec
     }
 
     # category id mapping

recommend.py

@@ -10,7 +10,7 @@ import config
 from log import Log
 from config import set_config
 from video_recall import PoolRecall
-from video_rank import video_new_rank,video_rank,refactor_video_rank, bottom_strategy, video_rank_by_w_h_rate, video_rank_with_old_video, bottom_strategy2
+from video_rank import video_new_rank2, video_sanke_rank,video_new_rank,video_rank,refactor_video_rank, bottom_strategy, video_rank_by_w_h_rate, video_rank_with_old_video, bottom_strategy2
 from db_helper import RedisHelper
 import gevent
 from utils import FilterVideos, get_user_has30day_return
@@ -197,10 +197,22 @@ def video_recommend(request_id, mid, uid, size, top_K, flow_pool_P, app_type, al
 
     if app_type in [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]:
         t = [gevent.spawn(pool_recall.rov_pool_recall_with_region, size, expire_time)]
+        if ab_code==60058:
+            t.append(gevent.spawn(pool_recall.get_hot_item_reall))
+        elif  ab_code==60059:
+            t.append(gevent.spawn(pool_recall.get_word2vec_item_reall))
+        elif  ab_code==60061:
+            t.append(gevent.spawn(pool_recall.get_sim_hot_item_reall_filter))
     else:
         t = [gevent.spawn(pool_recall.rov_pool_recall_with_region, size, expire_time),
              gevent.spawn(pool_recall.flow_pool_recall, size, config_.QUICK_FLOW_POOL_ID),
              gevent.spawn(pool_recall.flow_pool_recall, size)]
+        if ab_code==60058:
+            t.append(gevent.spawn(pool_recall.get_hot_item_reall))
+        elif ab_code == 60059:
+            t.append(gevent.spawn(pool_recall.get_word2vec_item_reall))
+        elif ab_code == 60061:
+            t.append(gevent.spawn(pool_recall.get_sim_hot_item_reall_filter))
 
     # 最惊奇相关推荐实验
     # elif ab_code == config_.AB_CODE['top_video_relevant_appType_19']:
@@ -281,10 +293,40 @@ def video_recommend(request_id, mid, uid, size, top_K, flow_pool_P, app_type, al
                 'rov_pool_recall': recall_result_list[0],
                 'flow_pool_recall': recall_result_list[2]
             }
+
+    if ab_code == 60058:
+        if len(recall_result_list)>=4:
+            data['hot_recall'] = recall_result_list[3]
+        else:
+            data['hot_recall'] = []
+    elif ab_code == 60059:
+        if len(recall_result_list)>=4:
+            data['w2v_recall'] = recall_result_list[3]
+        else:
+            data['w2v_recall'] = []
+    elif ab_code == 60061:
+        if len(recall_result_list)>=4:
+            data['sim_recall'] = recall_result_list[3]
+        else:
+            data['sim_recall'] = []
     #if ab_code=="ab_new_test":
     #    rank_result = video_new_rank(data=data, size=size, top_K=top_K, flow_pool_P=float(flow_pool_P))
     #else:
-    rank_result = video_rank(data=data, size=size, top_K=top_K, flow_pool_P=float(flow_pool_P))
+    #print("data['hot_recall']", data['hot_recall'])
+    if ab_code == 60058 or ab_code == 60059 or ab_code == 60060 or ab_code == 60061 :
+        test_config = None
+        if ab_code == 60058:
+            test_config =pool_recall.get_hotrecall_config()
+        elif ab_code == 60059:
+            test_config = pool_recall.get_w2v_config()
+        elif ab_code == 60060:
+            test_config = pool_recall.get_test_config()
+        elif ab_code == 60061:
+            test_config = pool_recall.get_simrecall_config()
+        #print("test config:",test_config)
+        rank_result = video_sanke_rank(data=data, size=size, top_K=top_K, flow_pool_P=float(flow_pool_P), ab_Code=ab_code, exp_config=test_config)
+    else:
+        rank_result = video_rank(data=data, size=size, top_K=top_K, flow_pool_P=float(flow_pool_P))
 
     # 老视频实验
     # if ab_code in [config_.AB_CODE['old_video']]:
@@ -336,6 +378,199 @@ def video_recommend(request_id, mid, uid, size, top_K, flow_pool_P, app_type, al
     #
     # result['rankResult'] = rank_result
 
+    return result
+    # return rank_result, last_rov_recall_key
+
+def video_old_recommend(request_id, mid, uid, size, top_K, flow_pool_P, app_type, algo_type, client_info,
+                    expire_time=24*3600, ab_code=config_.AB_CODE['initial'], rule_key='', data_key='',
+                    no_op_flag=False, old_video_index=-1, video_id=None, params=None, rule_key_30day=None,
+                    shield_config=None):
+    """
+    首页线上推荐逻辑
+    :param request_id: request_id
+    :param mid: mid type-string
+    :param uid: uid type-string
+    :param size: 请求视频数量 type-int
+    :param top_K: 保证topK为召回池视频 type-int
+    :param flow_pool_P: size-top_K视频为流量池视频的概率 type-float
+    :param app_type: 产品标识  type-int
+    :param algo_type: 算法类型  type-string
+    :param client_info: 用户位置信息 {"country": "国家",  "province": "省份",  "city": "城市"}
+    :param expire_time: 末位视频记录redis过期时间
+    :param ab_code: AB实验code
+    :param video_id: 相关推荐头部视频id
+    :param params:
+    :return:
+    """
+    result = {}
+    # ####### 多进程召回
+    start_recall = time.time()
+    # log_.info('====== recall')
+
+    #print("abcode",ab_code)
+    recall_result_list = []
+    pool_recall = PoolRecall(request_id=request_id,
+                             app_type=app_type, mid=mid, uid=uid, ab_code=ab_code,
+                             client_info=client_info, rule_key=rule_key, data_key=data_key, no_op_flag=no_op_flag,
+                             params=params, rule_key_30day=rule_key_30day, shield_config=shield_config, video_id= video_id)
+
+    if app_type in [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]:
+        t = [gevent.spawn(pool_recall.rov_pool_recall_with_region, size, expire_time)]
+        if ab_code ==60054:
+            t.append(gevent.spawn(pool_recall.get_sim_hot_item_reall_filter))
+        if ab_code == 60055:
+            t.append(gevent.spawn(pool_recall.get_3days_hot_item_reall))
+        if ab_code == 60056:
+            t.append(gevent.spawn(pool_recall.get_hot_item_reall))
+    else:
+        t = [gevent.spawn(pool_recall.rov_pool_recall_with_region, size, expire_time),
+             gevent.spawn(pool_recall.flow_pool_recall, size, config_.QUICK_FLOW_POOL_ID),
+             gevent.spawn(pool_recall.flow_pool_recall, size)]
+        if  ab_code == 60054:
+            t.append(gevent.spawn(pool_recall.get_sim_hot_item_reall_filter))
+        if ab_code == 60055:
+            t.append(gevent.spawn(pool_recall.get_3days_hot_item_reall))
+        if ab_code == 60056:
+            t.append(gevent.spawn(pool_recall.get_hot_item_reall))
+
+    gevent.joinall(t)
+    recall_result_list = [i.get() for i in t]
+
+    #print("recall:",recall_result_list)
+    if len(recall_result_list)<0:
+        result['recallResult']= []
+        result['rankResult'] = []
+        return result
+    #1. merge simrecall
+    if ab_code == 60054:
+        rov_pool_recall = []
+        if len(recall_result_list) >= 2:
+            region_recall = recall_result_list[0]
+            sim_recall = []
+            if app_type in [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]:
+                sim_recall = recall_result_list[1]
+            else:
+                if len(recall_result_list)>=4:
+                    sim_recall = recall_result_list[3]
+            now_video_ids = set('')
+            if len(region_recall) > 0:
+                for video in region_recall:
+                    video_id = video.get('videoId')
+                    if video_id not in now_video_ids:
+                        rov_pool_recall.append(video)
+                        now_video_ids.add(video_id)
+            if len(sim_recall) > 0:
+                for video in sim_recall:
+                    video_id = video.get('videoId')
+                    # print("sim video_id:", video_id)
+                    if video_id not in now_video_ids:
+                        rov_pool_recall.append(video)
+                        now_video_ids.add(video_id)
+            if len(rov_pool_recall) > 0:
+                recall_result_list[0] = rov_pool_recall
+    # merge hot 3day recall
+    if ab_code == 60055:
+        rov_pool_recall = []
+        if len(recall_result_list)>=2:
+            region_recall = recall_result_list[0]
+            hot_3_day_recall = []
+            if app_type in [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]:
+                hot_3_day_recall = recall_result_list[1]
+            else:
+                if len(recall_result_list)>=4:
+                    hot_3_day_recall = recall_result_list[3]
+            #print("sim:",sim_recall)
+            now_video_ids = set('')
+            if len(region_recall)>0:
+                for video in region_recall:
+                    video_id = video.get('videoId')
+                    if video_id not in now_video_ids:
+                        rov_pool_recall.append(video)
+                        now_video_ids.add(video_id)
+            if len(hot_3_day_recall)>0:
+                for video in hot_3_day_recall:
+                    video_id = video.get('videoId')
+                    #print("sim video_id:", video_id)
+                    if video_id not in now_video_ids:
+                        rov_pool_recall.append(video)
+                        now_video_ids.add(video_id)
+            if len(rov_pool_recall)>0:
+                recall_result_list[0] = rov_pool_recall
+    # merge old day recall
+    if ab_code == 60056:
+        rov_pool_recall = []
+        if len(recall_result_list)>=2:
+            region_recall = recall_result_list[0]
+            hot_7_day_recall = []
+            if app_type in [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]:
+                hot_7_day_recall = recall_result_list[1]
+            else:
+                if len(recall_result_list) >= 3:
+                    hot_7_day_recall = recall_result_list[3]
+            now_video_ids = set('')
+            if len(region_recall)>0:
+                for video in region_recall:
+                    video_id = video.get('videoId')
+                    if video_id not in now_video_ids:
+                        rov_pool_recall.append(video)
+                        now_video_ids.add(video_id)
+            if len(hot_7_day_recall)>0:
+                for video in hot_7_day_recall:
+                    video_id = video.get('videoId')
+                    #print("sim video_id:", video_id)
+                    if video_id not in now_video_ids:
+                        rov_pool_recall.append(video)
+                        now_video_ids.add(video_id)
+            if len(rov_pool_recall)>0:
+                recall_result_list[0] = rov_pool_recall
+
+    result['recallResult'] = recall_result_list
+    result['recallTime'] = (time.time() - start_recall) * 1000
+    #print("recall:", recall_result_list)
+    # ####### 排序
+    start_rank = time.time()
+    # log_.info('====== rank')
+    if app_type in [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]:
+        if ab_code in [
+            config_.AB_CODE['rov_rank_appType_18_19'],
+            config_.AB_CODE['rov_rank_appType_19'],
+            config_.AB_CODE['top_video_relevant_appType_19']
+        ]:
+            data = {
+                'rov_pool_recall': recall_result_list[0],
+                'flow_pool_recall': recall_result_list[1]
+            }
+        else:
+            data = {
+                'rov_pool_recall': recall_result_list[0],
+                'flow_pool_recall': []
+            }
+    else:
+        if recall_result_list[1]:
+            redis_helper = RedisHelper()
+            quick_flow_pool_P = redis_helper.get_data_from_redis(
+                key_name=f"{config_.QUICK_FLOWPOOL_DISTRIBUTE_RATE_KEY_NAME_PREFIX}{config_.QUICK_FLOW_POOL_ID}"
+            )
+            if quick_flow_pool_P:
+                flow_pool_P = quick_flow_pool_P
+            data = {
+                'rov_pool_recall': recall_result_list[0],
+                'flow_pool_recall': recall_result_list[1]
+            }
+        else:
+            data = {
+                'rov_pool_recall': recall_result_list[0],
+                'flow_pool_recall': recall_result_list[2]
+            }
+    #if ab_code=="ab_new_test":
+    #print("before data:", data)
+    rank_result = video_new_rank2(data=data, size=size, top_K=top_K, flow_pool_P=float(flow_pool_P), ab_code=ab_code)
+    #print(rank_result)
+
+    result['rankResult'] = rank_result
+    result['rankTime'] = (time.time() - start_rank) * 1000
+
+
     return result
     # return rank_result, last_rov_recall_key
 
@@ -394,6 +629,8 @@ def new_video_recommend(request_id, mid, uid, size, top_K, flow_pool_P, app_type
              gevent.spawn(pool_recall.get_selected_recall, size, region_code),
              gevent.spawn(pool_recall.get_no_selected_recall, size, region_code)
              ]
+        if ab_code == 60049:
+            t.append(gevent.spawn(pool_recall.get_sim_hot_item_reall))
     else:
         t = [
              gevent.spawn(pool_recall.get_region_hour_recall, size, region_code),
@@ -884,6 +1121,8 @@ def get_recommend_params(recommend_type, ab_exp_info, ab_info_data, mid, app_typ
                 flow_pool_P = 0.4
             elif '301' in ab_exp_code_list:
                 flow_pool_P = 0.6
+            elif '339' in ab_exp_code_list:
+                flow_pool_P = 0
 
             # if '136' in ab_exp_code_list:
             #     # 无回流 - 消费人群
@@ -1333,6 +1572,7 @@ def video_homepage_recommend(request_id, mid, uid, size, app_type, algo_type,
     # 简单召回 - 排序 - 兜底
     get_result_st = time.time()
     #print("ab_code:", ab_code)
+    #new pipeline
     if ab_code == 60047 or ab_code == 60048 or ab_code == 60049:
         result = new_video_recommend(request_id=request_id,
                              mid=mid, uid=uid, app_type=app_type,
@@ -1342,6 +1582,7 @@ def video_homepage_recommend(request_id, mid, uid, size, app_type, algo_type,
                              rule_key=rule_key, data_key=data_key,
                              no_op_flag=no_op_flag, old_video_index=old_video_index,
                              params=params, rule_key_30day=rule_key_30day, shield_config=shield_config)
+    # old base (test4, test5)
     elif ab_code == 60050 or ab_code == 60051:
         result = video_recommend(request_id=request_id,
                                  mid=mid, uid=uid, app_type=app_type,
@@ -1351,6 +1592,19 @@ def video_homepage_recommend(request_id, mid, uid, size, app_type, algo_type,
                                  rule_key=rule_key, data_key=data_key,
                                  no_op_flag=no_op_flag, old_video_index=old_video_index,
                                  params=params, rule_key_30day=rule_key_30day, shield_config=shield_config)
+    # 60052,60053,60057(test7, test8, test6)
+    # simrecal: 60054
+    # 3 days , 7days recall: 60005, test1
+    # old video: 60056, test2
+    elif ab_code == 60052 or ab_code == 60053 or ab_code == 60054 or ab_code == 60055 or ab_code == 60056 or ab_code==60057:
+        result = video_old_recommend(request_id=request_id,
+                                     mid=mid, uid=uid, app_type=app_type,
+                                     size=size, top_K=top_K, flow_pool_P=flow_pool_P,
+                                     algo_type='', client_info=client_info,
+                                     ab_code=ab_code, expire_time=expire_time,
+                                     rule_key=rule_key, data_key=data_key, no_op_flag=no_op_flag,
+                                     old_video_index=old_video_index, video_id= None,
+                                     params=params, rule_key_30day=rule_key_30day, shield_config=shield_config)
     else:
         result = video_recommend(request_id=request_id,
                              mid=mid, uid=uid, app_type=app_type,
@@ -1384,6 +1638,9 @@ def video_homepage_recommend(request_id, mid, uid, size, app_type, algo_type,
         update_flow_redis_data(result=rank_result, app_type=app_type, mid=mid, top_K=top_K)
     elif ab_code == 60050 or  ab_code == 60051:
         update_redis_data(result=rank_result, app_type=app_type, mid=mid, top_K=top_K)
+    elif ab_code == 60052 or ab_code == 60053 or ab_code == 60054 or ab_code == 60055 \
+            or ab_code == 60056 or ab_code==60057:
+        update_redis_data(result=rank_result, app_type=app_type, mid=mid, top_K=top_K)
     else:
         update_redis_data(result=rank_result, app_type=app_type, mid=mid, top_K=top_K)
     # log_.info({
@@ -1466,6 +1723,16 @@ def video_relevant_recommend(request_id, video_id, mid, uid, size, app_type, ab_
                                  rule_key=rule_key, data_key=data_key, no_op_flag=no_op_flag,
                                  old_video_index=old_video_index, video_id=video_id,
                                  params=params, rule_key_30day=rule_key_30day, shield_config=shield_config)
+    elif ab_code == 60052 or ab_code == 60053 or ab_code == 60054 or ab_code == 60055 or \
+            ab_code == 60056 or ab_code==60057:
+        result = video_old_recommend(request_id=request_id,
+                                 mid=mid, uid=uid, app_type=app_type,
+                                 size=size, top_K=top_K, flow_pool_P=flow_pool_P,
+                                 algo_type='', client_info=client_info,
+                                 ab_code=ab_code, expire_time=expire_time,
+                                 rule_key=rule_key, data_key=data_key, no_op_flag=no_op_flag,
+                                 old_video_index=old_video_index, video_id=video_id,
+                                 params=params, rule_key_30day=rule_key_30day, shield_config=shield_config)
     else:
         result = video_recommend(request_id=request_id,
                              mid=mid, uid=uid, app_type=app_type,
@@ -1497,11 +1764,14 @@ def video_relevant_recommend(request_id, video_id, mid, uid, size, app_type, ab_
     # redis数据刷新
     update_redis_st = time.time()
     if ab_code == 60047 or ab_code == 60048 or  ab_code == 60049:
-        update_flow_redis_data(result=rank_result, app_type=app_type, mid=mid, top_K=top_K)
+         update_flow_redis_data(result=rank_result, app_type=app_type, mid=mid, top_K=top_K)
     elif ab_code == 60050 or ab_code == 60051:
-        update_redis_data(result=rank_result, app_type=app_type, mid=mid, top_K=top_K)
+         update_redis_data(result=rank_result, app_type=app_type, mid=mid, top_K=top_K)
+    elif ab_code == 60052 or ab_code == 60053 or ab_code == 60054 or ab_code == 60055 or \
+            ab_code == 60056 or ab_code==60057:
+         update_redis_data(result=rank_result, app_type=app_type, mid=mid, top_K=top_K)
     else:
-        update_redis_data(result=rank_result, app_type=app_type, mid=mid, top_K=top_K)
+         update_redis_data(result=rank_result, app_type=app_type, mid=mid, top_K=top_K)
     # log_.info({
     #     'logTimestamp': int(time.time() * 1000),
     #     'request_id': request_id,

video_rank.py

@@ -592,6 +592,219 @@ def video_rank_with_old_video(rank_result, old_video_recall, size, top_K, old_vi
     return new_rank_result[:size]
 
 
+def video_new_rank2(data, size, top_K, flow_pool_P, ab_code):
+    """
+        视频分发排序
+        :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()
+    vidKeys = []
+    recall_list = []
+    pre_str = "k_p2:"
+    if ab_code == 60052 or ab_code == 60053  or ab_code==60057:
+        pre_str = "k_p2:"
+    elif ab_code == 60054:
+        pre_str = "k_p3:"
+    elif ab_code == 60055:
+        pre_str = "k_p4:"
+    elif ab_code == 60056:
+        pre_str = "k_p5:"
+    #print("pre_str:", pre_str)
+    for recall_item in data['rov_pool_recall']:
+        if len(recall_item)<=0:
+            continue
+        vid = recall_item.get("videoId",0)
+        vidKeys.append(pre_str+ str(vid))
+        recall_list.append(recall_item)
+    #print("vidKeys:", vidKeys)
+    video_scores = redisObj.get_batch_key(vidKeys)
+    #print("video_score:",video_scores)
+    for i in range(len(video_scores)):
+         try:
+            # print(video_scores[i])
+            if video_scores[i] is None:
+                recall_list[i]['sort_score']= 0.0
+            else:
+                video_score_str = json.loads(video_scores[i])
+                #print("video_score_str:", video_score_str)
+                recall_list[i]['sort_score'] = video_score_str[0]
+         except Exception :
+                recall_list[i]['sort_score'] = 0.0
+    #sort_items = sorted(video_items, key=lambda k: k[1], reverse=True)
+    rov_recall_rank =sorted(recall_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 个视频
+    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_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 not data['hot_rcall'] and not data['hot_rcall']:
+        return []
+    # 地域分组小时级规则更新数据
+    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 = []
+    if ab_Code==60058:
+        if len(data['hot_recall'])>0:
+            hot_recall = sorted(data['hot_recall'], key=lambda k: k.get('rovScore', 0), reverse=True)
+        recall_dict['hot_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:
+        if len(data['sim_recall'])>0:
+            recall_dict['sim_recall'] = data['sim_recall']
+        else:
+            recall_dict['sim_recall'] = sim_recall
+
+    recall_list = [('rov_recall_region_h',1, 1),('rov_recall_region_h',0.5, 1),('rov_recall_region_24h',1,1),
+                   ('hot_recall',0.5,1), ('w2v_recall',0.5,1),('rov_recall_24h',1,1), ('rov_recall_24h_dup',0.5,1)]
+    if exp_config is not  None:
+        recall_list = exp_config
+    #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:]
+
+    # 按概率 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]
+
+
+
 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},

video_recall.py

@@ -2124,6 +2124,43 @@ class PoolRecall(object):
                 )
         return recall_result[:200]
 
+    def get_sim_hot_item_reall_filter(self):
+        if self.video_id is None:
+            return  []
+        recall_key = "sim_hot_" + str(self.video_id)
+        #print("recall_key:", recall_key)
+        data = self.redis_helper.get_data_from_redis(key_name=recall_key)
+
+        #print(data)
+        recall_result = []
+        recall_dict  = {}
+        video_ids = []
+        if data is not None:
+            json_result = json.loads(data)
+            # print("json_result:", json_result)
+            for per_item in json_result:
+                try:
+                    vid = int(per_item[0])
+                    video_ids.append(vid)
+                    recall_dict[vid] = {'videoId': vid, 'flowPool': '',
+                         'rovScore': per_item[1], 'pushFrom': config_.PUSH_FROM['sim_hot_vid_recall'],
+                         'abCode': self.ab_code}
+                except Exception as e:
+                    continue
+        if len(video_ids)<=0:
+            return  recall_result
+        video_ids = video_ids[:50]
+        #print(video_ids)
+        filter_ = FilterVideos(request_id=self.request_id,
+                               app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
+        filtered_viewed_videos = filter_.filter_videos(pool_type='rov')
+        if filtered_viewed_videos is None:
+            return recall_result
+        print("filtered_viewed_videos:", filtered_viewed_videos)
+        for vid in filtered_viewed_videos:
+            if vid in recall_dict:
+                recall_result.append(recall_dict[vid])
+        return recall_result[:30]
     # get region_hour_recall
     def get_region_hour_recall(self, size=4, region_code='-1'):
         pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H
@@ -2141,7 +2178,7 @@ class PoolRecall(object):
                      'rovScore': per_item[1], 'pushFrom': config_.PUSH_FROM['rov_recall_region_h'],
                      'abCode': self.ab_code}
                 )
-        return recall_result[:200]
+        return recall_result[:30]
 
     # get region_day_recall
     def get_region_day_recall(self, size=4,region_code='-1'):
@@ -2325,4 +2362,215 @@ class PoolRecall(object):
         #     'executeTime': (time.time() - start_time) * 1000
         # })
 
-        return flow_pool_recall_result[:size]
+        return flow_pool_recall_result[:size]
+
+    def get_3days_hot_item_reall(self):
+        recall_key = "hot_3day:"
+        #print("recall_key:", recall_key)
+        data = self.redis_helper.get_data_from_redis(key_name=recall_key)
+
+        #print(data)
+        recall_result = []
+        recall_dict  = {}
+        video_ids = []
+        if data is not None:
+            json_result = json.loads(data)
+            #print("json_result:", json_result)
+            for per_item in json_result:
+                try:
+                    vid = int(per_item[0])
+                    video_ids.append(vid)
+                    recall_dict[vid] = {'videoId': vid, 'flowPool': '',
+                         'rovScore': per_item[1], 'pushFrom': config_.PUSH_FROM['hot_3_day_recall'],
+                         'abCode': self.ab_code}
+                except Exception as e:
+                    continue
+        #print("vid len:", len(video_ids))
+        if len(video_ids)<=0:
+            return  recall_result
+        video_ids = video_ids[:30]
+        #print(video_ids)
+        filter_ = FilterVideos(request_id=self.request_id,
+                               app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
+        filtered_viewed_videos = filter_.filter_videos(pool_type='rov')
+        if filtered_viewed_videos is None:
+            return recall_result
+        #print("filtered_viewed_videos:", filtered_viewed_videos)
+        for vid in filtered_viewed_videos:
+            if vid in recall_dict:
+                recall_result.append(recall_dict[vid])
+        #print("hot recall_result:", recall_result)
+        #print("recall_dict:", recall_dict)
+        return recall_result
+
+    def get_hot_item_reall(self):
+        #recall_key = "hot_video:"
+        recall_key = "hot_video:"
+        #print("recall_key:", recall_key)
+        data = self.redis_helper.get_data_from_redis(key_name=recall_key)
+
+        #print(data)
+        recall_result = []
+        recall_dict  = {}
+        video_ids = []
+        if data is not None:
+            json_result = json.loads(data)
+            # print("json_result:", json_result)
+            for per_item in json_result:
+                try:
+                    vid = int(per_item[0])
+                    video_ids.append(vid)
+                    recall_dict[vid] = {'videoId': vid, 'flowPool': '',
+                         'rovScore': per_item[1], 'pushFrom': config_.PUSH_FROM['hot_recall'],
+                         'abCode': self.ab_code}
+                except Exception as e:
+                    continue
+        if len(video_ids)<=0:
+            return  recall_result
+        video_ids = video_ids[:30]
+        #print(video_ids)
+        filter_ = FilterVideos(request_id=self.request_id,
+                               app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
+        filtered_viewed_videos = filter_.filter_videos(pool_type='rov')
+        if filtered_viewed_videos is None:
+            return recall_result
+        #print("filtered_viewed_videos:", filtered_viewed_videos)
+        for vid in filtered_viewed_videos:
+            if vid in recall_dict:
+                recall_result.append(recall_dict[vid])
+        #print("hot recall_result:", recall_result)
+        return recall_result
+
+    def get_title_recall(self):
+        if self.video_id is None:
+            return []
+        recall_key = "title_I2I:" + str(self.video_id)
+        # print("recall_key:", recall_key)
+        data = self.redis_helper.get_data_from_redis(key_name=recall_key)
+
+        # print(data)
+        recall_result = []
+        recall_dict = {}
+        video_ids = []
+        if data is not None:
+            json_result = json.loads(data)
+            # print("json_result:", json_result)
+            for per_item in json_result:
+                try:
+                    vid = int(per_item[0])
+                    video_ids.append(vid)
+                    recall_dict[vid] = {'videoId': vid, 'flowPool': '',
+                                        'rovScore': per_item[1], 'pushFrom': config_.PUSH_FROM['title_i2i_recall'],
+                                        'abCode': self.ab_code}
+                except Exception as e:
+                    continue
+        if len(video_ids) <= 0:
+            return recall_result
+        video_ids = video_ids[:50]
+        # print(video_ids)
+        filter_ = FilterVideos(request_id=self.request_id,
+                               app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
+        filtered_viewed_videos = filter_.filter_videos(pool_type='rov')
+        if filtered_viewed_videos is None:
+            return recall_result
+        # print("filtered_viewed_videos:", filtered_viewed_videos)
+        for vid in filtered_viewed_videos:
+            if vid in recall_dict:
+                recall_result.append(recall_dict[vid])
+        return recall_result
+
+    def get_word2vec_item_reall(self):
+        if self.video_id is None:
+            return  []
+        recall_key = "w2v:" + str(self.video_id)
+        #print("recall_key:", recall_key)
+        data = self.redis_helper.get_data_from_redis(key_name=recall_key)
+
+        #print(data)
+        recall_result = []
+        recall_dict  = {}
+        video_ids = []
+        if data is not None and data.strip()!="":
+            json_result = data.strip().split(",")
+            # print("json_result:", json_result)
+            for per_item in json_result:
+                try:
+                    vid = int(per_item)
+                    video_ids.append(vid)
+                    recall_dict[vid] = {'videoId': vid, 'flowPool': '',
+                         'rovScore': 0.0, 'pushFrom': config_.PUSH_FROM['w2v_recall'],
+                         'abCode': self.ab_code}
+                except Exception as e:
+                    continue
+        if len(video_ids)<=0:
+            return  recall_result
+        video_ids = video_ids[:50]
+        #print(video_ids)
+        filter_ = FilterVideos(request_id=self.request_id,
+                               app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
+        filtered_viewed_videos = filter_.filter_videos(pool_type='rov')
+        if filtered_viewed_videos is None:
+            return recall_result
+        #print("filtered_viewed_videos:", filtered_viewed_videos)
+        for vid in filtered_viewed_videos:
+            if vid in recall_dict:
+                recall_result.append(recall_dict[vid])
+        return recall_result[:30]
+
+
+    def get_test_config(self):
+        recall_key = "test_config"
+        # print("recall_key:", recall_key)
+        data = self.redis_helper.get_data_from_redis(key_name=recall_key)
+        if data is not None:
+            try:
+                #print(data)
+                json_result = json.loads(data)
+                #print(json_result)
+                return json_result
+            except Exception as e:
+                return None
+        else:
+            return None
+
+    def get_w2v_config(self):
+        recall_key = "w2v_config"
+        data = self.redis_helper.get_data_from_redis(key_name=recall_key)
+        if data is not None:
+            try:
+                #print(data)
+                json_result = json.loads(data)
+                #print(json_result)
+                return json_result
+            except Exception as e:
+                return None
+        else:
+            return None
+
+    def get_hotrecall_config(self):
+        recall_key = "ht_config"
+        data = self.redis_helper.get_data_from_redis(key_name=recall_key)
+        if data is not None:
+            try:
+                # print(data)
+                json_result = json.loads(data)
+                # print(json_result)
+                return json_result
+            except Exception as e:
+                return None
+        else:
+            return None
+
+    def get_simrecall_config(self):
+        recall_key = "simrecall_config"
+        data = self.redis_helper.get_data_from_redis(key_name=recall_key)
+        if data is not None:
+            try:
+                # print(data)
+                json_result = json.loads(data)
+                # print(json_result)
+                return json_result
+            except Exception as e:
+                return None
+        else:
+            return None