rov-offline/rule_rank_h_by_24h.py

import time

import pandas as pd
import multiprocessing
import math
import traceback
from functools import reduce
from odps import ODPS
from threading import Timer
from datetime import datetime, timedelta
from get_data import get_data_from_odps
from db_helper import RedisHelper
from my_utils import filter_video_status, check_table_partition_exits, filter_video_status_app, \
    request_post, send_msg_to_feishu
from my_config import set_config
from log import Log

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

features = [
    'apptype',
    'videoid',
    'preview人数',  # 过去24h预曝光人数
    'view人数',  # 过去24h曝光人数
    'play人数',  # 过去24h播放人数
    'share人数',  # 过去24h分享人数
    '回流人数',  # 过去24h分享，过去24h回流人数
    'preview次数',  # 过去24h预曝光次数
    'view次数',  # 过去24h曝光次数
    'play次数',  # 过去24h播放次数
    'share次数',  # 过去24h分享次数
    'platform_return',
    'platform_preview',
    'platform_preview_total',
    'platform_show',
    'platform_show_total',
    'platform_view',
    'platform_view_total',
]


def get_rov_redis_key(now_date):
    # 获取rov模型结果存放key
    redis_helper = RedisHelper()
    now_dt = 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.strftime(now_date - timedelta(days=1), '%Y%m%d')
        key_name = f'{config_.RECALL_KEY_NAME_PREFIX}{pre_dt}'
    return key_name


def h_data_check(project, table, now_date, now_h):
    """检查数据是否准备好"""
    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:
        # 23点开始到8点之前（不含8点），全部用22点生成那个列表
        if now_h == 23:
            dt = datetime.strftime(now_date - timedelta(hours=1), '%Y%m%d%H')
        elif now_h < 8:
            dt = f"{datetime.strftime(now_date - timedelta(days=1), '%Y%m%d')}22"
        else:
            dt = 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_feature_data(now_date, now_h, project, table):
    """获取特征数据"""
    # 23点开始到8点之前（不含8点），全部用22点生成那个列表
    if now_h == 23:
        dt = datetime.strftime(now_date - timedelta(hours=1), '%Y%m%d%H')
    elif now_h < 8:
        dt = f"{datetime.strftime(now_date - timedelta(days=1), '%Y%m%d')}22"
    else:
        dt = datetime.strftime(now_date, '%Y%m%d%H')
    log_.info({'feature_dt': dt})
    # dt = '20220425'
    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_score1(df):
    # score1计算公式: score = 回流人数/(view人数+10000)
    df = df.fillna(0)
    df['score'] = df['回流人数'] / (df['view人数'] + 1000)
    df = df.sort_values(by=['score'], ascending=False)
    return df


def cal_score2(df, param):
    # score2计算公式: score = share次数/(view+1000)+0.01*return/(share次数+100)
    df = df.fillna(0)
    if param.get('view_type', None) == 'video-show':
        df['share_rate'] = df['share次数'] / (df['platform_show'] + 1000)
    elif param.get('view_type', None) == 'preview':
        df['share_rate'] = df['share次数'] / (df['preview人数'] + 1000)
    else:
        df['share_rate'] = df['share次数'] / (df['view人数'] + 1000)
    df['back_rate'] = df['回流人数'] / (df['share次数'] + 100)
    df['score'] = df['share_rate'] + 0.01 * df['back_rate']
    df['platform_return_rate'] = df['platform_return'] / df['回流人数']
    df = df.sort_values(by=['score'], ascending=False)
    return df


