recommend.py 106 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130
  1. import json
  2. import random
  3. import time
  4. import multiprocessing
  5. import traceback
  6. import hashlib
  7. from datetime import datetime, timedelta
  8. import config
  9. from log import Log
  10. from config import set_config
  11. from video_recall import PoolRecall
  12. from video_rank import video_new_rank2, video_sank_pos_rank, video_new_rank, video_rank, refactor_video_rank, \
  13. bottom_strategy, video_rank_by_w_h_rate, video_rank_with_old_video, bottom_strategy2, video_new_rank3
  14. from db_helper import RedisHelper
  15. import gevent
  16. from utils import FilterVideos, get_user_has30day_return
  17. import ast
  18. log_ = Log()
  19. config_ = set_config()
  20. def relevant_video_top_recommend(request_id, app_type, mid, uid, head_vid, videos, size):
  21. """
  22. 相关推荐强插 运营给定置顶相关性视频
  23. :param request_id: request_id
  24. :param app_type: 产品标识 type-int
  25. :param mid: mid
  26. :param uid: uid
  27. :param head_vid: 相关推荐头部视频id type-int
  28. :param videos: 当前相关推荐结果 type-list
  29. :param size: 返回视频个数 type-int
  30. :return: rank_result
  31. """
  32. # 获取头部视频对应的相关性视频
  33. key_name = '{}{}'.format(config_.RELEVANT_VIDEOS_WITH_OP_KEY_NAME, head_vid)
  34. redis_helper = RedisHelper()
  35. relevant_videos = redis_helper.get_data_from_redis(key_name=key_name)
  36. if relevant_videos is None:
  37. # 该视频没有指定的相关性视频
  38. return videos
  39. relevant_videos = json.loads(relevant_videos)
  40. # 按照指定顺序排序
  41. relevant_videos_sorted = sorted(relevant_videos, key=lambda x: x['order'], reverse=False)
  42. # 过滤
  43. relevant_video_ids = [int(item['recommend_vid']) for item in relevant_videos_sorted]
  44. filter_helper = FilterVideos(request_id=request_id, app_type=app_type, video_ids=relevant_video_ids, mid=mid, uid=uid)
  45. filtered_ids = filter_helper.filter_videos()
  46. if filtered_ids is None:
  47. return videos
  48. # 获取生效中的视频
  49. now = int(time.time())
  50. relevant_videos_in_effect = [
  51. {'videoId': int(item['recommend_vid']), 'pushFrom': config_.PUSH_FROM['relevant_video_op'],
  52. 'abCode': config_.AB_CODE['relevant_video_op']}
  53. for item in relevant_videos_sorted
  54. if item['start_time'] < now < item['finish_time'] and int(item['recommend_vid']) in filtered_ids
  55. ]
  56. if len(relevant_videos_in_effect) == 0:
  57. return videos
  58. # 与现有排序结果 进行合并重排
  59. # 获取现有排序结果中流量池视频 及其位置
  60. relevant_ids = [item['videoId'] for item in relevant_videos_in_effect]
  61. flow_pool_videos = []
  62. other_videos = []
  63. for i, item in enumerate(videos):
  64. if item.get('pushFrom', None) == config_.PUSH_FROM['flow_recall'] and item.get('videoId') not in relevant_ids:
  65. flow_pool_videos.append((i, item))
  66. elif item.get('videoId') not in relevant_ids:
  67. other_videos.append(item)
  68. else:
  69. continue
  70. # 重排,保持流量池视频位置不变
  71. rank_result = relevant_videos_in_effect + other_videos
  72. for i, item in flow_pool_videos:
  73. rank_result.insert(i, item)
  74. return rank_result[:size]
  75. def video_position_recommend(request_id, mid, uid, app_type, videos):
  76. # videos = video_recommend(mid=mid, uid=uid, size=size, app_type=app_type,
  77. # algo_type=algo_type, client_info=client_info)
  78. redis_helper = RedisHelper()
  79. pos1_vids = redis_helper.get_data_from_redis(config.BaseConfig.RECALL_POSITION1_KEY_NAME)
  80. pos2_vids = redis_helper.get_data_from_redis(config.BaseConfig.RECALL_POSITION2_KEY_NAME)
  81. if pos1_vids is not None:
  82. pos1_vids = ast.literal_eval(pos1_vids)
  83. if pos2_vids is not None:
  84. pos2_vids = ast.literal_eval(pos2_vids)
  85. pos1_vids = [] if pos1_vids is None else pos1_vids
  86. pos2_vids = [] if pos2_vids is None else pos2_vids
  87. pos1_vids = [int(i) for i in pos1_vids]
  88. pos2_vids = [int(i) for i in pos2_vids]
  89. filter_1 = FilterVideos(request_id=request_id, app_type=app_type, video_ids=pos1_vids, mid=mid, uid=uid)
  90. filter_2 = FilterVideos(request_id=request_id, app_type=app_type, video_ids=pos2_vids, mid=mid, uid=uid)
  91. t = [gevent.spawn(filter_1.filter_videos), gevent.spawn(filter_2.filter_videos)]
  92. gevent.joinall(t)
  93. filted_list = [i.get() for i in t]
  94. pos1_vids = filted_list[0]
  95. pos2_vids = filted_list[1]
  96. videos = positon_duplicate(pos1_vids, pos2_vids, videos)
  97. if pos1_vids is not None and len(pos1_vids) >0 :
  98. videos.insert(0, {'videoId': int(pos1_vids[0]), 'rovScore': 100,
  99. 'pushFrom': config_.PUSH_FROM['position_insert'], 'abCode': config_.AB_CODE['position_insert']})
  100. if pos2_vids is not None and len(pos2_vids) >0 :
  101. videos.insert(1, {'videoId': int(pos2_vids[0]), 'rovScore': 100,
  102. 'pushFrom': config_.PUSH_FROM['position_insert'], 'abCode': config_.AB_CODE['position_insert']})
  103. return videos[:10]
  104. def positon_duplicate(pos1_vids, pos2_vids, videos):
  105. s = set()
  106. if pos1_vids is not None and len(pos1_vids) >0:
  107. s.add(int(pos1_vids[0]))
  108. if pos2_vids is not None and len(pos2_vids) >0:
  109. s.add(int(pos2_vids[0]))
  110. l = []
  111. for item in videos:
  112. if item['videoId'] in s:
  113. continue
  114. else:
  115. l.append(item)
  116. return l
  117. def video_recommend(request_id, mid, uid, size, top_K, flow_pool_P, app_type, algo_type, client_info,
  118. expire_time=24*3600, ab_code=config_.AB_CODE['initial'], rule_key='', data_key='',
  119. no_op_flag=False, old_video_index=-1, video_id=None, params=None, rule_key_30day=None,
  120. shield_config=None, level_weight=None, flow_pool_abtest_group=None):
  121. """
  122. 首页线上推荐逻辑
  123. :param request_id: request_id
  124. :param mid: mid type-string
  125. :param uid: uid type-string
  126. :param size: 请求视频数量 type-int
  127. :param top_K: 保证topK为召回池视频 type-int
  128. :param flow_pool_P: size-top_K视频为流量池视频的概率 type-float
  129. :param app_type: 产品标识 type-int
  130. :param algo_type: 算法类型 type-string
  131. :param client_info: 用户位置信息 {"country": "国家", "province": "省份", "city": "城市"}
  132. :param expire_time: 末位视频记录redis过期时间
  133. :param ab_code: AB实验code
  134. :param video_id: 相关推荐头部视频id
  135. :param params:
  136. :return:
  137. """
  138. result = {}
  139. # ####### 多进程召回
  140. start_recall = time.time()
  141. # log_.info('====== recall')
  142. '''
  143. cores = multiprocessing.cpu_count()
  144. pool = multiprocessing.Pool(processes=cores)
  145. pool_recall = PoolRecall(app_type=app_type, mid=mid, uid=uid, ab_code=ab_code)
  146. _, last_rov_recall_key, _ = pool_recall.get_video_last_idx()
  147. pool_list = [
  148. # rov召回池
  149. pool.apply_async(pool_recall.rov_pool_recall, (size,)),
  150. # 流量池
  151. pool.apply_async(pool_recall.flow_pool_recall, (size,))
  152. ]
  153. recall_result_list = [p.get() for p in pool_list]
  154. pool.close()
  155. pool.join()
  156. '''
  157. recall_result_list = []
  158. pool_recall = PoolRecall(request_id=request_id,
  159. app_type=app_type, mid=mid, uid=uid, ab_code=ab_code,
  160. client_info=client_info, rule_key=rule_key, data_key=data_key, no_op_flag=no_op_flag,
  161. params=params, rule_key_30day=rule_key_30day, shield_config=shield_config,
  162. video_id= video_id, level_weight=level_weight)
  163. # _, last_rov_recall_key, _ = pool_recall.get_video_last_idx()
  164. # # 小时级实验
  165. # if ab_code in [code for _, code in config_.AB_CODE['rank_by_h'].items()]:
  166. # t = [gevent.spawn(pool_recall.rule_recall_by_h, size, expire_time),
  167. # gevent.spawn(pool_recall.flow_pool_recall, size, config_.QUICK_FLOW_POOL_ID),
  168. # gevent.spawn(pool_recall.flow_pool_recall, size)]
  169. # # 小时级实验
  170. # elif ab_code in [code for _, code in config_.AB_CODE['rank_by_24h'].items()]:
  171. # t = [gevent.spawn(pool_recall.rov_pool_recall_by_h, size, expire_time),
  172. # gevent.spawn(pool_recall.flow_pool_recall, size, config_.QUICK_FLOW_POOL_ID),
  173. # gevent.spawn(pool_recall.flow_pool_recall, size)]
  174. # 地域分组实验
  175. # if ab_code in [code for _, code in config_.AB_CODE['region_rank_by_h'].items()]:
  176. # load test config
  177. exp_config = None
  178. if ab_code == 60058:
  179. exp_config = pool_recall.get_U2I_config()
  180. elif ab_code == 60059:
  181. exp_config = pool_recall.get_w2v_config()
  182. elif ab_code == 60060:
  183. exp_config = pool_recall.get_test_config()
  184. elif ab_code == 60061:
  185. exp_config = pool_recall.get_simrecall_config()
  186. elif ab_code == 60062:
  187. exp_config = pool_recall.get_u2u2i_config()
  188. elif ab_code == 60063:
  189. exp_config = pool_recall.get_simrecall_config_new()
  190. elif ab_code == 60064:
  191. exp_config = pool_recall.get_video_recall_config()
  192. #print("exp_config:", exp_config)
  193. if app_type in [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]:
  194. t = [gevent.spawn(pool_recall.rov_pool_recall_with_region, size, expire_time)]
  195. # if ab_code == 60058:
  196. # t.append(gevent.spawn(pool_recall.get_U2I_reall, mid))
  197. # t.append(gevent.spawn(pool_recall.get_play_reall, mid))
  198. # elif ab_code == 60059:
  199. # t.append(gevent.spawn(pool_recall.get_word2vec_item_reall))
  200. # elif ab_code == 60061 or ab_code == 60063:
  201. # t.append(gevent.spawn(pool_recall.get_sim_hot_item_reall_filter))
  202. # elif ab_code == 60062:
  203. # t.append(gevent.spawn(pool_recall.get_U2U2I_reall, mid))
  204. # elif ab_code == 60064:
  205. # t.append(gevent.spawn(pool_recall.get_return_video_reall))
  206. elif flow_pool_abtest_group == 'experimental_flow_set_level':
  207. t = [gevent.spawn(pool_recall.rov_pool_recall_with_region, size, expire_time),
  208. gevent.spawn(pool_recall.flow_pool_recall_new_with_level,
  209. size, flow_pool_id=config_.QUICK_FLOW_POOL_ID, flow_pool_abtest_group=flow_pool_abtest_group),
  210. gevent.spawn(pool_recall.flow_pool_recall_new_with_level,
  211. size, flow_pool_abtest_group=flow_pool_abtest_group)]
  212. else:
  213. t = [gevent.spawn(pool_recall.rov_pool_recall_with_region, size, expire_time),
  214. gevent.spawn(pool_recall.flow_pool_recall,
  215. size, flow_pool_id=config_.QUICK_FLOW_POOL_ID, flow_pool_abtest_group=flow_pool_abtest_group),
  216. gevent.spawn(pool_recall.flow_pool_recall,
  217. size, flow_pool_abtest_group=flow_pool_abtest_group)]
  218. if ab_code == 60058:
  219. t.append(gevent.spawn(pool_recall.get_U2I_reall, mid))
  220. t.append(gevent.spawn(pool_recall.get_play_reall, mid))
  221. elif ab_code == 60059:
  222. t.append(gevent.spawn(pool_recall.get_word2vec_item_reall))
  223. elif ab_code == 60061 or ab_code == 60063:
  224. t.append(gevent.spawn(pool_recall.get_sim_hot_item_reall_filter))
  225. elif ab_code == 60062:
  226. t.append(gevent.spawn(pool_recall.get_U2U2I_reall, mid))
  227. elif ab_code == 60064:
  228. t.append(gevent.spawn(pool_recall.get_return_video_reall))
  229. # 最惊奇相关推荐实验
  230. # elif ab_code == config_.AB_CODE['top_video_relevant_appType_19']:
  231. # t = [gevent.spawn(pool_recall.relevant_recall_19, video_id, size, expire_time),
  232. # gevent.spawn(pool_recall.flow_pool_recall_18_19, size)]
  233. # 最惊奇完整影视实验
  234. # elif ab_code == config_.AB_CODE['whole_movies']:
  235. # t = [gevent.spawn(pool_recall.rov_pool_recall_19, size, expire_time)]
  236. # 最惊奇/老好看实验
  237. # elif app_type in [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]:
  238. # t = [gevent.spawn(pool_recall.rov_pool_recall, size, expire_time),
  239. # gevent.spawn(pool_recall.flow_pool_recall_18_19, size)]
  240. # # 天级实验
  241. # elif ab_code in [code for _, code in config_.AB_CODE['rank_by_day'].items()]:
  242. # t = [gevent.spawn(pool_recall.rov_pool_recall_by_day, size, expire_time),
  243. # gevent.spawn(pool_recall.flow_pool_recall, size, config_.QUICK_FLOW_POOL_ID),
  244. # gevent.spawn(pool_recall.flow_pool_recall, size)]
  245. # 老视频实验
  246. # elif ab_code in [config_.AB_CODE['old_video']]:
  247. # t = [gevent.spawn(pool_recall.rov_pool_recall, size, expire_time),
  248. # gevent.spawn(pool_recall.flow_pool_recall, size),
  249. # gevent.spawn(pool_recall.old_videos_recall, size)]
  250. # else:
  251. # if app_type in [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]:
  252. # t = [gevent.spawn(pool_recall.rov_pool_recall, size, expire_time)]
  253. # else:
  254. # t = [gevent.spawn(pool_recall.rov_pool_recall, size, expire_time),
  255. # gevent.spawn(pool_recall.flow_pool_recall, size, config_.QUICK_FLOW_POOL_ID),
  256. # gevent.spawn(pool_recall.flow_pool_recall, size)]
  257. gevent.joinall(t)
  258. recall_result_list = [i.get() for i in t]
  259. # end_recall = time.time()
  260. # log_.info({
  261. # 'logTimestamp': int(time.time() * 1000),
  262. # 'request_id': request_id,
  263. # 'mid': mid,
  264. # 'uid': uid,
  265. # 'operation': 'recall',
  266. # 'recall_result': recall_result_list,
  267. # 'executeTime': (time.time() - start_recall) * 1000
  268. # })
  269. result['recallResult'] = recall_result_list
  270. result['recallTime'] = (time.time() - start_recall) * 1000
  271. # ####### 排序
  272. start_rank = time.time()
  273. # log_.info('====== rank')
  274. if app_type in [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]:
  275. if ab_code in [
  276. config_.AB_CODE['rov_rank_appType_18_19'],
  277. config_.AB_CODE['rov_rank_appType_19'],
  278. config_.AB_CODE['top_video_relevant_appType_19']
  279. ]:
  280. data = {
  281. 'rov_pool_recall': recall_result_list[0],
  282. 'flow_pool_recall': recall_result_list[1]
  283. }
  284. else:
  285. data = {
  286. 'rov_pool_recall': recall_result_list[0],
  287. 'flow_pool_recall': []
  288. }
  289. else:
  290. if recall_result_list[1]:
  291. redis_helper = RedisHelper()
  292. quick_flow_pool_P = redis_helper.get_data_from_redis(
  293. key_name=f"{config_.QUICK_FLOWPOOL_DISTRIBUTE_RATE_KEY_NAME_PREFIX}{config_.QUICK_FLOW_POOL_ID}"
  294. )
  295. if quick_flow_pool_P:
  296. flow_pool_P = quick_flow_pool_P
  297. data = {
  298. 'rov_pool_recall': recall_result_list[0],
  299. 'flow_pool_recall': recall_result_list[1]
  300. }
  301. else:
  302. data = {
  303. 'rov_pool_recall': recall_result_list[0],
  304. 'flow_pool_recall': recall_result_list[2]
  305. }
  306. data['u2i_recall'] = []
  307. data['u2i_play_recall'] = []
  308. data['w2v_recall'] = []
  309. data['sim_recall'] = []
  310. data['u2u2i_recall'] = []
  311. if ab_code == 60058:
  312. if len(recall_result_list)>=4:
  313. data['u2i_recall'] = recall_result_list[3]
  314. if len(recall_result_list)>=5:
  315. data['u2i_play_recall'] = recall_result_list[4]
  316. elif ab_code == 60059:
  317. if len(recall_result_list)>=4:
  318. data['w2v_recall'] = recall_result_list[3]
  319. elif (ab_code == 60061 or ab_code == 60063):
  320. if len(recall_result_list)>=4:
  321. data['sim_recall'] = recall_result_list[3]
  322. elif ab_code == 60062:
  323. if len(recall_result_list)>=4:
  324. data['u2u2i_recall'] = recall_result_list[3]
  325. elif ab_code == 60064:
  326. if len(recall_result_list)>=4: #if ab_code=="ab_new_test":
  327. data['return_video_recall'] = recall_result_list[3] # rank_result = video_new_rank(data=data, size=size, top_K=top_K, flow_pool_P=float(flow_pool_P))
  328. #else:
  329. #print("data['hot_recall']", data['hot_recall'])
  330. # 60058: u2itag, 60059:word2vec, 60061: sim_recall, 60062: u2u2i
  331. # if ab_code == 60058 or ab_code == 60059 or ab_code == 60060 or ab_code == 60061 \
  332. # or ab_code == 60062 or ab_code== 60063 or ab_code == 60064:
  333. # rank_result, flow_num = video_sank_pos_rank(data=data, size=size, top_K=top_K, flow_pool_P=float(flow_pool_P), ab_Code=ab_code, exp_config=exp_config)
  334. # result['flow_num'] = flow_num
  335. # if rank_result:
  336. # result['rank_num'] = len(rank_result)
  337. # else:
  338. rank_result = video_rank(data=data, size=size, top_K=top_K, flow_pool_P=float(flow_pool_P))
  339. # 老视频实验
  340. # if ab_code in [config_.AB_CODE['old_video']]:
  341. # rank_result = video_rank_with_old_video(rank_result=rank_result, old_video_recall=recall_result_list[2],
  342. # size=size, top_K=top_K, old_video_index=old_video_index)
  343. # end_rank = time.time()
  344. # log_.info({
  345. # 'logTimestamp': int(time.time() * 1000),
  346. # 'request_id': request_id,
  347. # 'mid': mid,
  348. # 'uid': uid,
  349. # 'operation': 'rank',
  350. # 'rank_result': rank_result,
  351. # 'executeTime': (time.time() - start_rank) * 1000
  352. # })
  353. result['rankResult'] = rank_result
  354. result['rankTime'] = (time.time() - start_rank) * 1000
  355. # if not rank_result:
  356. # # 兜底策略
  357. # # log_.info('====== bottom strategy')
  358. # start_bottom = time.time()
  359. # rank_result = bottom_strategy2(
  360. # size=size, app_type=app_type, mid=mid, uid=uid, ab_code=ab_code, client_info=client_info, params=params
  361. # )
  362. #
  363. # # if ab_code == config_.AB_CODE['region_rank_by_h'].get('abtest_130'):
  364. # # rank_result = bottom_strategy2(
  365. # # size=size, app_type=app_type, mid=mid, uid=uid, ab_code=ab_code, client_info=client_info, params=params
  366. # # )
  367. # # else:
  368. # # rank_result = bottom_strategy(
  369. # # request_id=request_id, size=size, app_type=app_type, ab_code=ab_code, params=params
  370. # # )
  371. #
  372. # # log_.info({
  373. # # 'logTimestamp': int(time.time() * 1000),
  374. # # 'request_id': request_id,
  375. # # 'mid': mid,
  376. # # 'uid': uid,
  377. # # 'operation': 'bottom',
  378. # # 'bottom_result': rank_result,
  379. # # 'executeTime': (time.time() - start_bottom) * 1000
  380. # # })
  381. # result['bottomResult'] = rank_result
  382. # result['bottomTime'] = (time.time() - start_bottom) * 1000
  383. #
  384. # result['rankResult'] = rank_result
  385. return result
  386. # return rank_result, last_rov_recall_key
  387. def video_old_recommend(request_id, mid, uid, size, top_K, flow_pool_P, app_type, algo_type, client_info,
  388. expire_time=24*3600, ab_code=config_.AB_CODE['initial'], rule_key='', data_key='',
  389. no_op_flag=False, old_video_index=-1, video_id=None, params=None, rule_key_30day=None,
  390. shield_config=None, env_dict=None, level_weight=None, flow_pool_abtest_group=None,
  391. rank_key_prefix=None):
  392. """
  393. 首页线上推荐逻辑
  394. :param request_id: request_id
  395. :param mid: mid type-string
  396. :param uid: uid type-string
  397. :param size: 请求视频数量 type-int
  398. :param top_K: 保证topK为召回池视频 type-int
  399. :param flow_pool_P: size-top_K视频为流量池视频的概率 type-float
  400. :param app_type: 产品标识 type-int
  401. :param algo_type: 算法类型 type-string
  402. :param client_info: 用户位置信息 {"country": "国家", "province": "省份", "city": "城市"}
  403. :param expire_time: 末位视频记录redis过期时间
  404. :param ab_code: AB实验code
  405. :param video_id: 相关推荐头部视频id
  406. :param params:
  407. :return:
  408. """
  409. result = {}
  410. # ####### 多进程召回
  411. start_recall = time.time()
  412. # log_.info('====== recall')
  413. #print("abcode",ab_code)
  414. recall_result_list = []
  415. pool_recall = PoolRecall(request_id=request_id,
  416. app_type=app_type, mid=mid, uid=uid, ab_code=ab_code,
  417. client_info=client_info, rule_key=rule_key, data_key=data_key, no_op_flag=no_op_flag,
  418. params=params, rule_key_30day=rule_key_30day, shield_config=shield_config,
  419. video_id=video_id, level_weight=level_weight)
  420. exp_config = pool_recall.get_sort_ab_codel_config()
  421. # 60054 全量: simrecall+融合排序
  422. if app_type in [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]:
  423. t = [gevent.spawn(pool_recall.rov_pool_recall_with_region, size, expire_time)]
  424. # if ab_code == 60054 or ab_code == 60066 or ab_code == 60072 or ab_code == 60073:
  425. # t.append(gevent.spawn(pool_recall.get_sim_hot_item_reall_filter))
  426. # if ab_code == 60056 or ab_code == 60071:
  427. # t.append(gevent.spawn(pool_recall.get_sim_hot_item_reall_filter))
  428. # t.append(gevent.spawn(pool_recall.get_U2I_reall, mid))
  429. # if ab_code ==60067 or ab_code == 60069:
  430. # t.append(gevent.spawn(pool_recall.get_sim_hot_item_reall_filter))
  431. # t.append(gevent.spawn(pool_recall.get_return_video_reall))
  432. # if ab_code == 60068 or ab_code == 60070:
  433. # t.append(gevent.spawn(pool_recall.get_sim_hot_item_reall_filter))
  434. # t.append(gevent.spawn(pool_recall.get_return_video_reall, 'rv2:'))
  435. elif flow_pool_abtest_group == 'experimental_flow_set_level':
  436. t = [gevent.spawn(pool_recall.rov_pool_recall_with_region, size, expire_time),
  437. gevent.spawn(pool_recall.flow_pool_recall_new_with_level,
  438. size, flow_pool_id=config_.QUICK_FLOW_POOL_ID, flow_pool_abtest_group=flow_pool_abtest_group),
  439. gevent.spawn(pool_recall.flow_pool_recall_new_with_level,
  440. size, flow_pool_abtest_group=flow_pool_abtest_group)]
  441. else:
  442. t = [gevent.spawn(pool_recall.rov_pool_recall_with_region, size, expire_time),
  443. gevent.spawn(pool_recall.flow_pool_recall,
  444. size, flow_pool_id=config_.QUICK_FLOW_POOL_ID, flow_pool_abtest_group=flow_pool_abtest_group),
  445. gevent.spawn(pool_recall.flow_pool_recall,
  446. size, flow_pool_abtest_group=flow_pool_abtest_group)]
  447. if ab_code == 60054 or ab_code == 60066 or ab_code == 60072 or ab_code == 60073 or ab_code == 60074 \
  448. or ab_code == 60075 or ab_code == 60076 or ab_code == 60077 or ab_code == 60078 or ab_code == 60079:
  449. t.append(gevent.spawn(pool_recall.get_sim_hot_item_reall_filter))
  450. elif ab_code == 60056 or ab_code == 60071:
  451. t.append(gevent.spawn(pool_recall.get_sim_hot_item_reall_filter))
  452. t.append(gevent.spawn(pool_recall.get_U2I_reall, mid))
  453. elif ab_code == 60067 or ab_code == 60069:
  454. t.append(gevent.spawn(pool_recall.get_sim_hot_item_reall_filter))
  455. t.append(gevent.spawn(pool_recall.get_return_video_reall))
  456. elif ab_code == 60068 or ab_code == 60070 or ab_code == 60080 or ab_code == 60081 or ab_code == 60082:
  457. t.append(gevent.spawn(pool_recall.get_sim_hot_item_reall_filter))
  458. t.append(gevent.spawn(pool_recall.get_return_video_reall, 'rv2:'))
  459. gevent.joinall(t)
  460. recall_result_list = [i.get() for i in t]
  461. #print("recall:",recall_result_list)
  462. if len(recall_result_list)<0:
  463. result['recallResult']= []
  464. result['rankResult'] = []
  465. return result
  466. #1. merge simrecall or deepfm
  467. if ab_code == 60054 or ab_code == 60066 or ab_code == 60072 or ab_code == 60073 or ab_code == 60074 \
  468. or ab_code == 60075 or ab_code == 60076 or ab_code == 60077 or ab_code == 60078 or ab_code == 60079 \
  469. or ab_code == 60080 or ab_code == 60081 or ab_code == 60082:
  470. rov_pool_recall = []
  471. if len(recall_result_list) >= 2:
  472. region_recall = recall_result_list[0]
  473. sim_recall = []
  474. if app_type in [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]:
  475. sim_recall = recall_result_list[1]
  476. else:
  477. if len(recall_result_list)>=4:
  478. sim_recall = recall_result_list[3]
  479. now_video_ids = set('')
  480. if len(region_recall) > 0:
  481. for video in region_recall:
  482. video_id = video.get('videoId')
  483. if video_id not in now_video_ids:
  484. rov_pool_recall.append(video)
  485. now_video_ids.add(video_id)
  486. if len(sim_recall) > 0:
  487. for video in sim_recall:
  488. video_id = video.get('videoId')
  489. # print("sim video_id:", video_id)
  490. if video_id not in now_video_ids:
  491. rov_pool_recall.append(video)
  492. now_video_ids.add(video_id)
  493. if len(rov_pool_recall) > 0:
  494. recall_result_list[0] = rov_pool_recall
  495. # merge simrecall, merge u2i title recall, deepfm
  496. if ab_code == 60056 or ab_code == 60071:
  497. rov_pool_recall = []
  498. if len(recall_result_list)>=2:
  499. region_recall = recall_result_list[0]
  500. sim_recall = []
  501. u2i_title_recall = []
  502. if app_type in [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]:
  503. sim_recall = recall_result_list[1]
  504. if len(recall_result_list) >= 3:
  505. u2i_title_recall = recall_result_list[2]
  506. else:
  507. if len(recall_result_list) >= 4:
  508. sim_recall = recall_result_list[3]
  509. if len(recall_result_list) >= 5:
  510. u2i_title_recall = recall_result_list[4]
  511. #print("u2i_title_recall:", u2i_title_recall)
  512. now_video_ids = set('')
  513. if len(region_recall)>0:
  514. for video in region_recall:
  515. video_id = video.get('videoId')
  516. if video_id not in now_video_ids:
  517. rov_pool_recall.append(video)
  518. now_video_ids.add(video_id)
  519. if len(sim_recall) > 0:
  520. for video in sim_recall:
  521. video_id = video.get('videoId')
  522. if video_id not in now_video_ids:
  523. rov_pool_recall.append(video)
  524. now_video_ids.add(video_id)
  525. if len(u2i_title_recall)>0:
  526. for video in u2i_title_recall:
  527. video_id = video.get('videoId')
  528. #print("sim video_id:", video_id)
  529. if video_id not in now_video_ids:
  530. rov_pool_recall.append(video)
  531. now_video_ids.add(video_id)
  532. if len(rov_pool_recall)>0:
  533. recall_result_list[0] = rov_pool_recall
  534. #2. merge simrecall, return video
  535. if ab_code == 60067 or ab_code == 60068 or ab_code == 60069 or ab_code == 60070 \
  536. or ab_code == 60080 or ab_code == 60081 or ab_code == 60082:
  537. rov_pool_recall = []
  538. if len(recall_result_list)>=2:
  539. region_recall = recall_result_list[0]
  540. return_video_reall = []
  541. sim_recall = []
  542. if app_type in [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]:
  543. sim_recall = recall_result_list[1]
  544. if len(recall_result_list) >= 3:
  545. return_video_reall = recall_result_list[2]
  546. else:
  547. if len(recall_result_list)>=4:
  548. sim_recall = recall_result_list[3]
  549. if len(recall_result_list)>=5:
  550. return_video_reall = recall_result_list[4]
  551. #print("sim:",sim_recall)
  552. now_video_ids = set('')
  553. if len(region_recall)>0:
  554. for video in region_recall:
  555. video_id = video.get('videoId')
  556. if video_id not in now_video_ids:
  557. rov_pool_recall.append(video)
  558. now_video_ids.add(video_id)
  559. if len(sim_recall) > 0:
  560. for video in sim_recall:
  561. video_id = video.get('videoId')
  562. if video_id not in now_video_ids:
  563. rov_pool_recall.append(video)
  564. now_video_ids.add(video_id)
  565. if len(return_video_reall)>0:
  566. for video in return_video_reall:
  567. video_id = video.get('videoId')
  568. if video_id not in now_video_ids:
  569. rov_pool_recall.append(video)
  570. now_video_ids.add(video_id)
  571. if len(rov_pool_recall)>0:
  572. recall_result_list[0] = rov_pool_recall
  573. result['recallResult'] = recall_result_list
  574. result['recallTime'] = (time.time() - start_recall) * 1000
  575. #print("recall:", recall_result_list)
  576. # ####### 排序
  577. start_rank = time.time()
  578. # log_.info('====== rank')
  579. if app_type in [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]:
  580. if ab_code in [
  581. config_.AB_CODE['rov_rank_appType_18_19'],
  582. config_.AB_CODE['rov_rank_appType_19'],
  583. config_.AB_CODE['top_video_relevant_appType_19']
  584. ]:
  585. data = {
  586. 'rov_pool_recall': recall_result_list[0],
  587. 'flow_pool_recall': recall_result_list[1]
  588. }
  589. else:
  590. data = {
  591. 'rov_pool_recall': recall_result_list[0],
  592. 'flow_pool_recall': []
  593. }
  594. else:
  595. if recall_result_list[1]:
  596. redis_helper = RedisHelper()
  597. quick_flow_pool_P = redis_helper.get_data_from_redis(
  598. key_name=f"{config_.QUICK_FLOWPOOL_DISTRIBUTE_RATE_KEY_NAME_PREFIX}{config_.QUICK_FLOW_POOL_ID}"
  599. )
  600. if quick_flow_pool_P:
  601. flow_pool_P = quick_flow_pool_P
  602. data = {
  603. 'rov_pool_recall': recall_result_list[0],
  604. 'flow_pool_recall': recall_result_list[1]
  605. }
  606. else:
  607. data = {
  608. 'rov_pool_recall': recall_result_list[0],
  609. 'flow_pool_recall': recall_result_list[2]
  610. }
  611. # 3. 特征回流
  612. #
  613. # for recall_item in data['rov_pool_recall']:
  614. # if len(recall_item) <= 0:
  615. # continue
  616. # vid = recall_item.get("videoId", 0)
  617. # rec_recall_list.append(vid)
  618. # rec_recall_item_list.append(recall_item)
  619. # redisObj = RedisHelper()
  620. # video_static_info = redisObj.get_batch_key(vidKeys)
  621. # video_hour_static_info = redisObj.get_batch_key(hour_vidKeys)
  622. # vid_day_fea_list = []
  623. # vid_hour_fea_list = []
  624. # if video_static_info:
  625. # vid_day_fea_list = video_static_info
  626. # if video_hour_static_info:
  627. # vid_hour_fea_list = video_hour_static_info
  628. if env_dict:
  629. province_code = client_info.get('provinceCode', -1)
  630. if province_code and province_code == "":
  631. province_code =-1
  632. city_code = client_info.get('cityCode', -1)
  633. if city_code and city_code == "":
  634. city_code = -1
  635. env_dict['mid'] = mid
  636. env_dict['province_code'] = province_code
  637. env_dict['city_code'] = city_code
  638. env_json = env_dict
  639. #4.
  640. # rank_result, flow_num = video_new_rank2(data=data, size=size, top_K=top_K, flow_pool_P=float(flow_pool_P), ab_code=ab_code, mid=mid, exp_config=exp_config, env_dict=env_dict)
  641. rank_result, flow_num = video_new_rank3(
  642. data=data, size=size, top_K=top_K, flow_pool_P=float(flow_pool_P), rank_key_prefix=rank_key_prefix
  643. )
  644. #print(rank_result)
  645. if rank_result:
  646. result['rank_num'] = len(rank_result)
  647. day_vidKeys = []
  648. hour_vidKeys = []
  649. rec_recall_list = []
  650. pre_str = "v_ctr:"
  651. pre_hour_str = "v_hour_ctr:"
  652. if env_dict and len(rank_result)>0:
  653. for rec_item in rank_result:
  654. vid = rec_item.get("videoId", 0)
  655. rec_recall_list.append(vid)
  656. day_vidKeys.append(pre_str+str(vid))
  657. hour_vidKeys.append(pre_hour_str+str(vid))
  658. redisObj = RedisHelper()
  659. video_static_info = redisObj.get_batch_key(day_vidKeys)
  660. video_hour_static_info = redisObj.get_batch_key(hour_vidKeys)
  661. vid_day_fea_list = []
  662. vid_hour_fea_list = []
  663. if video_static_info:
  664. vid_day_fea_list = video_static_info
  665. if video_hour_static_info:
  666. vid_hour_fea_list = video_hour_static_info
  667. env_dict['recall_list'] = rec_recall_list
  668. env_dict['vid_day_fea_list'] = vid_day_fea_list
  669. env_dict['vid_hour_fea_list'] = vid_hour_fea_list
  670. env_json = env_dict
  671. result['rankResult'] = rank_result
  672. result['flow_num'] = flow_num
  673. result['rankTime'] = (time.time() - start_rank) * 1000
  674. return result, env_json
  675. # return rank_result, last_rov_recall_key
  676. def new_video_recommend(request_id, mid, uid, size, top_K, flow_pool_P, app_type, algo_type, client_info,
  677. expire_time=24*3600, ab_code=config_.AB_CODE['initial'], rule_key='', data_key='',
  678. no_op_flag=False, old_video_index=-1, video_id=None, params=None, rule_key_30day=None,
  679. shield_config=None):
  680. """
  681. 首页线上推荐逻辑
  682. :param request_id: request_id
  683. :param mid: mid type-string
  684. :param uid: uid type-string
  685. :param size: 请求视频数量 type-int
  686. :param top_K: 保证topK为召回池视频 type-int
  687. :param flow_pool_P: size-top_K视频为流量池视频的概率 type-float
  688. :param app_type: 产品标识 type-int
  689. :param algo_type: 算法类型 type-string
  690. :param client_info: 用户位置信息 {"country": "国家", "province": "省份", "city": "城市"}
  691. :param expire_time: 末位视频记录redis过期时间
  692. :param ab_code: AB实验code
  693. :param video_id: 相关推荐头部视频id
  694. :param params:
  695. :return:
  696. """
  697. #1. recall
  698. result = {}
  699. result['rankResult'] = []
  700. # ####### 多进程召回
  701. start_recall = time.time()
  702. # 1. 根据城市或者省份获取region_code
  703. city_code_list = [code for _, code in config_.CITY_CODE.items()]
  704. # 获取provinceCode
  705. province_code = client_info.get('provinceCode', '-1')
  706. # 获取cityCode
  707. city_code = client_info.get('cityCode', '-1')
  708. if city_code in city_code_list:
  709. # 分城市数据存在时,获取城市分组数据
  710. region_code = city_code
  711. else:
  712. region_code = province_code
  713. if region_code == '':
  714. region_code = '-1'
  715. #print("region_code:", region_code)
  716. #size =1000
  717. pool_recall = PoolRecall(request_id=request_id,
  718. app_type=app_type, mid=mid, uid=uid, ab_code=ab_code,
  719. client_info=client_info, rule_key=rule_key, data_key=data_key, no_op_flag=no_op_flag,
  720. params=params, rule_key_30day=rule_key_30day, shield_config=shield_config, video_id= video_id)
  721. if app_type in [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]:
  722. t = [gevent.spawn(pool_recall.get_region_hour_recall, size, region_code),
  723. gevent.spawn(pool_recall.get_region_day_recall, size, region_code),
  724. gevent.spawn(pool_recall.get_selected_recall, size, region_code),
  725. gevent.spawn(pool_recall.get_no_selected_recall, size, region_code)
  726. ]
  727. if ab_code == 60049:
  728. t.append(gevent.spawn(pool_recall.get_sim_hot_item_reall))
  729. else:
  730. t = [
  731. gevent.spawn(pool_recall.get_region_hour_recall, size, region_code),
  732. gevent.spawn(pool_recall.get_region_day_recall, size, region_code),
  733. gevent.spawn(pool_recall.get_selected_recall, size, region_code),
  734. gevent.spawn(pool_recall.get_no_selected_recall, size, region_code),
  735. gevent.spawn(pool_recall.new_flow_pool_recall, size, config_.QUICK_FLOW_POOL_ID),
  736. gevent.spawn(pool_recall.new_flow_pool_recall, size)]
  737. if ab_code ==60049:
  738. t.append(gevent.spawn(pool_recall.get_sim_hot_item_reall))
  739. gevent.joinall(t)
  740. # all recall_result
  741. all_recall_result_list = [i.get() for i in t]
  742. all_recall_result = []
  743. #print(all_recall_result_list)
  744. result['recallTime'] = (time.time() - start_recall) * 1000
  745. #print("recall time:", result['recallTime'])
  746. if not all_recall_result_list or len(all_recall_result_list)==0:
  747. return result
  748. for recall_item in all_recall_result_list:
  749. if not recall_item or len(recall_item)==0:
  750. continue
  751. for per_item in recall_item:
  752. all_recall_result.append(per_item)
  753. #print("all_recall_result:", all_recall_result)
  754. #2. duplicate
  755. dup_time = time.time()
  756. recall_dict = {}
  757. fast_flow_set = set('')
  758. flow_flow_set = set('')
  759. region_h_recall = []
  760. region_day_recall = []
  761. select_day_recall = []
  762. no_selected_recall = []
  763. sim_hot_recall = []
  764. flow_recall = []
  765. flowFlag_dict = {}
  766. for per_item in all_recall_result:
  767. #print(per_item)
  768. try:
  769. vId = int(per_item.get("videoId",0))
  770. if vId==0:
  771. continue
  772. recall_name = per_item.get("pushFrom",'')
  773. flow_pool = per_item.get("flowPool", '')
  774. if flow_pool != '':
  775. flow_pool_id = int(flow_pool.split('#')[0])
  776. if flow_pool_id == config_.QUICK_FLOW_POOL_ID:
  777. fast_flow_set.add(vId)
  778. else:
  779. flow_flow_set.add(vId)
  780. flowFlag_dict[vId] = flow_pool
  781. #duplicate divide into
  782. if vId not in recall_dict:
  783. if recall_name == config_.PUSH_FROM['rov_recall_region_h']:
  784. region_h_recall.append(per_item)
  785. elif recall_name == config_.PUSH_FROM['rov_recall_region_24h']:
  786. region_day_recall.append(per_item)
  787. elif recall_name == config_.PUSH_FROM['rov_recall_24h']:
  788. select_day_recall.append(per_item)
  789. elif recall_name == config_.PUSH_FROM['rov_recall_24h_dup']:
  790. no_selected_recall.append(per_item)
  791. elif recall_name == config_.PUSH_FROM['sim_hot_vid_recall']:
  792. sim_hot_recall.append(per_item)
  793. elif recall_name == config_.PUSH_FROM['flow_recall']:
  794. flow_recall.append(per_item)
  795. if vId not in recall_dict:
  796. recall_dict[vId] = recall_name
  797. else:
  798. recall_name = recall_dict[vId] + "," + recall_name
  799. recall_dict[vId] = recall_name
  800. except:
  801. continue
  802. #print("recall_dict:", recall_dict)
  803. #print("recall time:", (time.time()-dup_time)*1000)
  804. #3. filter video, 先过预曝光
  805. filter_time = time.time()
  806. filter_ = FilterVideos(request_id=request_id,
  807. app_type=app_type, mid=mid, uid=uid, video_ids=list(recall_dict.keys()))
  808. #print("filer:", list(recall_dict.keys()))
  809. #a).expose filter
  810. #all_recall_list = list(recall_dict.keys())
  811. all_recall_list = filter_.filter_videos_new(region_code=region_code, shield_config=shield_config, flow_set=flowFlag_dict.keys())
  812. #print("filer after:", all_recall_list)
  813. #print("filter_ time:", (time.time() - filter_time) * 1000)
  814. #4. sort: old sort: flow 按概率出
  815. start_rank = time.time()
  816. #quick_flow_pool_P get from redis
  817. redis_helper = RedisHelper()
  818. quick_flow_pool_P = redis_helper.get_data_from_redis(
  819. key_name=f"{config_.QUICK_FLOWPOOL_DISTRIBUTE_RATE_KEY_NAME_PREFIX}{config_.QUICK_FLOW_POOL_ID}"
  820. )
  821. if quick_flow_pool_P:
  822. flow_pool_P = quick_flow_pool_P
  823. rank_result= []
  824. if ab_code==60048 or ab_code==60049:
  825. rank_ids, add_flow_set = video_new_rank(videoIds=all_recall_list,fast_flow_set=fast_flow_set, flow_set=flow_flow_set,size=size, top_K=top_K, flow_pool_P=float(flow_pool_P))
  826. #print("rank_ids:", rank_ids)
  827. for rank_item in rank_ids:
  828. rank_id = rank_item[0]
  829. rank_score = rank_item[1]
  830. pushFrom = recall_dict.get(rank_id, '')
  831. #print(pushFrom, rank_id)
  832. flowPoolFlag = ''
  833. if rank_id in add_flow_set:
  834. flowPoolFlag = flowFlag_dict.get(rank_id,'')
  835. rank_result.append({'videoId': rank_id, 'flowPool': flowPoolFlag,
  836. 'rovScore': rank_score, 'pushFrom': pushFrom,
  837. 'abCode': ab_code})
  838. #
  839. #print("rank_result:", rank_result)
  840. else:
  841. all_dup_recall_result = region_h_recall+region_day_recall+select_day_recall+no_selected_recall+flow_recall
  842. rank_result = refactor_video_rank(rov_recall_rank=all_dup_recall_result,fast_flow_set=fast_flow_set, flow_set=flow_flow_set, size=size, top_K=top_K, flow_pool_P=float(flow_pool_P))
  843. result['rankResult'] = rank_result
  844. result['rankTime'] = (time.time() - start_rank) * 1000
  845. #print("rank time:", result['rankTime'])
  846. return result
  847. # return rank_result, last_rov_recall_key
  848. def ab_test_op(rank_result, ab_code_list, app_type, mid, uid, **kwargs):
  849. """
  850. 对排序后的结果 按照AB实验进行对应的分组操作
  851. :param rank_result: 排序后的结果
  852. :param ab_code_list: 此次请求参与的 ab实验组
  853. :param app_type: 产品标识
  854. :param mid: mid
  855. :param uid: uid
  856. :param kwargs: 其他参数
  857. :return:
  858. """
  859. # ####### 视频宽高比AB实验
  860. # 对内容精选进行 视频宽高比分发实验
  861. # if config_.AB_CODE['w_h_rate'] in ab_code_list and app_type in config_.AB_TEST.get('w_h_rate', []):
  862. # rank_result = video_rank_by_w_h_rate(videos=rank_result)
  863. # log_.info('app_type: {}, mid: {}, uid: {}, rank_by_w_h_rate_result: {}'.format(
  864. # app_type, mid, uid, rank_result))
  865. # 按position位置排序
  866. if config_.AB_CODE['position_insert'] in ab_code_list and app_type in config_.AB_TEST.get('position_insert', []):
  867. rank_result = video_position_recommend(mid, uid, app_type, rank_result)
  868. print('===========================')
  869. print(rank_result)
  870. log_.info('app_type: {}, mid: {}, uid: {}, rank_by_position_insert_result: {}'.format(
  871. app_type, mid, uid, rank_result))
  872. # 相关推荐强插
  873. # if config_.AB_CODE['relevant_video_op'] in ab_code_list \
  874. # and app_type in config_.AB_TEST.get('relevant_video_op', []):
  875. # head_vid = kwargs['head_vid']
  876. # size = kwargs['size']
  877. # rank_result = relevant_video_top_recommend(
  878. # app_type=app_type, mid=mid, uid=uid, head_vid=head_vid, videos=rank_result, size=size
  879. # )
  880. # log_.info('app_type: {}, mid: {}, uid: {}, head_vid: {}, rank_by_relevant_video_op_result: {}'.format(
  881. # app_type, mid, uid, head_vid, rank_result))
  882. return rank_result
  883. def update_redis_data(result, app_type, mid, top_K, expire_time=24*3600, level_weight=None, flow_pool_abtest_group=None):
  884. """
  885. 根据最终的排序结果更新相关redis数据
  886. :param result: 排序结果
  887. :param app_type: 产品标识
  888. :param mid: mid
  889. :param top_K: 保证topK为召回池视频 type-int
  890. :param expire_time: 末位视频记录redis过期时间
  891. :return: None
  892. """
  893. # ####### redis数据刷新
  894. try:
  895. redis_helper = RedisHelper()
  896. # log_.info('====== update redis')
  897. if mid and mid != 'null':
  898. # mid为空时,不做预曝光和定位数据更新
  899. # 预曝光数据同步刷新到Redis, 过期时间为0.5h
  900. preview_key_name = f"{config_.PREVIEW_KEY_PREFIX}{app_type}:{mid}"
  901. preview_video_ids = [int(item['videoId']) for item in result]
  902. if preview_video_ids:
  903. # log_.error('key_name = {} \n values = {}'.format(preview_key_name, tuple(preview_video_ids)))
  904. redis_helper.add_data_with_set(key_name=preview_key_name, values=tuple(preview_video_ids), expire_time=30 * 60)
  905. # log_.info('preview redis update success!')
  906. # # 将此次获取的ROV召回池top_K末位视频id同步刷新到Redis中,方便下次快速定位到召回位置,过期时间为1天
  907. # rov_recall_video = [item['videoId'] for item in result[:top_K]
  908. # if item['pushFrom'] == config_.PUSH_FROM['rov_recall']]
  909. # if len(rov_recall_video) > 0:
  910. # if app_type == config_.APP_TYPE['APP']:
  911. # key_name = config_.UPDATE_ROV_KEY_NAME_APP
  912. # else:
  913. # key_name = config_.UPDATE_ROV_KEY_NAME
  914. # if not redis_helper.get_score_with_value(key_name=key_name, value=rov_recall_video[-1]):
  915. # redis_helper.set_data_to_redis(key_name=last_rov_recall_key, value=rov_recall_video[-1],
  916. # expire_time=expire_time)
  917. # log_.info('last video redis update success!')
  918. # 将此次获取的 地域分组小时级数据列表 中的视频id同步刷新到redis中,方便下次快速定位到召回位置
  919. rov_recall_h_video = [item['videoId'] for item in result[:top_K]
  920. if item['pushFrom'] == config_.PUSH_FROM['rov_recall_region_h']]
  921. if len(rov_recall_h_video) > 0:
  922. last_video_key = f'{config_.LAST_VIDEO_FROM_REGION_H_PREFIX}{app_type}:{mid}'
  923. redis_helper.set_data_to_redis(key_name=last_video_key, value=rov_recall_h_video[-1],
  924. expire_time=expire_time)
  925. # 将此次获取的 地域分组相对24h数据列表 中的视频id同步刷新到redis中,方便下次快速定位到召回位置
  926. rov_recall_24h_dup1_video = [item['videoId'] for item in result[:top_K]
  927. if item['pushFrom'] == config_.PUSH_FROM['rov_recall_region_24h']]
  928. if len(rov_recall_24h_dup1_video) > 0:
  929. last_video_key = f'{config_.LAST_VIDEO_FROM_REGION_DUP1_24H_PREFIX}{app_type}:{mid}'
  930. redis_helper.set_data_to_redis(key_name=last_video_key, value=rov_recall_24h_dup1_video[-1],
  931. expire_time=expire_time)
  932. # 将此次获取的 相对24h筛选数据列表 中的视频id同步刷新到redis中,方便下次快速定位到召回位置
  933. rov_recall_24h_dup2_video = [item['videoId'] for item in result[:top_K]
  934. if item['pushFrom'] == config_.PUSH_FROM['rov_recall_24h']]
  935. if len(rov_recall_24h_dup2_video) > 0:
  936. last_video_key = f'{config_.LAST_VIDEO_FROM_REGION_DUP2_24H_PREFIX}{app_type}:{mid}'
  937. redis_helper.set_data_to_redis(key_name=last_video_key, value=rov_recall_24h_dup2_video[-1],
  938. expire_time=expire_time)
  939. # 将此次获取的 相对24h筛选后剩余数据列表 中的视频id同步刷新到redis中,方便下次快速定位到召回位置
  940. rov_recall_24h_dup3_video = [item['videoId'] for item in result[:top_K]
  941. if item['pushFrom'] == config_.PUSH_FROM['rov_recall_24h_dup']]
  942. if len(rov_recall_24h_dup3_video) > 0:
  943. last_video_key = f'{config_.LAST_VIDEO_FROM_REGION_DUP3_24H_PREFIX}{app_type}:{mid}'
  944. redis_helper.set_data_to_redis(key_name=last_video_key, value=rov_recall_24h_dup3_video[-1],
  945. expire_time=expire_time)
  946. # # 将此次获取的 相对48h筛选数据列表 中的视频id同步刷新到redis中,方便下次快速定位到召回位置
  947. # rov_recall_48h_dup2_video = [item['videoId'] for item in result[:top_K]
  948. # if item['pushFrom'] == config_.PUSH_FROM['rov_recall_48h']]
  949. # if len(rov_recall_48h_dup2_video) > 0:
  950. # last_video_key = f'{config_.LAST_VIDEO_FROM_REGION_DUP2_48H_PREFIX}{app_type}:{mid}'
  951. # redis_helper.set_data_to_redis(key_name=last_video_key, value=rov_recall_48h_dup2_video[-1],
  952. # expire_time=expire_time)
  953. #
  954. # # 将此次获取的 相对48h筛选后剩余数据列表 中的视频id同步刷新到redis中,方便下次快速定位到召回位置
  955. # rov_recall_48h_dup3_video = [item['videoId'] for item in result[:top_K]
  956. # if item['pushFrom'] == config_.PUSH_FROM['rov_recall_48h_dup']]
  957. # if len(rov_recall_48h_dup3_video) > 0:
  958. # last_video_key = f'{config_.LAST_VIDEO_FROM_REGION_DUP3_48H_PREFIX}{app_type}:{mid}'
  959. # redis_helper.set_data_to_redis(key_name=last_video_key, value=rov_recall_48h_dup3_video[-1],
  960. # expire_time=expire_time)
  961. # 将此次分发的流量池视频,对 本地分发数-1 进行记录
  962. if app_type not in [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]:
  963. # 获取本地分发数-1策略开关
  964. switch = redis_helper.get_data_from_redis(key_name=config_.IN_FLOW_POOL_COUNT_SWITCH_KEY_NAME)
  965. if switch is not None:
  966. if int(switch) == 1:
  967. flow_recall_video = [item for item in result if item.get('flowPool', None) is not None]
  968. else:
  969. flow_recall_video = [item for item in result if
  970. item['pushFrom'] == config_.PUSH_FROM['flow_recall']]
  971. else:
  972. flow_recall_video = [item for item in result if item['pushFrom'] == config_.PUSH_FROM['flow_recall']]
  973. if flow_recall_video:
  974. if flow_pool_abtest_group == 'experimental_flow_set_level':
  975. update_local_distribute_count_new_with_level(flow_recall_video, level_weight)
  976. else:
  977. update_local_distribute_count(flow_recall_video)
  978. # update_local_distribute_count_new(flow_recall_video)
  979. # log_.info('update local distribute count success!')
  980. # 限流视频分发数记录
  981. if app_type == config_.APP_TYPE['APP']:
  982. # APP 不计入
  983. return
  984. limit_video_id_list = redis_helper.get_data_from_set(
  985. key_name=f"{config_.KEY_NAME_PREFIX_LIMIT_VIDEO_SET}{datetime.today().strftime('%Y%m%d')}"
  986. )
  987. if limit_video_id_list is not None:
  988. limit_video_id_list = [int(item) for item in limit_video_id_list]
  989. for item in result:
  990. video_id = item['videoId']
  991. if video_id in limit_video_id_list:
  992. key_name = f"{config_.KEY_NAME_PREFIX_LIMIT_VIDEO_DISTRIBUTE_COUNT}{video_id}"
  993. redis_helper.setnx_key(key_name=key_name, value=0, expire_time=24*2600)
  994. redis_helper.incr_key(key_name=key_name, amount=1, expire_time=24*3600)
  995. except Exception as e:
  996. log_.error("update redis data fail!")
  997. log_.error(traceback.format_exc())
  998. def update_flow_redis_data(result, app_type, mid, top_K, expire_time=24*3600):
  999. """
  1000. 根据最终的排序结果更新相关redis数据
  1001. :param result: 排序结果
  1002. :param app_type: 产品标识
  1003. :param mid: mid
  1004. :param top_K: 保证topK为召回池视频 type-int
  1005. :param expire_time: 末位视频记录redis过期时间
  1006. :return: None
  1007. """
  1008. # ####### redis数据刷新
  1009. try:
  1010. redis_helper = RedisHelper()
  1011. # log_.info('====== update redis')
  1012. if mid and mid != 'null':
  1013. # mid为空时,不做预曝光和定位数据更新
  1014. # 预曝光数据同步刷新到Redis, 过期时间为0.5h
  1015. preview_key_name = f"{config_.PREVIEW_KEY_PREFIX}{app_type}:{mid}"
  1016. preview_video_ids = [int(item['videoId']) for item in result]
  1017. if preview_video_ids:
  1018. # log_.error('key_name = {} \n values = {}'.format(preview_key_name, tuple(preview_video_ids)))
  1019. redis_helper.add_data_with_set(key_name=preview_key_name, values=tuple(preview_video_ids), expire_time=30 * 60)
  1020. # log_.info('preview redis update success!')
  1021. # 将此次分发的流量池视频,对 本地分发数-1 进行记录
  1022. if app_type not in [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]:
  1023. # 获取本地分发数-1策略开关
  1024. switch = redis_helper.get_data_from_redis(key_name=config_.IN_FLOW_POOL_COUNT_SWITCH_KEY_NAME)
  1025. if switch is not None:
  1026. if int(switch) == 1:
  1027. flow_recall_video = [item for item in result if item.get('flowPool', None) is not None]
  1028. else:
  1029. flow_recall_video = [item for item in result if
  1030. item['pushFrom'] == config_.PUSH_FROM['flow_recall']]
  1031. else:
  1032. flow_recall_video = [item for item in result if item['pushFrom'] == config_.PUSH_FROM['flow_recall']]
  1033. if flow_recall_video:
  1034. update_local_distribute_count(flow_recall_video)
  1035. # log_.info('update local distribute count success!')
  1036. # 限流视频分发数记录
  1037. if app_type == config_.APP_TYPE['APP']:
  1038. # APP 不计入
  1039. return
  1040. limit_video_id_list = redis_helper.get_data_from_set(
  1041. key_name=f"{config_.KEY_NAME_PREFIX_LIMIT_VIDEO_SET}{datetime.today().strftime('%Y%m%d')}"
  1042. )
  1043. if limit_video_id_list is not None:
  1044. limit_video_id_list = [int(item) for item in limit_video_id_list]
  1045. for item in result:
  1046. video_id = item['videoId']
  1047. if video_id in limit_video_id_list:
  1048. key_name = f"{config_.KEY_NAME_PREFIX_LIMIT_VIDEO_DISTRIBUTE_COUNT}{video_id}"
  1049. redis_helper.setnx_key(key_name=key_name, value=0, expire_time=24*2600)
  1050. redis_helper.incr_key(key_name=key_name, amount=1, expire_time=24*3600)
  1051. except Exception as e:
  1052. log_.error("update redis data fail!")
  1053. log_.error(traceback.format_exc())
  1054. def update_local_distribute_count(videos):
  1055. """
  1056. 更新本地分发数
  1057. :param videos: 视频列表 type-list [{'videoId':'', 'flowPool':'', 'distributeCount': '',
  1058. 'rovScore': '', 'pushFrom': 'flow_pool', 'abCode': self.ab_code}, ....]
  1059. :return:
  1060. """
  1061. try:
  1062. redis_helper = RedisHelper()
  1063. for item in videos:
  1064. video_id, flow_pool = item['videoId'], item['flowPool']
  1065. key_name = f"{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{video_id}:{flow_pool}"
  1066. # 本地记录的分发数 - 1
  1067. redis_helper.decr_key(key_name=key_name, amount=1, expire_time=15 * 60)
  1068. # 对该视频做分发数检查
  1069. cur_count = redis_helper.get_data_from_redis(key_name=key_name)
  1070. # 无记录
  1071. if cur_count is None:
  1072. continue
  1073. # 本地分发数 cur_count <= 0,从所有的流量召回池移除,删除本地分发记录key
  1074. if int(cur_count) <= 0:
  1075. add_remove_log = False
  1076. redis_helper.del_keys(key_name=key_name)
  1077. for app_name in config_.APP_TYPE:
  1078. app_type = config_.APP_TYPE.get(app_name)
  1079. flow_pool_key_list = [
  1080. f"{config_.FLOWPOOL_KEY_NAME_PREFIX}{app_type}",
  1081. f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX}{app_type}:{config_.QUICK_FLOW_POOL_ID}"
  1082. ]
  1083. for key in flow_pool_key_list:
  1084. remove_res = redis_helper.remove_value_from_zset(key_name=key, value=f"{video_id}-{flow_pool}")
  1085. if remove_res > 0:
  1086. add_remove_log = True
  1087. video_flow_pool_key_list = [
  1088. f"{config_.QUICK_FLOWPOOL_VIDEO_INFO_KEY_NAME_PREFIX}{app_type}:{config_.QUICK_FLOW_POOL_ID}:{video_id}",
  1089. f"{config_.FLOWPOOL_VIDEO_INFO_KEY_NAME_PREFIX}{app_type}:{video_id}"
  1090. ]
  1091. for key in video_flow_pool_key_list:
  1092. redis_helper.remove_value_from_set(key_name=key, values=(flow_pool, ))
  1093. if add_remove_log is True:
  1094. log_.info({'tag': 'remove video_id from flow_pool', 'video_id': video_id, 'flow_pool': flow_pool})
  1095. # if redis_helper.key_exists(key_name=key_name):
  1096. # # 该视频本地有记录,本地记录的分发数 - 1
  1097. # redis_helper.decr_key(key_name=key_name, amount=1, expire_time=5 * 60)
  1098. # else:
  1099. # # 该视频本地无记录,接口获取的分发数 - 1
  1100. # redis_helper.incr_key(key_name=key_name, amount=int(item['distributeCount']) - 1, expire_time=5 * 60)
  1101. except Exception as e:
  1102. log_.error('update_local_distribute_count error...')
  1103. log_.error(traceback.format_exc())
  1104. def update_local_distribute_count_new(videos):
  1105. """
  1106. 更新本地分发数
  1107. :param videos: 视频列表 type-list [{'videoId':'', 'flowPool':'', 'distributeCount': '',
  1108. 'rovScore': '', 'pushFrom': 'flow_pool', 'abCode': self.ab_code}, ....]
  1109. :return:
  1110. """
  1111. try:
  1112. redis_helper = RedisHelper()
  1113. for item in videos:
  1114. video_id, flow_pool = item['videoId'], item['flowPool']
  1115. key_name = f"{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{video_id}:{flow_pool}"
  1116. # 本地记录的分发数 - 1
  1117. redis_helper.decr_key(key_name=key_name, amount=1, expire_time=15 * 60)
  1118. # 对该视频做分发数检查
  1119. cur_count = redis_helper.get_data_from_redis(key_name=key_name)
  1120. # 无记录
  1121. if cur_count is None:
  1122. continue
  1123. # 本地分发数 cur_count <= 0,从所有的流量召回池移除,删除本地分发记录key
  1124. if int(cur_count) <= 0:
  1125. add_remove_log = False
  1126. redis_helper.del_keys(key_name=key_name)
  1127. for app_name in config_.APP_TYPE:
  1128. app_type = config_.APP_TYPE.get(app_name)
  1129. flow_pool_key_list = [
  1130. f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET}{app_type}",
  1131. f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX_SET}{app_type}:{config_.QUICK_FLOW_POOL_ID}"
  1132. ]
  1133. for key in flow_pool_key_list:
  1134. remove_res = redis_helper.remove_value_from_set(key_name=key,
  1135. values=(f"{video_id}-{flow_pool}", ))
  1136. if remove_res > 0:
  1137. add_remove_log = True
  1138. video_flow_pool_key_list = [
  1139. f"{config_.QUICK_FLOWPOOL_VIDEO_INFO_KEY_NAME_PREFIX}{app_type}:{config_.QUICK_FLOW_POOL_ID}:{video_id}",
  1140. f"{config_.FLOWPOOL_VIDEO_INFO_KEY_NAME_PREFIX}{app_type}:{video_id}"
  1141. ]
  1142. for key in video_flow_pool_key_list:
  1143. redis_helper.remove_value_from_set(key_name=key, values=(flow_pool, ))
  1144. if add_remove_log is True:
  1145. log_.info({'tag': 'remove video_id from flow_pool', 'video_id': video_id, 'flow_pool': flow_pool})
  1146. # if redis_helper.key_exists(key_name=key_name):
  1147. # # 该视频本地有记录,本地记录的分发数 - 1
  1148. # redis_helper.decr_key(key_name=key_name, amount=1, expire_time=5 * 60)
  1149. # else:
  1150. # # 该视频本地无记录,接口获取的分发数 - 1
  1151. # redis_helper.incr_key(key_name=key_name, amount=int(item['distributeCount']) - 1, expire_time=5 * 60)
  1152. except Exception as e:
  1153. log_.error('update_local_distribute_count error...')
  1154. log_.error(traceback.format_exc())
  1155. def update_local_distribute_count_new_with_level(videos, level_weight):
  1156. """
  1157. 更新本地分发数
  1158. :param videos: 视频列表 type-list [{'videoId':'', 'flowPool':'', 'distributeCount': '',
  1159. 'rovScore': '', 'pushFrom': 'flow_pool', 'abCode': self.ab_code}, ....]
  1160. :return:
  1161. """
  1162. try:
  1163. redis_helper = RedisHelper()
  1164. # level_weight = redis_helper.get_data_from_redis(key_name=config_.FLOWPOOL_LEVEL_WEIGHT_KEY_NAME)
  1165. # level_list = [level for level in json.loads(level_weight)]
  1166. if level_weight is None:
  1167. level_weight = {'1': 1, '2': 1, '3': 1, '4': 1, '5': 1, '6': 1}
  1168. level_list = [level for level in level_weight]
  1169. for item in videos:
  1170. video_id, flow_pool = item['videoId'], item['flowPool']
  1171. key_name = f"{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{video_id}:{flow_pool}"
  1172. # 本地记录的分发数 - 1
  1173. redis_helper.decr_key(key_name=key_name, amount=1, expire_time=15 * 60)
  1174. # 对该视频做分发数检查
  1175. cur_count = redis_helper.get_data_from_redis(key_name=key_name)
  1176. # 无记录
  1177. if cur_count is None:
  1178. continue
  1179. # 本地分发数 cur_count <= 0,从所有的流量召回池移除,删除本地分发记录key
  1180. if int(cur_count) <= 0:
  1181. add_remove_log = False
  1182. redis_helper.del_keys(key_name=key_name)
  1183. for app_name in config_.APP_TYPE:
  1184. app_type = config_.APP_TYPE.get(app_name)
  1185. flow_pool_key_list = [
  1186. f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX_SET}{app_type}:{config_.QUICK_FLOW_POOL_ID}"
  1187. ]
  1188. for level in level_list:
  1189. flow_pool_key_list.append(f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL}{app_type}:{level}")
  1190. for key in flow_pool_key_list:
  1191. remove_res = redis_helper.remove_value_from_set(key_name=key,
  1192. values=(f"{video_id}-{flow_pool}", ))
  1193. if remove_res > 0:
  1194. add_remove_log = True
  1195. if add_remove_log is True:
  1196. log_.info({'tag': 'remove video_id from flow_pool', 'video_id': video_id, 'flow_pool': flow_pool})
  1197. # if redis_helper.key_exists(key_name=key_name):
  1198. # # 该视频本地有记录,本地记录的分发数 - 1
  1199. # redis_helper.decr_key(key_name=key_name, amount=1, expire_time=5 * 60)
  1200. # else:
  1201. # # 该视频本地无记录,接口获取的分发数 - 1
  1202. # redis_helper.incr_key(key_name=key_name, amount=int(item['distributeCount']) - 1, expire_time=5 * 60)
  1203. except Exception as e:
  1204. log_.error('update_local_distribute_count error...')
  1205. log_.error(traceback.format_exc())
  1206. def get_religion_class_with_mid(mid, religion_class_name):
  1207. """
  1208. 判断用户是否属于对应的宗教类型
  1209. :param mid: mid type-string
  1210. :param religion_class_name: 宗教类型, type-string, (catholicism-天主教, christianity-基督教)
  1211. :return: religion_class_flag, type-int, (0-否,1-是), 默认: 0
  1212. """
  1213. religion_class_flag = 0
  1214. now_date = datetime.today()
  1215. redis_helper = RedisHelper()
  1216. if mid:
  1217. hash_mid = hashlib.md5(mid.encode('utf-8')).hexdigest()
  1218. hash_tag = hash_mid[-1:]
  1219. key_name_prefix = config_.KEY_NAME_PREFIX_RELIGION_USER.get(religion_class_name, None)
  1220. if key_name_prefix is None:
  1221. return religion_class_flag
  1222. key_name = f"{key_name_prefix}{hash_tag}:{datetime.strftime(now_date, '%Y%m%d')}"
  1223. if not redis_helper.key_exists(key_name=key_name):
  1224. key_name = f"{key_name_prefix}{hash_tag}:{datetime.strftime(now_date - timedelta(days=1), '%Y%m%d')}"
  1225. if redis_helper.data_exists_with_set(key_name=key_name, value=mid):
  1226. religion_class_flag = 1
  1227. return religion_class_flag
  1228. def get_recommend_params(recommend_type, ab_exp_info, ab_info_data, mid, app_type, page_type=0,
  1229. flow_pool_abtest_config=None):
  1230. """
  1231. 根据实验分组给定对应的推荐参数
  1232. :param recommend_type: 首页推荐和相关推荐区分参数(0-首页推荐,1-相关推荐)
  1233. :param ab_exp_info: AB实验组参数
  1234. :param ab_info_data: app实验组参数
  1235. :param mid: mid
  1236. :param app_type: app_type, type-int
  1237. :param page_type: 页面区分参数,默认:0(首页)
  1238. :param flow_pool_abtest_config: 流量池abtest配置
  1239. :return:
  1240. """
  1241. top_K = config_.K
  1242. flow_pool_P = config_.P
  1243. # 不获取人工干预数据标记
  1244. no_op_flag = True
  1245. expire_time = 3600
  1246. old_video_index = -1
  1247. # 获取对应的默认配置
  1248. ab_initial_config = config_.INITIAL_CONFIG.get(app_type, None)
  1249. if ab_initial_config is None:
  1250. ab_initial_config = config_.INITIAL_CONFIG.get('other')
  1251. param = config_.AB_EXP_CODE[ab_initial_config]
  1252. ab_code = param.get('ab_code')
  1253. rule_key = param.get('rule_key')
  1254. data_key = param.get('data_key')
  1255. rule_key_30day = param.get('30day_rule_key')
  1256. shield_config = config_.SHIELD_CONFIG
  1257. rank_key_prefix = 'rank:score1:'
  1258. # 默认使用 095 实验的配置
  1259. # ab_code = config_.AB_EXP_CODE['095'].get('ab_code')
  1260. # rule_key = config_.AB_EXP_CODE['095'].get('rule_key')
  1261. # data_key = config_.AB_EXP_CODE['095'].get('data_key')
  1262. # rule_key_30day = None
  1263. # 获取用户近30天是否有回流
  1264. # user_30day_return_result = get_user_has30day_return(mid=mid)
  1265. # 获取实验配置
  1266. if ab_exp_info:
  1267. ab_exp_code_list = []
  1268. config_value_dict = {}
  1269. for _, item in ab_exp_info.items():
  1270. if not item:
  1271. continue
  1272. for ab_item in item:
  1273. ab_exp_code = ab_item.get('abExpCode', None)
  1274. if not ab_exp_code:
  1275. continue
  1276. ab_exp_code_list.append(str(ab_exp_code))
  1277. config_value_dict[str(ab_exp_code)] = ab_item.get('configValue', None)
  1278. # 流量池视频分发概率实验
  1279. if '211' in ab_exp_code_list:
  1280. flow_pool_P = 0.9
  1281. elif '221' in ab_exp_code_list:
  1282. flow_pool_P = 0.7
  1283. elif '299' in ab_exp_code_list:
  1284. flow_pool_P = 0.5
  1285. elif '300' in ab_exp_code_list:
  1286. flow_pool_P = 0.4
  1287. elif '301' in ab_exp_code_list:
  1288. flow_pool_P = 0.6
  1289. elif '339' in ab_exp_code_list:
  1290. flow_pool_P = 0
  1291. # if '136' in ab_exp_code_list:
  1292. # # 无回流 - 消费人群
  1293. # if user_30day_return_result == 0:
  1294. # param = config_.AB_EXP_CODE.get('136')
  1295. # ab_code = param.get('ab_code')
  1296. # expire_time = 3600
  1297. # rule_key = param.get('rule_key')
  1298. # data_key = param.get('data_key')
  1299. # no_op_flag = True
  1300. # elif '137' in ab_exp_code_list:
  1301. # # 有回流 - 分享人群
  1302. # if user_30day_return_result == 1:
  1303. # param = config_.AB_EXP_CODE.get('137')
  1304. # ab_code = param.get('ab_code')
  1305. # expire_time = 3600
  1306. # rule_key = param.get('rule_key')
  1307. # data_key = param.get('data_key')
  1308. # no_op_flag = True
  1309. # elif '161' in ab_exp_code_list:
  1310. # # 无回流 - 消费人群
  1311. # if user_30day_return_result == 0:
  1312. # param = config_.AB_EXP_CODE.get('136')
  1313. # ab_code = param.get('ab_code')
  1314. # expire_time = 3600
  1315. # rule_key = param.get('rule_key')
  1316. # data_key = param.get('data_key')
  1317. # no_op_flag = True
  1318. # # 有回流 - 分享人群
  1319. # else:
  1320. # param = config_.AB_EXP_CODE.get('137')
  1321. # ab_code = param.get('ab_code')
  1322. # expire_time = 3600
  1323. # rule_key = param.get('rule_key')
  1324. # data_key = param.get('data_key')
  1325. # no_op_flag = True
  1326. # elif '162' in ab_exp_code_list:
  1327. # # 有回流
  1328. # if user_30day_return_result == 1:
  1329. # param = config_.AB_EXP_CODE.get('162')
  1330. # ab_code = param.get('ab_code')
  1331. # expire_time = 3600
  1332. # rule_key = param.get('rule_key')
  1333. # data_key = param.get('data_key')
  1334. # no_op_flag = True
  1335. # 老好看视频 宗教人群实验
  1336. if '228' in ab_exp_code_list:
  1337. # 天主教
  1338. religion_param = config_.AB_EXP_CODE['228']
  1339. religion_class_name = religion_param.get('religion_class_name')
  1340. religion_class_flag = get_religion_class_with_mid(mid=mid, religion_class_name=religion_class_name)
  1341. if religion_class_flag == 1:
  1342. ab_code = religion_param.get('ab_code')
  1343. rule_key = religion_param.get('rule_key')
  1344. data_key = religion_param.get('data_key')
  1345. rule_key_30day = religion_param.get('30day_rule_key')
  1346. elif '229' in ab_exp_code_list:
  1347. # 基督教
  1348. religion_param = config_.AB_EXP_CODE['229']
  1349. religion_class_name = religion_param.get('religion_class_name')
  1350. religion_class_flag = get_religion_class_with_mid(mid=mid, religion_class_name=religion_class_name)
  1351. if religion_class_flag == 1:
  1352. ab_code = religion_param.get('ab_code')
  1353. rule_key = religion_param.get('rule_key')
  1354. data_key = religion_param.get('data_key')
  1355. rule_key_30day = religion_param.get('30day_rule_key')
  1356. else:
  1357. for code, param in config_.AB_EXP_CODE.items():
  1358. if code in ab_exp_code_list:
  1359. ab_code = param.get('ab_code')
  1360. rule_key = param.get('rule_key')
  1361. data_key = param.get('data_key')
  1362. rule_key_30day = param.get('30day_rule_key')
  1363. shield_config = param.get('shield_config', config_.SHIELD_CONFIG)
  1364. rank_key_prefix = param.get('rank_key_prefix', 'rank:score1:')
  1365. break
  1366. """
  1367. # 推荐条数 10->4 实验
  1368. # if config_.AB_EXP_CODE['rec_size_home'] in ab_exp_code_list:
  1369. # config_value = config_value_dict.get(config_.AB_EXP_CODE['rec_size_home'], None)
  1370. # if config_value:
  1371. # config_value = eval(str(config_value))
  1372. # else:
  1373. # config_value = {}
  1374. # log_.info(f'config_value: {config_value}, type: {type(config_value)}')
  1375. # size = int(config_value.get('size', 4))
  1376. # top_K = int(config_value.get('K', 3))
  1377. # flow_pool_P = float(config_value.get('P', 0.3))
  1378. # else:
  1379. # size = size
  1380. # top_K = config_.K
  1381. # flow_pool_P = config_.P
  1382. # 算法实验相对对照组
  1383. # if config_.AB_EXP_CODE['ab_initial'] in ab_exp_code_list:
  1384. # ab_code = config_.AB_CODE['ab_initial']
  1385. # expire_time = 24 * 3600
  1386. # rule_key = config_.RULE_KEY['initial']
  1387. # no_op_flag = True
  1388. # 小时级更新-规则1 实验
  1389. # elif config_.AB_EXP_CODE['rule_rank1'] in ab_exp_code_list:
  1390. # ab_code = config_.AB_CODE['rank_by_h'].get('rule_rank1')
  1391. # expire_time = 3600
  1392. # rule_key = config_.RULE_KEY['rule_rank1']
  1393. # no_op_flag = True
  1394. # elif config_.AB_EXP_CODE['rule_rank2'] in ab_exp_code_list:
  1395. # ab_code = config_.AB_CODE['rank_by_h'].get('rule_rank2')
  1396. # expire_time = 3600
  1397. # rule_key = config_.RULE_KEY['rule_rank2']
  1398. # elif config_.AB_EXP_CODE['rule_rank3'] in ab_exp_code_list:
  1399. # ab_code = config_.AB_CODE['rank_by_h'].get('rule_rank3')
  1400. # expire_time = 3600
  1401. # rule_key = config_.RULE_KEY['rule_rank3']
  1402. # no_op_flag = True
  1403. # elif config_.AB_EXP_CODE['rule_rank4'] in ab_exp_code_list:
  1404. # ab_code = config_.AB_CODE['rank_by_h'].get('rule_rank4')
  1405. # expire_time = 3600
  1406. # rule_key = config_.RULE_KEY['rule_rank4']
  1407. # elif config_.AB_EXP_CODE['rule_rank5'] in ab_exp_code_list:
  1408. # ab_code = config_.AB_CODE['rank_by_h'].get('rule_rank5')
  1409. # expire_time = 3600
  1410. # rule_key = config_.RULE_KEY['rule_rank5']
  1411. # elif config_.AB_EXP_CODE['day_rule_rank1'] in ab_exp_code_list:
  1412. # ab_code = config_.AB_CODE['rank_by_day'].get('day_rule_rank1')
  1413. # expire_time = 24 * 3600
  1414. # rule_key = config_.RULE_KEY_DAY['day_rule_rank1']
  1415. # no_op_flag = True
  1416. # if config_.AB_EXP_CODE['rule_rank6'] in ab_exp_code_list:
  1417. # ab_code = config_.AB_CODE['rank_by_h'].get('rule_rank6')
  1418. # expire_time = 3600
  1419. # rule_key = config_.RULE_KEY['rule_rank6']
  1420. # no_op_flag = True
  1421. # elif config_.AB_EXP_CODE['day_rule_rank2'] in ab_exp_code_list:
  1422. # ab_code = config_.AB_CODE['rank_by_day'].get('day_rule_rank2')
  1423. # expire_time = 24 * 3600
  1424. # rule_key = config_.RULE_KEY_DAY['day_rule_rank2']
  1425. # no_op_flag = True
  1426. # elif config_.AB_EXP_CODE['region_rule_rank1'] in ab_exp_code_list:
  1427. # ab_code = config_.AB_CODE['region_rank_by_h'].get('region_rule_rank1')
  1428. # expire_time = 3600
  1429. # rule_key = config_.RULE_KEY_REGION['region_rule_rank1']
  1430. # no_op_flag = True
  1431. # elif config_.AB_EXP_CODE['24h_rule_rank1'] in ab_exp_code_list:
  1432. # ab_code = config_.AB_CODE['rank_by_24h'].get('24h_rule_rank1')
  1433. # expire_time = 3600
  1434. # rule_key = config_.RULE_KEY_24H['24h_rule_rank1']
  1435. # no_op_flag = True
  1436. # elif config_.AB_EXP_CODE['24h_rule_rank2'] in ab_exp_code_list:
  1437. # ab_code = config_.AB_CODE['rank_by_24h'].get('24h_rule_rank2')
  1438. # expire_time = 3600
  1439. # rule_key = config_.RULE_KEY_24H['24h_rule_rank2']
  1440. # no_op_flag = True
  1441. # elif config_.AB_EXP_CODE['region_rule_rank2'] in ab_exp_code_list:
  1442. # ab_code = config_.AB_CODE['region_rank_by_h'].get('region_rule_rank2')
  1443. # expire_time = 3600
  1444. # rule_key = config_.RULE_KEY_REGION['region_rule_rank2']
  1445. # no_op_flag = True
  1446. # if config_.AB_EXP_CODE['region_rule_rank3'] in ab_exp_code_list or\
  1447. # config_.AB_EXP_CODE['region_rule_rank3_appType_19'] in ab_exp_code_list or\
  1448. # config_.AB_EXP_CODE['region_rule_rank3_appType_4'] in ab_exp_code_list or\
  1449. # config_.AB_EXP_CODE['region_rule_rank3_appType_6'] in ab_exp_code_list or\
  1450. # config_.AB_EXP_CODE['region_rule_rank3_appType_18'] in ab_exp_code_list:
  1451. # ab_code = config_.AB_CODE['region_rank_by_h'].get('region_rule_rank3')
  1452. # expire_time = 3600
  1453. # rule_key = config_.RULE_KEY_REGION['region_rule_rank3'].get('rule_key')
  1454. # data_key = config_.RULE_KEY_REGION['region_rule_rank3'].get('data_key')
  1455. # no_op_flag = True
  1456. # if config_.AB_EXP_CODE['region_rule_rank4'] in ab_exp_code_list or\
  1457. if config_.AB_EXP_CODE['region_rule_rank4_appType_19'] in ab_exp_code_list or \
  1458. config_.AB_EXP_CODE['region_rule_rank4_appType_4'] in ab_exp_code_list or\
  1459. config_.AB_EXP_CODE['region_rule_rank4_appType_6'] in ab_exp_code_list or\
  1460. config_.AB_EXP_CODE['region_rule_rank4_appType_18'] in ab_exp_code_list:
  1461. ab_code = config_.AB_CODE['region_rank_by_h'].get('region_rule_rank4')
  1462. expire_time = 3600
  1463. rule_key = config_.RULE_KEY_REGION['region_rule_rank4'].get('rule_key')
  1464. data_key = config_.RULE_KEY_REGION['region_rule_rank4'].get('data_key')
  1465. no_op_flag = True
  1466. # elif config_.AB_EXP_CODE['region_rule_rank4_appType_5_data1'] in ab_exp_code_list:
  1467. # ab_code = config_.AB_CODE['region_rank_by_h'].get('region_rule_rank4')
  1468. # expire_time = 3600
  1469. # rule_key = config_.RULE_KEY_REGION['region_rule_rank4_appType_5_data1'].get('rule_key')
  1470. # data_key = config_.RULE_KEY_REGION['region_rule_rank4_appType_5_data1'].get('data_key')
  1471. # no_op_flag = True
  1472. # elif config_.AB_EXP_CODE['region_rule_rank3_appType_5_data2'] in ab_exp_code_list:
  1473. # ab_code = config_.AB_CODE['region_rank_by_h'].get('region_rule_rank3_appType_5_data2')
  1474. # expire_time = 3600
  1475. # rule_key = config_.RULE_KEY_REGION['region_rule_rank3_appType_5_data2'].get('rule_key')
  1476. # data_key = config_.RULE_KEY_REGION['region_rule_rank3_appType_5_data2'].get('data_key')
  1477. # no_op_flag = True
  1478. elif config_.AB_EXP_CODE['region_rule_rank4_appType_5_data3'] in ab_exp_code_list:
  1479. ab_code = config_.AB_CODE['region_rank_by_h'].get('region_rule_rank4_appType_5_data3')
  1480. expire_time = 3600
  1481. rule_key = config_.RULE_KEY_REGION['region_rule_rank4_appType_5_data3'].get('rule_key')
  1482. data_key = config_.RULE_KEY_REGION['region_rule_rank4_appType_5_data3'].get('data_key')
  1483. no_op_flag = True
  1484. elif config_.AB_EXP_CODE['region_rule_rank4_appType_5_data4'] in ab_exp_code_list:
  1485. ab_code = config_.AB_CODE['region_rank_by_h'].get('region_rule_rank4_appType_5_data4')
  1486. expire_time = 3600
  1487. rule_key = config_.RULE_KEY_REGION['region_rule_rank4_appType_5_data4'].get('rule_key')
  1488. data_key = config_.RULE_KEY_REGION['region_rule_rank4_appType_5_data4'].get('data_key')
  1489. no_op_flag = True
  1490. elif config_.AB_EXP_CODE['region_rule_rank4_appType_0_data2'] in ab_exp_code_list:
  1491. ab_code = config_.AB_CODE['region_rank_by_h'].get('region_rule_rank4_appType_0_data2')
  1492. expire_time = 3600
  1493. rule_key = config_.RULE_KEY_REGION['region_rule_rank4_appType_0_data2'].get('rule_key')
  1494. data_key = config_.RULE_KEY_REGION['region_rule_rank4_appType_0_data2'].get('data_key')
  1495. no_op_flag = True
  1496. # elif config_.AB_EXP_CODE['region_rule_rank4_appType_19_data2'] in ab_exp_code_list:
  1497. # ab_code = config_.AB_CODE['region_rank_by_h'].get('region_rule_rank4_appType_19_data2')
  1498. # expire_time = 3600
  1499. # rule_key = config_.RULE_KEY_REGION['region_rule_rank4_appType_19_data2'].get('rule_key')
  1500. # data_key = config_.RULE_KEY_REGION['region_rule_rank4_appType_19_data2'].get('data_key')
  1501. # no_op_flag = True
  1502. # elif config_.AB_EXP_CODE['region_rule_rank4_appType_19_data3'] in ab_exp_code_list:
  1503. # ab_code = config_.AB_CODE['region_rank_by_h'].get('region_rule_rank4_appType_19_data3')
  1504. # expire_time = 3600
  1505. # rule_key = config_.RULE_KEY_REGION['region_rule_rank4_appType_19_data3'].get('rule_key')
  1506. # data_key = config_.RULE_KEY_REGION['region_rule_rank4_appType_19_data3'].get('data_key')
  1507. # no_op_flag = True
  1508. elif config_.AB_EXP_CODE['region_rule_rank5_appType_0_data1'] in ab_exp_code_list:
  1509. ab_code = config_.AB_CODE['region_rank_by_h'].get('region_rule_rank5_appType_0_data1')
  1510. expire_time = 3600
  1511. rule_key = config_.RULE_KEY_REGION['region_rule_rank5_appType_0_data1'].get('rule_key')
  1512. data_key = config_.RULE_KEY_REGION['region_rule_rank5_appType_0_data1'].get('data_key')
  1513. no_op_flag = True
  1514. elif config_.AB_EXP_CODE['region_rule_rank4_appType_4_data2'] in ab_exp_code_list:
  1515. ab_code = config_.AB_CODE['region_rank_by_h'].get('region_rule_rank4_appType_4_data2')
  1516. expire_time = 3600
  1517. rule_key = config_.RULE_KEY_REGION['region_rule_rank4_appType_4_data2'].get('rule_key')
  1518. data_key = config_.RULE_KEY_REGION['region_rule_rank4_appType_4_data2'].get('data_key')
  1519. no_op_flag = True
  1520. elif config_.AB_EXP_CODE['region_rule_rank4_appType_4_data3'] in ab_exp_code_list:
  1521. ab_code = config_.AB_CODE['region_rank_by_h'].get('region_rule_rank4_appType_4_data3')
  1522. expire_time = 3600
  1523. rule_key = config_.RULE_KEY_REGION['region_rule_rank4_appType_4_data3'].get('rule_key')
  1524. data_key = config_.RULE_KEY_REGION['region_rule_rank4_appType_4_data3'].get('data_key')
  1525. no_op_flag = True
  1526. elif config_.AB_EXP_CODE['region_rule_rank4_appType_6_data2'] in ab_exp_code_list:
  1527. ab_code = config_.AB_CODE['region_rank_by_h'].get('region_rule_rank4_appType_6_data2')
  1528. expire_time = 3600
  1529. rule_key = config_.RULE_KEY_REGION['region_rule_rank4_appType_6_data2'].get('rule_key')
  1530. data_key = config_.RULE_KEY_REGION['region_rule_rank4_appType_6_data2'].get('data_key')
  1531. no_op_flag = True
  1532. elif config_.AB_EXP_CODE['region_rule_rank4_appType_6_data3'] in ab_exp_code_list:
  1533. ab_code = config_.AB_CODE['region_rank_by_h'].get('region_rule_rank4_appType_6_data3')
  1534. expire_time = 3600
  1535. rule_key = config_.RULE_KEY_REGION['region_rule_rank4_appType_6_data3'].get('rule_key')
  1536. data_key = config_.RULE_KEY_REGION['region_rule_rank4_appType_6_data3'].get('data_key')
  1537. no_op_flag = True
  1538. # elif config_.AB_EXP_CODE['region_rule_rank4_appType_18_data2'] in ab_exp_code_list:
  1539. # ab_code = config_.AB_CODE['region_rank_by_h'].get('region_rule_rank4_appType_18_data2')
  1540. # expire_time = 3600
  1541. # rule_key = config_.RULE_KEY_REGION['region_rule_rank4_appType_18_data2'].get('rule_key')
  1542. # data_key = config_.RULE_KEY_REGION['region_rule_rank4_appType_18_data2'].get('data_key')
  1543. # no_op_flag = True
  1544. # elif config_.AB_EXP_CODE['region_rule_rank6_appType_0_data1'] in ab_exp_code_list:
  1545. # ab_code = config_.AB_CODE['region_rank_by_h'].get('region_rule_rank6_appType_0_data1')
  1546. # expire_time = 3600
  1547. # rule_key = config_.RULE_KEY_REGION['region_rule_rank6_appType_0_data1'].get('rule_key')
  1548. # data_key = config_.RULE_KEY_REGION['region_rule_rank6_appType_0_data1'].get('data_key')
  1549. # no_op_flag = True
  1550. else:
  1551. ab_code = config_.AB_CODE['initial']
  1552. expire_time = 24 * 3600
  1553. rule_key = config_.RULE_KEY_REGION['initial'].get('rule_key')
  1554. data_key = config_.RULE_KEY_REGION['initial'].get('data_key')
  1555. # # 老好看视频 / 票圈最惊奇 首页/相关推荐逻辑更新实验
  1556. # if config_.AB_EXP_CODE['rov_rank_appType_18_19'] in ab_exp_code_list:
  1557. # ab_code = config_.AB_CODE['rov_rank_appType_18_19']
  1558. # expire_time = 3600
  1559. # flow_pool_P = config_.P_18_19
  1560. # no_op_flag = True
  1561. #
  1562. # elif config_.AB_EXP_CODE['rov_rank_appType_19'] in ab_exp_code_list:
  1563. # ab_code = config_.AB_CODE['rov_rank_appType_19']
  1564. # expire_time = 3600
  1565. # top_K = 0
  1566. # flow_pool_P = config_.P_18_19
  1567. # no_op_flag = True
  1568. #
  1569. # elif config_.AB_EXP_CODE['top_video_relevant_appType_19'] in ab_exp_code_list and page_type == 2:
  1570. # ab_code = config_.AB_CODE['top_video_relevant_appType_19']
  1571. # expire_time = 3600
  1572. # top_K = 1
  1573. # flow_pool_P = config_.P_18_19
  1574. # no_op_flag = True
  1575. #
  1576. # # 票圈最惊奇完整影视资源实验
  1577. # elif config_.AB_EXP_CODE['whole_movies'] in ab_exp_code_list:
  1578. # ab_code = config_.AB_CODE['whole_movies']
  1579. # expire_time = 24 * 3600
  1580. # no_op_flag = True
  1581. # 老视频实验
  1582. # if config_.AB_EXP_CODE['old_video'] in ab_exp_code_list:
  1583. # ab_code = config_.AB_CODE['old_video']
  1584. # no_op_flag = True
  1585. # old_video_index = 2
  1586. # else:
  1587. # old_video_index = -1
  1588. """
  1589. # APP实验组
  1590. if ab_info_data:
  1591. ab_info_app = {}
  1592. for page_code, item in json.loads(ab_info_data).items():
  1593. if not item:
  1594. continue
  1595. ab_info_code = item.get('eventId', None)
  1596. if ab_info_code:
  1597. ab_info_app[page_code] = ab_info_code
  1598. # print(f"======{ab_info_app}")
  1599. # 首页推荐
  1600. if recommend_type == 0:
  1601. app_ab_code = ab_info_app.get('10003', None)
  1602. for code, param in config_.APP_AB_CODE['10003'].items():
  1603. if code == app_ab_code:
  1604. ab_code = param.get('ab_code')
  1605. rule_key = param.get('rule_key')
  1606. data_key = param.get('data_key')
  1607. break
  1608. # # 相关推荐
  1609. # elif recommend_type == 1:
  1610. # if config_.APP_AB_CODE['10037'] == ab_info_app.get('10037', None):
  1611. # ab_code = config_.AB_CODE['region_rank_by_h'].get('region_rule_rank4')
  1612. # expire_time = 3600
  1613. # rule_key = 'rule3'
  1614. # data_key = 'data1'
  1615. # no_op_flag = True
  1616. # 流量池分发实验组划分
  1617. # 1. 随机选取流量池id
  1618. flow_pool_id_choice = random.choice(config_.FLOWPOOL_ID_LIST)
  1619. # 2. 判断流量id所属实验配置分组
  1620. flow_pool_abtest_group = 'control_group'
  1621. for key, items in flow_pool_abtest_config.items():
  1622. if int(flow_pool_id_choice) in items:
  1623. flow_pool_abtest_group = key
  1624. # log_.info(f"flow_pool_id_choice: {flow_pool_id_choice}, flow_pool_abtest_group: {flow_pool_abtest_group}")
  1625. return top_K, flow_pool_P, ab_code, rule_key, data_key, expire_time, no_op_flag, old_video_index, rule_key_30day, \
  1626. shield_config, flow_pool_abtest_group, rank_key_prefix
  1627. def video_homepage_recommend(request_id, mid, uid, size, app_type, algo_type,
  1628. client_info, ab_exp_info, params, ab_info_data, version_audit_status, env_dict,
  1629. level_weight, flow_pool_abtest_config):
  1630. """
  1631. 首页线上推荐逻辑
  1632. :param request_id: request_id
  1633. :param mid: mid type-string
  1634. :param uid: uid type-string
  1635. :param size: 请求视频数量 type-int
  1636. :param app_type: 产品标识 type-int
  1637. :param algo_type: 算法类型 type-string
  1638. :param client_info: 用户位置信息 {"country": "国家", "province": "省份", "city": "城市"}
  1639. :param ab_exp_info: ab实验分组参数 [{"expItemId":1, "configValue":{"size":4, "K":3, ...}}, ...]
  1640. :param params:
  1641. :param ab_info_data: app实验分组参数
  1642. :param version_audit_status: 小程序版本审核参数:1-审核中,2-审核通过
  1643. :return:
  1644. """
  1645. # 对 vlog 切换10%的流量做实验
  1646. # 对mid进行哈希
  1647. # hash_mid = hashlib.md5(mid.encode('utf-8')).hexdigest()
  1648. # if app_type in config_.AB_TEST['rank_by_h'] and hash_mid[-1:] in ['8', '0', 'a', 'b']:
  1649. # # 简单召回 - 排序 - 兜底
  1650. # rank_result, last_rov_recall_key = video_recommend(mid=mid, uid=uid, size=size, app_type=app_type,
  1651. # algo_type=algo_type, client_info=client_info,
  1652. # expire_time=3600,
  1653. # ab_code=config_.AB_CODE['rank_by_h'])
  1654. # # ab-test
  1655. # result = ab_test_op(rank_result=rank_result,
  1656. # ab_code_list=[config_.AB_CODE['position_insert']],
  1657. # app_type=app_type, mid=mid, uid=uid)
  1658. # # redis数据刷新
  1659. # update_redis_data(result=result, app_type=app_type, mid=mid, last_rov_recall_key=last_rov_recall_key,
  1660. # expire_time=3600)
  1661. # if app_type == config_.APP_TYPE['APP']:
  1662. # # 票圈视频APP
  1663. # top_K = config_.K
  1664. # flow_pool_P = config_.P
  1665. # # 简单召回 - 排序 - 兜底
  1666. # rank_result, last_rov_recall_key = video_recommend(request_id=request_id,
  1667. # mid=mid, uid=uid, app_type=app_type,
  1668. # size=size, top_K=top_K, flow_pool_P=flow_pool_P,
  1669. # algo_type=algo_type, client_info=client_info,
  1670. # expire_time=12 * 3600, params=params)
  1671. # # ab-test
  1672. # # result = ab_test_op(rank_result=rank_result,
  1673. # # ab_code_list=[config_.AB_CODE['position_insert']],
  1674. # # app_type=app_type, mid=mid, uid=uid)
  1675. # # redis数据刷新
  1676. # update_redis_data(result=rank_result, app_type=app_type, mid=mid, last_rov_recall_key=last_rov_recall_key,
  1677. # top_K=top_K, expire_time=12 * 3600)
  1678. #
  1679. # else:
  1680. recommend_result = {}
  1681. param_st = time.time()
  1682. recommend_result['fea_info'] = env_dict
  1683. # 特殊mid 和 小程序审核版本推荐处理
  1684. if mid in get_special_mid_list() or version_audit_status == 1:
  1685. rank_result = special_mid_recommend(request_id=request_id, mid=mid, uid=uid, app_type=app_type, size=size)
  1686. recommend_result['videos'] = rank_result
  1687. return recommend_result
  1688. # 普通mid推荐处理
  1689. top_K, flow_pool_P, ab_code, rule_key, data_key, expire_time, \
  1690. no_op_flag, old_video_index, rule_key_30day, shield_config, flow_pool_abtest_group, rank_key_prefix = \
  1691. get_recommend_params(recommend_type=0, ab_exp_info=ab_exp_info, ab_info_data=ab_info_data, mid=mid,
  1692. app_type=app_type, flow_pool_abtest_config=flow_pool_abtest_config)
  1693. # log_.info({
  1694. # 'logTimestamp': int(time.time() * 1000),
  1695. # 'request_id': request_id,
  1696. # 'app_type': app_type,
  1697. # 'mid': mid,
  1698. # 'uid': uid,
  1699. # 'operation': 'get_recommend_params',
  1700. # 'executeTime': (time.time() - param_st) * 1000
  1701. # })
  1702. recommend_result['getRecommendParamsTime'] = (time.time() - param_st) * 1000
  1703. # 简单召回 - 排序 - 兜底
  1704. get_result_st = time.time()
  1705. #print("ab_code:", ab_code)
  1706. #new pipeline
  1707. # if ab_code == 60047 or ab_code == 60048 or ab_code == 60049:
  1708. # result = new_video_recommend(request_id=request_id,
  1709. # mid=mid, uid=uid, app_type=app_type,
  1710. # size=size, top_K=top_K, flow_pool_P=flow_pool_P,
  1711. # algo_type=algo_type, client_info=client_info,
  1712. # ab_code=ab_code, expire_time=expire_time,
  1713. # rule_key=rule_key, data_key=data_key,
  1714. # no_op_flag=no_op_flag, old_video_index=old_video_index,
  1715. # params=params, rule_key_30day=rule_key_30day, shield_config=shield_config)
  1716. # simrecal: 60054 +融合, 全量
  1717. # return video, return video2
  1718. # old video: 60056, test2
  1719. if ab_code == 60054 or ab_code == 60056 or ab_code == 60067 or ab_code == 60068 or ab_code == 60066 \
  1720. or ab_code == 60069 or ab_code == 60070 or ab_code == 60071 or ab_code == 60072 or ab_code == 60073 \
  1721. or ab_code == 60074 or ab_code == 60075 or ab_code == 60076 or ab_code == 60077 \
  1722. or ab_code == 60078 or ab_code == 60079 or ab_code == 60080 or ab_code == 60081 or ab_code == 60082:
  1723. result, fea_info = video_old_recommend(request_id=request_id, mid=mid, uid=uid, app_type=app_type, size=size,
  1724. top_K=top_K, flow_pool_P=flow_pool_P, algo_type='',
  1725. client_info=client_info, ab_code=ab_code, expire_time=expire_time,
  1726. rule_key=rule_key, data_key=data_key, no_op_flag=no_op_flag,
  1727. old_video_index=old_video_index, video_id=None, params=params,
  1728. rule_key_30day=rule_key_30day, shield_config=shield_config,
  1729. env_dict=env_dict, level_weight=level_weight,
  1730. flow_pool_abtest_group=flow_pool_abtest_group,
  1731. rank_key_prefix=rank_key_prefix)
  1732. recommend_result['fea_info'] = fea_info
  1733. else:
  1734. result = video_recommend(request_id=request_id, mid=mid, uid=uid, app_type=app_type, size=size, top_K=top_K,
  1735. flow_pool_P=flow_pool_P, algo_type=algo_type, client_info=client_info,
  1736. ab_code=ab_code, expire_time=expire_time, rule_key=rule_key, data_key=data_key,
  1737. no_op_flag=no_op_flag, old_video_index=old_video_index, params=params,
  1738. rule_key_30day=rule_key_30day, shield_config=shield_config, level_weight=level_weight,
  1739. flow_pool_abtest_group=flow_pool_abtest_group)
  1740. # log_.info({
  1741. # 'logTimestamp': int(time.time() * 1000),
  1742. # 'request_id': request_id,
  1743. # 'app_type': app_type,
  1744. # 'mid': mid,
  1745. # 'uid': uid,
  1746. # 'operation': 'get_recommend_result',
  1747. # 'executeTime': (time.time() - get_result_st) * 1000
  1748. # })
  1749. recommend_result['recommendOperation'] = result
  1750. rank_result = result.get('rankResult')
  1751. for i, item in enumerate(rank_result):
  1752. item['position'] = i + 1
  1753. recommend_result['videos'] = rank_result
  1754. recommend_result['getRecommendResultTime'] = (time.time() - get_result_st) * 1000
  1755. # ab-test
  1756. # result = ab_test_op(rank_result=rank_result,
  1757. # ab_code_list=[config_.AB_CODE['position_insert']],
  1758. # app_type=app_type, mid=mid, uid=uid)
  1759. # redis数据刷新
  1760. update_redis_st = time.time()
  1761. # if ab_code == 60047 or ab_code == 60048 or ab_code == 60049:
  1762. # update_flow_redis_data(result=rank_result, app_type=app_type, mid=mid, top_K=top_K)
  1763. # else:
  1764. # update_redis_data(result=rank_result, app_type=app_type, mid=mid, top_K=top_K)
  1765. update_redis_data(result=rank_result, app_type=app_type, mid=mid, top_K=top_K,
  1766. level_weight=level_weight, flow_pool_abtest_group=flow_pool_abtest_group)
  1767. # log_.info({
  1768. # 'logTimestamp': int(time.time() * 1000),
  1769. # 'request_id': request_id,
  1770. # 'app_type': app_type,
  1771. # 'mid': mid,
  1772. # 'uid': uid,
  1773. # 'operation': 'update_redis_data',
  1774. # 'executeTime': (time.time() - update_redis_st) * 1000
  1775. # })
  1776. recommend_result['updateRedisDataTime'] = (time.time() - update_redis_st) * 1000
  1777. return recommend_result
  1778. # return rank_result
  1779. def video_relevant_recommend(request_id, video_id, mid, uid, size, app_type, ab_exp_info, client_info,
  1780. page_type, params, ab_info_data, version_audit_status, env_dict,
  1781. level_weight, flow_pool_abtest_config):
  1782. """
  1783. 相关推荐逻辑
  1784. :param request_id: request_id
  1785. :param video_id: 相关推荐的头部视频id
  1786. :param mid: mid type-string
  1787. :param uid: uid type-string
  1788. :param size: 请求视频数量 type-int
  1789. :param app_type: 产品标识 type-int
  1790. :param ab_exp_info: ab实验分组参数 [{"expItemId":1, "configValue":{"size":4, "K":3, ...}}, ...]
  1791. :param client_info: 地域参数
  1792. :param page_type: 页面区分参数 1:详情页;2:分享页
  1793. :param params:
  1794. :param ab_info_data: app实验分组参数
  1795. :param version_audit_status: 小程序版本审核参数:1-审核中,2-审核通过
  1796. :return: videos type-list
  1797. """
  1798. recommend_result = {}
  1799. param_st = time.time()
  1800. recommend_result['fea_info'] = env_dict
  1801. # 特殊mid 和 小程序审核版本推荐处理
  1802. if mid in get_special_mid_list() or version_audit_status == 1:
  1803. rank_result = special_mid_recommend(request_id=request_id, mid=mid, uid=uid, app_type=app_type, size=size)
  1804. recommend_result['videos'] = rank_result
  1805. return recommend_result
  1806. # return rank_result
  1807. # 普通mid推荐处理
  1808. top_K, flow_pool_P, ab_code, rule_key, data_key, expire_time, \
  1809. no_op_flag, old_video_index, rule_key_30day, shield_config, flow_pool_abtest_group, rank_key_prefix = \
  1810. get_recommend_params(recommend_type=1, ab_exp_info=ab_exp_info, ab_info_data=ab_info_data, page_type=page_type,
  1811. mid=mid, app_type=app_type, flow_pool_abtest_config=flow_pool_abtest_config)
  1812. # log_.info({
  1813. # 'logTimestamp': int(time.time() * 1000),
  1814. # 'request_id': request_id,
  1815. # 'app_type': app_type,
  1816. # 'mid': mid,
  1817. # 'uid': uid,
  1818. # 'operation': 'get_recommend_params',
  1819. # 'executeTime': (time.time() - param_st) * 1000
  1820. # })
  1821. recommend_result['getRecommendParamsTime'] = (time.time() - param_st) * 1000
  1822. # 简单召回 - 排序 - 兜底
  1823. get_result_st = time.time()
  1824. #print("ab_code:", ab_code)
  1825. # if ab_code == 60047 or ab_code == 60048 or ab_code == 60049:
  1826. # result = new_video_recommend(request_id=request_id,
  1827. # mid=mid, uid=uid, app_type=app_type,
  1828. # size=size, top_K=top_K, flow_pool_P=flow_pool_P,
  1829. # algo_type='', client_info=client_info,
  1830. # ab_code=ab_code, expire_time=expire_time,
  1831. # rule_key=rule_key, data_key=data_key, no_op_flag=no_op_flag,
  1832. # old_video_index=old_video_index, video_id=video_id,
  1833. # params=params, rule_key_30day=rule_key_30day, shield_config=shield_config)
  1834. if ab_code == 60054 or ab_code == 60056 or ab_code == 60067 or ab_code == 60068 or ab_code == 60066 \
  1835. or ab_code == 60069 or ab_code == 60070 or ab_code == 60071 or ab_code == 60072 or ab_code == 60073 \
  1836. or ab_code == 60074 or ab_code == 60075 or ab_code == 60076 or ab_code == 60077 \
  1837. or ab_code == 60078 or ab_code == 60079 or ab_code == 60080 or ab_code == 60081 or ab_code == 60082:
  1838. result, fea_info = video_old_recommend(request_id=request_id, mid=mid, uid=uid, app_type=app_type, size=size,
  1839. top_K=top_K, flow_pool_P=flow_pool_P, algo_type='',
  1840. client_info=client_info, ab_code=ab_code, expire_time=expire_time,
  1841. rule_key=rule_key, data_key=data_key, no_op_flag=no_op_flag,
  1842. old_video_index=old_video_index, video_id=video_id, params=params,
  1843. rule_key_30day=rule_key_30day, shield_config=shield_config,
  1844. env_dict=env_dict, level_weight=level_weight,
  1845. flow_pool_abtest_group=flow_pool_abtest_group,
  1846. rank_key_prefix=rank_key_prefix)
  1847. recommend_result['fea_info'] = fea_info
  1848. else:
  1849. result = video_recommend(request_id=request_id, mid=mid, uid=uid, app_type=app_type, size=size, top_K=top_K,
  1850. flow_pool_P=flow_pool_P, algo_type='', client_info=client_info, ab_code=ab_code,
  1851. expire_time=expire_time, rule_key=rule_key, data_key=data_key, no_op_flag=no_op_flag,
  1852. old_video_index=old_video_index, video_id=video_id, params=params,
  1853. rule_key_30day=rule_key_30day, shield_config=shield_config, level_weight=level_weight,
  1854. flow_pool_abtest_group=flow_pool_abtest_group)
  1855. # log_.info({
  1856. # 'logTimestamp': int(time.time() * 1000),
  1857. # 'request_id': request_id,
  1858. # 'app_type': app_type,
  1859. # 'mid': mid,
  1860. # 'uid': uid,
  1861. # 'operation': 'get_recommend_result',
  1862. # 'executeTime': (time.time() - get_result_st) * 1000
  1863. # })
  1864. recommend_result['recommendOperation'] = result
  1865. rank_result = result.get('rankResult')
  1866. for i, item in enumerate(rank_result):
  1867. item['position'] = i+1
  1868. recommend_result['videos'] = rank_result
  1869. recommend_result['getRecommendResultTime'] = (time.time() - get_result_st) * 1000
  1870. # ab-test
  1871. # result = ab_test_op(rank_result=rank_result,
  1872. # ab_code_list=[config_.AB_CODE['position_insert'], config_.AB_CODE['relevant_video_op']],
  1873. # app_type=app_type, mid=mid, uid=uid, head_vid=video_id, size=size)
  1874. # redis数据刷新
  1875. update_redis_st = time.time()
  1876. # if ab_code == 60047 or ab_code == 60048 or ab_code == 60049:
  1877. # update_flow_redis_data(result=rank_result, app_type=app_type, mid=mid, top_K=top_K)
  1878. # else:
  1879. # update_redis_data(result=rank_result, app_type=app_type, mid=mid, top_K=top_K)
  1880. update_redis_data(result=rank_result, app_type=app_type, mid=mid, top_K=top_K,
  1881. level_weight=level_weight, flow_pool_abtest_group=flow_pool_abtest_group)
  1882. # log_.info({
  1883. # 'logTimestamp': int(time.time() * 1000),
  1884. # 'request_id': request_id,
  1885. # 'app_type': app_type,
  1886. # 'mid': mid,
  1887. # 'uid': uid,
  1888. # 'operation': 'update_redis_data',
  1889. # 'executeTime': (time.time() - update_redis_st) * 1000
  1890. # })
  1891. recommend_result['updateRedisDataTime'] = (time.time() - update_redis_st) * 1000
  1892. return recommend_result
  1893. # return rank_result
  1894. def special_mid_recommend(request_id, mid, uid, app_type, size,
  1895. ab_code=config_.AB_CODE['special_mid'],
  1896. push_from=config_.PUSH_FROM['special_mid'],
  1897. expire_time=24*3600):
  1898. redis_helper = RedisHelper()
  1899. # 特殊mid推荐指定视频列表
  1900. pool_recall = PoolRecall(request_id=request_id, app_type=app_type,
  1901. mid=mid, uid=uid, ab_code=ab_code)
  1902. # 获取相关redis key
  1903. special_key_name, redis_date = pool_recall.get_pool_redis_key(pool_type='special')
  1904. # 用户上一次在rov召回池对应的位置
  1905. last_special_recall_key = f'{config_.LAST_VIDEO_FROM_SPECIAL_POOL_PREFIX}{app_type}:{mid}:{redis_date}'
  1906. value = redis_helper.get_data_from_redis(last_special_recall_key)
  1907. if value:
  1908. idx = redis_helper.get_index_with_data(special_key_name, value)
  1909. if not idx:
  1910. idx = 0
  1911. else:
  1912. idx += 1
  1913. else:
  1914. idx = 0
  1915. recall_result = []
  1916. # 每次获取的视频数
  1917. get_size = size * 5
  1918. # 记录获取频次
  1919. freq = 0
  1920. while len(recall_result) < size:
  1921. freq += 1
  1922. if freq > config_.MAX_FREQ_FROM_ROV_POOL:
  1923. break
  1924. # 获取数据
  1925. data = redis_helper.get_data_zset_with_index(key_name=special_key_name,
  1926. start=idx, end=idx + get_size - 1,
  1927. with_scores=True)
  1928. if not data:
  1929. break
  1930. # 获取视频id,并转换类型为int,并存储为key-value{videoId: score}
  1931. # 添加视频源参数 pushFrom, abCode
  1932. temp_result = [{'videoId': int(value[0]), 'rovScore': value[1],
  1933. 'pushFrom': push_from, 'abCode': ab_code}
  1934. for value in data]
  1935. recall_result.extend(temp_result)
  1936. idx += get_size
  1937. # 将此次获取的末位视频id同步刷新到Redis中,方便下次快速定位到召回位置,过期时间为1天
  1938. if mid and recall_result:
  1939. # mid为空时,不做记录
  1940. redis_helper.set_data_to_redis(key_name=last_special_recall_key,
  1941. value=recall_result[:size][-1]['videoId'],
  1942. expire_time=expire_time)
  1943. return recall_result[:size]
  1944. def get_special_mid_list():
  1945. redis_helper = RedisHelper()
  1946. special_mid_list = redis_helper.get_data_from_set(key_name=config_.KEY_NAME_SPECIAL_MID)
  1947. if special_mid_list:
  1948. return special_mid_list
  1949. else:
  1950. return []
  1951. if __name__ == '__main__':
  1952. videos = [
  1953. {"videoId": 10136461, "rovScore": 99.971, "pushFrom": "recall_pool", "abCode": 10000},
  1954. {"videoId": 10239014, "rovScore": 99.97, "pushFrom": "recall_pool", "abCode": 10000},
  1955. {"videoId": 9851154, "rovScore": 99.969, "pushFrom": "recall_pool", "abCode": 10000},
  1956. {"videoId": 10104347, "rovScore": 99.968, "pushFrom": "recall_pool", "abCode": 10000},
  1957. {"videoId": 10141507, "rovScore": 99.967, "pushFrom": "recall_pool", "abCode": 10000},
  1958. {"videoId": 10292817, "flowPool": "2#6#2#1641780979606", "rovScore": 53.926690610816486,
  1959. "pushFrom": "flow_pool", "abCode": 10000},
  1960. {"videoId": 10224932, "flowPool": "2#5#1#1641800279644", "rovScore": 53.47890460059617, "pushFrom": "flow_pool",
  1961. "abCode": 10000},
  1962. {"videoId": 9943255, "rovScore": 99.966, "pushFrom": "recall_pool", "abCode": 10000},
  1963. {"videoId": 10282970, "flowPool": "2#5#1#1641784814103", "rovScore": 52.682815076325575,
  1964. "pushFrom": "flow_pool", "abCode": 10000},
  1965. {"videoId": 10282205, "rovScore": 99.965, "pushFrom": "recall_pool", "abCode": 10000}
  1966. ]
  1967. res = relevant_video_top_recommend(app_type=4, mid='', uid=1111, head_vid=123, videos=videos, size=10)
  1968. print(res)