utils.py 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325
  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):
  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. key_prefix = f"{config_.REGION_H_VIDEO_FILER}{province_code}."
  138. else:
  139. key_prefix = config_.H_VIDEO_FILER
  140. filter_videos_list = redis_helper.get_data_from_set(key_name=f"{key_prefix}{rule_key}")
  141. if not filter_videos_list:
  142. return video_ids
  143. filter_videos = [int(video) for video in filter_videos_list]
  144. filtered_videos = [video_id for video_id in video_ids if video_id not in filter_videos]
  145. return filtered_videos
  146. def filter_videos_h(self, rule_key, ab_code, province_code):
  147. """召回小时级更新的视频过滤"""
  148. # 预曝光过滤
  149. st_pre = time.time()
  150. filtered_pre_result = self.filter_video_previewed(self.video_ids)
  151. et_pre = time.time()
  152. log_.info('filter by previewed: app_type = {}, mid = {}, uid = {}, result = {}, execute time = {}ms'.format(
  153. self.app_type, self.mid, self.uid, filtered_pre_result, (et_pre - st_pre) * 1000))
  154. if not filtered_pre_result:
  155. return None
  156. # 视频状态过滤
  157. st_status = time.time()
  158. filtered_status_result = self.filter_video_status_h(video_ids=filtered_pre_result, rule_key=rule_key,
  159. ab_code=ab_code, province_code=province_code)
  160. et_status = time.time()
  161. log_.info('filter by video status: result = {}, execute time = {}ms'.format(
  162. filtered_status_result, (et_status - st_status) * 1000))
  163. if not filtered_status_result:
  164. return None
  165. # 视频已曝光过滤
  166. st_viewed = time.time()
  167. filtered_viewed_result = self.filter_video_viewed(video_ids=filtered_status_result)
  168. et_viewed = time.time()
  169. log_.info('filter by viewed: app_type = {}, mid = {}, uid = {}, result = {}, execute time = {}ms'.format(
  170. self.app_type, self.mid, self.uid, filtered_viewed_result, (et_viewed - st_viewed) * 1000))
  171. if not filtered_viewed_result:
  172. return None
  173. else:
  174. return [int(video_id) for video_id in filtered_viewed_result]
  175. def filter_videos(self):
  176. """视频过滤"""
  177. # 预曝光过滤
  178. st_pre = time.time()
  179. filtered_pre_result = self.filter_video_previewed(self.video_ids)
  180. et_pre = time.time()
  181. log_.info('filter by previewed: app_type = {}, mid = {}, uid = {}, result = {}, execute time = {}ms'.format(
  182. self.app_type, self.mid, self.uid, filtered_pre_result, (et_pre - st_pre) * 1000))
  183. if not filtered_pre_result:
  184. return None
  185. # 视频状态过滤采用离线定时过滤方案
  186. # 视频状态过滤
  187. # st_status = time.time()
  188. # filtered_status_result = self.filter_video_status(video_ids=filtered_pre_result)
  189. # et_status = time.time()
  190. # log_.info('filter by video status: result = {}, execute time = {}ms'.format(
  191. # filtered_status_result, (et_status - st_status) * 1000))
  192. # if not filtered_status_result:
  193. # return None
  194. # 视频已曝光过滤
  195. st_viewed = time.time()
  196. filtered_viewed_result = self.filter_video_viewed(video_ids=filtered_pre_result)
  197. et_viewed = time.time()
  198. log_.info('filter by viewed: app_type = {}, mid = {}, uid = {}, result = {}, execute time = {}ms'.format(
  199. self.app_type, self.mid, self.uid, filtered_viewed_result, (et_viewed - st_viewed) * 1000))
  200. if not filtered_viewed_result:
  201. return None
  202. else:
  203. return [int(video_id) for video_id in filtered_viewed_result]
  204. def filter_video_previewed(self, video_ids):
  205. """
  206. 预曝光过滤
  207. :param video_ids: 需过滤的视频列表 type-list
  208. :return: filtered_videos 过滤后的列表 type-list
  209. """
  210. # 根据Redis缓存中的数据过滤
  211. redis_helper = RedisHelper()
  212. # key拼接
  213. key_name = config_.PREVIEW_KEY_PREFIX + '{}.{}'.format(self.app_type, self.mid)
  214. pe_videos_list = redis_helper.get_data_from_set(key_name)
  215. log_.info('****app_type = {}, mid = {}, uid = {}, pe_videos_list = {}'.format(
  216. self.app_type, self.mid, self.uid, pe_videos_list))
  217. log_.info('****app_type = {}, mid = {}, uid = {}, video_ids = {}'.format(
  218. self.app_type, self.mid, self.uid, video_ids))
  219. if not pe_videos_list:
  220. return video_ids
  221. pe_videos = [int(video) for video in pe_videos_list]
  222. filtered_videos = [video_id for video_id in video_ids if video_id not in pe_videos]
  223. return filtered_videos
  224. def filter_video_status(self, video_ids):
  225. """
  226. 对视频状态进行过滤
  227. :param video_ids: 视频id列表 type-list
  228. :return: filtered_videos
  229. """
  230. if len(video_ids) == 1:
  231. sql = "set hg_experimental_enable_shard_pruning=off; " \
  232. "SELECT video_id " \
  233. "FROM {} " \
  234. "WHERE audit_status = 5 " \
  235. "AND applet_rec_status IN (1, -6) " \
  236. "AND open_status = 1 " \
  237. "AND payment_status = 0 " \
  238. "AND encryption_status != 5 " \
  239. "AND transcoding_status = 3 " \
  240. "AND video_id IN ({});".format(config_.VIDEO_STATUS, video_ids[0])
  241. else:
  242. sql = "set hg_experimental_enable_shard_pruning=off; " \
  243. "SELECT video_id " \
  244. "FROM {} " \
  245. "WHERE audit_status = 5 " \
  246. "AND applet_rec_status IN (1, -6) " \
  247. "AND open_status = 1 " \
  248. "AND payment_status = 0 " \
  249. "AND encryption_status != 5 " \
  250. "AND transcoding_status = 3 " \
  251. "AND video_id IN {};".format(config_.VIDEO_STATUS, tuple(video_ids))
  252. hologres_helper = HologresHelper()
  253. data = hologres_helper.get_data(sql=sql)
  254. filtered_videos = [int(temp[0]) for temp in data]
  255. return filtered_videos
  256. def filter_video_viewed(self, video_ids, types=(1, 6,)):
  257. """
  258. 调用后端接口过滤用户已观看视频
  259. :param video_ids: 视频id列表 type-list
  260. :param types: 过滤参数 type-tuple, 默认(1, ) 1-已观看 2-视频状态 3-是否进入老年人社区 4-话题状态 5-推荐状态 6-白名单过滤
  261. :return: filtered_videos
  262. """
  263. # 调用http接口
  264. request_data = {"appType": self.app_type,
  265. "mid": self.mid,
  266. "uid": self.uid,
  267. "types": list(types),
  268. "videoIds": video_ids}
  269. result = request_post(request_url=config_.VIDEO_FILTER_URL, request_data=request_data, timeout=2.5)
  270. if result is None:
  271. log_.info('过滤失败,types: {}'.format(types))
  272. return video_ids
  273. if result['code'] != 0:
  274. log_.info('过滤失败,types: {}'.format(types))
  275. return video_ids
  276. filtered_videos = result['data']
  277. return filtered_videos
  278. if __name__ == '__main__':
  279. # filter_ = FilterVideos(app_type=1, mid='22', uid='www', video_ids=[1, 2, 3, 55])
  280. # filter_.filter_videos()
  281. # filter_.filter_video_status(video_ids=[1, 3, 5])
  282. # videos = [{'videoId': 9034659, 'flowPool': '3#11#3#1637824188547'}, {'videoId': 9035052, 'flowPool': '3#11#3#1637824172827'}]
  283. # res = get_videos_remain_view_count(4, videos)
  284. # print(res)
  285. # text = '测试 @李倩'
  286. # send_msg_to_feishu(text)
  287. update_video_w_h_rate(video_id=113, key_name='')