Merge branch 'feature_2023091919_liqian_ad_abtest_config_update' into dev

config.py

@@ -202,6 +202,7 @@ class BaseConfig(object):
         'data13': {APP_TYPE['VLOG']: 0.3, APP_TYPE['SHORT_VIDEO']: 0.7},  # [vlog, 票圈短视频]
         'data14': {APP_TYPE['VLOG']: 0.78, APP_TYPE['LOVE_LIVE']: 0.11, APP_TYPE['SHORT_VIDEO']: 0.08,
                    APP_TYPE['LONG_VIDEO']: 0.03},
+        'videos5': {APP_TYPE['LONG_VIDEO']: 1},  # [内容精选]
 
     }
 
@@ -279,6 +280,7 @@ class BaseConfig(object):
             # # {'data': 'data14', 'rule': 'rule4'},
             # {'data': 'data1', 'rule': 'rule7'},
             # {'data': 'data1', 'rule': 'rule8'},
+            {'data': 'videos5', 'rule': 'rule4'},  # [内容精选]
         ]
     }
 
@@ -328,6 +330,7 @@ class BaseConfig(object):
             # {'data': 'data14', 'rule': 'rule4'},
             # {'data': 'data1', 'rule': 'rule8'},
             # {'data': 'data1', 'rule': 'rule9'},
+            {'data': 'videos5', 'rule': 'rule4'},  # [内容精选]
         ]
     }
 
@@ -423,6 +426,7 @@ class BaseConfig(object):
             # {'data': 'data1', 'rule': 'rule16'},  # 214 vlog
             # {'data': 'data1', 'rule': 'rule17'},  # 215 vlog
             # {'data': 'data1', 'rule': 'rule18'},  # 224 vlog
+            {'data': 'videos5', 'rule': 'rule7-1'},  # 428 [内容精选]
         ],
         'params_list_new': [
             {'data': 'data10', 'rule': 'rule19'},  # 316 票圈视频 + 召回在线去重
@@ -634,6 +638,10 @@ class BaseConfig(object):
     FLOWPOOL_KEY_NAME_PREFIX_SET = 'flow:pool:item:'
     # 快速曝光流量池数据存放 redis key前缀，完整格式 flow:pool:quick:item:{appType}:{flowPool_id}
     QUICK_FLOWPOOL_KEY_NAME_PREFIX_SET = 'flow:pool:quick:item:'
+    # 流量池视频分层存放 redis key前缀，完整格式 flow:pool:level:item:{appType}:{level}
+    FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL = 'flow:pool:level:item:'
+    # 流量池各层分发概率权重存放 redis key，完整格式 flow:pool:level:recommend:weight
+    FLOWPOOL_LEVEL_WEIGHT_KEY_NAME = 'flow:pool:level:recommend:weight'
     # 快速曝光流量池分发概率 redis key前缀，完整格式 flow:pool:quick:distribute:rate:{flowPool_id}
     QUICK_FLOWPOOL_DISTRIBUTE_RATE_KEY_NAME_PREFIX = 'flow:pool:quick:distribute:rate:'
     # 本地记录视频的可分发数，控制分发，完整key格式：flow:pool:local:distribute:count:{videoId}:{flowPool}
@@ -1573,32 +1581,32 @@ class BaseConfig(object):
                 'ab0': {
                     'update_hours': list(range(7)), 'update_param': 0,
                     'special_update_config': {'special_hours': [0, 1, 7, 8], 'special_gradient': 0.01,
-                                              'special_max_update_step': 42}
+                                              'special_max_update_step': 38}
                 },  # 0-7点，uv控制在0%
                 'ab1': {
                     'update_hours': list(range(7)), 'update_param': 0,
                     'special_update_config': {'special_hours': [0, 1, 7, 8], 'special_gradient': 0.01,
-                                              'special_max_update_step': 42}
+                                              'special_max_update_step': 38}
                 },  # 0-7点，uv控制在0%
                 'ab2': {
                     'update_hours': list(range(7)), 'update_param': 0,
                     'special_update_config': {'special_hours': [0, 1, 7, 8], 'special_gradient': 0.01,
-                                              'special_max_update_step': 42}
+                                              'special_max_update_step': 38}
                 },  # 0-7点，uv控制在0%
                 'ab3': {
                     'update_hours': list(range(7)), 'update_param': 0,
                     'special_update_config': {'special_hours': [0, 1, 7, 8], 'special_gradient': 0.01,
-                                              'special_max_update_step': 42}
+                                              'special_max_update_step': 38}
                 },  # 0-7点，uv控制在0%
                 'ab4': {
                     'update_hours': list(range(7)), 'update_param': 0,
                     'special_update_config': {'special_hours': [0, 1, 7, 8], 'special_gradient': 0.01,
-                                              'special_max_update_step': 42}
+                                              'special_max_update_step': 38}
                 },  # 0-7点，uv控制在0%
                 'ab5': {
                     'update_hours': list(range(7)), 'update_param': 0,
                     'special_update_config': {'special_hours': [0, 1, 7, 8], 'special_gradient': 0.01,
-                                              'special_max_update_step': 42}
+                                              'special_max_update_step': 38}
                 },  # 0-7点，uv控制在0%
                 # 'ab6': {
                 #     'update_hours': list(range(9)), 'update_param': 1/2,
