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
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 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]
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)