rov-offline/videos_filter.py

import time
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_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()
    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()