# -*- coding: utf-8 -*- # @ModuleName: region_rule_rank_h # @Author: Liqian # @Time: 2022/5/5 15:54 # @Software: PyCharm import datetime import pandas as pd import math from odps import ODPS from threading import Timer from my_utils import RedisHelper, get_data_from_odps, filter_video_status from my_config import set_config from log import Log config_, _ = set_config() log_ = Log() region_code = { '河北省': '130000', '山西省': '140000', '辽宁省': '210000', '吉林省': '220000', '黑龙江省': '230000', '江苏省': '320000', '浙江省': '330000', '安徽省': '340000', '福建省': '350000', '江西省': '360000', '山东省': '370000', '河南省': '410000', '湖北省': '420000', '湖南省': '430000', '广东省': '440000', '海南省': '460000', '四川省': '510000', '贵州省': '520000', '云南省': '530000', '陕西省': '610000', '甘肃省': '620000', '青海省': '630000', '台湾省': '710000', '北京': '110000', '天津': '120000', '内蒙古': '150000', '上海': '310000', '广西': '450000', '重庆': '500000', '西藏': '540000', '宁夏': '640000', '新疆': '650000', '香港': '810000', '澳门': '820000', } features = [ 'code', # 省份编码 'videoid', 'lastday_preview', # 昨日预曝光人数 'lastday_view', # 昨日曝光人数 'lastday_play', # 昨日播放人数 'lastday_share', # 昨日分享人数 'lastday_return', # 昨日回流人数 'lastday_preview_total', # 昨日预曝光次数 'lastday_view_total', # 昨日曝光次数 'lastday_play_total', # 昨日播放次数 'lastday_share_total', # 昨日分享次数 ] def 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 - datetime.timedelta(days=1), '%Y%m%d') 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(project, table, now_date): """获取特征数据""" dt = datetime.datetime.strftime(now_date, '%Y%m%d') # 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 = lastday_share/(lastday_play+1000) # backrate = lastday_return/(lastday_share+10) # ctr = lastday_play/(lastday_preview+1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr # score = sharerate * backrate * LOG(lastday_return+1) * K2 df = df.fillna(0) df['share_rate'] = df['lastday_share'] / (df['lastday_play'] + 1000) df['back_rate'] = df['lastday_return'] / (df['lastday_share'] + 10) df['log_back'] = (df['lastday_return'] + 1).apply(math.log) df['ctr'] = df['lastday_play'] / (df['lastday_preview'] + 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(df, now_date, rule_key, param, region): """ 获取符合进入召回源条件的视频 :param df: :param now_date: :param rule_key: 小时级数据进入条件 :param param: 小时级数据进入条件参数 :param region: 所属地域 :return: """ redis_helper = RedisHelper() # 获取符合进入召回源条件的视频 return_count = param.get('return_count', 1) score_value = param.get('score_rule', 0) h_recall_df = df[(df['lastday_return'] >= return_count) & (df['score'] >= score_value)] # videoid重复时，保留分值高 h_recall_df = h_recall_df.sort_values(by=['score'], ascending=False) h_recall_df = h_recall_df.drop_duplicates(subset=['videoid'], keep='first') h_recall_df['videoid'] = h_recall_df['videoid'].astype(int) h_recall_videos = h_recall_df['videoid'].to_list() log_.info(f'day_recall videos count = {len(h_recall_videos)}') # 视频状态过滤 filtered_videos = filter_video_status(h_recall_videos) log_.info('filtered_videos count = {}'.format(len(filtered_videos))) # 写入对应的redis day_recall_result = {} for video_id in filtered_videos: score = h_recall_df[h_recall_df['videoid'] == video_id]['score'] # print(score) day_recall_result[int(video_id)] = float(score) day_recall_key_name = \ f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_DAY}{region}.{rule_key}.{datetime.datetime.strftime(now_date, '%Y%m%d')}" if len(day_recall_result) > 0: redis_helper.add_data_with_zset(key_name=day_recall_key_name, data=day_recall_result, expire_time=7 * 24 * 3600) def rank_by_day(project, table, now_date, rule_params, region_code_list): # 获取特征数据 feature_df = get_feature_data(project=project, table=table, now_date=now_date - datetime.timedelta(days=1)) # rank for key, value in rule_params.items(): log_.info(f"rule = {key}, param = {value}") for region in region_code_list: log_.info(f"region = {region}") # 计算score region_df = feature_df[feature_df['code'] == region] log_.info(f'region_df count = {len(region_df)}') score_df = cal_score(df=region_df) video_rank(df=score_df, now_date=now_date, rule_key=key, param=value, region=region) # to-csv score_filename = f"score_{region}_{key}_{datetime.datetime.strftime(now_date, '%Y%m%d')}.csv" score_df.to_csv(f'./data/{score_filename}') # to-logs log_.info({"date": datetime.datetime.strftime(now_date, '%Y%m%d'), "region_code": region, "redis_key_prefix": config_.RECALL_KEY_NAME_PREFIX_REGION_BY_DAY, "rule_key": key, "score_df": score_df[['videoid', 'score']]}) def h_timer_check(): rule_params = config_.RULE_PARAMS_REGION_DAY project = config_.PROJECT_REGION_DAY table = config_.TABLE_REGION_DAY region_code_list = [code for region, code in region_code.items()] now_date = datetime.datetime.today() log_.info(f"now_date: {datetime.datetime.strftime(now_date, '%Y%m%d')}") # 查看当天更新的数据是否已准备好 h_data_count = data_check(project=project, table=table, now_date=now_date) if h_data_count > 0: log_.info(f'day_data_count = {h_data_count}') # 数据准备好，进行更新 rank_by_day(now_date=now_date, rule_params=rule_params, project=project, table=table, region_code_list=region_code_list) else: # 数据没准备好，1分钟后重新检查 Timer(60, h_timer_check).start() if __name__ == '__main__': h_timer_check()