Merge branch 'feature_2023091516_liqian_flowpool_update' into dev

config.py

@@ -170,6 +170,7 @@ class BaseConfig(object):
             'abtest_409': 60071,
             'abtest_420': 60072,
             'abtest_421': 60073,
+            'abtest_430': 60074,
         },  # 地域分组小时级规则实验
 
         'rank_by_24h': {
@@ -450,6 +451,10 @@ class BaseConfig(object):
             'data_key': 'data2', 'rule_key': 'rule7-1',
             'ab_code': AB_CODE['region_rank_by_h'].get('abtest_421')
         },  # simrecall+融合+地域召回使用vlog数据&内容精选数据 data-[vlog: 0.5, 内容精选: 0.5]
+        '430': {
+            'data_key': 'videos5', 'rule_key': 'rule7-1',
+            'ab_code': AB_CODE['region_rank_by_h'].get('abtest_430')
+        },  # simrecall+融合+地域召回使用内容精选数据 data-[内容精选: 1]
     }
 
     # APP ab实验配置
@@ -704,6 +709,10 @@ class BaseConfig(object):
     QUICK_FLOWPOOL_KEY_NAME_PREFIX = 'flow:pool:quick:item:score:'
     # 流量池离线模型结果存放 redis key前缀，完整格式 flow:pool:item:{appType}
     FLOWPOOL_KEY_NAME_PREFIX_SET = 'flow:pool:item:'
+    # 流量池视频分层存放 redis key前缀，完整格式 flow:pool:level:item:{appType}:{level}
+    FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL = 'flow:pool:level:item:'
+    # 流量池各层分发概率权重存放 redis key，完整格式 flow:pool:level:
+    FLOWPOOL_LEVEL_WEIGHT_KEY_NAME = 'flow:pool:level:recommend:weight'
     # 快速曝光流量池数据存放 redis key前缀，完整格式 flow:pool:quick:item:{appType}:{flowPool_id}
     QUICK_FLOWPOOL_KEY_NAME_PREFIX_SET = 'flow:pool:quick:item:'
     # 快速曝光流量池分发概率 redis key前缀，完整格式 flow:pool:quick:distribute:rate:{flowPool_id}

recommend.py

@@ -229,8 +229,8 @@ def video_recommend(request_id, mid, uid, size, top_K, flow_pool_P, app_type, al
             t.append(gevent.spawn(pool_recall.get_return_video_reall))
     else:
         t = [gevent.spawn(pool_recall.rov_pool_recall_with_region, size, expire_time),
-             gevent.spawn(pool_recall.flow_pool_recall_new, size, config_.QUICK_FLOW_POOL_ID),
-             gevent.spawn(pool_recall.flow_pool_recall_new, size)]
+             gevent.spawn(pool_recall.flow_pool_recall_new_with_level, size, config_.QUICK_FLOW_POOL_ID),
+             gevent.spawn(pool_recall.flow_pool_recall_new_with_level, size)]
         if ab_code==60058:
             t.append(gevent.spawn(pool_recall.get_U2I_reall, mid))
             t.append(gevent.spawn(pool_recall.get_play_reall, mid))
@@ -458,9 +458,9 @@ def video_old_recommend(request_id, mid, uid, size, top_K, flow_pool_P, app_type
         #     t.append(gevent.spawn(pool_recall.get_return_video_reall, 'rv2:'))
     else:
         t = [gevent.spawn(pool_recall.rov_pool_recall_with_region, size, expire_time),
-             gevent.spawn(pool_recall.flow_pool_recall_new, size, config_.QUICK_FLOW_POOL_ID),
-             gevent.spawn(pool_recall.flow_pool_recall_new, size)]
-    if ab_code == 60054 or ab_code == 60066 or ab_code == 60072 or ab_code == 60073:
+             gevent.spawn(pool_recall.flow_pool_recall_new_with_level, size, config_.QUICK_FLOW_POOL_ID),
+             gevent.spawn(pool_recall.flow_pool_recall_new_with_level, size)]
+    if ab_code == 60054 or ab_code == 60066 or ab_code == 60072 or ab_code == 60073 or ab_code == 60074:
         t.append(gevent.spawn(pool_recall.get_sim_hot_item_reall_filter))
     elif ab_code == 60056 or ab_code == 60071:
         t.append(gevent.spawn(pool_recall.get_sim_hot_item_reall_filter))
