video_recall.py 188 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647264826492650265126522653265426552656265726582659266026612662266326642665266626672668266926702671267226732674267526762677267826792680268126822683268426852686268726882689269026912692269326942695269626972698269927002701270227032704270527062707270827092710271127122713271427152716271727182719272027212722272327242725272627272728272927302731273227332734273527362737273827392740274127422743274427452746274727482749275027512752275327542755275627572758275927602761276227632764276527662767276827692770277127722773277427752776277727782779278027812782278327842785278627872788278927902791279227932794279527962797279827992800280128022803280428052806280728082809281028112812281328142815281628172818281928202821282228232824282528262827282828292830283128322833283428352836283728382839284028412842284328442845284628472848284928502851285228532854285528562857285828592860286128622863286428652866286728682869287028712872287328742875287628772878287928802881288228832884288528862887288828892890289128922893289428952896289728982899290029012902290329042905290629072908290929102911291229132914291529162917291829192920292129222923292429252926292729282929293029312932293329342935293629372938293929402941294229432944294529462947294829492950295129522953295429552956295729582959296029612962296329642965296629672968296929702971297229732974297529762977297829792980298129822983298429852986298729882989299029912992299329942995299629972998299930003001300230033004300530063007300830093010301130123013301430153016301730183019302030213022302330243025302630273028302930303031303230333034303530363037303830393040304130423043304430453046304730483049305030513052305330543055305630573058305930603061306230633064306530663067306830693070307130723073307430753076307730783079308030813082308330843085308630873088308930903091309230933094309530963097309830993100310131023103310431053106310731083109311031113112311331143115311631173118311931203121312231233124312531263127312831293130313131323133313431353136313731383139314031413142314331443145314631473148314931503151315231533154315531563157315831593160316131623163316431653166316731683169317031713172317331743175317631773178317931803181318231833184318531863187318831893190319131923193319431953196319731983199320032013202320332043205320632073208320932103211321232133214321532163217321832193220322132223223322432253226322732283229323032313232323332343235323632373238323932403241324232433244324532463247324832493250325132523253325432553256325732583259326032613262326332643265326632673268326932703271327232733274327532763277327832793280328132823283328432853286328732883289329032913292329332943295329632973298329933003301330233033304330533063307330833093310331133123313331433153316331733183319332033213322332333243325332633273328332933303331333233333334333533363337333833393340334133423343334433453346334733483349335033513352335333543355335633573358335933603361336233633364336533663367336833693370337133723373337433753376337733783379338033813382338333843385338633873388338933903391339233933394339533963397339833993400340134023403340434053406340734083409341034113412341334143415341634173418341934203421342234233424342534263427342834293430343134323433343434353436343734383439344034413442344334443445344634473448344934503451345234533454345534563457345834593460346134623463346434653466346734683469347034713472347334743475347634773478347934803481348234833484348534863487348834893490349134923493349434953496349734983499350035013502350335043505350635073508350935103511351235133514351535163517351835193520352135223523352435253526352735283529353035313532353335343535353635373538353935403541354235433544354535463547354835493550355135523553355435553556355735583559356035613562356335643565356635673568356935703571357235733574357535763577357835793580358135823583358435853586358735883589359035913592359335943595359635973598359936003601360236033604360536063607360836093610361136123613361436153616361736183619362036213622362336243625362636273628362936303631363236333634363536363637363836393640364136423643364436453646364736483649365036513652365336543655365636573658365936603661366236633664366536663667366836693670367136723673367436753676367736783679368036813682368336843685368636873688368936903691369236933694369536963697369836993700370137023703370437053706370737083709371037113712371337143715371637173718371937203721372237233724372537263727372837293730373137323733373437353736373737383739374037413742374337443745374637473748374937503751375237533754375537563757375837593760376137623763376437653766376737683769377037713772377337743775377637773778377937803781378237833784378537863787378837893790379137923793379437953796379737983799380038013802380338043805380638073808380938103811381238133814381538163817381838193820382138223823382438253826382738283829383038313832383338343835383638373838383938403841384238433844384538463847384838493850385138523853385438553856385738583859386038613862386338643865386638673868386938703871387238733874387538763877387838793880388138823883388438853886388738883889389038913892389338943895389638973898389939003901390239033904390539063907390839093910391139123913391439153916391739183919392039213922392339243925392639273928392939303931393239333934393539363937393839393940394139423943394439453946394739483949395039513952395339543955395639573958395939603961396239633964396539663967396839693970397139723973397439753976397739783979398039813982398339843985398639873988398939903991399239933994399539963997399839994000400140024003400440054006400740084009401040114012401340144015401640174018401940204021402240234024402540264027402840294030403140324033403440354036
  1. import time
  2. import traceback
  3. import random
  4. from datetime import date, timedelta, datetime
  5. from log import Log
  6. from db_helper import RedisHelper
  7. from config import set_config
  8. from utils import FilterVideos, get_videos_remain_view_count, get_videos_local_distribute_count, send_msg_to_feishu
  9. import gevent
  10. import json
  11. import sys
  12. from parameter_update import param_update_expansion_factor
  13. from parameter_update import param_update_filter_flags
  14. from parameter_update import param_update_risk_rule
  15. from parameter_update import param_update_risk_videos
  16. from parameter_update import param_update_rule
  17. log_ = Log()
  18. config_ = set_config()
  19. class PoolRecall(object):
  20. """召回"""
  21. def __init__(self, request_id, app_type, client_info=None, mid='', uid='', ab_code='',
  22. rule_key='', data_key='', no_op_flag=False, params=None, rule_key_30day=None, shield_config=None,
  23. video_id=None, level_weight=None, h_data_key=None, h_rule_key=None):
  24. """
  25. 初始化
  26. :param request_id: request_id
  27. :param app_type: 产品标识 type-int
  28. :param client_info: 用户位置信息 {"cityCode": "100000"}
  29. :param mid: mid type-string
  30. :param uid: uid type-string
  31. :param ab_code: ab_code type-int
  32. :param params:
  33. """
  34. self.request_id = request_id
  35. self.app_type = app_type
  36. self.mid = mid
  37. self.uid = uid
  38. self.video_id = video_id
  39. self.ab_code = ab_code
  40. self.client_info = client_info
  41. self.rule_key = rule_key
  42. self.data_key = data_key
  43. self.no_op_flag = no_op_flag
  44. self.rule_key_30day = rule_key_30day
  45. self.shield_config = shield_config
  46. self.level_weight = level_weight
  47. self.redis_helper = RedisHelper(params=params)
  48. self.h_data_key = data_key
  49. self.h_rule_key = h_rule_key
  50. self.expansion_factor = param_update_expansion_factor()
  51. [self.risk_filter_flag, self.tags_filter_flag] = param_update_filter_flags()
  52. if self.risk_filter_flag:
  53. self.app_region_filtered = param_update_risk_rule()
  54. self.videos_with_risk = param_update_risk_videos()
  55. else:
  56. self.app_region_filtered = {}
  57. self.videos_with_risk = []
  58. if self.tags_filter_flag:
  59. self.tags_filter_rule = param_update_rule(self.redis_helper)
  60. else:
  61. self.tags_filter_rule = {}
  62. def copy_redis_zset_data(self, from_key_name, to_key_name):
  63. # 获取from_key_name中的数据
  64. records = self.redis_helper.get_data_zset_with_index(key_name=from_key_name, start=0, end=-1, with_scores=True)
  65. if records is not None:
  66. data = {}
  67. for video_id, score in records:
  68. data[int(video_id)] = score
  69. # 重新写入
  70. if self.redis_helper.key_exists(to_key_name):
  71. self.redis_helper.del_keys(key_name=to_key_name)
  72. self.redis_helper.add_data_with_zset(key_name=to_key_name, data=data, expire_time=1*3600)
  73. return True
  74. else:
  75. return False
  76. def update_mid_data(self, h_recall_mid_key, h_record_key, key_prefix):
  77. # 判断当前小时的小时级列表是否更新
  78. now_date = datetime.today()
  79. h = datetime.now().hour
  80. now_dt = datetime.strftime(now_date, '%Y%m%d')
  81. now_h_recall_key = f"{key_prefix}{self.app_type}.{self.data_key}.{self.rule_key}.{now_dt}.{h}"
  82. if self.redis_helper.key_exists(key_name=now_h_recall_key):
  83. flag = self.copy_redis_zset_data(from_key_name=now_h_recall_key, to_key_name=h_recall_mid_key)
  84. if flag:
  85. value = {'date': now_dt, 'h': h}
  86. self.redis_helper.set_data_to_redis(key_name=h_record_key, value=str(value), expire_time=1*3600)
  87. else:
  88. if h == 0:
  89. redis_dt = datetime.strftime(now_date - timedelta(days=1), '%Y%m%d')
  90. redis_h = 23
  91. else:
  92. redis_dt = now_dt
  93. redis_h = h - 1
  94. now_h_recall_key = f"{key_prefix}{self.app_type}.{self.data_key}.{self.rule_key}.{redis_dt}.{redis_h}"
  95. flag = self.copy_redis_zset_data(from_key_name=now_h_recall_key, to_key_name=h_recall_mid_key)
  96. if flag:
  97. value = {'date': redis_dt, 'h': redis_h}
  98. self.redis_helper.set_data_to_redis(key_name=h_record_key, value=str(value), expire_time=1*3600)
  99. def get_mid_h_key(self, province_code, key_flag=''):
  100. if key_flag == 'region_24h':
  101. # mid对应小时级视频列表 redis-key 地域分组相对24h
  102. h_recall_mid_key = f"{config_.H_WITH_MID_RECALL_KEY_NAME_PREFIX_REGION_24H}{self.app_type}.{self.mid}"
  103. # 判断mid对应小时级视频列表 时间记录
  104. h_record_key = f"{config_.H_WITH_MID_RECORD_KEY_NAME_PREFIX_REGION_24H}{self.app_type}.{self.mid}"
  105. elif key_flag in ['24h', 'day_24h']:
  106. # mid对应小时级视频列表 redis-key 相对24h
  107. h_recall_mid_key = f"{config_.H_WITH_MID_RECALL_KEY_NAME_PREFIX_24H}{self.app_type}.{self.mid}"
  108. # 判断mid对应小时级视频列表 时间记录
  109. h_record_key = f"{config_.H_WITH_MID_RECORD_KEY_NAME_PREFIX_24H}{self.app_type}.{self.mid}"
  110. else:
  111. # mid对应小时级视频列表 redis-key
  112. h_recall_mid_key = f"{config_.H_WITH_MID_RECALL_KEY_NAME_PREFIX}{self.app_type}.{self.mid}"
  113. # 判断mid对应小时级视频列表 时间记录
  114. h_record_key = f"{config_.H_WITH_MID_RECORD_KEY_NAME_PREFIX}{self.app_type}.{self.mid}"
  115. # 列表存储 redis-key prefix
  116. if self.ab_code in [code for _, code in config_.AB_CODE['region_rank_by_h'].items()]:
  117. if key_flag == 'region_24h':
  118. if self.ab_code == config_.AB_CODE['region_rank_by_h'].get('region_rule_rank2'):
  119. key_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_24H}{province_code}."
  120. else:
  121. key_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H}{province_code}."
  122. elif key_flag == 'day_24h':
  123. key_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_24H_H}{province_code}."
  124. else:
  125. key_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H}{province_code}."
  126. elif self.ab_code in [code for _, code in config_.AB_CODE['rank_by_24h'].items()]:
  127. key_prefix = config_.RECALL_KEY_NAME_PREFIX_BY_24H
  128. elif key_flag == '24h':
  129. key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP_24H_H
  130. else:
  131. key_prefix = config_.RECALL_KEY_NAME_PREFIX_BY_H
  132. if not self.redis_helper.key_exists(key_name=h_record_key):
  133. # ###### 记录key不存在,copy列表,更新记录
  134. self.update_mid_data(h_recall_mid_key=h_recall_mid_key, h_record_key=h_record_key, key_prefix=key_prefix)
  135. # return h_recall_mid_key
  136. else:
  137. # ###### 记录key存在,判断date, h
  138. now_date = datetime.today()
  139. h = datetime.now().hour
  140. # 获取记录的date, h
  141. record = self.redis_helper.get_data_from_redis(key_name=h_record_key)
  142. record_dt = eval(record).get('date')
  143. record_h = eval(record).get('h')
  144. now_dt = datetime.strftime(now_date, '%Y%m%d')
  145. if record_dt == now_dt and int(record_h) == h:
  146. # 已获取当前小时数据
  147. pass
  148. # return h_recall_mid_key
  149. elif (record_dt == now_dt and h-int(record_h) == 1) or (h == 0 and int(record_h) == 23):
  150. # 记录的h - 当前h = 1,判断当前h数据是否已更新
  151. now_h_recall_key = f"{key_prefix}{self.app_type}.{self.data_key}.{self.rule_key}.{now_dt}.{h}"
  152. # if not self.redis_helper.key_exists(key_name=now_h_recall_key):
  153. # 未更新
  154. # return h_recall_mid_key
  155. if self.redis_helper.key_exists(key_name=now_h_recall_key):
  156. # 已更新,重新获取更新mid对应列表及记录
  157. # self.redis_helper.del_keys(key_name=h_recall_mid_key)
  158. # self.redis_helper.del_keys(key_name=h_record_key)
  159. flag = self.copy_redis_zset_data(from_key_name=now_h_recall_key, to_key_name=h_recall_mid_key)
  160. if flag:
  161. new_record = {'date': now_dt, 'h': h}
  162. self.redis_helper.set_data_to_redis(key_name=h_record_key, value=str(new_record), expire_time=2*3600)
  163. # return h_recall_mid_key
  164. else:
  165. self.update_mid_data(h_recall_mid_key=h_recall_mid_key, h_record_key=h_record_key, key_prefix=key_prefix)
  166. # return h_recall_mid_key
  167. return h_recall_mid_key
  168. def rov_pool_recall_by_h(self, size=10, expire_time=24*3600):
  169. """
  170. 从小时级更新ROV召回池中获取视频
  171. :param size: 获取视频个数
  172. :param expire_time: 末位视频记录redis过期时间
  173. :return:
  174. """
  175. start_time = time.time()
  176. # 获取provinceCode
  177. province_code = self.client_info.get('provinceCode', '-1')
  178. if province_code == '':
  179. province_code = '-1'
  180. if self.ab_code in [code for _, code in config_.AB_CODE['rank_by_24h'].items()]:
  181. push_from = config_.PUSH_FROM['rov_recall_24h']
  182. else:
  183. push_from = config_.PUSH_FROM['rov_recall_h']
  184. # 获取mid对应的小时级列表redis-key
  185. h_recall_mid_key = self.get_mid_h_key(province_code=province_code)
  186. if not self.redis_helper.key_exists(h_recall_mid_key):
  187. recall_result = self.rov_pool_recall(size=size, expire_time=expire_time)
  188. else:
  189. # 过滤的视频
  190. fil_video_ids = []
  191. recall_result = []
  192. # 每次获取的视频数
  193. get_size = size * 5
  194. # 记录获取频次
  195. freq = 0
  196. while len(recall_result) < size:
  197. freq += 1
  198. if freq > config_.MAX_FREQ_FROM_ROV_POOL:
  199. break
  200. # 获取数据
  201. data = self.redis_helper.get_data_zset_with_index(key_name=h_recall_mid_key,
  202. start=(freq - 1) * get_size, end=freq * get_size - 1,
  203. with_scores=True)
  204. if not data:
  205. # log_.info('小时级更新视频已取完')
  206. break
  207. # 获取视频id,并转换类型为int,并存储为key-value{videoId: score}
  208. video_ids = []
  209. video_score = {}
  210. for value in data:
  211. video_id = int(value[0])
  212. video_ids.append(video_id)
  213. video_score[video_id] = value[1]
  214. # 过滤
  215. filter_ = FilterVideos(request_id=self.request_id,
  216. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
  217. ge = gevent.spawn(filter_.filter_videos_h, self.rule_key, self.ab_code, province_code)
  218. ge.join()
  219. filtered_result = ge.get()
  220. if filtered_result:
  221. # 添加视频源参数 pushFrom, abCode
  222. temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)],
  223. 'pushFrom': push_from, 'abCode': self.ab_code}
  224. for item in filtered_result if video_score.get(int(item)) is not None]
  225. recall_result.extend(temp_result)
  226. fil_video_ids.extend(list(set(video_ids) - set([item.get('videoId') for item in temp_result])))
  227. else:
  228. fil_video_ids.extend(video_ids)
  229. # 将被过滤的视频进行移除
  230. for value in fil_video_ids:
  231. self.redis_helper.remove_value_from_zset(key_name=h_recall_mid_key, value=value)
  232. # 判断获取到的小时级数据数量
  233. if len(recall_result) < size:
  234. # 补充数据
  235. rov_recall_result = self.rov_pool_recall(size=size, expire_time=expire_time)
  236. # 去重合并
  237. now_video_ids = [item.get('videoId') for item in recall_result]
  238. for video in rov_recall_result:
  239. vid = video.get('videoId')
  240. if vid not in now_video_ids:
  241. recall_result.append(video)
  242. now_video_ids.append(vid)
  243. if len(recall_result) >= size:
  244. break
  245. else:
  246. continue
  247. log_.info({
  248. 'logTimestamp': int(time.time() * 1000),
  249. 'request_id': self.request_id,
  250. 'operation': 'rov_pool_recall_by_h',
  251. 'executeTime': (time.time() - start_time) * 1000
  252. })
  253. return recall_result[:size]
  254. def rov_pool_recall_by_day(self, size=4, expire_time=24*3600):
  255. """
  256. 从天级规则更新列表中获取视频
  257. :param size: 获取视频个数
  258. :param expire_time: 末位视频记录redis过期时间
  259. :return:
  260. """
  261. start_time = time.time()
  262. # 获取天级规则更新列表相关redis key, 用户上一次在天级规则更新列表对应的位置
  263. rule_key_name, last_rule_day_recall_key, idx = self.get_video_last_idx_day()
  264. # 获取天级规则更新列表数据
  265. if not rule_key_name:
  266. # log_.info('天级规则更新列表中无视频')
  267. recall_result = self.rov_pool_recall(size=size, expire_time=expire_time)
  268. else:
  269. recall_result = []
  270. # 每次获取的视频数
  271. get_size = size * 5
  272. # 记录获取频次
  273. freq = 0
  274. while len(recall_result) < size:
  275. freq += 1
  276. if freq > config_.MAX_FREQ_FROM_ROV_POOL:
  277. break
  278. # 获取数据
  279. data = self.redis_helper.get_data_zset_with_index(key_name=rule_key_name,
  280. start=idx, end=idx + get_size - 1,
  281. with_scores=True)
  282. if not data:
  283. # log_.info('天级规则更新视频已取完')
  284. break
  285. # 获取视频id,并转换类型为int,并存储为key-value{videoId: score}
  286. video_ids = []
  287. video_score = {}
  288. for value in data:
  289. video_id = int(value[0])
  290. video_ids.append(video_id)
  291. video_score[video_id] = value[1]
  292. # 过滤
  293. filter_ = FilterVideos(request_id=self.request_id,
  294. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
  295. ge = gevent.spawn(filter_.filter_videos)
  296. ge.join()
  297. filtered_result = ge.get()
  298. if filtered_result:
  299. # 添加视频源参数 pushFrom, abCode
  300. temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)],
  301. 'pushFrom': config_.PUSH_FROM['rov_recall_day'], 'abCode': self.ab_code}
  302. for item in filtered_result if video_score.get(int(item)) is not None]
  303. recall_result.extend(temp_result)
  304. else:
  305. # 将此次获取的末位视频id同步刷新到Redis中,方便下次快速定位到召回位置,过期时间为1天
  306. self.redis_helper.set_data_to_redis(key_name=last_rule_day_recall_key, value=data[-1][0],
  307. expire_time=expire_time)
  308. idx += get_size
  309. # 判断获取到的天级规则数据数量
  310. if len(recall_result) < size:
  311. # 补充数据
  312. rov_recall_result = self.rov_pool_recall(size=size, expire_time=expire_time)
  313. # 去重合并
  314. now_video_ids = [item.get('videoId') for item in recall_result]
  315. for video in rov_recall_result:
  316. vid = video.get('videoId')
  317. if vid not in now_video_ids:
  318. recall_result.append(video)
  319. now_video_ids.append(vid)
  320. if len(recall_result) >= size:
  321. break
  322. else:
  323. continue
  324. log_.info({
  325. 'logTimestamp': int(time.time() * 1000),
  326. 'request_id': self.request_id,
  327. 'operation': 'rov_pool_recall_by_day',
  328. 'executeTime': (time.time() - start_time) * 1000
  329. })
  330. return recall_result[:size]
  331. def rov_pool_recall(self, size=10, expire_time=24*3600, video_type='', push_from=config_.PUSH_FROM['rov_recall']):
  332. """
  333. 从ROV召回池中获取视频
  334. :param size: 获取视频个数
  335. :param expire_time: 末位视频记录redis过期时间
  336. :param video_type: 视频列表类别
  337. :param push_from: 视频来源标记
  338. :return:
  339. """
  340. start_time = time.time()
  341. # 获取生效中的置顶视频
  342. if self.no_op_flag:
  343. top_video_ids, top_video_result = [], []
  344. elif self.client_info is None:
  345. # 无用户位置信息时,不获取置顶视频
  346. top_video_ids, top_video_result = [], []
  347. else:
  348. top_video_ids, top_video_result = self.get_top_videos()
  349. # log_.info('===top video result = {}'.format(top_video_ids))
  350. # 获取修改过rov的视频
  351. if self.no_op_flag:
  352. update_rov_video_ids, update_rov_result = [], []
  353. else:
  354. update_rov_video_ids, update_rov_result = self.get_update_rov_videos()
  355. # log_.info('update rov result = {}'.format(update_rov_video_ids))
  356. # 与置顶视频去重
  357. update_rov_video_ids_dup, update_rov_dup_result = [], []
  358. for item in update_rov_result:
  359. if item['videoId'] not in top_video_ids:
  360. update_rov_video_ids_dup.append(item['videoId'])
  361. update_rov_dup_result.append(item)
  362. # 获取相关redis key, 用户上一次在rov召回池对应的位置
  363. rov_pool_key, last_rov_recall_key, idx = self.get_video_last_idx(video_type=video_type)
  364. if not rov_pool_key:
  365. # log_.info('ROV召回池中无视频')
  366. if (not update_rov_dup_result) and (not top_video_result):
  367. return []
  368. rov_pool_recall_result = top_video_result.extend(update_rov_dup_result)
  369. rov_pool_recall_result.sort(key=lambda x: x.get('rovScore', 0), reverse=True)
  370. return rov_pool_recall_result[:size]
  371. rov_pool_recall_result = []
  372. # 每次获取的视频数
  373. get_size = size * 5
  374. # 记录获取频次
  375. freq = 0
  376. while len(rov_pool_recall_result) < size:
  377. freq += 1
  378. if freq > config_.MAX_FREQ_FROM_ROV_POOL:
  379. break
  380. # 获取数据
  381. st_get = time.time()
  382. data = self.redis_helper.get_data_zset_with_index(key_name=rov_pool_key,
  383. start=idx, end=idx + get_size - 1,
  384. with_scores=True)
  385. et_get = time.time()
  386. # log_.info('get data from rov pool redis: freq = {}, data = {}, execute time = {}ms'.format(
  387. # freq, data, (et_get - st_get) * 1000))
  388. if not data:
  389. # log_.info('ROV召回池中的视频已取完')
  390. break
  391. # 获取视频id,并转换类型为int,并存储为key-value{videoId: score}
  392. video_ids = []
  393. video_score = {}
  394. for value in data:
  395. video_id = int(value[0])
  396. # 视频id在 生效中的置顶视频 或 修改过rov的视频 中,跳过
  397. if video_id in update_rov_video_ids_dup or video_id in top_video_ids:
  398. continue
  399. video_ids.append(video_id)
  400. video_score[video_id] = value[1]
  401. # 过滤
  402. filter_ = FilterVideos(request_id=self.request_id,
  403. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
  404. ge = gevent.spawn(filter_.filter_videos)
  405. ge.join()
  406. filtered_result = ge.get()
  407. # filtered_result = filter_.filter_videos()
  408. if filtered_result:
  409. # 添加视频源参数 pushFrom, abCode
  410. temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)],
  411. 'pushFrom': push_from, 'abCode': self.ab_code}
  412. for item in filtered_result if video_score.get(int(item)) is not None]
  413. rov_pool_recall_result.extend(temp_result)
  414. else:
  415. # 将此次获取的末位视频id同步刷新到Redis中,方便下次快速定位到召回位置,过期时间为1天
  416. if self.mid:
  417. # mid为空时,不做记录
  418. self.redis_helper.set_data_to_redis(key_name=last_rov_recall_key, value=data[-1][0],
  419. expire_time=expire_time)
  420. idx += get_size
  421. # 生效中的置顶视频、被修改rov视频、rov召回池视频 归并排序
  422. if top_video_result:
  423. rov_pool_recall_result.extend(top_video_result)
  424. if update_rov_dup_result:
  425. rov_pool_recall_result.extend(update_rov_dup_result)
  426. rov_pool_recall_result.sort(key=lambda x: x.get('rovScore', 0), reverse=True)
  427. # log_.info({
  428. # 'logTimestamp': int(time.time() * 1000),
  429. # 'request_id': self.request_id,
  430. # 'operation': 'rov_pool_recall',
  431. # 'executeTime': (time.time() - start_time) * 1000
  432. # })
  433. return rov_pool_recall_result[:size]
  434. def flow_pool_recall(self, size=10, flow_pool_id=None, flow_pool_abtest_group=None):
  435. """从流量池中获取视频"""
  436. # add_flow_pool_recall_log
  437. flow_pool_recall_process = {}
  438. start_time = time.time()
  439. # 获取存在城市分组数据的城市编码列表
  440. city_code_list = [code for _, code in config_.CITY_CODE.items()]
  441. # 获取provinceCode
  442. province_code = self.client_info.get('provinceCode', '-1')
  443. # 获取cityCode
  444. city_code = self.client_info.get('cityCode', '-1')
  445. if city_code in city_code_list:
  446. # 分城市数据存在时,获取城市分组数据
  447. region_code = city_code
  448. else:
  449. region_code = province_code
  450. if region_code == '':
  451. region_code = '-1'
  452. flow_pool_key = self.get_pool_redis_key('flow', flow_pool_id=flow_pool_id)
  453. # add_flow_pool_recall_log
  454. flow_pool_recall_process['flow_pool_key'] = flow_pool_key
  455. # print(flow_pool_key)
  456. flow_pool_recall_result = []
  457. flow_pool_recall_videos = []
  458. # 每次获取的视频数
  459. get_size = size * 5
  460. # 记录获取频次
  461. freq = 0
  462. idx = 0
  463. while len(flow_pool_recall_result) < size:
  464. freq += 1
  465. if freq > config_.MAX_FREQ_FROM_FLOW_POOL:
  466. break
  467. # 获取数据
  468. # st_get = time.time()
  469. data = self.redis_helper.get_data_zset_with_index(key_name=flow_pool_key,
  470. start=idx, end=idx + get_size - 1,
  471. with_scores=True)
  472. # add_flow_pool_recall_log
  473. flow_pool_recall_process['initial_data'] = data
  474. # et_get = time.time()
  475. # log_.info('get data from flow pool redis: freq = {}, data = {}, execute time = {}ms'.format(
  476. # freq, data, (et_get - st_get) * 1000))
  477. if not data:
  478. # log_.info('流量池中的视频已取完')
  479. break
  480. # 将video_id 与 flow_pool, score做mapping整理
  481. video_ids = []
  482. video_mapping = {}
  483. video_score = {}
  484. for value in data:
  485. try:
  486. video_id, flow_pool = value[0].split('-')
  487. except Exception as e:
  488. log_.error({
  489. 'request_id': self.request_id,
  490. 'app_type': self.app_type,
  491. 'flow_pool_value': value
  492. })
  493. continue
  494. video_id = int(video_id)
  495. if video_id not in video_ids:
  496. video_ids.append(video_id)
  497. video_score[video_id] = value[1]
  498. if video_id not in video_mapping:
  499. video_mapping[video_id] = [flow_pool]
  500. else:
  501. video_mapping[video_id].append(flow_pool)
  502. # todo zhangbo 获取tags
  503. video_tag_dict = self.get_video_tags(video_ids)
  504. # 过滤
  505. filter_ = FilterVideos(request_id=self.request_id,
  506. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids,
  507. expansion_factor=self.expansion_factor,
  508. risk_filter_flag=self.risk_filter_flag,
  509. app_region_filtered=self.app_region_filtered,
  510. videos_with_risk=self.videos_with_risk
  511. )
  512. ge = gevent.spawn(filter_.filter_videos, pool_type='flow',
  513. region_code=region_code, shield_config=self.shield_config)
  514. ge.join()
  515. filtered_result = ge.get()
  516. # add_flow_pool_recall_log
  517. flow_pool_recall_process['filtered_data'] = filtered_result
  518. # 检查可分发数
  519. if filtered_result:
  520. st_check = time.time()
  521. ge = gevent.spawn(self.check_video_counts, video_ids=filtered_result, flow_pool_mapping=video_mapping)
  522. ge.join()
  523. check_result = ge.get()
  524. # add_flow_pool_recall_log
  525. flow_pool_recall_process['check_counts_data'] = check_result
  526. # log_.info({
  527. # 'logTimestamp': int(time.time() * 1000),
  528. # 'request_id': self.request_id,
  529. # 'app_type': self.app_type,
  530. # 'mid': self.mid,
  531. # 'uid': self.uid,
  532. # 'operation': 'check_video_counts',
  533. # 'executeTime': (time.time() - st_check) * 1000
  534. # })
  535. for item in check_result:
  536. video_id = int(item[0])
  537. flow_pool = item[1]
  538. if video_id not in flow_pool_recall_videos:
  539. # 取其中一个 flow_pool 作为召回结果
  540. # 添加视频源参数 pushFrom, abCode
  541. flow_pool_recall_result.append(
  542. {'videoId': video_id, 'flowPool': flow_pool,
  543. 'rovScore': video_score[video_id], 'pushFrom': config_.PUSH_FROM['flow_recall'],
  544. 'abCode': self.ab_code, 'flow_pool_abtest_group': flow_pool_abtest_group,
  545. "tags": video_tag_dict[video_id]}
  546. )
  547. flow_pool_recall_videos.append(video_id)
  548. # et_check = time.time()
  549. # log_.info('check result: result = {}, execute time = {}ms'.format(
  550. # check_result, (et_check - st_check) * 1000))
  551. # # 判断错误标记, True为错误
  552. # if error_flag:
  553. # # 结束流量池召回
  554. # break
  555. idx += get_size
  556. # log_.info({
  557. # 'logTimestamp': int(time.time() * 1000),
  558. # 'request_id': self.request_id,
  559. # 'operation': 'flow_pool_recall',
  560. # 'executeTime': (time.time() - start_time) * 1000
  561. # })
  562. return flow_pool_recall_result[:size], flow_pool_recall_process
  563. def flow_pool_recall_new(self, size=10, flow_pool_id=None):
  564. """从流量池中获取视频"""
  565. start_time = time.time()
  566. # 获取存在城市分组数据的城市编码列表
  567. city_code_list = [code for _, code in config_.CITY_CODE.items()]
  568. # 获取provinceCode
  569. province_code = self.client_info.get('provinceCode', '-1')
  570. # 获取cityCode
  571. city_code = self.client_info.get('cityCode', '-1')
  572. if city_code in city_code_list:
  573. # 分城市数据存在时,获取城市分组数据
  574. region_code = city_code
  575. else:
  576. region_code = province_code
  577. if region_code == '':
  578. region_code = '-1'
  579. flow_pool_key = self.get_pool_redis_key('flow_set', flow_pool_id=flow_pool_id)
  580. # print(flow_pool_key)
  581. flow_pool_recall_result = []
  582. flow_pool_recall_videos = []
  583. # 每次获取的视频数
  584. get_size = size * 5
  585. # 记录获取频次
  586. freq = 0
  587. idx = 0
  588. while len(flow_pool_recall_result) < size:
  589. freq += 1
  590. if freq > config_.MAX_FREQ_FROM_FLOW_POOL:
  591. break
  592. # 获取数据
  593. # st_get = time.time()
  594. data = self.redis_helper.get_data_with_count_from_set(key_name=flow_pool_key, count=get_size)
  595. # et_get = time.time()
  596. # log_.info('get data from flow pool redis: freq = {}, data = {}, execute time = {}ms'.format(
  597. # freq, data, (et_get - st_get) * 1000))
  598. if not data:
  599. # log_.info('流量池中的视频已取完')
  600. break
  601. # 将video_id 与 flow_pool, score做mapping整理
  602. video_ids = []
  603. video_mapping = {}
  604. video_score = {}
  605. for value in data:
  606. try:
  607. video_id, flow_pool = value.split('-')
  608. except Exception as e:
  609. log_.error({
  610. 'request_id': self.request_id,
  611. 'app_type': self.app_type,
  612. 'flow_pool_value': value
  613. })
  614. continue
  615. video_id = int(video_id)
  616. if video_id not in video_ids:
  617. video_ids.append(video_id)
  618. if video_id not in video_mapping:
  619. video_mapping[video_id] = [flow_pool]
  620. else:
  621. video_mapping[video_id].append(flow_pool)
  622. # 过滤
  623. filter_ = FilterVideos(request_id=self.request_id,
  624. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
  625. ge = gevent.spawn(filter_.filter_videos, pool_type='flow',
  626. region_code=region_code, shield_config=self.shield_config)
  627. ge.join()
  628. filtered_result = ge.get()
  629. # 检查可分发数
  630. if filtered_result:
  631. st_check = time.time()
  632. ge = gevent.spawn(self.check_video_counts_new, video_ids=filtered_result, flow_pool_mapping=video_mapping)
  633. ge.join()
  634. check_result = ge.get()
  635. # log_.info({
  636. # 'logTimestamp': int(time.time() * 1000),
  637. # 'request_id': self.request_id,
  638. # 'app_type': self.app_type,
  639. # 'mid': self.mid,
  640. # 'uid': self.uid,
  641. # 'operation': 'check_video_counts',
  642. # 'executeTime': (time.time() - st_check) * 1000
  643. # })
  644. for item in check_result:
  645. video_id = int(item[0])
  646. flow_pool = item[1]
  647. if video_id not in flow_pool_recall_videos:
  648. # 取其中一个 flow_pool 作为召回结果
  649. # 添加视频源参数 pushFrom, abCode
  650. flow_pool_recall_result.append(
  651. {'videoId': video_id, 'flowPool': flow_pool,
  652. 'rovScore': random.uniform(0, 100), 'pushFrom': config_.PUSH_FROM['flow_recall'],
  653. 'abCode': self.ab_code}
  654. )
  655. flow_pool_recall_videos.append(video_id)
  656. # et_check = time.time()
  657. # log_.info('check result: result = {}, execute time = {}ms'.format(
  658. # check_result, (et_check - st_check) * 1000))
  659. # # 判断错误标记, True为错误
  660. # if error_flag:
  661. # # 结束流量池召回
  662. # break
  663. idx += get_size
  664. # log_.info({
  665. # 'logTimestamp': int(time.time() * 1000),
  666. # 'request_id': self.request_id,
  667. # 'operation': 'flow_pool_recall',
  668. # 'executeTime': (time.time() - start_time) * 1000
  669. # })
  670. return flow_pool_recall_result[:size]
  671. def flow_pool_recall_new_with_level(self, size=10, flow_pool_id=None, flow_pool_abtest_group=None):
  672. """从流量池中获取视频"""
  673. # add_flow_pool_recall_log
  674. flow_pool_recall_process = {}
  675. start_time = time.time()
  676. # 获取存在城市分组数据的城市编码列表
  677. city_code_list = [code for _, code in config_.CITY_CODE.items()]
  678. # 获取provinceCode
  679. province_code = self.client_info.get('provinceCode', '-1')
  680. # 获取cityCode
  681. city_code = self.client_info.get('cityCode', '-1')
  682. if city_code in city_code_list:
  683. # 分城市数据存在时,获取城市分组数据
  684. region_code = city_code
  685. else:
  686. region_code = province_code
  687. if region_code == '':
  688. region_code = '-1'
  689. flow_pool_key, level = self.get_pool_redis_key('flow_set_level', flow_pool_id=flow_pool_id)
  690. # add_flow_pool_recall_log
  691. flow_pool_recall_process['flow_pool_key'] = flow_pool_key
  692. flow_pool_recall_process['level'] = level
  693. if flow_pool_key is None:
  694. return [], flow_pool_recall_process
  695. # print(flow_pool_key)
  696. flow_pool_recall_result = []
  697. flow_pool_recall_videos = []
  698. # 每次获取的视频数
  699. get_size = size * 5
  700. # 记录获取频次
  701. freq = 0
  702. idx = 0
  703. while len(flow_pool_recall_result) < size:
  704. freq += 1
  705. if freq > config_.MAX_FREQ_FROM_FLOW_POOL:
  706. break
  707. # 获取数据
  708. # st_get = time.time()
  709. data = self.redis_helper.get_data_with_count_from_set(key_name=flow_pool_key, count=get_size)
  710. # add_flow_pool_recall_log
  711. flow_pool_recall_process['initial_data'] = data
  712. # et_get = time.time()
  713. # log_.info('get data from flow pool redis: freq = {}, data = {}, execute time = {}ms'.format(
  714. # freq, data, (et_get - st_get) * 1000))
  715. if not data:
  716. # log_.info('流量池中的视频已取完')
  717. break
  718. # 将video_id 与 flow_pool, score做mapping整理
  719. video_ids = []
  720. video_mapping = {}
  721. video_score = {}
  722. for value in data:
  723. try:
  724. video_id, flow_pool = value.split('-')
  725. except Exception as e:
  726. log_.error({
  727. 'request_id': self.request_id,
  728. 'app_type': self.app_type,
  729. 'flow_pool_value': value
  730. })
  731. continue
  732. video_id = int(video_id)
  733. if video_id not in video_ids:
  734. video_ids.append(video_id)
  735. if video_id not in video_mapping:
  736. video_mapping[video_id] = [flow_pool]
  737. else:
  738. video_mapping[video_id].append(flow_pool)
  739. # todo zhangbo 获取tags
  740. video_tag_dict = self.get_video_tags(video_ids)
  741. # 过滤
  742. filter_ = FilterVideos(request_id=self.request_id,
  743. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids,
  744. expansion_factor=self.expansion_factor,
  745. risk_filter_flag=self.risk_filter_flag,
  746. app_region_filtered=self.app_region_filtered,
  747. videos_with_risk=self.videos_with_risk
  748. )
  749. ge = gevent.spawn(filter_.filter_videos, pool_type='flow',
  750. region_code=region_code, shield_config=self.shield_config)
  751. ge.join()
  752. filtered_result = ge.get()
  753. # add_flow_pool_recall_log
  754. flow_pool_recall_process['filtered_data'] = filtered_result
  755. # 检查可分发数
  756. if filtered_result:
  757. # st_check = time.time()
  758. ge = gevent.spawn(self.check_video_counts_new_with_level, video_ids=filtered_result, flow_pool_mapping=video_mapping)
  759. ge.join()
  760. check_result = ge.get()
  761. # add_flow_pool_recall_log
  762. flow_pool_recall_process['check_counts_data'] = check_result
  763. # log_.info({
  764. # 'logTimestamp': int(time.time() * 1000),
  765. # 'request_id': self.request_id,
  766. # 'app_type': self.app_type,
  767. # 'mid': self.mid,
  768. # 'uid': self.uid,
  769. # 'operation': 'check_video_counts',
  770. # 'executeTime': (time.time() - st_check) * 1000
  771. # })
  772. for item in check_result:
  773. video_id = int(item[0])
  774. flow_pool = item[1]
  775. if video_id not in flow_pool_recall_videos:
  776. # 取其中一个 flow_pool 作为召回结果
  777. # 添加视频源参数 pushFrom, abCode
  778. flow_pool_recall_result.append(
  779. {'videoId': video_id, 'flowPool': flow_pool, 'level': level,
  780. 'rovScore': random.uniform(0, 100), 'pushFrom': config_.PUSH_FROM['flow_recall'],
  781. 'abCode': self.ab_code, 'flow_pool_abtest_group': flow_pool_abtest_group,
  782. "tags": video_tag_dict[video_id]}
  783. )
  784. flow_pool_recall_videos.append(video_id)
  785. # et_check = time.time()
  786. # log_.info('check result: result = {}, execute time = {}ms'.format(
  787. # check_result, (et_check - st_check) * 1000))
  788. # # 判断错误标记, True为错误
  789. # if error_flag:
  790. # # 结束流量池召回
  791. # break
  792. idx += get_size
  793. # log_.info({
  794. # 'logTimestamp': int(time.time() * 1000),
  795. # 'request_id': self.request_id,
  796. # 'operation': 'flow_pool_recall',
  797. # 'executeTime': (time.time() - start_time) * 1000
  798. # })
  799. return flow_pool_recall_result[:size], flow_pool_recall_process
  800. def flow_pool_recall_new_with_level_score(self, size=10, flow_pool_id=None, flow_pool_abtest_group=None):
  801. """从流量池中获取视频"""
  802. # add_flow_pool_recall_log
  803. flow_pool_recall_process = {}
  804. # 获取存在城市分组数据的城市编码列表
  805. city_code_list = [code for _, code in config_.CITY_CODE.items()]
  806. # 获取provinceCode
  807. province_code = self.client_info.get('provinceCode', '-1')
  808. # 获取cityCode
  809. city_code = self.client_info.get('cityCode', '-1')
  810. if city_code in city_code_list:
  811. # 分城市数据存在时,获取城市分组数据
  812. region_code = city_code
  813. else:
  814. region_code = province_code
  815. if region_code == '':
  816. region_code = '-1'
  817. flow_pool_key, level = self.get_pool_redis_key('flow_set_level_score', flow_pool_id=flow_pool_id)
  818. # add_flow_pool_recall_log
  819. flow_pool_recall_process['flow_pool_key'] = flow_pool_key
  820. flow_pool_recall_process['level'] = level
  821. if flow_pool_key is None:
  822. return [], flow_pool_recall_process
  823. # print(flow_pool_key)
  824. flow_pool_recall_result = []
  825. flow_pool_recall_videos = []
  826. # 每次获取的视频数
  827. get_size = size * 5
  828. # 记录获取频次
  829. freq = 0
  830. idx = 0
  831. while len(flow_pool_recall_result) < size:
  832. freq += 1
  833. if freq > config_.MAX_FREQ_FROM_FLOW_POOL:
  834. break
  835. # 获取数据
  836. # st_get = time.time()
  837. data = self.redis_helper.get_data_zset_with_index(key_name=flow_pool_key,
  838. start=idx, end=idx + get_size - 1,
  839. with_scores=True)
  840. # add_flow_pool_recall_log
  841. # print(data)
  842. flow_pool_recall_process['initial_data'] = data
  843. # et_get = time.time()
  844. # log_.info('get data from flow pool redis: freq = {}, data = {}, execute time = {}ms'.format(
  845. # freq, data, (et_get - st_get) * 1000))
  846. if not data:
  847. # log_.info('流量池中的视频已取完')
  848. break
  849. # 将video_id 与 flow_pool, score做mapping整理
  850. video_ids = []
  851. video_mapping = {}
  852. video_score = {}
  853. for value in data:
  854. try:
  855. video_id, flow_pool = value[0].split('-')
  856. except Exception as e:
  857. log_.error({
  858. 'request_id': self.request_id,
  859. 'app_type': self.app_type,
  860. 'flow_pool_value': value
  861. })
  862. continue
  863. video_id = int(video_id)
  864. video_score[value[0]] = value[1]
  865. if video_id not in video_ids:
  866. video_ids.append(video_id)
  867. if video_id not in video_mapping:
  868. video_mapping[video_id] = [flow_pool]
  869. else:
  870. video_mapping[video_id].append(flow_pool)
  871. # 过滤
  872. filter_ = FilterVideos(request_id=self.request_id,
  873. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
  874. ge = gevent.spawn(filter_.filter_videos, pool_type='flow',
  875. region_code=region_code, shield_config=self.shield_config)
  876. ge.join()
  877. filtered_result = ge.get()
  878. # add_flow_pool_recall_log
  879. flow_pool_recall_process['filtered_data'] = filtered_result
  880. # 检查可分发数
  881. if filtered_result:
  882. # st_check = time.time()
  883. ge = gevent.spawn(self.check_video_counts_new_with_level_score,
  884. video_ids=filtered_result, flow_pool_mapping=video_mapping)
  885. ge.join()
  886. check_result = ge.get()
  887. # add_flow_pool_recall_log
  888. flow_pool_recall_process['check_counts_data'] = check_result
  889. # log_.info({
  890. # 'logTimestamp': int(time.time() * 1000),
  891. # 'request_id': self.request_id,
  892. # 'app_type': self.app_type,
  893. # 'mid': self.mid,
  894. # 'uid': self.uid,
  895. # 'operation': 'check_video_counts',
  896. # 'executeTime': (time.time() - st_check) * 1000
  897. # })
  898. for item in check_result:
  899. video_id = int(item[0])
  900. flow_pool = item[1]
  901. if video_id not in flow_pool_recall_videos:
  902. # 取其中一个 flow_pool 作为召回结果
  903. # 添加视频源参数 pushFrom, abCode
  904. flow_pool_recall_result.append(
  905. {'videoId': video_id, 'flowPool': flow_pool, 'level': level,
  906. 'rovScore': video_score[f"{video_id}-{flow_pool}"], 'pushFrom': config_.PUSH_FROM['flow_recall'],
  907. 'abCode': self.ab_code, 'flow_pool_abtest_group': flow_pool_abtest_group}
  908. )
  909. flow_pool_recall_videos.append(video_id)
  910. # et_check = time.time()
  911. # log_.info('check result: result = {}, execute time = {}ms'.format(
  912. # check_result, (et_check - st_check) * 1000))
  913. # # 判断错误标记, True为错误
  914. # if error_flag:
  915. # # 结束流量池召回
  916. # break
  917. idx += get_size
  918. # log_.info({
  919. # 'logTimestamp': int(time.time() * 1000),
  920. # 'request_id': self.request_id,
  921. # 'operation': 'flow_pool_recall',
  922. # 'executeTime': (time.time() - start_time) * 1000
  923. # })
  924. return flow_pool_recall_result[:size], flow_pool_recall_process
  925. def flow_pool_recall_new_with_level_score2(self, size=10, flow_pool_id=None, flow_pool_abtest_group=None):
  926. """从流量池中获取视频"""
  927. # add_flow_pool_recall_log
  928. flow_pool_recall_process = {}
  929. # 获取存在城市分组数据的城市编码列表
  930. city_code_list = [code for _, code in config_.CITY_CODE.items()]
  931. # 获取provinceCode
  932. province_code = self.client_info.get('provinceCode', '-1')
  933. # 获取cityCode
  934. city_code = self.client_info.get('cityCode', '-1')
  935. if city_code in city_code_list:
  936. # 分城市数据存在时,获取城市分组数据
  937. region_code = city_code
  938. else:
  939. region_code = province_code
  940. if region_code == '':
  941. region_code = '-1'
  942. flow_pool_key, level = self.get_pool_redis_key('flow_set_level_score', flow_pool_id=flow_pool_id)
  943. # add_flow_pool_recall_log
  944. flow_pool_recall_process['flow_pool_key'] = flow_pool_key
  945. flow_pool_recall_process['level'] = level
  946. if flow_pool_key is None:
  947. return [], flow_pool_recall_process
  948. # print(flow_pool_key)
  949. flow_pool_recall_result = []
  950. flow_pool_recall_videos = []
  951. # 每次获取的视频数
  952. get_size = size * 5
  953. # 获取数据
  954. idx = 0
  955. data = self.redis_helper.get_data_zset_with_index(key_name=flow_pool_key,
  956. start=idx, end=idx + get_size * 5 - 1,
  957. with_scores=True)
  958. flow_pool_recall_process['initial_data'] = data
  959. if not data:
  960. return [], flow_pool_recall_process
  961. # 将video_id 与 flow_pool, score做mapping整理
  962. video_ids = []
  963. video_mapping = {}
  964. video_score = {}
  965. for value in data:
  966. try:
  967. video_id, flow_pool = value[0].split('-')
  968. except Exception as e:
  969. log_.error({
  970. 'request_id': self.request_id,
  971. 'app_type': self.app_type,
  972. 'flow_pool_value': value
  973. })
  974. continue
  975. video_id = int(video_id)
  976. video_score[value[0]] = value[1]
  977. if video_id not in video_ids:
  978. video_ids.append(video_id)
  979. if video_id not in video_mapping:
  980. video_mapping[video_id] = [flow_pool]
  981. else:
  982. video_mapping[video_id].append(flow_pool)
  983. # 检查可分发数
  984. ge = gevent.spawn(self.check_video_counts_new_with_level_score,
  985. video_ids=video_ids, flow_pool_mapping=video_mapping)
  986. ge.join()
  987. check_result = ge.get()
  988. # add_flow_pool_recall_log
  989. flow_pool_recall_process['check_counts_data'] = check_result
  990. check_result_mapping = {}
  991. check_result_items = []
  992. if check_result:
  993. # 获取score top20 视频进入过滤
  994. for item in check_result:
  995. video_id = int(item[0])
  996. flow_pool = item[1]
  997. score = video_score[f"{video_id}-{flow_pool}"]
  998. if video_id not in flow_pool_recall_videos:
  999. check_result_mapping[video_id] = [flow_pool, score]
  1000. check_result_items.append([video_id, flow_pool, score])
  1001. check_result_items = sorted(check_result_items, key=lambda x: x[2], reverse=True)
  1002. to_filter_videos = [item[0] for item in check_result_items[:get_size]]
  1003. # todo zhangbo 获取tags
  1004. video_tag_dict = self.get_video_tags(to_filter_videos)
  1005. # 过滤
  1006. filter_ = FilterVideos(request_id=self.request_id,
  1007. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=to_filter_videos,
  1008. expansion_factor=self.expansion_factor,
  1009. risk_filter_flag=self.risk_filter_flag,
  1010. app_region_filtered=self.app_region_filtered,
  1011. videos_with_risk=self.videos_with_risk
  1012. )
  1013. ge = gevent.spawn(filter_.filter_videos, pool_type='flow',
  1014. region_code=region_code, shield_config=self.shield_config)
  1015. ge.join()
  1016. filtered_result = ge.get()
  1017. # add_flow_pool_recall_log
  1018. flow_pool_recall_process['filtered_data'] = filtered_result
  1019. if filtered_result:
  1020. for item in filtered_result:
  1021. video_id = int(item)
  1022. # 添加视频源参数 pushFrom, abCode
  1023. flow_pool_recall_result.append(
  1024. {'videoId': video_id, 'flowPool': check_result_mapping[video_id][0], 'level': level,
  1025. 'rovScore': check_result_mapping[video_id][1],
  1026. 'pushFrom': config_.PUSH_FROM['flow_recall'],
  1027. 'abCode': self.ab_code, 'flow_pool_abtest_group': flow_pool_abtest_group,
  1028. "tags": video_tag_dict[video_id]}
  1029. )
  1030. return flow_pool_recall_result[:size], flow_pool_recall_process
  1031. def check_video_counts(self, video_ids, flow_pool_mapping):
  1032. """
  1033. 检查视频剩余可分发数
  1034. :param video_ids: 视频id type-list
  1035. :param flow_pool_mapping: 视频id-流量池标记mapping, type-dict
  1036. :return: check_result, error_flag
  1037. """
  1038. # flow_pool_key = self.get_pool_redis_key('flow')
  1039. # videos = []
  1040. check_result = []
  1041. for video_id in video_ids:
  1042. video_id = int(video_id)
  1043. for flow_pool in flow_pool_mapping.get(video_id, []):
  1044. # 判断是否有本地分发记录
  1045. cur_count = get_videos_local_distribute_count(video_id=video_id, flow_pool=flow_pool)
  1046. # 无记录
  1047. if cur_count is None:
  1048. # videos.append({'videoId': video_id, 'flowPool': flow_pool})
  1049. continue
  1050. # 本地分发数 cur_count > 0
  1051. elif cur_count > 0:
  1052. check_result.append((video_id, flow_pool))
  1053. # 本地分发数 cur_count <= 0,从所有的流量召回池移除,删除本地分发记录key
  1054. else:
  1055. add_remove_log = False
  1056. remain_count_key = f"{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{video_id}:{flow_pool}"
  1057. self.redis_helper.del_keys(remain_count_key)
  1058. value = '{}-{}'.format(video_id, flow_pool)
  1059. for item in config_.APP_TYPE:
  1060. flow_pool_key = f"{config_.FLOWPOOL_KEY_NAME_PREFIX}{config_.APP_TYPE.get(item)}"
  1061. remove_res = self.redis_helper.remove_value_from_zset(key_name=flow_pool_key, value=value)
  1062. if remove_res > 0:
  1063. add_remove_log = True
  1064. quick_flow_pool_key = f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX}{config_.APP_TYPE.get(item)}" \
  1065. f":{config_.QUICK_FLOW_POOL_ID}"
  1066. remove_res = self.redis_helper.remove_value_from_zset(key_name=quick_flow_pool_key, value=value)
  1067. if remove_res > 0:
  1068. add_remove_log = True
  1069. if add_remove_log is True:
  1070. log_.info({'tag': 'remove video_id from flow_pool', 'video_id': video_id, 'flow_pool': flow_pool})
  1071. return check_result
  1072. def check_video_counts_new(self, video_ids, flow_pool_mapping):
  1073. """
  1074. 检查视频剩余可分发数
  1075. :param video_ids: 视频id type-list
  1076. :param flow_pool_mapping: 视频id-流量池标记mapping, type-dict
  1077. :return: check_result, error_flag
  1078. """
  1079. # flow_pool_key = self.get_pool_redis_key('flow')
  1080. # videos = []
  1081. check_result = []
  1082. for video_id in video_ids:
  1083. video_id = int(video_id)
  1084. for flow_pool in flow_pool_mapping.get(video_id, []):
  1085. # 判断是否有本地分发记录
  1086. cur_count = get_videos_local_distribute_count(video_id=video_id, flow_pool=flow_pool)
  1087. # 无记录
  1088. if cur_count is None:
  1089. # videos.append({'videoId': video_id, 'flowPool': flow_pool})
  1090. continue
  1091. # 本地分发数 cur_count > 0
  1092. elif cur_count > 0:
  1093. check_result.append((video_id, flow_pool))
  1094. # 本地分发数 cur_count <= 0,从所有的流量召回池移除,删除本地分发记录key
  1095. else:
  1096. add_remove_log = False
  1097. remain_count_key = f"{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{video_id}:{flow_pool}"
  1098. self.redis_helper.del_keys(remain_count_key)
  1099. value = '{}-{}'.format(video_id, flow_pool)
  1100. for item in config_.APP_TYPE:
  1101. flow_pool_key = f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET}{config_.APP_TYPE.get(item)}"
  1102. remove_res = self.redis_helper.remove_value_from_set(key_name=flow_pool_key, values=(value, ))
  1103. if remove_res > 0:
  1104. add_remove_log = True
  1105. quick_flow_pool_key = f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX_SET}{config_.APP_TYPE.get(item)}" \
  1106. f":{config_.QUICK_FLOW_POOL_ID}"
  1107. remove_res = self.redis_helper.remove_value_from_set(key_name=quick_flow_pool_key,
  1108. values=(value, ))
  1109. if remove_res > 0:
  1110. add_remove_log = True
  1111. if add_remove_log is True:
  1112. log_.info({'tag': 'remove video_id from flow_pool', 'video_id': video_id, 'flow_pool': flow_pool})
  1113. return check_result
  1114. def check_video_counts_new_with_level(self, video_ids, flow_pool_mapping):
  1115. """
  1116. 检查视频剩余可分发数
  1117. :param video_ids: 视频id type-list
  1118. :param flow_pool_mapping: 视频id-流量池标记mapping, type-dict
  1119. :return: check_result, error_flag
  1120. """
  1121. # flow_pool_key = self.get_pool_redis_key('flow')
  1122. # videos = []
  1123. # level_weight = self.redis_helper.get_data_from_redis(key_name=config_.FLOWPOOL_LEVEL_WEIGHT_KEY_NAME)
  1124. # level_list = [level for level in json.loads(level_weight)]
  1125. if self.level_weight is None:
  1126. level_weight = {'1': 1, '2': 1, '3': 1, '4': 1, '5': 1, '6': 1}
  1127. else:
  1128. level_weight = self.level_weight
  1129. level_list = [level for level in level_weight]
  1130. check_result = []
  1131. for video_id in video_ids:
  1132. video_id = int(video_id)
  1133. for flow_pool in flow_pool_mapping.get(video_id, []):
  1134. # 判断是否有本地分发记录
  1135. cur_count = get_videos_local_distribute_count(video_id=video_id, flow_pool=flow_pool)
  1136. # 无记录
  1137. if cur_count is None:
  1138. # videos.append({'videoId': video_id, 'flowPool': flow_pool})
  1139. continue
  1140. # 本地分发数 cur_count > 0
  1141. elif cur_count > 0:
  1142. check_result.append((video_id, flow_pool))
  1143. # 本地分发数 cur_count <= 0,从所有的流量召回池移除,删除本地分发记录key
  1144. else:
  1145. add_remove_log = False
  1146. remain_count_key = f"{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{video_id}:{flow_pool}"
  1147. self.redis_helper.del_keys(remain_count_key)
  1148. value = '{}-{}'.format(video_id, flow_pool)
  1149. for item in config_.APP_TYPE:
  1150. for level in level_list:
  1151. flow_pool_key = f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL}{config_.APP_TYPE.get(item)}:{level}"
  1152. remove_res = self.redis_helper.remove_value_from_set(key_name=flow_pool_key, values=(value, ))
  1153. if remove_res > 0:
  1154. add_remove_log = True
  1155. quick_flow_pool_key = f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX_SET}{config_.APP_TYPE.get(item)}" \
  1156. f":{config_.QUICK_FLOW_POOL_ID}"
  1157. remove_res = self.redis_helper.remove_value_from_set(key_name=quick_flow_pool_key,
  1158. values=(value, ))
  1159. if remove_res > 0:
  1160. add_remove_log = True
  1161. if add_remove_log is True:
  1162. log_.info({'tag': 'remove video_id from flow_pool', 'video_id': video_id, 'flow_pool': flow_pool})
  1163. return check_result
  1164. def check_video_counts_new_with_level_score(self, video_ids, flow_pool_mapping):
  1165. """
  1166. 检查视频剩余可分发数
  1167. :param video_ids: 视频id type-list
  1168. :param flow_pool_mapping: 视频id-流量池标记mapping, type-dict
  1169. :return: check_result, error_flag
  1170. """
  1171. # flow_pool_key = self.get_pool_redis_key('flow')
  1172. # videos = []
  1173. # level_weight = self.redis_helper.get_data_from_redis(key_name=config_.FLOWPOOL_LEVEL_WEIGHT_KEY_NAME)
  1174. # level_list = [level for level in json.loads(level_weight)]
  1175. if self.level_weight is None:
  1176. level_weight = {'1': 1, '2': 1, '3': 1, '4': 1, '5': 1, '6': 1}
  1177. else:
  1178. level_weight = self.level_weight
  1179. level_list = [level for level in level_weight]
  1180. check_result = []
  1181. for video_id in video_ids:
  1182. video_id = int(video_id)
  1183. for flow_pool in flow_pool_mapping.get(video_id, []):
  1184. # 判断是否有本地分发记录
  1185. cur_count = get_videos_local_distribute_count(video_id=video_id, flow_pool=flow_pool)
  1186. # 无记录
  1187. if cur_count is None:
  1188. # videos.append({'videoId': video_id, 'flowPool': flow_pool})
  1189. continue
  1190. # 本地分发数 cur_count > 0
  1191. elif cur_count > 0:
  1192. check_result.append((video_id, flow_pool))
  1193. # 本地分发数 cur_count <= 0,从所有的流量召回池移除,删除本地分发记录key
  1194. else:
  1195. add_remove_log = False
  1196. remain_count_key = f"{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{video_id}:{flow_pool}"
  1197. self.redis_helper.del_keys(remain_count_key)
  1198. value = '{}-{}'.format(video_id, flow_pool)
  1199. for item in config_.APP_TYPE:
  1200. for level in level_list:
  1201. flow_pool_key = f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL_SCORE}{config_.APP_TYPE.get(item)}:{level}"
  1202. remove_res = self.redis_helper.remove_value_from_zset(key_name=flow_pool_key, value=value)
  1203. if remove_res > 0:
  1204. add_remove_log = True
  1205. quick_flow_pool_key = f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX}{config_.APP_TYPE.get(item)}" \
  1206. f":{config_.QUICK_FLOW_POOL_ID}"
  1207. remove_res = self.redis_helper.remove_value_from_zset(key_name=quick_flow_pool_key, value=value)
  1208. if remove_res > 0:
  1209. add_remove_log = True
  1210. if add_remove_log is True:
  1211. log_.info({'tag': 'remove video_id from flow_pool', 'video_id': video_id, 'flow_pool': flow_pool})
  1212. return check_result
  1213. """
  1214. # 本次视频都有本地记录
  1215. if len(videos) == 0:
  1216. error_flag = False
  1217. return check_result, error_flag
  1218. # 本地无记录视频,检查实时分发数
  1219. st_remain_view_count = time.time()
  1220. view_count_result, error_flag = get_videos_remain_view_count(app_type=self.app_type, videos=videos)
  1221. log_.info({
  1222. 'logTimestamp': int(time.time() * 1000),
  1223. 'request_id': self.request_id,
  1224. 'app_type': self.app_type,
  1225. 'mid': self.mid,
  1226. 'uid': self.uid,
  1227. 'operation': 'remainViewCount',
  1228. 'executeTime': (time.time() - st_remain_view_count) * 1000
  1229. })
  1230. # 判断返回的错误标记,True为错误
  1231. if error_flag:
  1232. return check_result, error_flag
  1233. # 从流量召回池移除视频videos
  1234. # for item in videos:
  1235. # value = '{}-{}'.format(item['videoId'], item['flowPool'])
  1236. # self.redis_helper.remove_value_from_zset(key_name=flow_pool_key, value=value)
  1237. # redis_helper = RedisHelper()
  1238. for item in view_count_result:
  1239. try:
  1240. # 接口超时,item[2]可能为None
  1241. remain_count = int(item[2])
  1242. except Exception as e:
  1243. # log_.error('remain_count type error...')
  1244. log_.error(traceback.format_exc())
  1245. continue
  1246. if remain_count > 0:
  1247. # viewCount > 0
  1248. check_result.append(item)
  1249. # 将分发数更新到本地记录
  1250. key_name = f"{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{item[0]}:{item[1]}"
  1251. self.redis_helper.setnx_key(key_name=key_name, value=remain_count, expire_time=5 * 60)
  1252. else:
  1253. # viewCount <= 0
  1254. # 从流量召回池移除
  1255. value = '{}-{}'.format(item[0], item[1])
  1256. for item in config_.APP_TYPE:
  1257. flow_pool_key = f"{config_.FLOWPOOL_KEY_NAME_PREFIX}{config_.APP_TYPE.get(item)}"
  1258. self.redis_helper.remove_value_from_zset(key_name=flow_pool_key, value=value)
  1259. return check_result, error_flag
  1260. """
  1261. def get_pool_redis_key(self, pool_type, flow_pool_id=None):
  1262. """
  1263. 拼接key
  1264. :param pool_type: type-string {'rov': rov召回池, 'flow': 流量池}
  1265. :param flow_pool_id: 流量池ID
  1266. :return: key_name
  1267. """
  1268. if pool_type == 'rov':
  1269. # appType = 13 票圈视频app
  1270. if self.app_type == config_.APP_TYPE['APP']:
  1271. key_name_prefix = config_.RECALL_KEY_NAME_PREFIX_APP
  1272. # abCode = 30001 # 老好看视频 / 票圈最惊奇 首页/相关推荐逻辑更新实验
  1273. # elif self.ab_code == config_.AB_CODE['rov_rank_appType_18_19']:
  1274. # key_name_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_APP_TYPE}{self.app_type}."
  1275. # 其他
  1276. else:
  1277. key_name_prefix = config_.RECALL_KEY_NAME_PREFIX
  1278. # 判断热度列表是否更新,未更新则使用前一天的热度列表
  1279. key_name = key_name_prefix + time.strftime('%Y%m%d')
  1280. if self.redis_helper.key_exists(key_name):
  1281. redis_date = date.today().strftime('%Y%m%d')
  1282. else:
  1283. redis_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d')
  1284. key_name = key_name_prefix + redis_date
  1285. # if not self.redis_helper.key_exists(key_name):
  1286. # return None, None
  1287. # 判断当前时间是否晚于rov召回池更新时间 + 1h,发送消息到飞书
  1288. # now_h = datetime.now().hour
  1289. # now_m = datetime.now().minute
  1290. # feishu_text = '{} —— 今日ROV召回池数据未按时更新,请及时查看解决。'.format(config_.ENV_TEXT)
  1291. # if now_h == config_.ROV_UPDATE_H + 1 and now_m > config_.ROV_UPDATE_MINUTE:
  1292. # send_msg_to_feishu(feishu_text)
  1293. # elif now_h > config_.ROV_UPDATE_H + 2:
  1294. # send_msg_to_feishu(feishu_text)
  1295. return key_name, redis_date
  1296. elif pool_type == 'flow':
  1297. if flow_pool_id == config_.QUICK_FLOW_POOL_ID:
  1298. return f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX}{self.app_type}:{config_.QUICK_FLOW_POOL_ID}"
  1299. else:
  1300. return f"{config_.FLOWPOOL_KEY_NAME_PREFIX}{self.app_type}"
  1301. elif pool_type == 'flow_set':
  1302. if flow_pool_id == config_.QUICK_FLOW_POOL_ID:
  1303. return f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX_SET}{self.app_type}:{config_.QUICK_FLOW_POOL_ID}"
  1304. else:
  1305. return f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET}{self.app_type}"
  1306. elif pool_type == 'flow_set_level':
  1307. if flow_pool_id == config_.QUICK_FLOW_POOL_ID:
  1308. return f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX_SET}{self.app_type}:{config_.QUICK_FLOW_POOL_ID}", None
  1309. else:
  1310. # 1. 获取流量池各层级分发概率权重
  1311. # level_weight = self.redis_helper.get_data_from_redis(key_name=config_.FLOWPOOL_LEVEL_WEIGHT_KEY_NAME)
  1312. # if level_weight is None:
  1313. # level_weight = {'1': 1, '2': 1, '3': 1, '4': 1, '5': 1, '6': 1}
  1314. # else:
  1315. # level_weight = json.loads(level_weight)
  1316. if self.level_weight is None:
  1317. level_weight = {'1': 1, '2': 1, '3': 1, '4': 1, '5': 1, '6': 1}
  1318. else:
  1319. level_weight = self.level_weight
  1320. # print(level_weight)
  1321. # 2. 判断各层级是否有视频需分发
  1322. available_level = []
  1323. for level, weight in level_weight.items():
  1324. level_key = f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL}{self.app_type}:{level}"
  1325. if self.redis_helper.key_exists(key_name=level_key):
  1326. available_level.append((level, level_key, weight))
  1327. if len(available_level) == 0:
  1328. return None, None
  1329. # 3. 根据可分发层级权重设置分发概率
  1330. available_level = sorted(available_level, key=lambda x: x[2], reverse=False)
  1331. weight_sum = sum([int(item[2]) for item in available_level])
  1332. level_p_mapping = {}
  1333. level_p_low = 0
  1334. weight_temp = 0
  1335. for item in available_level:
  1336. level, level_key, weight = item[0], item[1], item[2]
  1337. level_p_up = (weight_temp + weight)/weight_sum
  1338. level_p_mapping[level] = {
  1339. 'key': level_key,
  1340. 'level_p': [round(level_p_low, 2), round(level_p_up, 2)]
  1341. }
  1342. level_p_low = round(level_p_up, 2)
  1343. weight_temp += weight
  1344. # log_.info(f"level_p_mapping: {level_p_mapping}")
  1345. # 4. 随机生成[0,1)之间数,返回相应概率区间的key
  1346. random_p = random.random()
  1347. for level, level_info in level_p_mapping.items():
  1348. level_p = level_info['level_p']
  1349. if level_p[0] <= random_p < level_p[1]:
  1350. # log_.info(f"random_p: {random_p}, level_p: {level_p}, level: {level}")
  1351. return level_info['key'], level
  1352. else:
  1353. continue
  1354. return None, None
  1355. elif pool_type == 'flow_set_level_score':
  1356. if flow_pool_id == config_.QUICK_FLOW_POOL_ID:
  1357. return f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX}{self.app_type}:{config_.QUICK_FLOW_POOL_ID}", None
  1358. else:
  1359. # 1. 获取流量池各层级分发概率权重
  1360. if self.level_weight is None:
  1361. level_weight = {'1': 1, '2': 1, '3': 1, '4': 1, '5': 1, '6': 1}
  1362. else:
  1363. level_weight = self.level_weight
  1364. # print(level_weight)
  1365. # 2. 判断各层级是否有视频需分发
  1366. available_level = []
  1367. for level, weight in level_weight.items():
  1368. level_key = f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL_SCORE}{self.app_type}:{level}"
  1369. if self.redis_helper.key_exists(key_name=level_key):
  1370. available_level.append((level, level_key, weight))
  1371. if len(available_level) == 0:
  1372. return None, None
  1373. # 3. 根据可分发层级权重设置分发概率
  1374. available_level = sorted(available_level, key=lambda x: x[2], reverse=False)
  1375. weight_sum = sum([int(item[2]) for item in available_level])
  1376. level_p_mapping = {}
  1377. level_p_low = 0
  1378. weight_temp = 0
  1379. for item in available_level:
  1380. level, level_key, weight = item[0], item[1], item[2]
  1381. level_p_up = (weight_temp + weight)/weight_sum
  1382. level_p_mapping[level] = {
  1383. 'key': level_key,
  1384. 'level_p': [round(level_p_low, 2), round(level_p_up, 2)]
  1385. }
  1386. level_p_low = round(level_p_up, 2)
  1387. weight_temp += weight
  1388. # log_.info(f"level_p_mapping: {level_p_mapping}")
  1389. # 4. 随机生成[0,1)之间数,返回相应概率区间的key
  1390. random_p = random.random()
  1391. for level, level_info in level_p_mapping.items():
  1392. level_p = level_info['level_p']
  1393. if level_p[0] <= random_p < level_p[1]:
  1394. # log_.info(f"random_p: {random_p}, level_p: {level_p}, level: {level}")
  1395. return level_info['key'], level
  1396. else:
  1397. continue
  1398. return None, None
  1399. elif pool_type == 'special':
  1400. key_name_prefix = config_.KEY_NAME_PREFIX_SPECIAL_VIDEOS
  1401. # 判断列表是否更新,未更新则使用前一天的列表
  1402. key_name = f"{key_name_prefix}{time.strftime('%Y%m%d')}"
  1403. if self.redis_helper.key_exists(key_name):
  1404. redis_date = date.today().strftime('%Y%m%d')
  1405. else:
  1406. redis_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d')
  1407. key_name = f"{key_name_prefix}{redis_date}"
  1408. # 判断当前时间是否晚于更新时间,发送消息到飞书
  1409. # now_h = datetime.now().hour
  1410. # feishu_text = '{} —— 今日special mid 数据未按时更新,请及时查看解决。'.format(config_.ENV_TEXT)
  1411. # if now_h > config_.ROV_UPDATE_H:
  1412. # send_msg_to_feishu(feishu_text)
  1413. return key_name, redis_date
  1414. else:
  1415. log_.error('pool type error')
  1416. return None, None
  1417. def get_video_last_idx(self, video_type=''):
  1418. """获取用户上一次在rov召回池对应的位置"""
  1419. # if self.ab_code in [config_.AB_CODE['rank_by_h']] or self.app_type == config_.APP_TYPE['APP']:
  1420. # abCode = 30001 # 老好看视频 / 票圈最惊奇 首页/相关推荐逻辑更新实验
  1421. if self.ab_code in [code for _, code in config_.AB_CODE['rank_by_h'].items()] + \
  1422. [code for _, code in config_.AB_CODE['region_rank_by_h'].items()] + \
  1423. [config_.AB_CODE['rov_rank_appType_18_19'], config_.AB_CODE['rov_rank_appType_19'],
  1424. config_.AB_CODE['top_video_relevant_appType_19']] + \
  1425. [code for _, code in config_.AB_CODE['rank_by_24h'].items()] or \
  1426. video_type == 'whole_movies':
  1427. rov_pool_key, redis_date = self.get_pool_redis_key_with_h('rov', video_type=video_type)
  1428. elif self.ab_code in [code for _, code in config_.AB_CODE['rank_by_day'].items()]:
  1429. rov_pool_key, redis_date = self.get_pool_redis_key_with_day('dup')
  1430. else:
  1431. rov_pool_key, redis_date = self.get_pool_redis_key('rov')
  1432. if not rov_pool_key:
  1433. return None, None, None
  1434. if self.ab_code in [code for _, code in config_.AB_CODE['rank_by_day'].items()]:
  1435. now_h = datetime.now().hour
  1436. if now_h < 7:
  1437. last_key_prefix = config_.LAST_VIDEO_FROM_ROV_POOL_PRE_PREFIX
  1438. else:
  1439. last_key_prefix = config_.LAST_VIDEO_FROM_ROV_POOL_NOW_PREFIX
  1440. elif video_type == 'whole_movies':
  1441. last_key_prefix = config_.LAST_VIDEO_FROM_WHOLE_MOVIES_PREFIX
  1442. else:
  1443. last_key_prefix = config_.LAST_VIDEO_FROM_ROV_POOL_PREFIX
  1444. last_rov_recall_key = f'{last_key_prefix}{self.app_type}:{self.mid}:{redis_date}'
  1445. value = self.redis_helper.get_data_from_redis(last_rov_recall_key)
  1446. if value:
  1447. idx = self.redis_helper.get_index_with_data(rov_pool_key, value)
  1448. if not idx:
  1449. idx = 0
  1450. else:
  1451. idx += 1
  1452. else:
  1453. idx = 0
  1454. return rov_pool_key, last_rov_recall_key, idx
  1455. def get_update_rov_videos(self):
  1456. """
  1457. 获取修改ROV的视频
  1458. :return: update_rov_video_ids, update_rov_result
  1459. """
  1460. try:
  1461. # 获取修改过ROV的视频
  1462. if self.app_type == config_.APP_TYPE['APP']:
  1463. key_name = config_.UPDATE_ROV_KEY_NAME_APP
  1464. else:
  1465. key_name = config_.UPDATE_ROV_KEY_NAME
  1466. # redis_helper = RedisHelper()
  1467. data = self.redis_helper.get_data_zset_with_index(key_name=key_name,
  1468. start=0, end=-1, with_scores=True)
  1469. # 获取视频id,并转换类型为int,将videoId和score做mapping,并存储为key-value{videoId: score}
  1470. if data is None:
  1471. return [], []
  1472. video_ids = []
  1473. video_score = {}
  1474. for value in data:
  1475. video_id = int(value[0])
  1476. video_ids.append(video_id)
  1477. video_score[video_id] = value[1]
  1478. # 过滤
  1479. filter_ = FilterVideos(request_id=self.request_id,
  1480. app_type=self.app_type, video_ids=video_ids, mid=self.mid, uid=self.uid)
  1481. ge = gevent.spawn(filter_.filter_videos)
  1482. ge.join()
  1483. filtered_result = ge.get()
  1484. # 添加视频源参数 pushFrom, abCode
  1485. update_rov_video_ids, update_rov_result = [], []
  1486. if not filtered_result:
  1487. return update_rov_video_ids, update_rov_result
  1488. for item in filtered_result:
  1489. video_id = int(item)
  1490. rov_score = video_score.get(video_id)
  1491. if rov_score is None:
  1492. continue
  1493. update_rov_video_ids.append(video_id)
  1494. update_rov_result.append({'videoId': video_id, 'rovScore': rov_score,
  1495. 'pushFrom': config_.PUSH_FROM['rov_recall'], 'abCode': self.ab_code})
  1496. return update_rov_video_ids, update_rov_result
  1497. except Exception as e:
  1498. log_.error(traceback.format_exc())
  1499. return [], []
  1500. def get_top_videos(self):
  1501. """
  1502. 获取置顶视频
  1503. :return: top_video_ids, top_video_result
  1504. """
  1505. try:
  1506. # 获取生效中的置顶视频列表
  1507. # redis_helper = RedisHelper()
  1508. if self.app_type == config_.APP_TYPE['APP']:
  1509. key_name = config_.TOP_VIDEO_LIST_KEY_NAME_APP
  1510. else:
  1511. key_name = config_.TOP_VIDEO_LIST_KEY_NAME
  1512. data = self.redis_helper.get_data_from_redis(key_name=key_name)
  1513. # log_.info('===1=== {}'.format(data))
  1514. if data is None:
  1515. return [], []
  1516. # 获取视频id,并转换类型为int,将videoId和score做mapping,并存储为key-value{videoId: score}
  1517. video_ids = []
  1518. video_info = {}
  1519. for item in eval(data):
  1520. # log_.info('=== uid: {}, item === {}'.format(self.uid, item))
  1521. video_id = int(item['videoId'])
  1522. # 判断 视频推荐区域与用户地址信息 是否匹配,市级别
  1523. city_code_list = item['cityCode'].split(',')
  1524. # log_.info('=== uid: {}, city_code_list: {},{}, cityCode: {},{} === '.format(
  1525. # self.uid, city_code_list, type(city_code_list[0]), self.client_info.get('cityCode'),
  1526. # type(self.client_info.get('cityCode'))))
  1527. if self.client_info.get('cityCode') in city_code_list or config_.ALL_AREA_CODE in city_code_list:
  1528. # log_.info('=== uid: {}, ===2=== {}, ===3=== video_id: {}, cityCode: {}'.format(
  1529. # self.uid, self.client_info.get('cityCode'), video_id, city_code_list))
  1530. if video_id in video_ids:
  1531. # 如果video_id有重复,score保留大分值
  1532. if item['score'] > video_info[video_id].get('score'):
  1533. video_info[video_id] = {'score': item['score'], 'cityCode': city_code_list}
  1534. else:
  1535. continue
  1536. else:
  1537. video_ids.append(video_id)
  1538. video_info[video_id] = {'score': item['score'], 'cityCode': city_code_list}
  1539. else:
  1540. continue
  1541. # 过滤
  1542. filter_ = FilterVideos(request_id=self.request_id,
  1543. app_type=self.app_type, video_ids=video_ids, mid=self.mid, uid=self.uid)
  1544. ge = gevent.spawn(filter_.filter_videos)
  1545. ge.join()
  1546. filtered_result = ge.get()
  1547. # 添加视频源参数 pushFrom = 'op_manual', abCode
  1548. top_video_ids, top_video_result = [], []
  1549. if not filtered_result:
  1550. return top_video_ids, top_video_result
  1551. for item in filtered_result:
  1552. video_id = int(item)
  1553. item_info = video_info.get(video_id)
  1554. if item_info is None:
  1555. continue
  1556. top_video_ids.append(video_id)
  1557. top_video_result.append({'videoId': video_id, 'rovScore': item_info.get('score'),
  1558. 'pushFrom': config_.PUSH_FROM['top'], 'abCode': self.ab_code})
  1559. return top_video_ids, top_video_result
  1560. except Exception as e:
  1561. log_.error(traceback.format_exc())
  1562. return [], []
  1563. def get_pool_redis_key_with_h(self, pool_type, video_type=''):
  1564. """
  1565. 拼接key,获取以小时级别更新的视频列表
  1566. :param pool_type: type-string {'rov': rov召回池, 'flow': 流量池}
  1567. :param video_type: 视频列表区分 whole_movies - 完整影视资源
  1568. :return: key_name
  1569. """
  1570. if pool_type == 'rov':
  1571. now_date = date.today().strftime('%Y%m%d')
  1572. # 获取当前所在小时
  1573. h = datetime.now().hour
  1574. # appType = 13, 票圈视频APP
  1575. # 数据更新周期:每天07:00-21:00, 2h/次
  1576. # if self.app_type == config_.APP_TYPE['APP']:
  1577. # if h < 7:
  1578. # key_h = 21
  1579. # key_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d')
  1580. # elif h > 21:
  1581. # key_h = 21
  1582. # key_date = now_date
  1583. # else:
  1584. # if h % 2 == 0:
  1585. # key_h = h - 1
  1586. # key_date = now_date
  1587. # else:
  1588. # key_h = h
  1589. # key_date = now_date
  1590. # # print(key_date, key_h)
  1591. # key_name = f'{config_.APP_FINAL_RECALL_KEY_NAME_PREFIX}{key_date}.{key_h}'
  1592. # if self.redis_helper.key_exists(key_name):
  1593. # return key_name, key_h
  1594. # else:
  1595. # if key_h == 7:
  1596. # redis_h = 21
  1597. # redis_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d')
  1598. # else:
  1599. # redis_h = key_h - 2
  1600. # redis_date = key_date
  1601. # # print(redis_date, redis_h)
  1602. # key_name = f'{config_.APP_FINAL_RECALL_KEY_NAME_PREFIX}{redis_date}.{redis_h}'
  1603. # return key_name, redis_h
  1604. # elif self.app_type in [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]:
  1605. # abCode = 30001 # 老好看视频 / 票圈最惊奇 首页/相关推荐逻辑更新实验
  1606. if self.ab_code in [
  1607. config_.AB_CODE['rov_rank_appType_18_19'], config_.AB_CODE['rov_rank_appType_19'],
  1608. config_.AB_CODE['top_video_relevant_appType_19']
  1609. ]:
  1610. # 判断热度列表是否更新,未更新则使用前一小时的热度列表
  1611. key_name_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_APP_TYPE}{self.app_type}:"
  1612. key_name = f"{key_name_prefix}{now_date}:{h}"
  1613. if self.redis_helper.key_exists(key_name):
  1614. return key_name, h
  1615. else:
  1616. if h == 0:
  1617. redis_h = 23
  1618. redis_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d')
  1619. else:
  1620. redis_h = h - 1
  1621. redis_date = now_date
  1622. key_name = f"{key_name_prefix}{redis_date}:{redis_h}"
  1623. # 判断当前时间是否晚于数据正常更新时间,发送消息到飞书
  1624. # now_m = datetime.now().minute
  1625. # feishu_text = '{} —— appType = {}, h = {} 数据未按时更新,请及时查看解决。'.format(
  1626. # config_.ENV_TEXT, self.app_type, h)
  1627. # if now_m > config_.ROV_H_UPDATE_MINUTE:
  1628. # send_msg_to_feishu(feishu_text)
  1629. return key_name, redis_h
  1630. # 完整影视资源
  1631. elif video_type == 'whole_movies':
  1632. # 判断完整影视资源列表是否更新,未更新则使用前一小时的热度列表
  1633. key_name_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_WHOLE_MOVIES}"
  1634. key_name = f"{key_name_prefix}{now_date}.{h}"
  1635. if self.redis_helper.key_exists(key_name):
  1636. return key_name, h
  1637. else:
  1638. if h == 0:
  1639. redis_h = 23
  1640. redis_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d')
  1641. else:
  1642. redis_h = h - 1
  1643. redis_date = now_date
  1644. key_name = f"{key_name_prefix}{redis_date}.{redis_h}"
  1645. # 判断当前时间是否晚于数据正常更新时间,发送消息到飞书
  1646. # now_m = datetime.now().minute
  1647. # feishu_text = '{} —— appType = {}, h = {} 完整影视资源数据未按时更新,请及时查看解决。'.format(
  1648. # config_.ENV_TEXT, self.app_type, h)
  1649. # if now_m > config_.ROV_H_UPDATE_MINUTE:
  1650. # send_msg_to_feishu(feishu_text)
  1651. return key_name, redis_h
  1652. else:
  1653. # 判断热度列表是否更新,未更新则使用前一小时的热度列表
  1654. if self.ab_code in [code for _, code in config_.AB_CODE['region_rank_by_h'].items()]:
  1655. # 获取存在城市分组数据的城市编码列表
  1656. city_code_list = [code for _, code in config_.CITY_CODE.items()]
  1657. # 获取provinceCode
  1658. province_code = self.client_info.get('provinceCode', '-1')
  1659. # 获取cityCode
  1660. city_code = self.client_info.get('cityCode', '-1')
  1661. if city_code in city_code_list:
  1662. # 分城市数据存在时,获取城市分组数据
  1663. region_code = city_code
  1664. else:
  1665. region_code = province_code
  1666. if region_code == '':
  1667. region_code = '-1'
  1668. if self.ab_code == config_.AB_CODE['region_rank_by_h'].get('region_rule_rank2'):
  1669. key_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_DUP_REGION_24H}{region_code}."
  1670. else:
  1671. key_prefix = f"{config_.RECALL_KEY_NAME_PREFIX_DUP_REGION_H}{region_code}:"
  1672. elif self.ab_code in [code for _, code in config_.AB_CODE['rank_by_24h'].items()]:
  1673. key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP_24H
  1674. else:
  1675. key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP_H
  1676. key_name = f"{key_prefix}{self.data_key}:{self.rule_key}:{now_date}:{h}"
  1677. if self.redis_helper.key_exists(key_name):
  1678. return key_name, h
  1679. else:
  1680. if h == 0:
  1681. redis_h = 23
  1682. redis_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d')
  1683. else:
  1684. redis_h = h - 1
  1685. redis_date = now_date
  1686. key_name = f"{key_prefix}{self.data_key}:{self.rule_key}:{redis_date}:{redis_h}"
  1687. # 判断当前时间是否晚于数据正常更新时间,发送消息到飞书
  1688. # now_m = datetime.now().minute
  1689. # feishu_text = '{} —— appType = {}, h = {}, key = {}, province_code = {} 数据未按时更新,请及时查看解决。'.format(
  1690. # config_.ENV_TEXT, self.app_type, h, key_name, self.client_info.get('provinceCode', '-1'))
  1691. # if now_m > config_.ROV_H_UPDATE_MINUTE:
  1692. # send_msg_to_feishu(feishu_text)
  1693. return key_name, redis_h
  1694. elif pool_type == 'flow':
  1695. return config_.FLOW_POOL_KEY_NAME_PREFIX + str(self.app_type)
  1696. else:
  1697. # log_.error('pool type error')
  1698. return None, None
  1699. def flow_pool_recall_18_19(self, size=4, push_from=config_.PUSH_FROM['flow_recall']):
  1700. """从流量池中获取视频"""
  1701. start_time = time.time()
  1702. flow_pool_key = self.get_pool_redis_key('flow')
  1703. flow_pool_recall_result = []
  1704. flow_pool_recall_videos = []
  1705. # 每次获取的视频数
  1706. get_size = size * 5
  1707. # 记录获取频次
  1708. freq = 0
  1709. idx = 0
  1710. while len(flow_pool_recall_result) < size:
  1711. freq += 1
  1712. if freq > config_.MAX_FREQ_FROM_FLOW_POOL_18_19:
  1713. break
  1714. # 获取数据
  1715. data = self.redis_helper.get_data_zset_with_index(key_name=flow_pool_key,
  1716. start=idx, end=idx + get_size - 1,
  1717. with_scores=True)
  1718. # log_.info('get data from flow pool redis: freq = {}, data = {}, execute time = {}ms'.format(
  1719. # freq, data, (et_get - st_get) * 1000))
  1720. if not data:
  1721. # log_.info('流量池中的视频已取完')
  1722. break
  1723. # 将video_id 与 score做mapping整理
  1724. video_ids = []
  1725. video_score = {}
  1726. for value in data:
  1727. video_id = int(value[0])
  1728. video_ids.append(video_id)
  1729. video_score[video_id] = value[1]
  1730. # 过滤
  1731. filter_ = FilterVideos(request_id=self.request_id,
  1732. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
  1733. ge = gevent.spawn(filter_.filter_videos)
  1734. ge.join()
  1735. filtered_result = ge.get()
  1736. # 添加视频源参数 pushFrom, abCode
  1737. if filtered_result:
  1738. # 添加视频源参数 pushFrom, abCode
  1739. temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)],
  1740. 'pushFrom': push_from, 'abCode': self.ab_code}
  1741. for item in filtered_result if video_score.get(int(item)) is not None]
  1742. flow_pool_recall_result.extend(temp_result)
  1743. idx += get_size
  1744. log_.info({
  1745. 'logTimestamp': int(time.time() * 1000),
  1746. 'request_id': self.request_id,
  1747. 'operation': 'flow_pool_recall_18_19',
  1748. 'executeTime': (time.time() - start_time) * 1000
  1749. })
  1750. return flow_pool_recall_result[:size]
  1751. def get_pool_redis_key_with_day(self, pool_type):
  1752. """
  1753. 拼接key,获取以天级别规则更新的视频列表
  1754. :param pool_type: type-string {'rov': rov召回池, 'flow': 流量池}
  1755. :return: key_name
  1756. """
  1757. now_date = datetime.today()
  1758. now_dt = datetime.strftime(now_date, '%Y%m%d')
  1759. if pool_type == 'rov':
  1760. # 判断列表是否更新,未更新则使用前一天的列表
  1761. rule_key_name = f'{config_.RECALL_KEY_NAME_PREFIX_BY_DAY}{self.rule_key}.{now_dt}'
  1762. if self.redis_helper.key_exists(key_name=rule_key_name):
  1763. return rule_key_name, now_dt
  1764. else:
  1765. redis_dt = datetime.strftime((now_date - timedelta(days=1)), '%Y%m%d')
  1766. rule_key_name = f'{config_.RECALL_KEY_NAME_PREFIX_BY_DAY}{self.rule_key}.{redis_dt}'
  1767. # 判断当前时间是否晚于数据正常更新时间,发送消息到飞书
  1768. # now_h = datetime.now().hour
  1769. # feishu_text = f'{config_.ENV_TEXT} —— appType = {self.app_type}, date = {now_dt}, ' \
  1770. # f'rule_key = {self.rule_key} 数据未按时更新,请及时查看解决。'
  1771. # if now_h > config_.ROV_DAY_UPDATE_MINUTE:
  1772. # send_msg_to_feishu(feishu_text)
  1773. return rule_key_name, redis_dt
  1774. elif pool_type == 'dup':
  1775. now_h = datetime.now().hour
  1776. if now_h < 7:
  1777. dup_key_name_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP_DAY_PRE
  1778. else:
  1779. dup_key_name_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP_DAY_NOW
  1780. dup_key_name = f'{dup_key_name_prefix}{self.rule_key}.{now_dt}'
  1781. if self.redis_helper.key_exists(key_name=dup_key_name):
  1782. return dup_key_name, now_dt
  1783. else:
  1784. redis_dt = datetime.strftime((now_date - timedelta(days=1)), '%Y%m%d')
  1785. dup_key_name = f'{dup_key_name_prefix}{self.rule_key}.{redis_dt}'
  1786. return dup_key_name, redis_dt
  1787. elif pool_type == 'flow':
  1788. return config_.FLOW_POOL_KEY_NAME_PREFIX + str(self.app_type)
  1789. else:
  1790. # log_.error('pool type error')
  1791. return None, None
  1792. def get_video_last_idx_day(self):
  1793. """获取用户上一次在天级规则更新列表中对应的位置"""
  1794. rule_key_name, redis_dt = self.get_pool_redis_key_with_day('rov')
  1795. if not rule_key_name:
  1796. return None, None, None
  1797. last_rule_day_recall_key = \
  1798. f'{config_.LAST_VIDEO_FROM_RULE_DAY_POOL_PREFIX}{self.app_type}.{self.mid}.{redis_dt}'
  1799. value = self.redis_helper.get_data_from_redis(last_rule_day_recall_key)
  1800. if value:
  1801. idx = self.redis_helper.get_index_with_data(rule_key_name, value)
  1802. if not idx:
  1803. idx = 0
  1804. else:
  1805. idx += 1
  1806. else:
  1807. idx = 0
  1808. return rule_key_name, last_rule_day_recall_key, idx
  1809. def old_videos_recall(self, size):
  1810. """老视频召回"""
  1811. # 获取老视频
  1812. now_dt = datetime.strftime(datetime.today(), '%Y%m%d')
  1813. key_name = f'{config_.RECALL_KEY_NAME_PREFIX_OLD_VIDEOS}{now_dt}'
  1814. old_videos = self.redis_helper.get_data_from_set(key_name=key_name)
  1815. if not old_videos:
  1816. return []
  1817. # 过滤
  1818. old_video_ids = [int(video_id) for video_id in old_videos]
  1819. filter_ = FilterVideos(request_id=self.request_id,
  1820. app_type=self.app_type, video_ids=old_video_ids, mid=self.mid, uid=self.uid)
  1821. ge = gevent.spawn(filter_.filter_videos)
  1822. ge.join()
  1823. filtered_videos = ge.get()
  1824. if not filtered_videos:
  1825. return []
  1826. # 添加视频源参数 pushFrom, abCode
  1827. old_video_result = [{'videoId': int(item), 'rovScore': 0,
  1828. 'pushFrom': config_.PUSH_FROM['old_video'], 'abCode': self.ab_code}
  1829. for item in filtered_videos]
  1830. # 随机抽取 size+1 条数据
  1831. random.shuffle(old_video_result)
  1832. return old_video_result[:size+1]
  1833. def get_region_dup_video_last_idx_h(self, province_code, region_dup=None):
  1834. """获取用户上一次在 地域分组 相关去重列表中对应的位置"""
  1835. now_date = date.today().strftime('%Y%m%d')
  1836. h = datetime.now().hour
  1837. if region_dup == 1:
  1838. # 小程序地域分组天级更新结果与小程序地域分组小时级更新结果去重
  1839. key_name_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_DAY_H
  1840. elif region_dup == 2:
  1841. if self.ab_code == config_.AB_CODE['region_rank_by_h'].get('region_rule_rank2'):
  1842. # 小程序天级更新结果与 小程序地域分组小时级更新24h结果 去重
  1843. key_name_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP_REGION_DAY_24H
  1844. else:
  1845. # 小程序天级更新结果与 小程序地域分组天级更新结果/小程序地域分组小时级更新结果 去重
  1846. key_name_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_DAY_H
  1847. else:
  1848. key_name_prefix = ''
  1849. key_name = f"{key_name_prefix}{province_code}.{self.rule_key}.{now_date}.{h}"
  1850. last_region_dup_key = \
  1851. f'{config_.LAST_VIDEO_FROM_REGION_DUP_PREFIX}{region_dup}.{self.app_type}.{self.mid}.{h}'
  1852. if not self.redis_helper.key_exists(key_name=key_name):
  1853. if h == 0:
  1854. redis_h = 23
  1855. redis_date = (date.today() - timedelta(days=1)).strftime('%Y%m%d')
  1856. else:
  1857. redis_h = h - 1
  1858. redis_date = now_date
  1859. key_name = f"{key_name_prefix}{province_code}.{self.rule_key}.{redis_date}.{redis_h}"
  1860. last_region_dup_key = \
  1861. f"{config_.LAST_VIDEO_FROM_REGION_DUP_PREFIX}{region_dup}.{self.app_type}.{self.mid}.{redis_h}"
  1862. # 判断当前时间是否晚于数据正常更新时间,发送消息到飞书
  1863. # now_m = datetime.now().minute
  1864. # feishu_text = f"{config_.ENV_TEXT} —— appType = {self.app_type}, h = {h}, key = {key_name}, " \
  1865. # f"province_code = {province_code} 数据未按时更新,请及时查看解决。"
  1866. # if now_m > config_.REGION_H_UPDATE_MINUTE:
  1867. # send_msg_to_feishu(feishu_text)
  1868. value = self.redis_helper.get_data_from_redis(last_region_dup_key)
  1869. if value:
  1870. idx = self.redis_helper.get_index_with_data(key_name, value)
  1871. if not idx:
  1872. idx = 0
  1873. else:
  1874. idx += 1
  1875. else:
  1876. idx = 0
  1877. return key_name, last_region_dup_key, idx
  1878. def rov_pool_recall_with_region_process(self, size=4, expire_time=24*3600):
  1879. """
  1880. 地域分组召回视频
  1881. :param size: 获取视频个数
  1882. :param expire_time: 末位视频记录redis过期时间
  1883. :return:
  1884. """
  1885. start_time = time.time()
  1886. # 获取存在城市分组数据的城市编码列表
  1887. city_code_list = [code for _, code in config_.CITY_CODE.items()]
  1888. # 获取provinceCode
  1889. province_code = self.client_info.get('provinceCode', '-1')
  1890. # 获取cityCode
  1891. city_code = self.client_info.get('cityCode', '-1')
  1892. if city_code in city_code_list:
  1893. # 分城市数据存在时,获取城市分组数据
  1894. region_code = city_code
  1895. else:
  1896. region_code = province_code
  1897. if region_code == '':
  1898. region_code = '-1'
  1899. if region_code == '-1':
  1900. t = [
  1901. gevent.spawn(self.recall_region_dup_24h, region_code, size, 'h', expire_time),
  1902. gevent.spawn(self.recall_region_dup_24h, region_code, size, '24h_dup2', expire_time),
  1903. gevent.spawn(self.recall_region_dup_24h, region_code, size, '24h_dup3', expire_time)
  1904. ]
  1905. else:
  1906. t = [
  1907. #add recall video
  1908. # gevent.spawn(self.rov_pool_recall_with_region_by_h, province_code, size),
  1909. # gevent.spawn(self.rov_pool_recall_with_region_by_h, province_code, size, 'region_24h'),
  1910. # gevent.spawn(self.rov_pool_recall_with_region_by_h, province_code, size, 'day_24h'),
  1911. gevent.spawn(self.recall_region_dup_24h, region_code, size, 'region_h', expire_time),
  1912. gevent.spawn(self.recall_region_dup_24h, region_code, size, 'h', expire_time),
  1913. gevent.spawn(self.recall_region_dup_24h, region_code, size, 'region_24h', expire_time),
  1914. gevent.spawn(self.recall_region_dup_24h, region_code, size, '24h_dup2', expire_time),
  1915. gevent.spawn(self.recall_region_dup_24h, region_code, size, '24h_dup3', expire_time)
  1916. ]
  1917. gevent.joinall(t)
  1918. region_recall_result_list = [i.get() for i in t]
  1919. # 将已获取到的视频按顺序去重合并
  1920. now_video_ids = []
  1921. recall_result = []
  1922. recall_num = size
  1923. # if ab_code and exp_config:
  1924. # if ab_code==60058 or ab_code==60059 or ab_code == 60060 \
  1925. # or ab_code == 60061 or ab_code==60052 \
  1926. # or ab_code==60053 or ab_code==60057 :
  1927. # try:
  1928. # recall_num = int(exp_config['recall_num'])
  1929. # except:
  1930. # recall_num = size
  1931. # if recall_num<size:
  1932. # recall_num = size
  1933. # todo zhangbo rank 放开地域的截断,给后续排序更多供给 做规则调控
  1934. if self.ab_code == 60098:
  1935. recall_num = size * 3
  1936. for region_result in region_recall_result_list:
  1937. for video in region_result:
  1938. video_id = video.get('videoId')
  1939. if video_id not in now_video_ids:
  1940. recall_result.append(video)
  1941. now_video_ids.append(video_id)
  1942. if len(recall_result) >= recall_num:
  1943. break
  1944. else:
  1945. continue
  1946. # # 130实验组不获取大列表的数据
  1947. # if self.ab_code != config_.AB_CODE['region_rank_by_h'].get('abtest_130'):
  1948. # # 判断获取到的小时级数据数量
  1949. # if len(recall_result) < size:
  1950. # # 补充数据
  1951. # rov_recall_result = self.rov_pool_recall(size=size, expire_time=expire_time)
  1952. # # 去重合并
  1953. # for video in rov_recall_result:
  1954. # vid = video.get('videoId')
  1955. # if vid not in now_video_ids:
  1956. # recall_result.append(video)
  1957. # now_video_ids.append(vid)
  1958. # if len(recall_result) >= size:
  1959. # break
  1960. # else:
  1961. # continue
  1962. # log_.info({
  1963. # 'logTimestamp': int(time.time() * 1000),
  1964. # 'request_id': self.request_id,
  1965. # 'operation': 'rov_pool_recall_with_region',
  1966. # 'executeTime': (time.time() - start_time) * 1000
  1967. # })
  1968. #print("recall_num:", recall_num)
  1969. #print("recall_result:", recall_result[:recall_num])
  1970. if self.h_data_key is None or self.h_rule_key is None:
  1971. return recall_result[:recall_num]
  1972. else:
  1973. # 补充不区分地域小时级结果(不做离线去重)
  1974. recall_h_h_result = self.recall_h(province_code=region_code, size=size, expire_time=expire_time)
  1975. # 去重
  1976. recall_result = recall_result[:recall_num]
  1977. recall_video_ids = [video.get('videoId') for video in recall_result]
  1978. for video in recall_h_h_result:
  1979. video_id = video.get('videoId')
  1980. if video_id not in recall_video_ids:
  1981. recall_result.append(video)
  1982. recall_video_ids.append(video_id)
  1983. return recall_result
  1984. def rov_pool_recall_with_region(self, size=4, expire_time=24*3600):
  1985. """召回池召回视频"""
  1986. # 获取召回池中视频
  1987. videos = self.rov_pool_recall_with_region_process(size=size, expire_time=expire_time)
  1988. # 对在流量池中存在的视频添加标记字段
  1989. result = []
  1990. for item in videos:
  1991. video_id = item['videoId']
  1992. t = [
  1993. gevent.spawn(self.get_video_flow_pool, video_id, True),
  1994. gevent.spawn(self.get_video_flow_pool, video_id, False)
  1995. ]
  1996. gevent.joinall(t)
  1997. flow_pool_list = [i.get() for i in t]
  1998. flow_pool_list = [item for item in flow_pool_list if item != '']
  1999. if len(flow_pool_list) > 0:
  2000. flow_pool = flow_pool_list[0]
  2001. item['flowPool'] = flow_pool
  2002. item['isInFlowPool'] = 1
  2003. result.append(item)
  2004. return result
  2005. def get_video_flow_pool(self, video_id, quick_flow_pool=False):
  2006. """
  2007. 获取videoId对应的任意一个flowPool
  2008. :param video_id: videoId
  2009. :param quick_flow_pool: 是否为快速曝光流量池标识,默认:否 False
  2010. :return: flow_pool
  2011. """
  2012. if quick_flow_pool is True:
  2013. isin_flow_pool_key = \
  2014. f"{config_.QUICK_FLOWPOOL_VIDEO_ID_KEY_NAME_PREFIX}{self.app_type}:{config_.QUICK_FLOW_POOL_ID}"
  2015. flow_pool_key = \
  2016. f"{config_.QUICK_FLOWPOOL_VIDEO_INFO_KEY_NAME_PREFIX}{self.app_type}:" \
  2017. f"{config_.QUICK_FLOW_POOL_ID}:{video_id}"
  2018. else:
  2019. isin_flow_pool_key = \
  2020. f"{config_.FLOWPOOL_VIDEO_ID_KEY_NAME_PREFIX}{self.app_type}"
  2021. flow_pool_key = \
  2022. f"{config_.FLOWPOOL_VIDEO_INFO_KEY_NAME_PREFIX}{self.app_type}:{video_id}"
  2023. # 判断是否在流量池中
  2024. isin_flow_pool = self.redis_helper.data_exists_with_set(key_name=isin_flow_pool_key, value=video_id)
  2025. flow_pool = ''
  2026. if isin_flow_pool:
  2027. # 随机获取一个flowPool标记
  2028. flow_pool_list = self.redis_helper.get_data_with_count_from_set(key_name=flow_pool_key, count=1)
  2029. if len(flow_pool_list) > 0:
  2030. flow_pool = flow_pool_list[0]
  2031. return flow_pool
  2032. def get_flow_pool_videos(self):
  2033. """获取当前可分发的流量池视频,以及对应的标记列表"""
  2034. video_id_list = []
  2035. videos_flow_pool = {}
  2036. # 快速曝光流量池
  2037. key_name_quick = f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX}{self.app_type}:{config_.QUICK_FLOW_POOL_ID}"
  2038. # 其他流量池
  2039. key_name_other = f"{config_.FLOWPOOL_KEY_NAME_PREFIX}{self.app_type}"
  2040. for key_name in [key_name_quick, key_name_other]:
  2041. data = self.redis_helper.get_all_data_from_zset(key_name=key_name, desc=True, with_scores=False)
  2042. if data is None or len(data) == 0:
  2043. continue
  2044. for item in data:
  2045. video_id, flow_pool = item.split('-')
  2046. video_id = int(video_id)
  2047. # ### 对该视频做分发数检查
  2048. cur_count = get_videos_local_distribute_count(video_id=video_id, flow_pool=flow_pool)
  2049. # 无记录
  2050. if cur_count is None:
  2051. continue
  2052. # 本地分发数 cur_count <= 0,从所有的流量召回池移除,删除本地分发记录key
  2053. if cur_count <= 0:
  2054. remain_count_key = f"{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{video_id}:{flow_pool}"
  2055. self.redis_helper.del_keys(remain_count_key)
  2056. for app_name in config_.APP_TYPE:
  2057. flow_pool_key = f"{config_.FLOWPOOL_KEY_NAME_PREFIX}{config_.APP_TYPE.get(app_name)}"
  2058. quick_flow_pool_key = f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX}" \
  2059. f"{config_.APP_TYPE.get(app_name)}:{config_.QUICK_FLOW_POOL_ID}"
  2060. self.redis_helper.remove_value_from_zset(key_name=flow_pool_key, value=item)
  2061. self.redis_helper.remove_value_from_zset(key_name=quick_flow_pool_key, value=item)
  2062. continue
  2063. # 本地分发数 cur_count > 0
  2064. if video_id in video_id_list:
  2065. videos_flow_pool[video_id].append(flow_pool)
  2066. else:
  2067. videos_flow_pool[video_id] = [flow_pool]
  2068. video_id_list.append(video_id)
  2069. return {'video_id_list': video_id_list, 'videos_flow_pool': videos_flow_pool}
  2070. def rov_pool_recall_with_region_by_h(self, province_code, size=4, key_flag=''):
  2071. """
  2072. 地域分组小时级视频召回
  2073. :param size: 视频数
  2074. :param province_code: 省份code
  2075. :param key_flag:
  2076. :return:
  2077. """
  2078. if key_flag == 'region_24h':
  2079. push_from = config_.PUSH_FROM['rov_recall_region_24h']
  2080. elif key_flag == 'day_24h':
  2081. push_from = config_.PUSH_FROM['rov_recall_24h']
  2082. else:
  2083. push_from = config_.PUSH_FROM['rov_recall_region_h']
  2084. # 获取mid对应的小时级列表redis-key
  2085. h_recall_mid_key = self.get_mid_h_key(province_code=province_code, key_flag=key_flag)
  2086. if not self.redis_helper.key_exists(h_recall_mid_key):
  2087. recall_result = []
  2088. else:
  2089. # 过滤的视频
  2090. fil_video_ids = []
  2091. recall_result = []
  2092. # 每次获取的视频数
  2093. get_size = size * 5
  2094. # 记录获取频次
  2095. freq = 0
  2096. while len(recall_result) < size:
  2097. freq += 1
  2098. if freq > config_.MAX_FREQ_FROM_ROV_POOL:
  2099. break
  2100. # 获取数据
  2101. data = self.redis_helper.get_data_zset_with_index(key_name=h_recall_mid_key,
  2102. start=(freq - 1) * get_size, end=freq * get_size - 1,
  2103. with_scores=True)
  2104. if not data:
  2105. # log_.info('地域分组小时级更新视频已取完')
  2106. break
  2107. # 获取视频id,并转换类型为int,并存储为key-value{videoId: score}
  2108. video_ids = []
  2109. video_score = {}
  2110. for value in data:
  2111. video_id = int(value[0])
  2112. video_ids.append(video_id)
  2113. video_score[video_id] = value[1]
  2114. # 过滤
  2115. filter_ = FilterVideos(request_id=self.request_id,
  2116. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
  2117. ge = gevent.spawn(filter_.filter_videos_h, self.rule_key, self.data_key,
  2118. self.ab_code, province_code, key_flag)
  2119. ge.join()
  2120. filtered_result = ge.get()
  2121. if filtered_result:
  2122. # 添加视频源参数 pushFrom, abCode
  2123. temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)],
  2124. 'pushFrom': push_from, 'abCode': self.ab_code}
  2125. for item in filtered_result if video_score.get(int(item)) is not None]
  2126. recall_result.extend(temp_result)
  2127. fil_video_ids.extend(list(set(video_ids) - set([item.get('videoId') for item in temp_result])))
  2128. else:
  2129. fil_video_ids.extend(video_ids)
  2130. # 将被过滤的视频进行移除
  2131. for value in fil_video_ids:
  2132. self.redis_helper.remove_value_from_zset(key_name=h_recall_mid_key, value=value)
  2133. return recall_result[:size]
  2134. def region_dup_recall(self, province_code, region_dup, size=4, expire_time=23*3600):
  2135. """
  2136. region dup 更新列表视频召回
  2137. :param province_code:
  2138. :param region_dup:
  2139. :param size:
  2140. :param expire_time:
  2141. :return:
  2142. """
  2143. if region_dup == 1:
  2144. push_from = config_.PUSH_FROM['rov_recall_region_day']
  2145. elif region_dup == 2:
  2146. push_from = config_.PUSH_FROM['rov_recall_day']
  2147. # 获取region dup更新列表相关redis key, 用户上一次在列表对应的位置
  2148. key_name, last_region_dup_key, idx = self.get_region_dup_video_last_idx_h(
  2149. province_code=province_code, region_dup=region_dup)
  2150. # 获取天级规则更新列表数据
  2151. if not key_name:
  2152. # log_.info(f'region dup 更新列表中无视频, region_dup = {region_dup}')
  2153. recall_result = []
  2154. else:
  2155. recall_result = []
  2156. # 每次获取的视频数
  2157. get_size = size * 5
  2158. # 记录获取频次
  2159. freq = 0
  2160. while len(recall_result) < size:
  2161. freq += 1
  2162. if freq > config_.MAX_FREQ_FROM_ROV_POOL:
  2163. break
  2164. # 获取数据
  2165. data = self.redis_helper.get_data_zset_with_index(key_name=key_name,
  2166. start=idx, end=idx + get_size - 1,
  2167. with_scores=True)
  2168. if not data:
  2169. # log_.info(f'region dup 更新视频已取完, region_dup = {region_dup}')
  2170. break
  2171. # 获取视频id,并转换类型为int,并存储为key-value{videoId: score}
  2172. video_ids = []
  2173. video_score = {}
  2174. for value in data:
  2175. video_id = int(value[0])
  2176. video_ids.append(video_id)
  2177. video_score[video_id] = value[1]
  2178. # 过滤
  2179. filter_ = FilterVideos(request_id=self.request_id,
  2180. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
  2181. ge = gevent.spawn(filter_.filter_videos)
  2182. ge.join()
  2183. filtered_result = ge.get()
  2184. if filtered_result:
  2185. # 添加视频源参数 pushFrom, abCode
  2186. temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)],
  2187. 'pushFrom': push_from, 'abCode': self.ab_code}
  2188. for item in filtered_result if video_score.get(int(item)) is not None]
  2189. recall_result.extend(temp_result)
  2190. else:
  2191. # 将此次获取的末位视频id同步刷新到Redis中,方便下次快速定位到召回位置,过期时间为1天
  2192. if self.mid:
  2193. # mid为空时,不做记录
  2194. self.redis_helper.set_data_to_redis(key_name=last_region_dup_key, value=data[-1][0],
  2195. expire_time=expire_time)
  2196. idx += get_size
  2197. return recall_result[:size]
  2198. def rule_recall_by_h(self, size=4, expire_time=24*3600):
  2199. """
  2200. 小时级召回视频
  2201. :param size: 获取视频个数
  2202. :param expire_time: 末位视频记录redis过期时间
  2203. :return:
  2204. """
  2205. start_time = time.time()
  2206. t = [gevent.spawn(self.rov_pool_recall_h, size),
  2207. gevent.spawn(self.rov_pool_recall_h, size, '24h')]
  2208. gevent.joinall(t)
  2209. h_recall_result_list = [i.get() for i in t]
  2210. # 将已获取到的视频按顺序去重合并
  2211. now_video_ids = []
  2212. recall_result = []
  2213. for h_result in h_recall_result_list:
  2214. for video in h_result:
  2215. video_id = video.get('videoId')
  2216. if video_id not in now_video_ids:
  2217. recall_result.append(video)
  2218. now_video_ids.append(video_id)
  2219. if len(recall_result) >= size:
  2220. break
  2221. else:
  2222. continue
  2223. # 判断获取到的小时级数据数量
  2224. if len(recall_result) < size:
  2225. # 补充数据
  2226. rov_recall_result = self.rov_pool_recall(size=size, expire_time=expire_time)
  2227. # 去重合并
  2228. for video in rov_recall_result:
  2229. vid = video.get('videoId')
  2230. if vid not in now_video_ids:
  2231. recall_result.append(video)
  2232. now_video_ids.append(vid)
  2233. if len(recall_result) >= size:
  2234. break
  2235. else:
  2236. continue
  2237. log_.info({
  2238. 'logTimestamp': int(time.time() * 1000),
  2239. 'request_id': self.request_id,
  2240. 'operation': 'rule_recall_by_h',
  2241. 'executeTime': (time.time() - start_time) * 1000
  2242. })
  2243. return recall_result[:size]
  2244. def rov_pool_recall_h(self, size=4, key_flag=''):
  2245. """
  2246. 小时级视频召回
  2247. :param size: 视频数
  2248. :param key_flag:
  2249. :return:
  2250. """
  2251. if key_flag == '24h':
  2252. push_from = config_.PUSH_FROM['rov_recall_24h']
  2253. else:
  2254. push_from = config_.PUSH_FROM['rov_recall_h']
  2255. # 获取mid对应的小时级列表redis-key
  2256. h_recall_mid_key = self.get_mid_h_key(province_code='', key_flag=key_flag)
  2257. if not self.redis_helper.key_exists(h_recall_mid_key):
  2258. recall_result = []
  2259. else:
  2260. # 过滤的视频
  2261. fil_video_ids = []
  2262. recall_result = []
  2263. # 每次获取的视频数
  2264. get_size = size * 5
  2265. # 记录获取频次
  2266. freq = 0
  2267. while len(recall_result) < size:
  2268. freq += 1
  2269. if freq > config_.MAX_FREQ_FROM_ROV_POOL:
  2270. break
  2271. # 获取数据
  2272. data = self.redis_helper.get_data_zset_with_index(key_name=h_recall_mid_key,
  2273. start=(freq - 1) * get_size, end=freq * get_size - 1,
  2274. with_scores=True)
  2275. if not data:
  2276. # log_.info('小时级更新视频已取完')
  2277. break
  2278. # 获取视频id,并转换类型为int,并存储为key-value{videoId: score}
  2279. video_ids = []
  2280. video_score = {}
  2281. for value in data:
  2282. video_id = int(value[0])
  2283. video_ids.append(video_id)
  2284. video_score[video_id] = value[1]
  2285. # 过滤
  2286. filter_ = FilterVideos(request_id=self.request_id,
  2287. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
  2288. ge = gevent.spawn(filter_.filter_videos_h, self.rule_key, self.ab_code, '', key_flag)
  2289. ge.join()
  2290. filtered_result = ge.get()
  2291. if filtered_result:
  2292. # 添加视频源参数 pushFrom, abCode
  2293. temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)],
  2294. 'pushFrom': push_from, 'abCode': self.ab_code}
  2295. for item in filtered_result if video_score.get(int(item)) is not None]
  2296. recall_result.extend(temp_result)
  2297. fil_video_ids.extend(list(set(video_ids) - set([item.get('videoId') for item in temp_result])))
  2298. else:
  2299. fil_video_ids.extend(video_ids)
  2300. # 将被过滤的视频进行移除
  2301. for value in fil_video_ids:
  2302. self.redis_helper.remove_value_from_zset(key_name=h_recall_mid_key, value=value)
  2303. return recall_result[:size]
  2304. def get_relevant_videos_19(self, video_id, size=4):
  2305. """
  2306. 获取最惊奇相关推荐视频
  2307. :param video_id: 头部视频id
  2308. :return: relevant_result
  2309. """
  2310. push_from = config_.PUSH_FROM['top_video_relevant_appType_19']
  2311. relevant_result = []
  2312. relevant_videos_key_name = f"{config_.MOVIE_RELEVANT_LIST_KEY_NAME_PREFIX}{video_id}"
  2313. # redis_helper = RedisHelper()
  2314. if not self.redis_helper.key_exists(key_name=relevant_videos_key_name):
  2315. return relevant_result
  2316. # 获取数据
  2317. data = self.redis_helper.get_data_zset_with_index(key_name=relevant_videos_key_name, start=0, end=-1,
  2318. with_scores=True)
  2319. if not data:
  2320. return relevant_result
  2321. # 获取视频id,并转换类型为int,并存储为key-value{videoId: score}
  2322. video_ids = []
  2323. video_score = {}
  2324. for value in data:
  2325. video_id = int(value[0])
  2326. video_ids.append(video_id)
  2327. video_score[video_id] = value[1]
  2328. # 过滤
  2329. filter_ = FilterVideos(request_id=self.request_id, app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
  2330. ge = gevent.spawn(filter_.filter_videos)
  2331. ge.join()
  2332. filtered_result = ge.get()
  2333. if filtered_result:
  2334. # 添加视频源参数 pushFrom, abCode
  2335. relevant_result = [{'videoId': int(item), 'rovScore': video_score[int(item)],
  2336. 'pushFrom': push_from, 'abCode': self.ab_code}
  2337. for item in filtered_result if video_score.get(int(item)) is not None]
  2338. return relevant_result[:size]
  2339. def relevant_recall_19(self, video_id, size=4, expire_time=24*3600):
  2340. """
  2341. 最惊奇相关推荐视频召回
  2342. :param video_id:
  2343. :param size:
  2344. :param expire_time:
  2345. :return:
  2346. """
  2347. t = [gevent.spawn(self.get_relevant_videos_19, video_id, size),
  2348. gevent.spawn(self.rov_pool_recall, size, expire_time)]
  2349. gevent.joinall(t)
  2350. relevant_recall_result_list = [i.get() for i in t]
  2351. # 将已获取到的视频按顺序去重合并
  2352. now_video_ids = []
  2353. recall_result = []
  2354. for relevant_result in relevant_recall_result_list:
  2355. for video in relevant_result:
  2356. video_id = video.get('videoId')
  2357. if video_id not in now_video_ids:
  2358. recall_result.append(video)
  2359. now_video_ids.append(video_id)
  2360. return recall_result[:size]
  2361. def rov_pool_recall_19(self, size=4, expire_time=24*3600):
  2362. """
  2363. 最惊奇视频召回
  2364. :param size: 获取视频个数
  2365. :param expire_time: 末位视频记录redis过期时间
  2366. :return:
  2367. """
  2368. t = [gevent.spawn(self.rov_pool_recall, size, expire_time=3600, video_type='whole_movies', push_from=config_.PUSH_FROM['whole_movies']),
  2369. gevent.spawn(self.flow_pool_recall_18_19, size, push_from=config_.PUSH_FROM['talk_videos'])]
  2370. gevent.joinall(t)
  2371. recall_result_list = [i.get() for i in t]
  2372. # 将已获取到的视频按顺序去重合并
  2373. now_video_ids = []
  2374. recall_result = []
  2375. for item in recall_result_list:
  2376. for video in item:
  2377. video_id = video.get('videoId')
  2378. if video_id not in now_video_ids:
  2379. recall_result.append(video)
  2380. now_video_ids.append(video_id)
  2381. if len(recall_result) >= size:
  2382. break
  2383. else:
  2384. continue
  2385. # 判断获取到的小时级数据数量
  2386. if len(recall_result) < size:
  2387. # 补充数据
  2388. rov_recall_result = self.rov_pool_recall(size=size, expire_time=expire_time)
  2389. # 去重合并
  2390. for video in rov_recall_result:
  2391. vid = video.get('videoId')
  2392. if vid not in now_video_ids:
  2393. recall_result.append(video)
  2394. now_video_ids.append(vid)
  2395. if len(recall_result) >= size:
  2396. break
  2397. else:
  2398. continue
  2399. return recall_result[:size]
  2400. def update_last_video_record(self, record_key, pool_key_prefix, province_code, data_key=None, rule_key=None):
  2401. if data_key is None or rule_key is None:
  2402. data_key = self.data_key
  2403. rule_key = self.rule_key
  2404. # 判断当前小时的小时级列表是否更新
  2405. now_date = datetime.today()
  2406. h = datetime.now().hour
  2407. now_dt = datetime.strftime(now_date, '%Y%m%d')
  2408. # now_pool_recall_key = f"{pool_key_prefix}{province_code}:{self.data_key}:{self.rule_key}:{now_dt}:{h}"
  2409. now_pool_recall_key = f"{pool_key_prefix}{province_code}:{data_key}:{rule_key}:{now_dt}:{h}"
  2410. if self.redis_helper.key_exists(key_name=now_pool_recall_key):
  2411. value = {'date': now_dt, 'h': h}
  2412. self.redis_helper.set_data_to_redis(key_name=record_key, value=str(value), expire_time=2 * 3600)
  2413. else:
  2414. if h == 0:
  2415. redis_dt = datetime.strftime(now_date - timedelta(days=1), '%Y%m%d')
  2416. redis_h = 23
  2417. else:
  2418. redis_dt = now_dt
  2419. redis_h = h - 1
  2420. # now_pool_recall_key = f"{pool_key_prefix}{province_code}:{self.data_key}:{self.rule_key}:{redis_dt}:{redis_h}"
  2421. now_pool_recall_key = f"{pool_key_prefix}{province_code}:{data_key}:{rule_key}:{redis_dt}:{redis_h}"
  2422. value = {'date': redis_dt, 'h': redis_h}
  2423. self.redis_helper.set_data_to_redis(key_name=record_key, value=str(value), expire_time=2 * 3600)
  2424. return now_pool_recall_key
  2425. def update_last_video_record_without_region(self, record_key, pool_key_prefix, data_key=None, rule_key=None):
  2426. if data_key is None or rule_key is None:
  2427. data_key = self.data_key
  2428. rule_key = self.rule_key
  2429. # 判断当前小时的小时级列表是否更新
  2430. now_date = datetime.today()
  2431. h = datetime.now().hour
  2432. now_dt = datetime.strftime(now_date, '%Y%m%d')
  2433. now_pool_recall_key = f"{pool_key_prefix}{data_key}:{rule_key}:{now_dt}:{h}"
  2434. if self.redis_helper.key_exists(key_name=now_pool_recall_key):
  2435. value = {'date': now_dt, 'h': h}
  2436. self.redis_helper.set_data_to_redis(key_name=record_key, value=str(value), expire_time=2 * 3600)
  2437. else:
  2438. if h == 0:
  2439. redis_dt = datetime.strftime(now_date - timedelta(days=1), '%Y%m%d')
  2440. redis_h = 23
  2441. else:
  2442. redis_dt = now_dt
  2443. redis_h = h - 1
  2444. now_pool_recall_key = f"{pool_key_prefix}{data_key}:{rule_key}:{redis_dt}:{redis_h}"
  2445. value = {'date': redis_dt, 'h': redis_h}
  2446. self.redis_helper.set_data_to_redis(key_name=record_key, value=str(value), expire_time=2 * 3600)
  2447. return now_pool_recall_key
  2448. def get_video_idx(self, pool_recall_key, last_video_key):
  2449. """
  2450. 获取上次视频所在位置
  2451. :param pool_recall_key: 视频所在列表 key
  2452. :param last_video_key: 上一次记录的视频
  2453. :return: idx
  2454. """
  2455. value = self.redis_helper.get_data_from_redis(last_video_key)
  2456. if value:
  2457. idx = self.redis_helper.get_index_with_data(key_name=pool_recall_key, value=value)
  2458. if not idx:
  2459. idx = 0
  2460. else:
  2461. idx += 1
  2462. else:
  2463. idx = 0
  2464. return idx
  2465. def get_last_recommend_video_idx(self, province_code, record_key_prefix, pool_key_prefix, last_video_key_prefix,
  2466. data_key=None, rule_key=None):
  2467. if data_key is None or rule_key is None:
  2468. data_key = self.data_key
  2469. rule_key = self.rule_key
  2470. # 判断mid对应上一次视频位置 时间记录
  2471. record_key = f"{record_key_prefix}{self.app_type}:{self.mid}"
  2472. last_video_key = f'{last_video_key_prefix}{self.app_type}:{self.mid}'
  2473. if not self.redis_helper.key_exists(key_name=record_key):
  2474. # ###### 记录key不存在
  2475. self.redis_helper.del_keys(key_name=last_video_key)
  2476. idx = 0
  2477. pool_recall_key = self.update_last_video_record(record_key=record_key, pool_key_prefix=pool_key_prefix,
  2478. province_code=province_code,
  2479. data_key=data_key, rule_key=rule_key)
  2480. else:
  2481. # ###### 记录key存在,判断date, h
  2482. now_date = datetime.today()
  2483. h = datetime.now().hour
  2484. # 获取记录的date, h
  2485. record = self.redis_helper.get_data_from_redis(key_name=record_key)
  2486. record_dt = eval(record).get('date')
  2487. record_h = eval(record).get('h')
  2488. now_dt = datetime.strftime(now_date, '%Y%m%d')
  2489. if record_dt == now_dt and int(record_h) == h:
  2490. # 已获取当前小时数据
  2491. # pool_recall_key = f"{pool_key_prefix}{province_code}:{self.data_key}:{self.rule_key}:{now_dt}:{h}"
  2492. pool_recall_key = f"{pool_key_prefix}{province_code}:{data_key}:{rule_key}:{now_dt}:{h}"
  2493. idx = self.get_video_idx(pool_recall_key=pool_recall_key, last_video_key=last_video_key)
  2494. elif (record_dt == now_dt and h-int(record_h) == 1) or (h == 0 and int(record_h) == 23):
  2495. # 记录的h - 当前h = 1,判断当前h数据是否已更新
  2496. # now_pool_recall_key = f"{pool_key_prefix}{province_code}:{self.data_key}:{self.rule_key}:{now_dt}:{h}"
  2497. now_pool_recall_key = f"{pool_key_prefix}{province_code}:{data_key}:{rule_key}:{now_dt}:{h}"
  2498. if self.redis_helper.key_exists(key_name=now_pool_recall_key):
  2499. new_record = {'date': now_dt, 'h': h}
  2500. self.redis_helper.set_data_to_redis(key_name=record_key, value=str(new_record), expire_time=2*3600)
  2501. idx = 0
  2502. self.redis_helper.del_keys(key_name=last_video_key)
  2503. pool_recall_key = now_pool_recall_key
  2504. else:
  2505. # pool_recall_key = f"{pool_key_prefix}{province_code}:{self.data_key}:{self.rule_key}:{record_dt}:{record_h}"
  2506. pool_recall_key = f"{pool_key_prefix}{province_code}:{data_key}:{rule_key}:{record_dt}:{record_h}"
  2507. idx = self.get_video_idx(pool_recall_key=pool_recall_key, last_video_key=last_video_key)
  2508. else:
  2509. idx = 0
  2510. self.redis_helper.del_keys(key_name=last_video_key)
  2511. pool_recall_key = self.update_last_video_record(record_key=record_key, pool_key_prefix=pool_key_prefix,
  2512. province_code=province_code,
  2513. data_key=data_key, rule_key=rule_key)
  2514. return pool_recall_key, idx
  2515. def get_last_recommend_video_idx_without_region(self, record_key_prefix, pool_key_prefix, last_video_key_prefix,
  2516. data_key=None, rule_key=None):
  2517. if data_key is None or rule_key is None:
  2518. data_key = self.data_key
  2519. rule_key = self.rule_key
  2520. # 判断mid对应上一次视频位置 时间记录
  2521. record_key = f"{record_key_prefix}{self.app_type}:{self.mid}"
  2522. last_video_key = f'{last_video_key_prefix}{self.app_type}:{self.mid}'
  2523. if not self.redis_helper.key_exists(key_name=record_key):
  2524. # ###### 记录key不存在
  2525. self.redis_helper.del_keys(key_name=last_video_key)
  2526. idx = 0
  2527. pool_recall_key = self.update_last_video_record_without_region(record_key=record_key,
  2528. pool_key_prefix=pool_key_prefix,
  2529. data_key=data_key, rule_key=rule_key)
  2530. else:
  2531. # ###### 记录key存在,判断date, h
  2532. now_date = datetime.today()
  2533. h = datetime.now().hour
  2534. # 获取记录的date, h
  2535. record = self.redis_helper.get_data_from_redis(key_name=record_key)
  2536. record_dt = eval(record).get('date')
  2537. record_h = eval(record).get('h')
  2538. now_dt = datetime.strftime(now_date, '%Y%m%d')
  2539. if record_dt == now_dt and int(record_h) == h:
  2540. # 已获取当前小时数据
  2541. pool_recall_key = f"{pool_key_prefix}{data_key}:{rule_key}:{now_dt}:{h}"
  2542. idx = self.get_video_idx(pool_recall_key=pool_recall_key, last_video_key=last_video_key)
  2543. elif (record_dt == now_dt and h-int(record_h) == 1) or (h == 0 and int(record_h) == 23):
  2544. # 记录的h - 当前h = 1,判断当前h数据是否已更新
  2545. now_pool_recall_key = f"{pool_key_prefix}{data_key}:{rule_key}:{now_dt}:{h}"
  2546. if self.redis_helper.key_exists(key_name=now_pool_recall_key):
  2547. new_record = {'date': now_dt, 'h': h}
  2548. self.redis_helper.set_data_to_redis(key_name=record_key, value=str(new_record), expire_time=2*3600)
  2549. idx = 0
  2550. self.redis_helper.del_keys(key_name=last_video_key)
  2551. pool_recall_key = now_pool_recall_key
  2552. else:
  2553. pool_recall_key = f"{pool_key_prefix}{data_key}:{rule_key}:{record_dt}:{record_h}"
  2554. idx = self.get_video_idx(pool_recall_key=pool_recall_key, last_video_key=last_video_key)
  2555. else:
  2556. idx = 0
  2557. self.redis_helper.del_keys(key_name=last_video_key)
  2558. pool_recall_key = self.update_last_video_record_without_region(record_key=record_key,
  2559. pool_key_prefix=pool_key_prefix,
  2560. data_key=data_key, rule_key=rule_key)
  2561. return pool_recall_key, idx
  2562. def recall_region_dup_24h(self, province_code, size=4, key_flag='', expire_time=2*3600):
  2563. """
  2564. 从小程序小时级24h数据 筛选后的剩余数据 更新结果中获取视频
  2565. :param size: 获取视频个数
  2566. :param key_flag: 视频表标记
  2567. :param expire_time: 末位视频记录redis过期时间
  2568. :return:
  2569. """
  2570. start_time = time.time()
  2571. if key_flag == 'region_h':
  2572. # 分地域小时级的筛选结果
  2573. # mid对应上一次视频位置 时间记录
  2574. record_key_prefix = config_.RECORD_KEY_NAME_PREFIX_LAST_VIDEO_REGION_H
  2575. # 视频列表
  2576. pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H
  2577. # mid对应上一次视频记录
  2578. last_video_key_prefix = config_.LAST_VIDEO_FROM_REGION_H_PREFIX
  2579. push_from = config_.PUSH_FROM['rov_recall_region_h']
  2580. elif key_flag == 'h':
  2581. # 不分地域小时级的筛选结果
  2582. # mid对应上一次视频位置 时间记录
  2583. record_key_prefix = config_.RECORD_KEY_NAME_PREFIX_LAST_VIDEO_REGION_DUP_H
  2584. # 视频列表
  2585. pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP_H_H
  2586. # mid对应上一次视频记录
  2587. last_video_key_prefix = config_.LAST_VIDEO_FROM_REGION_DUP_H_PREFIX
  2588. push_from = config_.PUSH_FROM['rov_recall_h_h']
  2589. elif key_flag == 'region_24h':
  2590. # 分地域相对24h的筛选结果
  2591. # mid对应上一次视频位置 时间记录
  2592. record_key_prefix = config_.RECORD_KEY_NAME_PREFIX_LAST_VIDEO_REGION_DUP1_24H
  2593. # 视频列表
  2594. pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H
  2595. # mid对应上一次视频记录
  2596. last_video_key_prefix = config_.LAST_VIDEO_FROM_REGION_DUP1_24H_PREFIX
  2597. push_from = config_.PUSH_FROM['rov_recall_region_24h']
  2598. elif key_flag == '24h_dup2':
  2599. # 不分地域相对24h的筛选结果
  2600. # mid对应上一次视频位置 时间记录
  2601. record_key_prefix = config_.RECORD_KEY_NAME_PREFIX_LAST_VIDEO_REGION_DUP2_24H
  2602. # 视频列表
  2603. pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_24H_H
  2604. # mid对应上一次视频记录
  2605. last_video_key_prefix = config_.LAST_VIDEO_FROM_REGION_DUP2_24H_PREFIX
  2606. push_from = config_.PUSH_FROM['rov_recall_24h']
  2607. elif key_flag == '24h_dup3':
  2608. # 不分地域相对24h的筛选后剩余数据
  2609. record_key_prefix = config_.RECORD_KEY_NAME_PREFIX_LAST_VIDEO_REGION_DUP3_24H
  2610. pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H
  2611. last_video_key_prefix = config_.LAST_VIDEO_FROM_REGION_DUP3_24H_PREFIX
  2612. push_from = config_.PUSH_FROM['rov_recall_24h_dup']
  2613. elif key_flag == '48h_dup2':
  2614. # 不分地域相对48h的筛选结果
  2615. # mid对应上一次视频位置 时间记录
  2616. record_key_prefix = config_.RECORD_KEY_NAME_PREFIX_LAST_VIDEO_REGION_DUP2_48H
  2617. # 视频列表
  2618. pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_48H_H
  2619. # mid对应上一次视频记录
  2620. last_video_key_prefix = config_.LAST_VIDEO_FROM_REGION_DUP2_48H_PREFIX
  2621. push_from = config_.PUSH_FROM['rov_recall_48h']
  2622. elif key_flag == '48h_dup3':
  2623. # 不分地域相对48h的筛选后剩余数据
  2624. record_key_prefix = config_.RECORD_KEY_NAME_PREFIX_LAST_VIDEO_REGION_DUP3_48H
  2625. pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_48H_H
  2626. last_video_key_prefix = config_.LAST_VIDEO_FROM_REGION_DUP3_48H_PREFIX
  2627. push_from = config_.PUSH_FROM['rov_recall_48h_dup']
  2628. else:
  2629. return []
  2630. # 获取相关redis key, 用户上一次在rov召回池对应的位置
  2631. pool_key, idx = self.get_last_recommend_video_idx(province_code=province_code,
  2632. record_key_prefix=record_key_prefix,
  2633. pool_key_prefix=pool_key_prefix,
  2634. last_video_key_prefix=last_video_key_prefix)
  2635. if not pool_key:
  2636. return []
  2637. recall_data = []
  2638. pool_recall_result = []
  2639. # 每次获取的视频数
  2640. # get_size = size * 5
  2641. get_size = size * self.expansion_factor
  2642. # 记录获取频次
  2643. freq = 0
  2644. while len(pool_recall_result) < size:
  2645. freq += 1
  2646. if freq > config_.MAX_FREQ_FROM_ROV_POOL:
  2647. break
  2648. # 获取数据
  2649. data = self.redis_helper.get_data_zset_with_index(key_name=pool_key,
  2650. start=idx, end=idx + get_size - 1,
  2651. with_scores=True)
  2652. if not data:
  2653. break
  2654. recall_data.extend(data)
  2655. # 获取视频id,并转换类型为int,并存储为key-value{videoId: score}
  2656. video_ids = []
  2657. video_score = {}
  2658. for value in data:
  2659. video_id = int(value[0])
  2660. video_ids.append(video_id)
  2661. video_score[video_id] = value[1]
  2662. # todo zhangbo 增加tags特征
  2663. video_tag_dict = self.get_video_tags(video_ids)
  2664. # 过滤
  2665. filter_ = FilterVideos(request_id=self.request_id,
  2666. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids,
  2667. expansion_factor=self.expansion_factor,
  2668. risk_filter_flag=self.risk_filter_flag,
  2669. app_region_filtered=self.app_region_filtered,
  2670. videos_with_risk=self.videos_with_risk,
  2671. force_truncation=20
  2672. )
  2673. ge = gevent.spawn(filter_.filter_videos, '', province_code, None)
  2674. ge.join()
  2675. filtered_result = ge.get()
  2676. if filtered_result:
  2677. # 添加视频源参数 pushFrom, abCode
  2678. temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)],
  2679. 'pushFrom': push_from, 'abCode': self.ab_code,
  2680. 'tags': video_tag_dict[item]}
  2681. for item in filtered_result if video_score.get(int(item)) is not None]
  2682. pool_recall_result.extend(temp_result)
  2683. # else:
  2684. # # 将此次获取的末位视频id同步刷新到Redis中,方便下次快速定位到召回位置,过期时间为1天
  2685. # if self.mid and self.ab_code != config_.AB_CODE['region_rank_by_h'].get('abtest_112'):
  2686. # # mid为空时,不做记录
  2687. # last_video_key = f'{last_video_key_prefix}{self.app_type}:{self.mid}'
  2688. # self.redis_helper.set_data_to_redis(key_name=last_video_key, value=data[-1][0],
  2689. # expire_time=expire_time)
  2690. idx += get_size
  2691. pool_recall_result.sort(key=lambda x: x.get('rovScore', 0), reverse=True)
  2692. # if len(recall_data) > 0 and len(pool_recall_result) == 0 \
  2693. # and self.ab_code == config_.AB_CODE['region_rank_by_h'].get('abtest_112') and self.mid:
  2694. if len(recall_data) > 0 and len(pool_recall_result) == 0 and self.mid:
  2695. # 召回数据不为空 & 过滤后结果为空 & 位于实验组 & mid不为空时,更新召回获取的末位视频id记录到定位的key中
  2696. last_video_key = f'{last_video_key_prefix}{self.app_type}:{self.mid}'
  2697. self.redis_helper.set_data_to_redis(key_name=last_video_key, value=recall_data[-1][0],
  2698. expire_time=expire_time)
  2699. # log_.info({
  2700. # 'logTimestamp': int(time.time() * 1000),
  2701. # 'request_id': self.request_id,
  2702. # 'operation': push_from,
  2703. # 'pool_recall_result': pool_recall_result,
  2704. # 'executeTime': (time.time() - start_time) * 1000
  2705. # })
  2706. return pool_recall_result[:size]
  2707. def recall_h(self, province_code, size=4, expire_time=2*3600):
  2708. """
  2709. 从不区分地域小时级列表 更新结果中获取视频
  2710. :param size: 获取视频个数
  2711. :param expire_time: 末位视频记录redis过期时间
  2712. :return:
  2713. """
  2714. # 不分地域小时级列表
  2715. # mid对应上一次视频位置 时间记录
  2716. record_key_prefix = config_.RECORD_KEY_NAME_PREFIX_LAST_VIDEO_H
  2717. # 视频列表
  2718. pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_BY_H_H
  2719. # mid对应上一次视频记录
  2720. last_video_key_prefix = config_.LAST_VIDEO_FROM_H_PREFIX
  2721. push_from = config_.PUSH_FROM['rov_recall_h_h_without_dup']
  2722. # 获取相关redis key, 用户上一次在rov召回池对应的位置
  2723. pool_key, idx = self.get_last_recommend_video_idx_without_region(record_key_prefix=record_key_prefix,
  2724. pool_key_prefix=pool_key_prefix,
  2725. last_video_key_prefix=last_video_key_prefix,
  2726. data_key=self.h_data_key,
  2727. rule_key=self.h_rule_key)
  2728. if not pool_key:
  2729. return []
  2730. recall_data = []
  2731. pool_recall_result = []
  2732. # 每次获取的视频数
  2733. get_size = size * 5
  2734. # 记录获取频次
  2735. freq = 0
  2736. while len(pool_recall_result) < size:
  2737. freq += 1
  2738. if freq > config_.MAX_FREQ_FROM_ROV_POOL:
  2739. break
  2740. # 获取数据
  2741. data = self.redis_helper.get_data_zset_with_index(key_name=pool_key,
  2742. start=idx, end=idx + get_size - 1,
  2743. with_scores=True)
  2744. if not data:
  2745. break
  2746. recall_data.extend(data)
  2747. # 获取视频id,并转换类型为int,并存储为key-value{videoId: score}
  2748. video_ids = []
  2749. video_score = {}
  2750. for value in data:
  2751. video_id = int(value[0])
  2752. video_ids.append(video_id)
  2753. video_score[video_id] = value[1]
  2754. # 过滤
  2755. filter_ = FilterVideos(request_id=self.request_id,
  2756. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
  2757. ge = gevent.spawn(filter_.filter_videos, pool_type='normal',
  2758. region_code=province_code, shield_config=self.shield_config)
  2759. ge.join()
  2760. filtered_result = ge.get()
  2761. if filtered_result:
  2762. # 添加视频源参数 pushFrom, abCode
  2763. temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)],
  2764. 'pushFrom': push_from, 'abCode': self.ab_code}
  2765. for item in filtered_result if video_score.get(int(item)) is not None]
  2766. pool_recall_result.extend(temp_result)
  2767. idx += get_size
  2768. pool_recall_result.sort(key=lambda x: x.get('rovScore', 0), reverse=True)
  2769. if len(recall_data) > 0 and len(pool_recall_result) == 0 and self.mid:
  2770. # 召回数据不为空 & 过滤后结果为空 & 位于实验组 & mid不为空时,更新召回获取的末位视频id记录到定位的key中
  2771. last_video_key = f'{last_video_key_prefix}{self.app_type}:{self.mid}'
  2772. self.redis_helper.set_data_to_redis(key_name=last_video_key, value=recall_data[-1][0],
  2773. expire_time=expire_time)
  2774. return pool_recall_result
  2775. def update_last_video_record_by_day(self, record_key, pool_key_prefix, expire_time):
  2776. # 判断当前日期的小时级列表是否更新
  2777. now_date = datetime.today()
  2778. now_dt = datetime.strftime(now_date, '%Y%m%d')
  2779. now_pool_recall_key = f"{pool_key_prefix}{self.data_key}:{self.rule_key_30day}:{now_dt}"
  2780. if self.redis_helper.key_exists(key_name=now_pool_recall_key):
  2781. value = {'date': now_dt}
  2782. self.redis_helper.set_data_to_redis(key_name=record_key, value=str(value), expire_time=expire_time)
  2783. else:
  2784. redis_dt = datetime.strftime(now_date - timedelta(days=1), '%Y%m%d')
  2785. now_pool_recall_key = f"{pool_key_prefix}{self.data_key}:{self.rule_key_30day}:{redis_dt}"
  2786. value = {'date': redis_dt}
  2787. self.redis_helper.set_data_to_redis(key_name=record_key, value=str(value), expire_time=expire_time)
  2788. return now_pool_recall_key
  2789. def get_last_recommend_video_idx_by_day(self, record_key_prefix, pool_key_prefix, last_video_key_prefix, expire_time):
  2790. # 判断mid对应上一次视频位置 时间记录
  2791. record_key = f"{record_key_prefix}{self.app_type}:{self.mid}"
  2792. last_video_key = f'{last_video_key_prefix}{self.app_type}:{self.mid}'
  2793. if not self.redis_helper.key_exists(key_name=record_key):
  2794. # ###### 记录key不存在
  2795. self.redis_helper.del_keys(key_name=last_video_key)
  2796. idx = 0
  2797. pool_recall_key = self.update_last_video_record_by_day(record_key=record_key,
  2798. pool_key_prefix=pool_key_prefix,
  2799. expire_time=expire_time)
  2800. else:
  2801. # ###### 记录key存在,判断date
  2802. now_date = datetime.today()
  2803. # 获取记录的date
  2804. record = self.redis_helper.get_data_from_redis(key_name=record_key)
  2805. record_dt = eval(record).get('date')
  2806. now_dt = datetime.strftime(now_date, '%Y%m%d')
  2807. if record_dt == now_dt:
  2808. # 已获取当前日期数据
  2809. pool_recall_key = f"{pool_key_prefix}{self.data_key}:{self.rule_key_30day}:{now_dt}"
  2810. idx = self.get_video_idx(pool_recall_key=pool_recall_key, last_video_key=last_video_key)
  2811. elif record_dt == datetime.strftime((now_date - timedelta(days=1)), '%Y%m%d'):
  2812. # 记录的dt - 当前dt = 1,判断当前h数据是否已更新
  2813. now_pool_recall_key = f"{pool_key_prefix}{self.data_key}:{self.rule_key_30day}:{now_dt}"
  2814. if self.redis_helper.key_exists(key_name=now_pool_recall_key):
  2815. new_record = {'date': now_dt}
  2816. self.redis_helper.set_data_to_redis(key_name=record_key,
  2817. value=str(new_record),
  2818. expire_time=expire_time)
  2819. idx = 0
  2820. self.redis_helper.del_keys(key_name=last_video_key)
  2821. pool_recall_key = now_pool_recall_key
  2822. else:
  2823. pool_recall_key = f"{pool_key_prefix}{self.data_key}:{self.rule_key_30day}:{record_dt}"
  2824. idx = self.get_video_idx(pool_recall_key=pool_recall_key, last_video_key=last_video_key)
  2825. else:
  2826. idx = 0
  2827. self.redis_helper.del_keys(key_name=last_video_key)
  2828. pool_recall_key = self.update_last_video_record_by_day(record_key=record_key,
  2829. pool_key_prefix=pool_key_prefix,
  2830. expire_time=expire_time)
  2831. return pool_recall_key, idx
  2832. def recall_update_by_day(self, size=4, key_flag='', expire_time=24*3600):
  2833. """
  2834. 从天级更新列表中获取视频
  2835. :param size: 获取视频个数
  2836. :param key_flag: 视频表标记
  2837. :param expire_time: 末位视频记录redis过期时间
  2838. :return:
  2839. """
  2840. if key_flag == '30day':
  2841. # 相对30天计算列表的筛选结果
  2842. # 视频列表
  2843. pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_30DAY
  2844. # mid对应上一次视频位置 时间记录
  2845. record_key_prefix = config_.RECORD_KEY_NAME_PREFIX_LAST_VIDEO_30DAY
  2846. # mid对应上一次视频记录
  2847. last_video_key_prefix = config_.LAST_VIDEO_FROM_30DAY_PREFIX
  2848. push_from = config_.PUSH_FROM['rov_recall_30day']
  2849. else:
  2850. return []
  2851. # 获取相关redis key, 用户上一次在rov召回池对应的位置
  2852. pool_key, idx = self.get_last_recommend_video_idx_by_day(record_key_prefix=record_key_prefix,
  2853. pool_key_prefix=pool_key_prefix,
  2854. last_video_key_prefix=last_video_key_prefix,
  2855. expire_time=expire_time)
  2856. if not pool_key:
  2857. return []
  2858. recall_data = []
  2859. pool_recall_result = []
  2860. # 每次获取的视频数
  2861. get_size = size * 5
  2862. # 记录获取频次
  2863. freq = 0
  2864. while len(pool_recall_result) < size:
  2865. freq += 1
  2866. if freq > config_.MAX_FREQ_FROM_ROV_POOL:
  2867. break
  2868. # 获取数据
  2869. data = self.redis_helper.get_data_zset_with_index(key_name=pool_key,
  2870. start=idx, end=idx + get_size - 1,
  2871. with_scores=True)
  2872. if not data:
  2873. break
  2874. recall_data.extend(data)
  2875. # 获取视频id,并转换类型为int,并存储为key-value{videoId: score}
  2876. video_ids = []
  2877. video_score = {}
  2878. for value in data:
  2879. video_id = int(value[0])
  2880. video_ids.append(video_id)
  2881. video_score[video_id] = value[1]
  2882. # 过滤
  2883. filter_ = FilterVideos(request_id=self.request_id,
  2884. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
  2885. ge = gevent.spawn(filter_.filter_videos)
  2886. ge.join()
  2887. filtered_result = ge.get()
  2888. if filtered_result:
  2889. # 添加视频源参数 pushFrom, abCode
  2890. temp_result = [{'videoId': int(item), 'rovScore': video_score[int(item)],
  2891. 'pushFrom': push_from, 'abCode': self.ab_code}
  2892. for item in filtered_result if video_score.get(int(item)) is not None]
  2893. pool_recall_result.extend(temp_result)
  2894. idx += get_size
  2895. pool_recall_result.sort(key=lambda x: x.get('rovScore', 0), reverse=True)
  2896. if len(recall_data) > 0 and len(pool_recall_result) == 0 and self.mid:
  2897. # 召回数据不为空 & 过滤后结果为空 & mid不为空时,更新召回获取的末位视频id记录到定位的key中
  2898. last_video_key = f'{last_video_key_prefix}{self.app_type}:{self.mid}'
  2899. self.redis_helper.set_data_to_redis(key_name=last_video_key, value=recall_data[-1][0],
  2900. expire_time=expire_time)
  2901. # log_.info({
  2902. # 'logTimestamp': int(time.time() * 1000),
  2903. # 'request_id': self.request_id,
  2904. # 'operation': push_from,
  2905. # 'pool_recall_result': pool_recall_result,
  2906. # 'executeTime': (time.time() - start_time) * 1000
  2907. # })
  2908. return pool_recall_result[:size]
  2909. #linfan
  2910. def get_sim_hot_item_reall(self):
  2911. recall_key = "sim_hot_"+str(self.video_id)
  2912. #print("recall_key:", recall_key)
  2913. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  2914. #print(data)
  2915. recall_result = []
  2916. if data is not None:
  2917. json_result =json.loads(data)
  2918. #print("json_result:", json_result)
  2919. for per_item in json_result:
  2920. recall_result.append(
  2921. {'videoId': per_item[0], 'flowPool': '',
  2922. 'rovScore': per_item[1], 'pushFrom': config_.PUSH_FROM['sim_hot_vid_recall'],
  2923. 'abCode': self.ab_code}
  2924. )
  2925. return recall_result[:200]
  2926. def get_sim_hot_item_reall_filter(self):
  2927. if self.video_id is None:
  2928. return []
  2929. recall_key = "sim_hot_" + str(self.video_id)
  2930. #print("recall_key:", recall_key)
  2931. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  2932. # todo zhangbo 获取tags
  2933. video_ids_4tags = []
  2934. if data is not None:
  2935. json_result = json.loads(data)
  2936. for per_item in json_result:
  2937. try:
  2938. vid = int(per_item[0])
  2939. video_ids_4tags.append(vid)
  2940. except Exception as e:
  2941. continue
  2942. video_tag_dict = self.get_video_tags(video_ids_4tags)
  2943. #print(data)
  2944. recall_result = []
  2945. recall_dict = {}
  2946. video_ids = []
  2947. if data is not None:
  2948. json_result = json.loads(data)
  2949. # print("json_result:", json_result)
  2950. for per_item in json_result:
  2951. try:
  2952. vid = int(per_item[0])
  2953. video_ids.append(vid)
  2954. recall_dict[vid] = {'videoId': vid, 'flowPool': '',
  2955. 'rovScore': per_item[1], 'pushFrom': config_.PUSH_FROM['sim_hot_vid_recall'],
  2956. 'abCode': self.ab_code, "tags": video_tag_dict[vid]}
  2957. except Exception as e:
  2958. continue
  2959. if len(video_ids)<=0:
  2960. return recall_result
  2961. recall_num = 20
  2962. #print("recall_num:", recall_num)
  2963. video_ids = video_ids[:recall_num]
  2964. #print(video_ids)
  2965. filter_ = FilterVideos(request_id=self.request_id,
  2966. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids,
  2967. expansion_factor=self.expansion_factor,
  2968. risk_filter_flag=self.risk_filter_flag,
  2969. app_region_filtered=self.app_region_filtered,
  2970. videos_with_risk=self.videos_with_risk
  2971. )
  2972. filtered_viewed_videos = filter_.filter_videos_status(pool_type='normal', region_code=self.get_region_code())
  2973. if filtered_viewed_videos is None:
  2974. return recall_result
  2975. #print("filtered_viewed_videos:", filtered_viewed_videos)
  2976. for vid in filtered_viewed_videos:
  2977. if vid in recall_dict:
  2978. recall_result.append(recall_dict[vid])
  2979. return recall_result
  2980. # get region_hour_recall
  2981. def get_region_hour_recall(self, size=4, region_code='-1'):
  2982. pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H
  2983. recall_key = f"{pool_key_prefix}{region_code}:{self.data_key}:{self.rule_key}"
  2984. #print("recall_key:", recall_key)
  2985. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  2986. #print(data)
  2987. recall_result = []
  2988. if data is not None:
  2989. json_result = json.loads(data)
  2990. #print("json_result:", json_result)
  2991. for per_item in json_result:
  2992. recall_result.append(
  2993. {'videoId': per_item[0], 'flowPool': '',
  2994. 'rovScore': per_item[1], 'pushFrom': config_.PUSH_FROM['rov_recall_region_h'],
  2995. 'abCode': self.ab_code}
  2996. )
  2997. return recall_result[:30]
  2998. # get region_day_recall
  2999. def get_region_day_recall(self, size=4,region_code='-1'):
  3000. """召回池召回视频"""
  3001. pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H
  3002. recall_key = f"{pool_key_prefix}{region_code}:{self.data_key}:{self.rule_key}"
  3003. #print("recall_key:", recall_key)
  3004. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3005. #print(data)
  3006. recall_result = []
  3007. if data is not None:
  3008. json_result = json.loads(data)
  3009. #print("json_result:", json_result)
  3010. for per_item in json_result:
  3011. recall_result.append(
  3012. {'videoId': per_item[0], 'flowPool': '',
  3013. 'rovScore': per_item[1], 'pushFrom': config_.PUSH_FROM['rov_recall_region_24h'],
  3014. 'abCode': self.ab_code}
  3015. )
  3016. return recall_result[:200]
  3017. def get_selected_recall(self, size=4, region_code='-1'):
  3018. """召回池召回视频"""
  3019. pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_24H_H
  3020. recall_key = f"{pool_key_prefix}{region_code}:{self.data_key}:{self.rule_key}"
  3021. #print("recall_key:", recall_key)
  3022. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3023. #print(data)
  3024. recall_result = []
  3025. if data is not None:
  3026. json_result = json.loads(data)
  3027. #print("json_result:", json_result)
  3028. for per_item in json_result:
  3029. recall_result.append(
  3030. {'videoId': per_item[0], 'flowPool': '',
  3031. 'rovScore': per_item[1], 'pushFrom': config_.PUSH_FROM['rov_recall_24h'],
  3032. 'abCode': self.ab_code}
  3033. )
  3034. #print("recall_result:", recall_result)
  3035. return recall_result[:200]
  3036. def get_no_selected_recall(self, size=4, region_code='-1'):
  3037. """未选择召回池召回视频"""
  3038. pool_key_prefix = config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H
  3039. recall_key = f"{pool_key_prefix}{region_code}:{self.data_key}:{self.rule_key}"
  3040. #print("recall_key:", recall_key)
  3041. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3042. #print(data)
  3043. recall_result = []
  3044. if data is not None:
  3045. json_result = json.loads(data)
  3046. #print("json_result:", json_result)
  3047. for per_item in json_result:
  3048. recall_result.append(
  3049. {'videoId': per_item[0], 'flowPool': '',
  3050. 'rovScore': per_item[1], 'pushFrom': config_.PUSH_FROM['rov_recall_24h_dup'],
  3051. 'abCode': self.ab_code}
  3052. )
  3053. return recall_result[:200]
  3054. def get_fast_flow_pool_recall(self, size=4):
  3055. """快速流量池召回视频"""
  3056. recall_key = f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX}{self.app_type}:{config_.QUICK_FLOW_POOL_ID}"
  3057. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3058. recall_result = []
  3059. if data is not None:
  3060. json_result = json.loads(data)
  3061. #print("json_result:", json_result)
  3062. for per_item in json_result:
  3063. recall_result.append(
  3064. {'videoId': per_item[0], 'flowPool': '',
  3065. 'rovScore': per_item[1], 'pushFrom': config_.PUSH_FROM['fast_flow_recall'],
  3066. 'abCode': self.ab_code}
  3067. )
  3068. return recall_result
  3069. def get_flow_pool_recall(self, size=4):
  3070. """流量池召回视频"""
  3071. recall_key = f"{config_.FLOWPOOL_KEY_NAME_PREFIX}{self.app_type}"
  3072. #print("recall_key:", recall_key)
  3073. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3074. #print(data)
  3075. recall_result = []
  3076. if data is not None:
  3077. json_result = json.loads(data)
  3078. #print("json_result:", json_result)
  3079. for per_item in json_result:
  3080. recall_result.append(
  3081. {'videoId': per_item[0], 'flowPool': '',
  3082. 'rovScore': per_item[1], 'pushFrom': config_.PUSH_FROM['normal_flow_recall'],
  3083. 'abCode': self.ab_code}
  3084. )
  3085. return recall_result
  3086. def new_flow_pool_recall(self, size=10, flow_pool_id=None):
  3087. """从流量池中获取视频"""
  3088. start_time = time.time()
  3089. flow_pool_key = self.get_pool_redis_key('flow', flow_pool_id=flow_pool_id)
  3090. flow_pool_recall_result = []
  3091. flow_pool_recall_videos = []
  3092. # 每次获取的视频数
  3093. get_size = size * 5
  3094. # 记录获取频次
  3095. freq = 0
  3096. idx = 0
  3097. while len(flow_pool_recall_result) < size:
  3098. freq += 1
  3099. if freq > config_.MAX_FREQ_FROM_FLOW_POOL:
  3100. break
  3101. # 获取数据
  3102. # st_get = time.time()
  3103. data = self.redis_helper.get_data_zset_with_index(key_name=flow_pool_key,
  3104. start=idx, end=idx + get_size - 1,
  3105. with_scores=True)
  3106. et_get = time.time()
  3107. # log_.info('get data from flow pool redis: freq = {}, data = {}, execute time = {}ms'.format(
  3108. # freq, data, (et_get - st_get) * 1000))
  3109. if not data:
  3110. # log_.info('流量池中的视频已取完')
  3111. break
  3112. # 将video_id 与 flow_pool, score做mapping整理
  3113. video_ids = []
  3114. video_mapping = {}
  3115. video_score = {}
  3116. for value in data:
  3117. try:
  3118. video_id, flow_pool = value[0].split('-')
  3119. except Exception as e:
  3120. log_.error({
  3121. 'request_id': self.request_id,
  3122. 'app_type': self.app_type,
  3123. 'flow_pool_value': value
  3124. })
  3125. continue
  3126. video_id = int(video_id)
  3127. if video_id not in video_ids:
  3128. video_ids.append(video_id)
  3129. video_score[video_id] = value[1]
  3130. if video_id not in video_mapping:
  3131. video_mapping[video_id] = [flow_pool]
  3132. else:
  3133. video_mapping[video_id].append(flow_pool)
  3134. # 过滤
  3135. # filter_ = FilterVideos(request_id=self.request_id,
  3136. # app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
  3137. # filtered_result = filter_.filter_videos(pool_type='flow', region_code=region_code, shield_config=self.shield_config)
  3138. # print("flow filter time:", (time.time()-et_get)*1000)
  3139. # 检查可分发数
  3140. if video_ids and len(video_ids)>0:
  3141. check_result = self.check_video_counts(video_ids=video_ids, flow_pool_mapping=video_mapping)
  3142. for item in check_result:
  3143. video_id = int(item[0])
  3144. flow_pool = item[1]
  3145. if video_id not in flow_pool_recall_videos:
  3146. # 取其中一个 flow_pool 作为召回结果
  3147. # 添加视频源参数 pushFrom, abCode
  3148. flow_pool_recall_result.append(
  3149. {'videoId': video_id, 'flowPool': flow_pool,
  3150. 'rovScore': video_score[video_id], 'pushFrom': config_.PUSH_FROM['flow_recall'],
  3151. 'abCode': self.ab_code}
  3152. )
  3153. flow_pool_recall_videos.append(video_id)
  3154. # et_check = time.time()
  3155. # log_.info('check result: result = {}, execute time = {}ms'.format(
  3156. # check_result, (et_check - st_check) * 1000))
  3157. # # 判断错误标记, True为错误
  3158. # if error_flag:
  3159. # # 结束流量池召回
  3160. # break
  3161. idx += get_size
  3162. # log_.info({
  3163. # 'logTimestamp': int(time.time() * 1000),
  3164. # 'request_id': self.request_id,
  3165. # 'operation': 'flow_pool_recall',
  3166. # 'executeTime': (time.time() - start_time) * 1000
  3167. # })
  3168. return flow_pool_recall_result[:size]
  3169. def get_3days_hot_item_reall(self, exp_config=None):
  3170. recall_key = "hot_3day:"
  3171. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3172. recall_result = []
  3173. recall_dict = {}
  3174. video_ids = []
  3175. if data is not None and data!="":
  3176. try:
  3177. json_result = json.loads(data)
  3178. for per_item in json_result:
  3179. vid = int(per_item[0])
  3180. video_ids.append(vid)
  3181. recall_dict[vid] = {'videoId': vid, 'flowPool': '',
  3182. 'rovScore': per_item[1], 'pushFrom': config_.PUSH_FROM['hot_3_day_recall'],
  3183. 'abCode': self.ab_code}
  3184. except Exception as e:
  3185. return recall_result
  3186. #print("vid len:", len(video_ids))
  3187. if len(video_ids)<=0:
  3188. return recall_result
  3189. recall_num = 20
  3190. try:
  3191. if exp_config and exp_config['recall_get_num']:
  3192. recall_num = int(exp_config['recall_get_num'])
  3193. except:
  3194. recall_num = 20
  3195. video_ids = video_ids[:recall_num]
  3196. #print(video_ids)
  3197. filter_ = FilterVideos(request_id=self.request_id,
  3198. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
  3199. filtered_viewed_videos = filter_.filter_videos_status(pool_type='normal')
  3200. if filtered_viewed_videos is None:
  3201. return recall_result
  3202. #print("filtered_viewed_videos:", filtered_viewed_videos)
  3203. for vid in filtered_viewed_videos:
  3204. if vid in recall_dict:
  3205. recall_result.append(recall_dict[vid])
  3206. #print("hot recall_result:", recall_result)
  3207. #print("recall_dict:", recall_dict)
  3208. return recall_result
  3209. def get_hot_item_reall(self,exp_config=None):
  3210. #recall_key = "hot_video:"
  3211. recall_key = "hot_video:"
  3212. #print("recall_key:", recall_key)
  3213. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3214. recall_result = []
  3215. recall_dict = {}
  3216. video_ids = []
  3217. if data is not None and data!="":
  3218. try:
  3219. json_result = json.loads(data)
  3220. for per_item in json_result:
  3221. vid = int(per_item[0])
  3222. video_ids.append(vid)
  3223. recall_dict[vid] = {'videoId': vid, 'flowPool': '',
  3224. 'rovScore': per_item[1], 'pushFrom': config_.PUSH_FROM['hot_recall'],
  3225. 'abCode': self.ab_code}
  3226. except Exception as e:
  3227. recall_result
  3228. if len(video_ids)<=0:
  3229. return recall_result
  3230. recall_num = 20
  3231. try:
  3232. if exp_config and exp_config['recall_get_num']:
  3233. recall_num = int(exp_config['recall_get_num'])
  3234. except:
  3235. recall_num = 20
  3236. #print("recall_num:", recall_num)
  3237. video_ids = video_ids[:recall_num]
  3238. #print(video_ids)
  3239. filter_ = FilterVideos(request_id=self.request_id,
  3240. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
  3241. filtered_viewed_videos = filter_.filter_videos(pool_type='rov')
  3242. if filtered_viewed_videos is None:
  3243. return recall_result
  3244. #print("filtered_viewed_videos:", filtered_viewed_videos)
  3245. for vid in filtered_viewed_videos:
  3246. if vid in recall_dict:
  3247. recall_result.append(recall_dict[vid])
  3248. #print("hot recall_result:", recall_result)
  3249. return recall_result
  3250. def get_title_recall(self):
  3251. if self.video_id is None:
  3252. return []
  3253. recall_key = "title_I2I:" + str(self.video_id)
  3254. # print("recall_key:", recall_key)
  3255. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3256. # print(data)
  3257. recall_result = []
  3258. recall_dict = {}
  3259. video_ids = []
  3260. if data is not None and data!="":
  3261. try:
  3262. json_result = json.loads(data)
  3263. for per_item in json_result:
  3264. vid = int(per_item[0])
  3265. video_ids.append(vid)
  3266. recall_dict[vid] = {'videoId': vid, 'flowPool': '',
  3267. 'rovScore': per_item[1], 'pushFrom': config_.PUSH_FROM['title_i2i_recall'],
  3268. 'abCode': self.ab_code}
  3269. except Exception as e:
  3270. return recall_result
  3271. if len(video_ids) <= 0:
  3272. return recall_result
  3273. video_ids = video_ids[:50]
  3274. # print(video_ids)
  3275. filter_ = FilterVideos(request_id=self.request_id,
  3276. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
  3277. filtered_viewed_videos = filter_.filter_videos(pool_type='normal')
  3278. if filtered_viewed_videos is None:
  3279. return recall_result
  3280. # print("filtered_viewed_videos:", filtered_viewed_videos)
  3281. for vid in filtered_viewed_videos:
  3282. if vid in recall_dict:
  3283. recall_result.append(recall_dict[vid])
  3284. return recall_result
  3285. def get_word2vec_item_reall(self, exp_config=None):
  3286. if self.video_id is None:
  3287. return []
  3288. recall_key = "w2v:" + str(self.video_id)
  3289. #print("recall_key:", recall_key)
  3290. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3291. #print(data)
  3292. recall_result = []
  3293. recall_dict = {}
  3294. video_ids = []
  3295. if data is not None and data.strip()!="":
  3296. try:
  3297. json_result = data.strip().split(",")
  3298. for per_item in json_result:
  3299. vid = int(per_item)
  3300. video_ids.append(vid)
  3301. recall_dict[vid] = {'videoId': vid, 'flowPool': '',
  3302. 'rovScore': 0.0, 'pushFrom': config_.PUSH_FROM['w2v_recall'],
  3303. 'abCode': self.ab_code}
  3304. except Exception as e:
  3305. return recall_result
  3306. if len(video_ids)<=0:
  3307. return recall_result
  3308. recall_num = 20
  3309. try:
  3310. if exp_config and exp_config['recall_get_num']:
  3311. recall_num = int(exp_config['recall_get_num'])
  3312. except:
  3313. recall_num = 20
  3314. #print("recall_num:", recall_num)
  3315. video_ids = video_ids[:recall_num]
  3316. #print(video_ids)
  3317. filter_ = FilterVideos(request_id=self.request_id,
  3318. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
  3319. filtered_viewed_videos = filter_.filter_videos_status(pool_type='normal')
  3320. if filtered_viewed_videos is None:
  3321. return recall_result
  3322. #print("filtered_viewed_videos:", filtered_viewed_videos)
  3323. for vid in filtered_viewed_videos:
  3324. if vid in recall_dict:
  3325. recall_result.append(recall_dict[vid])
  3326. return recall_result[:30]
  3327. def get_test_config(self):
  3328. recall_key = "test_exp_config_pos"
  3329. # print("recall_key:", recall_key)
  3330. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3331. if data is not None and data!="":
  3332. try:
  3333. #print(data)
  3334. json_result = json.loads(data)
  3335. #print(json_result)
  3336. return json_result
  3337. except Exception as e:
  3338. return None
  3339. else:
  3340. return None
  3341. def get_w2v_config(self):
  3342. recall_key = "w2v_exp_config_pos_range"
  3343. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3344. if data is not None and data!="":
  3345. try:
  3346. #print(data)
  3347. json_result = json.loads(data)
  3348. #print(json_result)
  3349. return json_result
  3350. except Exception as e:
  3351. return None
  3352. else:
  3353. return None
  3354. def get_simrecall_config(self):
  3355. recall_key = "simrecall_exp_config_pos"
  3356. # print("recall_key:", recall_key)
  3357. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3358. if data is not None:
  3359. try:
  3360. # print(data)
  3361. json_result = json.loads(data)
  3362. # print(json_result)
  3363. return json_result
  3364. except Exception as e:
  3365. return None
  3366. else:
  3367. return None
  3368. def get_simrecall_config_new(self):
  3369. recall_key = "simrecall_exp_config_range"
  3370. #print("recall_key:", recall_key)
  3371. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3372. if data is not None and data!="":
  3373. try:
  3374. # print(data)
  3375. json_result = json.loads(data)
  3376. # print(json_result)
  3377. return json_result
  3378. except Exception as e:
  3379. return None
  3380. else:
  3381. return None
  3382. def get_hotrecall_config(self):
  3383. recall_key = "ht_exp_config"
  3384. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3385. if data is not None:
  3386. try:
  3387. # print(data)
  3388. json_result = json.loads(data)
  3389. # print(json_result)
  3390. return json_result
  3391. except Exception as e:
  3392. return None
  3393. else:
  3394. return None
  3395. def get_U2I_config(self):
  3396. recall_key = "u2i_exp_config_pos_range_new"
  3397. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3398. if data is not None and data!="":
  3399. try:
  3400. # print(data)
  3401. json_result = json.loads(data)
  3402. # print(json_result)
  3403. return json_result
  3404. except Exception as e:
  3405. return None
  3406. else:
  3407. return None
  3408. def get_u2u2i_config(self):
  3409. recall_key = "u2u2i_exp_config_range"
  3410. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3411. if data is not None:
  3412. try:
  3413. # print(data)
  3414. json_result = json.loads(data)
  3415. # print(json_result)
  3416. return json_result
  3417. except Exception as e:
  3418. return None
  3419. else:
  3420. return None
  3421. def get_hotrecall_flow_config(self):
  3422. recall_key = "ht_flow_config:"
  3423. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3424. return data
  3425. def get_w2v_flow_config(self):
  3426. recall_key = "w2v_flow_config:"
  3427. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3428. return data
  3429. def get_flow_config(self):
  3430. recall_key = "flow_config:"
  3431. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3432. return data
  3433. def get_simrecall_flow_config(self):
  3434. recall_key = "simrecall_flow_config:"
  3435. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3436. return data
  3437. def get_flow_exp_7_config(self):
  3438. recall_key = "exp7_exp_config"
  3439. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3440. if data is not None:
  3441. try:
  3442. # print(data)
  3443. json_result = json.loads(data)
  3444. # print(json_result)
  3445. return json_result
  3446. except Exception as e:
  3447. return None
  3448. else:
  3449. return None
  3450. def get_flow_exp_8_config(self):
  3451. recall_key = "exp8_exp_config"
  3452. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3453. if data is not None:
  3454. try:
  3455. # print(data)
  3456. json_result = json.loads(data)
  3457. # print(json_result)
  3458. return json_result
  3459. except Exception as e:
  3460. return None
  3461. else:
  3462. return None
  3463. def get_sort_ab_codel_config(self):
  3464. ab_key = "sort_ab_config2"
  3465. data = self.redis_helper.get_data_from_redis(key_name=ab_key)
  3466. if data is not None:
  3467. try:
  3468. # print(data)
  3469. json_result = json.loads(data)
  3470. # print(json_result)
  3471. return json_result
  3472. except Exception as e:
  3473. return None
  3474. else:
  3475. return None
  3476. def get_U2I_reall(self, mid):
  3477. #recall_key = "hot_video:"
  3478. if not mid:
  3479. return []
  3480. recall_key = "u2i:"+mid
  3481. #print("recall_key:", recall_key)
  3482. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3483. #print(data)
  3484. recall_result = []
  3485. recall_dict = {}
  3486. video_ids = []
  3487. if data is not None and data!="":
  3488. try:
  3489. json_result = json.loads(data)
  3490. for per_item in json_result:
  3491. vid = int(per_item[0])
  3492. video_ids.append(vid)
  3493. recall_dict[vid] = {'videoId': vid, 'flowPool': '',
  3494. 'rovScore': float(per_item[1]), 'pushFrom': config_.PUSH_FROM['u2i_tag_recall'],
  3495. 'abCode': self.ab_code}
  3496. except Exception as e:
  3497. return recall_result
  3498. if len(video_ids)<=0:
  3499. return recall_result
  3500. recall_num = 20
  3501. #print("recall_num:", recall_num)
  3502. video_ids = video_ids[:recall_num]
  3503. #print(video_ids)
  3504. filter_ = FilterVideos(request_id=self.request_id,
  3505. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids,
  3506. expansion_factor=self.expansion_factor,
  3507. risk_filter_flag=self.risk_filter_flag,
  3508. app_region_filtered=self.app_region_filtered,
  3509. videos_with_risk=self.videos_with_risk
  3510. )
  3511. filtered_viewed_videos = filter_.filter_videos_status(pool_type='normal', region_code=self.get_region_code())
  3512. if filtered_viewed_videos is None:
  3513. return recall_result
  3514. #print("filtered_viewed_videos:", filtered_viewed_videos)
  3515. for vid in filtered_viewed_videos:
  3516. if vid in recall_dict:
  3517. recall_result.append(recall_dict[vid])
  3518. #print("u2i recall_result:", recall_result)
  3519. return recall_result
  3520. def get_U2U2I_reall(self, mid, exp_config=None):
  3521. #recall_key = "hot_video:"
  3522. if not mid:
  3523. return []
  3524. recall_key = "u2u2i:"+mid
  3525. #print("recall_key:", recall_key)
  3526. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3527. #print(data)
  3528. recall_result = []
  3529. recall_dict = {}
  3530. video_ids = []
  3531. if data is not None:
  3532. json_result = json.loads(data)
  3533. #print("json_result:", json_result)
  3534. for per_item in json_result:
  3535. try:
  3536. vid = int(per_item[0])
  3537. video_ids.append(vid)
  3538. recall_dict[vid] = {'videoId': vid, 'flowPool': '',
  3539. 'rovScore': float(per_item[1]), 'pushFrom': config_.PUSH_FROM['u2u2i_recall'],
  3540. 'abCode': self.ab_code}
  3541. except Exception as e:
  3542. continue
  3543. if len(video_ids)<=0:
  3544. return recall_result
  3545. recall_num = 20
  3546. try:
  3547. if exp_config and exp_config['recall_get_num']:
  3548. recall_num = int(exp_config['recall_get_num'])
  3549. except:
  3550. recall_num = 20
  3551. #print("recall_num:", recall_num)
  3552. video_ids = video_ids[:recall_num]
  3553. #print(video_ids)
  3554. filter_ = FilterVideos(request_id=self.request_id,
  3555. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
  3556. filtered_viewed_videos = filter_.filter_videos_status(pool_type='normal')
  3557. if filtered_viewed_videos is None:
  3558. return recall_result
  3559. #print("filtered_viewed_videos:", filtered_viewed_videos)
  3560. for vid in filtered_viewed_videos:
  3561. if vid in recall_dict:
  3562. recall_result.append(recall_dict[vid])
  3563. #print("u2i recall_result:", recall_result)
  3564. return recall_result
  3565. def get_video_recall_config(self):
  3566. recall_key = "vr_exp_pos_config_range"
  3567. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3568. if data is not None and data!="":
  3569. try:
  3570. # print(data)
  3571. json_result = json.loads(data)
  3572. # print(json_result)
  3573. return json_result
  3574. except Exception as e:
  3575. return None
  3576. else:
  3577. return None
  3578. def get_return_video_reall(self, pre_key=None):
  3579. if self.video_id is None:
  3580. return []
  3581. recall_key = "rv:"+ str(self.video_id)
  3582. if pre_key :
  3583. recall_key = pre_key + str(self.video_id)
  3584. #print("recall_key:", recall_key)
  3585. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3586. # todo zhangbo 获取tags
  3587. video_ids_4tag = []
  3588. if data is not None and data!="" :
  3589. try:
  3590. json_result = json.loads(data)
  3591. for per_item in json_result:
  3592. vid = int(per_item[0])
  3593. video_ids_4tag.append(vid)
  3594. except Exception as e:
  3595. video_ids_4tag = []
  3596. video_tag_dict = self.get_video_tags(video_ids_4tag)
  3597. #print(data)
  3598. recall_result = []
  3599. recall_dict = {}
  3600. video_ids = []
  3601. if data is not None and data!="" :
  3602. try:
  3603. json_result = json.loads(data)
  3604. for per_item in json_result:
  3605. vid = int(per_item[0])
  3606. video_ids.append(vid)
  3607. recall_dict[vid] = {'videoId': vid, 'flowPool': '',
  3608. 'rovScore': float(per_item[1]), 'pushFrom': config_.PUSH_FROM['return_video_recall'],
  3609. 'abCode': self.ab_code, "tags": video_tag_dict[vid]}
  3610. except Exception as e:
  3611. return recall_result
  3612. if len(video_ids)<=0:
  3613. return recall_result
  3614. recall_num = 20
  3615. #print("recall_num:", recall_num)
  3616. video_ids = video_ids[:recall_num]
  3617. #print(video_ids)
  3618. filter_ = FilterVideos(request_id=self.request_id,
  3619. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids,
  3620. expansion_factor=self.expansion_factor,
  3621. risk_filter_flag=self.risk_filter_flag,
  3622. app_region_filtered=self.app_region_filtered,
  3623. videos_with_risk=self.videos_with_risk
  3624. )
  3625. filtered_viewed_videos = filter_.filter_videos_status(pool_type='normal', region_code=self.get_region_code())
  3626. if filtered_viewed_videos is None:
  3627. return recall_result
  3628. #print("filtered_viewed_videos:", filtered_viewed_videos)
  3629. for vid in filtered_viewed_videos:
  3630. if vid in recall_dict:
  3631. recall_result.append(recall_dict[vid])
  3632. #print("u2i recall_result:", recall_result)
  3633. return recall_result
  3634. def get_play_reall(self, mid, exp_config=None):
  3635. #recall_key = "hot_video:"
  3636. if not mid:
  3637. return []
  3638. recall_key = "u2i_play:"+mid
  3639. #print("recall_key:", recall_key)
  3640. data = self.redis_helper.get_data_from_redis(key_name=recall_key)
  3641. #print(data)
  3642. recall_result = []
  3643. recall_dict = {}
  3644. video_ids = []
  3645. if data is not None and data!="":
  3646. try:
  3647. json_result = json.loads(data)
  3648. for per_item in json_result:
  3649. vid = int(per_item[0])
  3650. video_ids.append(vid)
  3651. recall_dict[vid] = {'videoId': vid, 'flowPool': '',
  3652. 'rovScore': float(per_item[1]), 'pushFrom': config_.PUSH_FROM['u2i_tag_play_recall'],
  3653. 'abCode': self.ab_code}
  3654. except Exception as e:
  3655. return recall_result
  3656. if len(video_ids)<=0:
  3657. return recall_result
  3658. recall_num = 20
  3659. try:
  3660. if exp_config and exp_config['recall_get_num']:
  3661. recall_num = int(exp_config['recall_get_num'])
  3662. except:
  3663. recall_num = 20
  3664. #print("recall_num:", recall_num)
  3665. video_ids = video_ids[:recall_num]
  3666. #print(video_ids)
  3667. filter_ = FilterVideos(request_id=self.request_id,
  3668. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=video_ids)
  3669. filtered_viewed_videos = filter_.filter_videos_status(pool_type='normal')
  3670. if filtered_viewed_videos is None:
  3671. return recall_result
  3672. #print("filtered_viewed_videos:", filtered_viewed_videos)
  3673. for vid in filtered_viewed_videos:
  3674. if vid in recall_dict:
  3675. recall_result.append(recall_dict[vid])
  3676. #print("u2i recall_result:", recall_result)
  3677. return recall_result
  3678. def get_region_code(self):
  3679. # 获取存在城市分组数据的城市编码列表
  3680. city_code_list = [code for _, code in config_.CITY_CODE.items()]
  3681. # 获取provinceCode
  3682. province_code = self.client_info.get('provinceCode', '-1')
  3683. # 获取cityCode
  3684. city_code = self.client_info.get('cityCode', '-1')
  3685. if city_code in city_code_list:
  3686. # 分城市数据存在时,获取城市分组数据
  3687. region_code = city_code
  3688. else:
  3689. region_code = province_code
  3690. if region_code == '':
  3691. region_code = '-1'
  3692. return region_code
  3693. def recall_strategy_trend_v1(self):
  3694. try:
  3695. # 1 获取trigger信息
  3696. region_code_province = self.client_info.get('provinceCode', '-1')
  3697. # 2 拼接redis key
  3698. key1 = "alg_recsys_recall_4h_region_trend_sum_" + region_code_province
  3699. key2 = "alg_recsys_recall_4h_region_trend_avg_" + region_code_province
  3700. # 3 取数据
  3701. data1 = self.redis_helper.get_data_from_redis(key_name=key1)
  3702. data2 = self.redis_helper.get_data_from_redis(key_name=key2)
  3703. data_for_filter = []
  3704. group_size = 20
  3705. if data1 is not None and not "" == data1:
  3706. # todo 类型转换没做兜底
  3707. data1_list = [int(i) for i in data1.split(",")]
  3708. data_for_filter.extend([data1_list[i:i + group_size] for i in range(0, len(data1_list), group_size)])
  3709. else:
  3710. data1_list = []
  3711. if data2 is not None and not "" == data2:
  3712. # todo 类型转换没做兜底
  3713. data2_list = [int(i) for i in data2.split(",")]
  3714. data2_list = [i for i in data2_list if i not in data1_list]
  3715. if len(data2_list) > 0:
  3716. data_for_filter.extend(
  3717. [data2_list[i:i + group_size] for i in range(0, len(data2_list), group_size)])
  3718. else:
  3719. data2_list = []
  3720. data_for_filter = [i for i in data_for_filter if len(i) > 0]
  3721. data_for_filter.append([17736990, 17734880, 17734759, 17726977])
  3722. data1_list.extend([17736990, 17734880, 17734759, 17726977])
  3723. # 3.9 获取item的tag特征
  3724. video_tag_dict = self.get_video_tags(list(set(data1_list) | set(data2_list)))
  3725. # 4 视频过滤
  3726. filter_ = FilterVideos(request_id=self.request_id,
  3727. app_type=self.app_type, mid=self.mid, uid=self.uid, video_ids=None,
  3728. expansion_factor=self.expansion_factor,
  3729. risk_filter_flag=self.risk_filter_flag,
  3730. app_region_filtered=self.app_region_filtered,
  3731. videos_with_risk=self.videos_with_risk
  3732. )
  3733. region_code = self.get_region_code()
  3734. t = [gevent.spawn(filter_.filter_videos_for_group, region_code, videos, video_tag_dict,
  3735. self.tags_filter_rule, self.tags_filter_flag) for videos in data_for_filter]
  3736. gevent.joinall(t)
  3737. result_list = [i.get() for i in t if i.get() is not None and len(i.get()) > 0]
  3738. # 5 返回结果
  3739. results = []
  3740. for g in result_list:
  3741. for v in g:
  3742. results.append({
  3743. 'tags': video_tag_dict[v],
  3744. 'videoId': v, 'flowPool': '',
  3745. 'rovScore': 0.0, 'pushFrom': config_.PUSH_FROM['recall_strategy_trend_v1'],
  3746. 'abCode': self.ab_code
  3747. })
  3748. # print("results-trend:" + str(results))
  3749. return results
  3750. except Exception as e:
  3751. log_.error("error in recall_strategy_trend_v1:{}".format(e))
  3752. return []
  3753. def get_video_tags(self, video_ids) -> dict:
  3754. REDIS_PREFIX = "alg_recsys_video_tags_"
  3755. redis_keys = [REDIS_PREFIX + str(i) for i in video_ids]
  3756. video_tags = self.redis_helper.get_batch_key(redis_keys)
  3757. video_tag_dict = {}
  3758. for i, tags_str in enumerate(video_tags):
  3759. tags = []
  3760. if tags_str is not None and len(tags_str) != 0:
  3761. tags = str(tags_str).split(",")
  3762. video_tag_dict[video_ids[i]] = tags
  3763. return video_tag_dict