rov-offline/check_video_limit_distribute.py

import gevent
import datetime
import numpy as np
from config import set_config
from log import Log
from utils import RedisHelper

config_, _ = set_config()
log_ = Log()
redis_helper = RedisHelper()


def update_limit_video_score(initial_videos, key_name):
    """
    调整限流视频的分数: 将视频移至所在列表的中位数之后，多个视频时，按照原本的顺序进行排列
    :param initial_videos: 视频列表及score type-dict, {videoId: score, ...}
    :param key_name: 视频列表对应的key
    :return:
    """
    if not initial_videos:
        return
    # 获取当前限流视频
    data = redis_helper.get_data_from_redis(key_name=config_.KEY_NAME_PREFIX_LIMIT_VIDEOS)
    if data is None:
        return
    # 获取限流视频对应的score
    limit_video_initial_score = []
    for video in eval(data):
        video_id = int(video[0])
        initial_score = initial_videos.get(video_id, None)
        if initial_score is not None:
            limit_video_initial_score.append((video_id, initial_score))

    log_.info(f"limit_video_initial_score = {limit_video_initial_score}")

    if len(limit_video_initial_score) == 0:
        return

    # 获取原始列表的分数的中位数
    initial_video_score_list = sorted([val for key, val in initial_videos.items()], reverse=False)
    media_score = np.median(initial_video_score_list)
    # 取中位数后一位
    if len(initial_video_score_list) % 2 == 0:
        temp_index = len(initial_video_score_list)//2
    else:
        temp_index = len(initial_video_score_list) // 2 + 1
    if len(initial_video_score_list) > 1:
        temp_score = initial_video_score_list[temp_index]
    else:
        temp_score = 0

    # 对限流视频score进行调整
    limit_video_final_score = {}
    limit_video_initial_score.sort(key=lambda x: x[1], reverse=True)
    limit_video_id_list = []
    for video_id, initial_score in limit_video_initial_score:
        if initial_score > media_score:
            limit_video_id_list.append(video_id)
    if len(limit_video_id_list) > 0:
        limit_score_step = (temp_score - media_score) / (len(limit_video_id_list) + 1)
        for i, video_id in enumerate(limit_video_id_list):
            final_score = media_score - limit_score_step * (i + 1)
            limit_video_final_score[int(video_id)] = final_score

    log_.info(f"media_score = {media_score}, temp_score = {temp_score}, "
              f"limit_video_final_score = {limit_video_final_score}")

    # 更新限流视频的score
    if len(limit_video_final_score) == 0:
        return
    redis_helper.add_data_with_zset(key_name=key_name, data=limit_video_final_score)


def check_videos_distribute():
    """
    检查当前限流视频分发数
    :return: stop_distribute_video_id_list
    """
    # 获取当前限流视频及最大分发数
    data = redis_helper.get_data_from_redis(key_name=config_.KEY_NAME_PREFIX_LIMIT_VIDEOS)
    if data is None:
        return []
    # 判断是否已超分发
    stop_distribute_video_id_list = []
    for video_id, max_distribute_count in eval(data):
        distributed_count = redis_helper.get_data_from_redis(
            key_name=f"{config_.KEY_NAME_PREFIX_LIMIT_VIDEO_DISTRIBUTE_COUNT}{video_id}"
        )
        if distributed_count is None:
            continue
        if int(distributed_count) >= int(max_distribute_count):
            stop_distribute_video_id_list.append(int(video_id))

    return stop_distribute_video_id_list