@@ -1608,22 +1616,22 @@ class BaseConfig(object):
                 'ab6': {
                     'update_hours': list(range(7)), 'update_param': 0,
                     'special_update_config': {'special_hours': [0, 1, 7, 8], 'special_gradient': 0.01,
-                                              'special_max_update_step': 42}
+                                              'special_max_update_step': 38}
                 },  # 0-7点，uv控制在设定目标uv的0%
                 'ab7': {
                     'update_hours': list(range(7)), 'update_param': 0,
                     'special_update_config': {'special_hours': [0, 1, 7, 8], 'special_gradient': 0.01,
-                                              'special_max_update_step': 42}
+                                              'special_max_update_step': 38}
                 },  # 0-9点，uv控制在0%
                 'ab8': {
                     'update_hours': list(range(7)), 'update_param': 0,
                     'special_update_config': {'special_hours': [0, 1, 7, 8], 'special_gradient': 0.01,
-                                              'special_max_update_step': 42}
+                                              'special_max_update_step': 38}
                 },  # 0-7点，uv控制在0%
                 'ab9': {
                     'update_hours': list(range(7)), 'update_param': 0,
                     'special_update_config': {'special_hours': [0, 1, 7, 8], 'special_gradient': 0.01,
-                                              'special_max_update_step': 42}
+                                              'special_max_update_step': 38}
                 },  # 0-7点，uv控制在0%
             },
         },

flow_pool_task.sh

@@ -2,8 +2,10 @@ source /etc/profile
 echo $ROV_OFFLINE_ENV
 if [[ $ROV_OFFLINE_ENV == 'test' ]]; then
 #    cd /data2/rov-offline && /root/anaconda3/bin/python /data2/rov-offline/pool_predict.py
-    cd /data2/rov-offline && /root/anaconda3/bin/python /data2/rov-offline/flowpool_data_update.py
+#    cd /data2/rov-offline && /root/anaconda3/bin/python /data2/rov-offline/flowpool_data_update.py
+    cd /data2/rov-offline && /root/anaconda3/bin/python /data2/rov-offline/flowpool_data_update_with_level.py
 elif [[ $ROV_OFFLINE_ENV == 'pro' ]]; then
 #    cd /data/rov-offline && /root/anaconda3/bin/python /data/rov-offline/pool_predict.py
-    cd /data/rov-offline && /root/anaconda3/bin/python /data/rov-offline/flowpool_data_update.py
+#    cd /data/rov-offline && /root/anaconda3/bin/python /data/rov-offline/flowpool_data_update.py
+    cd /data/rov-offline && /root/anaconda3/bin/python /data/rov-offline/flowpool_data_update_with_level.py
 fi

