# -*- coding: utf-8 -*- # @ModuleName: region_rule_rank_h # @Author: Liqian # @Time: 2022/5/5 15:54 # @Software: PyCharm from gevent import monkey monkey.patch_all() import gevent import datetime import pandas as pd import math from functools import reduce from odps import ODPS from threading import Timer from utils import MysqlHelper, RedisHelper, get_data_from_odps, filter_video_status, filter_shield_video, check_table_partition_exits from 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 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', # 地域分组 ] 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}' 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_feature_data(project, table, now_date): """获取特征数据""" dt = datetime.datetime.strftime(now_date, '%Y%m%d%H') # 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, 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['score'] = df['share_rate'] * df['back_rate'] * df['log_back'] * df['K2'] df['platform_return_rate'] = df['platform_return'] / df['lastonehour_return'] df = df.sort_values(by=['score'], ascending=False) return df def video_rank(df, now_date, now_h, rule_key, param, region, app_type, data_key): """ 获取符合进入召回源条件的视频，与每日更新的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)] # 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'h_recall videos count = {len(h_recall_videos)}') # 视频状态过滤 filtered_videos = filter_video_status(h_recall_videos) log_.info('filtered_videos count = {}'.format(len(filtered_videos))) # 屏蔽视频过滤 shield_key_name_list = config_.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) log_.info(f"shield filtered_videos count = {len(filtered_videos)}") # 写入对应的redis h_video_ids = [] h_recall_result = {} for video_id in filtered_videos: score = h_recall_df[h_recall_df['videoid'] == video_id]['score'] # print(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}.{app_type}.{data_key}.{rule_key}." \ f"{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}" if len(h_recall_result) > 0: redis_helper.add_data_with_zset(key_name=h_recall_key_name, data=h_recall_result, expire_time=23 * 3600) # 限流视频score调整 update_limit_video_score(initial_videos=h_recall_result, key_name=h_recall_key_name) # 清空线上过滤应用列表 redis_helper.del_keys(key_name=f"{config_.REGION_H_VIDEO_FILER}{region}.{app_type}.{data_key}.{rule_key}") region_24h_rule_key = param.get('region_24h_rule_key', 'rule1') by_24h_rule_key = param.get('24h_rule_key', None) # 与其他召回视频池去重，存入对应的redis dup_to_redis(h_video_ids=h_video_ids, now_date=now_date, now_h=now_h, rule_key=rule_key, region_24h_rule_key=region_24h_rule_key, by_24h_rule_key=by_24h_rule_key, region=region, app_type=app_type, data_key=data_key) def dup_to_redis(h_video_ids, now_date, now_h, rule_key, region_24h_rule_key, by_24h_rule_key, region, app_type, data_key): """将地域分组小时级数据与其他召回视频池去重，存入对应的redis""" redis_helper = RedisHelper() # # ##### 去重更新地域分组天级列表，并另存为redis中 # region_day_key_name = \ # f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_DAY}{region}.rule1." \ # f"{datetime.datetime.strftime(now_date, '%Y%m%d')}" # if redis_helper.key_exists(key_name=region_day_key_name): # region_day_data = redis_helper.get_data_zset_with_index( # key_name=region_day_key_name, start=0, end=-1, with_scores=True) # log_.info(f'region day data count = {len(region_day_data)}') # region_day_dup = {} # for video_id, score in region_day_data: # if int(video_id) not in h_video_ids: # region_day_dup[int(video_id)] = score # h_video_ids.append(int(video_id)) # log_.info(f"region day data dup count = {len(region_day_dup)}") # region_day_dup_key_name = \ # f"{config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_DAY_H}{region}.{rule_key}." \ # f"{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}" # if len(region_day_dup) > 0: # redis_helper.add_data_with_zset(key_name=region_day_dup_key_name, data=region_day_dup, expire_time=23 * 3600) # ##### 去重更新地域分组小时级24h列表，并另存为redis中 region_24h_key_name = \ f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_24H}{region}.{app_type}.{data_key}.{region_24h_rule_key}." \ f"{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}" if redis_helper.key_exists(key_name=region_24h_key_name): region_24h_data = redis_helper.get_all_data_from_zset(key_name=region_24h_key_name, with_scores=True) log_.info(f'region 24h data count = {len(region_24h_data)}') # 屏蔽视频过滤 region_24h_video_ids = [int(video_id) for video_id, _ in region_24h_data] shield_key_name_list = config_.SHIELD_CONFIG.get(region, None) if shield_key_name_list is not None: region_24h_video_ids = filter_shield_video(video_ids=region_24h_video_ids, shield_key_name_list=shield_key_name_list) log_.info(f"shield filtered_videos count = {len(region_24h_video_ids)}") region_24h_dup = {} for video_id, score in region_24h_data: if int(video_id) not in h_video_ids and int(video_id) in region_24h_video_ids: region_24h_dup[int(video_id)] = score h_video_ids.append(int(video_id)) log_.info(f"region 24h data dup count = {len(region_24h_dup)}") region_24h_dup_key_name = \ f"{config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H}{region}.{app_type}.{data_key}.{rule_key}." \ f"{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}" if len(region_24h_dup) > 0: redis_helper.add_data_with_zset(key_name=region_24h_dup_key_name, data=region_24h_dup, expire_time=23 * 3600) # 限流视频score调整 update_limit_video_score(initial_videos=region_24h_dup, key_name=region_24h_dup_key_name) # 清空线上过滤应用列表 # redis_helper.del_keys(key_name=f"{config_.REGION_H_VIDEO_FILER_24H}{app_type}.{data_key}.{region}.{rule_key}") # ##### 去重小程序天级更新结果，并另存为redis中 # day_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_DAY}rule2.{datetime.datetime.strftime(now_date, '%Y%m%d')}" # if redis_helper.key_exists(key_name=day_key_name): # day_data = redis_helper.get_data_zset_with_index( # key_name=day_key_name, start=0, end=-1, with_scores=True) # log_.info(f'day data count = {len(day_data)}') # day_dup = {} # for video_id, score in day_data: # if int(video_id) not in h_video_ids: # day_dup[int(video_id)] = score # h_video_ids.append(int(video_id)) # log_.info(f"day data dup count = {len(day_dup)}") # day_dup_key_name = \ # f"{config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_DAY_H}{region}.{rule_key}." \ # f"{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}" # if len(day_dup) > 0: # redis_helper.add_data_with_zset(key_name=day_dup_key_name, data=day_dup, expire_time=23 * 3600) # ##### 去重小程序相对24h更新结果，并另存为redis中 day_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_24H}{app_type}.{data_key}.{by_24h_rule_key}." \ f"{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}" if redis_helper.key_exists(key_name=day_key_name): day_data = redis_helper.get_all_data_from_zset(key_name=day_key_name, with_scores=True) log_.info(f'24h data count = {len(day_data)}') # 屏蔽视频过滤 day_video_ids = [int(video_id) for video_id, _ in day_data] shield_key_name_list = config_.SHIELD_CONFIG.get(region, None) if shield_key_name_list is not None: day_video_ids = filter_shield_video(video_ids=day_video_ids, shield_key_name_list=shield_key_name_list) log_.info(f"shield filtered_videos count = {len(day_video_ids)}") day_dup = {} for video_id, score in day_data: if int(video_id) not in h_video_ids and int(video_id) in day_video_ids: day_dup[int(video_id)] = score h_video_ids.append(int(video_id)) log_.info(f"24h data dup count = {len(day_dup)}") day_dup_key_name = \ f"{config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_24H_H}{region}.{app_type}.{data_key}.{rule_key}." \ f"{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}" if len(day_dup) > 0: redis_helper.add_data_with_zset(key_name=day_dup_key_name, data=day_dup, expire_time=23 * 3600) # 限流视频score调整 update_limit_video_score(initial_videos=day_dup, key_name=day_dup_key_name) # 清空线上过滤应用列表 redis_helper.del_keys(key_name=f"{config_.H_VIDEO_FILER_24H}{region}.{app_type}.{data_key}.{rule_key}") # ##### 去重小程序相对24h 筛选后剩余数据 更新结果，并另存为redis中 if by_24h_rule_key == 'rule3': other_h_24h_recall_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_BY_24H_OTHER}{app_type}.{data_key}." \ f"{by_24h_rule_key}.{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}" if redis_helper.key_exists(key_name=other_h_24h_recall_key_name): other_24h_data = redis_helper.get_all_data_from_zset(key_name=other_h_24h_recall_key_name, with_scores=True) log_.info(f'24h other data count = {len(other_24h_data)}') # 屏蔽视频过滤 other_24h_video_ids = [int(video_id) for video_id, _ in other_24h_data] shield_key_name_list = config_.SHIELD_CONFIG.get(region, None) if shield_key_name_list is not None: other_24h_video_ids = filter_shield_video(video_ids=other_24h_video_ids, shield_key_name_list=shield_key_name_list) log_.info(f"shield filtered_videos count = {len(other_24h_video_ids)}") other_24h_dup = {} for video_id, score in other_24h_data: if int(video_id) not in h_video_ids and int(video_id) in other_24h_video_ids: other_24h_dup[int(video_id)] = score h_video_ids.append(int(video_id)) log_.info(f"other 24h data dup count = {len(other_24h_dup)}") other_24h_dup_key_name = \ f"{config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H}{region}.{app_type}.{data_key}.{rule_key}." \ f"{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}" if len(other_24h_dup) > 0: redis_helper.add_data_with_zset(key_name=other_24h_dup_key_name, data=other_24h_dup, expire_time=23 * 3600) # 限流视频score调整 update_limit_video_score(initial_videos=other_24h_dup, key_name=other_24h_dup_key_name) # ##### 去重小程序模型更新结果，并另存为redis中 model_key_name = get_rov_redis_key(now_date=now_date) model_data = redis_helper.get_all_data_from_zset(key_name=model_key_name, with_scores=True) log_.info(f'model data count = {len(model_data)}') # 屏蔽视频过滤 model_video_ids = [int(video_id) for video_id, _ in model_data] shield_key_name_list = config_.SHIELD_CONFIG.get(region, None) if shield_key_name_list is not None: model_video_ids = filter_shield_video(video_ids=model_video_ids, shield_key_name_list=shield_key_name_list) log_.info(f"shield filtered_videos count = {len(model_video_ids)}") model_data_dup = {} for video_id, score in model_data: if int(video_id) not in h_video_ids and int(video_id) in model_video_ids: model_data_dup[int(video_id)] = score h_video_ids.append(int(video_id)) log_.info(f"model data dup count = {len(model_data_dup)}") model_data_dup_key_name = \ f"{config_.RECALL_KEY_NAME_PREFIX_DUP_REGION_H}{region}.{app_type}.{data_key}.{rule_key}." \ f"{datetime.datetime.strftime(now_date, '%Y%m%d')}.{now_h}" if len(model_data_dup) > 0: redis_helper.add_data_with_zset(key_name=model_data_dup_key_name, data=model_data_dup, expire_time=23 * 3600) # 限流视频score调整 update_limit_video_score(initial_videos=model_data_dup, key_name=model_data_dup_key_name) 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 process_with_region(region, df_merged, app_type, data_key, rule_key, rule_param, now_date, now_h): log_.info(f"region = {region}") # 计算score region_df = df_merged[df_merged['code'] == region] log_.info(f'region_df count = {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, app_type=app_type, data_key=data_key) def process_with_app_type(app_type, params, region_code_list, feature_df, now_date, now_h): log_.info(f"app_type = {app_type}") 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) for region in region_code_list ] ) gevent.joinall(task_list) # # 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 rank_by_h(project, table, now_date, now_h, rule_params, region_code_list): # 获取特征数据 feature_df = get_feature_data(project=project, table=table, now_date=now_date) feature_df['apptype'] = feature_df['apptype'].astype(int) t = [ gevent.spawn(process_with_app_type, app_type, params, region_code_list, feature_df, now_date, now_h) for app_type, params in rule_params.items() ] gevent.joinall(t) # for app_type, params in rule_params.items(): # log_.info(f"app_type = {app_type}") # 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 = [ # 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) # 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, param=value) # video_rank(df=score_df, now_date=now_date, now_h=now_h, rule_key=key, param=value, region=region) # # to-csv # score_filename = f"score_{region}_{key}_{datetime.datetime.strftime(now_date, '%Y%m%d%H')}.csv" # score_df.to_csv(f'./data/{score_filename}') # # to-logs # log_.info({"date": datetime.datetime.strftime(now_date, '%Y%m%d%H'), # "region_code": region, # "redis_key_prefix": config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H, # "rule_key": key, # # "score_df": score_df[['videoid', 'score']] # } # ) def h_rank_bottom(now_date, now_h, rule_params, region_code_list): """未按时更新数据，用上一小时结果作为当前小时的数据""" # 获取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 for app_type, params in rule_params.items(): log_.info(f"app_type = {app_type}") rule_params_item = params.get('rule_params') for param in params.get('params_list'): 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}") region_24h_rule_key = rule_param.get('region_24h_rule_key', 'rule1') for region in region_code_list: log_.info(f"region = {region}") key_name = f"{key_prefix}{region}.{app_type}.{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}.{app_type}.{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=23 * 3600) # 清空线上过滤应用列表 redis_helper.del_keys( key_name=f"{config_.REGION_H_VIDEO_FILER}{region}.{app_type}.{data_key}.{rule_key}") # 与其他召回视频池去重，存入对应的redis dup_to_redis(h_video_ids=h_video_ids, now_date=now_date, now_h=now_h, rule_key=rule_key, region_24h_rule_key=region_24h_rule_key, region=region, app_type=app_type, data_key=data_key) # for data_key, data_param in params['data_params'].items(): # log_.info(f"data_key = {data_key}, data_param = {data_param}") # for rule_key, rule_param in params['rule_params'].items(): # log_.info(f"rule_key = {rule_key}, rule_param = {rule_param}") # region_24h_rule_key = rule_param.get('region_24h_rule_key', 'rule1') # for region in region_code_list: # log_.info(f"region = {region}") # key_name = f"{key_prefix}{region}.{app_type}.{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}.{app_type}.{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=23 * 3600) # # 清空线上过滤应用列表 # redis_helper.del_keys(key_name=f"{config_.REGION_H_VIDEO_FILER}{region}.{app_type}.{data_key}.{rule_key}") # # 与其他召回视频池去重，存入对应的redis # dup_to_redis(h_video_ids=h_video_ids, now_date=now_date, now_h=now_h, rule_key=rule_key, # region_24h_rule_key=region_24h_rule_key, region=region, # app_type=app_type, data_key=data_key) def h_timer_check(): rule_params = config_.RULE_PARAMS_REGION_APP_TYPE 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"now_date: {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: h_rank_bottom(now_date=now_date, now_h=now_h, rule_params=rule_params, region_code_list=region_code_list) return # 查看当前小时更新的数据是否已准备好 h_data_count = h_data_check(project=project, table=table, now_date=now_date) if h_data_count > 0: log_.info(f'region_h_data_count = {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) elif now_min > 50: log_.info('h_recall data is None, use bottom data!') h_rank_bottom(now_date=now_date, now_h=now_h, rule_params=rule_params, region_code_list=region_code_list) else: # 数据没准备好，1分钟后重新检查 Timer(60, h_timer_check).start() if __name__ == '__main__': h_timer_check()