def process_with_region(app_type, data_key, rule_key, region, stop_distribute_video_id_list, now_date, now_h):
    log_.info(f"app_type = {app_type}, data_key = {data_key}, rule_key = {rule_key}, region = {region}")
    # 将已超分发视频加入到地域小时级线上过滤应用列表中
    redis_helper.add_data_with_set(
        key_name=f"{config_.REGION_H_VIDEO_FILER}{region}.{app_type}.{data_key}.{rule_key}",
        values=stop_distribute_video_id_list,
        expire_time=2 * 3600
    )
    # 将已超分发视频加入到地域分组24h的数据线上过滤应用列表中
    redis_helper.add_data_with_set(
        key_name=f"{config_.REGION_H_VIDEO_FILER_24H}{region}.{app_type}.{data_key}.{rule_key}",
        values=stop_distribute_video_id_list,
        expire_time=2 * 3600
    )
    # 将已超分发视频加入到不区分相对24h线上过滤应用列表中
    # redis_helper.add_data_with_set(
    #     key_name=f"{config_.H_VIDEO_FILER_24H}{region}.{app_type}.{data_key}.{rule_key}",
    #     values=stop_distribute_video_id_list,
    #     expire_time=2 * 3600
    # )

    if rule_key == 'rule4':
        key_prefix_list = [
            config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H,  # 地域分组相对24h列表
            config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_24H_H,  # 不区分地域相对24h列表
            config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H,  # 不区分地域相对24h列表2
            config_.RECALL_KEY_NAME_PREFIX_DUP_REGION_H,  # 大列表
        ]
    else:
        key_prefix_list = [
            config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H,  # 地域分组相对24h列表
            config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_24H_H,  # 不区分地域相对24h列表
            config_.RECALL_KEY_NAME_PREFIX_DUP_REGION_H,  # 大列表
        ]

    for key_prefix in key_prefix_list:
        key_name = f"{key_prefix}{region}.{app_type}.{data_key}.{rule_key}." \
                   f"{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
        if not redis_helper.key_exists(key_name=key_name):
            if now_h == 0:
                redis_date = now_date - datetime.timedelta(days=1)
                redis_h = 23
            else:
                redis_date = now_date
                redis_h = now_h - 1
            key_name = f"{key_prefix}{region}.{app_type}.{data_key}.{rule_key}." \
                       f"{datetime.datetime.strftime(redis_date, '%Y%m%d')}.{redis_h}"
        redis_helper.remove_value_from_zset(key_name=key_name, value=stop_distribute_video_id_list)

    """
    # 将已超分发视频 移除 不区分相对24h列表2
    if rule_key == 'rule4':
        key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H}" \
                   f"{region}.{app_type}.{data_key}.{rule_key}." \
                   f"{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
        if not redis_helper.key_exists(key_name=key_name):
            if now_h == 0:
                redis_date = now_date - datetime.timedelta(days=1)
                redis_h = 23
            else:
                redis_date = now_date
                redis_h = now_h - 1
            key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H}" \
                       f"{region}.{app_type}.{data_key}.{rule_key}." \
                       f"{datetime.datetime.strftime(redis_date, '%Y%m%d')}.{redis_h}"
        redis_helper.remove_value_from_zset(key_name=key_name, value=stop_distribute_video_id_list)

    # 将已超分发视频 移除 大列表
    key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP_REGION_H}" \
               f"{region}.{app_type}.{data_key}.{rule_key}." \
               f"{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
    if not redis_helper.key_exists(key_name=key_name):
        if now_h == 0:
            redis_date = now_date - datetime.timedelta(days=1)
            redis_h = 23
        else:
            redis_date = now_date
            redis_h = now_h - 1
        key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP_REGION_H}" \
                   f"{region}.{app_type}.{data_key}.{rule_key}." \
                   f"{datetime.datetime.strftime(redis_date, '%Y%m%d')}.{redis_h}"
    redis_helper.remove_value_from_zset(key_name=key_name, value=stop_distribute_video_id_list)
    """

    log_.info(f"app_type = {app_type}, data_key = {data_key}, rule_key = {rule_key}, region = {region} "
              f"videos check end!")


def check_region_videos():
    """检查限流视频分发数"""
    # 获取当前日期
    now_date = datetime.datetime.today()
    # 获取当前所在小时
    now_h = datetime.datetime.now().hour
    log_.info(f'now_date = {now_date}, now_h = {now_h}.')

    # 获取已超分发视频
    stop_distribute_video_id_list = check_videos_distribute()
    log_.info(f"stop_distribute_video_id_list = {stop_distribute_video_id_list}, "
              f"count = {len(stop_distribute_video_id_list)}")
    if len(stop_distribute_video_id_list) == 0:
        return

    # 对已超分发的视频进行移除
    region_code_list = [code for region, code in config_.REGION_CODE.items()]
    rule_params = config_.RULE_PARAMS_REGION_APP_TYPE

    for app_type, params in rule_params.items():
        log_.info(f"app_type = {app_type}")
        for param in params.get('params_list'):
            data_key = param.get('data')
            rule_key = param.get('rule')
            log_.info(f"data_key = {data_key}, rule_key = {rule_key}")
            task_list = [
                gevent.spawn(process_with_region, app_type, data_key, rule_key, region,
                             stop_distribute_video_id_list, now_date, now_h)
                for region in region_code_list
            ]
            gevent.joinall(task_list)

        # for data_key, data_param in params['data_params'].items():
        #     log_.info(f"data_key = {data_key}, data_param = {data_param}")
        #     for rule_key, rule_param in params['rule_params'].items():
        #         log_.info(f"rule_key = {rule_key}, rule_param = {rule_param}")
        #         task_list = [
        #             gevent.spawn(process_with_region, app_type, data_key, rule_key, region,
        #                          stop_distribute_video_id_list, now_date, now_h)
        #             for region in region_code_list
        #         ]
        #         gevent.joinall(task_list)

    # 将已超分发视频 移除 原始大列表
    key_name = f"{config_.RECALL_KEY_NAME_PREFIX}{datetime.datetime.strftime(now_date, '%Y%m%d')}"
    if not redis_helper.key_exists(key_name=key_name):
        redis_date = now_date - datetime.timedelta(days=1)
        key_name = f"{config_.RECALL_KEY_NAME_PREFIX}{datetime.datetime.strftime(redis_date, '%Y%m%d')}"
    redis_helper.remove_value_from_zset(key_name=key_name, value=stop_distribute_video_id_list)


if __name__ == '__main__':
    check_region_videos()