flowpool_data_update_with_level.py

@@ -0,0 +1,297 @@
+import datetime
+import random
+import time
+import os
+import traceback
+import random
+
+from config import set_config
+from utils import request_post, filter_video_status, send_msg_to_feishu, filter_video_status_app, \
+    filter_political_videos
+from log import Log
+from db_helper import RedisHelper
+from odps import ODPS
+
+config_, _ = set_config()
+log_ = Log()
+
+
+def get_videos_from_flow_pool(app_type, size=1000):
+    """
+    从流量池获取视频，循环获取，直到返回数据为None结束
+    :param app_type: 产品标识 type-int
+    :param size: 每次获取视频数量，type-int，默认1000
+    :return: videos  [{'videoId': 1111, 'flowPool': ''}, ...]
+    """
+    # 获取批次标识，利用首次获取数据时间戳为标记
+    batch_flag = int(time.time())
+    request_data = {'appType': app_type, 'batchFlag': batch_flag, 'size': size}
+    videos = []
+    retry = 0
+    while True:
+        result = request_post(request_url=config_.GET_VIDEOS_FROM_POOL_URL, request_data=request_data)
+        if result is None:
+            if retry > 2:
+                break
+            retry += 1
+            continue
+        if result['code'] != 0:
+            log_.info('batch_flag: {}, 获取流量池视频失败'.format(batch_flag))
+            if retry > 2:
+                break
+            retry += 1
+            continue
+        if not result['data']:
+            if retry > 2:
+                break
+            retry += 1
+            continue
+        videos.extend(result['data'])
+    return videos
+
+
+def get_videos_remain_view_count(video_info_list):
+    """
+    获取视频在流量池中的剩余可分发数，并存入对应的redis中
+    :param app_type: 产品标识 type-int
+    :param video_info_list: 视频信息 (视频id, 流量池标记) type-list，[(video_id, flow_pool), ...]
+    :return: data type-list，[(video_id, flow_pool, view_count), ...]
+    """
+    redis_helper = RedisHelper()
+    if not video_info_list:
+        return
+
+    # 每次请求10个
+    for i in range(len(video_info_list)//10 + 1):
+        remain_st_time = time.time()
+        videos = [{'videoId': info[0], 'flowPool': info[1]} for info in video_info_list[i*10:(i+1)*10]]
+        request_data = {'videos': videos}
+        result = request_post(request_url=config_.GET_REMAIN_VIEW_COUNT_URL,
+                              request_data=request_data, timeout=(0.5, 3))
+        log_.info(f"i = {i}, expend time = {(time.time()-remain_st_time)*1000}")
+        if result is None:
+            continue
+        if result['code'] != 0:
+            log_.error('获取视频在流量池中的剩余可分发数失败')
+            continue
+        for item in result['data']:
+            if item['distributeCount'] is None:
+                continue
+            distribute_count = int(item['distributeCount'])
+            if distribute_count > 0:
+                # 将分发数更新到本地记录
+                key_name = f"{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{item['videoId']}:{item['flowPool']}"
+                redis_helper.set_data_to_redis(key_name=key_name, value=distribute_count, expire_time=15 * 60)
+            else:
+                # 将本地记录删除
+                key_name = f"{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{item['videoId']}:{item['flowPool']}"
+                redis_helper.del_keys(key_name=key_name)
+                # 从流量召回池移除
+                value = '{}-{}'.format(item['videoId'], item['flowPool'])
+                for type_name in config_.APP_TYPE:
+                    flow_pool_key = f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET}{config_.APP_TYPE.get(type_name)}"
+                    redis_helper.remove_value_from_set(key_name=flow_pool_key, values=(value, ))
+                    quick_flow_pool_key = f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX_SET}{config_.APP_TYPE.get(type_name)}" \
+                                          f":{config_.QUICK_FLOW_POOL_ID}"
+                    redis_helper.remove_value_from_set(key_name=quick_flow_pool_key, values=(value, ))
+
+
+def get_flow_pool_recommend_config(flow_pool_id):
+    """获取流量池推荐分发配置"""
+    result = request_post(request_url=config_.GET_FLOW_POOL_RECOMMEND_CONFIG_URL)
+    if result is None:
+        return None
+    if result['code'] != 0:
+        return None
+    flow_pool_distribute_config = result['data'].get('flowPoolDistributeConfig')
+    if flow_pool_distribute_config:
+        if int(eval(flow_pool_distribute_config).get('flowPoolId')) == flow_pool_id:
+            return eval(eval(flow_pool_distribute_config).get('distributeRate'))
+        else:
+            return None
+    else:
+        return None
+
+
+def online_flow_pool_data_to_redis(app_type, video_ids_set, video_info_data):
+    redis_helper = RedisHelper()
+    for tag, values in video_ids_set.items():
+        if tag == 'quick_flow_pool':
+            key_name_video_ids = \
+                f"{config_.QUICK_FLOWPOOL_VIDEO_ID_KEY_NAME_PREFIX}{app_type}:{config_.QUICK_FLOW_POOL_ID}"
+            key_prefix_video_info = \
+                f"{config_.QUICK_FLOWPOOL_VIDEO_INFO_KEY_NAME_PREFIX}{app_type}:{config_.QUICK_FLOW_POOL_ID}"
+        else:
+            key_name_video_ids = f"{config_.FLOWPOOL_VIDEO_ID_KEY_NAME_PREFIX}{app_type}"
+            key_prefix_video_info = f"{config_.FLOWPOOL_VIDEO_INFO_KEY_NAME_PREFIX}{app_type}"
+        # 如果key已存在，删除key
+        if redis_helper.key_exists(key_name=key_name_video_ids):
+            redis_helper.del_keys(key_name=key_name_video_ids)
+        # 写入redis
+        if len(values) > 0:
+            redis_helper.add_data_with_set(key_name=key_name_video_ids, values=values, expire_time=3600)
+            info_values = video_info_data[tag]
+            if len(info_values) > 0:
+                for video_id, info_value in info_values.items():
+                    key_name_video_info = f"{key_prefix_video_info}:{video_id}"
+                    # 如果key已存在，删除key
+                    if redis_helper.key_exists(key_name=key_name_video_info):
+                        redis_helper.del_keys(key_name=key_name_video_info)
+                    if len(info_value) > 0:
+                        redis_helper.add_data_with_set(key_name=key_name_video_info, values=info_value,
+                                                       expire_time=3600)
+
+
+def get_flow_pool_data(app_type, video_info_list):
+    """
+    获取流量池可分发视频，并将结果上传Redis
+    :param app_type: 产品标识 type-int
+    :return: None
+    """
+    try:
+        # 从流量池获取数据
+        videos = get_videos_from_flow_pool(app_type=app_type)
+        log_.info(f"app_type: {app_type}")
+        # log_.info(f"videos: {videos}")
+        if len(videos) <= 0:
+            log_.info('流量池中无需分发的视频')
+            return video_info_list
+        # video_id 与 flow_pool, level 进行mapping
+        video_ids = set()
+        log_.info('流量池中视频数：{}'.format(len(videos)))
+        mapping = {}
+        for video in videos:
+            video_id = video['videoId']
+            video_ids.add(video_id)
+            item_info = {'flowPool': video['flowPool'], 'level': video['level']}
+            if video_id in mapping:
+                mapping[video_id].append(item_info)
+            else:
+                mapping[video_id] = [item_info]
+
+        # 对视频状态进行过滤
+        if app_type == config_.APP_TYPE['APP']:
+            filtered_videos = filter_video_status_app(list(video_ids))
+        else:
+            filtered_videos = filter_video_status(list(video_ids))
+        log_.info('filter videos status finished, filtered_videos nums={}'.format(len(filtered_videos)))
+
+        # 涉政视频过滤
+        if app_type not in config_.POLITICAL_RECOMMEND_APP_TYPE_LIST:
+            filtered_videos = filter_political_videos(video_ids=filtered_videos)
+
+        if not filtered_videos:
+            log_.info('流量池中视频状态不符合分发')
+            return video_info_list
+
+        # 上传数据到redis
+        quick_flow_pool_redis_data = set()
+        # 普通流量池视频按照层级存储
+        redis_data = dict()
+
+        for video_id in filtered_videos:
+            for item in mapping.get(video_id):
+                flow_pool = item['flowPool']
+                level = item['level']
+                # 判断是否为快速曝光流量池视频
+                value = '{}-{}'.format(video_id, flow_pool)
+                flow_pool_id = int(flow_pool.split('#')[0])  # flowPool: 流量池ID#分级ID#级别Level#生命周期ID
+                if flow_pool_id == config_.QUICK_FLOW_POOL_ID:
+                    quick_flow_pool_redis_data.add(value)
+                else:
+                    if level not in redis_data:
+                        redis_data[level] = set()
+                    redis_data[level].add(value)
+
+                video_info = (video_id, flow_pool)
+                if video_info not in video_info_list:
+                    video_info_list.append(video_info)
+
+        # 快速曝光流量池视频写入redis
+        redis_helper = RedisHelper()
+        quick_flow_pool_key_name = f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX_SET}{app_type}:{config_.QUICK_FLOW_POOL_ID}"
+        # 如果key已存在，删除key
+        if redis_helper.key_exists(quick_flow_pool_key_name):
+            redis_helper.del_keys(quick_flow_pool_key_name)
+        # 写入redis
+        if quick_flow_pool_redis_data:
+            log_.info(f"quick_flow_pool_redis_data = {quick_flow_pool_redis_data}")
+            redis_helper.add_data_with_set(key_name=quick_flow_pool_key_name, values=quick_flow_pool_redis_data,
+                                           expire_time=24 * 3600)
+            # 快速流量池分发概率存入redis
+            distribute_rate_key_name = f"{config_.QUICK_FLOWPOOL_DISTRIBUTE_RATE_KEY_NAME_PREFIX}{config_.QUICK_FLOW_POOL_ID}"
+            distribute_rate = get_flow_pool_recommend_config(flow_pool_id=config_.QUICK_FLOW_POOL_ID)
+            if distribute_rate is not None:
+                redis_helper.set_data_to_redis(key_name=distribute_rate_key_name, value=distribute_rate,
+                                               expire_time=15 * 60)
+
+        # 普通流量池视频写入redis - 分层存储
+        for level, videos in redis_data.items():
+            log_.info(f"level: {level}, videos_count: {len(videos)}")
+            flow_pool_key_name = f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL}{app_type}:{level}"
+            # 如果key已存在，删除key
+            if redis_helper.key_exists(flow_pool_key_name):
+                redis_helper.del_keys(flow_pool_key_name)
+            # 写入redis
+            if videos:
+                redis_helper.add_data_with_set(key_name=flow_pool_key_name, values=videos, expire_time=24 * 3600)
+
+        log_.info('data to redis finished!')
+
+        return video_info_list
+
+    except Exception as e:
+        log_.error('流量池更新失败, appType: {} exception: {}, traceback: {}'.format(
+            app_type, e, traceback.format_exc()))
+        send_msg_to_feishu(
+            webhook=config_.FEISHU_ROBOT['server_robot'].get('webhook'),
+            key_word=config_.FEISHU_ROBOT['server_robot'].get('key_word'),
+            msg_text='rov-offline{} - 流量池更新失败, appType: {}, exception: {}'.format(config_.ENV_TEXT, app_type, e)
+        )
+        return video_info_list
+
+
+def get_data_from_odps(project, sql):
+    """检查数据是否准备好"""
+    odps = ODPS(
+        access_id=config_.ODPS_CONFIG['ACCESSID'],
+        secret_access_key=config_.ODPS_CONFIG['ACCESSKEY'],
+        project=project,
+        endpoint=config_.ODPS_CONFIG['ENDPOINT'],
+        connect_timeout=3000,
+        read_timeout=500000,
+        pool_maxsize=1000,
+        pool_connections=1000
+    )
+
+    try:
+        with odps.execute_sql(sql=sql).open_reader() as reader:
+            data_df = reader.to_pandas()
+    except Exception as e:
+        data_df = None
+    return data_df
+
+
+if __name__ == '__main__':
+    st_time = time.time()
+    # 为避免第一个app_type获取数据不全，等待1min
+    time.sleep(60)
+    app_type_list = [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]
+    log_.info('flow pool predict start...')
+    video_info_list = []
+    for app_name, app_type in config_.APP_TYPE.items():
+        log_.info('{} predict start...'.format(app_name))
+        if app_type == config_.APP_TYPE['LAO_HAO_KAN_VIDEO']:
+            continue
+        elif app_type == config_.APP_TYPE['ZUI_JING_QI']:
+            continue
+        else:
+            video_info_list = get_flow_pool_data(app_type=app_type, video_info_list=video_info_list)
+        log_.info('{} predict end...'.format(app_name))
+
+    # 更新剩余分发数
+    log_.info(f"video_info_list count = {len(video_info_list)}")
+    get_videos_remain_view_count(video_info_list)
+    log_.info('flow pool predict end...')
+    log_.info(f"expend time = {(time.time() - st_time) * 1000}ms")
+

