sync redis

flowpool_data_update_with_level_v2.py

@@ -58,9 +58,9 @@ def update_remain_view_count(video_info_list):
     """
     redis_helper = RedisHelper()
     if not video_info_list:
-        return []
+        return {}
 
-    remain_videos = []
+    remain_videos = dict()
     # 每次请求10个
     for i in range(len(video_info_list)//10 + 1):
         remain_st_time = time.time()
@@ -79,7 +79,7 @@ def update_remain_view_count(video_info_list):
                 continue
             distribute_count = int(item['distributeCount'])
             if distribute_count > 0:
-                remain_videos.append(item['videoId'])
+                remain_videos[item['videoId']].add(distribute_count)
                 # 将分发数更新到本地记录
                 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=25 * 60)
@@ -168,8 +168,9 @@ def update_flow_pool(flow_pool_id_list):
 
         # 上传数据到redis
         quick_flow_pool_redis_data = set()
-        flow_pool_redis_data = set()
-        for video_id in remain_videos:
+        flow_pool_redis_data = dict()
+        level_weight = dict()
+        for video_id,distribute_count in remain_videos:
             for item in mapping.get(video_id):
                 flow_pool = item['flowPool']
                 # 判断是否为快速曝光流量池视频
@@ -178,7 +179,11 @@ def update_flow_pool(flow_pool_id_list):
                 if flow_pool_id == config_.QUICK_FLOW_POOL_ID:
                     quick_flow_pool_redis_data.add(value)
                 else:
-                    flow_pool_redis_data.add(value)
+                    if level not in flow_pool_redis_data:
+                        flow_pool_redis_data[level] = set()
+                        level_weight[level] = 0
+                    flow_pool_redis_data[level].add(value)
+                    level_weight[level]=level_weight[level]+distribute_count
 
 
         # 2. quick曝光池
@@ -199,19 +204,25 @@ def update_flow_pool(flow_pool_id_list):
                                                expire_time=15 * 60)
 
         # 3. 普通流量池
-        log_.info(f"videos_count: {len(flow_pool_redis_data)}")
-        flow_pool_key_name = f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL}{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_set(key_name=flow_pool_key_name, values=flow_pool_redis_data, expire_time=24 * 3600)
-            result = redis_helper.get_data_from_set(flow_pool_key_name)
-            if not result:
-                result = []
-            size = len(result)
-            log_.info(f'写入成功key={flow_pool_key_name}:{size}')
+        for level, videos in redis_data.items():
+            log_.info(f"level: {level}, videos_count: {len(videos)}")
+            flow_pool_key_name = f"flow:pool:level:item:v2:{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)
+                result = redis_helper.get_data_from_set(flow_pool_key_name)
+                if not result:
+                    result = []
+                size = len(result)
+                log_.info(f'写入成功key={flow_pool_key_name}:{size}')
+
+
+        # 4. 写入权重
+        redis_helper.set_data_to_redis(key_name="flow:pool:level:weight:v2", value=json.dump(level_weight), expire_time=24*3600): 
+
 
         log_.info('data to redis finished!')