rov-offline/region_rule_rank_h.py

# -*- coding: utf-8 -*-
# @ModuleName: region_rule_rank_h
# @Author: Liqian
# @Time: 2022/5/5 15:54
# @Software: PyCharm

import gevent
import datetime
import pandas as pd
import math
from functools import reduce
from odps import ODPS
from threading import Timer
from utils import MysqlHelper, RedisHelper, get_data_from_odps, filter_video_status, filter_shield_video, check_table_partition_exits
from config import set_config
from log import Log
from check_video_limit_distribute import update_limit_video_score

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

region_code = config_.REGION_CODE

features = [
    'apptype',
    'code',
    'videoid',
    'lastonehour_preview',  # 过去1小时预曝光人数
    'lastonehour_view',  # 过去1小时曝光人数
    'lastonehour_play',  # 过去1小时播放人数
    'lastonehour_share',  # 过去1小时分享人数
    'lastonehour_return',  # 过去1小时分享，过去1小时回流人数
    'lastonehour_preview_total',  # 过去1小时预曝光次数
    'lastonehour_view_total',  # 过去1小时曝光次数
    'lastonehour_play_total',  # 过去1小时播放次数
    'lastonehour_share_total',  # 过去1小时分享次数
    'platform_return',
    'lastonehour_show',  # 不区分地域
    'lastonehour_show_region',  # 地域分组
]


def get_region_code(region):
    """获取省份对应的code"""
    mysql_helper = MysqlHelper(mysql_info=config_.MYSQL_INFO)
    sql = f"SELECT ad_code FROM region_adcode WHERE parent_id = 0 AND region LIKE '{region}%';"
    ad_code = mysql_helper.get_data(sql=sql)
    return ad_code[0][0]


def h_data_check(project, table, now_date):
    """检查数据是否准备好"""
    odps = ODPS(
        access_id=config_.ODPS_CONFIG['ACCESSID'],
        secret_access_key=config_.ODPS_CONFIG['ACCESSKEY'],
        project=project,
        endpoint=config_.ODPS_CONFIG['ENDPOINT'],
        connect_timeout=3000,
        read_timeout=500000,
        pool_maxsize=1000,
        pool_connections=1000
    )

    try:
        dt = datetime.datetime.strftime(now_date, '%Y%m%d%H')
        check_res = check_table_partition_exits(date=dt, project=project, table=table)
        if check_res:
            sql = f'select * from {project}.{table} where dt = {dt}'
            with odps.execute_sql(sql=sql).open_reader() as reader:
                data_count = reader.count
        else:
            data_count = 0
    except Exception as e:
        data_count = 0
    return data_count


def get_rov_redis_key(now_date):
    """获取rov模型结果存放key"""
    redis_helper = RedisHelper()
    now_dt = datetime.datetime.strftime(now_date, '%Y%m%d')
    key_name = f'{config_.RECALL_KEY_NAME_PREFIX}{now_dt}'
    if not redis_helper.key_exists(key_name=key_name):
        pre_dt = datetime.datetime.strftime(now_date - datetime.timedelta(days=1), '%Y%m%d')
        key_name = f'{config_.RECALL_KEY_NAME_PREFIX}{pre_dt}'
    return key_name


def get_feature_data(project, table, now_date):
    """获取特征数据"""
    dt = datetime.datetime.strftime(now_date, '%Y%m%d%H')
    # dt = '2022041310'
    records = get_data_from_odps(date=dt, project=project, table=table)
    feature_data = []
    for record in records:
        item = {}
        for feature_name in features:
            item[feature_name] = record[feature_name]
        feature_data.append(item)
    feature_df = pd.DataFrame(feature_data)
    return feature_df


def cal_score(df, param):
    """
    计算score
    :param df: 特征数据
    :param param: 规则参数
    :return:
    """
    # score计算公式: sharerate*backrate*logback*ctr
    # sharerate = lastonehour_share/(lastonehour_play+1000)
    # backrate = lastonehour_return/(lastonehour_share+10)
    # ctr = lastonehour_play/(lastonehour_preview+1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr
    # score = sharerate * backrate * LOG(lastonehour_return+1) * K2

    df = df.fillna(0)
    df['share_rate'] = df['lastonehour_share'] / (df['lastonehour_play'] + 1000)
    df['back_rate'] = df['lastonehour_return'] / (df['lastonehour_share'] + 10)
    df['log_back'] = (df['lastonehour_return'] + 1).apply(math.log)
    if param.get('view_type', None) == 'video-show':
        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show'] + 1000)
    elif param.get('view_type', None) == 'video-show-region':
        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show_region'] + 1000)
    else:
        df['ctr'] = df['lastonehour_play'] / (df['lastonehour_preview'] + 1000)
    df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x)
    df['score'] = df['share_rate'] * df['back_rate'] * df['log_back'] * df['K2']
    df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return']
    df = df.sort_values(by=['score'], ascending=False)
    return df


