rov-offline/ad_video_data_update.py

import datetime
import traceback
import multiprocessing
from threading import Timer
from utils import RedisHelper, data_check, get_feature_data, send_msg_to_feishu, send_msg_to_feishu_new
from config import set_config
from log import Log
config_, _ = set_config()
log_ = Log()
redis_helper = RedisHelper()

features = [
    'apptype',
    'videoid',
    'sharerate_all',
    'sharerate_ad'
]


def get_top10_abnormal_videos_return(dt, filter_param):
    """获取昨日各端top10中的异常视频（裂变视频）"""
    abnormal_video_project = config_.ad_model_data['top10_videos'].get('project')
    abnormal_video_table = config_.ad_model_data['top10_videos'].get('table')
    abnormal_video_features = [
        'apptype', 'videoid', 'yesterday_return', 'rank', 'multiple'
    ]
    data_count = data_check(project=abnormal_video_project, table=abnormal_video_table, dt=dt)
    top10_abnormal_videos = {}
    if data_count > 0:
        abnormal_video_df = get_feature_data(project=abnormal_video_project, table=abnormal_video_table,
                                             features=abnormal_video_features, dt=dt)
        abnormal_video_df['multiple'].fillna(0, inplace=True)
        abnormal_video_df['apptype'] = abnormal_video_df['apptype'].astype(int)
        abnormal_video_df['videoid'] = abnormal_video_df['videoid'].astype(int)
        abnormal_video_df['yesterday_return'] = abnormal_video_df['yesterday_return'].astype(int)
        abnormal_video_df['rank'] = abnormal_video_df['rank'].astype(int)
        abnormal_video_df['multiple'] = abnormal_video_df['multiple'].astype(float)
        app_type_list = list(set(abnormal_video_df['apptype'].tolist()))
        for app_type in app_type_list:
            app_type_df = abnormal_video_df[abnormal_video_df['apptype'] == app_type]
            app_type_df = app_type_df.sort_values(by=['rank'], ascending=True)
            # print(app_type_df)
            temp_video_id_list = []
            for index, item in app_type_df.iterrows():
                # print(item['rank'], item['videoid'], item['multiple'])
                if item['multiple'] > filter_param:
                    # print(item['videoid'], item['multiple'])
                    abnormal_video_id_list = temp_video_id_list.copy()
                    abnormal_video_id_list.append(int(item['videoid']))
                    top10_abnormal_videos[app_type] = abnormal_video_id_list
                    temp_video_id_list.append(int(item['videoid']))
                else:
                    temp_video_id_list.append(int(item['videoid']))
            # print(top10_abnormal_videos)
    log_.info(f"top10_abnormal_videos = {top10_abnormal_videos}")
    return top10_abnormal_videos


def predict_video_share_rate(video_initial_df, dt, data_key, data_param, top10_abnormal_videos):
    """预估视频有广告时分享率"""
    # 获取对应的视频特征
    video_df = video_initial_df.copy()
    video_df['apptype'] = video_df['apptype'].astype(int)
    video_df = video_df[video_df['apptype'] == int(data_param)]
    log_.info(f"video_df length: {len(video_df)}")
    video_df['sharerate_all'].fillna(0, inplace=True)
    video_df['sharerate_ad'].fillna(0, inplace=True)
    video_df['sharerate_all'] = video_df['sharerate_all'].astype(float)
    video_df['sharerate_ad'] = video_df['sharerate_ad'].astype(float)
    # 获取有广告时所有视频近30天的分享率
    ad_all_videos_share_rate = video_df[video_df['videoid'] == 'allvideos']['sharerate_ad'].values[0]
    video_df = video_df[video_df['videoid'] != 'allvideos']
    # 剔除异常视频数据
    video_df['videoid'] = video_df['videoid'].astype(int)
    top10_abnormal_video_ids = top10_abnormal_videos.get(int(data_param), None)
    # print(int(data_param), len(video_df), top10_abnormal_video_ids)
    if top10_abnormal_video_ids is not None:
        video_df = video_df[~video_df['videoid'].isin(top10_abnormal_video_ids)]
        # print(len(video_df))
    # 计算视频有广告时分享率
    video_df['video_ad_share_rate'] = \
        video_df['sharerate_ad'] * float(ad_all_videos_share_rate) / video_df['sharerate_all']
    video_df['video_ad_share_rate'].fillna(0, inplace=True)
    # log_.info(f"video_df: {video_df}")
    video_df = video_df[video_df['video_ad_share_rate'] != 0]
    log_.info(f"video_df filtered 0 length: {len(video_df)}")
    # 结果写入redis
    key_name = f"{config_.KEY_NAME_PREFIX_AD_VIDEO}{data_key}:{dt}"
    redis_data = {}
    for index, item in video_df.iterrows():
        redis_data[int(item['videoid'])] = item['video_ad_share_rate']
    group_ad_share_rate_mean = video_df['video_ad_share_rate'].mean()
    redis_data[-1] = group_ad_share_rate_mean
    # 异常视频给定值：mean/3
    if top10_abnormal_video_ids is not None:
        abnormal_video_param = config_.AD_ABNORMAL_VIDEOS_PARAM.get(data_key, 1)
        print(data_key, data_param, abnormal_video_param)
        for abnormal_video_id in top10_abnormal_video_ids:
            print(abnormal_video_id, group_ad_share_rate_mean, group_ad_share_rate_mean * abnormal_video_param)
            redis_data[int(abnormal_video_id)] = group_ad_share_rate_mean * abnormal_video_param
    log_.info(f"redis_data count: {len(redis_data)}")
    if len(redis_data) > 0:
        redis_helper = RedisHelper()
        redis_helper.add_data_with_zset(key_name=key_name, data=redis_data, expire_time=2 * 24 * 3600)
    return video_df


