rov-offline/videos_filter.py

import time
import json
import traceback
from datetime import date, timedelta, datetime

from utils import filter_video_status, send_msg_to_feishu
from db_helper import RedisHelper
from config import set_config
from log import Log

config_, env = set_config()
log_ = Log()


def filter_relevant_videos():
    """运营强插相关推荐视频过滤"""
    log_.info("relevant videos with op filter start...")
    # 读取需要过滤的头部视频id
    redis_helper = RedisHelper()
    head_videos = redis_helper.get_data_from_set(key_name=config_.RELEVANT_TOP_VIDEOS_KEY_NAME)
    if head_videos is None or len(head_videos) == 0:
        log_.info("relevant videos with op filter end! head_videos = {}".format(head_videos))
        return

    # 过滤
    remove_head_vids = []
    for head_vid in head_videos:
        key_name = '{}{}'.format(config_.RELEVANT_VIDEOS_WITH_OP_KEY_NAME, head_vid)
        # 头部视频 对应的key不存在时，将head_vid移除对应redis
        if not redis_helper.key_exists(key_name=key_name):
            remove_head_vids.append(head_vid)
            log_.info('head_vid = {} relevant redis key not exist!'.format(head_vid))
            continue

        # 获取头部视频对应的相关视频
        relevant_videos = redis_helper.get_data_from_redis(key_name=key_name)
        # 该视频没有指定的相关性视频，将head_vid移除对应redis
        if relevant_videos is None:
            remove_head_vids.append(head_vid)
            log_.info('head_vid = {} not have relevant videos!'.format(head_vid))
            continue
        # 过滤
        relevant_videos = json.loads(relevant_videos)
        relevant_video_ids = [int(item['recommend_vid']) for item in relevant_videos]
        filtered_videos = filter_video_status(video_ids=relevant_video_ids)
        # 保留可推荐 且生效中 的视频
        relevant_videos_new = [
            item for item in relevant_videos
            if int(item['recommend_vid']) in filtered_videos and int(item['finish_time']) <= int(time.time())
        ]

        # 过滤后没有符合的视频，将head_vid移除对应redis，删除对应的相关推荐的key
        if len(relevant_videos_new) == 0:
            remove_head_vids.append(head_vid)
            redis_helper.del_keys(key_name=key_name)
            log_.info('head_vid = {} filtered finished! new relevant videos count = {}'.format(
                head_vid, len(relevant_videos_new)))
            continue

        # 重新写入redis
        # 以最晚结束的视频的结束时间 - 当前时间 + 5s 作为key的过期时间
        finish_time_list = [item['finish_time'] for item in relevant_videos_new]
        expire_time = max(finish_time_list) - int(time.time()) + 5
        if expire_time <= 0:
            log_.info('head_vid = {} expire_time <= 0!'.format(head_vid))
            continue
        # 存入redis
        redis_helper.set_data_to_redis(key_name=key_name,
                                       value=json.dumps(relevant_videos_new),
                                       expire_time=expire_time)
        log_.info('head_vid = {} filtered finished! new relevant videos count = {}'.format(
                head_vid, len(relevant_videos_new)))

    # 将需要移除的头部视频id进行移除
    redis_helper.remove_value_from_set(key_name=config_.RELEVANT_TOP_VIDEOS_KEY_NAME, values=tuple(remove_head_vids))
    log_.info('head videos remove finished! remove_head_vids = {}'.format(remove_head_vids))
    log_.info("relevant videos with op filter end!")


def filter_rov_pool(app_type=None):
    """ROV召回池视频过滤"""
    log_.info("rov recall pool filter start ...")
    # 拼接redis-key
    if app_type is None:
        key_name, _ = get_pool_redis_key(pool_type='rov')
    else:
        log_.info("appType = {}".format(app_type))
        key_name, _ = get_pool_redis_key(pool_type='rov', app_type=app_type)

    # 获取视频
    redis_helper = RedisHelper()
    data = redis_helper.get_data_zset_with_index(key_name=key_name, start=0, end=-1)
    if data is None:
        log_.info("data is None")
        log_.info("rov recall pool filter end!")
        return
    # 过滤
    video_ids = [int(video_id) for video_id in data]
    filtered_result = filter_video_status(video_ids=video_ids)
    # 求差集，获取需要过滤掉的视频，并从redis中移除
    filter_videos = set(video_ids) - set(filtered_result)
    log_.info("video_ids size = {}, filtered size = {}, filter sizer = {}".format(len(video_ids),
                                                                                  len(filtered_result),
                                                                                  len(filter_videos)))
    if len(filter_videos) == 0:
        log_.info("rov recall pool filter end!")
        return
    redis_helper.remove_value_from_zset(key_name=key_name, value=list(filter_videos))
    log_.info("rov recall pool filter end!")


