refactor recall

config.py

@@ -387,6 +387,9 @@ class BaseConfig(object):
         'special_mid': 'special_mid_videos',  # 特殊mid指定视频
         'rov_recall_30day': 'recall_pool_30day',  # 天级更新相对30天列表
         'sim_hot_vid_recall': 'sim_hot_vid_recall',  # 相似视频召回
+        'fast_flow_recall': 'fast_flow_recall', #快速流量池召回
+        'normal_flow_recall': 'normal_flow_recall',  # 普通流量池召回
+
     }
 
     # 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, bottom_strategy, video_rank_by_w_h_rate, video_rank_with_old_video, bottom_strategy2
+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 db_helper import RedisHelper
 import gevent
 from utils import FilterVideos, get_user_has30day_return
@@ -201,7 +201,7 @@ def video_recommend(request_id, mid, uid, size, top_K, flow_pool_P, app_type, al
         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),
-             gevent.spawn(pool_recall.get_sim_hot_item_reall, size)]
+             gevent.spawn(pool_recall.get_sim_hot_item_reall)]
 
     # 最惊奇相关推荐实验
     # elif ab_code == config_.AB_CODE['top_video_relevant_appType_19']:
@@ -340,6 +340,176 @@ def video_recommend(request_id, mid, uid, size, top_K, flow_pool_P, app_type, al
     return result
     # return rank_result, last_rov_recall_key
 
+def new_video_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:
+    """
+    #1. recall
+    result = {}
+    # ####### 多进程召回
+    start_recall = time.time()
+
+    # 1. 根据城市或者省份获取region_code
+    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'
+
+    size =1000
+    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.get_region_hour_recall, size, region_code),
+             gevent.spawn(pool_recall.get_region_day_recall, size, region_code),
+             gevent.spawn(pool_recall.get_selected_recall, size),
+             gevent.spawn(pool_recall.get_no_selected_recall, size)
+             ]
+    else:
+        t = [gevent.spawn(pool_recall.get_region_hour_recall, size),
+             gevent.spawn(pool_recall.get_region_day_recall, size),
+             gevent.spawn(pool_recall.get_selected_recall, size),
+             gevent.spawn(pool_recall.get_no_selected_recall, size),
+             gevent.spawn(pool_recall.get_fast_flow_pool_recall, size),
+             gevent.spawn(pool_recall.get_flow_pool_recall, size),
+             gevent.spawn(pool_recall.get_sim_hot_item_reall)]
+    gevent.joinall(t)
+    # all recall_result
+    all_recall_result_list = [i.get() for i in t]
+    result['recallTime'] = (time.time() - start_recall) * 1000
+
+    #2. duplicate
+    recall_dict = {}
+    fast_flow_set = set('')
+    flow_flow_set = set('')
+    all_flow_set = set('')
+    region_h_recall = []
+    region_day_recall = []
+    select_day_recall = []
+    no_selected_recall  = []
+    for per_item in all_recall_result_list:
+        vId = per_item.get("videoId",'0')
+        if vId=='0':
+            continue
+        recall_name = per_item.get("pushFrom",'')
+        if recall_name=='fast_flow_recall':
+            fast_flow_set.add(vId)
+        if recall_name=='flow_recall':
+            flow_flow_set.add(vId)
+        #duplicate divide into
+        if vId not in recall_dict:
+            if recall_name == config_.PUSH_FROM['rov_recall_region_h']:
+                region_h_recall.append(per_item)
+            elif recall_name == config_.PUSH_FROM['rov_recall_region_24h']:
+                region_day_recall.append(per_item)
+            elif recall_name == config_.PUSH_FROM['rov_recall_24h']:
+                select_day_recall.append(per_item)
+            elif recall_name == config_.PUSH_FROM['rov_recall_24h_dup']:
+                no_selected_recall.append(per_item)
+        if vId not in recall_dict:
+            recall_dict[vId] = recall_name
+        else:
+            recall_name = recall_dict[vId] + "," + recall_name
+            recall_dict[vId] = recall_name
+    all_flow_set.add(fast_flow_set)
+    all_flow_set.add(flow_flow_set)
+    #3. filter video, 先过预曝光
+    filter_ = FilterVideos(request_id=request_id,
+                           app_type=app_type, mid=mid, uid=uid, video_ids=recall_dict.keys())
+    #a).expose filter
+    expose_filterd_videos = filter_.new_filter_video()
+    if expose_filterd_videos is None:
+        return
+    #b). sep_filter
+    normal_video_list, flow_video_list = filter_.new_flow_video(expose_filterd_videos, all_flow_set, region_code, shield_config)
+    if len(normal_video_list) and len(flow_video_list)==0:
+        return
+    #4. sort: old sort: flow 按概率出
+    start_rank = time.time()
+    #quick_flow_pool_P get from redis
+    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
+
+    all_recall_list = normal_video_list+flow_video_list
+
+    rank_result= []
+    if ab_code=="ab_new_test":
+        rank_ids = video_new_rank(videoIds=all_recall_list,fast_flow_set=fast_flow_set, flow_set=flow_flow_set,size=size, top_K=top_K, flow_pool_P=float(flow_pool_P))
+        for rank_id in rank_ids:
+            if rank_id in recall_dict:
+                rank_result.append(recall_dict.get(rank_id))
+    else:
+        all_dup_recall_result = region_h_recall+region_day_recall+select_day_recall+no_selected_recall
+        rank_result = refactor_video_rank(rov_recall_rank=all_dup_recall_result,fast_flow_set=fast_flow_set, flow_set=flow_flow_set, size=size, top_K=top_K, flow_pool_P=float(flow_pool_P))
+
+    result['rankResult'] = rank_result
+    result['rankTime'] = (time.time() - start_rank) * 1000
+
+    # if not rank_result:
+    #     # 兜底策略
+    #     # log_.info('====== bottom strategy')
+    #     start_bottom = time.time()
+    #     rank_result = bottom_strategy2(
+    #         size=size, app_type=app_type, mid=mid, uid=uid, ab_code=ab_code, client_info=client_info, params=params
+    #     )
+    #
+    #     # if ab_code == config_.AB_CODE['region_rank_by_h'].get('abtest_130'):
+    #     #     rank_result = bottom_strategy2(
+    #     #         size=size, app_type=app_type, mid=mid, uid=uid, ab_code=ab_code, client_info=client_info, params=params
+    #     #     )
+    #     # else:
+    #     #     rank_result = bottom_strategy(
+    #     #         request_id=request_id, size=size, app_type=app_type, ab_code=ab_code, params=params
+    #     #     )
+    #
+    #     # log_.info({
+    #     #     'logTimestamp': int(time.time() * 1000),
+    #     #     'request_id': request_id,
+    #     #     'mid': mid,
+    #     #     'uid': uid,
+    #     #     'operation': 'bottom',
+    #     #     'bottom_result': rank_result,
+    #     #     'executeTime': (time.time() - start_bottom) * 1000
+    #     # })
+    #     result['bottomResult'] = rank_result
+    #     result['bottomTime'] = (time.time() - start_bottom) * 1000
+    #
+    # result['rankResult'] = rank_result
+
+    return result
+    # return rank_result, last_rov_recall_key
+
 
 def ab_test_op(rank_result, ab_code_list, app_type, mid, uid, **kwargs):
     """

utils.py

@@ -476,6 +476,49 @@ class FilterVideos(object):
             # video_ids = [int(video_id) for video_id in video_ids if int(video_id) not in shield_videos]
         return video_ids
 
+    def new_filter_video(self):
+        """视频过滤"""
+        # 1. 预曝光过滤
+        st_pre = time.time()
+        filtered_pre_result = self.filter_video_previewed(self.video_ids)
+        if not filtered_pre_result:
+            return None
+        log_.info({
+             'logTimestamp': int(time.time() * 1000),
+             'request_id': self.request_id,
+             'app_type': self.app_type,
+             'mid': self.mid,
+             'uid': self.uid,
+             'operation': 'preview_filter',
+             'request_videos': self.video_ids,
+             'preview_filter_result': filtered_pre_result,
+             'executeTime': (time.time() - st_pre) * 1000
+         })
+        #2. 视频已曝光过滤
+        st_viewed = time.time()
+        filtered_viewed_result = self.filter_video_viewed(video_ids=filtered_pre_result)
+        if not filtered_viewed_result:
+            return None
+        return filtered_viewed_result
+
+    def new_flow_video(self, vid_list, flow_vids_set, region_code, shield_config):
+        flow_video_list = []
+        normal_video_list = []
+        for v_id in vid_list:
+            if vid_list in flow_vids_set:
+                flow_video_list.append(v_id)
+            else:
+                normal_video_list.append(v_id)
+        shield_key_name_list = shield_config.get(region_code, None)
+        if shield_key_name_list is not None:
+            filtered_shield_video_ids = self.filter_shield_video(
+                video_ids=flow_video_list, shield_key_name_list=shield_key_name_list
+            )
+            return normal_video_list, filtered_shield_video_ids
+        else:
+            return normal_video_list, flow_video_list
+
+
 
 if __name__ == '__main__':
     # filter_ = FilterVideos(app_type=1, mid='22', uid='www', video_ids=[1, 2, 3, 55])

video_rank.py

@@ -151,7 +151,7 @@ def video_rank(data, size, top_K, flow_pool_P):
     return rank_result[:size]
 
 
-def video_new_rank(data, size, top_K, flow_pool_P):
+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': []}
@@ -160,66 +160,124 @@ def video_new_rank(data, size, top_K, flow_pool_P):
         :param flow_pool_P: size-top_K视频为流量池视频的概率 type-float
         :return: rank_result
         """
-    if not data['rov_pool_recall'] and not data['flow_pool_recall']:
+    if not videoIds or len(videoIds)==0:
         return []
 
     redisObj = RedisHelper()
-    video_id_list = []
-    for d in data:
-        video_id_list.append(d.get('rovScore', '0'))
-    video_scores = redisObj.get_batch_key(video_id_list)
+    video_scores = redisObj.get_batch_key(videoIds)
     video_items = []
-    for i in range(len(video_id_list)):
+    for i in range(len(video_scores)):
         try:
             video_score_str = json.load(video_scores[i])
-            video_items.append((video_id_list[i], video_score_str[0]))
+            video_items.append((videoIds[i], video_score_str[0]))
         except Exception:
-            video_items.append((video_id_list[i], 0.0))
+            video_items.append((videoIds[i], 0.0))
     sort_items = sorted(video_items, key=lambda k: k[1], reverse=True)
     rov_recall_rank = sort_items
-    # 流量池
-    flow_recall_rank = sorted(data['flow_pool_recall'], key=lambda k: k.get('rovScore', 0), reverse=True)
-    # 对各路召回的视频进行去重
-    rov_recall_rank, flow_recall_rank = remove_duplicate(rov_recall=rov_recall_rank, flow_recall=flow_recall_rank,
-                                                         top_K=top_K)
-    # log_.info('remove_duplicate finished! rov_recall_rank = {}, flow_recall_rank = {}'.format(
-    #     rov_recall_rank, flow_recall_rank))
-
-    # rank_result = relevant_recall_rank
+    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(flow_recall_rank[:top_K])
-        flow_recall_rank = flow_recall_rank[top_K:]
+        rank_result.extend(all_flow_recall_rank[:top_K])
+        all_flow_recall_rank = all_flow_recall_rank[top_K:]
 
-    # 按概率 p 及score排序获取 size - k 个视频
+    for rank_item in rank_result:
+        rank_set.add(rank_item[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:
-            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]
+            for flow_item in all_flow_recall_rank:
+                if flow_item[0] in rank_set:
+                    continue
+                else:
+                    rank_result.append(flow_item)
+                    rank_set.add(flow_item[0])
         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]
+            for recall_item in rov_recall_rank:
+                if recall_item[0] in rank_set:
+                    continue
+                else:
+                    rank_result.append(recall_item)
+                    rank_set.add(recall_item[0])
         i += 1
     return rank_result[:size]
 
 
+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:
+        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:]
+    #已存放了多少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):
     """

video_recall.py

@@ -32,6 +32,7 @@ class PoolRecall(object):
         self.app_type = app_type
         self.mid = mid
         self.uid = uid
+        self.video_id = video_id
         self.ab_code = ab_code
         self.client_info = client_info
         self.rule_key = rule_key
@@ -2113,7 +2114,7 @@ class PoolRecall(object):
         print(data)
         recall_result = []
         if data is not  None:
-            json_result =json.load(data)
+            json_result =json.loads(data)
             print("json_result:", json_result)
             for per_item in json_result:
                 recall_result.append(
@@ -2122,3 +2123,119 @@ class PoolRecall(object):
                      'abCode': self.ab_code}
                 )
         return recall_result
