import json
import random
import numpy
from log import Log
from config import set_config
from video_recall import PoolRecall
from db_helper import RedisHelper
from utils import FilterVideos, send_msg_to_feishu
from rank_service import get_featurs, get_tf_serving_sores
log_ = Log()
config_ = set_config()
def video_rank(data, size, top_K, flow_pool_P):
"""
视频分发排序
:param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []}
:param size: 请求数
:param top_K: 保证topK为召回池视频 type-int
:param flow_pool_P: size-top_K视频为流量池视频的概率 type-float
:return: rank_result
"""
if not data['rov_pool_recall'] and not data['flow_pool_recall']:
return []
# 将各路召回的视频按照score从大到小排序
# 最惊奇相关推荐相似视频
# relevant_recall = [item for item in data['rov_pool_recall']
# if item.get('pushFrom') == config_.PUSH_FROM['top_video_relevant_appType_19']]
# relevant_recall_rank = sorted(relevant_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
# 最惊奇完整影视视频
# whole_movies_recall = [item for item in data['rov_pool_recall']
# if item.get('pushFrom') == config_.PUSH_FROM['whole_movies']]
# whole_movies_recall_rank = sorted(whole_movies_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
# 最惊奇影视解说视频
# talk_videos_recall = [item for item in data['rov_pool_recall']
# if item.get('pushFrom') == config_.PUSH_FROM['talk_videos']]
# talk_videos_recall_rank = sorted(talk_videos_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
# 小时级更新数据
# h_recall = [item for item in data['rov_pool_recall'] if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_h']]
# h_recall_rank = sorted(h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
# 相对30天天级规则更新数据
day_30_recall = [item for item in data['rov_pool_recall']
if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_30day']]
day_30_recall_rank = sorted(day_30_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
# 地域分组小时级规则更新数据
region_h_recall = [item for item in data['rov_pool_recall']
if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_region_h']]
region_h_recall_rank = sorted(region_h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
# 地域分组小时级更新24h规则更新数据
region_24h_recall = [item for item in data['rov_pool_recall']
if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_region_24h']]
region_24h_recall_rank = sorted(region_24h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
# 地域分组天级规则更新数据
# region_day_recall = [item for item in data['rov_pool_recall']
# if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_region_day']]
# region_day_recall_rank = sorted(region_day_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
# 相对24h规则更新数据
rule_24h_recall = [item for item in data['rov_pool_recall']
if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_24h']]
rule_24h_recall_rank = sorted(rule_24h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
# 相对24h规则筛选后剩余更新数据
rule_24h_dup_recall = [item for item in data['rov_pool_recall']
if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_24h_dup']]
rule_24h_dup_recall_rank = sorted(rule_24h_dup_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
# 相对48h规则更新数据
rule_48h_recall = [item for item in data['rov_pool_recall']
if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_48h']]
rule_48h_recall_rank = sorted(rule_48h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
# 相对48h规则筛选后剩余更新数据
rule_48h_dup_recall = [item for item in data['rov_pool_recall']
if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_48h_dup']]
rule_48h_dup_recall_rank = sorted(rule_48h_dup_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
# 天级规则更新数据
# day_recall = [item for item in data['rov_pool_recall'] if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_day']]
# day_recall_rank = sorted(day_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
# ROV召回池
# rov_initial_recall = [
# item for item in data['rov_pool_recall']
# if item.get('pushFrom') not in
# [config_.PUSH_FROM['top_video_relevant_appType_19'],
# config_.PUSH_FROM['rov_recall_h'],
# config_.PUSH_FROM['rov_recall_region_h'],
# config_.PUSH_FROM['rov_recall_region_24h'],
# config_.PUSH_FROM['rov_recall_region_day'],
# config_.PUSH_FROM['rov_recall_24h'],
# config_.PUSH_FROM['rov_recall_24h_dup'],
# config_.PUSH_FROM['rov_recall_48h'],
# config_.PUSH_FROM['rov_recall_48h_dup'],
# config_.PUSH_FROM['rov_recall_day'],
# config_.PUSH_FROM['whole_movies'],
# config_.PUSH_FROM['talk_videos']]
# ]
# rov_initial_recall_rank = sorted(rov_initial_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
# rov_recall_rank = whole_movies_recall_rank + talk_videos_recall_rank + h_recall_rank + \
# day_30_recall_rank + region_h_recall_rank + region_24h_recall_rank + \
# region_day_recall_rank + rule_24h_recall_rank + rule_24h_dup_recall_rank + \
# rule_48h_recall_rank + rule_48h_dup_recall_rank + \
# day_recall_rank + rov_initial_recall_rank
rov_recall_rank = day_30_recall_rank + \
region_h_recall_rank + region_24h_recall_rank + \
rule_24h_recall_rank + rule_24h_dup_recall_rank + \
rule_48h_recall_rank + rule_48h_dup_recall_rank
# 流量池
flow_recall_rank = sorted(data['flow_pool_recall'], key=lambda k: k.get('rovScore', 0), reverse=True)
# 对各路召回的视频进行去重
rov_recall_rank, flow_recall_rank = remove_duplicate(rov_recall=rov_recall_rank, flow_recall=flow_recall_rank,
top_K=top_K)
# log_.info('remove_duplicate finished! rov_recall_rank = {}, flow_recall_rank = {}'.format(
# rov_recall_rank, flow_recall_rank))
# rank_result = relevant_recall_rank
rank_result = []
# 从ROV召回池中获取top k
if len(rov_recall_rank) > 0:
rank_result.extend(rov_recall_rank[:top_K])
rov_recall_rank = rov_recall_rank[top_K:]
else:
rank_result.extend(flow_recall_rank[:top_K])
flow_recall_rank = flow_recall_rank[top_K:]
# 按概率 p 及score排序获取 size - k 个视频
i = 0
while i < size - top_K:
# 随机生成[0, 1)浮点数
rand = random.random()
# log_.info('rand: {}'.format(rand))
if rand < flow_pool_P:
if flow_recall_rank:
rank_result.append(flow_recall_rank[0])
flow_recall_rank.remove(flow_recall_rank[0])
else:
rank_result.extend(rov_recall_rank[:size - top_K - i])
return rank_result[:size]
else:
if rov_recall_rank:
rank_result.append(rov_recall_rank[0])
rov_recall_rank.remove(rov_recall_rank[0])
else:
rank_result.extend(flow_recall_rank[:size - top_K - i])
return rank_result[:size]
i += 1
return rank_result[:size]
def video_new_rank(videoIds, fast_flow_set, flow_set, size, top_K, flow_pool_P):
"""
视频分发排序
:param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []}
:param size: 请求数
:param top_K: 保证topK为召回池视频 type-int
:param flow_pool_P: size-top_K视频为流量池视频的概率 type-float
:return: rank_result
"""
add_flow_set = set('')
if not videoIds or len(videoIds)==0:
return [], add_flow_set
redisObj = RedisHelper()
vidKeys = []
for vid in videoIds:
vidKeys.append("k_p:"+str(vid))
#print("vidKeys:", vidKeys)
video_scores = redisObj.get_batch_key(vidKeys)
#print(video_scores)
video_items = []
for i in range(len(video_scores)):
try:
#print(video_scores[i])
if video_scores[i] is None:
video_items.append((videoIds[i], 0.0))
else:
video_score_str = json.loads(video_scores[i])
#print("video_score_str:",video_score_str)
video_items.append((videoIds[i], video_score_str[0]))
except Exception:
video_items.append((videoIds[i], 0.0))
sort_items = sorted(video_items, key=lambda k: k[1], reverse=True)
#print("sort_items:", sort_items)
rov_recall_rank = sort_items
fast_flow_recall_rank = []
flow_recall_rank = []
for item in sort_items:
if item[0] in fast_flow_set:
fast_flow_recall_rank.append(item)
elif item[0] in flow_set:
flow_recall_rank.append(item)
# all flow result
all_flow_recall_rank = fast_flow_recall_rank+flow_recall_rank
rank_result = []
rank_set = set('')
# 从ROV召回池中获取top k
if len(rov_recall_rank) > 0:
rank_result.extend(rov_recall_rank[:top_K])
rov_recall_rank = rov_recall_rank[top_K:]
else:
rank_result.extend(all_flow_recall_rank[:top_K])
all_flow_recall_rank = all_flow_recall_rank[top_K:]
for rank_item in rank_result:
rank_set.add(rank_item[0])
#print("rank_result:", rank_result)
# 按概率 p 及score排序获取 size - k 个视频， 第4个位置按概率取流量池
i = 0
left_quato = size - top_K
j = 0
jj = 0
while i < left_quato and (j<len(all_flow_recall_rank) or jj<len(rov_recall_rank)):
# 随机生成[0, 1)浮点数
rand = random.random()
# log_.info('rand: {}'.format(rand))
if rand < flow_pool_P:
for flow_item in all_flow_recall_rank:
j+=1
if flow_item[0] in rank_set:
continue
else:
rank_result.append(flow_item)
rank_set.add(flow_item[0])
add_flow_set.add(flow_item[0])
i += 1
if i>= left_quato:
break
else:
for recall_item in rov_recall_rank:
jj+=1
if recall_item[0] in rank_set:
continue
else:
rank_result.append(recall_item)
rank_set.add(recall_item[0])
i += 1
if i>= left_quato:
break
#print("rank_result:", rank_result)
#print("add_flow_set:", add_flow_set)
return rank_result[:size], add_flow_set
def refactor_video_rank(rov_recall_rank, fast_flow_set, flow_set, size, top_K, flow_pool_P):
"""
视频分发排序
:param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []}
:param size: 请求数
:param top_K: 保证topK为召回池视频 type-int
:param flow_pool_P: size-top_K视频为流量池视频的概率 type-float
:return: rank_result
"""
if not rov_recall_rank or len(rov_recall_rank) == 0:
return []
fast_flow_recall_rank = []
flow_recall_rank = []
for item in rov_recall_rank:
vid = item.get('videoId', 0)
#print(item)
if vid in fast_flow_set:
fast_flow_recall_rank.append(item)
elif vid in flow_set:
flow_recall_rank.append(item)
# all flow result
all_flow_recall_rank = fast_flow_recall_rank + flow_recall_rank
rank_result = []
rank_set = set('')
# 从ROV召回池中获取top k
if len(rov_recall_rank) > 0:
rank_result.extend(rov_recall_rank[:top_K])
rov_recall_rank = rov_recall_rank[top_K:]
else:
rank_result.extend(all_flow_recall_rank[:top_K])
all_flow_recall_rank = all_flow_recall_rank[top_K:]
#已存放了多少VID
for rank_item in rank_result:
rank_set.add(rank_item.get('videoId', 0))
# 按概率 p 及score排序获取 size - k 个视频， 第4个位置按概率取流量池
i = 0
while i < size - top_K:
# 随机生成[0, 1)浮点数
rand = random.random()
# log_.info('rand: {}'.format(rand))
if rand < flow_pool_P:
for flow_item in all_flow_recall_rank:
flow_vid = flow_item.get('videoId', 0)
if flow_vid in rank_set:
continue
else:
rank_result.append(flow_item)
rank_set.add(flow_vid)
else:
for recall_item in rov_recall_rank:
flow_vid = recall_item.get('videoId', 0)
if flow_vid in rank_set:
continue
else:
rank_result.append(recall_item)
rank_set.add(flow_vid)
i += 1
return rank_result[:size]
def remove_duplicate(rov_recall, flow_recall, top_K):
"""
对多路召回的视频去重
去重原则：
如果视频在ROV召回池topK，则保留ROV召回池，否则保留流量池
:param rov_recall: ROV召回池-已排序
:param flow_recall: 流量池-已排序
:param top_K: 保证topK为召回池视频 type-int
:return:
"""
flow_recall_result = []
rov_recall_remove = []
flow_recall_video_ids = [item['videoId'] for item in flow_recall]
# rov_recall topK
for item in rov_recall[:top_K]:
if item['videoId'] in flow_recall_video_ids:
flow_recall_video_ids.remove(item['videoId'])
# other
for item in rov_recall[top_K:]:
if item['videoId'] in flow_recall_video_ids:
rov_recall_remove.append(item)
# rov recall remove
for item in rov_recall_remove:
rov_recall.remove(item)
# flow recall remove
for item in flow_recall:
if item['videoId'] in flow_recall_video_ids:
flow_recall_result.append(item)
return rov_recall, flow_recall_result
def bottom_strategy(request_id, size, app_type, ab_code, params):
"""
兜底策略: 从ROV召回池中获取top1000，进行状态过滤后的视频
:param request_id: request_id
:param size: 需要获取的视频数
:param app_type: 产品标识 type-int
:param ab_code: abCode
:param params:
:return:
"""
pool_recall = PoolRecall(request_id=request_id, app_type=app_type, ab_code=ab_code)
key_name, _ = pool_recall.get_pool_redis_key(pool_type='rov')
redis_helper = RedisHelper(params=params)
data = redis_helper.get_data_zset_with_index(key_name=key_name, start=0, end=1000)
if not data:
log_.info('{} —— ROV推荐进入了二次兜底, data = {}'.format(config_.ENV_TEXT, data))
send_msg_to_feishu('{} —— ROV推荐进入了二次兜底，请查看是否有数据更新失败问题。'.format(config_.ENV_TEXT))
# 二次兜底
bottom_data = bottom_strategy_last(size=size, app_type=app_type, ab_code=ab_code, params=params)
return bottom_data
# 视频状态过滤采用离线定时过滤方案
# 状态过滤
# filter_videos = FilterVideos(app_type=app_type, video_ids=data)
# filtered_data = filter_videos.filter_video_status(video_ids=data)
if len(data) > size:
random_data = numpy.random.choice(data, size, False)
else:
random_data = data
bottom_data = [{'videoId': int(item), 'pushFrom': config_.PUSH_FROM['bottom'], 'abCode': ab_code}
for item in random_data]
return bottom_data
def bottom_strategy_last(size, app_type, ab_code, params):
"""
兜底策略: 从兜底视频中随机获取视频，进行状态过滤后的视频
:param size: 需要获取的视频数
:param app_type: 产品标识 type-int
:param ab_code: abCode
:param params:
:return:
"""
redis_helper = RedisHelper(params=params)
bottom_data = redis_helper.get_data_zset_with_index(key_name=config_.BOTTOM_KEY_NAME, start=0, end=-1)
random_data = numpy.random.choice(bottom_data, size * 30, False)
# 视频状态过滤采用离线定时过滤方案
# 状态过滤
# filter_videos = FilterVideos(app_type=app_type, video_ids=random_data)
# filtered_data = filter_videos.filter_video_status(video_ids=random_data)
bottom_data = [{'videoId': int(video_id), 'pushFrom': config_.PUSH_FROM['bottom_last'], 'abCode': ab_code}
for video_id in random_data[:size]]
return bottom_data
def bottom_strategy2(size, app_type, mid, uid, ab_code, client_info, params):
"""
兜底策略: 从兜底视频中随机获取视频，进行过滤后的视频
:param size: 需要获取的视频数
:param app_type: 产品标识 type-int
:param mid: mid
:param uid: uid
:param ab_code: abCode
:param client_info: 地域信息
:param params:
:return:
"""
# 获取存在城市分组数据的城市编码列表
city_code_list = [code for _, code in config_.CITY_CODE.items()]
# 获取provinceCode
province_code = client_info.get('provinceCode', '-1')
# 获取cityCode
city_code = client_info.get('cityCode', '-1')
if city_code in city_code_list:
# 分城市数据存在时，获取城市分组数据
region_code = city_code
else:
region_code = province_code
if region_code == '':
region_code = '-1'
redis_helper = RedisHelper(params=params)
bottom_data = redis_helper.get_data_from_set(key_name=config_.BOTTOM2_KEY_NAME)
bottom_result = []
if bottom_data is None:
return bottom_result
if len(bottom_data) > 0:
try:
random_data = numpy.random.choice(bottom_data, size * 5, False)
except Exception as e:
random_data = bottom_data
video_ids = [int(item) for item in random_data]
# 过滤
filter_ = FilterVideos(request_id=params.request_id, app_type=app_type, mid=mid, uid=uid, video_ids=video_ids)
filtered_data = filter_.filter_videos(pool_type='flow', region_code=region_code)
if filtered_data:
bottom_result = [{'videoId': int(video_id), 'pushFrom': config_.PUSH_FROM['bottom2'], 'abCode': ab_code}
for video_id in filtered_data[:size]]
return bottom_result
def video_rank_by_w_h_rate(videos):
"""
视频宽高比实验(每组的前两个视频调整为横屏视频)，根据视频宽高比信息对视频进行重排
:param videos:
:return:
"""
redis_helper = RedisHelper()
# ##### 判断前两个视频是否是置顶视频 或者 流量池视频
top_2_push_from_flag = [False, False]
for i, video in enumerate(videos[:2]):
if video['pushFrom'] in [config_.PUSH_FROM['top'], config_.PUSH_FROM['flow_recall']]:
top_2_push_from_flag[i] = True
if top_2_push_from_flag[0] and top_2_push_from_flag[1]:
return videos
# ##### 判断前两个视频是否为横屏
top_2_w_h_rate_flag = [False, False]
for i, video in enumerate(videos[:2]):
if video['pushFrom'] in [config_.PUSH_FROM['top'], config_.PUSH_FROM['flow_recall']]:
# 视频来源为置顶 或 流量池时，不做判断
top_2_w_h_rate_flag[i] = True
elif video['pushFrom'] in [config_.PUSH_FROM['rov_recall'], config_.PUSH_FROM['bottom']]:
# 视频来源为 rov召回池 或 一层兜底时，判断是否是横屏
w_h_rate = redis_helper.get_score_with_value(
key_name=config_.W_H_RATE_UP_1_VIDEO_LIST_KEY_NAME['rov_recall'], value=video['videoId'])
if w_h_rate is not None:
top_2_w_h_rate_flag[i] = True
elif video['pushFrom'] == config_.PUSH_FROM['bottom_last']:
# 视频来源为 二层兜底时，判断是否是横屏
w_h_rate = redis_helper.get_score_with_value(
key_name=config_.W_H_RATE_UP_1_VIDEO_LIST_KEY_NAME['bottom_last'], value=video['videoId'])
if w_h_rate is not None:
top_2_w_h_rate_flag[i] = True
if top_2_w_h_rate_flag[0] and top_2_w_h_rate_flag[1]:
return videos
# ##### 前两个视频中有不符合前面两者条件的，对视频进行位置调整
# 记录横屏视频位置
horizontal_video_index = []
# 记录流量池视频位置
flow_video_index = []
# 记录置顶视频位置
top_video_index = []
for i, video in enumerate(videos):
# 视频来源为置顶
if video['pushFrom'] == config_.PUSH_FROM['top']:
top_video_index.append(i)
# 视频来源为流量池
elif video['pushFrom'] == config_.PUSH_FROM['flow_recall']:
flow_video_index.append(i)
# 视频来源为rov召回池 或 一层兜底
elif video['pushFrom'] in [config_.PUSH_FROM['rov_recall'], config_.PUSH_FROM['bottom']]:
w_h_rate = redis_helper.get_score_with_value(
key_name=config_.W_H_RATE_UP_1_VIDEO_LIST_KEY_NAME['rov_recall'], value=video['videoId'])
if w_h_rate is not None:
horizontal_video_index.append(i)
else:
continue
# 视频来源为 二层兜底
elif video['pushFrom'] == config_.PUSH_FROM['bottom_last']:
w_h_rate = redis_helper.get_score_with_value(
key_name=config_.W_H_RATE_UP_1_VIDEO_LIST_KEY_NAME['bottom_last'], value=video['videoId'])
if w_h_rate is not None:
horizontal_video_index.append(i)
else:
continue
# 重新排序
top2_index = []
for i in range(2):
if i in top_video_index:
top2_index.append(i)
elif i in flow_video_index:
top2_index.append(i)
flow_video_index.remove(i)
elif i in horizontal_video_index:
top2_index.append(i)
horizontal_video_index.remove(i)
elif len(horizontal_video_index) > 0:
# 调整横屏视频到第一位
top2_index.append(horizontal_video_index[0])
# 从横屏位置记录中移除
horizontal_video_index.pop(0)
elif i == 0:
return videos
# 重排
flow_result = [videos[i] for i in flow_video_index]
other_result = [videos[i] for i in range(len(videos)) if i not in top2_index and i not in flow_video_index]
top2_result = []
for i, j in enumerate(top2_index):
item = videos[j]
if i != j:
# 修改abCode
item['abCode'] = config_.AB_CODE['w_h_rate']
top2_result.append(item)
new_rank_result = top2_result
for i in range(len(top2_index), len(videos)):
if i in flow_video_index:
new_rank_result.append(flow_result[0])
flow_result.pop(0)
else:
new_rank_result.append(other_result[0])
other_result.pop(0)
return new_rank_result
def video_rank_with_old_video(rank_result, old_video_recall, size, top_K, old_video_index=2):
"""
视频分发排序 - 包含老视频， 老视频插入固定位置
:param rank_result: 排序后的结果
:param size: 请求数
:param old_video_index: 老视频插入的位置索引，默认为2
:return: new_rank_result
"""
if not old_video_recall:
return rank_result
if not rank_result:
return old_video_recall[:size]
# 视频去重
rank_video_ids = [item['videoId'] for item in rank_result]
old_video_remove = []
for old_video in old_video_recall:
if old_video['videoId'] in rank_video_ids:
old_video_remove.append(old_video)
for item in old_video_remove:
old_video_recall.remove(item)
if not old_video_recall:
return rank_result
# 插入老视频
# 随机获取一个视频
ind = random.randint(0, len(old_video_recall) - 1)
old_video = old_video_recall[ind]
# 插入
if len(rank_result) < top_K:
new_rank_result = rank_result + [old_video]
else:
new_rank_result = rank_result[:old_video_index] + [old_video] + rank_result[old_video_index:]
if len(new_rank_result) > size:
# 判断后两位视频来源
push_from_1 = new_rank_result[-1]['pushFrom']
push_from_2 = new_rank_result[-2]['pushFrom']
if push_from_2 == config_.PUSH_FROM['rov_recall'] and push_from_1 == config_.PUSH_FROM['flow_recall']:
new_rank_result = new_rank_result[:-2] + new_rank_result[-1:]
return new_rank_result[:size]
def video_new_rank2(data, size, top_K, flow_pool_P, ab_code, mid, exp_config=None, env_dict=None):
"""
视频分发排序
:param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []}
:param size: 请求数
:param top_K: 保证topK为召回池视频 type-int
:param flow_pool_P: size-top_K视频为流量池视频的概率 type-float
:return: rank_result
"""
if not data['rov_pool_recall'] and not data['flow_pool_recall']:
return [], 0
#全量的是vlog,票圈精选， 334，60057,
# 60054: simrecall,
# 60052: 票圈精选，融合排序，60053:空置
# 60055: video_reall, 60065: video_recall2
# 60056: get_U2I_reall
pre_str = "k_p2:"
if ab_code==60057:
pre_str = "k_p2:"
elif ab_code == 60052:
pre_str = "k_p5:"
elif ab_code == 60053:
pre_str = "k_p8:"
elif ab_code == 60054:
pre_str = "k_p3:"
elif ab_code == 60055:
pre_str = "k_p4:"
elif ab_code == 60056:
pre_str = "k_p7:"
#print("pre_str:", pre_str)
recall_list = []
rov_recall_rank = data['rov_pool_recall']
#call rank service
#flag_call_service = 0
if ab_code == 60066:
feature_dict, recall_list = get_featurs(mid, data, size, top_K, flow_pool_P, env_dict)
score_result = get_tf_serving_sores(feature_dict)
if score_result and len(score_result) > 0 and len(score_result) == len(recall_list):
for i in range(len(score_result)):
recall_list[i]['sort_score'] = score_result[i][0]
recall_list[i]['flag_call_service'] = 1
rov_recall_rank = sorted(recall_list, key=lambda k: k.get('sort_score', 0), reverse=True)
else:
rov_recall_rank = sup_rank(data, pre_str, recall_list, rov_recall_rank)
else:
redisObj = RedisHelper()
vidKeys = []
for recall_item in data['rov_pool_recall']:
if len(recall_item) <= 0:
continue
vid = recall_item.get("videoId", 0)
vidKeys.append(pre_str + str(vid))
recall_list.append(recall_item)
video_scores = redisObj.get_batch_key(vidKeys)
if video_scores and len(recall_list) > 0:
for i in range(len(video_scores)):
try:
if video_scores[i] is None:
recall_list[i]['sort_score'] = 0.0
else:
video_score_str = json.loads(video_scores[i])
# print("video_score_str:", video_score_str)
recall_list[i]['sort_score'] = video_score_str[0]
except Exception:
recall_list[i]['sort_score'] = 0.0
rov_recall_rank = sorted(recall_list, key=lambda k: k.get('sort_score', 0), reverse=True)
#print(rov_recall_rank)
flow_recall_rank = sorted(data['flow_pool_recall'], key=lambda k: k.get('rovScore', 0), reverse=True)
rov_recall_rank, flow_recall_rank = remove_duplicate(rov_recall=rov_recall_rank, flow_recall=flow_recall_rank,
top_K=top_K)
rank_result = []
rank_set = set('')
# 从ROV召回池中获取top k
if len(rov_recall_rank) > 0:
rank_result.extend(rov_recall_rank[:top_K])
rov_recall_rank = rov_recall_rank[top_K:]
else:
rank_result.extend(flow_recall_rank[:top_K])
flow_recall_rank = flow_recall_rank[top_K:]
# 按概率 p 及score排序获取 size - k 个视频
flow_num = 0
flowConfig = 0
if exp_config and exp_config['flowConfig']:
flowConfig = exp_config['flowConfig']
if flowConfig == 1 and len(rov_recall_rank) > 0:
for recall_item in rank_result:
flow_recall_name = recall_item.get("flowPool", '')
flow_num = flow_num + 1
all_recall_rank = rov_recall_rank + flow_recall_rank
if flow_num > 0:
rank_result.extend(all_recall_rank[:size - top_K])
return rank_result, flow_num
else:
i = 0
while i < size - top_K:
# 随机生成[0, 1)浮点数
rand = random.random()
# log_.info('rand: {}'.format(rand))
if rand < flow_pool_P:
if flow_recall_rank:
rank_result.append(flow_recall_rank[0])
flow_recall_rank.remove(flow_recall_rank[0])
else:
rank_result.extend(rov_recall_rank[:size - top_K - i])
return rank_result[:size], flow_num
else:
if rov_recall_rank:
rank_result.append(rov_recall_rank[0])
rov_recall_rank.remove(rov_recall_rank[0])
else:
rank_result.extend(flow_recall_rank[:size - top_K - i])
return rank_result[:size], flow_num
i += 1
else:
i = 0
while i < size - top_K:
# 随机生成[0, 1)浮点数
rand = random.random()
# log_.info('rand: {}'.format(rand))
if rand < flow_pool_P:
if flow_recall_rank:
rank_result.append(flow_recall_rank[0])
flow_recall_rank.remove(flow_recall_rank[0])
else:
rank_result.extend(rov_recall_rank[:size - top_K - i])
return rank_result[:size], flow_num
else:
if rov_recall_rank:
rank_result.append(rov_recall_rank[0])
rov_recall_rank.remove(rov_recall_rank[0])
else:
rank_result.extend(flow_recall_rank[:size - top_K - i])
return rank_result[:size], flow_num
i += 1
return rank_result[:size], flow_num
# 排序服务兜底
def sup_rank(data, pre_str, recall_list, rov_recall_rank):
redisObj = RedisHelper()
vidKeys = []
for recall_item in data['rov_pool_recall']:
if len(recall_item) <= 0:
continue
vid = recall_item.get("videoId", 0)
vidKeys.append(pre_str + str(vid))
recall_list.append(recall_item)
video_scores = redisObj.get_batch_key(vidKeys)
#print("vidKeys:", video_scores, "\t", vidKeys)
#print(len(video_scores), len(recall_list))
if video_scores and len(recall_list) > 0:
for i in range(len(video_scores)):
try:
if video_scores[i] is None:
recall_list[i]['sort_score'] = 0.0
else:
video_score_str = json.loads(video_scores[i])
recall_list[i]['flag_call_service'] = 0
recall_list[i]['sort_score'] = video_score_str[0]
except Exception:
recall_list[i]['sort_score'] = 0.0
rov_recall_rank = sorted(recall_list, key=lambda k: k.get('sort_score', 0), reverse=True)
#print("rov_recall_rank:", rov_recall_rank)
else:
rov_recall_rank = recall_list
return rov_recall_rank
def video_sanke_rank(data, size, top_K, flow_pool_P, ab_Code='', exp_config=None):
"""
视频分发排序
:param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []}
:param size: 请求数
:param top_K: 保证topK为召回池视频 type-int
:param flow_pool_P: size-top_K视频为流量池视频的概率 type-float
:return: rank_result
"""
if not data['rov_pool_recall'] and not data['flow_pool_recall'] \
and len(data['u2i_recall'])==0 and len(data['w2v_recall'])==0 \
and len(data['sim_recall']) == 0 and len(data['u2u2i_recall']) == 0 :
return [], 0
# 地域分组小时级规则更新数据
recall_dict = {}
region_h_recall = [item for item in data['rov_pool_recall']
if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_region_h']]
region_h_recall_rank = sorted(region_h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
recall_dict['rov_recall_region_h'] = region_h_recall_rank
# 地域分组小时级更新24h规则更新数据
region_24h_recall = [item for item in data['rov_pool_recall']
if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_region_24h']]
region_24h_recall_rank = sorted(region_24h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
recall_dict['rov_recall_region_24h'] = region_24h_recall_rank
# 相对24h规则更新数据
rule_24h_recall = [item for item in data['rov_pool_recall']
if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_24h']]
rule_24h_recall_rank = sorted(rule_24h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
recall_dict['rov_recall_24h'] = rule_24h_recall_rank
# 相对24h规则筛选后剩余更新数据
rule_24h_dup_recall = [item for item in data['rov_pool_recall']
if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_24h_dup']]
rule_24h_dup_recall_rank = sorted(rule_24h_dup_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
recall_dict['rov_recall_24h_dup'] = rule_24h_dup_recall_rank
hot_recall = []
w2v_recall =[]
sim_recall = []
u2u2i_recall = []
if ab_Code==60058:
if len(data['u2i_recall'])>0:
hot_recall = sorted(data['u2i_recall'], key=lambda k: k.get('rovScore', 0), reverse=True)
recall_dict['u2i_recall'] = hot_recall
elif ab_Code==60059:
if len(data['w2v_recall'])>0:
recall_dict['w2v_recall'] = data['w2v_recall']
else:
recall_dict['w2v_recall'] = w2v_recall
elif ab_Code==60061 or ab_Code==60063:
if len(data['sim_recall'])>0:
recall_dict['sim_recall'] = data['sim_recall']
else:
recall_dict['sim_recall'] = sim_recall
elif ab_Code==60062:
if len(data['u2u2i_recall'])>0:
recall_dict['u2u2i_recall'] = data['u2u2i_recall']
else:
recall_dict['u2u2i_recall'] = u2u2i_recall
recall_list = [('rov_recall_region_h',1, 1),('rov_recall_region_h',0.5, 1),('rov_recall_region_24h',1,1),
('u2i_recall',0.5,1), ('w2v_recall',0.5,1),('rov_recall_24h',1,1), ('rov_recall_24h_dup',0.5,1)]
if exp_config and exp_config['recall_list']:
recall_list = exp_config['recall_list']
#print("recall_config:", recall_list)
rov_recall_rank = []
select_ids = set('')
for i in range(3):
if len(rov_recall_rank)>8:
break
for per_recall_item in recall_list:
per_recall_name = per_recall_item[0]
per_recall_freq = per_recall_item[1]
per_limt_num = per_recall_item[2]
rand_num = random.random()
#print(recall_dict[per_recall_name])
if rand_num<per_recall_freq and per_recall_name in recall_dict:
per_recall = recall_dict[per_recall_name]
#print("per_recall_item:", per_recall_item)
cur_recall_num = 0
for recall_item in per_recall:
vid = recall_item['videoId']
if vid in select_ids:
continue
rov_recall_rank.append(recall_item)
select_ids.add(vid)
cur_recall_num+=1
if cur_recall_num>=per_limt_num:
break
# print("rov_recall_rank:")
# print(rov_recall_rank)
#rov_recall_rank = region_h_recall_rank + region_24h_recall_rank + \
# rule_24h_recall_rank + rule_24h_dup_recall_rank
# 流量池
flow_recall_rank = sorted(data['flow_pool_recall'], key=lambda k: k.get('rovScore', 0), reverse=True)
# 对各路召回的视频进行去重
rov_recall_rank, flow_recall_rank = remove_duplicate(rov_recall=rov_recall_rank, flow_recall=flow_recall_rank,
top_K=top_K)
# log_.info('remove_duplicate finished! rov_recall_rank = {}, flow_recall_rank = {}'.format(
# rov_recall_rank, flow_recall_rank))
# rank_result = relevant_recall_rank
rank_result = []
# 从ROV召回池中获取top k
if len(rov_recall_rank) > 0:
rank_result.extend(rov_recall_rank[:top_K])
rov_recall_rank = rov_recall_rank[top_K:]
else:
rank_result.extend(flow_recall_rank[:top_K])
flow_recall_rank = flow_recall_rank[top_K:]
flow_num = 0
flowConfig =0
if exp_config and exp_config['flowConfig']:
flowConfig = exp_config['flowConfig']
if flowConfig == 1 and len(rov_recall_rank) > 0:
rank_result.extend(rov_recall_rank[:top_K])
for recall_item in rank_result:
flow_recall_name = recall_item.get("flowPool", '')
if flow_recall_name is not None and flow_recall_name.find("#") > -1:
flow_num = flow_num + 1
all_recall_rank = rov_recall_rank + flow_recall_rank
if flow_num > 0:
rank_result.extend(all_recall_rank[:size - top_K])
return rank_result[:size], flow_num
else:
# 按概率 p 及score排序获取 size - k 个视频
i = 0
while i < size - top_K:
# 随机生成[0, 1)浮点数
rand = random.random()
# log_.info('rand: {}'.format(rand))
if rand < flow_pool_P:
if flow_recall_rank:
rank_result.append(flow_recall_rank[0])
flow_recall_rank.remove(flow_recall_rank[0])
else:
rank_result.extend(rov_recall_rank[:size - top_K - i])
return rank_result[:size], flow_num
else:
if rov_recall_rank:
rank_result.append(rov_recall_rank[0])
rov_recall_rank.remove(rov_recall_rank[0])
else:
rank_result.extend(flow_recall_rank[:size - top_K - i])
return rank_result[:size], flow_num
i += 1
else:
# 按概率 p 及score排序获取 size - k 个视频
i = 0
while i < size - top_K:
# 随机生成[0, 1)浮点数
rand = random.random()
# log_.info('rand: {}'.format(rand))
if rand < flow_pool_P:
if flow_recall_rank:
rank_result.append(flow_recall_rank[0])
flow_recall_rank.remove(flow_recall_rank[0])
else:
rank_result.extend(rov_recall_rank[:size - top_K - i])
return rank_result[:size], flow_num
else:
if rov_recall_rank:
rank_result.append(rov_recall_rank[0])
rov_recall_rank.remove(rov_recall_rank[0])
else:
rank_result.extend(flow_recall_rank[:size - top_K - i])
return rank_result[:size],flow_num
i += 1
return rank_result[:size], flow_num
def video_sank_pos_rank(data, size, top_K, flow_pool_P, ab_Code='', exp_config=None):
"""
视频分发排序
:param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []}
:param size: 请求数
:param top_K: 保证topK为召回池视频 type-int
:param flow_pool_P: size-top_K视频为流量池视频的概率 type-float
:return: rank_result
"""
if not data['rov_pool_recall'] and not data['flow_pool_recall'] \
and len(data['u2i_recall'])==0 and len(data['w2v_recall'])==0 \
and len(data['sim_recall']) == 0 and len(data['u2u2i_recall']) == 0 :
return [], 0
# 地域分组小时级规则更新数据
recall_dict = {}
region_h_recall = [item for item in data['rov_pool_recall']
if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_region_h']]
region_h_recall_rank = sorted(region_h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
recall_dict['rov_recall_region_h'] = region_h_recall_rank
# 地域分组小时级更新24h规则更新数据
region_24h_recall = [item for item in data['rov_pool_recall']
if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_region_24h']]
region_24h_recall_rank = sorted(region_24h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
recall_dict['rov_recall_region_24h'] = region_24h_recall_rank
# 相对24h规则更新数据
rule_24h_recall = [item for item in data['rov_pool_recall']
if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_24h']]
rule_24h_recall_rank = sorted(rule_24h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
recall_dict['rov_recall_24h'] = rule_24h_recall_rank
# 相对24h规则筛选后剩余更新数据
rule_24h_dup_recall = [item for item in data['rov_pool_recall']
if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_24h_dup']]
rule_24h_dup_recall_rank = sorted(rule_24h_dup_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
recall_dict['rov_recall_24h_dup'] = rule_24h_dup_recall_rank
u2i_recall = []
u2i_play_recall = []
w2v_recall =[]
sim_recall = []
u2u2i_recall = []
return_video_recall = []
#print("")
if ab_Code==60058:
if len(data['u2i_recall'])>0:
recall_dict['u2i_recall'] = data['u2i_recall']
else:
recall_dict['u2i_recall'] = u2i_recall
if len(data['u2i_play_recall']) > 0:
recall_dict['u2i_play_recall'] = data['u2i_play_recall']
else:
recall_dict['u2i_play_recall'] = u2i_play_recall
elif ab_Code==60059:
if len(data['w2v_recall'])>0:
recall_dict['w2v_recall'] = data['w2v_recall']
else:
recall_dict['w2v_recall'] = w2v_recall
elif ab_Code==60061 or ab_Code==60063:
if len(data['sim_recall'])>0:
recall_dict['sim_recall'] = data['sim_recall']
else:
recall_dict['sim_recall'] = sim_recall
elif ab_Code==60062:
if len(data['u2u2i_recall'])>0:
recall_dict['u2u2i_recall'] = data['u2u2i_recall']
else:
recall_dict['u2u2i_recall'] = u2u2i_recall
elif ab_Code==60064:
if len(data['return_video_recall'])>0:
recall_dict['return_video_recall'] = data['return_video_recall']
else:
recall_dict['return_video_recall'] = return_video_recall
recall_pos1 = [('rov_recall_region_h',0, 0.98),('rov_recall_24h',0.98, 1),('rov_recall_region_24h',0,1),
('rov_recall_24h',0,1), ('rov_recall_24h_dup',0,1)]
recall_pos2 = [('rov_recall_region_h',0,0.98),('rov_recall_24h',0.98,1),('rov_recall_region_24h',0,1),
('rov_recall_24h',0,1),('rov_recall_24h_dup',0,1)]
recall_pos3 = [('rov_recall_region_h', 0,0.98), ('rov_recall_24h', 0.98,1), ('rov_recall_region_24h', 0,1),
('rov_recall_24h', 0,1), ('rov_recall_24h_dup', 0,1)]
recall_pos4 = [('rov_recall_region_h', 0,0.98), ('rov_recall_24h', 0,0.02), ('rov_recall_region_24h', 0,1),
('rov_recall_24h', 0,1), ('rov_recall_24h_dup', 0,1)]
if exp_config and 'recall_pos1' in exp_config \
and 'recall_pos2' in exp_config \
and 'recall_pos3' in exp_config \
and 'recall_pos4' in exp_config :
recall_pos1 = exp_config['recall_pos1']
recall_pos2 = exp_config['recall_pos2']
recall_pos3 = exp_config['recall_pos3']
recall_pos4 = exp_config['recall_pos4']
#print("recall_config:", recall_pos1)
rov_recall_rank = []
recall_list = []
recall_list.append(recall_pos1)
recall_list.append(recall_pos2)
recall_list.append(recall_pos3)
recall_list.append(recall_pos4)
select_ids = set('')
recall_num_limit_dict = {}
if exp_config and 'recall_num_limit' in exp_config:
recall_num_limit_dict = exp_config['recall_num_limit']
exp_recall_dict = {}
#index_pos = 0
for j in range(3):
if len(rov_recall_rank)>12:
break
# choose pos
for recall_pos_config in recall_list:
rand_num = random.random()
index_pos = 0
# choose pos recall
for per_recall_item in recall_pos_config:
if index_pos == 1:
break
if len(per_recall_item)<3:
continue
per_recall_name = per_recall_item[0]
per_recall_min = float(per_recall_item[1])
per_recall_max = float(per_recall_item[2])
per_recall_num = exp_recall_dict.get(per_recall_name, 0)
per_recall_total_num = recall_num_limit_dict.get(per_recall_name, 0)
# recall set total num
if len(recall_num_limit_dict)>0 and per_recall_total_num>0 and per_recall_num>= per_recall_total_num:
continue
if rand_num >= per_recall_min and rand_num < per_recall_max and per_recall_name in recall_dict:
per_recall = recall_dict[per_recall_name]
for recall_item in per_recall:
vid = recall_item['videoId']
if vid in select_ids:
continue
recall_item['rand'] = rand_num
rov_recall_rank.append(recall_item)
select_ids.add(vid)
if per_recall_name in exp_recall_dict:
exp_recall_dict[per_recall_name] +=1
else:
exp_recall_dict[per_recall_name] = 1
index_pos = 1
break
#print("rov_recall_rank:", rov_recall_rank)
if len(rov_recall_rank)<4:
rov_doudi_rank = region_h_recall_rank + sim_recall + u2i_recall + u2u2i_recall + w2v_recall +return_video_recall+u2i_play_recall+ region_24h_recall_rank + rule_24h_recall_rank + rule_24h_dup_recall_rank
for recall_item in rov_doudi_rank:
vid = recall_item['videoId']
if vid in select_ids:
continue
rov_recall_rank.append(recall_item)
select_ids.add(vid)
if len(rov_recall_rank)>12:
break
# print("rov_recall_rank:")
#print(rov_recall_rank)
# 流量池
flow_recall_rank = sorted(data['flow_pool_recall'], key=lambda k: k.get('rovScore', 0), reverse=True)
# 对各路召回的视频进行去重
rov_recall_rank, flow_recall_rank = remove_duplicate(rov_recall=rov_recall_rank, flow_recall=flow_recall_rank,
top_K=top_K)
# log_.info('remove_duplicate finished! rov_recall_rank = {}, flow_recall_rank = {}'.format(
# rov_recall_rank, flow_recall_rank))
# rank_result = relevant_recall_rank
rank_result = []
# 从ROV召回池中获取top k
if len(rov_recall_rank) > 0:
rank_result.extend(rov_recall_rank[:top_K])
rov_recall_rank = rov_recall_rank[top_K:]
else:
rank_result.extend(flow_recall_rank[:top_K])
flow_recall_rank = flow_recall_rank[top_K:]
flow_num = 0
flowConfig =0
if exp_config and exp_config['flowConfig']:
flowConfig = exp_config['flowConfig']
if flowConfig == 1 and len(rov_recall_rank) > 0:
rank_result.extend(rov_recall_rank[:top_K])
for recall_item in rank_result:
flow_recall_name = recall_item.get("flowPool", '')
if flow_recall_name is not None and flow_recall_name.find("#") > -1:
flow_num = flow_num + 1
all_recall_rank = rov_recall_rank + flow_recall_rank
if flow_num > 0:
rank_result.extend(all_recall_rank[:size - top_K])
return rank_result[:size], flow_num
else:
# 按概率 p 及score排序获取 size - k 个视频
i = 0
while i < size - top_K:
# 随机生成[0, 1)浮点数
rand = random.random()
# log_.info('rand: {}'.format(rand))
if rand < flow_pool_P:
if flow_recall_rank:
rank_result.append(flow_recall_rank[0])
flow_recall_rank.remove(flow_recall_rank[0])
else:
rank_result.extend(rov_recall_rank[:size - top_K - i])
return rank_result[:size], flow_num
else:
if rov_recall_rank:
rank_result.append(rov_recall_rank[0])
rov_recall_rank.remove(rov_recall_rank[0])
else:
rank_result.extend(flow_recall_rank[:size - top_K - i])
return rank_result[:size], flow_num
i += 1
else:
# 按概率 p 及score排序获取 size - k 个视频
i = 0
while i < size - top_K:
# 随机生成[0, 1)浮点数
rand = random.random()
# log_.info('rand: {}'.format(rand))
if rand < flow_pool_P:
if flow_recall_rank:
rank_result.append(flow_recall_rank[0])
flow_recall_rank.remove(flow_recall_rank[0])
else:
rank_result.extend(rov_recall_rank[:size - top_K - i])
return rank_result[:size], flow_num
else:
if rov_recall_rank:
rank_result.append(rov_recall_rank[0])
rov_recall_rank.remove(rov_recall_rank[0])
else:
rank_result.extend(flow_recall_rank[:size - top_K - i])
return rank_result[:size],flow_num
i += 1
return rank_result[:size], flow_num
if __name__ == '__main__':
d_test = [{'videoId': 10028734, 'rovScore': 99.977, 'pushFrom': 'recall_pool', 'abCode': 10000},
{'videoId': 1919925, 'rovScore': 99.974, 'pushFrom': 'recall_pool', 'abCode': 10000},
{'videoId': 9968118, 'rovScore': 99.972, 'pushFrom': 'recall_pool', 'abCode': 10000},
{'videoId': 9934863, 'rovScore': 99.971, 'pushFrom': 'recall_pool', 'abCode': 10000},
{'videoId': 10219869, 'flowPool': '1#1#1#1640830818883', 'rovScore': 82.21929728934731, 'pushFrom': 'flow_pool', 'abCode': 10000},
{'videoId': 10212814, 'flowPool': '1#1#1#1640759014984', 'rovScore': 81.26694187726412, 'pushFrom': 'flow_pool', 'abCode': 10000},
{'videoId': 10219437, 'flowPool': '1#1#1#1640827620520', 'rovScore': 81.21634156641908, 'pushFrom': 'flow_pool', 'abCode': 10000},
{'videoId': 1994050, 'rovScore': 99.97, 'pushFrom': 'recall_pool', 'abCode': 10000},
{'videoId': 9894474, 'rovScore': 99.969, 'pushFrom': 'recall_pool', 'abCode': 10000},
{'videoId': 10028081, 'rovScore': 99.966, 'pushFrom': 'recall_pool', 'abCode': 10000}]
res = video_rank_by_w_h_rate(videos=d_test)
for tmp in res:
print(tmp)