def cal_score(df, param):
    # score计算公式: score1 = share次数/(view+1000)+0.01*return/(share次数+100)
    # ctr = lastonehour_play/(lastonehour_preview+1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr
    # score = 0.3 * score1 + 0.7 * K2
    df = df.fillna(0)
    if param.get('view_type', None) == 'video-show':
        df['share_rate'] = df['share次数'] / (df['platform_show'] + 1000)
        df['ctr'] = df['play人数'] / (df['platform_show'] + 1000)
    elif param.get('view_type', None) == 'preview':
        df['share_rate'] = df['share次数'] / (df['preview人数'] + 1000)
        df['ctr'] = df['play人数'] / (df['preview人数'] + 1000)
    else:
        df['share_rate'] = df['share次数'] / (df['platform_show'] + 1000)
        df['ctr'] = df['play人数'] / (df['platform_show'] + 1000)

    df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x)
    df['back_rate'] = df['回流人数'] / (df['share次数'] + 100)
    df['platform_return_rate'] = df['platform_return'] / df['回流人数']

    df['score1'] = df['share_rate'] + 0.01 * df['back_rate']

    click_score_rate = param.get('click_score_rate', None)
    back_score_rate = param.get('click_score_rate', None)
    if click_score_rate is not None:
        df['score'] = (1 - click_score_rate) * df['score1'] + click_score_rate * df['K2']
    elif back_score_rate is not None:
        df['score'] = (1 - back_score_rate) * df['score1'] + back_score_rate * df['back_rate']
    else:
        df['score'] = df['score1']

    df = df.sort_values(by=['score'], ascending=False)
    return df


def video_rank_h(df, now_date, now_h, rule_key, param, data_key, notify_backend):
    """
    获取符合进入召回源条件的视频，与每日更新的rov模型结果视频列表进行合并
    :param df:
    :param now_date:
    :param now_h:
    :param rule_key: 天级规则数据进入条件
    :param param: 天级规则数据进入条件参数
    :param data_key: 使用数据标识
    :param notify_backend: 是否同步给后端标识
    :return:
    """
    redis_helper = RedisHelper()
    log_.info(f"videos_count = {len(df)}")

    # videoid重复时，保留分值高
    df = df.sort_values(by=['score'], ascending=False)
    df = df.drop_duplicates(subset=['videoid'], keep='first')
    df['videoid'] = df['videoid'].astype(int)

    # 获取符合进入召回源条件的视频
    return_count = param.get('return_count')
    if return_count:
        day_recall_df = df[df['回流人数'] > return_count]
    else:
        day_recall_df = df
    platform_return_rate = param.get('platform_return_rate', 0)
    day_recall_df = day_recall_df[day_recall_df['platform_return_rate'] > platform_return_rate]
    day_recall_videos = day_recall_df['videoid'].to_list()
    log_.info(f'h_by24h_recall videos count = {len(day_recall_videos)}')
    # 视频状态过滤
    if data_key in ['data7', ]:
        filtered_videos = filter_video_status_app(day_recall_videos)
    else:
        filtered_videos = filter_video_status(day_recall_videos)
    # log_.info('filtered_videos count = {}'.format(len(filtered_videos)))

    # 写入对应的redis
    now_dt = datetime.strftime(now_date, '%Y%m%d')
    day_video_ids = []
    day_recall_result = {}
    # json_data = []
    for video_id in filtered_videos:
        score = day_recall_df[day_recall_df['videoid'] == video_id]['score']
        day_recall_result[int(video_id)] = float(score)
        day_video_ids.append(int(video_id))
        # json_data.append({'videoId': video_id, 'rovScore': float(score)})

    h_24h_recall_key_name = \
        f"{config_.RECALL_KEY_NAME_PREFIX_BY_24H}{data_key}:{rule_key}:{now_dt}:{now_h}"
    # log_.info("h_24h_recall_key_name:redis:{}".format(h_24h_recall_key_name))
    if len(day_recall_result) > 0:
        log_.info(f"count = {len(day_recall_result)}, key = {h_24h_recall_key_name}")
        redis_helper.add_data_with_zset(key_name=h_24h_recall_key_name, data=day_recall_result, expire_time=2 * 3600)
        # 清空线上过滤应用列表
        # redis_helper.del_keys(key_name=f"{config_.H_VIDEO_FILER_24H}{app_type}.{data_key}.{rule_key}")

    # 去重筛选结果，保留剩余数据并写入Redis
    all_videos = df['videoid'].to_list()
    log_.info(f'h_by24h_recall all videos count = {len(all_videos)}')
    # 视频状态过滤
    st_time = time.time()
    if data_key in ['data7', ]:
        all_filtered_videos = filter_video_status_app(all_videos)
    else:
        all_filtered_videos = filter_video_status(all_videos)
    log_.info(f'all_filtered_videos count = {len(all_filtered_videos)}, param = {param}, execute_time = {int(time.time() - st_time)*1000}ms')
    # 与筛选结果去重
    other_videos = [video for video in all_filtered_videos if video not in day_video_ids]
    log_.info(f'other_videos count = {len(other_videos)}')
    # 写入对应的redis
    other_24h_recall_result = {}
    json_data = []
    for video_id in other_videos:
        score = df[df['videoid'] == video_id]['score']
        other_24h_recall_result[int(video_id)] = float(score)
        json_data.append({'videoId': video_id, 'rovScore': float(score)})
    # other_h_24h_recall_key_name = \
    #     f"{config_.RECALL_KEY_NAME_PREFIX_BY_24H_OTHER}{app_type}:{data_key}:{rule_key}:{now_dt}:{now_h}"
    other_h_24h_recall_key_name = \
        f"{config_.RECALL_KEY_NAME_PREFIX_BY_24H_OTHER}{data_key}:{rule_key}:{now_dt}:{now_h}"
    if len(other_24h_recall_result) > 0:
        log_.info(f"count = {len(other_24h_recall_result)}")
        redis_helper.add_data_with_zset(key_name=other_h_24h_recall_key_name, data=other_24h_recall_result,
                                        expire_time=2 * 3600)
    # 通知后端更新兜底视频数据
    if notify_backend is True:
        log_.info('json_data count = {}'.format(len(json_data[:5000])))
        # log_.info(f"json_data = {json_data}")
        result = request_post(request_url=config_.NOTIFY_BACKEND_updateFallBackVideoList_URL,
                              request_data={'videos': json_data[:5000]})
        if result is None:
            log_.error('notify backend updateFallBackVideoList fail!')
        elif result['code'] == 0:
            log_.info('notify backend updateFallBackVideoList success!')
        else:
            log_.error('notify backend updateFallBackVideoList fail!')

    # 去重更新rov模型结果，并另存为redis中
    # initial_data_dup = {}
    # for video_id, score in initial_data:
    #     if int(video_id) not in day_video_ids:
    #         initial_data_dup[int(video_id)] = score
    # log_.info(f"initial data dup count = {len(initial_data_dup)}")
    #
    # initial_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP_24H}{rule_key}.{now_dt}.{now_h}"
    # if len(initial_data_dup) > 0:
    #     redis_helper.add_data_with_zset(key_name=initial_key_name, data=initial_data_dup, expire_time=23 * 3600)


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


