update route update_rov && video recall

app.py

@@ -100,27 +100,25 @@ def relevant_recommend():
 # 管理后台实时修改rov
 @app.route('/applet/video/update/rov', methods=['GET', 'POST'])
 def update_rov():
-    result = {'code': 200, 'message': 'success'}
-    log_.info('result: {}'.format(result))
-    return json.dumps(result)
+    # result = {'code': 200, 'message': 'success'}
+    # log_.info('result: {}'.format(result))
+    # return json.dumps(result)
     # ##### 暂时关闭该接口
-    # try:
-    #     request_data = json.loads(request.get_data())
-    #     log_.info('request data: {}'.format(request_data))
-    #     app_type = request_data.get('appType')
-    #     video_id = request_data.get('videoId')
-    #     rov_score = request_data.get('rovScore')
-    #     pool_recall = PoolRecall(app_type=app_type)
-    #     rov_pool_key, _ = pool_recall.get_pool_redis_key('rov')
-    #     redis_helper = RedisHelper()
-    #     redis_helper.update_score_with_value(key_name=rov_pool_key, value=video_id, score=rov_score)
-    #     result = {'code': 200, 'message': 'success'}
-    #     log_.info('result: {}'.format(result))
-    #     return json.dumps(result)
-    # except Exception as e:
-    #     log_.error(e)
-    #     result = {'code': -1, 'message': 'fail'}
-    #     return json.dumps(result)
+    try:
+        request_data = json.loads(request.get_data())
+        log_.info('request data: {}'.format(request_data))
+        # app_type = request_data.get('appType')
+        video_id = request_data.get('videoId')
+        rov_score = request_data.get('rovScore')
+        redis_helper = RedisHelper()
+        redis_helper.update_score_with_value(key_name=config_.UPDATE_ROV_KEY_NAME, value=video_id, score=rov_score)
+        result = {'code': 200, 'message': 'update rov success'}
+        log_.info('result: {}'.format(result))
+        return json.dumps(result)
+    except Exception as e:
+        log_.error(traceback.format_exc())
+        result = {'code': -1, 'message': 'update rov fail'}
+        return json.dumps(result)
 
 
 # app热榜

config.py

@@ -27,6 +27,9 @@ class BaseConfig(object):
     # 流量池redis key前缀，完整格式 com.weiqu.video.flowpool.hot.item.score.{appType}
     FLOW_POOL_KEY_NAME_PREFIX = 'com.weiqu.video.flowpool.hot.item.score.'
 
+    # 修改ROV的视频 redis key
+    UPDATE_ROV_KEY_NAME = 'com.weiqu.video.update.rov.item.score'
+
     # 首页推荐预曝光列表redis key 前缀，完整key格式：com.weiqu.video.hot.recommend.previewed.{appType}.{mid}
     PREVIEW_KEY_PREFIX = 'com.weiqu.video.hot.recommend.previewed.'
 
@@ -166,7 +169,7 @@ class ProductionConfig(BaseConfig):
 
 
 def set_config():
-    # return DevelopmentConfig()
+    return DevelopmentConfig()
     # return TestConfig()
     # return PreProductionConfig()
-    return ProductionConfig()
+    # return ProductionConfig()

db_helper.py

@@ -124,13 +124,13 @@ class RedisHelper(object):
             return None
         return conn.zscore(key_name, value)
 
-    def update_score_with_value(self, key_name, value, score, expire_time=7*24*3600):
+    def update_score_with_value(self, key_name, value, score, expire_time=24*3600):
         """
         在zset中，修改元素value对应的score
         :param key_name: key
         :param value: 元素的值
         :param score: value对应的score更新值
-        :param expire_time: 过期时间，单位：s，默认7天，type-int
+        :param expire_time: 过期时间，单位：s，默认1天，type-int
         """
         conn = self.connect()
         if conn.exists(key_name):

recommend.py

@@ -89,9 +89,11 @@ def video_recommend(mid, uid, size, app_type, algo_type):
     # 将此次获取的ROV召回池config_.K末位视频id同步刷新到Redis中，方便下次快速定位到召回位置，过期时间为1天
     rov_recall_video = [item['videoId'] for item in rank_result if item['pushFrom'] == 'recall_pool']
     if 0 < len(rov_recall_video) <= config_.K:
-        redis_helper.set_data_to_redis(key_name=last_rov_recall_key, value=rov_recall_video[-1])
+        if not redis_helper.get_score_with_value(key_name=config_.UPDATE_ROV_KEY_NAME, value=rov_recall_video[-1]):
+            redis_helper.set_data_to_redis(key_name=last_rov_recall_key, value=rov_recall_video[-1])
     elif len(rov_recall_video) > config_.K:
