rov-offline/alg_recsys_recall_24h_noregion.py

# -*- coding: utf-8 -*-
import pandas as pd
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()


RULE_PARAMS = {
    'rule_params': {
        'rule66': {'cal_score_func': 2, 'return_count': 100, 'platform_return_rate': 0.001, 'view_type': 'preview'},
    },
    'data_params': config_.DATA_PARAMS,
    'params_list': [
        {'data': 'data66', 'rule': 'rule66'},
    ]
}

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"一共有多少个视频 = {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
    log_.info(f"回流量-过滤后，一共有多少个视频 = {len(day_recall_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"回流率-过滤后，一共有多少个视频 = {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(f"视频状态-过滤后，一共有多少个视频 = {len(filtered_videos)}")

    # 写入对应的redis
    now_dt = datetime.strftime(now_date, '%Y%m%d')
    day_video_ids = []
    day_recall_result = {}
    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))

    # recall:item:score:24h:
    h_24h_recall_key_name = \
        f"{config_.RECALL_KEY_NAME_PREFIX_BY_24H}{data_key}:{rule_key}:{now_dt}:{now_h}"
    log_.info("打印非地域24小时redis key:{}".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)
    else:
        log_.info(f"无数据，不写入。")

    # ---------------处理剩余结果---------------
    log_.info('开始处理剩余结果other')
    all_videos = df['videoid'].to_list()
    if data_key in ['data7', ]:
        all_filtered_videos = filter_video_status_app(all_videos)
    else:
        all_filtered_videos = filter_video_status(all_videos)
    other_videos = [video for video in all_filtered_videos if video not in day_video_ids]
    log_.info(f'过滤后剩余视频数量 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)})
    # recall:item:score:24h:other:
    other_h_24h_recall_key_name = \
        f"{config_.RECALL_KEY_NAME_PREFIX_BY_24H_OTHER}{data_key}:{rule_key}:{now_dt}:{now_h}"
    log_.info("打印非地域24小时（剩余）redis key:{}".format(other_h_24h_recall_key_name))
    if len(other_24h_recall_result) > 0:
        log_.info(f"开始写入尾部数据：count = {len(other_24h_recall_result)}, key = {other_h_24h_recall_key_name}")
        redis_helper.add_data_with_zset(key_name=other_h_24h_recall_key_name, data=other_24h_recall_result,
                                        expire_time=2 * 3600)
    else:
        log_.info(f"无尾部数据，不写入。")

    # 通知后端更新兜底视频数据
    if notify_backend is True:
        log_.info('json_data count = {}'.format(len(json_data[:5000])))
        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 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'):
        score_df_list = []
        notify_backend = param.get('notify_backend', False)
        data_key = param.get('data')
        data_param = data_params_item.get(data_key)
        rule_key = param.get('rule')
        rule_param = rule_params_item.get(rule_key)
        merge_func = rule_param.get('merge_func', 1)
        log_.info("数据采用:{},统计采用{}.".format(data_key, rule_key))
        log_.info("具体的规则是:{}.".format(rule_param))

        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)



def h_timer_check():
    try:
        project = config_.PROJECT_24H_APP_TYPE
        table = config_.TABLE_24H_APP_TYPE
        rule_params = RULE_PARAMS
        now_date = datetime.today()
        log_.info(f"开始执行: {datetime.strftime(now_date, '%Y%m%d%H')}")
        now_min = datetime.now().minute
        now_h = datetime.now().hour
        # 查看当前天级更新的数据是否已准备好
        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("当前时间{}小时，使用bottom的data，开始。".format(now_h))
            h_rank_bottom(now_date=now_date, now_h=now_h, rule_params=rule_params)
            log_.info("----------当前时间{}小时，使用bottom的data，完成----------".format(now_h))
        elif h_data_count > 0:
            log_.info('上游数据表查询数据条数 h_data_count = {}，开始计算。'.format(h_data_count))
            rank_by_h(now_date=now_date, now_h=now_h, rule_params=rule_params, project=project, table=table)
            log_.info("数据2----------正常完成----------")
        elif now_min > 40:
            log_.info('当前分钟超过40，预计执行无法完成，使用 bottom data!')
            h_rank_bottom(now_date=now_date, now_h=now_h, rule_params=rule_params)
            log_.info('----------当前分钟超过40，使用bottom的data，完成----------')
        else:
            # 数据没准备好，1分钟后重新检查
            log_.info("上游数据未就绪，等待...")
            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("文件alg_recsys_recall_24h_noregion.py:「24小时无地域」 开始执行")
    h_timer_check()