def merge_df_with_score(df_left, df_right):
    """
    df 按照videoid合并，平台回流人数、回流人数、分数 分别求和
    :param df_left:
    :param df_right:
    :return:
    """
    df_merged = pd.merge(df_left, df_right, on=['videoid'], how='outer', suffixes=['_x', '_y'])
    df_merged.fillna(0, inplace=True)
    feature_list = ['videoid', '回流人数', 'platform_return', 'score']
    for feature in feature_list[1:]:
        df_merged[feature] = df_merged[f'{feature}_x'] + df_merged[f'{feature}_y']
    return df_merged[feature_list]


def process_with_param(param, data_params_item, rule_params_item, feature_df, now_date, now_h):
    log_.info(f"param = {param} start...")
    score_df_list = []
    notify_backend = param.get('notify_backend', False)
    data_key = param.get('data')
    data_param = data_params_item.get(data_key)
    log_.info(f"data_key = {data_key}, data_param = {data_param}")
    rule_key = param.get('rule')
    rule_param = rule_params_item.get(rule_key)
    log_.info(f"rule_key = {rule_key}, rule_param = {rule_param}")
    # cal_score_func = rule_param.get('cal_score_func', 1)
    merge_func = rule_param.get('merge_func', 1)

    if merge_func == 2:
        for apptype, weight in data_param.items():
            df = feature_df[feature_df['apptype'] == apptype]
            # 计算score
            score_df = cal_score(df=df, param=rule_param)
            score_df['score'] = score_df['score'] * weight
            score_df_list.append(score_df)
        # 分数合并
        df_merged = reduce(merge_df_with_score, score_df_list)
        # 更新平台回流比
        df_merged['platform_return_rate'] = df_merged['platform_return'] / df_merged['回流人数']
        video_rank_h(df=df_merged, now_date=now_date, now_h=now_h,
                     rule_key=rule_key, param=rule_param, data_key=data_key,
                     notify_backend=notify_backend)
    else:
        df_list = [feature_df[feature_df['apptype'] == apptype] for apptype, _ in data_param.items()]
        df_merged = reduce(merge_df, df_list)
        score_df = cal_score(df=df_merged, param=rule_param)
        video_rank_h(df=score_df, now_date=now_date, now_h=now_h,
                     rule_key=rule_key, param=rule_param, data_key=data_key,
                     notify_backend=notify_backend)
    log_.info(f"param = {param} end!")


