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