flowpool_data_update_with_level_tomson.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327
  1. import datetime
  2. import random
  3. import time
  4. import os
  5. import traceback
  6. import random
  7. import json
  8. from config import set_config
  9. from utils import request_post, filter_video_status, send_msg_to_feishu, filter_video_status_app, \
  10. filter_political_videos
  11. from log import Log
  12. from db_helper import RedisHelper
  13. from odps import ODPS
  14. config_, _ = set_config()
  15. log_ = Log()
  16. def get_videos_from_flow_pool(app_type, size=1000):
  17. """
  18. 从流量池获取视频,循环获取,直到返回数据为None结束
  19. :param app_type: 产品标识 type-int
  20. :param size: 每次获取视频数量,type-int,默认1000
  21. :return: videos [{'videoId': 1111, 'flowPool': ''}, ...]
  22. """
  23. # 获取批次标识,利用首次获取数据时间戳为标记
  24. batch_flag = int(time.time())
  25. request_data = {'appType': app_type, 'batchFlag': batch_flag, 'size': size}
  26. videos = []
  27. retry = 0
  28. print("zhangbo1")
  29. while True:
  30. print("zhangbo2")
  31. print(config_.GET_VIDEOS_FROM_POOL_URL)
  32. result = request_post(request_url=config_.GET_VIDEOS_FROM_POOL_URL, request_data=request_data)
  33. if result is None:
  34. if retry > 2:
  35. break
  36. retry += 1
  37. continue
  38. if result['code'] != 0:
  39. log_.info('batch_flag: {}, 获取流量池视频失败'.format(batch_flag))
  40. if retry > 2:
  41. break
  42. retry += 1
  43. continue
  44. if not result['data']:
  45. if retry > 2:
  46. break
  47. retry += 1
  48. continue
  49. videos.extend(result['data'])
  50. return videos
  51. def get_videos_remain_view_count(video_info_list):
  52. """
  53. 获取视频在流量池中的剩余可分发数,并存入对应的redis中
  54. :param app_type: 产品标识 type-int
  55. :param video_info_list: 视频信息 (视频id, 流量池标记) type-list,[(video_id, flow_pool), ...]
  56. :return: data type-list,[(video_id, flow_pool, view_count), ...]
  57. """
  58. redis_helper = RedisHelper()
  59. if not video_info_list:
  60. return
  61. # 每次请求10个
  62. for i in range(len(video_info_list)//10 + 1):
  63. remain_st_time = time.time()
  64. videos = [{'videoId': info[0], 'flowPool': info[1]} for info in video_info_list[i*10:(i+1)*10]]
  65. request_data = {'videos': videos}
  66. result = request_post(request_url=config_.GET_REMAIN_VIEW_COUNT_URL,
  67. request_data=request_data, timeout=(0.5, 3))
  68. log_.info(f"i = {i}, expend time = {(time.time()-remain_st_time)*1000}")
  69. if result is None:
  70. continue
  71. if result['code'] != 0:
  72. log_.error('获取视频在流量池中的剩余可分发数失败')
  73. continue
  74. for item in result['data']:
  75. if item['distributeCount'] is None:
  76. continue
  77. distribute_count = int(item['distributeCount'])
  78. if distribute_count > 0:
  79. # 将分发数更新到本地记录
  80. key_name = f"{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{item['videoId']}:{item['flowPool']}"
  81. redis_helper.set_data_to_redis(key_name=key_name, value=distribute_count, expire_time=15 * 60)
  82. else:
  83. # 将本地记录删除
  84. key_name = f"{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{item['videoId']}:{item['flowPool']}"
  85. redis_helper.del_keys(key_name=key_name)
  86. # 从流量召回池移除
  87. value = '{}-{}'.format(item['videoId'], item['flowPool'])
  88. for type_name in config_.APP_TYPE:
  89. for level in range(1, 7):
  90. flow_pool_key = \
  91. f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL}{config_.APP_TYPE.get(type_name)}:{level}"
  92. redis_helper.remove_value_from_set(key_name=flow_pool_key, values=(value, ))
  93. quick_flow_pool_key = f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX_SET}{config_.APP_TYPE.get(type_name)}" \
  94. f":{config_.QUICK_FLOW_POOL_ID}"
  95. redis_helper.remove_value_from_set(key_name=quick_flow_pool_key, values=(value, ))
  96. def get_flow_pool_recommend_config(flow_pool_id):
  97. """获取流量池推荐分发配置"""
  98. result = request_post(request_url=config_.GET_FLOW_POOL_RECOMMEND_CONFIG_URL)
  99. if result is None:
  100. return None
  101. if result['code'] != 0:
  102. return None
  103. flow_pool_distribute_config = result['data'].get('flowPoolDistributeConfig')
  104. if flow_pool_distribute_config:
  105. if int(eval(flow_pool_distribute_config).get('flowPoolId')) == flow_pool_id:
  106. return eval(eval(flow_pool_distribute_config).get('distributeRate'))
  107. else:
  108. return None
  109. else:
  110. return None
  111. def online_flow_pool_data_to_redis(app_type, video_ids_set, video_info_data):
  112. redis_helper = RedisHelper()
  113. for tag, values in video_ids_set.items():
  114. if tag == 'quick_flow_pool':
  115. key_name_video_ids = \
  116. f"{config_.QUICK_FLOWPOOL_VIDEO_ID_KEY_NAME_PREFIX}{app_type}:{config_.QUICK_FLOW_POOL_ID}"
  117. key_prefix_video_info = \
  118. f"{config_.QUICK_FLOWPOOL_VIDEO_INFO_KEY_NAME_PREFIX}{app_type}:{config_.QUICK_FLOW_POOL_ID}"
  119. else:
  120. key_name_video_ids = f"{config_.FLOWPOOL_VIDEO_ID_KEY_NAME_PREFIX}{app_type}"
  121. key_prefix_video_info = f"{config_.FLOWPOOL_VIDEO_INFO_KEY_NAME_PREFIX}{app_type}"
  122. # 如果key已存在,删除key
  123. if redis_helper.key_exists(key_name=key_name_video_ids):
  124. redis_helper.del_keys(key_name=key_name_video_ids)
  125. # 写入redis
  126. if len(values) > 0:
  127. redis_helper.add_data_with_set(key_name=key_name_video_ids, values=values, expire_time=3600)
  128. info_values = video_info_data[tag]
  129. if len(info_values) > 0:
  130. for video_id, info_value in info_values.items():
  131. key_name_video_info = f"{key_prefix_video_info}:{video_id}"
  132. # 如果key已存在,删除key
  133. if redis_helper.key_exists(key_name=key_name_video_info):
  134. redis_helper.del_keys(key_name=key_name_video_info)
  135. if len(info_value) > 0:
  136. redis_helper.add_data_with_set(key_name=key_name_video_info, values=info_value,
  137. expire_time=3600)
  138. def get_flow_pool_data(app_type, video_info_list, flow_pool_id_list):
  139. """
  140. 获取流量池可分发视频,并将结果上传Redis
  141. :param app_type: 产品标识 type-int
  142. :return: None
  143. """
  144. try:
  145. # 从流量池获取数据
  146. videos = get_videos_from_flow_pool(app_type=app_type)
  147. log_.info(f"app_type: {app_type}")
  148. # log_.info(f"videos: {videos}")
  149. if len(videos) <= 0:
  150. log_.info('流量池中无需分发的视频')
  151. return video_info_list
  152. # video_id 与 flow_pool, level 进行mapping
  153. video_ids = set()
  154. log_.info('流量池中视频数:{}'.format(len(videos)))
  155. mapping = {}
  156. for video in videos:
  157. flow_pool_id = video['flowPoolId'] # 召回使用的切分ID是在这里做的。流量池中的视频是不区分ID的,也不区分层。
  158. if int(flow_pool_id) not in flow_pool_id_list:
  159. continue
  160. # print(f"flow_pool_id: {flow_pool_id}")
  161. video_id = video['videoId']
  162. video_ids.add(video_id)
  163. item_info = {'flowPool': video['flowPool'], 'level': video['level']}
  164. if video_id in mapping:
  165. mapping[video_id].append(item_info)
  166. else:
  167. mapping[video_id] = [item_info]
  168. log_.info(f"需更新流量池视频数: {len(video_ids)}")
  169. # 对视频状态进行过滤
  170. if app_type == config_.APP_TYPE['APP']:
  171. filtered_videos = filter_video_status_app(list(video_ids))
  172. else:
  173. filtered_videos = filter_video_status(list(video_ids))
  174. log_.info('filter videos status finished, filtered_videos nums={}'.format(len(filtered_videos)))
  175. # 涉政视频过滤
  176. if app_type not in config_.POLITICAL_RECOMMEND_APP_TYPE_LIST:
  177. filtered_videos = filter_political_videos(video_ids=filtered_videos)
  178. if not filtered_videos:
  179. log_.info('流量池中视频状态不符合分发')
  180. return video_info_list
  181. # 上传数据到redis
  182. quick_flow_pool_redis_data = set()
  183. # 普通流量池视频按照层级存储
  184. redis_data = dict()
  185. for video_id in filtered_videos:
  186. for item in mapping.get(video_id):
  187. flow_pool = item['flowPool']
  188. level = item['level']
  189. # 判断是否为快速曝光流量池视频
  190. value = '{}-{}'.format(video_id, flow_pool)
  191. flow_pool_id = int(flow_pool.split('#')[0]) # flowPool: 流量池ID#分级ID#级别Level#生命周期ID
  192. if flow_pool_id == config_.QUICK_FLOW_POOL_ID:
  193. quick_flow_pool_redis_data.add(value)
  194. else:
  195. if level not in redis_data:
  196. redis_data[level] = set()
  197. redis_data[level].add(value)
  198. video_info = (video_id, flow_pool)
  199. if video_info not in video_info_list:
  200. video_info_list.append(video_info)
  201. # 快速曝光流量池视频写入redis
  202. redis_helper = RedisHelper()
  203. quick_flow_pool_key_name = f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX_SET}{app_type}:{config_.QUICK_FLOW_POOL_ID}"
  204. # 如果key已存在,删除key
  205. if redis_helper.key_exists(quick_flow_pool_key_name):
  206. redis_helper.del_keys(quick_flow_pool_key_name)
  207. # 写入redis
  208. if quick_flow_pool_redis_data:
  209. log_.info(f"quick_flow_pool_redis_data = {quick_flow_pool_redis_data}")
  210. redis_helper.add_data_with_set(key_name=quick_flow_pool_key_name, values=quick_flow_pool_redis_data,
  211. expire_time=24 * 3600)
  212. # 快速流量池分发概率存入redis
  213. distribute_rate_key_name = f"{config_.QUICK_FLOWPOOL_DISTRIBUTE_RATE_KEY_NAME_PREFIX}{config_.QUICK_FLOW_POOL_ID}"
  214. distribute_rate = get_flow_pool_recommend_config(flow_pool_id=config_.QUICK_FLOW_POOL_ID)
  215. if distribute_rate is not None:
  216. redis_helper.set_data_to_redis(key_name=distribute_rate_key_name, value=distribute_rate,
  217. expire_time=15 * 60)
  218. # 普通流量池视频写入redis - 分层存储
  219. level_list = []
  220. for level, videos in redis_data.items():
  221. log_.info(f"level: {level}, videos_count: {len(videos)}")
  222. level_list.append(level)
  223. flow_pool_key_name = f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL}{app_type}:{level}"
  224. # 如果key已存在,删除key
  225. if redis_helper.key_exists(flow_pool_key_name):
  226. redis_helper.del_keys(flow_pool_key_name)
  227. # 写入redis
  228. if videos:
  229. redis_helper.add_data_with_set(key_name=flow_pool_key_name, values=videos, expire_time=24 * 3600)
  230. # 删除此时不存在的level key
  231. for i in range(1, 7):
  232. if i not in level_list:
  233. flow_pool_key_name = f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL}{app_type}:{i}"
  234. # 如果key已存在,删除key
  235. if redis_helper.key_exists(flow_pool_key_name):
  236. redis_helper.del_keys(flow_pool_key_name)
  237. log_.info('data to redis finished!')
  238. return video_info_list
  239. except Exception as e:
  240. log_.error('流量池更新失败, appType: {} exception: {}, traceback: {}'.format(
  241. app_type, e, traceback.format_exc()))
  242. send_msg_to_feishu(
  243. webhook=config_.FEISHU_ROBOT['server_robot'].get('webhook'),
  244. key_word=config_.FEISHU_ROBOT['server_robot'].get('key_word'),
  245. msg_text='rov-offline{} - 流量池更新失败, appType: {}, exception: {}'.format(config_.ENV_TEXT, app_type, e)
  246. )
  247. return video_info_list
  248. def get_data_from_odps(project, sql):
  249. """检查数据是否准备好"""
  250. odps = ODPS(
  251. access_id=config_.ODPS_CONFIG['ACCESSID'],
  252. secret_access_key=config_.ODPS_CONFIG['ACCESSKEY'],
  253. project=project,
  254. endpoint=config_.ODPS_CONFIG['ENDPOINT'],
  255. connect_timeout=3000,
  256. read_timeout=500000,
  257. pool_maxsize=1000,
  258. pool_connections=1000
  259. )
  260. try:
  261. with odps.execute_sql(sql=sql).open_reader() as reader:
  262. data_df = reader.to_pandas()
  263. except Exception as e:
  264. data_df = None
  265. return data_df
  266. if __name__ == '__main__':
  267. st_time = time.time()
  268. # 为避免第一个app_type获取数据不全,等待1min
  269. time.sleep(60)
  270. app_type_list = [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]
  271. log_.info('flow pool predict start...')
  272. # 获取对应流量池id列表
  273. redis_helper = RedisHelper()
  274. flow_pool_abtest_config = redis_helper.get_data_from_redis(key_name=config_.FLOWPOOL_ABTEST_KEY_NAME)
  275. if flow_pool_abtest_config is not None:
  276. flow_pool_abtest_config = json.loads(flow_pool_abtest_config)
  277. else:
  278. flow_pool_abtest_config = {}
  279. flow_pool_id_list = flow_pool_abtest_config.get('control_group', [])
  280. video_info_list = []
  281. for app_name, app_type in config_.APP_TYPE.items():
  282. log_.info('{} predict start...'.format(app_name))
  283. if app_type == config_.APP_TYPE['LAO_HAO_KAN_VIDEO']:
  284. continue
  285. elif app_type == config_.APP_TYPE['ZUI_JING_QI']:
  286. continue
  287. else:
  288. video_info_list = get_flow_pool_data(app_type=app_type, video_info_list=video_info_list,
  289. flow_pool_id_list=flow_pool_id_list)
  290. log_.info('{} predict end...'.format(app_name))
  291. # 更新剩余分发数
  292. log_.info(f"video_info_list count = {len(video_info_list)}")
  293. get_videos_remain_view_count(video_info_list)
  294. log_.info('flow pool predict end...')
  295. log_.info(f"expend time = {(time.time() - st_time) * 1000}ms")