def rank_by_h(now_date, now_h, rule_params, project, table):
    # 获取特征数据
    feature_df = get_feature_data(now_date=now_date, now_h=now_h, project=project, table=table)
    feature_df['apptype'] = feature_df['apptype'].astype(int)
    # rank
    data_params_item = rule_params.get('data_params')
    rule_params_item = rule_params.get('rule_params')
    """
    for param in rule_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}")
        # 计算score
        cal_score_func = rule_param.get('cal_score_func', 1)
        if cal_score_func == 2:
            score_df = cal_score2(df=df_merged, param=rule_param)
        else:
            score_df = cal_score1(df=df_merged)
        video_rank_h(df=score_df, now_date=now_date, now_h=now_h,
                     rule_key=rule_key, param=rule_param, data_key=data_key)
    """
    params_list = rule_params.get('params_list')
    pool = multiprocessing.Pool(processes=len(params_list))
    for param in params_list:
        pool.apply_async(
            func=process_with_param,
            args=(param, data_params_item, rule_params_item, feature_df, now_date, now_h)
        )
    pool.close()
    pool.join()

    # for param in rule_params.get('params_list'):
    #     score_df_list = []
    #     notify_backend = param.get('notify_backend', False)
    #     data_key = param.get('data')
    #     data_param = data_params_item.get(data_key)
    #     log_.info(f"data_key = {data_key}, data_param = {data_param}")
    #     rule_key = param.get('rule')
    #     rule_param = rule_params_item.get(rule_key)
    #     log_.info(f"rule_key = {rule_key}, rule_param = {rule_param}")
    #     # cal_score_func = rule_param.get('cal_score_func', 1)
    #     merge_func = rule_param.get('merge_func', 1)
    #
    #     if merge_func == 2:
    #         for apptype, weight in data_param.items():
    #             df = feature_df[feature_df['apptype'] == apptype]
    #             # 计算score
    #             score_df = cal_score(df=df, param=rule_param)
    #             score_df['score'] = score_df['score'] * weight
    #             score_df_list.append(score_df)
    #         # 分数合并
    #         df_merged = reduce(merge_df_with_score, score_df_list)
    #         # 更新平台回流比
    #         df_merged['platform_return_rate'] = df_merged['platform_return'] / df_merged['回流人数']
    #         video_rank_h(df=df_merged, now_date=now_date, now_h=now_h,
    #                      rule_key=rule_key, param=rule_param, data_key=data_key,
    #                      notify_backend=notify_backend)
    #     else:
    #         df_list = [feature_df[feature_df['apptype'] == apptype] for apptype, _ in data_param.items()]
    #         df_merged = reduce(merge_df, df_list)
    #         score_df = cal_score(df=df_merged, param=rule_param)
    #         video_rank_h(df=score_df, now_date=now_date, now_h=now_h,
    #                      rule_key=rule_key, param=rule_param, data_key=data_key,
    #                      notify_backend=notify_backend)

    #     # to-csv
    #     score_filename = f"score_by24h_{key}_{datetime.strftime(now_date, '%Y%m%d%H')}.csv"
    #     score_df.to_csv(f'./data/{score_filename}')
    #     # to-logs
    #     log_.info({"date": datetime.strftime(now_date, '%Y%m%d%H'),
    #                "redis_key_prefix": config_.RECALL_KEY_NAME_PREFIX_BY_24H,
    #                "rule_key": key,
    #                # "score_df": score_df[['videoid', 'score']]
    #                })