flowpool_level_weight_update.py

@@ -0,0 +1,23 @@
+import json
+
+from config import set_config
+from log import Log
+from db_helper import RedisHelper
+
+config_, _ = set_config()
+log_ = Log()
+
+flowpool_level_weight = {
+    '1': 100,
+    '2': 100,
+    '3': 100,
+    '4': 100,
+    '5': 100,
+    '6': 100,
+}
+log_.info(f"flowpool_level_weight: {flowpool_level_weight}")
+redis_helper = RedisHelper()
+redis_helper.set_data_to_redis(
+    key_name=config_.FLOWPOOL_LEVEL_WEIGHT_KEY_NAME, value=json.dumps(flowpool_level_weight), expire_time=365 * 24 *3600
+)
+log_.info(f"flowpool_level_weight update finished!")

videos_filter.py

@@ -268,6 +268,79 @@ def filter_flow_pool_new():
     log_.info("flow pool filter end!")
 
 
+def filter_flow_pool_process(app_type, flow_pool_id, key_name):
+    # 获取视频
+    redis_helper = RedisHelper()
+    data = redis_helper.get_data_from_set(key_name=key_name)
+    if data is None:
+        log_.info(f"flow_pool_id = {flow_pool_id}, data is None")
+        log_.info(f"app_type = {app_type}, flow_pool_id = {flow_pool_id}, videos filter end!")
+        return
+    # videoId与flowPool做mapping
+    video_ids = []
+    mapping = {}
+    for video in data:
+        video_id, flow_pool = video.split('-')
+        video_id = int(video_id)
+        if video_id not in video_ids:
+            video_ids.append(video_id)
+            mapping[video_id] = [flow_pool]
+        else:
+            mapping[video_id].append(flow_pool)
+    # 过滤
+    if len(video_ids) == 0:
+        log_.info(f"data size = {len(data)}, video_ids size = {len(video_ids)}, data = {data}")
+        log_.info(f"app_type = {app_type}, flow_pool_id = {flow_pool_id}, videos filter end!")
+        return
+    if app_type == config_.APP_TYPE['APP']:
+        filtered_result = filter_video_status_app(video_ids=video_ids)
+    else:
+        filtered_result = filter_video_status(video_ids=video_ids)
+    # 求差集，获取需要过滤掉的视频，并从redis中移除
+    filter_videos = set(video_ids) - set(filtered_result)
+    log_.info("data size = {}, video_ids size = {}, filtered size = {}, filter sizer = {}".format(
+        len(data), len(video_ids), len(filtered_result), len(filter_videos)))
+    # 移除
+    if len(filter_videos) == 0:
+        log_.info(f"app_type = {app_type}, flow_pool_id = {flow_pool_id} videos filter end!")
+        return
+    remove_videos = ['{}-{}'.format(video_id, flow_pool)
+                     for video_id in filter_videos
+                     for flow_pool in mapping[video_id]]
+    redis_helper.remove_value_from_set(key_name=key_name, values=tuple(remove_videos))
+    log_.info(f"app_type = {app_type}, flow_pool_id = {flow_pool_id} videos filter end!")
+
+
+def filter_flow_pool_new_with_level():
+    """流量池视频过滤"""
+    log_.info("flow pool filter start ...")
+    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)]
+    log_.info(f"level_list: {level_list}")
+    app_type_list = [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]
+    for _, app_type in config_.APP_TYPE.items():
+        log_.info('app_type {} videos filter start...'.format(app_type))
+        if app_type in app_type_list:
+            filter_flow_pool_18_19(app_type=app_type)
+        else:
+            for flow_pool_id in [None, config_.QUICK_FLOW_POOL_ID]:
+                log_.info(f"flow_pool_id = {flow_pool_id}")
+                # 拼接redis-key
+                if flow_pool_id == config_.QUICK_FLOW_POOL_ID:
+                    key_name = get_pool_redis_key(pool_type='flow_set', app_type=app_type, flow_pool_id=flow_pool_id)
+                    filter_flow_pool_process(app_type, flow_pool_id, key_name)
+                else:
+                    for level in level_list:
+                        key_name = get_pool_redis_key(pool_type='flow_set', app_type=app_type,
+                                                      flow_pool_id=flow_pool_id, level=level)
+                        filter_flow_pool_process(app_type, flow_pool_id, key_name)
+                        log_.info(f"app_type = {app_type}, level = {level} videos filter end!")
+
+            log_.info(f"app_type = {app_type} videos filter end!")
+
+    log_.info("flow pool filter end!")
+
+
 def filter_flow_pool_18_19(app_type):
     """流量池视频过滤"""
     log_.info('app_type {} videos filter start...'.format(app_type))
