1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324 |
- import copy
- 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, flow_pool_recall_process=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
- """
- # add_flow_pool_recall_log
- if flow_pool_recall_process is None:
- flow_pool_recall_process = {}
- if not data['rov_pool_recall'] and not data['flow_pool_recall']:
- # add_flow_pool_recall_log
- return [], flow_pool_recall_process
- # 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)
- # 不分地域小时级规则更新数据
- rule_h_recall = [item for item in data['rov_pool_recall']
- if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_h_h']]
- rule_h_recall_rank = sorted(rule_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 + rule_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 = []
- # add_flow_pool_recall_log
- flow_pool_recall_process['recall_duplicate_res'] = {'rov_recall_rank': rov_recall_rank,
- 'flow_recall_rank': copy.deepcopy(flow_recall_rank)}
- # 从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()
- # add_flow_pool_recall_log
- flow_pool_recall_process['flow_pool_P'] = flow_pool_P
- flow_pool_recall_process[f'{i}_rand'] = rand
- # 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_pool_recall_process
- 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_pool_recall_process
- i += 1
- return rank_result[:size], flow_pool_recall_process
- 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,
- pre_str = "k_p2:"
- rov_recall_rank = data['rov_pool_recall']
- #print(rov_recall_rank)
- #call rank service
- #flag_call_service = 0
- sort_index = 0
- if exp_config and "sort_flag" in exp_config:
- sort_index = exp_config["sort_flag"]
- #print("sort_index:", sort_index)
- redisObj = RedisHelper()
- vidKeys = []
- rec_recall_item_list = []
- rec_recall_vid_list = []
- day_vidKeys = []
- hour_vidKeys = []
- pre_day_str = "v_ctr:"
- pre_hour_str = "v_hour_ctr:"
- for recall_item in data['rov_pool_recall']:
- try:
- vid = int(recall_item.get("videoId", 0))
- rec_recall_vid_list.append(vid)
- rec_recall_item_list.append(recall_item)
- vidKeys.append(pre_str + str(vid))
- day_vidKeys.append(pre_day_str+str(vid))
- hour_vidKeys.append(pre_hour_str+str(vid))
- except:
- continue
- video_scores = redisObj.get_batch_key(vidKeys)
- #print("video_scores:", video_scores)
- if (ab_code == 60066 or ab_code == 60069 or ab_code == 60070 or ab_code == 60071) and len(rec_recall_vid_list)>0:
- video_static_info = redisObj.get_batch_key(day_vidKeys)
- video_hour_static_info = redisObj.get_batch_key(hour_vidKeys)
- #print("env_dict:", env_dict)
- feature_dict = get_featurs(mid, data, size, top_K, flow_pool_P, rec_recall_vid_list,env_dict, video_static_info, video_hour_static_info)
- score_result = get_tf_serving_sores(feature_dict)
- #print("score_result:", score_result)
- if video_scores and len(video_scores)>0 and rec_recall_item_list and score_result and len(score_result) > 0\
- and len(score_result) == len(rec_recall_item_list) and len(video_scores)== len(score_result):
- for i in range(len(score_result)):
- try:
- if video_scores[i] is None and len(score_result[i])>0:
- return_score = 0.000000001
- # sore_index :10 = model score
- if sort_index == 10:
- total_score = score_result[i][0]
- else:
- total_score = return_score * score_result[i][0]
- rec_recall_item_list[i]['sort_score'] = total_score
- rec_recall_item_list[i]['base_rov_score'] = 0.0
- rec_recall_item_list[i]['share_score'] = return_score
- rec_recall_item_list[i]['model_score'] = score_result[i][0]
- else:
- video_score_str = json.loads(video_scores[i])
- # sore_index :10 = model score
- return_score = 0.000000001
- if sort_index == 10:
- total_score = score_result[i][0]
- else:
- if len(video_score_str)>= sort_index and len(video_score_str)>0:
- return_score = video_score_str[sort_index]
- total_score = return_score * score_result[i][0]
- #print("total_score:", total_score, " model score :", score_result[i][0], "return_score:",
- # return_score)
- rec_recall_item_list[i]['sort_score'] = total_score
- rec_recall_item_list[i]['base_rov_score'] = video_score_str[0]
- rec_recall_item_list[i]['share_score'] = return_score
- rec_recall_item_list[i]['model_score'] = score_result[i][0]
- except Exception as e:
- #print('exception: {}:', e)
- return_score = 0.000000001
- if sort_index == 10:
- total_score = 0.00000001
- else:
- total_score = return_score * 0.00000001
- rec_recall_item_list[i]['sort_score'] = total_score
- rec_recall_item_list[i]['base_rov_score'] = 0
- rec_recall_item_list[i]['share_score'] = return_score
- rec_recall_item_list[i]['model_score'] = 0.00000001
- rec_recall_item_list[i]['flag_call_service'] = 1
- rov_recall_rank = sorted(rec_recall_item_list, key=lambda k: k.get('sort_score', 0), reverse=True)
- else:
- rov_recall_rank = sup_rank(video_scores, rec_recall_item_list)
- else:
- if video_scores and len(rec_recall_item_list) > 0 and len(video_scores)>0:
- for i in range(len(video_scores)):
- try:
- if video_scores[i] is None:
- rec_recall_item_list[i]['sort_score'] = 0.0
- else:
- video_score_str = json.loads(video_scores[i])
- # print("video_score_str:", video_score_str)
- rec_recall_item_list[i]['sort_score'] = video_score_str[0]
- except Exception:
- rec_recall_item_list[i]['sort_score'] = 0.0
- rov_recall_rank = sorted(rec_recall_item_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 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 video_new_rank3(data, size, top_K, flow_pool_P, rank_key_prefix='rank:score1:', flow_pool_recall_process=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
- :param rank_key_prefix:
- :return: rank_result
- """
- redis_helper = RedisHelper()
- # add_flow_pool_recall_log
- if flow_pool_recall_process is None:
- flow_pool_recall_process = {}
- if not data['rov_pool_recall'] and not data['flow_pool_recall']:
- # add_flow_pool_recall_log
- return [], 0, flow_pool_recall_process
- # return [], 0
- rov_recall_rank = data['rov_pool_recall']
- vid_keys = []
- rec_recall_item_list = []
- rec_recall_vid_list = []
- for recall_item in data['rov_pool_recall']:
- try:
- vid = int(recall_item.get("videoId", 0))
- rec_recall_vid_list.append(vid)
- rec_recall_item_list.append(recall_item)
- vid_keys.append(f"{rank_key_prefix}{vid}")
- except:
- continue
- video_scores = redis_helper.get_batch_key(vid_keys)
- if video_scores and len(rec_recall_item_list) > 0 and len(rec_recall_item_list) == len(video_scores):
- for i in range(len(video_scores)):
- try:
- if video_scores[i] is None:
- rec_recall_item_list[i]['sort_score'] = 0.0
- else:
- rec_recall_item_list[i]['sort_score'] = float(video_scores[i])
- except Exception:
- rec_recall_item_list[i]['sort_score'] = 0.0
- rov_recall_rank = sorted(rec_recall_item_list, key=lambda k: k.get('sort_score', 0), reverse=True)
- 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 = []
- # add_flow_pool_recall_log
- flow_pool_recall_process['recall_duplicate_res'] = {'rov_recall_rank': rov_recall_rank,
- 'flow_recall_rank': copy.deepcopy(flow_recall_rank)}
- # 从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
- i = 0
- while i < size - top_K:
- # 随机生成[0, 1)浮点数
- rand = random.random()
- # add_flow_pool_recall_log
- flow_pool_recall_process['flow_pool_P'] = flow_pool_P
- flow_pool_recall_process[f'{i}_rand'] = rand
- # 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, flow_pool_recall_process
- 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, flow_pool_recall_process
- i += 1
- return rank_result[:size], flow_num, flow_pool_recall_process
- # 排序服务兜底
- def sup_rank(video_scores, 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)
|