add flow_pool_18_19

recommend.py

@@ -176,6 +176,9 @@ def video_recommend(mid, uid, size, top_K, flow_pool_P, app_type, algo_type, cli
     if ab_code in config_.AB_CODE['rank_by_h']:
         t = [gevent.spawn(pool_recall.rov_pool_recall_by_h, size, expire_time),
              gevent.spawn(pool_recall.flow_pool_recall, size)]
+    elif ab_code == config_.AB_CODE['rov_rank_appType_18_19']:
+        t = [gevent.spawn(pool_recall.rov_pool_recall, size, expire_time),
+             gevent.spawn(pool_recall.flow_pool_recall_18_19, size)]
     else:
         t = [gevent.spawn(pool_recall.rov_pool_recall, size, expire_time),
              gevent.spawn(pool_recall.flow_pool_recall, size)]

video_recall.py

@@ -716,3 +716,51 @@ class PoolRecall(object):
         else:
             log_.error('pool type error')
             return None, None
+
+    def flow_pool_recall_18_19(self, size=10):
+        """从流量池中获取视频"""
+        # log_.info('====== flow pool recall')
+        flow_pool_key = self.get_pool_redis_key('flow')
+        flow_pool_recall_result = []
+        flow_pool_recall_videos = []
+        # 每次获取的视频数
+        get_size = size * 3
+        # 记录获取频次
+        freq = 0
+        idx = 0
+        while len(flow_pool_recall_result) < size:
+            freq += 1
+            # 获取数据
+            st_get = time.time()
+            data = self.redis_helper.get_data_zset_with_index(key_name=flow_pool_key,
+                                                              start=idx, end=idx + get_size - 1,
+                                                              with_scores=True)
+            et_get = time.time()
+            # log_.info('get data from flow pool redis: freq = {}, data = {}, execute time = {}ms'.format(
+            #     freq, data, (et_get - st_get) * 1000))
+            if not data:
+                log_.info('流量池中的视频已取完')
+                break
+            # 将video_id 与 score做mapping整理
+            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, mid=self.mid, uid=self.uid, video_ids=video_ids)
+            ge = gevent.spawn(filter_.filter_videos)
+            ge.join()
+            filtered_result = ge.get()
+            # 添加视频源参数 pushFrom, abCode
+            if filtered_result:
+                # 添加视频源参数 pushFrom, abCode
+                temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)],
+                                'pushFrom': config_.PUSH_FROM['flow_recall'], 'abCode': self.ab_code}
+                               for item in filtered_result if video_score.get(int(item)) is not None]
+                flow_pool_recall_result.extend(temp_result)
+            idx += get_size
+
+        return flow_pool_recall_result[:size]