video_rank.py 55 KB

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