video_rank.py 59 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324
  1. import copy
  2. import json
  3. import random
  4. import numpy
  5. from log import Log
  6. from config import set_config
  7. from video_recall import PoolRecall
  8. from db_helper import RedisHelper
  9. from utils import FilterVideos, send_msg_to_feishu
  10. from rank_service import get_featurs, get_tf_serving_sores
  11. log_ = Log()
  12. config_ = set_config()
  13. def video_rank(data, size, top_K, flow_pool_P, flow_pool_recall_process=None):
  14. """
  15. 视频分发排序
  16. :param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []}
  17. :param size: 请求数
  18. :param top_K: 保证topK为召回池视频 type-int
  19. :param flow_pool_P: size-top_K视频为流量池视频的概率 type-float
  20. :return: rank_result
  21. """
  22. # add_flow_pool_recall_log
  23. if flow_pool_recall_process is None:
  24. flow_pool_recall_process = {}
  25. if not data['rov_pool_recall'] and not data['flow_pool_recall']:
  26. # add_flow_pool_recall_log
  27. return [], flow_pool_recall_process
  28. # return []
  29. # 将各路召回的视频按照score从大到小排序
  30. # 最惊奇相关推荐相似视频
  31. # relevant_recall = [item for item in data['rov_pool_recall']
  32. # if item.get('pushFrom') == config_.PUSH_FROM['top_video_relevant_appType_19']]
  33. # relevant_recall_rank = sorted(relevant_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  34. # 最惊奇完整影视视频
  35. # whole_movies_recall = [item for item in data['rov_pool_recall']
  36. # if item.get('pushFrom') == config_.PUSH_FROM['whole_movies']]
  37. # whole_movies_recall_rank = sorted(whole_movies_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  38. # 最惊奇影视解说视频
  39. # talk_videos_recall = [item for item in data['rov_pool_recall']
  40. # if item.get('pushFrom') == config_.PUSH_FROM['talk_videos']]
  41. # talk_videos_recall_rank = sorted(talk_videos_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  42. # 小时级更新数据
  43. # h_recall = [item for item in data['rov_pool_recall'] if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_h']]
  44. # h_recall_rank = sorted(h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  45. # 相对30天天级规则更新数据
  46. day_30_recall = [item for item in data['rov_pool_recall']
  47. if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_30day']]
  48. day_30_recall_rank = sorted(day_30_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  49. # 地域分组小时级规则更新数据
  50. region_h_recall = [item for item in data['rov_pool_recall']
  51. if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_region_h']]
  52. region_h_recall_rank = sorted(region_h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  53. # 不分地域小时级规则更新数据
  54. rule_h_recall = [item for item in data['rov_pool_recall']
  55. if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_h_h']]
  56. rule_h_recall_rank = sorted(rule_h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  57. # 地域分组小时级更新24h规则更新数据
  58. region_24h_recall = [item for item in data['rov_pool_recall']
  59. if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_region_24h']]
  60. region_24h_recall_rank = sorted(region_24h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  61. # 地域分组天级规则更新数据
  62. # region_day_recall = [item for item in data['rov_pool_recall']
  63. # if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_region_day']]
  64. # region_day_recall_rank = sorted(region_day_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  65. # 相对24h规则更新数据
  66. rule_24h_recall = [item for item in data['rov_pool_recall']
  67. if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_24h']]
  68. rule_24h_recall_rank = sorted(rule_24h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  69. # 相对24h规则筛选后剩余更新数据
  70. rule_24h_dup_recall = [item for item in data['rov_pool_recall']
  71. if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_24h_dup']]
  72. rule_24h_dup_recall_rank = sorted(rule_24h_dup_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  73. # 相对48h规则更新数据
  74. rule_48h_recall = [item for item in data['rov_pool_recall']
  75. if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_48h']]
  76. rule_48h_recall_rank = sorted(rule_48h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  77. # 相对48h规则筛选后剩余更新数据
  78. rule_48h_dup_recall = [item for item in data['rov_pool_recall']
  79. if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_48h_dup']]
  80. rule_48h_dup_recall_rank = sorted(rule_48h_dup_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  81. # 天级规则更新数据
  82. # day_recall = [item for item in data['rov_pool_recall'] if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_day']]
  83. # day_recall_rank = sorted(day_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  84. # ROV召回池
  85. # rov_initial_recall = [
  86. # item for item in data['rov_pool_recall']
  87. # if item.get('pushFrom') not in
  88. # [config_.PUSH_FROM['top_video_relevant_appType_19'],
  89. # config_.PUSH_FROM['rov_recall_h'],
  90. # config_.PUSH_FROM['rov_recall_region_h'],
  91. # config_.PUSH_FROM['rov_recall_region_24h'],
  92. # config_.PUSH_FROM['rov_recall_region_day'],
  93. # config_.PUSH_FROM['rov_recall_24h'],
  94. # config_.PUSH_FROM['rov_recall_24h_dup'],
  95. # config_.PUSH_FROM['rov_recall_48h'],
  96. # config_.PUSH_FROM['rov_recall_48h_dup'],
  97. # config_.PUSH_FROM['rov_recall_day'],
  98. # config_.PUSH_FROM['whole_movies'],
  99. # config_.PUSH_FROM['talk_videos']]
  100. # ]
  101. # rov_initial_recall_rank = sorted(rov_initial_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  102. # rov_recall_rank = whole_movies_recall_rank + talk_videos_recall_rank + h_recall_rank + \
  103. # day_30_recall_rank + region_h_recall_rank + region_24h_recall_rank + \
  104. # region_day_recall_rank + rule_24h_recall_rank + rule_24h_dup_recall_rank + \
  105. # rule_48h_recall_rank + rule_48h_dup_recall_rank + \
  106. # day_recall_rank + rov_initial_recall_rank
  107. rov_recall_rank = day_30_recall_rank + \
  108. region_h_recall_rank + rule_h_recall_rank + region_24h_recall_rank + \
  109. rule_24h_recall_rank + rule_24h_dup_recall_rank + \
  110. rule_48h_recall_rank + rule_48h_dup_recall_rank
  111. # 流量池
  112. flow_recall_rank = sorted(data['flow_pool_recall'], key=lambda k: k.get('rovScore', 0), reverse=True)
  113. # 对各路召回的视频进行去重
  114. rov_recall_rank, flow_recall_rank = remove_duplicate(rov_recall=rov_recall_rank, flow_recall=flow_recall_rank,
  115. top_K=top_K)
  116. # log_.info('remove_duplicate finished! rov_recall_rank = {}, flow_recall_rank = {}'.format(
  117. # rov_recall_rank, flow_recall_rank))
  118. # rank_result = relevant_recall_rank
  119. rank_result = []
  120. # add_flow_pool_recall_log
  121. flow_pool_recall_process['recall_duplicate_res'] = {'rov_recall_rank': rov_recall_rank,
  122. 'flow_recall_rank': copy.deepcopy(flow_recall_rank)}
  123. # 从ROV召回池中获取top k
  124. if len(rov_recall_rank) > 0:
  125. rank_result.extend(rov_recall_rank[:top_K])
  126. rov_recall_rank = rov_recall_rank[top_K:]
  127. else:
  128. rank_result.extend(flow_recall_rank[:top_K])
  129. flow_recall_rank = flow_recall_rank[top_K:]
  130. # 按概率 p 及score排序获取 size - k 个视频
  131. i = 0
  132. while i < size - top_K:
  133. # 随机生成[0, 1)浮点数
  134. rand = random.random()
  135. # add_flow_pool_recall_log
  136. flow_pool_recall_process['flow_pool_P'] = flow_pool_P
  137. flow_pool_recall_process[f'{i}_rand'] = rand
  138. # log_.info('rand: {}'.format(rand))
  139. if rand < flow_pool_P:
  140. if flow_recall_rank:
  141. rank_result.append(flow_recall_rank[0])
  142. flow_recall_rank.remove(flow_recall_rank[0])
  143. else:
  144. rank_result.extend(rov_recall_rank[:size - top_K - i])
  145. return rank_result[:size], flow_pool_recall_process
  146. else:
  147. if rov_recall_rank:
  148. rank_result.append(rov_recall_rank[0])
  149. rov_recall_rank.remove(rov_recall_rank[0])
  150. else:
  151. rank_result.extend(flow_recall_rank[:size - top_K - i])
  152. return rank_result[:size], flow_pool_recall_process
  153. i += 1
  154. return rank_result[:size], flow_pool_recall_process
  155. def video_new_rank(videoIds, fast_flow_set, flow_set, size, top_K, flow_pool_P):
  156. """
  157. 视频分发排序
  158. :param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []}
  159. :param size: 请求数
  160. :param top_K: 保证topK为召回池视频 type-int
  161. :param flow_pool_P: size-top_K视频为流量池视频的概率 type-float
  162. :return: rank_result
  163. """
  164. add_flow_set = set('')
  165. if not videoIds or len(videoIds)==0:
  166. return [], add_flow_set
  167. redisObj = RedisHelper()
  168. vidKeys = []
  169. for vid in videoIds:
  170. vidKeys.append("k_p:"+str(vid))
  171. #print("vidKeys:", vidKeys)
  172. video_scores = redisObj.get_batch_key(vidKeys)
  173. #print(video_scores)
  174. video_items = []
  175. for i in range(len(video_scores)):
  176. try:
  177. #print(video_scores[i])
  178. if video_scores[i] is None:
  179. video_items.append((videoIds[i], 0.0))
  180. else:
  181. video_score_str = json.loads(video_scores[i])
  182. #print("video_score_str:",video_score_str)
  183. video_items.append((videoIds[i], video_score_str[0]))
  184. except Exception:
  185. video_items.append((videoIds[i], 0.0))
  186. sort_items = sorted(video_items, key=lambda k: k[1], reverse=True)
  187. #print("sort_items:", sort_items)
  188. rov_recall_rank = sort_items
  189. fast_flow_recall_rank = []
  190. flow_recall_rank = []
  191. for item in sort_items:
  192. if item[0] in fast_flow_set:
  193. fast_flow_recall_rank.append(item)
  194. elif item[0] in flow_set:
  195. flow_recall_rank.append(item)
  196. # all flow result
  197. all_flow_recall_rank = fast_flow_recall_rank+flow_recall_rank
  198. rank_result = []
  199. rank_set = set('')
  200. # 从ROV召回池中获取top k
  201. if len(rov_recall_rank) > 0:
  202. rank_result.extend(rov_recall_rank[:top_K])
  203. rov_recall_rank = rov_recall_rank[top_K:]
  204. else:
  205. rank_result.extend(all_flow_recall_rank[:top_K])
  206. all_flow_recall_rank = all_flow_recall_rank[top_K:]
  207. for rank_item in rank_result:
  208. rank_set.add(rank_item[0])
  209. #print("rank_result:", rank_result)
  210. # 按概率 p 及score排序获取 size - k 个视频, 第4个位置按概率取流量池
  211. i = 0
  212. left_quato = size - top_K
  213. j = 0
  214. jj = 0
  215. while i < left_quato and (j<len(all_flow_recall_rank) or jj<len(rov_recall_rank)):
  216. # 随机生成[0, 1)浮点数
  217. rand = random.random()
  218. # log_.info('rand: {}'.format(rand))
  219. if rand < flow_pool_P:
  220. for flow_item in all_flow_recall_rank:
  221. j+=1
  222. if flow_item[0] in rank_set:
  223. continue
  224. else:
  225. rank_result.append(flow_item)
  226. rank_set.add(flow_item[0])
  227. add_flow_set.add(flow_item[0])
  228. i += 1
  229. if i>= left_quato:
  230. break
  231. else:
  232. for recall_item in rov_recall_rank:
  233. jj+=1
  234. if recall_item[0] in rank_set:
  235. continue
  236. else:
  237. rank_result.append(recall_item)
  238. rank_set.add(recall_item[0])
  239. i += 1
  240. if i>= left_quato:
  241. break
  242. #print("rank_result:", rank_result)
  243. #print("add_flow_set:", add_flow_set)
  244. return rank_result[:size], add_flow_set
  245. def refactor_video_rank(rov_recall_rank, fast_flow_set, flow_set, size, top_K, flow_pool_P):
  246. """
  247. 视频分发排序
  248. :param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []}
  249. :param size: 请求数
  250. :param top_K: 保证topK为召回池视频 type-int
  251. :param flow_pool_P: size-top_K视频为流量池视频的概率 type-float
  252. :return: rank_result
  253. """
  254. if not rov_recall_rank or len(rov_recall_rank) == 0:
  255. return []
  256. fast_flow_recall_rank = []
  257. flow_recall_rank = []
  258. for item in rov_recall_rank:
  259. vid = item.get('videoId', 0)
  260. #print(item)
  261. if vid in fast_flow_set:
  262. fast_flow_recall_rank.append(item)
  263. elif vid in flow_set:
  264. flow_recall_rank.append(item)
  265. # all flow result
  266. all_flow_recall_rank = fast_flow_recall_rank + flow_recall_rank
  267. rank_result = []
  268. rank_set = set('')
  269. # 从ROV召回池中获取top k
  270. if len(rov_recall_rank) > 0:
  271. rank_result.extend(rov_recall_rank[:top_K])
  272. rov_recall_rank = rov_recall_rank[top_K:]
  273. else:
  274. rank_result.extend(all_flow_recall_rank[:top_K])
  275. all_flow_recall_rank = all_flow_recall_rank[top_K:]
  276. #已存放了多少VID
  277. for rank_item in rank_result:
  278. rank_set.add(rank_item.get('videoId', 0))
  279. # 按概率 p 及score排序获取 size - k 个视频, 第4个位置按概率取流量池
  280. i = 0
  281. while i < size - top_K:
  282. # 随机生成[0, 1)浮点数
  283. rand = random.random()
  284. # log_.info('rand: {}'.format(rand))
  285. if rand < flow_pool_P:
  286. for flow_item in all_flow_recall_rank:
  287. flow_vid = flow_item.get('videoId', 0)
  288. if flow_vid in rank_set:
  289. continue
  290. else:
  291. rank_result.append(flow_item)
  292. rank_set.add(flow_vid)
  293. else:
  294. for recall_item in rov_recall_rank:
  295. flow_vid = recall_item.get('videoId', 0)
  296. if flow_vid in rank_set:
  297. continue
  298. else:
  299. rank_result.append(recall_item)
  300. rank_set.add(flow_vid)
  301. i += 1
  302. return rank_result[:size]
  303. def remove_duplicate(rov_recall, flow_recall, top_K):
  304. """
  305. 对多路召回的视频去重
  306. 去重原则:
  307. 如果视频在ROV召回池topK,则保留ROV召回池,否则保留流量池
  308. :param rov_recall: ROV召回池-已排序
  309. :param flow_recall: 流量池-已排序
  310. :param top_K: 保证topK为召回池视频 type-int
  311. :return:
  312. """
  313. flow_recall_result = []
  314. rov_recall_remove = []
  315. flow_recall_video_ids = [item['videoId'] for item in flow_recall]
  316. # rov_recall topK
  317. for item in rov_recall[:top_K]:
  318. if item['videoId'] in flow_recall_video_ids:
  319. flow_recall_video_ids.remove(item['videoId'])
  320. # other
  321. for item in rov_recall[top_K:]:
  322. if item['videoId'] in flow_recall_video_ids:
  323. rov_recall_remove.append(item)
  324. # rov recall remove
  325. for item in rov_recall_remove:
  326. rov_recall.remove(item)
  327. # flow recall remove
  328. for item in flow_recall:
  329. if item['videoId'] in flow_recall_video_ids:
  330. flow_recall_result.append(item)
  331. return rov_recall, flow_recall_result
  332. def bottom_strategy(request_id, size, app_type, ab_code, params):
  333. """
  334. 兜底策略: 从ROV召回池中获取top1000,进行状态过滤后的视频
  335. :param request_id: request_id
  336. :param size: 需要获取的视频数
  337. :param app_type: 产品标识 type-int
  338. :param ab_code: abCode
  339. :param params:
  340. :return:
  341. """
  342. pool_recall = PoolRecall(request_id=request_id, app_type=app_type, ab_code=ab_code)
  343. key_name, _ = pool_recall.get_pool_redis_key(pool_type='rov')
  344. redis_helper = RedisHelper(params=params)
  345. data = redis_helper.get_data_zset_with_index(key_name=key_name, start=0, end=1000)
  346. if not data:
  347. log_.info('{} —— ROV推荐进入了二次兜底, data = {}'.format(config_.ENV_TEXT, data))
  348. send_msg_to_feishu('{} —— ROV推荐进入了二次兜底,请查看是否有数据更新失败问题。'.format(config_.ENV_TEXT))
  349. # 二次兜底
  350. bottom_data = bottom_strategy_last(size=size, app_type=app_type, ab_code=ab_code, params=params)
  351. return bottom_data
  352. # 视频状态过滤采用离线定时过滤方案
  353. # 状态过滤
  354. # filter_videos = FilterVideos(app_type=app_type, video_ids=data)
  355. # filtered_data = filter_videos.filter_video_status(video_ids=data)
  356. if len(data) > size:
  357. random_data = numpy.random.choice(data, size, False)
  358. else:
  359. random_data = data
  360. bottom_data = [{'videoId': int(item), 'pushFrom': config_.PUSH_FROM['bottom'], 'abCode': ab_code}
  361. for item in random_data]
  362. return bottom_data
  363. def bottom_strategy_last(size, app_type, ab_code, params):
  364. """
  365. 兜底策略: 从兜底视频中随机获取视频,进行状态过滤后的视频
  366. :param size: 需要获取的视频数
  367. :param app_type: 产品标识 type-int
  368. :param ab_code: abCode
  369. :param params:
  370. :return:
  371. """
  372. redis_helper = RedisHelper(params=params)
  373. bottom_data = redis_helper.get_data_zset_with_index(key_name=config_.BOTTOM_KEY_NAME, start=0, end=-1)
  374. random_data = numpy.random.choice(bottom_data, size * 30, False)
  375. # 视频状态过滤采用离线定时过滤方案
  376. # 状态过滤
  377. # filter_videos = FilterVideos(app_type=app_type, video_ids=random_data)
  378. # filtered_data = filter_videos.filter_video_status(video_ids=random_data)
  379. bottom_data = [{'videoId': int(video_id), 'pushFrom': config_.PUSH_FROM['bottom_last'], 'abCode': ab_code}
  380. for video_id in random_data[:size]]
  381. return bottom_data
  382. def bottom_strategy2(size, app_type, mid, uid, ab_code, client_info, params):
  383. """
  384. 兜底策略: 从兜底视频中随机获取视频,进行过滤后的视频
  385. :param size: 需要获取的视频数
  386. :param app_type: 产品标识 type-int
  387. :param mid: mid
  388. :param uid: uid
  389. :param ab_code: abCode
  390. :param client_info: 地域信息
  391. :param params:
  392. :return:
  393. """
  394. # 获取存在城市分组数据的城市编码列表
  395. city_code_list = [code for _, code in config_.CITY_CODE.items()]
  396. # 获取provinceCode
  397. province_code = client_info.get('provinceCode', '-1')
  398. # 获取cityCode
  399. city_code = client_info.get('cityCode', '-1')
  400. if city_code in city_code_list:
  401. # 分城市数据存在时,获取城市分组数据
  402. region_code = city_code
  403. else:
  404. region_code = province_code
  405. if region_code == '':
  406. region_code = '-1'
  407. redis_helper = RedisHelper(params=params)
  408. bottom_data = redis_helper.get_data_from_set(key_name=config_.BOTTOM2_KEY_NAME)
  409. bottom_result = []
  410. if bottom_data is None:
  411. return bottom_result
  412. if len(bottom_data) > 0:
  413. try:
  414. random_data = numpy.random.choice(bottom_data, size * 5, False)
  415. except Exception as e:
  416. random_data = bottom_data
  417. video_ids = [int(item) for item in random_data]
  418. # 过滤
  419. filter_ = FilterVideos(request_id=params.request_id, app_type=app_type, mid=mid, uid=uid, video_ids=video_ids)
  420. filtered_data = filter_.filter_videos(pool_type='flow', region_code=region_code)
  421. if filtered_data:
  422. bottom_result = [{'videoId': int(video_id), 'pushFrom': config_.PUSH_FROM['bottom2'], 'abCode': ab_code}
  423. for video_id in filtered_data[:size]]
  424. return bottom_result
  425. def video_rank_by_w_h_rate(videos):
  426. """
  427. 视频宽高比实验(每组的前两个视频调整为横屏视频),根据视频宽高比信息对视频进行重排
  428. :param videos:
  429. :return:
  430. """
  431. redis_helper = RedisHelper()
  432. # ##### 判断前两个视频是否是置顶视频 或者 流量池视频
  433. top_2_push_from_flag = [False, False]
  434. for i, video in enumerate(videos[:2]):
  435. if video['pushFrom'] in [config_.PUSH_FROM['top'], config_.PUSH_FROM['flow_recall']]:
  436. top_2_push_from_flag[i] = True
  437. if top_2_push_from_flag[0] and top_2_push_from_flag[1]:
  438. return videos
  439. # ##### 判断前两个视频是否为横屏
  440. top_2_w_h_rate_flag = [False, False]
  441. for i, video in enumerate(videos[:2]):
  442. if video['pushFrom'] in [config_.PUSH_FROM['top'], config_.PUSH_FROM['flow_recall']]:
  443. # 视频来源为置顶 或 流量池时,不做判断
  444. top_2_w_h_rate_flag[i] = True
  445. elif video['pushFrom'] in [config_.PUSH_FROM['rov_recall'], config_.PUSH_FROM['bottom']]:
  446. # 视频来源为 rov召回池 或 一层兜底时,判断是否是横屏
  447. w_h_rate = redis_helper.get_score_with_value(
  448. key_name=config_.W_H_RATE_UP_1_VIDEO_LIST_KEY_NAME['rov_recall'], value=video['videoId'])
  449. if w_h_rate is not None:
  450. top_2_w_h_rate_flag[i] = True
  451. elif video['pushFrom'] == config_.PUSH_FROM['bottom_last']:
  452. # 视频来源为 二层兜底时,判断是否是横屏
  453. w_h_rate = redis_helper.get_score_with_value(
  454. key_name=config_.W_H_RATE_UP_1_VIDEO_LIST_KEY_NAME['bottom_last'], value=video['videoId'])
  455. if w_h_rate is not None:
  456. top_2_w_h_rate_flag[i] = True
  457. if top_2_w_h_rate_flag[0] and top_2_w_h_rate_flag[1]:
  458. return videos
  459. # ##### 前两个视频中有不符合前面两者条件的,对视频进行位置调整
  460. # 记录横屏视频位置
  461. horizontal_video_index = []
  462. # 记录流量池视频位置
  463. flow_video_index = []
  464. # 记录置顶视频位置
  465. top_video_index = []
  466. for i, video in enumerate(videos):
  467. # 视频来源为置顶
  468. if video['pushFrom'] == config_.PUSH_FROM['top']:
  469. top_video_index.append(i)
  470. # 视频来源为流量池
  471. elif video['pushFrom'] == config_.PUSH_FROM['flow_recall']:
  472. flow_video_index.append(i)
  473. # 视频来源为rov召回池 或 一层兜底
  474. elif video['pushFrom'] in [config_.PUSH_FROM['rov_recall'], config_.PUSH_FROM['bottom']]:
  475. w_h_rate = redis_helper.get_score_with_value(
  476. key_name=config_.W_H_RATE_UP_1_VIDEO_LIST_KEY_NAME['rov_recall'], value=video['videoId'])
  477. if w_h_rate is not None:
  478. horizontal_video_index.append(i)
  479. else:
  480. continue
  481. # 视频来源为 二层兜底
  482. elif video['pushFrom'] == config_.PUSH_FROM['bottom_last']:
  483. w_h_rate = redis_helper.get_score_with_value(
  484. key_name=config_.W_H_RATE_UP_1_VIDEO_LIST_KEY_NAME['bottom_last'], value=video['videoId'])
  485. if w_h_rate is not None:
  486. horizontal_video_index.append(i)
  487. else:
  488. continue
  489. # 重新排序
  490. top2_index = []
  491. for i in range(2):
  492. if i in top_video_index:
  493. top2_index.append(i)
  494. elif i in flow_video_index:
  495. top2_index.append(i)
  496. flow_video_index.remove(i)
  497. elif i in horizontal_video_index:
  498. top2_index.append(i)
  499. horizontal_video_index.remove(i)
  500. elif len(horizontal_video_index) > 0:
  501. # 调整横屏视频到第一位
  502. top2_index.append(horizontal_video_index[0])
  503. # 从横屏位置记录中移除
  504. horizontal_video_index.pop(0)
  505. elif i == 0:
  506. return videos
  507. # 重排
  508. flow_result = [videos[i] for i in flow_video_index]
  509. other_result = [videos[i] for i in range(len(videos)) if i not in top2_index and i not in flow_video_index]
  510. top2_result = []
  511. for i, j in enumerate(top2_index):
  512. item = videos[j]
  513. if i != j:
  514. # 修改abCode
  515. item['abCode'] = config_.AB_CODE['w_h_rate']
  516. top2_result.append(item)
  517. new_rank_result = top2_result
  518. for i in range(len(top2_index), len(videos)):
  519. if i in flow_video_index:
  520. new_rank_result.append(flow_result[0])
  521. flow_result.pop(0)
  522. else:
  523. new_rank_result.append(other_result[0])
  524. other_result.pop(0)
  525. return new_rank_result
  526. def video_rank_with_old_video(rank_result, old_video_recall, size, top_K, old_video_index=2):
  527. """
  528. 视频分发排序 - 包含老视频, 老视频插入固定位置
  529. :param rank_result: 排序后的结果
  530. :param size: 请求数
  531. :param old_video_index: 老视频插入的位置索引,默认为2
  532. :return: new_rank_result
  533. """
  534. if not old_video_recall:
  535. return rank_result
  536. if not rank_result:
  537. return old_video_recall[:size]
  538. # 视频去重
  539. rank_video_ids = [item['videoId'] for item in rank_result]
  540. old_video_remove = []
  541. for old_video in old_video_recall:
  542. if old_video['videoId'] in rank_video_ids:
  543. old_video_remove.append(old_video)
  544. for item in old_video_remove:
  545. old_video_recall.remove(item)
  546. if not old_video_recall:
  547. return rank_result
  548. # 插入老视频
  549. # 随机获取一个视频
  550. ind = random.randint(0, len(old_video_recall) - 1)
  551. old_video = old_video_recall[ind]
  552. # 插入
  553. if len(rank_result) < top_K:
  554. new_rank_result = rank_result + [old_video]
  555. else:
  556. new_rank_result = rank_result[:old_video_index] + [old_video] + rank_result[old_video_index:]
  557. if len(new_rank_result) > size:
  558. # 判断后两位视频来源
  559. push_from_1 = new_rank_result[-1]['pushFrom']
  560. push_from_2 = new_rank_result[-2]['pushFrom']
  561. if push_from_2 == config_.PUSH_FROM['rov_recall'] and push_from_1 == config_.PUSH_FROM['flow_recall']:
  562. new_rank_result = new_rank_result[:-2] + new_rank_result[-1:]
  563. return new_rank_result[:size]
  564. def video_new_rank2(data, size, top_K, flow_pool_P, ab_code, mid, exp_config=None, env_dict=None):
  565. """
  566. 视频分发排序
  567. :param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []}
  568. :param size: 请求数
  569. :param top_K: 保证topK为召回池视频 type-int
  570. :param flow_pool_P: size-top_K视频为流量池视频的概率 type-float
  571. :return: rank_result
  572. """
  573. if not data['rov_pool_recall'] and not data['flow_pool_recall']:
  574. return [], 0
  575. #全量的是vlog,票圈精选, 334,60057,
  576. # 60054: simrecall,
  577. pre_str = "k_p2:"
  578. rov_recall_rank = data['rov_pool_recall']
  579. #print(rov_recall_rank)
  580. #call rank service
  581. #flag_call_service = 0
  582. sort_index = 0
  583. if exp_config and "sort_flag" in exp_config:
  584. sort_index = exp_config["sort_flag"]
  585. #print("sort_index:", sort_index)
  586. redisObj = RedisHelper()
  587. vidKeys = []
  588. rec_recall_item_list = []
  589. rec_recall_vid_list = []
  590. day_vidKeys = []
  591. hour_vidKeys = []
  592. pre_day_str = "v_ctr:"
  593. pre_hour_str = "v_hour_ctr:"
  594. for recall_item in data['rov_pool_recall']:
  595. try:
  596. vid = int(recall_item.get("videoId", 0))
  597. rec_recall_vid_list.append(vid)
  598. rec_recall_item_list.append(recall_item)
  599. vidKeys.append(pre_str + str(vid))
  600. day_vidKeys.append(pre_day_str+str(vid))
  601. hour_vidKeys.append(pre_hour_str+str(vid))
  602. except:
  603. continue
  604. video_scores = redisObj.get_batch_key(vidKeys)
  605. #print("video_scores:", video_scores)
  606. if (ab_code == 60066 or ab_code == 60069 or ab_code == 60070 or ab_code == 60071) and len(rec_recall_vid_list)>0:
  607. video_static_info = redisObj.get_batch_key(day_vidKeys)
  608. video_hour_static_info = redisObj.get_batch_key(hour_vidKeys)
  609. #print("env_dict:", env_dict)
  610. 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)
  611. score_result = get_tf_serving_sores(feature_dict)
  612. #print("score_result:", score_result)
  613. if video_scores and len(video_scores)>0 and rec_recall_item_list and score_result and len(score_result) > 0\
  614. and len(score_result) == len(rec_recall_item_list) and len(video_scores)== len(score_result):
  615. for i in range(len(score_result)):
  616. try:
  617. if video_scores[i] is None and len(score_result[i])>0:
  618. return_score = 0.000000001
  619. # sore_index :10 = model score
  620. if sort_index == 10:
  621. total_score = score_result[i][0]
  622. else:
  623. total_score = return_score * score_result[i][0]
  624. rec_recall_item_list[i]['sort_score'] = total_score
  625. rec_recall_item_list[i]['base_rov_score'] = 0.0
  626. rec_recall_item_list[i]['share_score'] = return_score
  627. rec_recall_item_list[i]['model_score'] = score_result[i][0]
  628. else:
  629. video_score_str = json.loads(video_scores[i])
  630. # sore_index :10 = model score
  631. return_score = 0.000000001
  632. if sort_index == 10:
  633. total_score = score_result[i][0]
  634. else:
  635. if len(video_score_str)>= sort_index and len(video_score_str)>0:
  636. return_score = video_score_str[sort_index]
  637. total_score = return_score * score_result[i][0]
  638. #print("total_score:", total_score, " model score :", score_result[i][0], "return_score:",
  639. # return_score)
  640. rec_recall_item_list[i]['sort_score'] = total_score
  641. rec_recall_item_list[i]['base_rov_score'] = video_score_str[0]
  642. rec_recall_item_list[i]['share_score'] = return_score
  643. rec_recall_item_list[i]['model_score'] = score_result[i][0]
  644. except Exception as e:
  645. #print('exception: {}:', e)
  646. return_score = 0.000000001
  647. if sort_index == 10:
  648. total_score = 0.00000001
  649. else:
  650. total_score = return_score * 0.00000001
  651. rec_recall_item_list[i]['sort_score'] = total_score
  652. rec_recall_item_list[i]['base_rov_score'] = 0
  653. rec_recall_item_list[i]['share_score'] = return_score
  654. rec_recall_item_list[i]['model_score'] = 0.00000001
  655. rec_recall_item_list[i]['flag_call_service'] = 1
  656. rov_recall_rank = sorted(rec_recall_item_list, key=lambda k: k.get('sort_score', 0), reverse=True)
  657. else:
  658. rov_recall_rank = sup_rank(video_scores, rec_recall_item_list)
  659. else:
  660. if video_scores and len(rec_recall_item_list) > 0 and len(video_scores)>0:
  661. for i in range(len(video_scores)):
  662. try:
  663. if video_scores[i] is None:
  664. rec_recall_item_list[i]['sort_score'] = 0.0
  665. else:
  666. video_score_str = json.loads(video_scores[i])
  667. # print("video_score_str:", video_score_str)
  668. rec_recall_item_list[i]['sort_score'] = video_score_str[0]
  669. except Exception:
  670. rec_recall_item_list[i]['sort_score'] = 0.0
  671. rov_recall_rank = sorted(rec_recall_item_list, key=lambda k: k.get('sort_score', 0), reverse=True)
  672. #print(rov_recall_rank)
  673. flow_recall_rank = sorted(data['flow_pool_recall'], key=lambda k: k.get('rovScore', 0), reverse=True)
  674. rov_recall_rank, flow_recall_rank = remove_duplicate(rov_recall=rov_recall_rank, flow_recall=flow_recall_rank,
  675. top_K=top_K)
  676. rank_result = []
  677. rank_set = set('')
  678. # 从ROV召回池中获取top k
  679. if len(rov_recall_rank) > 0:
  680. rank_result.extend(rov_recall_rank[:top_K])
  681. rov_recall_rank = rov_recall_rank[top_K:]
  682. else:
  683. rank_result.extend(flow_recall_rank[:top_K])
  684. flow_recall_rank = flow_recall_rank[top_K:]
  685. # 按概率 p 及score排序获取 size - k 个视频
  686. flow_num = 0
  687. flowConfig = 0
  688. # 本段代码控制流量池,通过实验传参,现不动
  689. if flowConfig == 1 and len(rov_recall_rank) > 0:
  690. for recall_item in rank_result:
  691. flow_recall_name = recall_item.get("flowPool", '')
  692. flow_num = flow_num + 1
  693. all_recall_rank = rov_recall_rank + flow_recall_rank
  694. if flow_num > 0:
  695. rank_result.extend(all_recall_rank[:size - top_K])
  696. return rank_result, flow_num
  697. else:
  698. i = 0
  699. while i < size - top_K:
  700. # 随机生成[0, 1)浮点数
  701. rand = random.random()
  702. # log_.info('rand: {}'.format(rand))
  703. if rand < flow_pool_P:
  704. if flow_recall_rank:
  705. rank_result.append(flow_recall_rank[0])
  706. flow_recall_rank.remove(flow_recall_rank[0])
  707. else:
  708. rank_result.extend(rov_recall_rank[:size - top_K - i])
  709. return rank_result[:size], flow_num
  710. else:
  711. if rov_recall_rank:
  712. rank_result.append(rov_recall_rank[0])
  713. rov_recall_rank.remove(rov_recall_rank[0])
  714. else:
  715. rank_result.extend(flow_recall_rank[:size - top_K - i])
  716. return rank_result[:size], flow_num
  717. i += 1
  718. else:
  719. i = 0
  720. while i < size - top_K:
  721. # 随机生成[0, 1)浮点数
  722. rand = random.random()
  723. # log_.info('rand: {}'.format(rand))
  724. if rand < flow_pool_P:
  725. if flow_recall_rank:
  726. rank_result.append(flow_recall_rank[0])
  727. flow_recall_rank.remove(flow_recall_rank[0])
  728. else:
  729. rank_result.extend(rov_recall_rank[:size - top_K - i])
  730. return rank_result[:size], flow_num
  731. else:
  732. if rov_recall_rank:
  733. rank_result.append(rov_recall_rank[0])
  734. rov_recall_rank.remove(rov_recall_rank[0])
  735. else:
  736. rank_result.extend(flow_recall_rank[:size - top_K - i])
  737. return rank_result[:size], flow_num
  738. i += 1
  739. return rank_result[:size], flow_num
  740. def video_new_rank3(data, size, top_K, flow_pool_P, rank_key_prefix='rank:score1:', flow_pool_recall_process=None):
  741. """
  742. 视频分发排序
  743. :param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []}
  744. :param size: 请求数
  745. :param top_K: 保证topK为召回池视频 type-int
  746. :param flow_pool_P: size-top_K视频为流量池视频的概率 type-float
  747. :param rank_key_prefix:
  748. :return: rank_result
  749. """
  750. redis_helper = RedisHelper()
  751. # add_flow_pool_recall_log
  752. if flow_pool_recall_process is None:
  753. flow_pool_recall_process = {}
  754. if not data['rov_pool_recall'] and not data['flow_pool_recall']:
  755. # add_flow_pool_recall_log
  756. return [], 0, flow_pool_recall_process
  757. # return [], 0
  758. rov_recall_rank = data['rov_pool_recall']
  759. vid_keys = []
  760. rec_recall_item_list = []
  761. rec_recall_vid_list = []
  762. for recall_item in data['rov_pool_recall']:
  763. try:
  764. vid = int(recall_item.get("videoId", 0))
  765. rec_recall_vid_list.append(vid)
  766. rec_recall_item_list.append(recall_item)
  767. vid_keys.append(f"{rank_key_prefix}{vid}")
  768. except:
  769. continue
  770. video_scores = redis_helper.get_batch_key(vid_keys)
  771. if video_scores and len(rec_recall_item_list) > 0 and len(rec_recall_item_list) == len(video_scores):
  772. for i in range(len(video_scores)):
  773. try:
  774. if video_scores[i] is None:
  775. rec_recall_item_list[i]['sort_score'] = 0.0
  776. else:
  777. rec_recall_item_list[i]['sort_score'] = float(video_scores[i])
  778. except Exception:
  779. rec_recall_item_list[i]['sort_score'] = 0.0
  780. rov_recall_rank = sorted(rec_recall_item_list, key=lambda k: k.get('sort_score', 0), reverse=True)
  781. flow_recall_rank = sorted(data['flow_pool_recall'], key=lambda k: k.get('rovScore', 0), reverse=True)
  782. rov_recall_rank, flow_recall_rank = remove_duplicate(
  783. rov_recall=rov_recall_rank, flow_recall=flow_recall_rank, top_K=top_K
  784. )
  785. rank_result = []
  786. # add_flow_pool_recall_log
  787. flow_pool_recall_process['recall_duplicate_res'] = {'rov_recall_rank': rov_recall_rank,
  788. 'flow_recall_rank': copy.deepcopy(flow_recall_rank)}
  789. # 从ROV召回池中获取top k
  790. if len(rov_recall_rank) > 0:
  791. rank_result.extend(rov_recall_rank[:top_K])
  792. rov_recall_rank = rov_recall_rank[top_K:]
  793. else:
  794. rank_result.extend(flow_recall_rank[:top_K])
  795. flow_recall_rank = flow_recall_rank[top_K:]
  796. # 按概率 p 及score排序获取 size - k 个视频
  797. flow_num = 0
  798. i = 0
  799. while i < size - top_K:
  800. # 随机生成[0, 1)浮点数
  801. rand = random.random()
  802. # add_flow_pool_recall_log
  803. flow_pool_recall_process['flow_pool_P'] = flow_pool_P
  804. flow_pool_recall_process[f'{i}_rand'] = rand
  805. # log_.info('rand: {}'.format(rand))
  806. if rand < flow_pool_P:
  807. if flow_recall_rank:
  808. rank_result.append(flow_recall_rank[0])
  809. flow_recall_rank.remove(flow_recall_rank[0])
  810. else:
  811. rank_result.extend(rov_recall_rank[:size - top_K - i])
  812. return rank_result[:size], flow_num, flow_pool_recall_process
  813. else:
  814. if rov_recall_rank:
  815. rank_result.append(rov_recall_rank[0])
  816. rov_recall_rank.remove(rov_recall_rank[0])
  817. else:
  818. rank_result.extend(flow_recall_rank[:size - top_K - i])
  819. return rank_result[:size], flow_num, flow_pool_recall_process
  820. i += 1
  821. return rank_result[:size], flow_num, flow_pool_recall_process
  822. # 排序服务兜底
  823. def sup_rank(video_scores, recall_list):
  824. if video_scores and len(recall_list) > 0:
  825. for i in range(len(video_scores)):
  826. try:
  827. if video_scores[i] is None:
  828. recall_list[i]['sort_score'] = 0.0
  829. else:
  830. video_score_str = json.loads(video_scores[i])
  831. recall_list[i]['flag_call_service'] = 0
  832. recall_list[i]['sort_score'] = video_score_str[0]
  833. except Exception:
  834. recall_list[i]['sort_score'] = 0.0
  835. rov_recall_rank = sorted(recall_list, key=lambda k: k.get('sort_score', 0), reverse=True)
  836. #print("rov_recall_rank:", rov_recall_rank)
  837. else:
  838. rov_recall_rank = recall_list
  839. return rov_recall_rank
  840. def video_sanke_rank(data, size, top_K, flow_pool_P, ab_Code='', exp_config=None):
  841. """
  842. 视频分发排序
  843. :param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []}
  844. :param size: 请求数
  845. :param top_K: 保证topK为召回池视频 type-int
  846. :param flow_pool_P: size-top_K视频为流量池视频的概率 type-float
  847. :return: rank_result
  848. """
  849. if not data['rov_pool_recall'] and not data['flow_pool_recall'] \
  850. and len(data['u2i_recall'])==0 and len(data['w2v_recall'])==0 \
  851. and len(data['sim_recall']) == 0 and len(data['u2u2i_recall']) == 0 :
  852. return [], 0
  853. # 地域分组小时级规则更新数据
  854. recall_dict = {}
  855. region_h_recall = [item for item in data['rov_pool_recall']
  856. if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_region_h']]
  857. region_h_recall_rank = sorted(region_h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  858. recall_dict['rov_recall_region_h'] = region_h_recall_rank
  859. # 地域分组小时级更新24h规则更新数据
  860. region_24h_recall = [item for item in data['rov_pool_recall']
  861. if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_region_24h']]
  862. region_24h_recall_rank = sorted(region_24h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  863. recall_dict['rov_recall_region_24h'] = region_24h_recall_rank
  864. # 相对24h规则更新数据
  865. rule_24h_recall = [item for item in data['rov_pool_recall']
  866. if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_24h']]
  867. rule_24h_recall_rank = sorted(rule_24h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  868. recall_dict['rov_recall_24h'] = rule_24h_recall_rank
  869. # 相对24h规则筛选后剩余更新数据
  870. rule_24h_dup_recall = [item for item in data['rov_pool_recall']
  871. if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_24h_dup']]
  872. rule_24h_dup_recall_rank = sorted(rule_24h_dup_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  873. recall_dict['rov_recall_24h_dup'] = rule_24h_dup_recall_rank
  874. hot_recall = []
  875. w2v_recall =[]
  876. sim_recall = []
  877. u2u2i_recall = []
  878. if ab_Code==60058:
  879. if len(data['u2i_recall'])>0:
  880. hot_recall = sorted(data['u2i_recall'], key=lambda k: k.get('rovScore', 0), reverse=True)
  881. recall_dict['u2i_recall'] = hot_recall
  882. elif ab_Code==60059:
  883. if len(data['w2v_recall'])>0:
  884. recall_dict['w2v_recall'] = data['w2v_recall']
  885. else:
  886. recall_dict['w2v_recall'] = w2v_recall
  887. elif ab_Code==60061 or ab_Code==60063:
  888. if len(data['sim_recall'])>0:
  889. recall_dict['sim_recall'] = data['sim_recall']
  890. else:
  891. recall_dict['sim_recall'] = sim_recall
  892. elif ab_Code==60062:
  893. if len(data['u2u2i_recall'])>0:
  894. recall_dict['u2u2i_recall'] = data['u2u2i_recall']
  895. else:
  896. recall_dict['u2u2i_recall'] = u2u2i_recall
  897. recall_list = [('rov_recall_region_h',1, 1),('rov_recall_region_h',0.5, 1),('rov_recall_region_24h',1,1),
  898. ('u2i_recall',0.5,1), ('w2v_recall',0.5,1),('rov_recall_24h',1,1), ('rov_recall_24h_dup',0.5,1)]
  899. if exp_config and exp_config['recall_list']:
  900. recall_list = exp_config['recall_list']
  901. #print("recall_config:", recall_list)
  902. rov_recall_rank = []
  903. select_ids = set('')
  904. for i in range(3):
  905. if len(rov_recall_rank)>8:
  906. break
  907. for per_recall_item in recall_list:
  908. per_recall_name = per_recall_item[0]
  909. per_recall_freq = per_recall_item[1]
  910. per_limt_num = per_recall_item[2]
  911. rand_num = random.random()
  912. #print(recall_dict[per_recall_name])
  913. if rand_num<per_recall_freq and per_recall_name in recall_dict:
  914. per_recall = recall_dict[per_recall_name]
  915. #print("per_recall_item:", per_recall_item)
  916. cur_recall_num = 0
  917. for recall_item in per_recall:
  918. vid = recall_item['videoId']
  919. if vid in select_ids:
  920. continue
  921. rov_recall_rank.append(recall_item)
  922. select_ids.add(vid)
  923. cur_recall_num+=1
  924. if cur_recall_num>=per_limt_num:
  925. break
  926. # print("rov_recall_rank:")
  927. # print(rov_recall_rank)
  928. #rov_recall_rank = region_h_recall_rank + region_24h_recall_rank + \
  929. # rule_24h_recall_rank + rule_24h_dup_recall_rank
  930. # 流量池
  931. flow_recall_rank = sorted(data['flow_pool_recall'], key=lambda k: k.get('rovScore', 0), reverse=True)
  932. # 对各路召回的视频进行去重
  933. rov_recall_rank, flow_recall_rank = remove_duplicate(rov_recall=rov_recall_rank, flow_recall=flow_recall_rank,
  934. top_K=top_K)
  935. # log_.info('remove_duplicate finished! rov_recall_rank = {}, flow_recall_rank = {}'.format(
  936. # rov_recall_rank, flow_recall_rank))
  937. # rank_result = relevant_recall_rank
  938. rank_result = []
  939. # 从ROV召回池中获取top k
  940. if len(rov_recall_rank) > 0:
  941. rank_result.extend(rov_recall_rank[:top_K])
  942. rov_recall_rank = rov_recall_rank[top_K:]
  943. else:
  944. rank_result.extend(flow_recall_rank[:top_K])
  945. flow_recall_rank = flow_recall_rank[top_K:]
  946. flow_num = 0
  947. flowConfig =0
  948. if exp_config and exp_config['flowConfig']:
  949. flowConfig = exp_config['flowConfig']
  950. if flowConfig == 1 and len(rov_recall_rank) > 0:
  951. rank_result.extend(rov_recall_rank[:top_K])
  952. for recall_item in rank_result:
  953. flow_recall_name = recall_item.get("flowPool", '')
  954. if flow_recall_name is not None and flow_recall_name.find("#") > -1:
  955. flow_num = flow_num + 1
  956. all_recall_rank = rov_recall_rank + flow_recall_rank
  957. if flow_num > 0:
  958. rank_result.extend(all_recall_rank[:size - top_K])
  959. return rank_result[:size], flow_num
  960. else:
  961. # 按概率 p 及score排序获取 size - k 个视频
  962. i = 0
  963. while i < size - top_K:
  964. # 随机生成[0, 1)浮点数
  965. rand = random.random()
  966. # log_.info('rand: {}'.format(rand))
  967. if rand < flow_pool_P:
  968. if flow_recall_rank:
  969. rank_result.append(flow_recall_rank[0])
  970. flow_recall_rank.remove(flow_recall_rank[0])
  971. else:
  972. rank_result.extend(rov_recall_rank[:size - top_K - i])
  973. return rank_result[:size], flow_num
  974. else:
  975. if rov_recall_rank:
  976. rank_result.append(rov_recall_rank[0])
  977. rov_recall_rank.remove(rov_recall_rank[0])
  978. else:
  979. rank_result.extend(flow_recall_rank[:size - top_K - i])
  980. return rank_result[:size], flow_num
  981. i += 1
  982. else:
  983. # 按概率 p 及score排序获取 size - k 个视频
  984. i = 0
  985. while i < size - top_K:
  986. # 随机生成[0, 1)浮点数
  987. rand = random.random()
  988. # log_.info('rand: {}'.format(rand))
  989. if rand < flow_pool_P:
  990. if flow_recall_rank:
  991. rank_result.append(flow_recall_rank[0])
  992. flow_recall_rank.remove(flow_recall_rank[0])
  993. else:
  994. rank_result.extend(rov_recall_rank[:size - top_K - i])
  995. return rank_result[:size], flow_num
  996. else:
  997. if rov_recall_rank:
  998. rank_result.append(rov_recall_rank[0])
  999. rov_recall_rank.remove(rov_recall_rank[0])
  1000. else:
  1001. rank_result.extend(flow_recall_rank[:size - top_K - i])
  1002. return rank_result[:size],flow_num
  1003. i += 1
  1004. return rank_result[:size], flow_num
  1005. def video_sank_pos_rank(data, size, top_K, flow_pool_P, ab_Code='', exp_config=None):
  1006. """
  1007. 视频分发排序
  1008. :param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []}
  1009. :param size: 请求数
  1010. :param top_K: 保证topK为召回池视频 type-int
  1011. :param flow_pool_P: size-top_K视频为流量池视频的概率 type-float
  1012. :return: rank_result
  1013. """
  1014. if not data['rov_pool_recall'] and not data['flow_pool_recall'] \
  1015. and len(data['u2i_recall'])==0 and len(data['w2v_recall'])==0 \
  1016. and len(data['sim_recall']) == 0 and len(data['u2u2i_recall']) == 0 :
  1017. return [], 0
  1018. # 地域分组小时级规则更新数据
  1019. recall_dict = {}
  1020. region_h_recall = [item for item in data['rov_pool_recall']
  1021. if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_region_h']]
  1022. region_h_recall_rank = sorted(region_h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  1023. recall_dict['rov_recall_region_h'] = region_h_recall_rank
  1024. # 地域分组小时级更新24h规则更新数据
  1025. region_24h_recall = [item for item in data['rov_pool_recall']
  1026. if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_region_24h']]
  1027. region_24h_recall_rank = sorted(region_24h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  1028. recall_dict['rov_recall_region_24h'] = region_24h_recall_rank
  1029. # 相对24h规则更新数据
  1030. rule_24h_recall = [item for item in data['rov_pool_recall']
  1031. if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_24h']]
  1032. rule_24h_recall_rank = sorted(rule_24h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  1033. recall_dict['rov_recall_24h'] = rule_24h_recall_rank
  1034. # 相对24h规则筛选后剩余更新数据
  1035. rule_24h_dup_recall = [item for item in data['rov_pool_recall']
  1036. if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_24h_dup']]
  1037. rule_24h_dup_recall_rank = sorted(rule_24h_dup_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  1038. recall_dict['rov_recall_24h_dup'] = rule_24h_dup_recall_rank
  1039. u2i_recall = []
  1040. u2i_play_recall = []
  1041. w2v_recall =[]
  1042. sim_recall = []
  1043. u2u2i_recall = []
  1044. return_video_recall = []
  1045. #print("")
  1046. if ab_Code==60058:
  1047. if len(data['u2i_recall'])>0:
  1048. recall_dict['u2i_recall'] = data['u2i_recall']
  1049. else:
  1050. recall_dict['u2i_recall'] = u2i_recall
  1051. if len(data['u2i_play_recall']) > 0:
  1052. recall_dict['u2i_play_recall'] = data['u2i_play_recall']
  1053. else:
  1054. recall_dict['u2i_play_recall'] = u2i_play_recall
  1055. elif ab_Code==60059:
  1056. if len(data['w2v_recall'])>0:
  1057. recall_dict['w2v_recall'] = data['w2v_recall']
  1058. else:
  1059. recall_dict['w2v_recall'] = w2v_recall
  1060. elif ab_Code==60061 or ab_Code==60063:
  1061. if len(data['sim_recall'])>0:
  1062. recall_dict['sim_recall'] = data['sim_recall']
  1063. else:
  1064. recall_dict['sim_recall'] = sim_recall
  1065. elif ab_Code==60062:
  1066. if len(data['u2u2i_recall'])>0:
  1067. recall_dict['u2u2i_recall'] = data['u2u2i_recall']
  1068. else:
  1069. recall_dict['u2u2i_recall'] = u2u2i_recall
  1070. elif ab_Code==60064:
  1071. if len(data['return_video_recall'])>0:
  1072. recall_dict['return_video_recall'] = data['return_video_recall']
  1073. else:
  1074. recall_dict['return_video_recall'] = return_video_recall
  1075. recall_pos1 = [('rov_recall_region_h',0, 0.98),('rov_recall_24h',0.98, 1),('rov_recall_region_24h',0,1),
  1076. ('rov_recall_24h',0,1), ('rov_recall_24h_dup',0,1)]
  1077. recall_pos2 = [('rov_recall_region_h',0,0.98),('rov_recall_24h',0.98,1),('rov_recall_region_24h',0,1),
  1078. ('rov_recall_24h',0,1),('rov_recall_24h_dup',0,1)]
  1079. recall_pos3 = [('rov_recall_region_h', 0,0.98), ('rov_recall_24h', 0.98,1), ('rov_recall_region_24h', 0,1),
  1080. ('rov_recall_24h', 0,1), ('rov_recall_24h_dup', 0,1)]
  1081. recall_pos4 = [('rov_recall_region_h', 0,0.98), ('rov_recall_24h', 0,0.02), ('rov_recall_region_24h', 0,1),
  1082. ('rov_recall_24h', 0,1), ('rov_recall_24h_dup', 0,1)]
  1083. if exp_config and 'recall_pos1' in exp_config \
  1084. and 'recall_pos2' in exp_config \
  1085. and 'recall_pos3' in exp_config \
  1086. and 'recall_pos4' in exp_config :
  1087. recall_pos1 = exp_config['recall_pos1']
  1088. recall_pos2 = exp_config['recall_pos2']
  1089. recall_pos3 = exp_config['recall_pos3']
  1090. recall_pos4 = exp_config['recall_pos4']
  1091. #print("recall_config:", recall_pos1)
  1092. rov_recall_rank = []
  1093. recall_list = []
  1094. recall_list.append(recall_pos1)
  1095. recall_list.append(recall_pos2)
  1096. recall_list.append(recall_pos3)
  1097. recall_list.append(recall_pos4)
  1098. select_ids = set('')
  1099. recall_num_limit_dict = {}
  1100. if exp_config and 'recall_num_limit' in exp_config:
  1101. recall_num_limit_dict = exp_config['recall_num_limit']
  1102. exp_recall_dict = {}
  1103. #index_pos = 0
  1104. for j in range(3):
  1105. if len(rov_recall_rank)>12:
  1106. break
  1107. # choose pos
  1108. for recall_pos_config in recall_list:
  1109. rand_num = random.random()
  1110. index_pos = 0
  1111. # choose pos recall
  1112. for per_recall_item in recall_pos_config:
  1113. if index_pos == 1:
  1114. break
  1115. if len(per_recall_item)<3:
  1116. continue
  1117. per_recall_name = per_recall_item[0]
  1118. per_recall_min = float(per_recall_item[1])
  1119. per_recall_max = float(per_recall_item[2])
  1120. per_recall_num = exp_recall_dict.get(per_recall_name, 0)
  1121. per_recall_total_num = recall_num_limit_dict.get(per_recall_name, 0)
  1122. # recall set total num
  1123. if len(recall_num_limit_dict)>0 and per_recall_total_num>0 and per_recall_num>= per_recall_total_num:
  1124. continue
  1125. if rand_num >= per_recall_min and rand_num < per_recall_max and per_recall_name in recall_dict:
  1126. per_recall = recall_dict[per_recall_name]
  1127. for recall_item in per_recall:
  1128. vid = recall_item['videoId']
  1129. if vid in select_ids:
  1130. continue
  1131. recall_item['rand'] = rand_num
  1132. rov_recall_rank.append(recall_item)
  1133. select_ids.add(vid)
  1134. if per_recall_name in exp_recall_dict:
  1135. exp_recall_dict[per_recall_name] +=1
  1136. else:
  1137. exp_recall_dict[per_recall_name] = 1
  1138. index_pos = 1
  1139. break
  1140. #print("rov_recall_rank:", rov_recall_rank)
  1141. if len(rov_recall_rank)<4:
  1142. 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
  1143. for recall_item in rov_doudi_rank:
  1144. vid = recall_item['videoId']
  1145. if vid in select_ids:
  1146. continue
  1147. rov_recall_rank.append(recall_item)
  1148. select_ids.add(vid)
  1149. if len(rov_recall_rank)>12:
  1150. break
  1151. # print("rov_recall_rank:")
  1152. #print(rov_recall_rank)
  1153. # 流量池
  1154. flow_recall_rank = sorted(data['flow_pool_recall'], key=lambda k: k.get('rovScore', 0), reverse=True)
  1155. # 对各路召回的视频进行去重
  1156. rov_recall_rank, flow_recall_rank = remove_duplicate(rov_recall=rov_recall_rank, flow_recall=flow_recall_rank,
  1157. top_K=top_K)
  1158. # log_.info('remove_duplicate finished! rov_recall_rank = {}, flow_recall_rank = {}'.format(
  1159. # rov_recall_rank, flow_recall_rank))
  1160. # rank_result = relevant_recall_rank
  1161. rank_result = []
  1162. # 从ROV召回池中获取top k
  1163. if len(rov_recall_rank) > 0:
  1164. rank_result.extend(rov_recall_rank[:top_K])
  1165. rov_recall_rank = rov_recall_rank[top_K:]
  1166. else:
  1167. rank_result.extend(flow_recall_rank[:top_K])
  1168. flow_recall_rank = flow_recall_rank[top_K:]
  1169. flow_num = 0
  1170. flowConfig =0
  1171. if exp_config and exp_config['flowConfig']:
  1172. flowConfig = exp_config['flowConfig']
  1173. if flowConfig == 1 and len(rov_recall_rank) > 0:
  1174. rank_result.extend(rov_recall_rank[:top_K])
  1175. for recall_item in rank_result:
  1176. flow_recall_name = recall_item.get("flowPool", '')
  1177. if flow_recall_name is not None and flow_recall_name.find("#") > -1:
  1178. flow_num = flow_num + 1
  1179. all_recall_rank = rov_recall_rank + flow_recall_rank
  1180. if flow_num > 0:
  1181. rank_result.extend(all_recall_rank[:size - top_K])
  1182. return rank_result[:size], flow_num
  1183. else:
  1184. # 按概率 p 及score排序获取 size - k 个视频
  1185. i = 0
  1186. while i < size - top_K:
  1187. # 随机生成[0, 1)浮点数
  1188. rand = random.random()
  1189. # log_.info('rand: {}'.format(rand))
  1190. if rand < flow_pool_P:
  1191. if flow_recall_rank:
  1192. rank_result.append(flow_recall_rank[0])
  1193. flow_recall_rank.remove(flow_recall_rank[0])
  1194. else:
  1195. rank_result.extend(rov_recall_rank[:size - top_K - i])
  1196. return rank_result[:size], flow_num
  1197. else:
  1198. if rov_recall_rank:
  1199. rank_result.append(rov_recall_rank[0])
  1200. rov_recall_rank.remove(rov_recall_rank[0])
  1201. else:
  1202. rank_result.extend(flow_recall_rank[:size - top_K - i])
  1203. return rank_result[:size], flow_num
  1204. i += 1
  1205. else:
  1206. # 按概率 p 及score排序获取 size - k 个视频
  1207. i = 0
  1208. while i < size - top_K:
  1209. # 随机生成[0, 1)浮点数
  1210. rand = random.random()
  1211. # log_.info('rand: {}'.format(rand))
  1212. if rand < flow_pool_P:
  1213. if flow_recall_rank:
  1214. rank_result.append(flow_recall_rank[0])
  1215. flow_recall_rank.remove(flow_recall_rank[0])
  1216. else:
  1217. rank_result.extend(rov_recall_rank[:size - top_K - i])
  1218. return rank_result[:size], flow_num
  1219. else:
  1220. if rov_recall_rank:
  1221. rank_result.append(rov_recall_rank[0])
  1222. rov_recall_rank.remove(rov_recall_rank[0])
  1223. else:
  1224. rank_result.extend(flow_recall_rank[:size - top_K - i])
  1225. return rank_result[:size],flow_num
  1226. i += 1
  1227. return rank_result[:size], flow_num
  1228. if __name__ == '__main__':
  1229. d_test = [{'videoId': 10028734, 'rovScore': 99.977, 'pushFrom': 'recall_pool', 'abCode': 10000},
  1230. {'videoId': 1919925, 'rovScore': 99.974, 'pushFrom': 'recall_pool', 'abCode': 10000},
  1231. {'videoId': 9968118, 'rovScore': 99.972, 'pushFrom': 'recall_pool', 'abCode': 10000},
  1232. {'videoId': 9934863, 'rovScore': 99.971, 'pushFrom': 'recall_pool', 'abCode': 10000},
  1233. {'videoId': 10219869, 'flowPool': '1#1#1#1640830818883', 'rovScore': 82.21929728934731, 'pushFrom': 'flow_pool', 'abCode': 10000},
  1234. {'videoId': 10212814, 'flowPool': '1#1#1#1640759014984', 'rovScore': 81.26694187726412, 'pushFrom': 'flow_pool', 'abCode': 10000},
  1235. {'videoId': 10219437, 'flowPool': '1#1#1#1640827620520', 'rovScore': 81.21634156641908, 'pushFrom': 'flow_pool', 'abCode': 10000},
  1236. {'videoId': 1994050, 'rovScore': 99.97, 'pushFrom': 'recall_pool', 'abCode': 10000},
  1237. {'videoId': 9894474, 'rovScore': 99.969, 'pushFrom': 'recall_pool', 'abCode': 10000},
  1238. {'videoId': 10028081, 'rovScore': 99.966, 'pushFrom': 'recall_pool', 'abCode': 10000}]
  1239. res = video_rank_by_w_h_rate(videos=d_test)
  1240. for tmp in res:
  1241. print(tmp)