@@ -380,7 +453,7 @@ def filter_rov_updated_app():
     log_.info("update rov videos app filter end!")
 
 
-def get_pool_redis_key(pool_type, app_type=None, flow_pool_id=None):
+def get_pool_redis_key(pool_type, app_type=None, flow_pool_id=None, level=None):
     """
     拼接key
     :param pool_type: type-string {'rov': rov召回池, 'flow': 流量池}
@@ -457,7 +530,10 @@ def get_pool_redis_key(pool_type, app_type=None, flow_pool_id=None):
         if flow_pool_id == config_.QUICK_FLOW_POOL_ID:
             return f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX_SET}{app_type}:{flow_pool_id}"
         else:
-            return f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET}{app_type}"
+            if level is None:
+                return f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET}{app_type}"
+            else:
+                return f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL}{app_type}:{level}"
 
     else:
         log_.error('pool type error')
@@ -1006,7 +1082,8 @@ def main():
         # filter_rov_pool(app_type=config_.APP_TYPE['ZUI_JING_QI'])
         # 流量池视频过滤
         # filter_flow_pool()
-        filter_flow_pool_new()
+        # filter_flow_pool_new()
+        filter_flow_pool_new_with_level()
         # 兜底视频过滤
         # filter_bottom()
         # 修改过ROV的视频过滤
@@ -1036,7 +1113,7 @@ def main():
         # 过滤完整电影数据
         # filter_whole_movies()
         # 过滤小程序天级更新30天数据
-        filter_day_30day()
+        # filter_day_30day()
     except Exception as e:
         log_.error(traceback.format_exc())
         send_msg_to_feishu(