-        redis_helper.set_data_to_redis(key_name=last_rov_recall_key, value=rov_recall_video[config_.K - 1])
+        if not redis_helper.get_score_with_value(key_name=config_.UPDATE_ROV_KEY_NAME, value=rov_recall_video[config_.K - 1]):
+            redis_helper.set_data_to_redis(key_name=last_rov_recall_key, value=rov_recall_video[config_.K - 1])
     log_.info('last video redis update success!')
 
     # 将此次分发的流量池视频，对 本地分发数-1 进行记录

video_recall.py

@@ -31,11 +31,18 @@ class PoolRecall(object):
     def rov_pool_recall(self, size=10):
         """从ROV召回池中获取视频"""
         log_.info('====== rov pool recall')
+
+        # 获取修改过rov的视频
+        update_rov_video_ids, update_rov_result = self.get_update_rov_videos()
+        log_.info('update rov result = {}'.format(update_rov_video_ids))
+
         # 获取相关redis key, 用户上一次在rov召回池对应的位置
         rov_pool_key, last_rov_recall_key, idx = self.get_video_last_idx()
         if not rov_pool_key:
             log_.info('ROV召回池中无视频')
-            return []
+            if not update_rov_result:
+                return []
+            return update_rov_result
         rov_pool_recall_result = []
         # 每次获取的视频数
         get_size = size * 2
@@ -61,6 +68,8 @@ class PoolRecall(object):
             video_score = {}
             for value in data:
                 video_id = int(value[0])
+                if video_id in update_rov_video_ids:
+                    continue
                 video_ids.append(video_id)
                 video_score[video_id] = value[1]
             # 过滤
@@ -69,8 +78,9 @@ class PoolRecall(object):
             ge = gevent.spawn(filter_.filter_videos)
             ge.join()
             filtered_result = ge.get()
-            #filtered_result = filter_.filter_videos()
+            # filtered_result = filter_.filter_videos()
             debug_tm_e = time.time()
+
             if filtered_result:
                 # 添加视频源参数 pushFrom, abCode
                 temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)],
@@ -81,6 +91,11 @@ class PoolRecall(object):
                 # 将此次获取的末位视频id同步刷新到Redis中，方便下次快速定位到召回位置，过期时间为1天
                 self.redis_helper.set_data_to_redis(key_name=last_rov_recall_key, value=data[-1][0])
             idx += get_size
+
+        # 被修改rov视频、rov召回池视频 归并排序
+        if update_rov_result:
+            rov_pool_recall_result.extend(update_rov_result)
+            rov_pool_recall_result.sort(key=lambda x: x.get('rovScore', 0), reverse=True)
         return rov_pool_recall_result[:size]
 
     def flow_pool_recall(self, size=10):
@@ -296,3 +311,48 @@ class PoolRecall(object):
         else:
             idx = 0
         return rov_pool_key, last_rov_recall_key, idx
+
+    def get_update_rov_videos(self):
+        """
+        获取修改ROV的视频
+        :return: update_rov_video_ids, update_rov_result
+        """
+        try:
+            # 获取修改过ROV的视频
+            redis_helper = RedisHelper()
+            data = redis_helper.get_data_zset_with_index(key_name=config_.UPDATE_ROV_KEY_NAME,
+                                                         start=0, end=-1, with_scores=True)
+            # 获取视频id,并转换类型为int,将videoId和score做mapping，并存储为key-value{videoId: score}
+            if data is None:
+                return [], []
+            video_ids = []
+            video_score = {}
+            for value in data:
+                video_id = int(value[0])
+                video_ids.append(video_id)
+                video_score[video_id] = value[1]
+            # 过滤
+            filter_ = FilterVideos(app_type=self.app_type, video_ids=video_ids, mid=self.mid, uid=self.uid)
+            ge = gevent.spawn(filter_.filter_videos)
+            ge.join()
+            filtered_result = ge.get()
+
+            # 添加视频源参数 pushFrom, abCode
+            update_rov_video_ids, update_rov_result = [], []
+            if not filtered_result:
+                return update_rov_video_ids, update_rov_result
+            for item in filtered_result:
+                video_id = int(item)
+                rov_score = video_score.get(video_id)
+                if rov_score is None:
+                    continue
+                update_rov_video_ids.append(video_id)
+                update_rov_result.append({'videoId': video_id, 'rovScore': rov_score,
+                                          'pushFrom': 'recall_pool', 'abCode': self.ab_code})
+            return update_rov_video_ids, update_rov_result
+
+        except Exception as e:
+            log_.error(traceback.format_exc())
+            return [], []
+
+