add recall

video_recall.py

@@ -0,0 +1,187 @@
+import time
+
+from datetime import date, timedelta
+from log import Log
+from db_helper import RedisHelper
+from config import set_config
+from utils import FilterVideos, get_videos_remain_view_count
+
+log_ = Log()
+config_ = set_config()
+
+
+class PoolRecall(object):
+    """召回"""
+    def __init__(self, app_type, mid, uid):
+        """
+        初始化
+        :param app_type: 产品标识 type-int
+        :param mid: mid type-string
+        :param uid: uid type-string
+        """
+        self.app_type = app_type
+        self.mid = mid
+        self.uid = uid
+        self.redis_helper = RedisHelper()
+
+    def rov_pool_recall(self, size=10):
+        """从ROV召回池中获取视频"""
+        log_.info('====== rov pool recall')
+        # 获取相关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 []
+        rov_pool_recall_result = []
+        # 每次获取的视频数
+        get_size = size * 2
+        # 记录获取频次
+        freq = 0
+        while len(rov_pool_recall_result) < size:
+            freq += 1
+            if freq > config_.MAX_FREQ_FROM_ROV_POOL:
+                break
+            # 获取数据
+            st_get = time.time()
+            data = self.redis_helper.get_data_zset_with_index(key_name=rov_pool_key,
+                                                              start=idx, end=idx + get_size - 1,
+                                                              with_scores=True)
+            et_get = time.time()
+            log_.info('get data from rov pool redis: freq = {}, data = {}, execute time = {}ms'.format(
+                freq, data, (et_get - st_get) * 1000))
+            if not data:
+                log_.info('ROV召回池中的视频已取完')
+                break
+            # 获取视频id,并转换类型为int,并存储为key-value{videoId: score}
+            video_ids = []
+            video_score = {}
+            for value in data:
+                video_ids.append(eval(value[0]))
+                video_score[eval(value[0])] = value[1]
+            # 过滤
+            filter_ = FilterVideos(app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
+            filtered_result = filter_.filter_videos()
+            if filtered_result:
+                temp_result = [{'videoId': item, 'rovScore': video_score[item]} for item in filtered_result]
+                rov_pool_recall_result.extend(temp_result)
+            else:
+                # 将此次获取的末位视频id同步刷新到Redis中，方便下次快速定位到召回位置，过期时间为1天
+                self.redis_helper.set_data_to_redis(key_name=last_rov_recall_key, value=data[-1])
+            idx += get_size
+        return rov_pool_recall_result[:size]
+
+    def flow_pool_recall(self, size=10):
+        """从流量池中获取视频"""
+        log_.info('====== flow pool recall')
+        flow_pool_key = config_.FLOW_POOL_KEY_NAME
+        flow_pool_recall_result = []
+        flow_pool_recall_videos = []
+        # 每次获取的视频数
+        get_size = size * 3
+        # 记录获取频次
+        freq = 0
+        idx = 0
+        while len(flow_pool_recall_result) < size:
+            # 获取数据
+            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 与 flow_pool, score做mapping整理
+            video_ids = []
+            video_mapping = {}
+            video_score = {}
+            for value in data:
+                video_id, flow_pool = value[0].split('-')
+                video_id = eval(video_id)
+                if video_id not in video_ids:
+                    video_ids.append(video_id)
+                    video_score[video_id] = value[1]
+                if video_id not in video_mapping:
+                    video_mapping[video_id] = [flow_pool]
+                else:
+                    video_mapping[video_id].append(flow_pool)
+            # 过滤
+            filter_ = FilterVideos(app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
+            filtered_result = filter_.filter_videos()
+            # 检查可分发数
+            if filtered_result:
+                st_check = time.time()
+                check_result = self.check_video_counts(video_ids=filtered_result, flow_pool_mapping=video_mapping)
+                for item in check_result:
+                    if item[0] not in flow_pool_recall_videos:
+                        # 取其中一个 flow_pool 作为召回结果
+                        flow_pool_recall_result.append(
+                            {'videoId': item[0], 'flowPool': item[1], 'rovScore': video_score[item[0]]}
+                        )
+                        flow_pool_recall_videos.append(item[0])
+                et_check = time.time()
+                log_.info('check result: result = {}, execute time = {}ms'.format(
+                    check_result, (et_check - st_check) * 1000))
+            idx += get_size
+
+        return flow_pool_recall_result[:size]
+
+    def check_video_counts(self, video_ids, flow_pool_mapping):
+        """
+        检查视频剩余可分发数
+        :param video_ids: 视频id type-list
+        :param flow_pool_mapping: 视频id-流量池标记mapping, type-dict
+        :return:
+        """
+        videos = []
+        for video_id in video_ids:
+            for flow_pool in flow_pool_mapping[video_id]:
+                videos.append({'videoId': video_id, 'flowPool': flow_pool})
+        check_result = get_videos_remain_view_count(app_type=self.app_type, videos=videos)
+        if not check_result:
+            return None
+        return check_result
+
+    def get_pool_redis_key(self, pool_type):
+        """
+        拼接key
+        :param pool_type: type-string {'rov': rov召回池, 'flow': 流量池}
+        :return: key_name
+        """
+        if pool_type == 'rov':
+            # 判断热度列表是否更新，未更新则使用前一天的热度列表
+            key_name = config_.RECALL_KEY_NAME_PREFIX + time.strftime('%Y%m%d')
+            if self.redis_helper.key_exists(key_name):
+                redis_date = date.today().strftime('%Y%m%d')
+            else:
+                redis_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d')
+                key_name = config_.RECALL_KEY_NAME_PREFIX + redis_date
+                if not self.redis_helper.key_exists(key_name):
+                    return None, None
+            return key_name, redis_date
+
+        elif pool_type == 'flow':
+            return config_.FLOW_POOL_KEY_NAME
+
+        else:
+            log_.error('pool type error')
+            return None, None
+
+    def get_video_last_idx(self):
+        """获取用户上一次在rov召回池对应的位置"""
+        rov_pool_key, redis_date = self.get_pool_redis_key('rov')
+        if not rov_pool_key:
+            return None, None, None
+        last_rov_recall_key = config_.LAST_VIDEO_FROM_ROV_POOL_PREFIX + '{}.{}'.format(self.mid, redis_date)
+        value = self.redis_helper.get_data_from_redis(last_rov_recall_key)
+        if value:
+            idx = self.redis_helper.get_index_with_data(rov_pool_key, value)
+            if not idx:
+                idx = 0
+            else:
+                idx += 1
+        else:
+            idx = 0
+        return rov_pool_key, last_rov_recall_key, idx