def h_rank_bottom(now_date, now_h, rule_params):
    """未按时更新数据，用模型召回数据作为当前的数据"""
    redis_helper = RedisHelper()
    if now_h == 0:
        redis_dt = datetime.strftime(now_date - timedelta(days=1), '%Y%m%d')
        redis_h = 23
    else:
        redis_dt = datetime.strftime(now_date, '%Y%m%d')
        redis_h = now_h - 1
    key_prefix_list = [config_.RECALL_KEY_NAME_PREFIX_BY_24H, config_.RECALL_KEY_NAME_PREFIX_BY_24H_OTHER]

    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}")
        for key_prefix in key_prefix_list:
            key_name = f"{key_prefix}{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()
            for video_id, score in initial_data:
                final_data[video_id] = score
            # 存入对应的redis
            final_key_name = \
                f"{key_prefix}{data_key}:{rule_key}:{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=2 * 3600)

    """
    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}")
            for key_prefix in key_prefix_list:
                key_name = f"{key_prefix}{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()
                for video_id, score in initial_data:
                    final_data[video_id] = score
                # 存入对应的redis
                final_key_name = \
                    f"{key_prefix}{app_type}:{data_key}:{rule_key}:{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=2 * 3600)
                # 清空线上过滤应用列表
                # redis_helper.del_keys(key_name=f"{config_.H_VIDEO_FILER_24H}{app_type}.{data_key}.{rule_key}")
    """


def h_timer_check():
    try:
        project = config_.PROJECT_24H_APP_TYPE
        table = config_.TABLE_24H_APP_TYPE
        rule_params = config_.RULE_PARAMS_24H_APP_TYPE
        now_date = datetime.today()
        log_.info(f"now_date: {datetime.strftime(now_date, '%Y%m%d%H')}")
        now_min = datetime.now().minute
        now_h = datetime.now().hour
        redis_helper = RedisHelper()
        # 查看当前天级更新的数据是否已准备好
        h_data_count = h_data_check(project=project, table=table, now_date=now_date, now_h=now_h)
        if now_h == 23 or now_h < 8:
            log_.info(f'now_h = {now_h} use bottom data!')
            h_rank_bottom(now_date=now_date, now_h=now_h, rule_params=rule_params)
            log_.info(f"24h_data end!")
            redis_helper.set_data_to_redis(
                key_name=f"{config_.RULE_24H_DATA_STATUS}:{datetime.strftime(now_date, '%Y%m%d%H')}", value='1', expire_time=2 * 3600
            )
            log_.info(f"rule_24h_data status update to '1' finished!")
        elif h_data_count > 0:
            log_.info(f'h_by24h_data_count = {h_data_count}')
            # 数据准备好，进行更新
            rank_by_h(now_date=now_date, now_h=now_h, rule_params=rule_params, project=project, table=table)
            log_.info(f"24h_data end!")
            redis_helper.set_data_to_redis(
                key_name=f"{config_.RULE_24H_DATA_STATUS}:{datetime.strftime(now_date, '%Y%m%d%H')}", value='1', expire_time=2 * 3600
            )
            log_.info(f"rule_24h_data status update to '1' finished!")
        elif now_min > 40:
            log_.info('h_by24h_recall data is None, use bottom data!')
            h_rank_bottom(now_date=now_date, now_h=now_h, rule_params=rule_params)
            log_.info(f"24h_data end!")
            redis_helper.set_data_to_redis(
                key_name=f"{config_.RULE_24H_DATA_STATUS}:{datetime.strftime(now_date, '%Y%m%d%H')}", value='1', expire_time=2 * 3600
            )
            log_.info(f"rule_24h_data status update to '1' finished!")
        else:
            # 数据没准备好，1分钟后重新检查
            Timer(60, h_timer_check).start()

    except Exception as e:
        log_.error(f"不区分地域24h数据更新失败, exception: {e}, traceback: {traceback.format_exc()}")
        send_msg_to_feishu(
            webhook=config_.FEISHU_ROBOT['server_robot'].get('webhook'),
            key_word=config_.FEISHU_ROBOT['server_robot'].get('key_word'),
            msg_text=f"rov-offline{config_.ENV_TEXT} - 不区分地域24h数据更新失败\n"
                     f"exception: {e}\n"
                     f"traceback: {traceback.format_exc()}"
        )


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