# -*- coding: utf-8 -*- import multiprocessing import traceback import gevent import datetime import pandas as pd import math from functools import reduce from odps import ODPS from threading import Timer from my_utils import MysqlHelper, RedisHelper, get_data_from_odps, filter_video_status, filter_shield_video, \ check_table_partition_exits, filter_video_status_app, send_msg_to_feishu, filter_political_videos from my_config import set_config from log import Log from check_video_limit_distribute import update_limit_video_score config_, _ = set_config() log_ = Log() region_code = config_.REGION_CODE RULE_PARAMS = { 'rule_params': { 'rule66': { 'view_type': 'video-show-region', # 'score_func': '20240223', # 'lastonehour_allreturn': "1", 'region_24h_rule_key': 'rule66', '24h_rule_key': 'rule66' }, 'rule67': { 'view_type': 'video-show-region', 'platform_return_rate': 0.001, 'region_24h_rule_key': 'rule66', '24h_rule_key': 'rule66', 'h_rule_key': 'rule66' }, 'rule68': { 'view_type': 'video-show-region', 'platform_return_rate': 0.001, 'region_24h_rule_key': 'rule66', '24h_rule_key': 'rule66', 'score_func': 'back_rate_exponential_weighting1' }, }, 'data_params': config_.DATA_PARAMS, 'params_list': [ # 532 {'data': 'data66', 'rule': 'rule66'}, # 523-> 523 & 518 # {'data': 'data66', 'rule': 'rule67'}, # 523->510 # {'data': 'data66', 'rule': 'rule68'}, # 523->514 # {'data': 'data66', 'rule': 'rule69'}, # 523->518 ], } features = [ 'apptype', 'code', 'videoid', 'lastonehour_preview', # 过去1小时预曝光人数 - 区分地域 'lastonehour_view', # 过去1小时曝光人数 - 区分地域 'lastonehour_play', # 过去1小时播放人数 - 区分地域 'lastonehour_share', # 过去1小时分享人数 - 区分地域 'lastonehour_return', # 过去1小时分享，过去1小时回流人数 - 区分地域 'lastonehour_preview_total', # 过去1小时预曝光次数 - 区分地域 'lastonehour_view_total', # 过去1小时曝光次数 - 区分地域 'lastonehour_play_total', # 过去1小时播放次数 - 区分地域 'lastonehour_share_total', # 过去1小时分享次数 - 区分地域 'platform_return', 'lastonehour_show', # 不区分地域 'lastonehour_show_region', # 地域分组 'lasttwohour_share', # h-2小时分享人数 'lasttwohour_return_now', # h-2分享，过去1小时回流人数 'lasttwohour_return', # h-2分享，h-2回流人数 'lastthreehour_share', # h-3小时分享人数 'lastthreehour_return_now', # h-3分享，过去1小时回流人数 'lastthreehour_return', # h-3分享，h-3回流人数 'lastonehour_return_new', # 过去1小时分享，过去1小时回流人数（回流统计为对应地域分享带回的回流，分享限制地域，回流不限制地域） 'lasttwohour_return_now_new', # h-2分享，过去1小时回流人数（回流统计为对应地域分享带回的回流，分享限制地域，回流不限制地域） 'lasttwohour_return_new', # h-2分享，h-2回流人数（回流统计为对应地域分享带回的回流，分享限制地域，回流不限制地域） 'lastthreehour_return_now_new', # h-3分享，过去1小时回流人数（回流统计为对应地域分享带回的回流，分享限制地域，回流不限制地域） 'lastthreehour_return_new', # h-3分享，h-3回流人数（回流统计为对应地域分享带回的回流，分享限制地域，回流不限制地域） 'platform_return_new', # 平台分发回流（回流统计为对应地域分享带回的回流，分享限制地域，回流不限制地域） 'lastonehour_allreturn', 'lastonehour_allreturn_sharecnt' ] def get_region_code(region): """获取省份对应的code""" mysql_helper = MysqlHelper(mysql_info=config_.MYSQL_INFO) sql = f"SELECT ad_code FROM region_adcode WHERE parent_id = 0 AND region LIKE '{region}%';" ad_code = mysql_helper.get_data(sql=sql) return ad_code[0][0] 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') # 测试 张博 check_res = check_table_partition_exits(date=dt, project=project, table=table) if check_res: sql = f'select * from {project}.{table} where dt = "{dt}"' print("zhangbo-sql-是否有数据") print(sql) 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_rov_redis_key(now_date): """获取rov模型结果存放key""" redis_helper = RedisHelper() now_dt = datetime.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.datetime.strftime(now_date - datetime.timedelta(days=1), '%Y%m%d') key_name = f'{config_.RECALL_KEY_NAME_PREFIX}{pre_dt}' return key_name def get_day_30day_videos(now_date, data_key, rule_key): """获取天级更新相对30天的视频id""" redis_helper = RedisHelper() day_30day_recall_key_prefix = config_.RECALL_KEY_NAME_PREFIX_30DAY now_dt = datetime.datetime.strftime(now_date, '%Y%m%d') day_30day_recall_key_name = f"{day_30day_recall_key_prefix}{data_key}:{rule_key}:{now_dt}" if not redis_helper.key_exists(key_name=day_30day_recall_key_name): redis_dt = datetime.datetime.strftime((now_date - datetime.timedelta(days=1)), '%Y%m%d') day_30day_recall_key_name = f"{day_30day_recall_key_prefix}{data_key}:{rule_key}:{redis_dt}" data = redis_helper.get_all_data_from_zset(key_name=day_30day_recall_key_name, with_scores=True) if data is None: return None video_ids = [int(video_id) for video_id, _ in data] return video_ids def get_feature_data(project, table, now_date): """获取特征数据""" dt = datetime.datetime.strftime(now_date, '%Y%m%d%H') # 张博 测试 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_initial_20240223(df, param): """ 计算score :param df: 特征数据 :param param: 规则参数 :return: """ log_.info("进入了cal_score_initial_20240223") 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['back_rate_new'] = (df['lastonehour_return'] + 1) / (df['lastonehour_share'] + 10) df['back_rate_all'] = df['lastonehour_allreturn'] / (df['lastonehour_allreturn_sharecnt'] + 10) df['log_back'] = (df['lastonehour_return'] + 1).apply(math.log) df['log_back_all'] = (df['lastonehour_allreturn'] + 1).apply(math.log) if param.get('view_type', None) == 'video-show': df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show'] + 1000) elif param.get('view_type', None) == 'video-show-region': df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show_region'] + 1000) else: df['ctr'] = df['lastonehour_play'] / (df['lastonehour_preview'] + 1000) df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x) df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return'] df['score'] = df['share_rate'] * ( df['back_rate_new'] + 0.01 * df['back_rate_all'] ) * ( df['log_back'] + 0.01 * df['log_back_all'] ) * df['K2'] df = df.sort_values(by=['score'], ascending=False) return df def cal_score_initial(df, param): """ 计算score :param df: 特征数据 :param param: 规则参数 :return: """ # score计算公式: sharerate*backrate*logback*ctr # sharerate = lastonehour_share/(lastonehour_play+1000) # backrate = lastonehour_return/(lastonehour_share+10) # ctr = lastonehour_play/(lastonehour_preview+1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr # score = sharerate * backrate * LOG(lastonehour_return+1) * K2 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) if param.get('view_type', None) == 'video-show': df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show'] + 1000) elif param.get('view_type', None) == 'video-show-region': df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show_region'] + 1000) else: df['ctr'] = df['lastonehour_play'] / (df['lastonehour_preview'] + 1000) df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x) df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return'] df['score1'] = df['share_rate'] * df['back_rate'] * df['log_back'] * df['K2'] 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 cal_score_add_return(df, param): # score计算公式: sharerate*(backrate*logback + backrate2*logback_now2 + backrate3*logback_now3)*ctr # sharerate = lastonehour_share/(lastonehour_play+1000) # backrate = lastonehour_return/(lastonehour_share+10) # backrate2 = lasttwohour_return_now/(lasttwohour_share+10) # backrate3 = lastthreehour_return_now/(lastthreehour_share+10) # ctr = lastonehour_play/(lastonehour_preview+1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr # score = k2 * sharerate * (backrate * LOG(lastonehour_return+1) + backrate_2 * LOG(lasttwohour_return_now+1) + backrate_3 * LOG(lastthreehour_return_now+1)) 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['back_rate2'] = df['lasttwohour_return_now'] / (df['lasttwohour_share'] + 10) df['log_back2'] = (df['lasttwohour_return_now'] + 1).apply(math.log) df['back_rate3'] = df['lastthreehour_return_now'] / (df['lastthreehour_share'] + 10) df['log_back3'] = (df['lastthreehour_return_now'] + 1).apply(math.log) if param.get('view_type', None) == 'video-show': df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show'] + 1000) elif param.get('view_type', None) == 'video-show-region': df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show_region'] + 1000) else: df['ctr'] = df['lastonehour_play'] / (df['lastonehour_preview'] + 1000) df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x) df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return'] df['score'] = df['K2'] * df['share_rate'] * ( df['back_rate'] * df['log_back'] + df['back_rate2'] * df['log_back2'] + df['back_rate3'] * df['log_back3'] ) df = df.sort_values(by=['score'], ascending=False) return df def cal_score_multiply_return_retention(df, param): # score计算公式: k2 * sharerate * backrate * LOG(lastonehour_return+1) * 前两小时回流留存 # sharerate = lastonehour_share/(lastonehour_play+1000) # backrate = lastonehour_return/(lastonehour_share+10) # ctr = lastonehour_play/(lastonehour_preview+1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr # 前两小时回流留存 return_retention_initial = (lasttwohour_return_now + lastthreehour_return_now)/(lasttwohour_return + lastthreehour_return + 1) # return_retention = 0.5 if return_retention_initial == 0 else return_retention_initial # score = k2 * sharerate * backrate * LOG(lastonehour_return+1) * return_retention 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) if param.get('view_type', None) == 'video-show': df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show'] + 1000) elif param.get('view_type', None) == 'video-show-region': df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show_region'] + 1000) else: df['ctr'] = df['lastonehour_play'] / (df['lastonehour_preview'] + 1000) df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x) df['return_retention_initial'] = (df['lasttwohour_return_now'] + df['lastthreehour_return_now']) / \ (df['lasttwohour_return'] + df['lastthreehour_return'] + 1) df['return_retention'] = df['return_retention_initial'].apply(lambda x: 0.5 if x == 0 else x) df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return'] df['score'] = df['K2'] * df['share_rate'] * df['back_rate'] * df['log_back'] * df['return_retention'] df = df.sort_values(by=['score'], ascending=False) return df def cal_score_update_backrate(df, param): # score计算公式: k2 * sharerate * (backrate + backrate * backrate_2 * backrate_3) * LOG(lastonehour_return+1) # sharerate = lastonehour_share/(lastonehour_play+1000) # backrate = lastonehour_return/(lastonehour_share+10) # backrate2 = lasttwohour_return_now/(lasttwohour_share+10) # backrate3 = lastthreehour_return_now/(lastthreehour_share+10) # ctr = lastonehour_play/(lastonehour_preview+1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr # backrate1_3_initial = backrate * backrate_2 * backrate_3 # backrate1_3 = 0.02 if backrate1_3_initial == 0 else backrate1_3_initial # score = k2 * sharerate * (backrate + backrate1_3) * LOG(lastonehour_return+1) 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['back_rate2'] = df['lasttwohour_return_now'] / (df['lasttwohour_share'] + 10) df['back_rate3'] = df['lastthreehour_return_now'] / (df['lastthreehour_share'] + 10) df['log_back'] = (df['lastonehour_return'] + 1).apply(math.log) if param.get('view_type', None) == 'video-show': df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show'] + 1000) elif param.get('view_type', None) == 'video-show-region': df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show_region'] + 1000) else: df['ctr'] = df['lastonehour_play'] / (df['lastonehour_preview'] + 1000) df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x) df['backrate1_3_initial'] = df['back_rate'] * df['back_rate2'] * df['back_rate3'] df['backrate1_3'] = df['backrate1_3_initial'].apply(lambda x: 0.02 if x == 0 else x) df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return'] df['score'] = df['K2'] * df['share_rate'] * (df['back_rate'] + df['backrate1_3']) * df['log_back'] df = df.sort_values(by=['score'], ascending=False) return df def cal_score_with_new_return(df, param): # 回流数据使用 分享限制地域，回流不限制地域 统计数据 # score计算公式: sharerate*backrate*logback*ctr # sharerate = lastonehour_share/(lastonehour_play+1000) # backrate = lastonehour_return_new/(lastonehour_share+10) # ctr = lastonehour_play/(lastonehour_preview+1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr # score = sharerate * backrate * LOG(lastonehour_return_new+1) * K2 df = df.fillna(0) df['share_rate'] = df['lastonehour_share'] / (df['lastonehour_play'] + 1000) df['back_rate'] = df['lastonehour_return_new'] / (df['lastonehour_share'] + 10) df['log_back'] = (df['lastonehour_return_new'] + 1).apply(math.log) if param.get('view_type', None) == 'video-show': df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show'] + 1000) elif param.get('view_type', None) == 'video-show-region': df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show_region'] + 1000) else: df['ctr'] = df['lastonehour_play'] / (df['lastonehour_preview'] + 1000) df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x) df['platform_return_rate'] = df['platform_return_new'] / df['lastonehour_return_new'] df['score'] = df['share_rate'] * df['back_rate'] * df['log_back'] * df['K2'] df = df.sort_values(by=['score'], ascending=False) return df def cal_score_multiply_return_retention_with_new_return(df, param): # 回流数据使用 分享限制地域，回流不限制地域 统计数据 # score计算公式: k2 * sharerate * backrate * LOG(lastonehour_return_new+1) * 前两小时回流留存 # sharerate = lastonehour_share/(lastonehour_play+1000) # backrate = lastonehour_return_new/(lastonehour_share+10) # ctr = lastonehour_play/(lastonehour_preview+1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr # 前两小时回流留存 return_retention_initial = (lasttwohour_return_now_new + lastthreehour_return_now_new)/(lasttwohour_return_new + lastthreehour_return_new + 1) # return_retention = 0.5 if return_retention_initial == 0 else return_retention_initial # score = k2 * sharerate * backrate * LOG(lastonehour_return_new+1) * return_retention df = df.fillna(0) df['share_rate'] = df['lastonehour_share'] / (df['lastonehour_play'] + 1000) df['back_rate'] = df['lastonehour_return_new'] / (df['lastonehour_share'] + 10) df['log_back'] = (df['lastonehour_return_new'] + 1).apply(math.log) if param.get('view_type', None) == 'video-show': df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show'] + 1000) elif param.get('view_type', None) == 'video-show-region': df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show_region'] + 1000) else: df['ctr'] = df['lastonehour_play'] / (df['lastonehour_preview'] + 1000) df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x) df['return_retention_initial'] = (df['lasttwohour_return_now_new'] + df['lastthreehour_return_now_new']) / \ (df['lasttwohour_return_new'] + df['lastthreehour_return_new'] + 1) df['return_retention'] = df['return_retention_initial'].apply(lambda x: 0.5 if x == 0 else x) df['platform_return_rate'] = df['platform_return_new'] / df['lastonehour_return_new'] df['score'] = df['K2'] * df['share_rate'] * df['back_rate'] * df['log_back'] * df['return_retention'] df = df.sort_values(by=['score'], ascending=False) return df def cal_score_with_back_view0(df, param): # score = sharerate*backrate*log(return+1)*CTR, # sharerate=(lastonehour_share+1)/(lastonehour_play+1000) # backrate=(lastonehour_return+1)/(lastonehour_share+10) # CTR=(lastonehour_play+1)/(lastonehour_view+100), ctr不进行校正 df = df.fillna(0) df['share_rate'] = (df['lastonehour_share'] + 1) / (df['lastonehour_play'] + 1000) df['back_rate'] = (df['lastonehour_return'] + 1) / (df['lastonehour_share'] + 10) df['log_back'] = (df['lastonehour_return'] + 1).apply(math.log) df['ctr'] = (df['lastonehour_play'] + 1) / (df['lastonehour_view'] + 100) df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return'] df['score'] = df['share_rate'] * df['back_rate'] * df['log_back'] * df['ctr'] df = df.sort_values(by=['score'], ascending=False) return df def cal_score_with_back_view1(df, param): # score = back_play_rate*log(return+1)*CTR, # back_play_rate=(lastonehour_return+1)/(lastonehour_play+1000) # CTR=(lastonehour_play+1)/(lastonehour_view+100), ctr不进行校正 df = df.fillna(0) df['back_play_rate'] = (df['lastonehour_return'] + 1) / (df['lastonehour_play'] + 1000) df['log_back'] = (df['lastonehour_return'] + 1).apply(math.log) df['ctr'] = (df['lastonehour_play'] + 1) / (df['lastonehour_view'] + 100) df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return'] df['score'] = df['back_play_rate'] * df['log_back'] * df['ctr'] df = df.sort_values(by=['score'], ascending=False) return df def cal_score_with_back_rate_exponential_weighting1(df, param): """ 计算score :param df: 特征数据 :param param: 规则参数 :return: """ # score计算公式: score = sharerate * backrate ^ 2 * LOG(lastonehour_return + 1) * K2 # sharerate = lastonehour_share / (lastonehour_play + 1000) # backrate = lastonehour_return / (lastonehour_share + 10) # ctr = lastonehour_play / (lastonehour_show + 1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr 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) if param.get('view_type', None) == 'video-show': df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show'] + 1000) elif param.get('view_type', None) == 'video-show-region': df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show_region'] + 1000) else: df['ctr'] = df['lastonehour_play'] / (df['lastonehour_preview'] + 1000) df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x) df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return'] df['score'] = df['share_rate'] * df['back_rate'] ** 2 * df['log_back'] * df['K2'] df = df.sort_values(by=['score'], ascending=False) return df def cal_score_with_back_rate_exponential_weighting2(df, param): """ 计算score :param df: 特征数据 :param param: 规则参数 :return: """ # score计算公式: score = sharerate ^ 0.5 * backrate ^ 2 * LOG(lastonehour_return + 1) * K2 ^ 0.5 # sharerate = lastonehour_share / (lastonehour_play + 1000) # backrate = lastonehour_return / (lastonehour_share + 10) # ctr = lastonehour_play / (lastonehour_show + 1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr 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) if param.get('view_type', None) == 'video-show': df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show'] + 1000) elif param.get('view_type', None) == 'video-show-region': df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show_region'] + 1000) else: df['ctr'] = df['lastonehour_play'] / (df['lastonehour_preview'] + 1000) df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x) df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return'] df['score'] = df['share_rate'] ** 0.5 * df['back_rate'] ** 2 * df['log_back'] * df['K2'] ** 0.5 df = df.sort_values(by=['score'], ascending=False) return df def cal_score_with_back_rate_by_rank_weighting(df, param): """ add by sunmingze 20231123 计算score :param df: 特征数据 :param param: 规则参数 :return: """ # score计算公式: score = 1 / sharerate(rank)^0.5 + 5 / backrate(rank)^0.5 + 10 / LOG(lastonehour_return +1)(rank) ^0.5 # + 1 / K2(rank)^0.5 # sharerate = lastonehour_share / (lastonehour_play + 1000) # backrate = lastonehour_return / (lastonehour_share + 10) # ctr = lastonehour_play / (lastonehour_show + 1000), 对ctr限最大值：K2 = 0.6 if ctr > 0.6 else ctr 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) if param.get('view_type', None) == 'video-show': df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show'] + 1000) elif param.get('view_type', None) == 'video-show-region': df['ctr'] = df['lastonehour_play'] / (df['lastonehour_show_region'] + 1000) else: df['ctr'] = df['lastonehour_play'] / (df['lastonehour_preview'] + 1000) df['K2'] = df['ctr'].apply(lambda x: 0.6 if x > 0.6 else x) df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return'] # 分别的得到sharerate、backrate、K值、return人数的序关系 df['rank_by_sharerate'] = df['share_rate'].rank(ascending=0, method='dense') df['rank_by_backrate'] = df['back_rate'].rank(ascending=0, method='dense') df['rank_by_K2'] = df['K2'].rank(ascending=0, method='dense') df['rank_by_logback'] = df['log_back'].rank(ascending=0, method='dense') # 计算基于序的加法关系函数 df['score'] = 1/(df['rank_by_sharerate'] + 10) + 5/(df['rank_by_backrate'] + 10) df['score'] = df['score'] + 5/(df['rank_by_logback'] + 10) + 1/(df['rank_by_K2'] + 10) df = df.sort_values(by=['score'], ascending=False) return df def cal_score(df, param): if param.get('return_data', None) == 'share_region_return': if param.get('score_func', None) == 'multiply_return_retention': df = cal_score_multiply_return_retention_with_new_return(df=df, param=param) else: df = cal_score_with_new_return(df=df, param=param) else: if param.get('score_func', None) == 'add_backrate*log(return+1)': df = cal_score_add_return(df=df, param=param) elif param.get('score_func', None) == 'multiply_return_retention': df = cal_score_multiply_return_retention(df=df, param=param) elif param.get('score_func', None) == 'update_backrate': df = cal_score_update_backrate(df=df, param=param) elif param.get('score_func', None) == 'back_view0': df = cal_score_with_back_view0(df=df, param=param) elif param.get('score_func', None) == 'back_view1': df = cal_score_with_back_view1(df=df, param=param) elif param.get('score_func', None) == 'back_rate_exponential_weighting1': df = cal_score_with_back_rate_exponential_weighting1(df=df, param=param) elif param.get('score_func', None) == 'back_rate_exponential_weighting2': df = cal_score_with_back_rate_exponential_weighting2(df=df, param=param) elif param.get('score_func', None) == 'back_rate_rank_weighting': df = cal_score_with_back_rate_by_rank_weighting(df=df, param=param) elif param.get('score_func', None) == '20240223': df = cal_score_initial_20240223(df=df, param=param) else: df = cal_score_initial(df=df, param=param) return df def add_func1(initial_df, pre_h_df): """当前小时级数据与前几个小时数据合并""" score_list = initial_df['score'].to_list() if len(score_list) > 0: min_score = min(score_list) else: min_score = 0 pre_h_df = pre_h_df[pre_h_df['score'] > min_score] df = pd.concat([initial_df, pre_h_df], ignore_index=True) # videoid去重，保留分值高 df['videoid'] = df['videoid'].astype(int) df = df.sort_values(by=['score'], ascending=False) df = df.drop_duplicates(subset=['videoid'], keep="first") return df def add_func2(initial_df, pre_h_df): """当前小时级数据与前几个小时数据合并: 当前小时存在的视频以当前小时为准，否则以高分为主""" score_list = initial_df['score'].to_list() if len(score_list) > 0: min_score = min(score_list) else: min_score = 0 initial_video_id_list = initial_df['videoid'].to_list() pre_h_df = pre_h_df[pre_h_df['score'] > min_score] pre_h_df = pre_h_df[~pre_h_df['videoid'].isin(initial_video_id_list)] df = pd.concat([initial_df, pre_h_df], ignore_index=True) # videoid去重，保留分值高 df['videoid'] = df['videoid'].astype(int) df = df.sort_values(by=['score'], ascending=False) df = df.drop_duplicates(subset=['videoid'], keep="first") return df def add_videos(initial_df, now_date, rule_key, region, data_key, hour_count, top, add_func): """ 地域小时级数据列表中增加前6h优质视频 :param initial_df: 地域小时级筛选结果 :param now_date: :param data_key: :param region: :param rule_key: :param hour_count: 前几个小时, type-int :param top: type-int :return: df """ redis_helper = RedisHelper() pre_h_data = [] for i in range(1, hour_count+1): pre_date = now_date - datetime.timedelta(hours=i) pre_h = pre_date.hour pre_h_recall_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H}{region}:{data_key}:{rule_key}:" \ f"{datetime.datetime.strftime(pre_date, '%Y%m%d')}:{pre_h}" pre_h_top_data = redis_helper.get_data_zset_with_index(key_name=pre_h_recall_key_name, start=0, end=top-1, desc=True, with_scores=True) if pre_h_top_data is None: continue pre_h_data.extend(pre_h_top_data) pre_h_df = pd.DataFrame(data=pre_h_data, columns=['videoid', 'score']) if add_func == 'func2': df = add_func2(initial_df=initial_df, pre_h_df=pre_h_df) else: df = add_func1(initial_df=initial_df, pre_h_df=pre_h_df) return df def video_rank(df, now_date, now_h, rule_key, param, region, data_key, rule_rank_h_flag, add_videos_with_pre_h=False, hour_count=0): """ 获取符合进入召回源条件的视频，与每日更新的rov模型结果视频列表进行合并 :param df: :param now_date: :param now_h: :param rule_key: 小时级数据进入条件 :param param: 小时级数据进入条件参数 :param region: 所属地域 :return: """ redis_helper = RedisHelper() # 获取符合进入召回源条件的视频，进入条件：小时级回流>=20 && score>=0.005 return_count = param.get('return_count', 1) score_value = param.get('score_rule', 0) platform_return_rate = param.get('platform_return_rate', 0) # h_recall_df = df[(df['lastonehour_return'] >= return_count) & (df['score'] >= score_value) # & (df['platform_return_rate'] >= platform_return_rate)] h_recall_df = df[ (df['lastonehour_return'] >= return_count) & (df['score'] >= score_value) & (df['platform_return_rate'] >= platform_return_rate) ] if "lastonehour_allreturn" in param.keys(): log_.info("采用 lastonehour_allreturn 过滤") h_recall_df = df[ (df['lastonehour_allreturn'] > 0) ] # try: # if "return_countv2" in param.keys() and "platform_return_ratev2" in param.keys(): # return_countv2 = param["return_countv2"] # platform_return_ratev2 = param["platform_return_ratev2"] # h_recall_df = h_recall_df[ # df['platform_return_rate'] >= platform_return_ratev2 | # (df['platform_return_rate'] < platform_return_ratev2 & df['lastonehour_return'] > return_countv2) # ] # except Exception as e: # log_.error("return_countv2 is wrong with{}".format(e)) # 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) log_.info(f"各种规则过滤后，一共有多少个视频 = {len(h_recall_df)}") # 增加打捞的优质视频 if add_videos_with_pre_h is True: add_func = param.get('add_func', None) h_recall_df = add_videos(initial_df=h_recall_df, now_date=now_date, rule_key=rule_key, region=region, data_key=data_key, hour_count=hour_count, top=10, add_func=add_func) log_.info(f"打捞优质视频完成") h_recall_videos = h_recall_df['videoid'].to_list() log_.info(f"各种规则增加后，一共有多少个视频 = {len(h_recall_videos)}") # 视频状态过滤 if data_key in ['data7', ]: filtered_videos = filter_video_status_app(h_recall_videos) else: filtered_videos = filter_video_status(h_recall_videos) # 屏蔽视频过滤 shield_config = param.get('shield_config', config_.SHIELD_CONFIG) shield_key_name_list = shield_config.get(region, None) if shield_key_name_list is not None: filtered_videos = filter_shield_video(video_ids=filtered_videos, shield_key_name_list=shield_key_name_list) # 涉政视频过滤 political_filter = param.get('political_filter', None) if political_filter is True: filtered_videos = filter_political_videos(video_ids=filtered_videos) log_.info(f"视频状态-涉政等-过滤后，一共有多少个视频 = {len(filtered_videos)}") h_video_ids = [] by_30day_rule_key = param.get('30day_rule_key', None) if by_30day_rule_key is not None: # 与相对30天列表去重 h_video_ids = get_day_30day_videos(now_date=now_date, data_key=data_key, rule_key=by_30day_rule_key) if h_video_ids is not None: filtered_videos = [video_id for video_id in filtered_videos if int(video_id) not in h_video_ids] # 写入对应的redis h_recall_result = {} for video_id in filtered_videos: score = h_recall_df[h_recall_df['videoid'] == video_id]['score'] h_recall_result[int(video_id)] = float(score) h_video_ids.append(int(video_id)) h_recall_key_name = \ f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H}{region}:{data_key}:{rule_key}:" \ f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}" log_.info("打印地域1小时的某个地域{},redis key:{}".format(region, h_recall_key_name)) if len(h_recall_result) > 0: log_.info(f"开始写入头部数据：count = {len(h_recall_result)}, key = {h_recall_key_name}") redis_helper.add_data_with_zset(key_name=h_recall_key_name, data=h_recall_result, expire_time=2 * 24 * 3600) # 限流视频score调整 tmp = update_limit_video_score(initial_videos=h_recall_result, key_name=h_recall_key_name) if tmp: log_.info(f"走了限流逻辑后：count = {len(h_recall_result)}, key = {h_recall_key_name}") else: log_.info("走了限流逻辑，但没更改redis，未生效。") # 清空线上过滤应用列表 # redis_helper.del_keys(key_name=f"{config_.REGION_H_VIDEO_FILER}{region}.{app_type}.{data_key}.{rule_key}") else: log_.info(f"无数据，不写入。") # h_rule_key = param.get('h_rule_key', None) # region_24h_rule_key = param.get('region_24h_rule_key', 'rule1') # by_24h_rule_key = param.get('24h_rule_key', None) # by_48h_rule_key = param.get('48h_rule_key', None) # dup_remove = param.get('dup_remove', True) # # 与其他召回视频池去重，存入对应的redis # dup_to_redis(h_video_ids=h_video_ids, now_date=now_date, now_h=now_h, rule_key=rule_key, h_rule_key=h_rule_key, # region_24h_rule_key=region_24h_rule_key, by_24h_rule_key=by_24h_rule_key, # by_48h_rule_key=by_48h_rule_key, region=region, data_key=data_key, # rule_rank_h_flag=rule_rank_h_flag, political_filter=political_filter, # shield_config=shield_config, dup_remove=dup_remove) def dup_data(h_video_ids, initial_key_name, dup_key_name, region, political_filter, shield_config, dup_remove): redis_helper = RedisHelper() if redis_helper.key_exists(key_name=initial_key_name): initial_data = redis_helper.get_all_data_from_zset(key_name=initial_key_name, with_scores=True) # 屏蔽视频过滤 initial_video_ids = [int(video_id) for video_id, _ in initial_data] shield_key_name_list = shield_config.get(region, None) if shield_key_name_list is not None: initial_video_ids = filter_shield_video(video_ids=initial_video_ids, shield_key_name_list=shield_key_name_list) # 涉政视频过滤 if political_filter is True: initial_video_ids = filter_political_videos(video_ids=initial_video_ids) dup_data = {} # 视频去重逻辑 if dup_remove is True: for video_id, score in initial_data: if int(video_id) not in h_video_ids and int(video_id) in initial_video_ids: dup_data[int(video_id)] = score h_video_ids.append(int(video_id)) else: for video_id, score in initial_data: if int(video_id) in initial_video_ids: dup_data[int(video_id)] = score if len(dup_data) > 0: redis_helper.add_data_with_zset(key_name=dup_key_name, data=dup_data, expire_time=2 * 24 * 3600) # 限流视频score调整 update_limit_video_score(initial_videos=dup_data, key_name=dup_key_name) return h_video_ids def dup_to_redis(h_video_ids, now_date, now_h, rule_key, h_rule_key, region_24h_rule_key, by_24h_rule_key, by_48h_rule_key, region, data_key, rule_rank_h_flag, political_filter, shield_config, dup_remove): """将地域分组小时级数据与其他召回视频池去重，存入对应的redis""" # ##### 去重更新不区分地域小时级列表，并另存为redis中 if h_rule_key is not None: h_key_name = \ f"{config_.RECALL_KEY_NAME_PREFIX_BY_H_H}{data_key}:{h_rule_key}:" \ f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}" h_dup_key_name = \ f"{config_.RECALL_KEY_NAME_PREFIX_DUP_H_H}{region}:{data_key}:{rule_key}:" \ f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}" h_video_ids = dup_data(h_video_ids=h_video_ids, initial_key_name=h_key_name, dup_key_name=h_dup_key_name, region=region, political_filter=political_filter, shield_config=shield_config, dup_remove=dup_remove) # ##### 去重更新地域分组小时级24h列表，并另存为redis中 region_24h_key_name = \ f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_24H}{region}:{data_key}:{region_24h_rule_key}:" \ f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}" region_24h_dup_key_name = \ f"{config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H}{region}:{data_key}:{rule_key}:" \ f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}" h_video_ids = dup_data(h_video_ids=h_video_ids, initial_key_name=region_24h_key_name, dup_key_name=region_24h_dup_key_name, region=region, political_filter=political_filter, shield_config=shield_config, dup_remove=dup_remove) if rule_rank_h_flag == '48h': # ##### 去重小程序相对48h更新结果，并另存为redis中 h_48h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_48H}{data_key}:{by_48h_rule_key}:" \ f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}" h_48h_dup_key_name = \ f"{config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_48H_H}{region}:{data_key}:{rule_key}:" \ f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}" h_video_ids = dup_data(h_video_ids=h_video_ids, initial_key_name=h_48h_key_name, dup_key_name=h_48h_dup_key_name, region=region, political_filter=political_filter, shield_config=shield_config, dup_remove=dup_remove) # ##### 去重小程序相对48h 筛选后剩余数据 更新结果，并另存为redis中 if by_48h_rule_key == 'rule1': other_h_48h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_48H_OTHER}{data_key}:" \ f"{by_48h_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}" other_h_48h_dup_key_name = \ f"{config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_48H_H}{region}:{data_key}:{rule_key}:" \ f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}" h_video_ids = dup_data(h_video_ids=h_video_ids, initial_key_name=other_h_48h_key_name, dup_key_name=other_h_48h_dup_key_name, region=region, political_filter=political_filter, shield_config=shield_config, dup_remove=dup_remove) else: # ##### 去重小程序相对24h更新结果，并另存为redis中 h_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_24H}{data_key}:{by_24h_rule_key}:" \ f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}" h_24h_dup_key_name = \ f"{config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_24H_H}{region}:{data_key}:{rule_key}:" \ f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}" h_video_ids = dup_data(h_video_ids=h_video_ids, initial_key_name=h_24h_key_name, dup_key_name=h_24h_dup_key_name, region=region, political_filter=political_filter, shield_config=shield_config, dup_remove=dup_remove) # ##### 去重小程序相对24h 筛选后剩余数据 更新结果，并另存为redis中 # if by_24h_rule_key in ['rule3', 'rule4']: other_h_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_24H_OTHER}{data_key}:" \ f"{by_24h_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}" other_h_24h_dup_key_name = \ f"{config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H}{region}:{data_key}:{rule_key}:" \ f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}" h_video_ids = dup_data(h_video_ids=h_video_ids, initial_key_name=other_h_24h_key_name, dup_key_name=other_h_24h_dup_key_name, region=region, political_filter=political_filter, shield_config=shield_config, dup_remove=dup_remove) # ##### 去重小程序模型更新结果，并另存为redis中 # model_key_name = get_rov_redis_key(now_date=now_date) # model_data_dup_key_name = \ # f"{config_.RECALL_KEY_NAME_PREFIX_DUP_REGION_H}{region}:{data_key}:{rule_key}:" \ # f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}" # h_video_ids = dup_data(h_video_ids=h_video_ids, initial_key_name=model_key_name, # dup_key_name=model_data_dup_key_name, region=region) def merge_df(df_left, df_right): """ df按照videoid, code 合并，对应特征求和 :param df_left: :param df_right: :return: """ df_merged = pd.merge(df_left, df_right, on=['videoid', 'code'], how='outer', suffixes=['_x', '_y']) df_merged.fillna(0, inplace=True) feature_list = ['videoid', 'code'] for feature in features: if feature in ['apptype', 'videoid', 'code']: 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, code]合并，平台回流人数、回流人数、分数 分别求和 :param df_left: :param df_right: :return: """ df_merged = pd.merge(df_left, df_right, on=['videoid', 'code'], how='outer', suffixes=['_x', '_y']) df_merged.fillna(0, inplace=True) feature_list = ['videoid', 'code', 'lastonehour_return', 'platform_return', 'score'] for feature in feature_list[2:]: df_merged[feature] = df_merged[f'{feature}_x'] + df_merged[f'{feature}_y'] return df_merged[feature_list] def process_with_region(region, df_merged, data_key, rule_key, rule_param, now_date, now_h, rule_rank_h_flag, add_videos_with_pre_h, hour_count): log_.info(f"多协程的region = {region} 开始执行") region_df = df_merged[df_merged['code'] == region] log_.info(f'该区域region = {region}, 下有多少数据量 = {len(region_df)}') score_df = cal_score(df=region_df, param=rule_param) video_rank(df=score_df, now_date=now_date, now_h=now_h, rule_key=rule_key, param=rule_param, region=region, data_key=data_key, rule_rank_h_flag=rule_rank_h_flag, add_videos_with_pre_h=add_videos_with_pre_h, hour_count=hour_count) log_.info(f"多协程的region = {region} 完成执行") def process_with_region2(region, df_merged, data_key, rule_key, rule_param, now_date, now_h, rule_rank_h_flag, add_videos_with_pre_h, hour_count): log_.info(f"region = {region} start...") region_score_df = df_merged[df_merged['code'] == region] log_.info(f'region = {region}, region_score_df count = {len(region_score_df)}') video_rank(df=region_score_df, now_date=now_date, now_h=now_h, region=region, rule_key=rule_key, param=rule_param, data_key=data_key, rule_rank_h_flag=rule_rank_h_flag, add_videos_with_pre_h=add_videos_with_pre_h, hour_count=hour_count) log_.info(f"region = {region} end!") def process_with_app_type(app_type, params, region_code_list, feature_df, now_date, now_h, rule_rank_h_flag): log_.info(f"app_type = {app_type} start...") data_params_item = params.get('data_params') rule_params_item = params.get('rule_params') task_list = [] for param in 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}") task_list.extend( [ gevent.spawn(process_with_region, region, df_merged, app_type, data_key, rule_key, rule_param, now_date, now_h, rule_rank_h_flag) for region in region_code_list ] ) gevent.joinall(task_list) log_.info(f"app_type = {app_type} end!") # log_.info(f"app_type = {app_type}") # task_list = [] # for data_key, data_param in params['data_params'].items(): # 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) # for rule_key, rule_param in params['rule_params'].items(): # log_.info(f"rule_key = {rule_key}, rule_param = {rule_param}") # task_list.extend( # [ # gevent.spawn(process_with_region, region, df_merged, app_type, data_key, rule_key, rule_param, # now_date, now_h) # for region in region_code_list # ] # ) # gevent.joinall(task_list) def copy_data_for_city(region, city_code, data_key, rule_key, now_date, now_h, shield_config): """copy 对应数据到城市对应redis，并做相应屏蔽视频过滤""" log_.info(f"city_code = {city_code} start ...") redis_helper = RedisHelper() key_prefix_list = [ config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H, # 地域小时级 config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H, # 地域相对24h config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_24H_H, # 不区分地域相对24h config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H, # 不区分地域相对24h筛选后 config_.RECALL_KEY_NAME_PREFIX_DUP_REGION_H, # rov大列表 ] for key_prefix in key_prefix_list: region_key = f"{key_prefix}{region}:{data_key}:{rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}" city_key = f"{key_prefix}{city_code}:{data_key}:{rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}" if not redis_helper.key_exists(key_name=region_key): continue region_data = redis_helper.get_all_data_from_zset(key_name=region_key, with_scores=True) if not region_data: continue # 屏蔽视频过滤 region_video_ids = [int(video_id) for video_id, _ in region_data] shield_key_name_list = shield_config.get(city_code, None) # shield_key_name_list = config_.SHIELD_CONFIG.get(city_code, None) if shield_key_name_list is not None: filtered_video_ids = filter_shield_video(video_ids=region_video_ids, shield_key_name_list=shield_key_name_list) else: filtered_video_ids = region_video_ids city_data = {} for video_id, score in region_data: if int(video_id) in filtered_video_ids: city_data[int(video_id)] = score if len(city_data) > 0: redis_helper.add_data_with_zset(key_name=city_key, data=city_data, expire_time=2 * 24 * 3600) log_.info(f"city_code = {city_code} end!") def process_with_param(param, data_params_item, rule_params_item, region_code_list, feature_df, now_date, now_h, rule_rank_h_flag): data_key = param.get('data') data_param = data_params_item.get(data_key) rule_key = param.get('rule') rule_param = rule_params_item.get(rule_key) merge_func = rule_param.get('merge_func', None) log_.info("数据采用:{},统计采用{}.".format(data_key, rule_key)) log_.info("具体的规则是:{}.".format(rule_param)) # 是否在地域小时级数据中增加打捞的优质视频 add_videos_with_pre_h = rule_param.get('add_videos_with_pre_h', False) hour_count = rule_param.get('hour_count', 0) if merge_func == 2: score_df_list = [] 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['lastonehour_return'] task_list = [ gevent.spawn(process_with_region2, region, df_merged, data_key, rule_key, rule_param, now_date, now_h, rule_rank_h_flag, add_videos_with_pre_h, hour_count) for region in region_code_list ] else: df_list = [feature_df[feature_df['apptype'] == apptype] for apptype in data_param] df_merged = reduce(merge_df, df_list) task_list = [ gevent.spawn(process_with_region, region, df_merged, data_key, rule_key, rule_param, now_date, now_h, rule_rank_h_flag, add_videos_with_pre_h, hour_count) for region in region_code_list ] gevent.joinall(task_list) # 特殊城市视频数据准备 # 屏蔽视频过滤 # shield_config = rule_param.get('shield_config', config_.SHIELD_CONFIG) # for region, city_list in config_.REGION_CITY_MAPPING.items(): # t = [ # gevent.spawn( # copy_data_for_city, # region, city_code, data_key, rule_key, now_date, now_h, shield_config # ) # for city_code in city_list # ] # gevent.joinall(t) log_.info(f"多进程的 param = {param} 完成执行!") def rank_by_h(project, table, now_date, now_h, rule_params, region_code_list, rule_rank_h_flag): # 获取特征数据 feature_df = get_feature_data(project=project, table=table, now_date=now_date) feature_df['apptype'] = feature_df['apptype'].astype(int) data_params_item = rule_params.get('data_params') rule_params_item = rule_params.get('rule_params') params_list = rule_params.get('params_list') pool = multiprocessing.Pool(processes=len(params_list)) for param in params_list: pool.apply_async( func=process_with_param, args=(param, data_params_item, rule_params_item, region_code_list, feature_df, now_date, now_h, rule_rank_h_flag) ) pool.close() pool.join() def h_bottom_process(param, rule_params_item, region_code_list, key_prefix, redis_dt, redis_h, now_date, now_h, rule_rank_h_flag): redis_helper = RedisHelper() data_key = param.get('data') rule_key = param.get('rule') rule_param = rule_params_item.get(rule_key) log_.info(f"data_key = {data_key}, rule_key = {rule_key}, rule_param = {rule_param}") h_rule_key = rule_param.get('h_rule_key', None) region_24h_rule_key = rule_param.get('region_24h_rule_key', 'rule1') by_24h_rule_key = rule_param.get('24h_rule_key', None) by_48h_rule_key = rule_param.get('48h_rule_key', None) # 涉政视频过滤 political_filter = param.get('political_filter', None) # 屏蔽视频过滤 shield_config = param.get('shield_config', config_.SHIELD_CONFIG) dup_remove = param.get('dup_remove', True) for region in region_code_list: log_.info(f"region = {region}") key_name = f"{key_prefix}{region}:{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() h_video_ids = [] for video_id, score in initial_data: final_data[video_id] = score h_video_ids.append(int(video_id)) # 存入对应的redis final_key_name = \ f"{key_prefix}{region}:{data_key}:{rule_key}:{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=2 * 24 * 3600) # 与其他召回视频池去重，存入对应的redis dup_to_redis(h_video_ids=h_video_ids, now_date=now_date, now_h=now_h, rule_key=rule_key, h_rule_key=h_rule_key, region_24h_rule_key=region_24h_rule_key, region=region, data_key=data_key, by_24h_rule_key=by_24h_rule_key, by_48h_rule_key=by_48h_rule_key, rule_rank_h_flag=rule_rank_h_flag, political_filter=political_filter, shield_config=shield_config, dup_remove=dup_remove) # 特殊城市视频数据准备 for region, city_list in config_.REGION_CITY_MAPPING.items(): t = [ gevent.spawn( copy_data_for_city, region, city_code, data_key, rule_key, now_date, now_h, shield_config ) for city_code in city_list ] gevent.joinall(t) def h_rank_bottom(now_date, now_h, rule_params, region_code_list, rule_rank_h_flag): """未按时更新数据，用上一小时结果作为当前小时的数据""" # 获取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_prefix = config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H rule_params_item = rule_params.get('rule_params') params_list = rule_params.get('params_list') pool = multiprocessing.Pool(processes=len(params_list)) for param in params_list: pool.apply_async( func=h_bottom_process, args=(param, rule_params_item, region_code_list, key_prefix, redis_dt, redis_h, now_date, now_h, rule_rank_h_flag) ) pool.close() pool.join() def h_timer_check(): try: rule_rank_h_flag = "24h" rule_params = RULE_PARAMS project = config_.PROJECT_REGION_APP_TYPE table = config_.TABLE_REGION_APP_TYPE region_code_list = [code for region, code in region_code.items()] now_date = datetime.datetime.today() log_.info(f"开始执行: {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: log_.info("当前时间{}小时，使用bottom的data，开始。".format(now_h)) h_rank_bottom(now_date=now_date, now_h=now_h, rule_params=rule_params, region_code_list=region_code_list, rule_rank_h_flag=rule_rank_h_flag) log_.info("----------当前时间{}小时，使用bottom的data，完成----------".format(now_h)) return # 查看当前小时更新的数据是否已准备好 h_data_count = h_data_check(project=project, table=table, now_date=now_date) if h_data_count > 0: log_.info('上游数据表查询数据条数 h_data_count = {}，开始计算。'.format(h_data_count)) # 数据准备好，进行更新 rank_by_h(now_date=now_date, now_h=now_h, rule_params=rule_params, project=project, table=table, region_code_list=region_code_list, rule_rank_h_flag=rule_rank_h_flag) log_.info("数据1----------正常完成----------") elif now_min > 40: log_.info('当前分钟超过40，预计执行无法完成，使用 bottom data!') h_rank_bottom(now_date=now_date, now_h=now_h, rule_params=rule_params, region_code_list=region_code_list, rule_rank_h_flag=rule_rank_h_flag) log_.info('----------当前分钟超过40，使用bottom的data，完成----------') else: # 数据没准备好，1分钟后重新检查 log_.info("上游数据未就绪，等待...") Timer(60, h_timer_check).start() # 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"now_date: {datetime.datetime.strftime(now_date, '%Y%m%d')}\n" # f"now_h: {now_h}\n" # f"finished time: {datetime.datetime.strftime(datetime.datetime.now(), '%Y%m%d %H:%M:%S')}" # ) 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__': log_.info("文件alg_recsys_recall_1h_region.py:「1小时地域」 开始执行") h_timer_check()