rov-offline/check_video_limit_distribute.py

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 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

    for region in region_code_list:
        log_.info(f"region = {region}")
        for key, value in rule_params.items():
            log_.info(f"rule = {key}, param = {value}")
            # 将已超分发视频加入到地域小时级线上过滤应用列表中
            redis_helper.add_data_with_set(
                key_name=f"{config_.REGION_H_VIDEO_FILER}{region}.{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}.{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}.{key}",
                values=stop_distribute_video_id_list,
                expire_time=2 * 3600
            )
            # 将已超分发视频 移除 大列表
            key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP_REGION_H}{region}.{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}{region}.{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"region = {region} videos check end!")
    log_.info("region_h videos check end!")

    # 将已超分发视频 移除 原始大列表
    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()