video_recall.py 184 KB

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