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 multiprocessing
import os
import gevent
import datetime
import pandas as pd
import math
from functools import reduce
from odps import ODPS
from threading import Timer, Thread
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

# os.environ['NUMEXPR_MAX_THREADS'] = '16'

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} start...")
    # 计算score
    region_df = df_merged[df_merged['code'] == region]
    log_.info(f'region = {region}, 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)
    log_.info(f"region = {region} end!")



def process_with_app_type(app_type, params, region_code_list, feature_df, now_date, now_h):
    log_.info(f"app_type = {app_type} start...")
    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} end!")


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

    pool = multiprocessing.Pool(processes=len(config_.APP_TYPE))
    for app_type, params in rule_params.items():
        pool.apply_async(func=process_with_app_type,
                         args=(app_type, params, region_code_list, feature_df, now_date, now_h))
    pool.close()
    pool.join()

    """
    for app_type, params in rule_params.items():
        log_.info(f"app_type = {app_type} start...")
        data_params_item = params.get('data_params')
        rule_params_item = params.get('rule_params')
        for param in params.get('params_list'):
            log_.info(f"param = {param} start...")
            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 = []
            for region in region_code_list:
                t = Thread(target=process_with_region,
                           args=(region, df_merged, app_type, data_key, rule_key, rule_param, now_date, now_h)
                           )
                t.start()
                task_list.append(t)
            for t in task_list:
                t.join()
            log_.info(f"param = {param} end!")
        log_.info(f"app_type = {app_type} end!")
    """

    # 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}")

                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, region=region,
                             app_type=app_type, data_key=data_key, by_24h_rule_key=by_24h_rule_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)
        log_.info(f"region_h_data end!")
    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)
        log_.info(f"region_h_data end!")
    else:
        # 数据没准备好，1分钟后重新检查
        Timer(60, h_timer_check).start()


if __name__ == '__main__':
    log_.info(f"region_h_data start...")
    h_timer_check()