utils.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348
  1. import traceback
  2. import requests
  3. import json
  4. import time
  5. from datetime import datetime
  6. # from db_helper import HologresHelper, RedisHelper, MysqlHelper
  7. from db_helper import RedisHelper, MysqlHelper
  8. from config import set_config
  9. from log import Log
  10. config_ = set_config()
  11. log_ = Log()
  12. def send_msg_to_feishu(msg_text):
  13. """发送消息到飞书"""
  14. # webhook地址
  15. webhook = 'https://open.feishu.cn/open-apis/bot/v2/hook/8de4de35-30ed-4692-8854-7a154e89b2f2'
  16. # 自定义关键词key_word
  17. key_word = '服务报警'
  18. headers = {'Content-Type': 'application/json'}
  19. payload_message = {
  20. "msg_type": "text",
  21. "content": {
  22. "text": '{}: {}'.format(key_word, msg_text)
  23. }
  24. }
  25. response = requests.request('POST', url=webhook, headers=headers, data=json.dumps(payload_message))
  26. print(response.text)
  27. def request_post(request_url, request_data, timeout):
  28. """
  29. post 请求 HTTP接口
  30. :param request_url: 接口URL
  31. :param request_data: 请求参数
  32. :param timeout: 超时时间,单位为秒,type-float or tuple(connect_timeout, read_timeout)
  33. :return: res_data json格式
  34. """
  35. try:
  36. headers = {"Connection": "close"}
  37. response = requests.post(url=request_url, json=request_data, timeout=timeout, headers=headers)
  38. if response.status_code == 200:
  39. res_data = json.loads(response.text)
  40. return res_data
  41. else:
  42. return None
  43. except Exception as e:
  44. log_.error('url: {}, exception: {}, traceback: {}'.format(request_url, e, traceback.format_exc()))
  45. return None
  46. def get_videos_remain_view_count(app_type, videos):
  47. """
  48. 获取视频在流量池中的剩余可分发数
  49. :param app_type: 产品标识 type-int
  50. :param videos: 视频信息 (视频id, 流量池标记) type-list,[{'videoId': video_id, 'flowPool': flow_pool}, ...]
  51. :return: data type-list,[(video_id, flow_pool, view_count), ...]
  52. error_flag 错误标记,True为错误
  53. """
  54. error_flag = False
  55. if not videos:
  56. return [], error_flag
  57. request_data = {'appType': app_type, 'videos': videos}
  58. result = request_post(request_url=config_.GET_REMAIN_VIEW_COUNT_URL, request_data=request_data, timeout=(0.1, 1))
  59. if result is None:
  60. error_flag = True
  61. return [], error_flag
  62. if result['code'] != 0:
  63. log_.info('获取视频在流量池中的剩余可分发数失败')
  64. error_flag = True
  65. return [], error_flag
  66. data = [(item['videoId'], item['flowPool'], item['distributeCount']) for item in result['data']]
  67. return data, error_flag
  68. def get_videos_local_distribute_count(video_id, flow_pool):
  69. """
  70. 获取流量池视频本地分发数
  71. :param video_id: video_id
  72. :param flow_pool: 流量池标记
  73. :return: current_count 本地记录的分发数
  74. """
  75. # redis_h = datetime.now().hour
  76. # if datetime.now().minute >= 30:
  77. # redis_h += 0.5
  78. # key_name = config_.LOCAL_DISTRIBUTE_COUNT_PREFIX + str(redis_h)
  79. key_name = '{}{}.{}'.format(config_.LOCAL_DISTRIBUTE_COUNT_PREFIX, video_id, flow_pool)
  80. redis_helper = RedisHelper()
  81. # video = '{}-{}'.format(video_id, flow_pool)
  82. # current_count = redis_helper.get_score_with_value(key_name=key_name, value=video)
  83. current_count = redis_helper.get_data_from_redis(key_name=key_name)
  84. if current_count is not None:
  85. return int(current_count)
  86. else:
  87. return None
  88. def update_video_w_h_rate(video_id, key_name):
  89. """
  90. 获取横屏视频的宽高比,并存入redis中 (width/height>1)
  91. :param video_id: videoId type-int
  92. :param key_name: redis key
  93. :return: None
  94. """
  95. # 获取数据
  96. sql = "SELECT id, width, height, rotate FROM longvideo.wx_video WHERE id = {};".format(video_id)
  97. mysql_helper = MysqlHelper()
  98. data = mysql_helper.get_data(sql=sql)
  99. if len(data) == 0:
  100. return
  101. # 更新到redis
  102. width, height, rotate = int(data[0][1]), int(data[0][2]), int(data[0][3])
  103. if width == 0 or height == 0:
  104. return
  105. if rotate in (90, 270):
  106. w_h_rate = height / width
  107. else:
  108. w_h_rate = width / height
  109. if w_h_rate > 1:
  110. info_data = {int(video_id): w_h_rate}
  111. else:
  112. return
  113. redis_helper = RedisHelper()
  114. # 写入新数据
  115. if len(info_data) > 0:
  116. redis_helper.add_data_with_zset(key_name=key_name, data=info_data)
  117. class FilterVideos(object):
  118. """视频过滤"""
  119. def __init__(self, app_type, video_ids, mid='', uid=''):
  120. """
  121. 初始化
  122. :param app_type: 产品标识 type-int
  123. :param video_ids: 需过滤的视频列表 type-list
  124. :param mid: mid type-string
  125. :param uid: uid type-string
  126. """
  127. self.app_type = app_type
  128. self.mid = mid
  129. self.uid = uid
  130. self.video_ids = video_ids
  131. def filter_video_status_h(self, video_ids, rule_key, ab_code, province_code, key_flag=''):
  132. """召回小时级更新的视频状态过滤"""
  133. # 根据Redis缓存中的数据过滤
  134. redis_helper = RedisHelper()
  135. # 获取不符合推荐状态的视频
  136. if ab_code in [code for _, code in config_.AB_CODE['region_rank_by_h'].items()]:
  137. if key_flag == 'region_24h':
  138. key_prefix = f"{config_.REGION_H_VIDEO_FILER_24H}{province_code}."
  139. elif key_flag == 'day_24h':
  140. key_prefix = f"{config_.H_VIDEO_FILER_24H}{province_code}."
  141. else:
  142. key_prefix = f"{config_.REGION_H_VIDEO_FILER}{province_code}."
  143. elif ab_code in [code for _, code in config_.AB_CODE['rank_by_24h'].items()]:
  144. key_prefix = config_.H_VIDEO_FILER_24H
  145. elif key_flag == '24h':
  146. key_prefix = config_.H_VIDEO_FILER_24H
  147. else:
  148. key_prefix = config_.H_VIDEO_FILER
  149. filter_videos_list = redis_helper.get_data_from_set(key_name=f"{key_prefix}{rule_key}")
  150. if not filter_videos_list:
  151. return video_ids
  152. filter_videos = [int(video) for video in filter_videos_list]
  153. filtered_videos = [video_id for video_id in video_ids if video_id not in filter_videos]
  154. return filtered_videos
  155. def filter_videos_h(self, rule_key, ab_code, province_code, key_flag=''):
  156. """召回小时级更新的视频过滤"""
  157. # 预曝光过滤
  158. st_pre = time.time()
  159. filtered_pre_result = self.filter_video_previewed(self.video_ids)
  160. et_pre = time.time()
  161. log_.info('filter by previewed: app_type = {}, mid = {}, uid = {}, request_videos = {}, '
  162. 'result = {}, executeTime = {}'.
  163. format(self.app_type, self.mid, self.uid, self.video_ids,
  164. filtered_pre_result, (et_pre - st_pre) * 1000))
  165. if not filtered_pre_result:
  166. return None
  167. # 视频状态过滤
  168. st_status = time.time()
  169. filtered_status_result = self.filter_video_status_h(video_ids=filtered_pre_result, rule_key=rule_key,
  170. ab_code=ab_code, province_code=province_code,
  171. key_flag=key_flag)
  172. et_status = time.time()
  173. log_.info('filter by video status: app_type = {}, mid = {}, uid = {}, request_videos = {}, '
  174. 'result = {}, executeTime = {}'.
  175. format(self.app_type, self.mid, self.uid, filtered_pre_result,
  176. filtered_status_result, (et_status - st_status) * 1000))
  177. if not filtered_status_result:
  178. return None
  179. # 视频已曝光过滤
  180. st_viewed = time.time()
  181. filtered_viewed_result = self.filter_video_viewed(video_ids=filtered_status_result)
  182. et_viewed = time.time()
  183. log_.info('filter by viewed: app_type = {}, mid = {}, uid = {}, request_videos = {}, '
  184. 'result = {}, executeTime = {}'.
  185. format(self.app_type, self.mid, self.uid, filtered_status_result,
  186. filtered_viewed_result, (et_viewed - st_viewed) * 1000))
  187. if not filtered_viewed_result:
  188. return None
  189. else:
  190. return [int(video_id) for video_id in filtered_viewed_result]
  191. def filter_videos(self):
  192. """视频过滤"""
  193. # 预曝光过滤
  194. st_pre = time.time()
  195. filtered_pre_result = self.filter_video_previewed(self.video_ids)
  196. et_pre = time.time()
  197. log_.info('filter by previewed: app_type = {}, mid = {}, uid = {}, request_videos = {}, '
  198. 'result = {}, executeTime = {}'.
  199. format(self.app_type, self.mid, self.uid, self.video_ids,
  200. filtered_pre_result, (et_pre - st_pre) * 1000))
  201. if not filtered_pre_result:
  202. return None
  203. # 视频状态过滤采用离线定时过滤方案
  204. # 视频状态过滤
  205. # st_status = time.time()
  206. # filtered_status_result = self.filter_video_status(video_ids=filtered_pre_result)
  207. # et_status = time.time()
  208. # log_.info('filter by video status: result = {}, execute time = {}ms'.format(
  209. # filtered_status_result, (et_status - st_status) * 1000))
  210. # if not filtered_status_result:
  211. # return None
  212. # 视频已曝光过滤
  213. st_viewed = time.time()
  214. filtered_viewed_result = self.filter_video_viewed(video_ids=filtered_pre_result)
  215. et_viewed = time.time()
  216. log_.info('filter by viewed: app_type = {}, mid = {}, uid = {}, request_videos = {}, '
  217. 'result = {}, executeTime = {}'.
  218. format(self.app_type, self.mid, self.uid, filtered_pre_result,
  219. filtered_viewed_result, (et_viewed - st_viewed) * 1000))
  220. if not filtered_viewed_result:
  221. return None
  222. else:
  223. return [int(video_id) for video_id in filtered_viewed_result]
  224. def filter_video_previewed(self, video_ids):
  225. """
  226. 预曝光过滤
  227. :param video_ids: 需过滤的视频列表 type-list
  228. :return: filtered_videos 过滤后的列表 type-list
  229. """
  230. if not self.mid:
  231. # mid为空时,不做预曝光过滤
  232. return video_ids
  233. # 根据Redis缓存中的数据过滤
  234. redis_helper = RedisHelper()
  235. # key拼接
  236. key_name = config_.PREVIEW_KEY_PREFIX + '{}.{}'.format(self.app_type, self.mid)
  237. pe_videos_list = redis_helper.get_data_from_set(key_name)
  238. # log_.info('****app_type = {}, mid = {}, uid = {}, pe_videos_list = {}'.format(
  239. # self.app_type, self.mid, self.uid, pe_videos_list))
  240. # log_.info('****app_type = {}, mid = {}, uid = {}, video_ids = {}'.format(
  241. # self.app_type, self.mid, self.uid, video_ids))
  242. if not pe_videos_list:
  243. return video_ids
  244. pe_videos = [int(video) for video in pe_videos_list]
  245. filtered_videos = [video_id for video_id in video_ids if video_id not in pe_videos]
  246. return filtered_videos
  247. def filter_video_status(self, video_ids):
  248. """
  249. 对视频状态进行过滤
  250. :param video_ids: 视频id列表 type-list
  251. :return: filtered_videos
  252. """
  253. if len(video_ids) == 1:
  254. sql = "set hg_experimental_enable_shard_pruning=off; " \
  255. "SELECT video_id " \
  256. "FROM {} " \
  257. "WHERE audit_status = 5 " \
  258. "AND applet_rec_status IN (1, -6) " \
  259. "AND open_status = 1 " \
  260. "AND payment_status = 0 " \
  261. "AND encryption_status != 5 " \
  262. "AND transcoding_status = 3 " \
  263. "AND video_id IN ({});".format(config_.VIDEO_STATUS, video_ids[0])
  264. else:
  265. sql = "set hg_experimental_enable_shard_pruning=off; " \
  266. "SELECT video_id " \
  267. "FROM {} " \
  268. "WHERE audit_status = 5 " \
  269. "AND applet_rec_status IN (1, -6) " \
  270. "AND open_status = 1 " \
  271. "AND payment_status = 0 " \
  272. "AND encryption_status != 5 " \
  273. "AND transcoding_status = 3 " \
  274. "AND video_id IN {};".format(config_.VIDEO_STATUS, tuple(video_ids))
  275. hologres_helper = HologresHelper()
  276. data = hologres_helper.get_data(sql=sql)
  277. filtered_videos = [int(temp[0]) for temp in data]
  278. return filtered_videos
  279. def filter_video_viewed(self, video_ids, types=(1, 6,)):
  280. """
  281. 调用后端接口过滤用户已观看视频
  282. :param video_ids: 视频id列表 type-list
  283. :param types: 过滤参数 type-tuple, 默认(1, ) 1-已观看 2-视频状态 3-是否进入老年人社区 4-话题状态 5-推荐状态 6-白名单过滤
  284. :return: filtered_videos
  285. """
  286. # 调用http接口
  287. request_data = {"appType": self.app_type,
  288. "mid": self.mid,
  289. "uid": self.uid,
  290. "types": list(types),
  291. "videoIds": video_ids}
  292. result = request_post(request_url=config_.VIDEO_FILTER_URL, request_data=request_data, timeout=1)
  293. if result is None:
  294. log_.info('过滤失败,types: {}'.format(types))
  295. return video_ids
  296. if result['code'] != 0:
  297. log_.info('过滤失败,types: {}'.format(types))
  298. return video_ids
  299. filtered_videos = result['data']
  300. return filtered_videos
  301. if __name__ == '__main__':
  302. # filter_ = FilterVideos(app_type=1, mid='22', uid='www', video_ids=[1, 2, 3, 55])
  303. # filter_.filter_videos()
  304. # filter_.filter_video_status(video_ids=[1, 3, 5])
  305. # videos = [{'videoId': 9034659, 'flowPool': '3#11#3#1637824188547'}, {'videoId': 9035052, 'flowPool': '3#11#3#1637824172827'}]
  306. # res = get_videos_remain_view_count(4, videos)
  307. # print(res)
  308. # text = '测试 @李倩'
  309. # send_msg_to_feishu(text)
  310. update_video_w_h_rate(video_id=113, key_name='')