import pandas as pd 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 utils import filter_video_status, check_table_partition_exits, filter_video_status_app, \ request_post, send_msg_to_feishu from 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}" 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)}') # 视频状态过滤 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)}') # 与筛选结果去重 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 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) """ 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 # 查看当前天级更新的数据是否已准备好 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!") 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!") 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!") 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()