rov-offline/check_video_limit_distribute.py

import gevent
import datetime
import numpy as np
from my_config import set_config
from log import Log
from my_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, expire_time=2 * 24 * 3600)
    return 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(data_key, rule_key, region, stop_distribute_video_id_list, now_date, now_h):
    log_.info(f"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
    # )

    key_prefix_list = [
        config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H,  # 地域分组小时级列表
        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,  # 大列表
    ]

    # if rule_key == 'rule4':
    #     key_prefix_list = [
    #         config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H,  # 地域分组小时级列表
    #         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,  # 大列表
    #     ]
    # elif rule_key == 'rule5':
    #     key_prefix_list = [
    #         config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H,  # 地域分组小时级列表
    #         config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H,  # 地域分组相对24h列表
    #         config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_48H_H,  # 不区分地域相对48h列表
    #         config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_48H_H,  # 不区分地域相对48h列表2
    #         config_.RECALL_KEY_NAME_PREFIX_DUP_REGION_H,  # 大列表
    #     ]
    # else:
    #     key_prefix_list = [
    #         config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H,  # 地域分组小时级列表
    #         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}:{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}:{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"data_key = {data_key}, rule_key = {rule_key}, region = {region} "
              f"videos check end!")


def check_region_videos(rule_params):
    """检查限流视频分发数"""
    # 获取当前日期
    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()] + \
                       [code for city, code in config_.CITY_CODE.items()]

    for param in rule_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,
                         data_key, rule_key, region, stop_distribute_video_id_list, now_date, now_h)
            for region in region_code_list
        ]
        gevent.joinall(task_list)

    # 将已超分发视频 移除 天级更新30天列表
    log_.info("day_by_30day check start...")
    day30_key_prefix = config_.RECALL_KEY_NAME_PREFIX_30DAY
    for param in config_.RULE_PARAMS_30DAY_APP_TYPE.get('params_list'):
        data_key = param.get('data')
        rule_key = param.get('rule')
        log_.info(f"data_key = {data_key}, rule_key = {rule_key}")
        key_name = f"{day30_key_prefix}:{data_key}:{rule_key}:{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"{day30_key_prefix}:{data_key}:{rule_key}:{datetime.datetime.strftime(redis_date, '%Y%m%d')}"
        redis_helper.remove_value_from_zset(key_name=key_name, value=stop_distribute_video_id_list)
    log_.info("day_by_30day 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__':
    log_.info("start...")
    check_region_videos(rule_params=config_.RULE_PARAMS_REGION_APP_TYPE)
    # check_region_videos(rule_params=config_.RULE_PARAMS_REGION_APP_TYPE_48H)
    log_.info("end!")