rov-server/video_recall.py

import time
import traceback
import random

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, client_info=None, mid='', uid='', ab_code='',
                 rule_key='', no_op_flag=False):
        """
        初始化
        :param app_type: 产品标识 type-int
        :param client_info: 用户位置信息 {"cityCode": "100000"}
        :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.client_info = client_info
        self.rule_key = rule_key
        self.no_op_flag = no_op_flag
        self.redis_helper = RedisHelper()

    def copy_redis_zset_data(self, from_key_name, to_key_name):
        # 获取from_key_name中的数据
        records = self.redis_helper.get_data_zset_with_index(key_name=from_key_name, start=0, end=-1, with_scores=True)
        if records is not None:
            data = {}
            for video_id, score in records:
                data[int(video_id)] = score
            # 重新写入
            if self.redis_helper.key_exists(to_key_name):
                self.redis_helper.del_keys(key_name=to_key_name)
            self.redis_helper.add_data_with_zset(key_name=to_key_name, data=data, expire_time=2*3600)
            return True
        else:
            return False

    def update_mid_data(self, province_code):
        # mid对应小时级视频列表 redis-key
        h_recall_mid_key = f"{config_.H_WITH_MID_RECALL_KEY_NAME_PREFIX}{self.mid}"
        # 判断mid对应小时级视频列表 时间记录
        h_record_key = f"{config_.H_WITH_MID_RECORD_KEY_NAME_PREFIX}{self.mid}"
        # 判断当前小时的小时级列表是否更新
        now_date = datetime.today()
        h = datetime.now().hour
        now_dt = datetime.strftime(now_date, '%Y%m%d')
        if self.ab_code in [code for _, code in config_.AB_CODE['region_rank_by_h'].items()]:
            key_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H}{province_code}."
        else:
            key_prefix = config_.RECALL_KEY_NAME_PREFIX_BY_H
        now_h_recall_key = f"{key_prefix}{self.rule_key}.{now_dt}.{h}"
        if self.redis_helper.key_exists(key_name=now_h_recall_key):
            flag = self.copy_redis_zset_data(from_key_name=now_h_recall_key, to_key_name=h_recall_mid_key)
            if flag:
                value = {'date': now_dt, 'h': h}
                self.redis_helper.set_data_to_redis(key_name=h_record_key, value=str(value), expire_time=2*3600)
        else:
            if h == 0:
                redis_dt = datetime.strftime(now_date - timedelta(days=1), '%Y%m%d')
                redis_h = 23
            else:
                redis_dt = now_dt
                redis_h = h - 1
            now_h_recall_key = f"{key_prefix}{self.rule_key}.{redis_dt}.{redis_h}"
            flag = self.copy_redis_zset_data(from_key_name=now_h_recall_key, to_key_name=h_recall_mid_key)
            if flag:
                value = {'date': redis_dt, 'h': redis_h}
                self.redis_helper.set_data_to_redis(key_name=h_record_key, value=str(value), expire_time=2*3600)

    def get_mid_h_key(self, province_code):
        # mid对应小时级视频列表 redis-key
        h_recall_mid_key = f"{config_.H_WITH_MID_RECALL_KEY_NAME_PREFIX}{self.mid}"
        # 判断mid对应小时级视频列表 时间记录
        h_record_key = f"{config_.H_WITH_MID_RECORD_KEY_NAME_PREFIX}{self.mid}"
        if not self.redis_helper.key_exists(key_name=h_record_key):
            # ###### 记录key不存在，copy列表，更新记录
            self.update_mid_data(province_code=province_code)
            # return h_recall_mid_key
        else:
            # ###### 记录key存在，判断date, h
            now_date = datetime.today()
            h = datetime.now().hour
            # 获取记录的date, h
            record = self.redis_helper.get_data_from_redis(key_name=h_record_key)
            record_dt = eval(record).get('date')
            record_h = eval(record).get('h')
            now_dt = datetime.strftime(now_date, '%Y%m%d')
            if record_dt == now_dt and int(record_h) == h:
                # 已获取当前小时数据
                pass
                # return h_recall_mid_key
            elif (record_dt == now_dt and h-int(record_h) == 1) or (h == 0 and int(record_h) == 23):
                # 记录的h - 当前h = 1，判断当前h数据是否已更新
                if self.ab_code in [code for _, code in config_.AB_CODE['region_rank_by_h'].items()]:
                    key_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H}{province_code}."
                else:
                    key_prefix = config_.RECALL_KEY_NAME_PREFIX_BY_H
                now_h_recall_key = f"{key_prefix}{self.rule_key}.{now_dt}.{h}"
                # if not self.redis_helper.key_exists(key_name=now_h_recall_key):
                    # 未更新
                    # return h_recall_mid_key
                if self.redis_helper.key_exists(key_name=now_h_recall_key):
                    # 已更新，重新获取更新mid对应列表及记录
                    # self.redis_helper.del_keys(key_name=h_recall_mid_key)
                    # self.redis_helper.del_keys(key_name=h_record_key)
                    flag = self.copy_redis_zset_data(from_key_name=now_h_recall_key, to_key_name=h_recall_mid_key)
                    if flag:
                        new_record = {'date': now_dt, 'h': h}
                        self.redis_helper.set_data_to_redis(key_name=h_record_key, value=str(new_record), expire_time=2*3600)
                    # return h_recall_mid_key
            else:
                self.update_mid_data(province_code=province_code)
                # return h_recall_mid_key

        return h_recall_mid_key

    def rov_pool_recall_by_h(self, size=10, expire_time=24*3600):
        """
        从小时级更新ROV召回池中获取视频
        :param size: 获取视频个数
        :param expire_time: 末位视频记录redis过期时间
        :return:
        """
        # 获取provinceCode
        province_code = self.client_info.get('provinceCode', '-1')
        if province_code == '':
            province_code = '-1'
        # 获取mid对应的小时级列表redis-key
        h_recall_mid_key = self.get_mid_h_key(province_code=province_code)
        if not self.redis_helper.key_exists(h_recall_mid_key):
            recall_result = self.rov_pool_recall(size=size, expire_time=expire_time)
        else:
            # 过滤的视频
            fil_video_ids = []
            recall_result = []
            # 每次获取的视频数
            get_size = size * 5
            # 记录获取频次
            freq = 0
            while len(recall_result) < size:
                freq += 1
                if freq > config_.MAX_FREQ_FROM_ROV_POOL:
                    break
                # 获取数据
                data = self.redis_helper.get_data_zset_with_index(key_name=h_recall_mid_key,
                                                                  start=(freq - 1) * get_size, end=freq * get_size - 1,
                                                                  with_scores=True)
                if not data:
                    log_.info('小时级更新视频已取完')
                    break
                # 获取视频id,并转换类型为int,并存储为key-value{videoId: score}
                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_h, self.rule_key, self.ab_code, province_code)
                ge.join()
                filtered_result = ge.get()

                if filtered_result:
                    # 添加视频源参数 pushFrom, abCode
                    temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)],
                                    'pushFrom': config_.PUSH_FROM['rov_recall_h'], 'abCode': self.ab_code}
                                   for item in filtered_result if video_score.get(int(item)) is not None]
                    recall_result.extend(temp_result)
                    fil_video_ids.extend(list(set(video_ids) - set([item.get('videoId') for item in temp_result])))
                else:
                    fil_video_ids.extend(video_ids)
            # 将被过滤的视频进行移除
            for value in fil_video_ids:
                self.redis_helper.remove_value_from_zset(key_name=h_recall_mid_key, value=value)

            # 判断获取到的小时级数据数量
            if len(recall_result) < size:
                # 补充数据
                rov_recall_result = self.rov_pool_recall(size=size, expire_time=expire_time)
                # 去重合并
                now_video_ids = [item.get('videoId') for item in recall_result]
                for video in rov_recall_result:
                    vid = video.get('videoId')
                    if vid not in now_video_ids:
                        recall_result.append(video)
                        now_video_ids.append(vid)
                        if len(recall_result) >= size:
                            break
                        else:
                            continue

        return recall_result[:size]

    def rov_pool_recall_by_day(self, size=4, expire_time=24*3600):
        """
        从天级规则更新列表中获取视频
        :param size: 获取视频个数
        :param expire_time: 末位视频记录redis过期时间
        :return:
        """
        # 获取天级规则更新列表相关redis key, 用户上一次在天级规则更新列表对应的位置
        rule_key_name, last_rule_day_recall_key, idx = self.get_video_last_idx_day()
        # 获取天级规则更新列表数据
        if not rule_key_name:
            log_.info('天级规则更新列表中无视频')
            recall_result = self.rov_pool_recall(size=size, expire_time=expire_time)
        else:
            recall_result = []
            # 每次获取的视频数
            get_size = size * 5
            # 记录获取频次
            freq = 0
            while len(recall_result) < size:
                freq += 1
                if freq > config_.MAX_FREQ_FROM_ROV_POOL:
                    break
                # 获取数据
                data = self.redis_helper.get_data_zset_with_index(key_name=rule_key_name,
                                                                  start=idx, end=idx + get_size - 1,
                                                                  with_scores=True)
                if not data:
                    log_.info('天级规则更新视频已取完')
                    break
                # 获取视频id,并转换类型为int,并存储为key-value{videoId: score}
                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()

                if filtered_result:
                    # 添加视频源参数 pushFrom, abCode
                    temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)],
                                    'pushFrom': config_.PUSH_FROM['rov_recall_day'], 'abCode': self.ab_code}
                                   for item in filtered_result if video_score.get(int(item)) is not None]
                    recall_result.extend(temp_result)
                else:
                    # 将此次获取的末位视频id同步刷新到Redis中，方便下次快速定位到召回位置，过期时间为1天
                    self.redis_helper.set_data_to_redis(key_name=last_rule_day_recall_key, value=data[-1][0],
                                                        expire_time=expire_time)
                idx += get_size

            # 判断获取到的天级规则数据数量
            if len(recall_result) < size:
                # 补充数据
                rov_recall_result = self.rov_pool_recall(size=size, expire_time=expire_time)
                # 去重合并
                now_video_ids = [item.get('videoId') for item in recall_result]
                for video in rov_recall_result:
                    vid = video.get('videoId')
                    if vid not in now_video_ids:
                        recall_result.append(video)
                        now_video_ids.append(vid)
                        if len(recall_result) >= size:
                            break
                        else:
                            continue
        return recall_result[:size]

    def rov_pool_recall(self, size=10, expire_time=24*3600):
        """
        从ROV召回池中获取视频
        :param size: 获取视频个数
        :param expire_time: 末位视频记录redis过期时间
        :return:
        """
        # log_.info('====== rov pool recall')

        # 获取生效中的置顶视频
        if self.no_op_flag:
            top_video_ids, top_video_result = [], []
        elif 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的视频
        if self.no_op_flag:
            update_rov_video_ids, update_rov_result = [], []
        else:
            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()
        if not rov_pool_key:
            log_.info('ROV召回池中无视频')
            if (not update_rov_dup_result) and (not top_video_result):
                return []
            rov_pool_recall_result = top_video_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]
        rov_pool_recall_result = []
        # 每次获取的视频数
        get_size = size * 5
        # 记录获取频次
        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])
                # 视频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]
            # 过滤
            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:
                # 添加视频源参数 pushFrom, abCode
                temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)],
                                'pushFrom': config_.PUSH_FROM['rov_recall'], '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],
                                                    expire_time=expire_time)
            idx += get_size

        # 生效中的置顶视频、被修改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):
        """从流量池中获取视频"""
        # 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 * 5
        # 记录获取频次
        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': config_.PUSH_FROM['flow_recall'],
                             '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 = 13  票圈视频app
            if self.app_type == config_.APP_TYPE['APP']:
                key_name_prefix = config_.RECALL_KEY_NAME_PREFIX_APP
            # abCode = 30001   # 老好看视频 / 票圈最惊奇 首页/相关推荐逻辑更新实验
            # elif self.ab_code == config_.AB_CODE['rov_rank_appType_18_19']:
            #     key_name_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_APP_TYPE}{self.app_type}."
            # 其他
            else:
                key_name_prefix = config_.RECALL_KEY_NAME_PREFIX

            # 判断热度列表是否更新，未更新则使用前一天的热度列表
            key_name = 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 = 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召回池数据未按时更新，请及时查看解决。'.format(config_.ENV_TEXT)
                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召回池对应的位置"""
        # if self.ab_code in [config_.AB_CODE['rank_by_h']] or self.app_type == config_.APP_TYPE['APP']:
        # abCode = 30001   # 老好看视频 / 票圈最惊奇 首页/相关推荐逻辑更新实验
        if self.ab_code in [code for _, code in config_.AB_CODE['rank_by_h'].items()] + \
                [code for _, code in config_.AB_CODE['region_rank_by_h'].items()] + \
                [config_.AB_CODE['rov_rank_appType_18_19'], config_.AB_CODE['rov_rank_appType_19']] or \
                self.app_type == config_.APP_TYPE['APP']:
            rov_pool_key, redis_date = self.get_pool_redis_key_with_h('rov')

        elif self.ab_code in [code for _, code in config_.AB_CODE['rank_by_day'].items()]:
            rov_pool_key, redis_date = self.get_pool_redis_key_with_day('dup')

        else:
            rov_pool_key, redis_date = self.get_pool_redis_key('rov')

        if not rov_pool_key:
            return None, None, None
        if self.ab_code in [code for _, code in config_.AB_CODE['rank_by_day'].items()]:
            now_h = datetime.now().hour
            if now_h < 7:
                last_key_prefix = config_.LAST_VIDEO_FROM_ROV_POOL_PRE_PREFIX
            else:
                last_key_prefix = config_.LAST_VIDEO_FROM_ROV_POOL_NOW_PREFIX
        else:
            last_key_prefix = config_.LAST_VIDEO_FROM_ROV_POOL_PREFIX
        last_rov_recall_key = f'{last_key_prefix}{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的视频
            if self.app_type == config_.APP_TYPE['APP']:
                key_name = config_.UPDATE_ROV_KEY_NAME_APP
            else:
                key_name = config_.UPDATE_ROV_KEY_NAME
            redis_helper = RedisHelper()
            data = redis_helper.get_data_zset_with_index(key_name=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': config_.PUSH_FROM['rov_recall'], 'abCode': self.ab_code})
            return update_rov_video_ids, update_rov_result

        except Exception as e:
            log_.error(traceback.format_exc())
            return [], []

    def get_top_videos(self):
        """
        获取置顶视频
        :return: top_video_ids, top_video_result
        """
        try:
            # 获取生效中的置顶视频列表
            redis_helper = RedisHelper()
            if self.app_type == config_.APP_TYPE['APP']:
                key_name = config_.TOP_VIDEO_LIST_KEY_NAME_APP
            else:
                key_name = config_.TOP_VIDEO_LIST_KEY_NAME
            data = redis_helper.get_data_from_redis(key_name=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('=== uid: {},  item === {}'.format(self.uid, item))
                video_id = int(item['videoId'])
                # 判断 视频推荐区域与用户地址信息 是否匹配，市级别
                city_code_list = item['cityCode'].split(',')
                # log_.info('=== uid: {},  city_code_list: {},{}, cityCode: {},{} === '.format(
                #     self.uid, city_code_list, type(city_code_list[0]), self.client_info.get('cityCode'),
                #     type(self.client_info.get('cityCode'))))
                if self.client_info.get('cityCode') in city_code_list or config_.ALL_AREA_CODE in city_code_list:
                    # log_.info('=== uid: {}, ===2=== {}, ===3=== video_id: {}, cityCode: {}'.format(
                    #     self.uid, self.client_info.get('cityCode'), video_id, city_code_list))
                    if video_id in video_ids:
                        # 如果video_id有重复，score保留大分值
                        if item['score'] > video_info[video_id].get('score'):
                            video_info[video_id] = {'score': item['score'], 'cityCode': city_code_list}
                        else:
                            continue
                    else:
                        video_ids.append(video_id)
                        video_info[video_id] = {'score': item['score'], 'cityCode': city_code_list}
                else:
                    continue

            # 过滤
            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': config_.PUSH_FROM['top'], 'abCode': self.ab_code})
            return top_video_ids, top_video_result

        except Exception as e:
            log_.error(traceback.format_exc())
            return [], []

    def get_pool_redis_key_with_h(self, pool_type):
        """
        拼接key，获取以小时级别更新的视频列表
        :param pool_type: type-string {'rov': rov召回池, 'flow': 流量池}
        :return: key_name
        """
        if pool_type == 'rov':
            now_date = date.today().strftime('%Y%m%d')
            # 获取当前所在小时
            h = datetime.now().hour

            # appType = 13, 票圈视频APP
            # 数据更新周期：每天07:00-21:00, 2h/次
            if self.app_type == config_.APP_TYPE['APP']:
                if h < 7:
                    key_h = 21
                    key_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d')
                elif h > 21:
                    key_h = 21
                    key_date = now_date
                else:
                    if h % 2 == 0:
                        key_h = h - 1
                        key_date = now_date
                    else:
                        key_h = h
                        key_date = now_date
                # print(key_date, key_h)
                key_name = f'{config_.APP_FINAL_RECALL_KEY_NAME_PREFIX}{key_date}.{key_h}'
                if self.redis_helper.key_exists(key_name):
                    return key_name, key_h
                else:
                    if key_h == 7:
                        redis_h = 21
                        redis_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d')
                    else:
                        redis_h = key_h - 2
                        redis_date = key_date
                    # print(redis_date, redis_h)
                    key_name = f'{config_.APP_FINAL_RECALL_KEY_NAME_PREFIX}{redis_date}.{redis_h}'
                    return key_name, redis_h

            # elif self.app_type in [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]:
            # abCode = 30001   # 老好看视频 / 票圈最惊奇 首页/相关推荐逻辑更新实验
            elif self.ab_code in [config_.AB_CODE['rov_rank_appType_18_19'], config_.AB_CODE['rov_rank_appType_19']]:
                # 判断热度列表是否更新，未更新则使用前一小时的热度列表
                key_name_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_APP_TYPE}{self.app_type}."
                key_name = f"{key_name_prefix}{now_date}.{h}"
                if self.redis_helper.key_exists(key_name):
                    return key_name, h
                else:
                    if h == 0:
                        redis_h = 23
                        redis_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d')
                    else:
                        redis_h = h - 1
                        redis_date = now_date
                    key_name = f"{key_name_prefix}{redis_date}.{redis_h}"
                    # 判断当前时间是否晚于数据正常更新时间，发送消息到飞书
                    now_m = datetime.now().minute
                    feishu_text = '{} —— appType = {}, h = {} 数据未按时更新，请及时查看解决。'.format(
                        config_.ENV_TEXT, self.app_type, h)
                    if now_m > config_.ROV_H_UPDATE_MINUTE:
                        send_msg_to_feishu(feishu_text)
                    return key_name, redis_h

            else:
                # 判断热度列表是否更新，未更新则使用前一小时的热度列表
                if self.ab_code in [code for _, code in config_.AB_CODE['region_rank_by_h'].items()]:
                    # 获取provinceCode
                    province_code = self.client_info.get('provinceCode', '-1')
                    if province_code == '':
                        province_code = '-1'
                    key_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_DUP_REGION_H}{province_code}."
                else:
                    key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP_H
                key_name = f"{key_prefix}{self.rule_key}.{now_date}.{h}"
                if self.redis_helper.key_exists(key_name):
                    return key_name, h
                else:
                    if h == 0:
                        redis_h = 23
                        redis_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d')
                    else:
                        redis_h = h - 1
                        redis_date = now_date
                    key_name = f"{key_prefix}{self.rule_key}.{redis_date}.{redis_h}"
                    # 判断当前时间是否晚于数据正常更新时间，发送消息到飞书
                    now_m = datetime.now().minute
                    feishu_text = '{} —— appType = {}, h = {}, key = {}, province_code = {} 数据未按时更新，请及时查看解决。'.format(
                        config_.ENV_TEXT, self.app_type, h, key_name, self.client_info.get('provinceCode', '-1'))
                    if now_m > config_.ROV_H_UPDATE_MINUTE:
                        send_msg_to_feishu(feishu_text)
                    return key_name, redis_h

        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 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 * 5
        # 记录获取频次
        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]

    def get_pool_redis_key_with_day(self, pool_type):
        """
        拼接key，获取以天级别规则更新的视频列表
        :param pool_type: type-string {'rov': rov召回池, 'flow': 流量池}
        :return: key_name
        """
        now_date = datetime.today()
        now_dt = datetime.strftime(now_date, '%Y%m%d')
        if pool_type == 'rov':
            # 判断列表是否更新，未更新则使用前一天的列表
            rule_key_name = f'{config_.RECALL_KEY_NAME_PREFIX_BY_DAY}{self.rule_key}.{now_dt}'
            if self.redis_helper.key_exists(key_name=rule_key_name):
                return rule_key_name, now_dt
            else:
                redis_dt = datetime.strftime((now_date - timedelta(days=1)), '%Y%m%d')
                rule_key_name = f'{config_.RECALL_KEY_NAME_PREFIX_BY_DAY}{self.rule_key}.{redis_dt}'

                # 判断当前时间是否晚于数据正常更新时间，发送消息到飞书
                now_h = datetime.now().hour
                feishu_text = f'{config_.ENV_TEXT} —— appType = {self.app_type}, date = {now_dt}, ' \
                              f'rule_key = {self.rule_key} 数据未按时更新，请及时查看解决。'
                if now_h > config_.ROV_DAY_UPDATE_MINUTE:
                    send_msg_to_feishu(feishu_text)
                return rule_key_name, redis_dt
        elif pool_type == 'dup':
            now_h = datetime.now().hour
            if now_h < 7:
                dup_key_name_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP_DAY_PRE
            else:
                dup_key_name_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP_DAY_NOW
            dup_key_name = f'{dup_key_name_prefix}{self.rule_key}.{now_dt}'
            if self.redis_helper.key_exists(key_name=dup_key_name):
                return dup_key_name, now_dt
            else:
                redis_dt = datetime.strftime((now_date - timedelta(days=1)), '%Y%m%d')
                dup_key_name = f'{dup_key_name_prefix}{self.rule_key}.{redis_dt}'
                return dup_key_name, redis_dt

        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_day(self):
        """获取用户上一次在天级规则更新列表中对应的位置"""
        rule_key_name, redis_dt = self.get_pool_redis_key_with_day('rov')
        if not rule_key_name:
            return None, None, None
        last_rule_day_recall_key = \
            f'{config_.LAST_VIDEO_FROM_RULE_DAY_POOL_PREFIX}{self.app_type}.{self.mid}.{redis_dt}'
        value = self.redis_helper.get_data_from_redis(last_rule_day_recall_key)
        if value:
            idx = self.redis_helper.get_index_with_data(rule_key_name, value)
            if not idx:
                idx = 0
            else:
                idx += 1
        else:
            idx = 0
        return rule_key_name, last_rule_day_recall_key, idx

    def old_videos_recall(self, size):
        """老视频召回"""
        # 获取老视频
        now_dt = datetime.strftime(datetime.today(), '%Y%m%d')
        key_name = f'{config_.RECALL_KEY_NAME_PREFIX_OLD_VIDEOS}{now_dt}'
        old_videos = self.redis_helper.get_data_from_set(key_name=key_name)
        if not old_videos:
            return []
        # 过滤
        old_video_ids = [int(video_id) for video_id in old_videos]
        filter_ = FilterVideos(app_type=self.app_type, video_ids=old_video_ids, mid=self.mid, uid=self.uid)
        ge = gevent.spawn(filter_.filter_videos)
        ge.join()
        filtered_videos = ge.get()
        if not filtered_videos:
            return []
        # 添加视频源参数 pushFrom, abCode
        old_video_result = [{'videoId': int(item), 'rovScore': 0,
                             'pushFrom': config_.PUSH_FROM['old_video'], 'abCode': self.ab_code}
                            for item in filtered_videos]
        # 随机抽取 size+1 条数据
        random.shuffle(old_video_result)
        return old_video_result[:size+1]

    def get_region_dup_video_last_idx_h(self, province_code, region_dup=None):
        """获取用户上一次在 地域分组 相关去重列表中对应的位置"""
        now_date = date.today().strftime('%Y%m%d')
        h = datetime.now().hour

        if region_dup == 1:
            # 小程序地域分组天级更新结果与小程序地域分组小时级更新结果去重
            key_name_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_DAY_H
        elif region_dup == 2:
            # 小程序天级更新结果与 小程序地域分组天级更新结果/小程序地域分组小时级更新结果 去重
            key_name_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_DAY_H
        else:
            key_name_prefix = ''
        key_name = f"{key_name_prefix}{province_code}.{self.rule_key}.{now_date}.{h}"
        last_region_dup_key = \
            f'{config_.LAST_VIDEO_FROM_REGION_DUP_PREFIX}{region_dup}.{self.app_type}.{self.mid}.{h}'

        if not self.redis_helper.key_exists(key_name=key_name):
            if h == 0:
                redis_h = 23
                redis_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d')
            else:
                redis_h = h - 1
                redis_date = now_date
            key_name = f"{key_name_prefix}{province_code}.{self.rule_key}.{redis_date}.{redis_h}"
            last_region_dup_key = \
                f'{config_.LAST_VIDEO_FROM_REGION_DUP_PREFIX}{region_dup}.{self.app_type}.{self.mid}.{redis_h}'

        value = self.redis_helper.get_data_from_redis(last_region_dup_key)
        if value:
            idx = self.redis_helper.get_index_with_data(key_name, value)
            if not idx:
                idx = 0
            else:
                idx += 1
        else:
            idx = 0
        return key_name, last_region_dup_key, idx

    def rov_pool_recall_with_region(self, size=4, expire_time=24*3600):
        """
        地域分组召回视频
        :param size: 获取视频个数
        :param expire_time: 末位视频记录redis过期时间
        :return:
        """
        # 获取provinceCode
        province_code = self.client_info.get('provinceCode', '-1')
        if province_code == '':
            province_code = '-1'
        t = [gevent.spawn(self.rov_pool_recall_with_region_by_h, province_code, size),
             gevent.spawn(self.region_dup_recall, province_code, 1, size, expire_time),
             gevent.spawn(self.region_dup_recall, province_code, 2, size, expire_time)]
        gevent.joinall(t)
        region_recall_result_list = [i.get() for i in t]
        # 将已获取到的视频按顺序去重合并
        now_video_ids = []
        recall_result = []
        for region_result in region_recall_result_list:
            for video in region_result:
                video_id = video.get('videoId')
                if video_id not in now_video_ids:
                    recall_result.append(video)
                    now_video_ids.append(video_id)
                    if len(recall_result) >= size:
                        break
                    else:
                        continue
        # 判断获取到的小时级数据数量
        if len(recall_result) < size:
            # 补充数据
            rov_recall_result = self.rov_pool_recall(size=size, expire_time=expire_time)
            # 去重合并
            for video in rov_recall_result:
                vid = video.get('videoId')
                if vid not in now_video_ids:
                    recall_result.append(video)
                    now_video_ids.append(vid)
                    if len(recall_result) >= size:
                        break
                    else:
                        continue

        return recall_result[:size]

    def rov_pool_recall_with_region_by_h(self, province_code, size=4):
        """
        地域分组小时级视频召回
        :param size: 视频数
        :param province_code: 省份code
        :return:
        """
        # 获取mid对应的小时级列表redis-key
        h_recall_mid_key = self.get_mid_h_key(province_code=province_code)
        if not self.redis_helper.key_exists(h_recall_mid_key):
            recall_result = []
        else:
            # 过滤的视频
            fil_video_ids = []
            recall_result = []
            # 每次获取的视频数
            get_size = size * 5
            # 记录获取频次
            freq = 0
            while len(recall_result) < size:
                freq += 1
                if freq > config_.MAX_FREQ_FROM_ROV_POOL:
                    break
                # 获取数据
                data = self.redis_helper.get_data_zset_with_index(key_name=h_recall_mid_key,
                                                                  start=(freq - 1) * get_size, end=freq * get_size - 1,
                                                                  with_scores=True)
                if not data:
                    log_.info('地域分组小时级更新视频已取完')
                    break
                # 获取视频id,并转换类型为int,并存储为key-value{videoId: score}
                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_h, self.rule_key, self.ab_code, province_code)
                ge.join()
                filtered_result = ge.get()

                if filtered_result:
                    # 添加视频源参数 pushFrom, abCode
                    temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)],
                                    'pushFrom': config_.PUSH_FROM['rov_recall_region_h'], 'abCode': self.ab_code}
                                   for item in filtered_result if video_score.get(int(item)) is not None]
                    recall_result.extend(temp_result)
                    fil_video_ids.extend(list(set(video_ids) - set([item.get('videoId') for item in temp_result])))
                else:
                    fil_video_ids.extend(video_ids)
            # 将被过滤的视频进行移除
            for value in fil_video_ids:
                self.redis_helper.remove_value_from_zset(key_name=h_recall_mid_key, value=value)

        return recall_result[:size]

    def region_dup_recall(self, province_code, region_dup, size=4, expire_time=23*3600):
        """
        region dup 更新列表视频召回
        :param province_code:
        :param region_dup:
        :param size:
        :param expire_time:
        :return:
        """
        if region_dup == 1:
            push_from = config_.PUSH_FROM['rov_recall_region_day']
        elif region_dup == 2:
            push_from = config_.PUSH_FROM['rov_recall_day']

        # 获取region dup更新列表相关redis key, 用户上一次在列表对应的位置
        key_name, last_region_dup_key, idx = self.get_region_dup_video_last_idx_h(
            province_code=province_code, region_dup=region_dup)
        # 获取天级规则更新列表数据
        if not key_name:
            log_.info(f'region dup 更新列表中无视频, region_dup = {region_dup}')
            recall_result = []
        else:
            recall_result = []
            # 每次获取的视频数
            get_size = size * 5
            # 记录获取频次
            freq = 0
            while len(recall_result) < size:
                freq += 1
                if freq > config_.MAX_FREQ_FROM_ROV_POOL:
                    break
                # 获取数据
                data = self.redis_helper.get_data_zset_with_index(key_name=key_name,
                                                                  start=idx, end=idx + get_size - 1,
                                                                  with_scores=True)
                if not data:
                    log_.info(f'region dup 更新视频已取完, region_dup = {region_dup}')
                    break
                # 获取视频id,并转换类型为int,并存储为key-value{videoId: score}
                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()

                if filtered_result:
                    # 添加视频源参数 pushFrom, abCode
                    temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)],
                                    'pushFrom': push_from, 'abCode': self.ab_code}
                                   for item in filtered_result if video_score.get(int(item)) is not None]
                    recall_result.extend(temp_result)
                else:
                    # 将此次获取的末位视频id同步刷新到Redis中，方便下次快速定位到召回位置，过期时间为1天
                    self.redis_helper.set_data_to_redis(key_name=last_region_dup_key, value=data[-1][0],
                                                        expire_time=expire_time)
                idx += get_size

        return recall_result[:size]