video_rank.py 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340
  1. import random
  2. import numpy
  3. from log import Log
  4. from config import set_config
  5. from video_recall import PoolRecall
  6. from db_helper import RedisHelper
  7. from utils import FilterVideos, send_msg_to_feishu
  8. log_ = Log()
  9. config_ = set_config()
  10. def video_rank(data, size, top_K, flow_pool_P):
  11. """
  12. 视频分发排序
  13. :param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []}
  14. :param size: 请求数
  15. :param top_K: 保证topK为召回池视频 type-int
  16. :param flow_pool_P: size-top_K视频为流量池视频的概率 type-float
  17. :return: rank_result
  18. """
  19. if not data['rov_pool_recall'] and not data['flow_pool_recall']:
  20. return None
  21. # 将各路召回的视频按照score从大到小排序
  22. # 小时级更新数据
  23. h_recall = [item for item in data['rov_pool_recall'] if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_h']]
  24. h_recall_rank = sorted(h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  25. # 地域分组小时级规则更新数据
  26. region_h_recall = [item for item in data['rov_pool_recall']
  27. if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_region_h']]
  28. region_h_recall_rank = sorted(region_h_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  29. # 地域分组天级规则更新数据
  30. region_day_recall = [item for item in data['rov_pool_recall']
  31. if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_region_day']]
  32. region_day_recall_rank = sorted(region_day_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  33. # 天级规则更新数据
  34. day_recall = [item for item in data['rov_pool_recall'] if item.get('pushFrom') == config_.PUSH_FROM['rov_recall_day']]
  35. day_recall_rank = sorted(day_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  36. # ROV召回池
  37. rov_initial_recall = [
  38. item for item in data['rov_pool_recall']
  39. if item.get('pushFrom') not in
  40. [config_.PUSH_FROM['rov_recall_h'],
  41. config_.PUSH_FROM['rov_recall_region_h'],
  42. config_.PUSH_FROM['rov_recall_region_day'],
  43. config_.PUSH_FROM['rov_recall_day']]
  44. ]
  45. rov_initial_recall_rank = sorted(rov_initial_recall, key=lambda k: k.get('rovScore', 0), reverse=True)
  46. rov_recall_rank = h_recall_rank + region_h_recall_rank + region_day_recall_rank \
  47. + day_recall_rank + rov_initial_recall_rank
  48. # 流量池
  49. flow_recall_rank = sorted(data['flow_pool_recall'], key=lambda k: k.get('rovScore', 0), reverse=True)
  50. # 对各路召回的视频进行去重
  51. rov_recall_rank, flow_recall_rank = remove_duplicate(rov_recall=rov_recall_rank, flow_recall=flow_recall_rank,
  52. top_K=top_K)
  53. # log_.info('remove_duplicate finished! rov_recall_rank = {}, flow_recall_rank = {}'.format(
  54. # rov_recall_rank, flow_recall_rank))
  55. # 从ROV召回池中获取top k
  56. if len(rov_recall_rank) > 0:
  57. rank_result = rov_recall_rank[:top_K]
  58. rov_recall_rank = rov_recall_rank[top_K:]
  59. else:
  60. rank_result = flow_recall_rank[:top_K]
  61. flow_recall_rank = flow_recall_rank[top_K:]
  62. # 按概率 p 及score排序获取 size - k 个视频
  63. i = 0
  64. while i < size - top_K:
  65. # 随机生成[0, 1)浮点数
  66. rand = random.random()
  67. # log_.info('rand: {}'.format(rand))
  68. if rand < flow_pool_P:
  69. if flow_recall_rank:
  70. rank_result.append(flow_recall_rank[0])
  71. flow_recall_rank.remove(flow_recall_rank[0])
  72. else:
  73. rank_result.extend(rov_recall_rank[:size - top_K - i])
  74. return rank_result
  75. else:
  76. if rov_recall_rank:
  77. rank_result.append(rov_recall_rank[0])
  78. rov_recall_rank.remove(rov_recall_rank[0])
  79. else:
  80. rank_result.extend(flow_recall_rank[:size - top_K - i])
  81. return rank_result
  82. i += 1
  83. return rank_result
  84. def remove_duplicate(rov_recall, flow_recall, top_K):
  85. """
  86. 对多路召回的视频去重
  87. 去重原则:
  88. 如果视频在ROV召回池topK,则保留ROV召回池,否则保留流量池
  89. :param rov_recall: ROV召回池-已排序
  90. :param flow_recall: 流量池-已排序
  91. :param top_K: 保证topK为召回池视频 type-int
  92. :return:
  93. """
  94. flow_recall_result = []
  95. rov_recall_remove = []
  96. flow_recall_video_ids = [item['videoId'] for item in flow_recall]
  97. # rov_recall topK
  98. for item in rov_recall[:top_K]:
  99. if item['videoId'] in flow_recall_video_ids:
  100. flow_recall_video_ids.remove(item['videoId'])
  101. # other
  102. for item in rov_recall[top_K:]:
  103. if item['videoId'] in flow_recall_video_ids:
  104. rov_recall_remove.append(item)
  105. # rov recall remove
  106. for item in rov_recall_remove:
  107. rov_recall.remove(item)
  108. # flow recall remove
  109. for item in flow_recall:
  110. if item['videoId'] in flow_recall_video_ids:
  111. flow_recall_result.append(item)
  112. return rov_recall, flow_recall_result
  113. def bottom_strategy(size, app_type, ab_code):
  114. """
  115. 兜底策略: 从ROV召回池中获取top1000,进行状态过滤后的视频
  116. :param size: 需要获取的视频数
  117. :param app_type: 产品标识 type-int
  118. :param ab_code: abCode
  119. :return:
  120. """
  121. pool_recall = PoolRecall(app_type=app_type, ab_code=ab_code)
  122. key_name, _ = pool_recall.get_pool_redis_key(pool_type='rov')
  123. redis_helper = RedisHelper()
  124. data = redis_helper.get_data_zset_with_index(key_name=key_name, start=0, end=1000)
  125. if not data:
  126. log_.info('{} —— ROV推荐进入了二次兜底, data = {}'.format(config_.ENV_TEXT, data))
  127. send_msg_to_feishu('{} —— ROV推荐进入了二次兜底,请查看是否有数据更新失败问题。'.format(config_.ENV_TEXT))
  128. # 二次兜底
  129. bottom_data = bottom_strategy_last(size=size, app_type=app_type, ab_code=ab_code)
  130. return bottom_data
  131. # 视频状态过滤采用离线定时过滤方案
  132. # 状态过滤
  133. # filter_videos = FilterVideos(app_type=app_type, video_ids=data)
  134. # filtered_data = filter_videos.filter_video_status(video_ids=data)
  135. if len(data) > size:
  136. random_data = numpy.random.choice(data, size, False)
  137. else:
  138. random_data = data
  139. bottom_data = [{'videoId': int(item), 'pushFrom': config_.PUSH_FROM['bottom'], 'abCode': ab_code}
  140. for item in random_data]
  141. return bottom_data
  142. def bottom_strategy_last(size, app_type, ab_code):
  143. """
  144. 兜底策略: 从兜底视频中随机获取视频,进行状态过滤后的视频
  145. :param size: 需要获取的视频数
  146. :param app_type: 产品标识 type-int
  147. :param ab_code: abCode
  148. :return:
  149. """
  150. redis_helper = RedisHelper()
  151. bottom_data = redis_helper.get_data_zset_with_index(key_name=config_.BOTTOM_KEY_NAME, start=0, end=-1)
  152. random_data = numpy.random.choice(bottom_data, size * 30, False)
  153. # 视频状态过滤采用离线定时过滤方案
  154. # 状态过滤
  155. # filter_videos = FilterVideos(app_type=app_type, video_ids=random_data)
  156. # filtered_data = filter_videos.filter_video_status(video_ids=random_data)
  157. bottom_data = [{'videoId': int(video_id), 'pushFrom': config_.PUSH_FROM['bottom_last'], 'abCode': ab_code}
  158. for video_id in random_data[:size]]
  159. return bottom_data
  160. def video_rank_by_w_h_rate(videos):
  161. """
  162. 视频宽高比实验(每组的前两个视频调整为横屏视频),根据视频宽高比信息对视频进行重排
  163. :param videos:
  164. :return:
  165. """
  166. redis_helper = RedisHelper()
  167. # ##### 判断前两个视频是否是置顶视频 或者 流量池视频
  168. top_2_push_from_flag = [False, False]
  169. for i, video in enumerate(videos[:2]):
  170. if video['pushFrom'] in [config_.PUSH_FROM['top'], config_.PUSH_FROM['flow_recall']]:
  171. top_2_push_from_flag[i] = True
  172. if top_2_push_from_flag[0] and top_2_push_from_flag[1]:
  173. return videos
  174. # ##### 判断前两个视频是否为横屏
  175. top_2_w_h_rate_flag = [False, False]
  176. for i, video in enumerate(videos[:2]):
  177. if video['pushFrom'] in [config_.PUSH_FROM['top'], config_.PUSH_FROM['flow_recall']]:
  178. # 视频来源为置顶 或 流量池时,不做判断
  179. top_2_w_h_rate_flag[i] = True
  180. elif video['pushFrom'] in [config_.PUSH_FROM['rov_recall'], config_.PUSH_FROM['bottom']]:
  181. # 视频来源为 rov召回池 或 一层兜底时,判断是否是横屏
  182. w_h_rate = redis_helper.get_score_with_value(
  183. key_name=config_.W_H_RATE_UP_1_VIDEO_LIST_KEY_NAME['rov_recall'], value=video['videoId'])
  184. if w_h_rate is not None:
  185. top_2_w_h_rate_flag[i] = True
  186. elif video['pushFrom'] == config_.PUSH_FROM['bottom_last']:
  187. # 视频来源为 二层兜底时,判断是否是横屏
  188. w_h_rate = redis_helper.get_score_with_value(
  189. key_name=config_.W_H_RATE_UP_1_VIDEO_LIST_KEY_NAME['bottom_last'], value=video['videoId'])
  190. if w_h_rate is not None:
  191. top_2_w_h_rate_flag[i] = True
  192. if top_2_w_h_rate_flag[0] and top_2_w_h_rate_flag[1]:
  193. return videos
  194. # ##### 前两个视频中有不符合前面两者条件的,对视频进行位置调整
  195. # 记录横屏视频位置
  196. horizontal_video_index = []
  197. # 记录流量池视频位置
  198. flow_video_index = []
  199. # 记录置顶视频位置
  200. top_video_index = []
  201. for i, video in enumerate(videos):
  202. # 视频来源为置顶
  203. if video['pushFrom'] == config_.PUSH_FROM['top']:
  204. top_video_index.append(i)
  205. # 视频来源为流量池
  206. elif video['pushFrom'] == config_.PUSH_FROM['flow_recall']:
  207. flow_video_index.append(i)
  208. # 视频来源为rov召回池 或 一层兜底
  209. elif video['pushFrom'] in [config_.PUSH_FROM['rov_recall'], config_.PUSH_FROM['bottom']]:
  210. w_h_rate = redis_helper.get_score_with_value(
  211. key_name=config_.W_H_RATE_UP_1_VIDEO_LIST_KEY_NAME['rov_recall'], value=video['videoId'])
  212. if w_h_rate is not None:
  213. horizontal_video_index.append(i)
  214. else:
  215. continue
  216. # 视频来源为 二层兜底
  217. elif video['pushFrom'] == config_.PUSH_FROM['bottom_last']:
  218. w_h_rate = redis_helper.get_score_with_value(
  219. key_name=config_.W_H_RATE_UP_1_VIDEO_LIST_KEY_NAME['bottom_last'], value=video['videoId'])
  220. if w_h_rate is not None:
  221. horizontal_video_index.append(i)
  222. else:
  223. continue
  224. # 重新排序
  225. top2_index = []
  226. for i in range(2):
  227. if i in top_video_index:
  228. top2_index.append(i)
  229. elif i in flow_video_index:
  230. top2_index.append(i)
  231. flow_video_index.remove(i)
  232. elif i in horizontal_video_index:
  233. top2_index.append(i)
  234. horizontal_video_index.remove(i)
  235. elif len(horizontal_video_index) > 0:
  236. # 调整横屏视频到第一位
  237. top2_index.append(horizontal_video_index[0])
  238. # 从横屏位置记录中移除
  239. horizontal_video_index.pop(0)
  240. elif i == 0:
  241. return videos
  242. # 重排
  243. flow_result = [videos[i] for i in flow_video_index]
  244. other_result = [videos[i] for i in range(len(videos)) if i not in top2_index and i not in flow_video_index]
  245. top2_result = []
  246. for i, j in enumerate(top2_index):
  247. item = videos[j]
  248. if i != j:
  249. # 修改abCode
  250. item['abCode'] = config_.AB_CODE['w_h_rate']
  251. top2_result.append(item)
  252. new_rank_result = top2_result
  253. for i in range(len(top2_index), len(videos)):
  254. if i in flow_video_index:
  255. new_rank_result.append(flow_result[0])
  256. flow_result.pop(0)
  257. else:
  258. new_rank_result.append(other_result[0])
  259. other_result.pop(0)
  260. return new_rank_result
  261. def video_rank_with_old_video(rank_result, old_video_recall, size, top_K, old_video_index=2):
  262. """
  263. 视频分发排序 - 包含老视频, 老视频插入固定位置
  264. :param rank_result: 排序后的结果
  265. :param size: 请求数
  266. :param old_video_index: 老视频插入的位置索引,默认为2
  267. :return: new_rank_result
  268. """
  269. if not old_video_recall:
  270. return rank_result
  271. if not rank_result:
  272. return old_video_recall[:size]
  273. # 视频去重
  274. rank_video_ids = [item['videoId'] for item in rank_result]
  275. old_video_remove = []
  276. for old_video in old_video_recall:
  277. if old_video['videoId'] in rank_video_ids:
  278. old_video_remove.append(old_video)
  279. for item in old_video_remove:
  280. old_video_recall.remove(item)
  281. if not old_video_recall:
  282. return rank_result
  283. # 插入老视频
  284. # 随机获取一个视频
  285. ind = random.randint(0, len(old_video_recall) - 1)
  286. old_video = old_video_recall[ind]
  287. # 插入
  288. if len(rank_result) < top_K:
  289. new_rank_result = rank_result + [old_video]
  290. else:
  291. new_rank_result = rank_result[:old_video_index] + [old_video] + rank_result[old_video_index:]
  292. if len(new_rank_result) > size:
  293. # 判断后两位视频来源
  294. push_from_1 = new_rank_result[-1]['pushFrom']
  295. push_from_2 = new_rank_result[-2]['pushFrom']
  296. if push_from_2 == config_.PUSH_FROM['rov_recall'] and push_from_1 == config_.PUSH_FROM['flow_recall']:
  297. new_rank_result = new_rank_result[:-2] + new_rank_result[-1:]
  298. return new_rank_result[:size]
  299. if __name__ == '__main__':
  300. d_test = [{'videoId': 10028734, 'rovScore': 99.977, 'pushFrom': 'recall_pool', 'abCode': 10000},
  301. {'videoId': 1919925, 'rovScore': 99.974, 'pushFrom': 'recall_pool', 'abCode': 10000},
  302. {'videoId': 9968118, 'rovScore': 99.972, 'pushFrom': 'recall_pool', 'abCode': 10000},
  303. {'videoId': 9934863, 'rovScore': 99.971, 'pushFrom': 'recall_pool', 'abCode': 10000},
  304. {'videoId': 10219869, 'flowPool': '1#1#1#1640830818883', 'rovScore': 82.21929728934731, 'pushFrom': 'flow_pool', 'abCode': 10000},
  305. {'videoId': 10212814, 'flowPool': '1#1#1#1640759014984', 'rovScore': 81.26694187726412, 'pushFrom': 'flow_pool', 'abCode': 10000},
  306. {'videoId': 10219437, 'flowPool': '1#1#1#1640827620520', 'rovScore': 81.21634156641908, 'pushFrom': 'flow_pool', 'abCode': 10000},
  307. {'videoId': 1994050, 'rovScore': 99.97, 'pushFrom': 'recall_pool', 'abCode': 10000},
  308. {'videoId': 9894474, 'rovScore': 99.969, 'pushFrom': 'recall_pool', 'abCode': 10000},
  309. {'videoId': 10028081, 'rovScore': 99.966, 'pushFrom': 'recall_pool', 'abCode': 10000}]
  310. res = video_rank_by_w_h_rate(videos=d_test)
  311. for tmp in res:
  312. print(tmp)