def video_rank(df, now_date, now_h, rule_key, param, region, app_type, data_key):
    """
    获取符合进入召回源条件的视频，与每日更新的rov模型结果视频列表进行合并
    :param df:
    :param now_date:
    :param now_h:
    :param rule_key: 小时级数据进入条件
    :param param: 小时级数据进入条件参数
    :param region: 所属地域
    :return:
    """
    redis_helper = RedisHelper()

    # 获取符合进入召回源条件的视频，进入条件：小时级回流>=20 && score>=0.005
    return_count = param.get('return_count', 1)
    score_value = param.get('score_rule', 0)
    platform_return_rate = param.get('platform_return_rate', 0)
    h_recall_df = df[(df['lastonehour_return'] >= return_count) & (df['score'] >= score_value)
                     & (df['platform_return_rate'] >= platform_return_rate)]
    # videoid重复时，保留分值高
    h_recall_df = h_recall_df.sort_values(by=['score'], ascending=False)
    h_recall_df = h_recall_df.drop_duplicates(subset=['videoid'], keep='first')
    h_recall_df['videoid'] = h_recall_df['videoid'].astype(int)
    h_recall_videos = h_recall_df['videoid'].to_list()
    log_.info(f'h_recall videos count = {len(h_recall_videos)}')

    # 视频状态过滤
    filtered_videos = filter_video_status(h_recall_videos)
    log_.info('filtered_videos count = {}'.format(len(filtered_videos)))

    # 屏蔽视频过滤
    shield_key_name_list = config_.SHIELD_CONFIG.get(region, None)
    if shield_key_name_list is not None:
        filtered_videos = filter_shield_video(video_ids=filtered_videos, shield_key_name_list=shield_key_name_list)
        log_.info(f"shield filtered_videos count = {len(filtered_videos)}")

    # 写入对应的redis
    h_video_ids = []
    h_recall_result = {}
    for video_id in filtered_videos:
        score = h_recall_df[h_recall_df['videoid'] == video_id]['score']
        # print(score)
        h_recall_result[int(video_id)] = float(score)
        h_video_ids.append(int(video_id))
    h_recall_key_name = \
        f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H}{region}.{app_type}.{data_key}.{rule_key}." \
        f"{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
    if len(h_recall_result) > 0:
        redis_helper.add_data_with_zset(key_name=h_recall_key_name, data=h_recall_result, expire_time=23 * 3600)
        # 限流视频score调整
        update_limit_video_score(initial_videos=h_recall_result, key_name=h_recall_key_name)
        # 清空线上过滤应用列表
        redis_helper.del_keys(key_name=f"{config_.REGION_H_VIDEO_FILER}{region}.{app_type}.{data_key}.{rule_key}")

    region_24h_rule_key = param.get('region_24h_rule_key', 'rule1')
    by_24h_rule_key = param.get('24h_rule_key', None)
    # 与其他召回视频池去重，存入对应的redis
    dup_to_redis(h_video_ids=h_video_ids, now_date=now_date, now_h=now_h, rule_key=rule_key,
                 region_24h_rule_key=region_24h_rule_key, by_24h_rule_key=by_24h_rule_key,
                 region=region, app_type=app_type, data_key=data_key)


def dup_to_redis(h_video_ids, now_date, now_h, rule_key, region_24h_rule_key, by_24h_rule_key, region, app_type, data_key):
    """将地域分组小时级数据与其他召回视频池去重，存入对应的redis"""
    redis_helper = RedisHelper()
    # # ##### 去重更新地域分组天级列表，并另存为redis中
    # region_day_key_name = \
    #     f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_DAY}{region}.rule1." \
    #     f"{datetime.datetime.strftime(now_date, '%Y%m%d')}"
    # if redis_helper.key_exists(key_name=region_day_key_name):
    #     region_day_data = redis_helper.get_data_zset_with_index(
    #         key_name=region_day_key_name, start=0, end=-1, with_scores=True)
    #     log_.info(f'region day data count = {len(region_day_data)}')
    #     region_day_dup = {}
    #     for video_id, score in region_day_data:
    #         if int(video_id) not in h_video_ids:
    #             region_day_dup[int(video_id)] = score
    #             h_video_ids.append(int(video_id))
    #     log_.info(f"region day data dup count = {len(region_day_dup)}")
    #     region_day_dup_key_name = \
    #         f"{config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_DAY_H}{region}.{rule_key}." \
    #         f"{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
    #     if len(region_day_dup) > 0:
    #         redis_helper.add_data_with_zset(key_name=region_day_dup_key_name, data=region_day_dup, expire_time=23 * 3600)

    # ##### 去重更新地域分组小时级24h列表，并另存为redis中
    region_24h_key_name = \
        f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_24H}{region}.{app_type}.{data_key}.{region_24h_rule_key}." \
        f"{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
    if redis_helper.key_exists(key_name=region_24h_key_name):
        region_24h_data = redis_helper.get_all_data_from_zset(key_name=region_24h_key_name, with_scores=True)
        log_.info(f'region 24h data count = {len(region_24h_data)}')

        # 屏蔽视频过滤
        region_24h_video_ids = [int(video_id) for video_id, _ in region_24h_data]
        shield_key_name_list = config_.SHIELD_CONFIG.get(region, None)
        if shield_key_name_list is not None:
            region_24h_video_ids = filter_shield_video(video_ids=region_24h_video_ids, shield_key_name_list=shield_key_name_list)
            log_.info(f"shield filtered_videos count = {len(region_24h_video_ids)}")

        region_24h_dup = {}
        for video_id, score in region_24h_data:
            if int(video_id) not in h_video_ids and int(video_id) in region_24h_video_ids:
                region_24h_dup[int(video_id)] = score
                h_video_ids.append(int(video_id))
        log_.info(f"region 24h data dup count = {len(region_24h_dup)}")
        region_24h_dup_key_name = \
            f"{config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H}{region}.{app_type}.{data_key}.{rule_key}." \
            f"{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
        if len(region_24h_dup) > 0:
            redis_helper.add_data_with_zset(key_name=region_24h_dup_key_name, data=region_24h_dup, expire_time=23 * 3600)
            # 限流视频score调整
            update_limit_video_score(initial_videos=region_24h_dup, key_name=region_24h_dup_key_name)
            # 清空线上过滤应用列表
            # redis_helper.del_keys(key_name=f"{config_.REGION_H_VIDEO_FILER_24H}{app_type}.{data_key}.{region}.{rule_key}")

    # ##### 去重小程序天级更新结果，并另存为redis中
    # day_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_DAY}rule2.{datetime.datetime.strftime(now_date, '%Y%m%d')}"
    # if redis_helper.key_exists(key_name=day_key_name):
    #     day_data = redis_helper.get_data_zset_with_index(
    #         key_name=day_key_name, start=0, end=-1, with_scores=True)
    #     log_.info(f'day data count = {len(day_data)}')
    #     day_dup = {}
    #     for video_id, score in day_data:
    #         if int(video_id) not in h_video_ids:
    #             day_dup[int(video_id)] = score
    #             h_video_ids.append(int(video_id))
    #     log_.info(f"day data dup count = {len(day_dup)}")
    #     day_dup_key_name = \
    #         f"{config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_DAY_H}{region}.{rule_key}." \
    #         f"{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
    #     if len(day_dup) > 0:
    #         redis_helper.add_data_with_zset(key_name=day_dup_key_name, data=day_dup, expire_time=23 * 3600)

    # ##### 去重小程序相对24h更新结果，并另存为redis中
    day_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_24H}{app_type}.{data_key}.{by_24h_rule_key}." \
                   f"{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
    if redis_helper.key_exists(key_name=day_key_name):
        day_data = redis_helper.get_all_data_from_zset(key_name=day_key_name, with_scores=True)
        log_.info(f'24h data count = {len(day_data)}')

        # 屏蔽视频过滤
        day_video_ids = [int(video_id) for video_id, _ in day_data]
        shield_key_name_list = config_.SHIELD_CONFIG.get(region, None)
        if shield_key_name_list is not None:
            day_video_ids = filter_shield_video(video_ids=day_video_ids, shield_key_name_list=shield_key_name_list)
            log_.info(f"shield filtered_videos count = {len(day_video_ids)}")

        day_dup = {}
        for video_id, score in day_data:
            if int(video_id) not in h_video_ids and int(video_id) in day_video_ids:
                day_dup[int(video_id)] = score
                h_video_ids.append(int(video_id))
        log_.info(f"24h data dup count = {len(day_dup)}")
        day_dup_key_name = \
            f"{config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_24H_H}{region}.{app_type}.{data_key}.{rule_key}." \
            f"{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
        if len(day_dup) > 0:
            redis_helper.add_data_with_zset(key_name=day_dup_key_name, data=day_dup, expire_time=23 * 3600)
            # 限流视频score调整
            update_limit_video_score(initial_videos=day_dup, key_name=day_dup_key_name)
            # 清空线上过滤应用列表
            redis_helper.del_keys(key_name=f"{config_.H_VIDEO_FILER_24H}{region}.{app_type}.{data_key}.{rule_key}")

    # ##### 去重小程序相对24h 筛选后剩余数据 更新结果，并另存为redis中
    if by_24h_rule_key == 'rule3':
        other_h_24h_recall_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_24H_OTHER}{app_type}.{data_key}." \
                                      f"{by_24h_rule_key}.{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
        if redis_helper.key_exists(key_name=other_h_24h_recall_key_name):
            other_24h_data = redis_helper.get_all_data_from_zset(key_name=other_h_24h_recall_key_name, with_scores=True)
            log_.info(f'24h other data count = {len(other_24h_data)}')

            # 屏蔽视频过滤
            other_24h_video_ids = [int(video_id) for video_id, _ in other_24h_data]
            shield_key_name_list = config_.SHIELD_CONFIG.get(region, None)
            if shield_key_name_list is not None:
                other_24h_video_ids = filter_shield_video(video_ids=other_24h_video_ids, shield_key_name_list=shield_key_name_list)
                log_.info(f"shield filtered_videos count = {len(other_24h_video_ids)}")

            other_24h_dup = {}
            for video_id, score in other_24h_data:
                if int(video_id) not in h_video_ids and int(video_id) in other_24h_video_ids:
                    other_24h_dup[int(video_id)] = score
                    h_video_ids.append(int(video_id))
            log_.info(f"other 24h data dup count = {len(other_24h_dup)}")
            other_24h_dup_key_name = \
                f"{config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H}{region}.{app_type}.{data_key}.{rule_key}." \
                f"{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
            if len(other_24h_dup) > 0:
                redis_helper.add_data_with_zset(key_name=other_24h_dup_key_name, data=other_24h_dup, expire_time=23 * 3600)
                # 限流视频score调整
                update_limit_video_score(initial_videos=other_24h_dup, key_name=other_24h_dup_key_name)

    # ##### 去重小程序模型更新结果，并另存为redis中
    model_key_name = get_rov_redis_key(now_date=now_date)
    model_data = redis_helper.get_all_data_from_zset(key_name=model_key_name, with_scores=True)
    log_.info(f'model data count = {len(model_data)}')

    # 屏蔽视频过滤
    model_video_ids = [int(video_id) for video_id, _ in model_data]
    shield_key_name_list = config_.SHIELD_CONFIG.get(region, None)
    if shield_key_name_list is not None:
        model_video_ids = filter_shield_video(video_ids=model_video_ids, shield_key_name_list=shield_key_name_list)
        log_.info(f"shield filtered_videos count = {len(model_video_ids)}")

    model_data_dup = {}
    for video_id, score in model_data:
        if int(video_id) not in h_video_ids and int(video_id) in model_video_ids:
            model_data_dup[int(video_id)] = score
            h_video_ids.append(int(video_id))
    log_.info(f"model data dup count = {len(model_data_dup)}")
    model_data_dup_key_name = \
        f"{config_.RECALL_KEY_NAME_PREFIX_DUP_REGION_H}{region}.{app_type}.{data_key}.{rule_key}." \
        f"{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
    if len(model_data_dup) > 0:
        redis_helper.add_data_with_zset(key_name=model_data_dup_key_name, data=model_data_dup, expire_time=23 * 3600)
        # 限流视频score调整
        update_limit_video_score(initial_videos=model_data_dup, key_name=model_data_dup_key_name)


def merge_df(df_left, df_right):
    """
    df按照videoid, code 合并，对应特征求和
    :param df_left:
    :param df_right:
    :return:
    """
    df_merged = pd.merge(df_left, df_right, on=['videoid', 'code'], how='outer', suffixes=['_x', '_y'])
    df_merged.fillna(0, inplace=True)
    feature_list = ['videoid', 'code']
    for feature in features:
        if feature in ['apptype', 'videoid', 'code']:
            continue
        df_merged[feature] = df_merged[f'{feature}_x'] + df_merged[f'{feature}_y']
        feature_list.append(feature)
    return df_merged[feature_list]


def process_with_region(region, df_merged, app_type, data_key, rule_key, rule_param, now_date, now_h):
    log_.info(f"region = {region}")
    # 计算score
    region_df = df_merged[df_merged['code'] == region]
    log_.info(f'region_df count = {len(region_df)}')
    score_df = cal_score(df=region_df, param=rule_param)
    video_rank(df=score_df, now_date=now_date, now_h=now_h, rule_key=rule_key, param=rule_param,
               region=region, app_type=app_type, data_key=data_key)


def process_with_app_type(app_type, params, region_code_list, feature_df, now_date, now_h):
    log_.info(f"app_type = {app_type}")
    data_params_item = params.get('data_params')
    rule_params_item = params.get('rule_params')
    task_list = []
    for param in params.get('params_list'):
        data_key = param.get('data')
        data_param = data_params_item.get(data_key)
        log_.info(f"data_key = {data_key}, data_param = {data_param}")
        df_list = [feature_df[feature_df['apptype'] == apptype] for apptype in data_param]
        df_merged = reduce(merge_df, df_list)

        rule_key = param.get('rule')
        rule_param = rule_params_item.get(rule_key)
        log_.info(f"rule_key = {rule_key}, rule_param = {rule_param}")
        task_list.extend(
            [
                gevent.spawn(process_with_region, region, df_merged, app_type, data_key, rule_key, rule_param,
                             now_date, now_h)
                for region in region_code_list
            ]
        )
    gevent.joinall(task_list)



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


def rank_by_h(project, table, now_date, now_h, rule_params, region_code_list):
    # 获取特征数据
    feature_df = get_feature_data(project=project, table=table, now_date=now_date)
    feature_df['apptype'] = feature_df['apptype'].astype(int)
    t = [
        gevent.spawn(process_with_app_type, app_type, params, region_code_list, feature_df, now_date, now_h)
        for app_type, params in rule_params.items()
    ]
    gevent.joinall(t)


    # for app_type, params in rule_params.items():
    #     log_.info(f"app_type = {app_type}")
    #     for data_key, data_param in params['data_params'].items():
    #         log_.info(f"data_key = {data_key}, data_param = {data_param}")
    #         df_list = [feature_df[feature_df['apptype'] == apptype] for apptype in data_param]
    #         df_merged = reduce(merge_df, df_list)
    #         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, region, df_merged, app_type, data_key, rule_key, rule_param, now_date, now_h)
    #                 for region in region_code_list
    #             ]
    #             gevent.joinall(task_list)

    # rank
    # for key, value in rule_params.items():
    #     log_.info(f"rule = {key}, param = {value}")
    #     for region in region_code_list:
    #         log_.info(f"region = {region}")
    #         # 计算score
    #         region_df = feature_df[feature_df['code'] == region]
    #         log_.info(f'region_df count = {len(region_df)}')
    #         score_df = cal_score(df=region_df, param=value)
    #         video_rank(df=score_df, now_date=now_date, now_h=now_h, rule_key=key, param=value, region=region)
    #         # to-csv
    #         score_filename = f"score_{region}_{key}_{datetime.datetime.strftime(now_date, '%Y%m%d%H')}.csv"
    #         score_df.to_csv(f'./data/{score_filename}')
    #         # to-logs
    #         log_.info({"date": datetime.datetime.strftime(now_date, '%Y%m%d%H'),
    #                    "region_code": region,
    #                    "redis_key_prefix": config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H,
    #                    "rule_key": key,
    #                    # "score_df": score_df[['videoid', 'score']]
    #                    }
    #                   )


def h_rank_bottom(now_date, now_h, rule_params, region_code_list):
    """未按时更新数据，用上一小时结果作为当前小时的数据"""
    # 获取rov模型结果
    redis_helper = RedisHelper()
    if now_h == 0:
        redis_dt = datetime.datetime.strftime(now_date - datetime.timedelta(days=1), '%Y%m%d')
        redis_h = 23
    else:
        redis_dt = datetime.datetime.strftime(now_date, '%Y%m%d')
        redis_h = now_h - 1

    # 以上一小时的地域分组数据作为当前小时的数据
    key_prefix = config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H
    for app_type, params in rule_params.items():
        log_.info(f"app_type = {app_type}")
        rule_params_item = params.get('rule_params')
        for param in params.get('params_list'):
            data_key = param.get('data')
            rule_key = param.get('rule')
            rule_param = rule_params_item.get(rule_key)
            log_.info(f"data_key = {data_key}, rule_key = {rule_key}, rule_param = {rule_param}")
            region_24h_rule_key = rule_param.get('region_24h_rule_key', 'rule1')
            for region in region_code_list:
                log_.info(f"region = {region}")
                key_name = f"{key_prefix}{region}.{app_type}.{data_key}.{rule_key}.{redis_dt}.{redis_h}"
                initial_data = redis_helper.get_all_data_from_zset(key_name=key_name, with_scores=True)
                if initial_data is None:
                    initial_data = []
                final_data = dict()
                h_video_ids = []
                for video_id, score in initial_data:
                    final_data[video_id] = score
                    h_video_ids.append(int(video_id))
                # 存入对应的redis
                final_key_name = \
                    f"{key_prefix}{region}.{app_type}.{data_key}.{rule_key}.{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
                if len(final_data) > 0:
                    redis_helper.add_data_with_zset(key_name=final_key_name, data=final_data, expire_time=23 * 3600)
                # 清空线上过滤应用列表
                redis_helper.del_keys(
                    key_name=f"{config_.REGION_H_VIDEO_FILER}{region}.{app_type}.{data_key}.{rule_key}")
                # 与其他召回视频池去重，存入对应的redis
                dup_to_redis(h_video_ids=h_video_ids, now_date=now_date, now_h=now_h, rule_key=rule_key,
                             region_24h_rule_key=region_24h_rule_key, region=region,
                             app_type=app_type, data_key=data_key)


        # 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}")
        #         region_24h_rule_key = rule_param.get('region_24h_rule_key', 'rule1')
        #         for region in region_code_list:
        #             log_.info(f"region = {region}")
        #             key_name = f"{key_prefix}{region}.{app_type}.{data_key}.{rule_key}.{redis_dt}.{redis_h}"
        #             initial_data = redis_helper.get_all_data_from_zset(key_name=key_name, with_scores=True)
        #             if initial_data is None:
        #                 initial_data = []
        #             final_data = dict()
        #             h_video_ids = []
        #             for video_id, score in initial_data:
        #                 final_data[video_id] = score
        #                 h_video_ids.append(int(video_id))
        #             # 存入对应的redis
        #             final_key_name = \
        #                 f"{key_prefix}{region}.{app_type}.{data_key}.{rule_key}.{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}"
        #             if len(final_data) > 0:
        #                 redis_helper.add_data_with_zset(key_name=final_key_name, data=final_data, expire_time=23 * 3600)
        #             # 清空线上过滤应用列表
        #             redis_helper.del_keys(key_name=f"{config_.REGION_H_VIDEO_FILER}{region}.{app_type}.{data_key}.{rule_key}")
        #             # 与其他召回视频池去重，存入对应的redis
        #             dup_to_redis(h_video_ids=h_video_ids, now_date=now_date, now_h=now_h, rule_key=rule_key,
        #                          region_24h_rule_key=region_24h_rule_key, region=region,
        #                          app_type=app_type, data_key=data_key)


def h_timer_check():
    rule_params = config_.RULE_PARAMS_REGION_APP_TYPE
    project = config_.PROJECT_REGION_APP_TYPE
    table = config_.TABLE_REGION_APP_TYPE
    region_code_list = [code for region, code in region_code.items()]
    now_date = datetime.datetime.today()
    log_.info(f"now_date: {datetime.datetime.strftime(now_date, '%Y%m%d%H')}")
    now_h = datetime.datetime.now().hour
    now_min = datetime.datetime.now().minute
    if now_h == 0:
        h_rank_bottom(now_date=now_date, now_h=now_h, rule_params=rule_params, region_code_list=region_code_list)
        return
    # 查看当前小时更新的数据是否已准备好
    h_data_count = h_data_check(project=project, table=table, now_date=now_date)
    if h_data_count > 0:
        log_.info(f'region_h_data_count = {h_data_count}')
        # 数据准备好，进行更新
        rank_by_h(now_date=now_date, now_h=now_h, rule_params=rule_params,
                  project=project, table=table, region_code_list=region_code_list)
    elif now_min > 50:
        log_.info('h_recall data is None, use bottom data!')
        h_rank_bottom(now_date=now_date, now_h=now_h, rule_params=rule_params, region_code_list=region_code_list)
    else:
        # 数据没准备好，1分钟后重新检查
        Timer(60, h_timer_check).start()


if __name__ == '__main__':
    h_timer_check()