merge top-video

app.py

@@ -39,13 +39,15 @@ def homepage_recommend():
         size = request_data.get('size')
         app_type = request_data.get('appType')
         algo_type = request_data.get('algoType')
+        client_info = request_data.get('clientInfo')
         log_.info('homepage_recommend request data: {}'.format(request_data))
         # size默认为10
         if not size:
             size = 10
         if category_id in config_.CATEGORY['recommend']:
             # 推荐
-            videos = video_recommend(mid=mid, uid=uid, size=size, app_type=app_type, algo_type=algo_type)
+            videos = video_recommend(mid=mid, uid=uid, size=size, app_type=app_type,
+                                     algo_type=algo_type, client_info=client_info)
             result = {'code': 200, 'message': 'success', 'data': {'videos': videos}}
             log_.info('category_id: {}, mid: {}, uid: {}, result: {}, execute time = {}ms'.format(
                 category_id, mid, uid, result, (time.time() - start_time)*1000))

config.py

@@ -50,10 +50,16 @@ class BaseConfig(object):
     # 兜底视频redis存储key
     BOTTOM_KEY_NAME = 'com.weiqu.video.bottom'
 
+    # 生效中的置顶视频列表 redis key
+    TOP_VIDEO_LIST_KEY_NAME = 'com.weiqu.video.top.item.score.area'
+
     # rov召回池更新时间
     ROV_UPDATE_H = 6
     ROV_UPDATE_MINUTE = 30
 
+    # 置顶视频区域 为 全部 的code
+    ALL_AREA_CODE = '000000'
+
 
 class DevelopmentConfig(BaseConfig):
     """测试环境配置"""

recommend.py

@@ -14,7 +14,7 @@ log_ = Log()
 config_ = set_config()
 
 
-def video_recommend(mid, uid, size, app_type, algo_type):
+def video_recommend(mid, uid, size, app_type, algo_type, client_info):
     """
     首页线上推荐逻辑
     :param mid: mid type-string
@@ -22,6 +22,7 @@ def video_recommend(mid, uid, size, app_type, algo_type):
     :param size: 请求视频数量 type-int
     :param app_type: 产品标识  type-int
     :param algo_type: 算法类型  type-string
+    :param client_info: 用户位置信息 {"country": "国家",  "province": "省份",  "city": "城市"}
     :return:
     """
     ab_code = config_.AB_CODE