@@ -481,7 +481,7 @@ def video_old_recommend(request_id, mid, uid, size, top_K, flow_pool_P, app_type
         result['rankResult'] = []
         return result
     #1. merge simrecall or  deepfm
-    if ab_code == 60054 or ab_code == 60066 or ab_code == 60072 or ab_code == 60073:
+    if ab_code == 60054 or ab_code == 60066 or ab_code == 60072 or ab_code == 60073 or ab_code == 60074:
         rov_pool_recall = []
         if len(recall_result_list) >= 2:
             region_recall = recall_result_list[0]
@@ -1009,7 +1009,8 @@ def update_redis_data(result, app_type, mid, top_K, expire_time=24*3600):
                 flow_recall_video = [item for item in result if item['pushFrom'] == config_.PUSH_FROM['flow_recall']]
             if flow_recall_video:
                 # update_local_distribute_count(flow_recall_video)
-                update_local_distribute_count_new(flow_recall_video)
+                # update_local_distribute_count_new(flow_recall_video)
+                update_local_distribute_count_new_with_level(flow_recall_video)
                 # log_.info('update local distribute count success!')
 
         # 限流视频分发数记录
@@ -1202,6 +1203,58 @@ def update_local_distribute_count_new(videos):
         log_.error(traceback.format_exc())
 
 
+def update_local_distribute_count_new_with_level(videos):
+    """
+    更新本地分发数
+    :param videos: 视频列表 type-list [{'videoId':'', 'flowPool':'', 'distributeCount': '',
+                                    'rovScore': '', 'pushFrom': 'flow_pool', 'abCode': self.ab_code}, ....]
+    :return:
+    """
+    try:
+        redis_helper = RedisHelper()
+        level_weight = redis_helper.get_data_from_redis(key_name=config_.FLOWPOOL_LEVEL_WEIGHT_KEY_NAME)
+        level_list = [level for level in json.loads(level_weight)]
+        for item in videos:
+            video_id, flow_pool = item['videoId'], item['flowPool']
+            key_name = f"{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{video_id}:{flow_pool}"
+            # 本地记录的分发数 - 1
+            redis_helper.decr_key(key_name=key_name, amount=1, expire_time=15 * 60)
+            # 对该视频做分发数检查
+            cur_count = redis_helper.get_data_from_redis(key_name=key_name)
+            # 无记录
+            if cur_count is None:
+                continue
+            # 本地分发数 cur_count <= 0，从所有的流量召回池移除，删除本地分发记录key
+            if int(cur_count) <= 0:
+                add_remove_log = False
+                redis_helper.del_keys(key_name=key_name)
+                for app_name in config_.APP_TYPE:
+                    app_type = config_.APP_TYPE.get(app_name)
+                    flow_pool_key_list = [
+                        f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX_SET}{app_type}:{config_.QUICK_FLOW_POOL_ID}"
+                    ]
+                    for level in level_list:
+                        flow_pool_key_list.append(f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL}{app_type}:{level}")
+                    for key in flow_pool_key_list:
+                        remove_res = redis_helper.remove_value_from_set(key_name=key,
+                                                                        values=(f"{video_id}-{flow_pool}", ))
+                        if remove_res > 0:
+                            add_remove_log = True
+                if add_remove_log is True:
+                    log_.info({'tag': 'remove video_id from flow_pool', 'video_id': video_id, 'flow_pool': flow_pool})
+
+            # if redis_helper.key_exists(key_name=key_name):
+            #     # 该视频本地有记录，本地记录的分发数 - 1
+            #     redis_helper.decr_key(key_name=key_name, amount=1, expire_time=5 * 60)
+            # else:
+            #     # 该视频本地无记录，接口获取的分发数 - 1
+            #     redis_helper.incr_key(key_name=key_name, amount=int(item['distributeCount']) - 1, expire_time=5 * 60)
+
+    except Exception as e:
+        log_.error('update_local_distribute_count error...')
+        log_.error(traceback.format_exc())
+
+
 def get_religion_class_with_mid(mid, religion_class_name):
     """
     判断用户是否属于对应的宗教类型
@@ -1755,7 +1808,8 @@ def video_homepage_recommend(request_id, mid, uid, size, app_type, algo_type,
     # return video, return video2
     # old video: 60056, test2
     if ab_code == 60054 or ab_code == 60056 or ab_code == 60067 or ab_code == 60068 or ab_code == 60066 \
-            or ab_code == 60069 or ab_code == 60070 or ab_code == 60071 or ab_code == 60072 or ab_code == 60073:
+            or ab_code == 60069 or ab_code == 60070 or ab_code == 60071 or ab_code == 60072 or ab_code == 60073 \
+            or ab_code == 60074:
         result, fea_info = 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,
@@ -1874,7 +1928,8 @@ def video_relevant_recommend(request_id, video_id, mid, uid, size, app_type, ab_
     #                              old_video_index=old_video_index, video_id=video_id,
     #                              params=params, rule_key_30day=rule_key_30day, shield_config=shield_config)
     if ab_code == 60054 or ab_code == 60056 or ab_code == 60067 or ab_code == 60068 or ab_code == 60066 \
-            or ab_code == 60069 or ab_code == 60070 or ab_code == 60071 or ab_code == 60072 or ab_code == 60073:
+            or ab_code == 60069 or ab_code == 60070 or ab_code == 60071 or ab_code == 60072 or ab_code == 60073 \
+            or ab_code == 60074:
         result, fea_info = 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,

video_recall.py

@@ -672,6 +672,125 @@ class PoolRecall(object):
 
         return flow_pool_recall_result[:size]
 
+    def flow_pool_recall_new_with_level(self, size=10, flow_pool_id=None):
+        """从流量池中获取视频"""
+        start_time = time.time()
+        # 获取存在城市分组数据的城市编码列表
+        city_code_list = [code for _, code in config_.CITY_CODE.items()]
+        # 获取provinceCode
+        province_code = self.client_info.get('provinceCode', '-1')
+        # 获取cityCode
+        city_code = self.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'
+
+        flow_pool_key, level = self.get_pool_redis_key('flow_set_level', flow_pool_id=flow_pool_id)
+        if flow_pool_key is None:
+            return []
+        # print(flow_pool_key)
+        flow_pool_recall_result = []
+        flow_pool_recall_videos = []
+        # 每次获取的视频数
+        get_size = size * 5
+        # 记录获取频次
+        freq = 0
+        idx = 0
+        while len(flow_pool_recall_result) < size:
+            freq += 1
+            if freq > config_.MAX_FREQ_FROM_FLOW_POOL:
+                break
+            # 获取数据
+            # st_get = time.time()
+            data = self.redis_helper.get_data_with_count_from_set(key_name=flow_pool_key, count=get_size)
+            # et_get = time.time()
+            # log_.info('get data from flow pool redis: freq = {}, data = {}, execute time = {}ms'.format(
+            #     freq, data, (et_get - st_get) * 1000))
+            if not data:
+                # log_.info('流量池中的视频已取完')
+                break
+            # 将video_id 与 flow_pool, score做mapping整理
+            video_ids = []
+            video_mapping = {}
+            video_score = {}
+            for value in data:
+                try:
+                    video_id, flow_pool = value.split('-')
+                except Exception as e:
+                    log_.error({
+                        'request_id': self.request_id,
+                        'app_type': self.app_type,
+                        'flow_pool_value': value
+                    })
+                    continue
+                video_id = int(video_id)
+                if video_id not in video_ids:
+                    video_ids.append(video_id)
+                if video_id not in video_mapping:
+                    video_mapping[video_id] = [flow_pool]
+                else:
+                    video_mapping[video_id].append(flow_pool)
+            # 过滤
+            filter_ = FilterVideos(request_id=self.request_id,
+                                   app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
+            ge = gevent.spawn(filter_.filter_videos, pool_type='flow',
+                              region_code=region_code, shield_config=self.shield_config)
+            ge.join()
+            filtered_result = ge.get()
+            # 检查可分发数
+            if filtered_result:
+                # st_check = time.time()
+                ge = gevent.spawn(self.check_video_counts_new_with_level, video_ids=filtered_result, flow_pool_mapping=video_mapping)
+                ge.join()
+                check_result = ge.get()
+                # log_.info({
+                #     'logTimestamp': int(time.time() * 1000),
+                #     'request_id': self.request_id,
+                #     'app_type': self.app_type,
+                #     'mid': self.mid,
+                #     'uid': self.uid,
+                #     'operation': 'check_video_counts',
+                #     'executeTime': (time.time() - st_check) * 1000
+                # })
+
+                for item in check_result:
+                    video_id = int(item[0])
+                    flow_pool = item[1]
+                    if video_id not in flow_pool_recall_videos:
+                        # 取其中一个 flow_pool 作为召回结果
+                        # 添加视频源参数 pushFrom, abCode
+                        flow_pool_recall_result.append(
+                            {'videoId': video_id, 'flowPool': flow_pool, 'level': level,
+                             'rovScore': random.uniform(0, 100), 'pushFrom': config_.PUSH_FROM['flow_recall'],
+                             'abCode': self.ab_code}
+                        )
+
+                        flow_pool_recall_videos.append(video_id)
+                # et_check = time.time()
+                # log_.info('check result: result = {}, execute time = {}ms'.format(
+                #     check_result, (et_check - st_check) * 1000))
+
+                # # 判断错误标记, True为错误
+                # if error_flag:
+                #     # 结束流量池召回
+                #     break
+
+            idx += get_size
+
+        # log_.info({
+        #     'logTimestamp': int(time.time() * 1000),
+        #     'request_id': self.request_id,
+        #     'operation': 'flow_pool_recall',
+        #     'executeTime': (time.time() - start_time) * 1000
+        # })
+
+        return flow_pool_recall_result[:size]
+
     def check_video_counts(self, video_ids, flow_pool_mapping):
         """
         检查视频剩余可分发数
