video_rank.py 50 KB

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