|
|
@@ -79,6 +79,11 @@ class PoolRecall(object):
|
|
|
h_recall_mid_key = f"{config_.H_WITH_MID_RECALL_KEY_NAME_PREFIX_REGION_24H}{self.app_type}.{self.mid}"
|
|
|
# 判断mid对应小时级视频列表 时间记录
|
|
|
h_record_key = f"{config_.H_WITH_MID_RECORD_KEY_NAME_PREFIX_REGION_24H}{self.app_type}.{self.mid}"
|
|
|
+ elif key_flag == '24h':
|
|
|
+ # mid对应小时级视频列表 redis-key
|
|
|
+ h_recall_mid_key = f"{config_.H_WITH_MID_RECALL_KEY_NAME_PREFIX_24H}{self.app_type}.{self.mid}"
|
|
|
+ # 判断mid对应小时级视频列表 时间记录
|
|
|
+ h_record_key = f"{config_.H_WITH_MID_RECORD_KEY_NAME_PREFIX_24H}{self.app_type}.{self.mid}"
|
|
|
else:
|
|
|
# mid对应小时级视频列表 redis-key
|
|
|
h_recall_mid_key = f"{config_.H_WITH_MID_RECALL_KEY_NAME_PREFIX}{self.app_type}.{self.mid}"
|
|
|
@@ -96,6 +101,8 @@ class PoolRecall(object):
|
|
|
key_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H}{province_code}."
|
|
|
elif self.ab_code in [code for _, code in config_.AB_CODE['rank_by_24h'].items()]:
|
|
|
key_prefix = config_.RECALL_KEY_NAME_PREFIX_BY_24H
|
|
|
+ elif key_flag == '24h':
|
|
|
+ key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP_24H_H
|
|
|
else:
|
|
|
key_prefix = config_.RECALL_KEY_NAME_PREFIX_BY_H
|
|
|
|
|
|
@@ -149,6 +156,10 @@ class PoolRecall(object):
|
|
|
province_code = self.client_info.get('provinceCode', '-1')
|
|
|
if province_code == '':
|
|
|
province_code = '-1'
|
|
|
+ if self.ab_code in [code for _, code in config_.AB_CODE['rank_by_24h'].items()]:
|
|
|
+ push_from = config_.PUSH_FROM['rov_recall_24h']
|
|
|
+ else:
|
|
|
+ push_from = config_.PUSH_FROM['rov_recall_h']
|
|
|
# 获取mid对应的小时级列表redis-key
|
|
|
h_recall_mid_key = self.get_mid_h_key(province_code=province_code)
|
|
|
if not self.redis_helper.key_exists(h_recall_mid_key):
|
|
|
@@ -188,7 +199,7 @@ class PoolRecall(object):
|
|
|
if filtered_result:
|
|
|
# 添加视频源参数 pushFrom, abCode
|
|
|
temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)],
|
|
|
- 'pushFrom': config_.PUSH_FROM['rov_recall_h'], 'abCode': self.ab_code}
|
|
|
+ 'pushFrom': push_from, 'abCode': self.ab_code}
|
|
|
for item in filtered_result if video_score.get(int(item)) is not None]
|
|
|
recall_result.extend(temp_result)
|
|
|
fil_video_ids.extend(list(set(video_ids) - set([item.get('videoId') for item in temp_result])))
|
|
|
@@ -1210,3 +1221,106 @@ class PoolRecall(object):
|
|
|
idx += get_size
|
|
|
|
|
|
return recall_result[:size]
|
|
|
+
|
|
|
+ def rule_recall_by_h(self, size=4, expire_time=24*3600):
|
|
|
+ """
|
|
|
+ 小时级召回视频
|
|
|
+ :param size: 获取视频个数
|
|
|
+ :param expire_time: 末位视频记录redis过期时间
|
|
|
+ :return:
|
|
|
+ """
|
|
|
+ t = [gevent.spawn(self.rov_pool_recall_h, size),
|
|
|
+ gevent.spawn(self.rov_pool_recall_h, size, '24h')]
|
|
|
+ gevent.joinall(t)
|
|
|
+ h_recall_result_list = [i.get() for i in t]
|
|
|
+ # 将已获取到的视频按顺序去重合并
|
|
|
+ now_video_ids = []
|
|
|
+ recall_result = []
|
|
|
+ for h_result in h_recall_result_list:
|
|
|
+ for video in h_result:
|
|
|
+ video_id = video.get('videoId')
|
|
|
+ if video_id not in now_video_ids:
|
|
|
+ recall_result.append(video)
|
|
|
+ now_video_ids.append(video_id)
|
|
|
+ if len(recall_result) >= size:
|
|
|
+ break
|
|
|
+ else:
|
|
|
+ continue
|
|
|
+ # 判断获取到的小时级数据数量
|
|
|
+ if len(recall_result) < size:
|
|
|
+ # 补充数据
|
|
|
+ rov_recall_result = self.rov_pool_recall(size=size, expire_time=expire_time)
|
|
|
+ # 去重合并
|
|
|
+ for video in rov_recall_result:
|
|
|
+ vid = video.get('videoId')
|
|
|
+ if vid not in now_video_ids:
|
|
|
+ recall_result.append(video)
|
|
|
+ now_video_ids.append(vid)
|
|
|
+ if len(recall_result) >= size:
|
|
|
+ break
|
|
|
+ else:
|
|
|
+ continue
|
|
|
+
|
|
|
+ return recall_result[:size]
|
|
|
+
|
|
|
+ def rov_pool_recall_h(self, size=4, key_flag=''):
|
|
|
+ """
|
|
|
+ 小时级视频召回
|
|
|
+ :param size: 视频数
|
|
|
+ :param key_flag:
|
|
|
+ :return:
|
|
|
+ """
|
|
|
+ if key_flag == '24h':
|
|
|
+ push_from = config_.PUSH_FROM['rov_recall_24h']
|
|
|
+ else:
|
|
|
+ push_from = config_.PUSH_FROM['rov_recall_h']
|
|
|
+ # 获取mid对应的小时级列表redis-key
|
|
|
+ h_recall_mid_key = self.get_mid_h_key(province_code='', key_flag=key_flag)
|
|
|
+ if not self.redis_helper.key_exists(h_recall_mid_key):
|
|
|
+ recall_result = []
|
|
|
+ else:
|
|
|
+ # 过滤的视频
|
|
|
+ fil_video_ids = []
|
|
|
+ recall_result = []
|
|
|
+ # 每次获取的视频数
|
|
|
+ get_size = size * 5
|
|
|
+ # 记录获取频次
|
|
|
+ freq = 0
|
|
|
+ while len(recall_result) < size:
|
|
|
+ freq += 1
|
|
|
+ if freq > config_.MAX_FREQ_FROM_ROV_POOL:
|
|
|
+ break
|
|
|
+ # 获取数据
|
|
|
+ data = self.redis_helper.get_data_zset_with_index(key_name=h_recall_mid_key,
|
|
|
+ start=(freq - 1) * get_size, end=freq * get_size - 1,
|
|
|
+ with_scores=True)
|
|
|
+ if not data:
|
|
|
+ log_.info('小时级更新视频已取完')
|
|
|
+ break
|
|
|
+ # 获取视频id,并转换类型为int,并存储为key-value{videoId: score}
|
|
|
+ video_ids = []
|
|
|
+ video_score = {}
|
|
|
+ for value in data:
|
|
|
+ video_id = int(value[0])
|
|
|
+ video_ids.append(video_id)
|
|
|
+ video_score[video_id] = value[1]
|
|
|
+ # 过滤
|
|
|
+ filter_ = FilterVideos(app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
|
|
|
+ ge = gevent.spawn(filter_.filter_videos_h, self.rule_key, self.ab_code, '', key_flag)
|
|
|
+ ge.join()
|
|
|
+ filtered_result = ge.get()
|
|
|
+
|
|
|
+ if filtered_result:
|
|
|
+ # 添加视频源参数 pushFrom, abCode
|
|
|
+ temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)],
|
|
|
+ 'pushFrom': push_from, 'abCode': self.ab_code}
|
|
|
+ for item in filtered_result if video_score.get(int(item)) is not None]
|
|
|
+ recall_result.extend(temp_result)
|
|
|
+ fil_video_ids.extend(list(set(video_ids) - set([item.get('videoId') for item in temp_result])))
|
|
|
+ else:
|
|
|
+ fil_video_ids.extend(video_ids)
|
|
|
+ # 将被过滤的视频进行移除
|
|
|
+ for value in fil_video_ids:
|
|
|
+ self.redis_helper.remove_value_from_zset(key_name=h_recall_mid_key, value=value)
|
|
|
+
|
|
|
+ return recall_result[:size]
|
