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() mid_group_list = [ 'apptype', 'return1mids', 'return2_3mids', 'return4_8mids', 'return9_24mids', 'return25_nmids', 'return0share1mids', 'return0share2_nmids' ] def update_user_group_to_redis(project, table, dt, app_type): """更新mid对应分组到redis中""" # 获取用户分组数据 feature_df = get_feature_data(project=project, table=table, features=mid_group_list, dt=dt) feature_df['apptype'] = feature_df['apptype'].astype(int) feature_df = feature_df[feature_df['apptype'] == app_type] for group in mid_group_list: log_.info(f"group = {group} update redis start ...") mid_list = feature_df[group].tolist() for mid in mid_list: key_name = f"{config_.KEY_NAME_PREFIX_MID_GROUP}{mid}" redis_helper.set_data_to_redis(key_name=key_name, value=group, expire_time=25 * 3600) log_.info(f"group = {group}, mid count = {len(mid_list)}, update redis finished!") def timer_check(): try: app_type = config_.APP_TYPE['VLOG'] project = config_.ad_model_data['user_group'].get('project') table = config_.ad_model_data['user_group'].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"user group data count = {data_count}") # 数据准备好,进行更新 update_user_group_to_redis(project=project, table=table, dt=dt, app_type=app_type) log_.info(f"user group data update end!") elif now_min > 45: log_.info('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()