rov-offline/ad_video_data_update.py

import datetime
import traceback
from threading import Timer
from utils import RedisHelper, data_check, get_feature_data, send_msg_to_feishu
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 predict_video_share_rate(video_initial_df, dt, data_key, data_param):
    """预估视频有广告时分享率"""
    # 获取对应的视频特征
    video_df = video_initial_df.copy()
    video_df['apptype'] = video_df['apptype'].astype(int)
    video_df = video_df[video_df['apptype'] == int(data_param)]
    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['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)
    video_df = video_df[video_df['video_ad_share_rate'] != 0]
    # 结果写入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
    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):
    """预估视频有广告时分享率"""
    # 获取视频特征
    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)
        log_.info(f"data_key = {data_key} update end!")


def timer_check():
    try:
        update_params = config_.AD_VIDEO_DATA_PARAMS
        project = config_.ad_model_data['videos_share_rate'].get('project')
        table = config_.ad_model_data['videos_share_rate'].get('table')
        now_date = datetime.datetime.today()
        dt = datetime.datetime.strftime(now_date, '%Y%m%d')
        log_.info(f"now_date: {dt}")
        now_min = datetime.datetime.now().minute
        # 查看当前更新的数据是否已准备好
        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=update_params)
            log_.info(f"ad video data update end!")
        # elif now_min > 45:
        #     log_.info('ad video data is None!')
        #     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} - 视频分享率数据未准备好！\n"
        #                  f"traceback: {traceback.format_exc()}"
        #     )
        else:
            # 数据没准备好，1分钟后重新检查
            Timer(60, timer_check).start()

    except Exception as e:
        log_.error(f"视频分享率预测数据更新失败, 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} - 视频分享率预测数据更新失败\n"
                     f"exception: {e}\n"
                     f"traceback: {traceback.format_exc()}"
        )


if __name__ == '__main__':
    timer_check()