+
+    # get region_hour_recall
+    def get_region_hour_recall(self, size=4, region_code='',):
+        pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H
+        recall_key = f"{pool_key_prefix}:{region_code}:{self.data_key}:{self.rule_key}"
+        print("recall_key:", recall_key)
+        data = self.redis_helper.get_data_from_redis(key_name=recall_key)
+        print(data)
+        recall_result = []
+        if data is not None:
+            json_result = json.loads(data)
+            print("json_result:", json_result)
+            for per_item in json_result:
+                recall_result.append(
+                    {'videoId': per_item[0], 'flowPool': '',
+                     'rovScore': per_item[1], 'pushFrom': config_.PUSH_FROM['rov_recall_region_h'],
+                     'abCode': self.ab_code}
+                )
+        return recall_result
+
+    # get region_day_recall
+    def get_region_day_recall(self, size=4,region_code=''):
+        """召回池召回视频"""
+        pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H
+        recall_key = f"{pool_key_prefix}:{region_code}:{self.data_key}:{self.rule_key}"
+        print("recall_key:", recall_key)
+        data = self.redis_helper.get_data_from_redis(key_name=recall_key)
+        print(data)
+        recall_result = []
+        if data is not None:
+            json_result = json.loads(data)
+            print("json_result:", json_result)
+            for per_item in json_result:
+                recall_result.append(
+                    {'videoId': per_item[0], 'flowPool': '',
+                     'rovScore': per_item[1], 'pushFrom': config_.PUSH_FROM['rov_recall_region_24h'],
+                     'abCode': self.ab_code}
+                )
+        return recall_result
+
+
+    def get_selected_recall(self, size=4, region_code=''):
+        """召回池召回视频"""
+        pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_24H_H
+        recall_key = f"{pool_key_prefix}:{region_code}:{self.data_key}:{self.rule_key}"
+        print("recall_key:", recall_key)
+        data = self.redis_helper.get_data_from_redis(key_name=recall_key)
+        print(data)
+        recall_result = []
+        if data is not None:
+            json_result = json.loads(data)
+            print("json_result:", json_result)
+            for per_item in json_result:
+                recall_result.append(
+                    {'videoId': per_item[0], 'flowPool': '',
+                     'rovScore': per_item[1], 'pushFrom': config_.PUSH_FROM['rov_recall_24h'],
+                     'abCode': self.ab_code}
+                )
+        return recall_result
+
+    def get_no_selected_recall(self, size=4, region_code=''):
+        """未选择召回池召回视频"""
+        pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H
+        recall_key = f"{pool_key_prefix}:{region_code}:{self.data_key}:{self.rule_key}"
+        print("recall_key:", recall_key)
+        data = self.redis_helper.get_data_from_redis(key_name=recall_key)
+        print(data)
+        recall_result = []
+        if data is not None:
+            json_result = json.loads(data)
+            print("json_result:", json_result)
+            for per_item in json_result:
+                recall_result.append(
+                    {'videoId': per_item[0], 'flowPool': '',
+                     'rovScore': per_item[1], 'pushFrom': config_.PUSH_FROM['rov_recall_24h_dup'],
+                     'abCode': self.ab_code}
+                )
+        return recall_result
+
+    def get_fast_flow_pool_recall(self, size=4):
+        """快速流量池召回视频"""
+        pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H
+        recall_key = f"{pool_key_prefix}:{region_code}:{self.data_key}:{self.rule_key}"
+        #recall_key = "recall:item:score:region:dup3:24h:" :{data_key}: {rule_key}
+        print("recall_key:", recall_key)
+        data = self.redis_helper.get_data_from_redis(key_name=recall_key)
+        print(data)
+        recall_result = []
+        if data is not None:
+            json_result = json.loads(data)
+            print("json_result:", json_result)
+            for per_item in json_result:
+                recall_result.append(
+                    {'videoId': per_item[0], 'flowPool': '',
+                     'rovScore': per_item[1], 'pushFrom': config_.PUSH_FROM['fast_flow_recall'],
+                     'abCode': self.ab_code}
+                )
+        return  recall_result
+
+    def get_flow_pool_recall(self, size=4):
+        """流量池召回视频"""
+        recall_key = ""
+        print("recall_key:", recall_key)
+        data = self.redis_helper.get_data_from_redis(key_name=recall_key)
+        print(data)
+        recall_result = []
+        if data is not None:
+            json_result = json.loads(data)
+            print("json_result:", json_result)
+            for per_item in json_result:
+                recall_result.append(
+                    {'videoId': per_item[0], 'flowPool': '',
+                     'rovScore': per_item[1], 'pushFrom': config_.PUSH_FROM['normal_flow_recall'],
+                     'abCode': self.ab_code}
+                )
+        return  recall_result