@@ -44,7 +45,7 @@ def video_recommend(mid, uid, size, app_type, algo_type):
     pool.join()
     '''
     recall_result_list = []
-    pool_recall = PoolRecall(app_type=app_type, mid=mid, uid=uid, ab_code=ab_code)
+    pool_recall = PoolRecall(app_type=app_type, mid=mid, uid=uid, ab_code=ab_code, client_info=client_info)
     _, last_rov_recall_key, _ = pool_recall.get_video_last_idx()
     t = [gevent.spawn(pool_recall.rov_pool_recall, size), gevent.spawn(pool_recall.flow_pool_recall, size) ]
     gevent.joinall(t)
@@ -87,7 +88,7 @@ def video_recommend(mid, uid, size, app_type, algo_type):
         log_.info('preview redis update success!')
 
     # 将此次获取的ROV召回池config_.K末位视频id同步刷新到Redis中，方便下次快速定位到召回位置，过期时间为1天
-    rov_recall_video = [item['videoId'] for item in rank_result if item['pushFrom'] == 'recall_pool']
+    rov_recall_video = [item['videoId'] for item in rank_result[:3] if item['pushFrom'] == 'recall_pool']
     if 0 < len(rov_recall_video) <= config_.K:
         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])
@@ -138,10 +139,9 @@ def video_relevant_recommend(mid, uid, size, app_type):
     :param uid: uid type-string
     :param size: 请求视频数量 type-int
     :param app_type: 产品标识  type-int
-    :param algo_type: 算法类型  type-string
     :return: videos type-list
     """
-    videos = video_recommend(mid=mid, uid=uid, size=size, app_type=app_type, algo_type='')
+    videos = video_recommend(mid=mid, uid=uid, size=size, app_type=app_type, algo_type='', client_info=None)
     return videos
 
 

video_recall.py

@@ -14,10 +14,11 @@ config_ = set_config()
     
 class PoolRecall(object):
     """召回"""
-    def __init__(self, app_type, mid='', uid='', ab_code=''):
+    def __init__(self, app_type, client_info=None, mid='', uid='', ab_code=''):
         """
         初始化
         :param app_type: 产品标识 type-int
+        :param client_info: 用户位置信息 {"country": "国家",  "province": "省份",  "city": "城市"}
         :param mid: mid type-string
         :param uid: uid type-string
         :param ab_code: ab_code type-int
@@ -26,15 +27,30 @@ class PoolRecall(object):
         self.mid = mid
         self.uid = uid
         self.ab_code = ab_code
+        self.client_info = client_info
         self.redis_helper = RedisHelper()
 
     def rov_pool_recall(self, size=10):
         """从ROV召回池中获取视频"""
         log_.info('====== rov pool recall')
 
+        # 获取生效中的置顶视频
+        if self.client_info is None:
+            # 无用户位置信息时，不获取置顶视频
+            top_video_ids, top_video_result = [], []
+        else:
+            top_video_ids, top_video_result = self.get_top_videos()
+            log_.info('top video result = {}'.format(top_video_ids))
+
         # 获取修改过rov的视频
         update_rov_video_ids, update_rov_result = self.get_update_rov_videos()
         log_.info('update rov result = {}'.format(update_rov_video_ids))
+        # 与置顶视频去重
+        update_rov_video_ids_dup, update_rov_dup_result = [], []
+        for item in update_rov_result:
+            if item['videoId'] not in top_video_ids:
+                update_rov_video_ids_dup.append(item['videoId'])
+                update_rov_dup_result.append(item)
 
         # 获取相关redis key, 用户上一次在rov召回池对应的位置
         rov_pool_key, last_rov_recall_key, idx = self.get_video_last_idx()
@@ -68,18 +84,17 @@ class PoolRecall(object):
             video_score = {}
             for value in data:
                 video_id = int(value[0])
-                if video_id in update_rov_video_ids:
+                # 视频id在 生效中的置顶视频 或 修改过rov的视频 中，跳过
+                if video_id in update_rov_video_ids_dup or video_id in top_video_ids:
                     continue
                 video_ids.append(video_id)
                 video_score[video_id] = value[1]
             # 过滤
-            debug_tm_b = time.time()
             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()
             # filtered_result = filter_.filter_videos()
-            debug_tm_e = time.time()
 
             if filtered_result:
                 # 添加视频源参数 pushFrom, abCode
@@ -92,10 +107,12 @@ class PoolRecall(object):
                 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)
+        # 生效中的置顶视频、被修改rov视频、rov召回池视频 归并排序
+        if top_video_result:
+            rov_pool_recall_result.extend(top_video_result)
+        if update_rov_dup_result:
+            rov_pool_recall_result.extend(update_rov_dup_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):
@@ -355,4 +372,52 @@ class PoolRecall(object):
             log_.error(traceback.format_exc())
             return [], []
 
+    def get_top_videos(self):
+        """
+        获取置顶视频
+        :return: top_video_ids, top_video_result
+        """
+        try:
+            # 获取生效中的置顶视频列表
+            redis_helper = RedisHelper()
+            data = redis_helper.get_data_from_redis(key_name=config_.TOP_VIDEO_LIST_KEY_NAME)
+            log_.info('===1===  {}'.format(data))
+            if data is None:
+                return [], []
+            # 获取视频id,并转换类型为int,将videoId和score做mapping，并存储为key-value{videoId: score}
+            video_ids = []
+            video_info = {}
+            for item in eval(data):
+                log_.info('=== item === {}'.format(item))
+                video_id = int(item['videoId'])
+                # 判断 视频推荐区域与用户地址信息 是否匹配，市级别
+                city_code_list = item['cityCode'].split(',')
+                if self.client_info.get('cityCode') in city_code_list or config_.ALL_AREA_CODE in city_code_list:
+                    log_.info('===2=== {}, ===3=== video_id: {}, cityCode: {}'.format(
+                        self.client_info.get('cityCode'), video_id, city_code_list))
+                    video_ids.append(video_id)
+                    video_info[video_id] = {'score': item['score'], 'cityCode': city_code_list}
 
+            # 过滤
+            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 = 'op_manual', abCode
+            top_video_ids, top_video_result = [], []
+            if not filtered_result:
+                return top_video_ids, top_video_result
+            for item in filtered_result:
+                video_id = int(item)
+                item_info = video_info.get(video_id)
+                if item_info is None:
+                    continue
+                top_video_ids.append(video_id)
+                top_video_result.append({'videoId': video_id, 'rovScore': item_info.get('score'),
+                                         'pushFrom': 'op_manual', 'abCode': self.ab_code})
+            return top_video_ids, top_video_result
+
+        except Exception as e:
+            log_.error(traceback.format_exc())
+            return [], []