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',
'group',
'sharerate_all',
'sharerate_ad'
]
def predict_user_group_share_rate(user_group_initial_df, dt, data_params, rule_params, param):
"""预估用户组对应的有广告时分享率"""
# 获取对应的参数
data_key = param.get('data')
data_param = data_params.get(data_key)
rule_key = param.get('rule')
rule_param = rule_params.get(rule_key)
# 获取对应的用户组特征
user_group_df = user_group_initial_df.copy()
user_group_df['apptype'] = user_group_df['apptype'].astype(int)
user_group_df = user_group_df[user_group_df['apptype'] == data_param]
user_group_df['sharerate_all'].fillna(0, inplace=True)
user_group_df['sharerate_ad'].fillna(0, inplace=True)
user_group_df['sharerate_all'] = user_group_df['sharerate_all'].astype(float)
user_group_df['sharerate_ad'] = user_group_df['sharerate_ad'].astype(float)
# 获取有广告时所有用户组近30天的分享率
ad_all_group_share_rate = user_group_df[user_group_df['group'] == 'allmids']['sharerate_ad'].values[0]
# 获取对应的用户分组数据
user_group_list = rule_param.get('group_list')
user_group_df = user_group_df[user_group_df['group'].isin(user_group_list)]
# 去除对应无广告用户组
if rule_param.get('remove_no_ad_group') is True:
user_group_df = user_group_df[~user_group_df['group'].isin(rule_param.get('no_ad_mid_group_list'))]
# 计算用户组有广告时分享率
user_group_df['group_ad_share_rate'] = \
user_group_df['sharerate_ad'] * float(ad_all_group_share_rate) / user_group_df['sharerate_all']
user_group_df['group_ad_share_rate'].fillna(0, inplace=True)
# 结果写入redis
key_name = f"{config_.KEY_NAME_PREFIX_AD_GROUP}{data_key}:{rule_key}:{dt}"
redis_data = {}
for index, item in user_group_df.iterrows():
redis_data[item['group']] = item['group_ad_share_rate']
group_ad_share_rate_mean = user_group_df['group_ad_share_rate'].mean()
redis_data['mean_group'] = 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 user_group_df
def update_users_data(project, table, dt, update_params):
"""预估用户组有广告时分享率"""
# 获取用户组特征
user_group_initial_df = get_feature_data(project=project, table=table, features=features, dt=dt)
data_params = update_params.get('data_params')
rule_params = update_params.get('rule_params')
for param in update_params.get('params_list'):
log_.info(f"param = {param} update start...")
predict_user_group_share_rate(user_group_initial_df=user_group_initial_df,
dt=dt,
data_params=data_params,
rule_params=rule_params,
param=param)
log_.info(f"param = {param} update end!")
# for data_key, data_param in update_params.items():
# log_.info(f"data_key = {data_key} update start...")
# predict_user_group_share_rate(user_group_initial_df=user_group_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_USER_PARAMS
project = config_.ad_model_data['users_share_rate'].get('project')
table = config_.ad_model_data['users_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 user group data count = {data_count}")
# 数据准备好,进行更新
update_users_data(project=project, table=table, dt=dt, update_params=update_params)
log_.info(f"ad user group data update end!")
# elif now_min > 45:
# log_.info('ad user group 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()