@@ -757,6 +876,52 @@ class PoolRecall(object):
                         log_.info({'tag': 'remove video_id from flow_pool', 'video_id': video_id, 'flow_pool': flow_pool})
         return check_result
 
+    def check_video_counts_new_with_level(self, video_ids, flow_pool_mapping):
+        """
+        检查视频剩余可分发数
+        :param video_ids: 视频id type-list
+        :param flow_pool_mapping: 视频id-流量池标记mapping, type-dict
+        :return: check_result, error_flag
+        """
+        # flow_pool_key = self.get_pool_redis_key('flow')
+        # videos = []
+        level_weight = self.redis_helper.get_data_from_redis(key_name=config_.FLOWPOOL_LEVEL_WEIGHT_KEY_NAME)
+        level_list = [level for level in json.loads(level_weight)]
+        check_result = []
+        for video_id in video_ids:
+            video_id = int(video_id)
+            for flow_pool in flow_pool_mapping.get(video_id, []):
+                # 判断是否有本地分发记录
+                cur_count = get_videos_local_distribute_count(video_id=video_id, flow_pool=flow_pool)
+                # 无记录
+                if cur_count is None:
+                    # videos.append({'videoId': video_id, 'flowPool': flow_pool})
+                    continue
+                # 本地分发数 cur_count > 0
+                elif cur_count > 0:
+                    check_result.append((video_id, flow_pool))
+                # 本地分发数 cur_count <= 0，从所有的流量召回池移除，删除本地分发记录key
+                else:
+                    add_remove_log = False
+                    remain_count_key = f"{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{video_id}:{flow_pool}"
+                    self.redis_helper.del_keys(remain_count_key)
+                    value = '{}-{}'.format(video_id, flow_pool)
+                    for item in config_.APP_TYPE:
+                        for level in level_list:
+                            flow_pool_key = f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL}{config_.APP_TYPE.get(item)}:{level}"
+                            remove_res = self.redis_helper.remove_value_from_set(key_name=flow_pool_key, values=(value, ))
+                            if remove_res > 0:
+                                add_remove_log = True
+                        quick_flow_pool_key = f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX_SET}{config_.APP_TYPE.get(item)}" \
+                                              f":{config_.QUICK_FLOW_POOL_ID}"
+                        remove_res = self.redis_helper.remove_value_from_set(key_name=quick_flow_pool_key,
+                                                                             values=(value, ))
+                        if remove_res > 0:
+                            add_remove_log = True
+                    if add_remove_log is True:
+                        log_.info({'tag': 'remove video_id from flow_pool', 'video_id': video_id, 'flow_pool': flow_pool})
+        return check_result
+
     """
         # 本次视频都有本地记录
         if len(videos) == 0:
@@ -860,6 +1025,47 @@ class PoolRecall(object):
             else:
                 return f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET}{self.app_type}"
 
+        elif pool_type == 'flow_set_level':
+            if flow_pool_id == config_.QUICK_FLOW_POOL_ID:
+                return f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX_SET}{self.app_type}:{config_.QUICK_FLOW_POOL_ID}", None
+            else:
+                # 1. 获取流量池各层级分发概率权重
+                level_weight = self.redis_helper.get_data_from_redis(key_name=config_.FLOWPOOL_LEVEL_WEIGHT_KEY_NAME)
+                if level_weight is None:
+                    level_weight = {'1': 1, '2': 1, '3': 1, '4': 1, '5': 1, '6': 1}
+                else:
+                    level_weight = json.loads(level_weight)
+                # 2. 判断各层级是否有视频需分发
+                available_level = []
+                for level, weight in level_weight.items():
+                    level_key = f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL}{self.app_type}:{level}"
+                    if self.redis_helper.key_exists(key_name=level_key):
+                        available_level.append((level, level_key, weight))
+                if len(available_level) == 0:
+                    return None, None
+                # 3. 根据可分发层级权重设置分发概率
+                available_level = sorted(available_level, key=lambda x: x[2], reverse=False)
+                weight_sum = sum([int(item[2]) for item in available_level])
+                level_p_mapping = {}
+                level_p_low = 0
+                for level, level_key, weight in available_level:
+                    level_p_up = level_p_low + weight/weight_sum
+                    level_p_mapping[level] = {
+                        'key': level_key,
+                        'level_p': [round(level_p_low, 2), round(level_p_up, 2)]
+                    }
+                    level_p_low = round(level_p_up, 2)
+                # log_.info(f"level_p_mapping: {level_p_mapping}")
+                # 4. 随机生成[0,1)之间数，返回相应概率区间的key
+                random_p = random.random()
+                for level, level_info in level_p_mapping.items():
+                    level_p = level_info['level_p']
+                    if level_p[0] <= random_p < level_p[1]:
+                        # log_.info(f"random_p: {random_p}, level_p: {level_p}, level: {level}")
+                        return level_info['key'], level
+                    else:
+                        continue
+
         elif pool_type == 'special':
             key_name_prefix = config_.KEY_NAME_PREFIX_SPECIAL_VIDEOS
             # 判断列表是否更新，未更新则使用前一天的列表