utils.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365
  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, request_id, app_type, video_ids, mid='', uid=''):
  120. """
  121. 初始化
  122. :param request_id: request_id
  123. :param app_type: 产品标识 type-int
  124. :param video_ids: 需过滤的视频列表 type-list
  125. :param mid: mid type-string
  126. :param uid: uid type-string
  127. """
  128. self.request_id = request_id
  129. self.app_type = app_type
  130. self.mid = mid
  131. self.uid = uid
  132. self.video_ids = video_ids
  133. def filter_video_status_h(self, video_ids, rule_key, ab_code, province_code, key_flag=''):
  134. """召回小时级更新的视频状态过滤"""
  135. # 根据Redis缓存中的数据过滤
  136. redis_helper = RedisHelper()
  137. # 获取不符合推荐状态的视频
  138. if ab_code in [code for _, code in config_.AB_CODE['region_rank_by_h'].items()]:
  139. if key_flag == 'region_24h':
  140. key_prefix = f"{config_.REGION_H_VIDEO_FILER_24H}{province_code}."
  141. elif key_flag == 'day_24h':
  142. key_prefix = f"{config_.H_VIDEO_FILER_24H}{province_code}."
  143. else:
  144. key_prefix = f"{config_.REGION_H_VIDEO_FILER}{province_code}."
  145. elif ab_code in [code for _, code in config_.AB_CODE['rank_by_24h'].items()]:
  146. key_prefix = config_.H_VIDEO_FILER_24H
  147. elif key_flag == '24h':
  148. key_prefix = config_.H_VIDEO_FILER_24H
  149. else:
  150. key_prefix = config_.H_VIDEO_FILER
  151. filter_videos_list = redis_helper.get_data_from_set(key_name=f"{key_prefix}{rule_key}")
  152. if not filter_videos_list:
  153. return video_ids
  154. filter_videos = [int(video) for video in filter_videos_list]
  155. filtered_videos = [video_id for video_id in video_ids if video_id not in filter_videos]
  156. return filtered_videos
  157. def filter_videos_h(self, rule_key, ab_code, province_code, key_flag=''):
  158. """召回小时级更新的视频过滤"""
  159. # 预曝光过滤
  160. st_pre = time.time()
  161. filtered_pre_result = self.filter_video_previewed(self.video_ids)
  162. et_pre = time.time()
  163. log_.info({'request_id': self.request_id,
  164. 'app_type': self.app_type,
  165. 'mid': self.mid,
  166. 'uid': self.uid,
  167. 'request_videos': self.video_ids,
  168. 'preview_filter_result': filtered_pre_result,
  169. 'executeTime': (et_pre - st_pre) * 1000})
  170. if not filtered_pre_result:
  171. return None
  172. # 视频状态过滤
  173. st_status = time.time()
  174. filtered_status_result = self.filter_video_status_h(video_ids=filtered_pre_result, rule_key=rule_key,
  175. ab_code=ab_code, province_code=province_code,
  176. key_flag=key_flag)
  177. et_status = time.time()
  178. log_.info({'request_id': self.request_id,
  179. 'app_type': self.app_type,
  180. 'mid': self.mid,
  181. 'uid': self.uid,
  182. 'request_videos': filtered_pre_result,
  183. 'status_filter_result': filtered_status_result,
  184. 'executeTime': (et_status - st_status) * 1000})
  185. if not filtered_status_result:
  186. return None
  187. # 视频已曝光过滤
  188. st_viewed = time.time()
  189. filtered_viewed_result = self.filter_video_viewed(video_ids=filtered_status_result)
  190. et_viewed = time.time()
  191. log_.info({'request_id': self.request_id,
  192. 'app_type': self.app_type,
  193. 'mid': self.mid,
  194. 'uid': self.uid,
  195. 'request_videos': filtered_status_result,
  196. 'view_filter_result': filtered_viewed_result,
  197. 'executeTime': (et_viewed - st_viewed) * 1000})
  198. if not filtered_viewed_result:
  199. return None
  200. else:
  201. return [int(video_id) for video_id in filtered_viewed_result]
  202. def filter_videos(self):
  203. """视频过滤"""
  204. # 预曝光过滤
  205. st_pre = time.time()
  206. filtered_pre_result = self.filter_video_previewed(self.video_ids)
  207. et_pre = time.time()
  208. log_.info({'request_id': self.request_id,
  209. 'app_type': self.app_type,
  210. 'mid': self.mid,
  211. 'uid': self.uid,
  212. 'request_videos': self.video_ids,
  213. 'preview_filter_result': filtered_pre_result,
  214. 'executeTime': (et_pre - st_pre) * 1000})
  215. if not filtered_pre_result:
  216. return None
  217. # 视频状态过滤采用离线定时过滤方案
  218. # 视频状态过滤
  219. # st_status = time.time()
  220. # filtered_status_result = self.filter_video_status(video_ids=filtered_pre_result)
  221. # et_status = time.time()
  222. # log_.info('filter by video status: result = {}, execute time = {}ms'.format(
  223. # filtered_status_result, (et_status - st_status) * 1000))
  224. # if not filtered_status_result:
  225. # return None
  226. # 视频已曝光过滤
  227. st_viewed = time.time()
  228. filtered_viewed_result = self.filter_video_viewed(video_ids=filtered_pre_result)
  229. et_viewed = time.time()
  230. log_.info({'request_id': self.request_id,
  231. 'app_type': self.app_type,
  232. 'mid': self.mid,
  233. 'uid': self.uid,
  234. 'request_videos': filtered_pre_result,
  235. 'view_filter_result': filtered_viewed_result,
  236. 'executeTime': (et_viewed - st_viewed) * 1000})
  237. if not filtered_viewed_result:
  238. return None
  239. else:
  240. return [int(video_id) for video_id in filtered_viewed_result]
  241. def filter_video_previewed(self, video_ids):
  242. """
  243. 预曝光过滤
  244. :param video_ids: 需过滤的视频列表 type-list
  245. :return: filtered_videos 过滤后的列表 type-list
  246. """
  247. if not self.mid:
  248. # mid为空时,不做预曝光过滤
  249. return video_ids
  250. # 根据Redis缓存中的数据过滤
  251. redis_helper = RedisHelper()
  252. # key拼接
  253. key_name = config_.PREVIEW_KEY_PREFIX + '{}.{}'.format(self.app_type, self.mid)
  254. pe_videos_list = redis_helper.get_data_from_set(key_name)
  255. # log_.info('****app_type = {}, mid = {}, uid = {}, pe_videos_list = {}'.format(
  256. # self.app_type, self.mid, self.uid, pe_videos_list))
  257. # log_.info('****app_type = {}, mid = {}, uid = {}, video_ids = {}'.format(
  258. # self.app_type, self.mid, self.uid, video_ids))
  259. if not pe_videos_list:
  260. return video_ids
  261. pe_videos = [int(video) for video in pe_videos_list]
  262. filtered_videos = [video_id for video_id in video_ids if video_id not in pe_videos]
  263. return filtered_videos
  264. def filter_video_status(self, video_ids):
  265. """
  266. 对视频状态进行过滤
  267. :param video_ids: 视频id列表 type-list
  268. :return: filtered_videos
  269. """
  270. if len(video_ids) == 1:
  271. sql = "set hg_experimental_enable_shard_pruning=off; " \
  272. "SELECT video_id " \
  273. "FROM {} " \
  274. "WHERE audit_status = 5 " \
  275. "AND applet_rec_status IN (1, -6) " \
  276. "AND open_status = 1 " \
  277. "AND payment_status = 0 " \
  278. "AND encryption_status != 5 " \
  279. "AND transcoding_status = 3 " \
  280. "AND video_id IN ({});".format(config_.VIDEO_STATUS, video_ids[0])
  281. else:
  282. sql = "set hg_experimental_enable_shard_pruning=off; " \
  283. "SELECT video_id " \
  284. "FROM {} " \
  285. "WHERE audit_status = 5 " \
  286. "AND applet_rec_status IN (1, -6) " \
  287. "AND open_status = 1 " \
  288. "AND payment_status = 0 " \
  289. "AND encryption_status != 5 " \
  290. "AND transcoding_status = 3 " \
  291. "AND video_id IN {};".format(config_.VIDEO_STATUS, tuple(video_ids))
  292. hologres_helper = HologresHelper()
  293. data = hologres_helper.get_data(sql=sql)
  294. filtered_videos = [int(temp[0]) for temp in data]
  295. return filtered_videos
  296. def filter_video_viewed(self, video_ids, types=(1, 6,)):
  297. """
  298. 调用后端接口过滤用户已观看视频
  299. :param video_ids: 视频id列表 type-list
  300. :param types: 过滤参数 type-tuple, 默认(1, ) 1-已观看 2-视频状态 3-是否进入老年人社区 4-话题状态 5-推荐状态 6-白名单过滤
  301. :return: filtered_videos
  302. """
  303. # 调用http接口
  304. request_data = {"appType": self.app_type,
  305. "mid": self.mid,
  306. "uid": self.uid,
  307. "types": list(types),
  308. "videoIds": video_ids}
  309. result = request_post(request_url=config_.VIDEO_FILTER_URL, request_data=request_data, timeout=1)
  310. if result is None:
  311. log_.info('过滤失败,types: {}'.format(types))
  312. return video_ids
  313. if result['code'] != 0:
  314. log_.info('过滤失败,types: {}'.format(types))
  315. return video_ids
  316. filtered_videos = result['data']
  317. return filtered_videos
  318. if __name__ == '__main__':
  319. # filter_ = FilterVideos(app_type=1, mid='22', uid='www', video_ids=[1, 2, 3, 55])
  320. # filter_.filter_videos()
  321. # filter_.filter_video_status(video_ids=[1, 3, 5])
  322. # videos = [{'videoId': 9034659, 'flowPool': '3#11#3#1637824188547'}, {'videoId': 9035052, 'flowPool': '3#11#3#1637824172827'}]
  323. # res = get_videos_remain_view_count(4, videos)
  324. # print(res)
  325. # text = '测试 @李倩'
  326. # send_msg_to_feishu(text)
  327. update_video_w_h_rate(video_id=113, key_name='')