rov-server/video_recall.py

import time
import traceback

from datetime import date, timedelta, datetime
from log import Log
from db_helper import RedisHelper
from config import set_config
from utils import FilterVideos, get_videos_remain_view_count, get_videos_local_distribute_count, send_msg_to_feishu
import gevent

log_ = Log()
config_ = set_config()

    
class PoolRecall(object):
    """召回"""
    def __init__(self, app_type, mid='', uid='', ab_code=''):
        """
        初始化
        :param app_type: 产品标识 type-int
        :param mid: mid type-string
        :param uid: uid type-string
        :param ab_code: ab_code type-int
        """
        self.app_type = app_type
        self.mid = mid
        self.uid = uid
        self.ab_code = ab_code
        self.redis_helper = RedisHelper()

    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召回池中无视频')
            if not update_rov_result:
                return []
            return update_rov_result
        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_id = int(value[0])
                if video_id in update_rov_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
                temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)],
                                'pushFrom': 'recall_pool', 'abCode': self.ab_code}
                               for item in filtered_result if video_score.get(int(item)) is not None]
                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][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):
        """从流量池中获取视频"""
        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 与 flow_pool, score做mapping整理
            video_ids = []
            video_mapping = {}
            video_score = {}
            for value in data:
                video_id, flow_pool = value[0].split('-')
                video_id = int(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)
            ge = gevent.spawn(filter_.filter_videos)
            ge.join()
            filtered_result = ge.get()
            #filtered_result = filter_.filter_videos()
            # 检查可分发数
            if filtered_result:
                st_check = time.time()
                ge = gevent.spawn(self.check_video_counts, video_ids=filtered_result, flow_pool_mapping=video_mapping)
                ge.join()
                check_result, error_flag = ge.get()
                #check_result, error_flag = self.check_video_counts(video_ids=filtered_result, flow_pool_mapping=video_mapping)

                for item in check_result:
                    video_id = int(item[0])
                    flow_pool = item[1]
                    if video_id not in flow_pool_recall_videos:
                        # 取其中一个 flow_pool 作为召回结果
                        # 添加视频源参数 pushFrom, abCode
                        flow_pool_recall_result.append(
                            {'videoId': video_id, 'flowPool': flow_pool,
                             'rovScore': video_score[video_id], 'pushFrom': 'flow_pool', 'abCode': self.ab_code}
                        )
                        # flow_pool_recall_result.append(
                        #     {'videoId': item[0], 'flowPool': item[1], 'distributeCount': item[2],
                        #      'rovScore': video_score[item[0]], 'pushFrom': 'flow_pool', 'abCode': self.ab_code}
                        # )
                        flow_pool_recall_videos.append(video_id)
                et_check = time.time()
                log_.info('check result: result = {}, execute time = {}ms'.format(
                    check_result, (et_check - st_check) * 1000))

                # 判断错误标记, True为错误
                if error_flag:
                    # 结束流量池召回
                    break

            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: check_result, error_flag
        """
        # flow_pool_key = self.get_pool_redis_key('flow')
        videos = []
        check_result = []
        for video_id in video_ids:
            video_id = int(video_id)
            for flow_pool in flow_pool_mapping.get(video_id, []):
                # 判断是否有本地分发记录
                cur_count = get_videos_local_distribute_count(video_id=video_id, flow_pool=flow_pool)
                # 无记录
                if cur_count is None:
                    videos.append({'videoId': video_id, 'flowPool': flow_pool})
                # 本地分发数 cur_count > 0
                elif cur_count > 0:
                    check_result.append((video_id, flow_pool))
                # 本地分发数 cur_count <= 0，从所有的流量召回池移除
                else:
                    value = '{}-{}'.format(video_id, flow_pool)
                    for item in config_.APP_TYPE:
                        flow_pool_key = config_.FLOW_POOL_KEY_NAME_PREFIX + str(config_.APP_TYPE.get(item))
                        self.redis_helper.remove_value_from_zset(key_name=flow_pool_key, value=value)
        # 本次视频都有本地记录
        if len(videos) == 0:
            error_flag = False
            return check_result, error_flag

        # 本地无记录视频，检查实时分发数
        view_count_result, error_flag = get_videos_remain_view_count(app_type=self.app_type, videos=videos)
        log_.info('error_flag = {}, view_count_result = {}'.format(error_flag, view_count_result))

        # 判断返回的错误标记，True为错误
        if error_flag:
            return check_result, error_flag
            # 从流量召回池移除视频videos
            # for item in videos:
            #     value = '{}-{}'.format(item['videoId'], item['flowPool'])
            #     self.redis_helper.remove_value_from_zset(key_name=flow_pool_key, value=value)

        redis_helper = RedisHelper()
        for item in view_count_result:
            try:
                # 接口超时，item[2]可能为None
                remain_count = int(item[2])
            except Exception as e:
                log_.error('remain_count type error...')
                log_.error(traceback.format_exc())
                continue
            if remain_count > 0:
                # viewCount > 0
                check_result.append(item)
                # 将分发数更新到本地记录
                key_name = '{}{}.{}'.format(config_.LOCAL_DISTRIBUTE_COUNT_PREFIX, item[0], item[1])
                redis_helper.setnx_key(key_name=key_name, value=remain_count, expire_time=5 * 60)
            else:
                # viewCount <= 0
                # 从流量召回池移除
                value = '{}-{}'.format(item[0], item[1])
                for item in config_.APP_TYPE:
                    flow_pool_key = config_.FLOW_POOL_KEY_NAME_PREFIX + str(config_.APP_TYPE.get(item))
                    self.redis_helper.remove_value_from_zset(key_name=flow_pool_key, value=value)
        return check_result, error_flag

    def get_pool_redis_key(self, pool_type):
        """
        拼接key
        :param pool_type: type-string {'rov': rov召回池, 'flow': 流量池}
        :return: key_name
        """
        if pool_type == 'rov':
            # appType = 6
            if self.app_type == config_.APP_TYPE['SHORT_VIDEO']:
                # 获取当前所在小时
                redis_date = datetime.now().hour
                # 判断热度列表是否更新，未更新则使用前一小时的热度列表
                key_name = '{}{}.{}'.format(config_.RECALL_KEY_NAME_PREFIX_APP_TYPE, self.app_type, redis_date)
                if self.redis_helper.key_exists(key_name):
                    return key_name, redis_date
                else:
                    key_name = '{}{}.{}'.format(config_.RECALL_KEY_NAME_PREFIX_APP_TYPE, self.app_type, redis_date - 1)
                    return key_name, redis_date - 1
                # 其他
            else:
                # 判断热度列表是否更新，未更新则使用前一天的热度列表
                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

                    # 判断当前时间是否晚于rov召回池更新时间 + 1h，发送消息到飞书
                    now_h = datetime.now().hour
                    now_m = datetime.now().minute
                    feishu_text = '生产环境 —— 今日ROV召回池数据未按时更新，请及时查看解决。'
                    if now_h == config_.ROV_UPDATE_H + 1 and now_m > config_.ROV_UPDATE_MINUTE:
                        send_msg_to_feishu(feishu_text)
                    elif now_h > config_.ROV_UPDATE_H + 2:
                        send_msg_to_feishu(feishu_text)

                return key_name, redis_date

        elif pool_type == 'flow':
            return config_.FLOW_POOL_KEY_NAME_PREFIX + str(self.app_type)

        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.app_type, 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

    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 [], []