add flowpool videos remainViewCount check

config.py

@@ -685,6 +685,8 @@ class BaseConfig(object):
     QUICK_FLOWPOOL_KEY_NAME_PREFIX = 'flow:pool:quick:item:score:'
     # 快速曝光流量池分发概率 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}
+    LOCAL_DISTRIBUTE_COUNT_PREFIX = 'flow:pool:local:distribute:count:'
 
     # 兜底视频redis存储key
     BOTTOM_KEY_NAME = 'bottom:videos'

db_helper.py

@@ -266,6 +266,17 @@ class RedisHelper(object):
         conn = self.connect()
         conn.persist(key_name)
 
+    def setnx_key(self, key_name, value, expire_time=5*60):
+        """
+        当key不存在时，将value塞入key中，key存在时不做操作
+        :param key_name: key
+        :param value: value
+        :return: 过期时间，单位：s，默认5分钟 type-int
+        """
+        conn = self.connect()
+        conn.setnx(name=key_name, value=value)
+        conn.expire(name=key_name, time=int(expire_time))
+
 
 class HologresHelper(object):
     def __init__(self):

pool_predict.py

@@ -39,29 +39,45 @@ def get_videos_from_flow_pool(app_type, size=1000):
     return videos
 
 
-def get_videos_remain_view_count(app_type, videos_info):
+def get_videos_remain_view_count(app_type, video_info_list):
     """
-    获取视频在流量池中的剩余可分发数
+    获取视频在流量池中的剩余可分发数，并存入对应的redis中
     :param app_type: 产品标识 type-int
-    :param videos_info: 视频信息 (视频id, 流量池标记) type-list，[(video_id, flow_pool), ...]
+    :param video_info_list: 视频信息 (视频id, 流量池标记) type-list，[(video_id, flow_pool), ...]
     :return: data type-list，[(video_id, flow_pool, view_count), ...]
     """
-    if not videos_info:
-        return []
-
-    videos = [{'videoId': info[0], 'flowPool': info[1]} for info in videos_info]
-    request_data = {'appType': app_type, 'videos': videos}
-    result = request_post(request_url=config_.GET_REMAIN_VIEW_COUNT_URL, request_data=request_data)
-
-    if result is None:
-        return []
-
-    if result['code'] != 0:
-        log_.info('获取视频在流量池中的剩余可分发数失败')
-        return []
-
-    data = [(item['videoId'], item['flowPool'], item['viewCount']) for item in result['data']]
-    return data
+    redis_helper = RedisHelper()
+    if not video_info_list:
+        return
+
+    # 每次请求10个
+    for i in range(len(video_info_list)//10 + 1):
+        videos = [{'videoId': info[0], 'flowPool': info[1]} for info in video_info_list[i*10:(i+1)*10]]
+        request_data = {'appType': app_type, 'videos': videos}
+        result = request_post(request_url=config_.GET_REMAIN_VIEW_COUNT_URL,
+                              request_data=request_data, timeout=(0.1, 3))
+        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.setnx_key(key_name=key_name, value=distribute_count, expire_time=10 * 60)
+            else:
+                # 从流量召回池移除
+                value = '{}-{}'.format(item['videoId'], item['flowPool'])
+                for type_name in config_.APP_TYPE:
+                    flow_pool_key = f"{config_.FLOWPOOL_KEY_NAME_PREFIX}{config_.APP_TYPE.get(type_name)}"
+                    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(item)}" \
+                                          f":{config_.QUICK_FLOW_POOL_ID}"
+                    redis_helper.remove_value_from_zset(key_name=quick_flow_pool_key, value=value)
 
 
 def get_flow_pool_recommend_config(flow_pool_id):
@@ -125,6 +141,7 @@ def predict(app_type):
         # 上传数据到redis
         redis_data = {}
         quick_flow_pool_redis_data = {}
+        video_info_list = []
         for i in range(len(video_score)):
             video_id = filtered_videos[i]
             score = video_score[i]
@@ -136,6 +153,9 @@ def predict(app_type):
                     quick_flow_pool_redis_data[value] = score
                 else:
                     redis_data[value] = score
+                video_info = (video_id, flow_pool)
+                if video_info not in video_info_list:
+                    video_info_list.append(video_info)
 
         # 快速曝光流量池视频写入redis
         redis_helper = RedisHelper()
@@ -165,6 +185,12 @@ def predict(app_type):
             redis_helper.add_data_with_zset(key_name=flow_pool_key_name, data=redis_data, expire_time=24 * 3600)
 
         log_.info('data to redis finished!')
+
+        # 更新剩余分发数
+        get_videos_remain_view_count(app_type, video_info_list)
+
+        return video_info_list
+
     except Exception as e:
         log_.error('流量池更新失败, appType: {} exception: {}, traceback: {}'.format(
             app_type, e, traceback.format_exc()))
@@ -297,12 +323,15 @@ if __name__ == '__main__':
     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']:
-            predict_18_19(app_type=app_type)
+            # predict_18_19(app_type=app_type)
+            continue
         elif app_type == config_.APP_TYPE['ZUI_JING_QI']:
-            predict_19(app_type=app_type)
+            # predict_19(app_type=app_type)
+            continue
         else:
             predict(app_type=app_type)
         log_.info('{} predict end...'.format(app_name))
+
     log_.info('flow pool predict end...')
 
 

utils.py

@@ -127,7 +127,7 @@ def send_msg_to_feishu(webhook, key_word, msg_text):
     print(response.text)
 
 
-def request_post(request_url, request_data=None):
+def request_post(request_url, request_data=None, **kwargs):
     """
     post 请求 HTTP接口
     :param request_url: 接口URL
@@ -135,7 +135,7 @@ def request_post(request_url, request_data=None):
     :return: res_data json格式
     """
     try:
-        response = requests.post(url=request_url, json=request_data)
+        response = requests.post(url=request_url, json=request_data, **kwargs)
         if response.status_code == 200:
             res_data = json.loads(response.text)
             return res_data