user_group_update.py 3.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081
  1. import datetime
  2. import traceback
  3. from threading import Timer
  4. from utils import RedisHelper, data_check, get_feature_data, send_msg_to_feishu
  5. from config import set_config
  6. from log import Log
  7. config_, _ = set_config()
  8. log_ = Log()
  9. redis_helper = RedisHelper()
  10. features = [
  11. 'apptype',
  12. 'return1mids',
  13. 'return2_3mids',
  14. 'return4_8mids',
  15. 'return9_24mids',
  16. 'return25_nmids',
  17. 'return0share1mids',
  18. 'return0share2_nmids'
  19. ]
  20. def update_user_group_to_redis(project, table, dt, app_type):
  21. """更新mid对应分组到redis中"""
  22. # 获取用户分组数据
  23. feature_df = get_feature_data(project=project, table=table, features=features, dt=dt)
  24. feature_df['apptype'] = feature_df['apptype'].astype(int)
  25. feature_df = feature_df[feature_df['apptype'] == app_type]
  26. group_list = features[1:]
  27. for group in group_list:
  28. log_.info(f"group = {group} update redis start ...")
  29. mid_list = feature_df[group].tolist()
  30. mid_list = [mid for mid in mid_list if mid is not None]
  31. log_.info(f"mid count = {len(mid_list)}")
  32. for mid in mid_list:
  33. key_name = f"{config_.KEY_NAME_PREFIX_MID_GROUP}{mid}"
  34. redis_helper.set_data_to_redis(key_name=key_name, value=group, expire_time=25 * 3600)
  35. log_.info(f"group = {group}, mid count = {len(mid_list)}, update redis finished!")
  36. def timer_check():
  37. try:
  38. app_type = config_.APP_TYPE['VLOG']
  39. project = config_.ad_model_data['user_group'].get('project')
  40. table = config_.ad_model_data['user_group'].get('table')
  41. now_date = datetime.datetime.today() - datetime.timedelta(days=2)
  42. dt = datetime.datetime.strftime(now_date, '%Y%m%d')
  43. log_.info(f"now_date: {dt}")
  44. now_min = datetime.datetime.now().minute
  45. # 查看当前更新的数据是否已准备好
  46. data_count = data_check(project=project, table=table, dt=dt)
  47. if data_count > 0:
  48. log_.info(f"user group data count = {data_count}")
  49. # 数据准备好,进行更新
  50. update_user_group_to_redis(project=project, table=table, dt=dt, app_type=app_type)
  51. log_.info(f"user group data update end!")
  52. elif now_min > 45:
  53. log_.info('user group data is None!')
  54. send_msg_to_feishu(
  55. webhook=config_.FEISHU_ROBOT['server_robot'].get('webhook'),
  56. key_word=config_.FEISHU_ROBOT['server_robot'].get('key_word'),
  57. msg_text=f"rov-offline{config_.ENV_TEXT} - 用户分组数据未准备好!\n"
  58. f"traceback: {traceback.format_exc()}"
  59. )
  60. else:
  61. # 数据没准备好,1分钟后重新检查
  62. Timer(60, timer_check).start()
  63. except Exception as e:
  64. log_.error(f"用户分组数据更新失败, exception: {e}, traceback: {traceback.format_exc()}")
  65. send_msg_to_feishu(
  66. webhook=config_.FEISHU_ROBOT['server_robot'].get('webhook'),
  67. key_word=config_.FEISHU_ROBOT['server_robot'].get('key_word'),
  68. msg_text=f"rov-offline{config_.ENV_TEXT} - 用户分组数据更新失败\n"
  69. f"exception: {e}\n"
  70. f"traceback: {traceback.format_exc()}"
  71. )
  72. if __name__ == '__main__':
  73. timer_check()