flowpool

flowpool_data_update_with_level_score_v2.py

@@ -47,7 +47,7 @@ def get_videos_from_flow_pool(app_type, size=1000):
     return videos
 
 
-def get_videos_remain_view_count(video_info_list):
+def update_remain_view_count(video_info_list):
     """
     获取视频在流量池中的剩余可分发数，并存入对应的redis中
     :param app_type: 产品标识 type-int
@@ -56,8 +56,9 @@ def get_videos_remain_view_count(video_info_list):
     """
     redis_helper = RedisHelper()
     if not video_info_list:
-        return
+        return []
 
+    remain_videos = []
     # 每次请求10个
     for i in range(len(video_info_list)//10 + 1):
         remain_st_time = time.time()
@@ -76,24 +77,16 @@ def get_videos_remain_view_count(video_info_list):
                 continue
             distribute_count = int(item['distributeCount'])
             if distribute_count > 0:
+                remain_videos.append(item['videoId'])
                 # 将分发数更新到本地记录
                 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)
+                redis_helper.set_data_to_redis(key_name=key_name, value=distribute_count, expire_time=25 * 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:
-                    for level in range(1, 7):
-                        flow_pool_key = \
-                            f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL_SCORE}{config_.APP_TYPE.get(type_name)}:{level}"
-                        redis_helper.remove_value_from_zset(key_name=flow_pool_key, value=(value, ))
-                    quick_flow_pool_key = f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX}{config_.APP_TYPE.get(type_name)}" \
-                                          f":{config_.QUICK_FLOW_POOL_ID}"
-                    redis_helper.remove_value_from_zset(key_name=quick_flow_pool_key, value=(value, ))
-
+    log_.info(f"新增加不分发过滤前后整体数量: {len(video_info_list)}:{len(remain_videos)}")
+	return remain_videos
 
 def get_flow_pool_recommend_config(flow_pool_id):
     """获取流量池推荐分发配置"""
@@ -111,50 +104,21 @@ def get_flow_pool_recommend_config(flow_pool_id):
     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, flow_pool_id_list):
+def update_flow_pool(flow_pool_id_list):
     """
     获取流量池可分发视频，并将结果上传Redis
     :param app_type: 产品标识 type-int
     :return: None
     """
+    # 所有产品都从0取数据
+    app_type = 0
+
     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
+            return
         # video_id 与 flow_pool, level 进行mapping
         video_ids = set()
         log_.info('流量池中视频数：{}'.format(len(videos)))
@@ -174,48 +138,51 @@ def get_flow_pool_data(app_type, video_info_list, flow_pool_id_list):
         log_.info(f"需更新流量池视频数: {len(video_ids)}")
 
         # 对视频状态进行过滤
-        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))
+        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
+            return
+
+
+
+        # 1. 更新分发数量，过滤分发数量为0的视频
+        video_info_list = []
+        for video_id in filtered_videos:
+            for item in mapping.get(video_id):
+                flow_pool = item['flowPool']
+                video_info = (video_id, flow_pool)
+                if video_info not in video_info_list:
+                    video_info_list.append(video_info)
+
+        log_.info(f"video_info_list count = {len(video_info_list)}")
+        remain_videos = update_remain_view_count(video_info_list)
+        if not remain_videos:
+            log_.info('流量池中视频状态不符合分发')
+            return
+
 
         # 上传数据到redis
         quick_flow_pool_redis_data = dict()
         # 普通流量池视频按照层级存储
         redis_data = dict()
-
-        for video_id in filtered_videos:
+        for video_id in remain_videos:
             for item in mapping.get(video_id):
                 flow_pool = item['flowPool']
-                level = item['level']
                 try:
                     score = float(item['score'])
                 except:
                     continue
-                # 判断是否为快速曝光流量池视频
                 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[value] = score
-                else:
-                    if level not in redis_data:
-                        redis_data[level] = dict()
-                    redis_data[level][value] = score
-
-                video_info = (video_id, flow_pool)
-                if video_info not in video_info_list:
-                    video_info_list.append(video_info)
+                else:                    
+				    flow_pool_redis_data[value] = score
 