def update_videos_data(project, table, dt, update_params, top10_abnormal_videos):
    """预估视频有广告时分享率"""
    # 获取视频特征
    video_initial_df = get_feature_data(project=project, table=table, features=features, dt=dt)
    for data_key, data_param in update_params.items():
        log_.info(f"data_key = {data_key} update start...")
        predict_video_share_rate(video_initial_df=video_initial_df, dt=dt, data_key=data_key, data_param=data_param,
                                 top10_abnormal_videos=top10_abnormal_videos)
        log_.info(f"data_key = {data_key} update end!")


def timer_check(dt, video_key, video_params, top10_abnormal_videos):
    log_.info(f"video_key = {video_key}")
    project = config_.ad_model_data[video_key].get('project')
    table = config_.ad_model_data[video_key].get('table')
    # 查看当前更新的数据是否已准备好
    data_count = data_check(project=project, table=table, dt=dt)
    if data_count > 0:
        log_.info(f"ad video data count = {data_count}")
        # 数据准备好，进行更新
        update_videos_data(project=project, table=table, dt=dt, update_params=video_params,
                           top10_abnormal_videos=top10_abnormal_videos)
        log_.info(f"video_key = {video_key} ad video data update end!")
        msg_list = [
            f"env: rov-offline {config_.ENV_TEXT}",
            f"video_key: {video_key}",
            f"now_date: {dt}",
            f"finished time: {datetime.datetime.strftime(datetime.datetime.now(), '%Y%m%d %H:%M:%S')}",
        ]
        send_msg_to_feishu_new(
            webhook=config_.FEISHU_ROBOT['ad_video_update_robot'].get('webhook'),
            key_word=config_.FEISHU_ROBOT['ad_video_update_robot'].get('key_word'),
            title='广告模型视频分享率预测数据更新完成',
            msg_list=msg_list
        )

    else:
        # 数据没准备好，1分钟后重新检查
        Timer(60, timer_check, args=[dt, video_key, video_params, top10_abnormal_videos]).start()


def main():
    try:
        now_date = datetime.datetime.today()
        dt = datetime.datetime.strftime(now_date, '%Y%m%d')
        log_.info(f"now_date: {dt}")
        # 获取昨天top10中的异常视频（裂变视频）
        top10_abnormal_videos = get_top10_abnormal_videos_return(
            dt=dt, filter_param=config_.ad_model_data['top10_videos'].get('abnormal_filter_param')
        )
        update_params = config_.AD_VIDEO_DATA_PARAMS
        pool = multiprocessing.Pool(processes=len(update_params))
        for video_key, video_params in update_params.items():
            pool.apply_async(
                func=timer_check,
                args=(dt, video_key, video_params, top10_abnormal_videos)
            )
        pool.close()
        pool.join()
        # for video_key, video_params in update_params.items():
        #     timer_check(dt, video_key, video_params, top10_abnormal_videos)

    except Exception as e:
        log_.error(f"视频分享率预测数据更新失败, exception: {e}, traceback: {traceback.format_exc()}")
        msg_list = [
            f"env: rov-offline {config_.ENV_TEXT}",
            f"now time: {datetime.datetime.strftime(datetime.datetime.now(), '%Y%m%d %H:%M:%S')}",
            f"exception: {e}",
            f"traceback: {traceback.format_exc()}",
        ]
        send_msg_to_feishu_new(
            webhook=config_.FEISHU_ROBOT['ad_video_update_robot'].get('webhook'),
            key_word=config_.FEISHU_ROBOT['ad_video_update_robot'].get('key_word'),
            title='广告模型视频分享率预测数据更新失败',
            msg_list=msg_list
        )


if __name__ == '__main__':
    # timer_check()
    main()