# -*- coding: utf-8 -*- # @ModuleName: rule_rank_h_18_19 # @Author: Liqian # @Time: 2022/4/21 下午4:31 # @Software: PyCharm import time import datetime import pandas as pd import math import random from odps import ODPS from threading import Timer from get_data import get_data_from_odps from db_helper import RedisHelper, MysqlHelper from my_config import set_config from log import Log from my_utils import filter_video_status config_, env = set_config() log_ = Log() features = [ 'videoid', 'lastonehour_view', # 过去1小时曝光 'lastonehour_play', # 过去1小时播放 'lastonehour_share', # 过去1小时分享 'lastonehour_return', # 过去1小时分享，过去1小时回流 ] def h_data_check(project, table, now_date): """检查数据是否准备好""" 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: dt = datetime.datetime.strftime(now_date, '%Y%m%d%H') sql = f'select * from {project}.{table} where dt = {dt}' with odps.execute_sql(sql=sql).open_reader() as reader: data_count = reader.count except Exception as e: data_count = 0 return data_count def get_feature_data(now_date, project, table): """获取特征数据""" dt = datetime.datetime.strftime(now_date, '%Y%m%d%H') # dt = '2022041310' 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_score(df): """ 计算score :param df: 特征数据 :return: """ # score计算公式: sharerate*backrate*logback*ctr # sharerate = lastonehour_share/(lastonehour_play+1000) # backrate = lastonehour_return/(lastonehour_share+10) # ctr = lastonehour_play/(lastonehour_view+1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr # score = sharerate * backrate * LOG(lastonehour_return+1) * K2 # 视频状态过滤 log_.info(f'initial_df count = {len(df)}') video_ids = [int(video_id) for video_id in df['videoid']] filtered_result = filter_video_status(video_ids=video_ids) filter_result = set(video_ids) - set(filtered_result) df['videoid'] = df['videoid'].astype(int) filter_df = df[df['videoid'].isin(filter_result)] df = df.append(filter_df) df = df.drop_duplicates(['videoid'], keep=False) log_.info(f'filtered_df count = {len(df)}') # 计算score df = df.fillna(0) df['share_rate'] = df['lastonehour_share'] / (df['lastonehour_play'] + 1000) df['back_rate'] = df['lastonehour_return'] / (df['lastonehour_share'] + 10) df['log_back'] = (df['lastonehour_return'] + 1).apply(math.log) df['ctr'] = df['lastonehour_play'] / (df['lastonehour_view'] + 1000) df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x) df['score'] = df['share_rate'] * df['back_rate'] * df['log_back'] * df['K2'] df = df.sort_values(by=['score'], ascending=False) return df def video_rank(app_type, df, now_date, now_h, return_count): """ 根据回流数量，对视频进行二次排序 :param app_type: :param df: :param now_date: :param now_h: :param return_count: 小时级数据回流限制数 :return: """ log_.info(f'df length = {len(df)}') # 获取符合进入召回源条件的视频，进入条件：小时级回流>=20 && score>=0.005 h_recall_df = df[(df['lastonehour_return'] >= return_count) & (df['score'] >= 0.005)] h_recall_videos = h_recall_df['videoid'].to_list() log_.info(f'h_recall videos count = {len(h_recall_videos)}') # 不符合进入召回源条件的视频 df = df.append(h_recall_df) h_else_df = df.drop_duplicates(['videoid'], keep=False) h_else_df = h_else_df.sort_values(by=['score'], ascending=False) h_else_videos = h_else_df['videoid'].to_list() # 合并，给定分数 final_videos = h_recall_videos + h_else_videos final_result = {} step = round(100/len(final_videos), 3) for i, video_id in enumerate(final_videos): score = 100 - i * step final_result[int(video_id)] = score # 写入对应的redis key_name = \ f"{config_.RECALL_KEY_NAME_PREFIX_APP_TYPE}{app_type}.{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}" if len(final_result) > 0: redis_helper = RedisHelper() redis_helper.add_data_with_zset(key_name=key_name, data=final_result, expire_time=23 * 3600) def rank_by_h(app_type, now_date, now_h, return_count_list, project, table): # 获取特征数据 feature_df = get_feature_data(now_date=now_date, project=project, table=table) # 计算score score_df = cal_score(df=feature_df) # rank for cnt in return_count_list: log_.info(f"return_count = {cnt}") video_rank(app_type=app_type, df=score_df, now_date=now_date, now_h=now_h, return_count=cnt) # to-csv score_filename = f"score_{app_type}_{datetime.datetime.strftime(now_date, '%Y%m%d%H')}.csv" score_df.to_csv(f'./data/{score_filename}') def h_rank_bottom(app_type, now_date, now_h): """未按时更新数据，用上一小时结果作为当前小时的数据""" log_.info(f"app_type = {app_type}") # 获取rov模型结果 redis_helper = RedisHelper() if now_h == 0: redis_dt = datetime.datetime.strftime(now_date - datetime.timedelta(days=1), '%Y%m%d') redis_h = 23 else: redis_dt = datetime.datetime.strftime(now_date, '%Y%m%d') redis_h = now_h - 1 key_name = f"{config_.RECALL_KEY_NAME_PREFIX_APP_TYPE}{app_type}.{redis_dt}.{redis_h}" initial_data = redis_helper.get_all_data_from_zset(key_name=key_name, with_scores=True) final_data = dict() for video_id, score in initial_data: final_data[video_id] = score # 存入对应的redis final_key_name = \ f"{config_.RECALL_KEY_NAME_PREFIX_APP_TYPE}{app_type}.{datetime.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=23 * 3600) def h_timer_check(app_type): log_.info(f"app_type = {app_type}") return_count_list = [20] now_date = datetime.datetime.today() log_.info(f"now_date: {datetime.datetime.strftime(now_date, '%Y%m%d%H')}") now_h = datetime.datetime.now().hour now_min = datetime.datetime.now().minute if now_h == 0: h_rank_bottom(app_type=app_type, now_date=now_date, now_h=now_h) return # 查看当前小时更新的数据是否已准备好 project = config_.PREDICT_PROJECT_18_19[str(app_type)] table = config_.PREDICT_TABLE_18_19[str(app_type)] h_data_count = h_data_check(project=project, table=table, now_date=now_date) if h_data_count > 0: log_.info(f'h_data_count = {h_data_count}') # 数据准备好，进行更新 rank_by_h(app_type=app_type, now_date=now_date, now_h=now_h, return_count_list=return_count_list, project=project, table=table) elif now_min > 50: log_.info('h_recall data is None, use bottom data!') h_rank_bottom(app_type=app_type, now_date=now_date, now_h=now_h) else: # 数据没准备好，1分钟后重新检查 Timer(60, h_timer_check, args=[app_type]).start() def predict(app_type_list): for app_type in app_type_list: h_timer_check(app_type=app_type) def predict_test(app_type_list, count): now_date = datetime.datetime.today() now_h = datetime.datetime.now().hour log_.info(f"now_date = {datetime.datetime.strftime(now_date, '%Y%m%d%H')}, now_h = {now_h}") # 获取测试环境中最近发布的40000条视频 sql = "SELECT id FROM wx_video ORDER BY id DESC LIMIT 40000;" mysql_helper = MysqlHelper(mysql_info=config_.MYSQL_INFO) data = mysql_helper.get_data(sql=sql) video_ids = [int(video[0]) for video in data] # 视频状态过滤 filtered_videos = filter_video_status(video_ids) log_.info('filtered_videos count = {}'.format(len(filtered_videos))) for app_type in app_type_list: log_.info(f"app_type = {app_type}") videos_temp = random.sample(filtered_videos, count) redis_data_temp = {} csv_data_temp = [] for video_id in videos_temp: score = random.uniform(0, 100) redis_data_temp[video_id] = score csv_data_temp.append({'video_id': video_id, 'rov_score': score}) # 打包预测结果存入csv score_df = pd.DataFrame(data=csv_data_temp, columns=['video_id', 'rov_score']) score_df = score_df.sort_values(by=['rov_score'], ascending=False) score_filename = f"score_{app_type}_{datetime.datetime.strftime(now_date, '%Y%m%d%H')}.csv" score_df.to_csv(f'./data/{score_filename}', index=False) # 存入对应的redis key_name = \ f"{config_.RECALL_KEY_NAME_PREFIX_APP_TYPE}{app_type}.{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}" redis_helper = RedisHelper() redis_helper.add_data_with_zset(key_name=key_name, data=redis_data_temp) log_.info('data to redis finished!') if __name__ == '__main__': app_type_list = [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']] log_.info(f'appType: {app_type_list} predict start...') predict_start = time.time() if env in ['dev', 'test']: predict_test(app_type_list=app_type_list, count=300) elif env in ['pre', 'pro']: predict(app_type_list=app_type_list) else: log_.error('env error') predict_end = time.time() log_.info(f'appType: {app_type_list} predict end, execute time = {(predict_end - predict_start) * 1000}ms')