-        # 快速曝光流量池视频写入redis
+        # 2. quick曝光池
         redis_helper = RedisHelper()
         quick_flow_pool_key_name = f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX}{app_type}:{config_.QUICK_FLOW_POOL_ID}"
         # 如果key已存在，删除key
@@ -233,30 +200,19 @@ def get_flow_pool_data(app_type, video_info_list, flow_pool_id_list):
                 redis_helper.set_data_to_redis(key_name=distribute_rate_key_name, value=distribute_rate,
                                                expire_time=15 * 60)
 
-        # 普通流量池视频写入redis - 分层存储
-        level_list = []
-        for level, videos in redis_data.items():
-            log_.info(f"level: {level}, videos_count: {len(videos)}")
-            level_list.append(level)
-            flow_pool_key_name = f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL_SCORE}{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_zset(key_name=flow_pool_key_name, data=videos, expire_time=24 * 3600)
+        # 3. 普通流量池
+        log_.info(f"videos_count: {len(flow_pool_redis_data)}")
+        flow_pool_key_name = f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL_SCORE}{app_type}"
+        # 如果key已存在，删除key
+        if redis_helper.key_exists(flow_pool_key_name):
+            redis_helper.del_keys(flow_pool_key_name)
+        # 写入redis
+        if flow_pool_redis_data:
+            redis_helper.add_data_with_zset(key_name=flow_pool_key_name, data=flow_pool_redis_data, expire_time=24 * 3600)
 
-        # 删除此时不存在的level key
-        for i in range(1, 7):
-            if i not in level_list:
-                flow_pool_key_name = f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL_SCORE}{app_type}:{i}"
-                # 如果key已存在，删除key
-                if redis_helper.key_exists(flow_pool_key_name):
-                    redis_helper.del_keys(flow_pool_key_name)
 
         log_.info('data to redis finished!')
 
-        return video_info_list
 
     except Exception as e:
         log_.error('流量池更新失败, appType: {} exception: {}, traceback: {}'.format(
@@ -266,35 +222,10 @@ def get_flow_pool_data(app_type, video_info_list, flow_pool_id_list):
             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(40)
-    app_type_list = [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]
     log_.info('flow pool predict start...')
     # 获取对应流量池id列表
     redis_helper = RedisHelper()
@@ -304,21 +235,9 @@ if __name__ == '__main__':
     else:
         flow_pool_abtest_config = {}
     flow_pool_id_list = flow_pool_abtest_config.get('experimental_flow_set_level_score', [])
-    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,
-                                                 flow_pool_id_list=flow_pool_id_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('predict start...')
+    update_flow_pool(flow_pool_id_list=flow_pool_id_list)
+    log_.info('predict end...')
     log_.info(f"expend time = {(time.time() - st_time) * 1000}ms")
+    log_.info('flow pool predict end...')
 

flowpool_data_update_with_level_v2.py

@@ -1,9 +1,5 @@
-import datetime
-import random
 import time
-import os
 import traceback
-import random
 import json
 
 from my_config import set_config
@@ -30,7 +26,6 @@ def get_videos_from_flow_pool(app_type, size=1000):
     videos = []
     retry = 0
     while True:
-        print(config_.GET_VIDEOS_FROM_POOL_URL)
         result = request_post(request_url=config_.GET_VIDEOS_FROM_POOL_URL, request_data=request_data)
         if result is None:
             if retry > 2:
@@ -92,6 +87,7 @@ def update_remain_view_count(video_info_list):
                 redis_helper.del_keys(key_name=key_name)
     log_.info(f"新增加不分发过滤前后整体数量: {len(video_info_list)}:{len(remain_videos)}")
     return remain_videos
+
 def get_flow_pool_recommend_config(flow_pool_id):
     """获取流量池推荐分发配置"""
     result = request_post(request_url=config_.GET_FLOW_POOL_RECOMMEND_CONFIG_URL)