|
|
@@ -1182,6 +1182,15 @@ class PoolRecall(object):
|
|
|
else:
|
|
|
t = [gevent.spawn(self.rov_pool_recall_with_region_by_h, province_code, size, 'region_24h'),
|
|
|
gevent.spawn(self.region_dup_recall, province_code, 2, size, expire_time)]
|
|
|
+ elif self.ab_code == config_.AB_CODE['region_rank_by_h'].get('region_rule_rank5_appType_0_data1'):
|
|
|
+ if province_code == '-1':
|
|
|
+ t = [gevent.spawn(self.rov_pool_recall_with_region_by_h, province_code, size, 'day_24h'),
|
|
|
+ gevent.spawn(self.recall_region_dup_24h, province_code, size)]
|
|
|
+ else:
|
|
|
+ t = [gevent.spawn(self.rov_pool_recall_with_region_by_h, province_code, size),
|
|
|
+ gevent.spawn(self.rov_pool_recall_with_region_by_h, province_code, size, 'region_24h'),
|
|
|
+ gevent.spawn(self.rov_pool_recall_with_region_by_h, province_code, size, 'day_24h'),
|
|
|
+ gevent.spawn(self.recall_region_dup_24h, province_code, size)]
|
|
|
else:
|
|
|
if province_code == '-1':
|
|
|
t = [gevent.spawn(self.rov_pool_recall_with_region_by_h, province_code, size, 'day_24h')]
|
|
|
@@ -1577,3 +1586,149 @@ class PoolRecall(object):
|
|
|
|
|
|
return recall_result[:size]
|
|
|
|
|
|
+
|
|
|
+ def update_last_video_record(self, record_key, pool_key_prefix, province_code):
|
|
|
+ # 判断当前小时的小时级列表是否更新
|
|
|
+ now_date = datetime.today()
|
|
|
+ h = datetime.now().hour
|
|
|
+ now_dt = datetime.strftime(now_date, '%Y%m%d')
|
|
|
+ now_pool_recall_key = f"{pool_key_prefix}{province_code}.{self.app_type}.{self.data_key}.{self.rule_key}.{now_dt}.{h}"
|
|
|
+ if self.redis_helper.key_exists(key_name=now_pool_recall_key):
|
|
|
+ value = {'date': now_dt, 'h': h}
|
|
|
+ self.redis_helper.set_data_to_redis(key_name=record_key, value=str(value), expire_time=2 * 3600)
|
|
|
+ else:
|
|
|
+ if h == 0:
|
|
|
+ redis_dt = datetime.strftime(now_date - timedelta(days=1), '%Y%m%d')
|
|
|
+ redis_h = 23
|
|
|
+ else:
|
|
|
+ redis_dt = now_dt
|
|
|
+ redis_h = h - 1
|
|
|
+ now_pool_recall_key = f"{pool_key_prefix}{province_code}.{self.app_type}.{self.data_key}.{self.rule_key}.{redis_dt}.{redis_h}"
|
|
|
+ value = {'date': redis_dt, 'h': redis_h}
|
|
|
+ self.redis_helper.set_data_to_redis(key_name=record_key, value=str(value), expire_time=2 * 3600)
|
|
|
+ return now_pool_recall_key
|
|
|
+
|
|
|
+ def get_video_idx(self, pool_recall_key):
|
|
|
+ """
|
|
|
+ 获取上次视频所在位置
|
|
|
+ :param pool_recall_key: 视频所在列表 key
|
|
|
+ :return: idx
|
|
|
+ """
|
|
|
+ last_video_key = f'{config_.LAST_VIDEO_FROM_REGION_DUP3_24H_PREFIX}{self.app_type}.{self.mid}'
|
|
|
+ value = self.redis_helper.get_data_from_redis(last_video_key)
|
|
|
+ if value:
|
|
|
+ idx = self.redis_helper.get_index_with_data(key_name=pool_recall_key, value=value)
|
|
|
+ if not idx:
|
|
|
+ idx = 0
|
|
|
+ else:
|
|
|
+ idx += 1
|
|
|
+ else:
|
|
|
+ idx = 0
|
|
|
+ return idx
|
|
|
+
|
|
|
+ def get_last_recommend_video_idx(self, province_code):
|
|
|
+ # 判断mid对应上一次视频位置 时间记录
|
|
|
+ record_key = f"{config_.RECORD_KEY_NAME_PREFIX_LAST_VIDEO_REGION_DUP3_24H}{self.app_type}.{self.mid}"
|
|
|
+ pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H
|
|
|
+ if not self.redis_helper.key_exists(key_name=record_key):
|
|
|
+ # ###### 记录key不存在
|
|
|
+ idx = 0
|
|
|
+ pool_recall_key = self.update_last_video_record(record_key=record_key, pool_key_prefix=pool_key_prefix,
|
|
|
+ province_code=province_code)
|
|
|
+ else:
|
|
|
+ # ###### 记录key存在，判断date, h
|
|
|
+ now_date = datetime.today()
|
|
|
+ h = datetime.now().hour
|
|
|
+ # 获取记录的date, h
|
|
|
+ record = self.redis_helper.get_data_from_redis(key_name=record_key)
|
|
|
+ record_dt = eval(record).get('date')
|
|
|
+ record_h = eval(record).get('h')
|
|
|
+ now_dt = datetime.strftime(now_date, '%Y%m%d')
|
|
|
+ if record_dt == now_dt and int(record_h) == h:
|
|
|
+ # 已获取当前小时数据
|
|
|
+ pool_recall_key = f"{pool_key_prefix}{province_code}.{self.app_type}.{self.data_key}.{self.rule_key}.{now_dt}.{h}"
|
|
|
+ idx = self.get_video_idx(pool_recall_key=pool_recall_key)
|
|
|
+ elif (record_dt == now_dt and h-int(record_h) == 1) or (h == 0 and int(record_h) == 23):
|
|
|
+ # 记录的h - 当前h = 1，判断当前h数据是否已更新
|
|
|
+ now_pool_recall_key = f"{pool_key_prefix}{province_code}.{self.app_type}.{self.data_key}.{self.rule_key}.{now_dt}.{h}"
|
|
|
+ if self.redis_helper.key_exists(key_name=now_pool_recall_key):
|
|
|
+ new_record = {'date': now_dt, 'h': h}
|
|
|
+ self.redis_helper.set_data_to_redis(key_name=record_key, value=str(new_record), expire_time=2*3600)
|
|
|
+ idx = 0
|
|
|
+ pool_recall_key = now_pool_recall_key
|
|
|
+ else:
|
|
|
+ pool_recall_key = f"{pool_key_prefix}{province_code}.{self.app_type}.{self.data_key}.{self.rule_key}.{record_dt}.{record_h}"
|
|
|
+ idx = self.get_video_idx(pool_recall_key=pool_recall_key)
|
|
|
+ else:
|
|
|
+ idx = 0
|
|
|
+ pool_recall_key = self.update_last_video_record(record_key=record_key, pool_key_prefix=pool_key_prefix,
|
|
|
+ province_code=province_code)
|
|
|
+
|
|
|
+ return pool_recall_key, idx
|
|
|
+
|
|
|
+ def recall_region_dup_24h(self, province_code, size=4, expire_time=2*3600, push_from=config_.PUSH_FROM['rov_recall_24h_dup']):
|
|
|
+ """
|
|
|
+ 从小程序小时级24h数据 筛选后的剩余数据 更新结果中获取视频
|
|
|
+ :param size: 获取视频个数
|
|
|
+ :param expire_time: 末位视频记录redis过期时间
|
|
|
+ :param push_from: 视频来源标记
|
|
|
+ :return:
|
|
|
+ """
|
|
|
+ start_time = time.time()
|
|
|
+ # 获取相关redis key, 用户上一次在rov召回池对应的位置
|
|
|
+ pool_key, idx = self.get_last_recommend_video_idx(province_code=province_code)
|
|
|
+ if not pool_key:
|
|
|
+ return []
|
|
|
+ pool_recall_result = []
|
|
|
+ # 每次获取的视频数
|
|
|
+ get_size = size * 5
|
|
|
+ # 记录获取频次
|
|
|
+ freq = 0
|
|
|
+ while len(pool_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=pool_key,
|
|
|
+ start=idx, end=idx + get_size - 1,
|
|
|
+ with_scores=True)
|
|
|
+ if not data:
|
|
|
+ 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(request_id=self.request_id,
|
|
|
+ app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
|
|
|
+ ge = gevent.spawn(filter_.filter_videos)
|
|
|
+ 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]
|
|
|
+ pool_recall_result.extend(temp_result)
|
|
|
+ else:
|
|
|
+ # 将此次获取的末位视频id同步刷新到Redis中，方便下次快速定位到召回位置，过期时间为1天
|
|
|
+ if self.mid:
|
|
|
+ # mid为空时，不做记录
|
|
|
+ last_video_key = f'{config_.LAST_VIDEO_FROM_REGION_DUP3_24H_PREFIX}{self.app_type}.{self.mid}'
|
|
|
+ self.redis_helper.set_data_to_redis(key_name=last_video_key, value=data[-1][0],
|
|
|
+ expire_time=expire_time)
|
|
|
+ idx += get_size
|
|
|
+
|
|
|
+ pool_recall_result.sort(key=lambda x: x.get('rovScore', 0), reverse=True)
|
|
|
+
|
|
|
+ log_.info({
|
|
|
+ 'logTimestamp': int(time.time() * 1000),
|
|
|
+ 'request_id': self.request_id,
|
|
|
+ 'operation': 'rov_recall_24h_dup',
|
|
|
+ 'executeTime': (time.time() - start_time) * 1000
|
|
|
+ })
|
|
|
+ return pool_recall_result[:size]
|
liqian