def filter_flow_pool():
    """流量池视频过滤"""
    log_.info("flow pool filter start ...")
    for _, app_type in config_.APP_TYPE.items():
        log_.info('app_type {} videos filter start...'.format(app_type))
        # 拼接redis-key
        key_name = get_pool_redis_key(pool_type='flow', app_type=app_type)
        # 获取视频
        redis_helper = RedisHelper()
        data = redis_helper.get_data_zset_with_index(key_name=key_name, start=0, end=-1)
        if data is None:
            log_.info("data is None")
            log_.info("app_type {} videos filter end!".format(app_type))
            continue
        # videoId与flowPool做mapping
        video_ids = []
        mapping = {}
        for video in data:
            video_id, flow_pool = video.split('-')
            video_id = int(video_id)
            if video_id not in video_ids:
                video_ids.append(video_id)
                mapping[video_id] = [flow_pool]
            else:
                mapping[video_id].append(flow_pool)
        # 过滤
        if len(video_ids) == 0:
            log_.info("data size = {}, video_ids size = {}, data = {}".format(len(data), len(video_ids), data))
            log_.info("app_type {} videos filter end!".format(app_type))
            continue
        filtered_result = filter_video_status(video_ids=video_ids)
        # 求差集，获取需要过滤掉的视频，并从redis中移除
        filter_videos = set(video_ids) - set(filtered_result)
        log_.info("data size = {}, video_ids size = {}, filtered size = {}, filter sizer = {}".format(
            len(data), len(video_ids), len(filtered_result), len(filter_videos)))
        # 移除
        if len(filter_videos) == 0:
            log_.info("app_type {} videos filter end!".format(app_type))
            continue
        remove_videos = ['{}-{}'.format(video_id, flow_pool)
                         for video_id in filter_videos
                         for flow_pool in mapping[video_id]]
        redis_helper.remove_value_from_zset(key_name=key_name, value=remove_videos)
        log_.info("app_type {} videos filter end!".format(app_type))
    log_.info("flow pool filter end!")


def filter_bottom():
    """兜底视频过滤"""
    log_.info("bottom videos filter start ...")
    # 获取视频
    redis_helper = RedisHelper()
    data = redis_helper.get_data_zset_with_index(key_name=config_.BOTTOM_KEY_NAME, start=0, end=-1)
    if data is None:
        log_.info("data is None")
        log_.info("bottom videos filter end!")
        return
    # 过滤
    video_ids = [int(video_id) for video_id in data]
    filtered_result = filter_video_status(video_ids=video_ids)
    # 求差集，获取需要过滤掉的视频，并从redis中移除
    filter_videos = set(video_ids) - set(filtered_result)
    log_.info("video_ids size = {}, filtered size = {}, filter sizer = {}".format(len(video_ids),
                                                                                  len(filtered_result),
                                                                                  len(filter_videos)))
    if len(filter_videos) == 0:
        log_.info("bottom videos filter end!")
        return
    redis_helper.remove_value_from_zset(key_name=config_.BOTTOM_KEY_NAME, value=list(filter_videos))
    log_.info("bottom videos filter end!")


def filter_rov_updated():
    """修改过ROV的视频过滤"""
    log_.info("update rov videos filter start ...")
    # 获取视频
    redis_helper = RedisHelper()
    data = redis_helper.get_data_zset_with_index(key_name=config_.UPDATE_ROV_KEY_NAME, start=0, end=-1)
    if data is None:
        log_.info("data is None")
        log_.info("update rov videos filter end!")
        return
    # 过滤
    video_ids = [int(video_id) for video_id in data]
    filtered_result = filter_video_status(video_ids=video_ids)
    # 求差集，获取需要过滤掉的视频，并从redis中移除
    filter_videos = set(video_ids) - set(filtered_result)
    log_.info("video_ids size = {}, filtered size = {}, filter sizer = {}".format(len(video_ids),
                                                                                  len(filtered_result),
                                                                                  len(filter_videos)))
    if len(filter_videos) == 0:
        log_.info("update rov videos filter end!")
        return
    redis_helper.remove_value_from_zset(key_name=config_.UPDATE_ROV_KEY_NAME, value=list(filter_videos))
    log_.info("update rov videos filter end!")


def get_pool_redis_key(pool_type, app_type=None):
    """
    拼接key
    :param pool_type: type-string {'rov': rov召回池, 'flow': 流量池}
    :param app_type: 产品标识
    :return: key_name
    """
    redis_helper = RedisHelper()
    if pool_type == 'rov':
        # appType = 6
        if app_type == config_.APP_TYPE['SHORT_VIDEO']:
            # 获取当前所在小时
            redis_date = datetime.now().hour
            # 判断热度列表是否更新，未更新则使用前一小时的热度列表
            key_name = '{}{}.{}'.format(config_.RECALL_KEY_NAME_PREFIX_APP_TYPE, app_type, redis_date)
            if redis_helper.key_exists(key_name):
                return key_name, redis_date
            else:
                if redis_date == 0:
                    redis_date = 23
                else:
                    redis_date = redis_date - 1
                key_name = '{}{}.{}'.format(config_.RECALL_KEY_NAME_PREFIX_APP_TYPE, app_type, redis_date)

                return key_name, redis_date
        # 其他
        else:
            # 判断热度列表是否更新，未更新则使用前一天的热度列表
            key_name = config_.RECALL_KEY_NAME_PREFIX + time.strftime('%Y%m%d')
            if 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

            return key_name, redis_date

    elif pool_type == 'flow':
        # 流量池
        return config_.FLOWPOOL_KEY_NAME_PREFIX + str(app_type)

    else:
        log_.error('pool type error')
        return None, None


def main():
    try:
        # ROV召回池视频过滤
        filter_rov_pool()
        # appType = 6，ROV召回池视频过滤
        filter_rov_pool(app_type=config_.APP_TYPE['SHORT_VIDEO'])
        # 流量池视频过滤
        filter_flow_pool()
        # 兜底视频过滤
        filter_bottom()
        # 修改过ROV的视频过滤
        filter_rov_updated()
        # 运营强插相关推荐视频过滤
        filter_relevant_videos()
    except Exception as e:
        log_.error(traceback.format_exc())
        send_msg_to_feishu('{} - 过滤失败 \n {}'.format(config_.ENV_TEXT, traceback.format_exc()))
        return


if __name__ == '__main__':
    main()