import traceback import requests import json import time from datetime import datetime from db_helper import HologresHelper, RedisHelper from config import set_config from log import Log config_ = set_config() log_ = Log() def send_msg_to_feishu(msg_text): """发送消息到飞书""" # webhook地址 webhook = 'https://open.feishu.cn/open-apis/bot/v2/hook/8de4de35-30ed-4692-8854-7a154e89b2f2' # 自定义关键词key_word key_word = '服务报警' headers = {'Content-Type': 'application/json'} payload_message = { "msg_type": "text", "content": { "text": '{}: {}'.format(key_word, msg_text) } } response = requests.request('POST', url=webhook, headers=headers, data=json.dumps(payload_message)) print(response.text) def request_post(request_url, request_data, timeout): """ post 请求 HTTP接口 :param request_url: 接口URL :param request_data: 请求参数 :param timeout: 超时时间，单位为秒，type-float or tuple(connect_timeout, read_timeout) :return: res_data json格式 """ try: headers = {"Connection": "close"} response = requests.post(url=request_url, json=request_data, timeout=timeout, headers=headers) if response.status_code == 200: res_data = json.loads(response.text) return res_data else: return None except Exception as e: log_.error('url: {}, exception: {}, traceback: {}'.format(request_url, e, traceback.format_exc())) return None def get_videos_remain_view_count(app_type, videos): """ 获取视频在流量池中的剩余可分发数 :param app_type: 产品标识 type-int :param videos: 视频信息 (视频id, 流量池标记) type-list，[{'videoId': video_id, 'flowPool': flow_pool}, ...] :return: data type-list，[(video_id, flow_pool, view_count), ...] error_flag 错误标记，True为错误 """ error_flag = False if not videos: return [], error_flag request_data = {'appType': app_type, 'videos': videos} result = request_post(request_url=config_.GET_REMAIN_VIEW_COUNT_URL, request_data=request_data, timeout=(0.1, 1)) if result is None: error_flag = True return [], error_flag if result['code'] != 0: log_.info('获取视频在流量池中的剩余可分发数失败') error_flag = True return [], error_flag data = [(item['videoId'], item['flowPool'], item['distributeCount']) for item in result['data']] return data, error_flag def get_videos_local_distribute_count(video_id, flow_pool): """ 获取流量池视频本地分发数 :param video_id: video_id :param flow_pool: 流量池标记 :return: current_count 本地记录的分发数 """ # redis_h = datetime.now().hour # if datetime.now().minute >= 30: # redis_h += 0.5 # key_name = config_.LOCAL_DISTRIBUTE_COUNT_PREFIX + str(redis_h) key_name = '{}{}.{}'.format(config_.LOCAL_DISTRIBUTE_COUNT_PREFIX, video_id, flow_pool) redis_helper = RedisHelper() # video = '{}-{}'.format(video_id, flow_pool) # current_count = redis_helper.get_score_with_value(key_name=key_name, value=video) current_count = redis_helper.get_data_from_redis(key_name=key_name) if current_count is not None: return int(current_count) else: return None class FilterVideos(object): """视频过滤""" def __init__(self, app_type, video_ids, mid='', uid=''): """ 初始化 :param app_type: 产品标识 type-int :param video_ids: 需过滤的视频列表 type-list :param mid: mid type-string :param uid: uid type-string """ self.app_type = app_type self.mid = mid self.uid = uid self.video_ids = video_ids def filter_videos(self): """视频过滤""" # 预曝光过滤 st_pre = time.time() filtered_pre_result = self.filter_video_previewed(self.video_ids) et_pre = time.time() log_.info('filter by previewed: app_type = {}, result = {}, execute time = {}ms'.format( self.app_type, filtered_pre_result, (et_pre - st_pre) * 1000)) if not filtered_pre_result: return None # 视频状态过滤 st_status = time.time() filtered_status_result = self.filter_video_status(video_ids=filtered_pre_result) et_status = time.time() log_.info('filter by video status: result = {}, execute time = {}ms'.format( filtered_status_result, (et_status - st_status) * 1000)) if not filtered_status_result: return None # 视频已曝光过滤 st_viewed = time.time() filtered_viewed_result = self.filter_video_viewed(video_ids=filtered_status_result) et_viewed = time.time() log_.info('filter by viewed: app_type = {}, mid = {}, uid = {}, result = {}, execute time = {}ms'.format( self.app_type, self.mid, self.uid, filtered_viewed_result, (et_viewed - st_viewed) * 1000)) if not filtered_viewed_result: return None else: return [int(video_id) for video_id in filtered_viewed_result] def filter_video_previewed(self, video_ids): """ 预曝光过滤 :param video_ids: 需过滤的视频列表 type-list :return: filtered_videos 过滤后的列表 type-list """ # 根据Redis缓存中的数据过滤 redis_helper = RedisHelper() # key拼接 key_name = config_.PREVIEW_KEY_PREFIX + '{}.{}'.format(self.app_type, self.mid) pe_videos_list = redis_helper.get_data_from_set(key_name) if not pe_videos_list: return video_ids pe_videos = [int(video) for video in pe_videos_list] filtered_videos = [video_id for video_id in video_ids if video_id not in pe_videos] return filtered_videos def filter_video_status(self, video_ids): """ 对视频状态进行过滤 :param video_ids: 视频id列表 type-list :return: filtered_videos """ if len(video_ids) == 1: sql = "set hg_experimental_enable_shard_pruning=off; " \ "SELECT video_id " \ "FROM {} " \ "WHERE audit_status = 5 " \ "AND applet_rec_status IN (1, -6) " \ "AND open_status = 1 " \ "AND payment_status = 0 " \ "AND encryption_status IS NULL " \ "AND transcoding_status = 3 " \ "AND video_id IN ({});".format(config_.VIDEO_STATUS, video_ids[0]) else: sql = "set hg_experimental_enable_shard_pruning=off; " \ "SELECT video_id " \ "FROM {} " \ "WHERE audit_status = 5 " \ "AND applet_rec_status IN (1, -6) " \ "AND open_status = 1 " \ "AND payment_status = 0 " \ "AND encryption_status IS NULL " \ "AND transcoding_status = 3 " \ "AND video_id IN {};".format(config_.VIDEO_STATUS, tuple(video_ids)) hologres_helper = HologresHelper() data = hologres_helper.get_data(sql=sql) filtered_videos = [int(temp[0]) for temp in data] return filtered_videos def filter_video_viewed(self, video_ids, types=(1,)): """ 调用后端接口过滤用户已观看视频 :param video_ids: 视频id列表 type-list :param types: 过滤参数 type-tuple, 默认(1, ) 1-已观看 2-视频状态 3-是否进入老年人社区 4-话题状态 5-推荐状态 :return: filtered_videos """ # 调用http接口 request_data = {"appType": self.app_type, "mid": self.mid, "uid": self.uid, "types": list(types), "videoIds": video_ids} result = request_post(request_url=config_.VIDEO_FILTER_URL, request_data=request_data, timeout=2.5) if result is None: log_.info('过滤失败，types: {}'.format(types)) return video_ids if result['code'] != 0: log_.info('过滤失败，types: {}'.format(types)) return video_ids filtered_videos = result['data'] return filtered_videos if __name__ == '__main__': # filter_ = FilterVideos(app_type=1, mid='22', uid='www', video_ids=[1, 2, 3, 55]) # filter_.filter_videos() # filter_.filter_video_status(video_ids=[1, 3, 5]) # videos = [{'videoId': 9034659, 'flowPool': '3#11#3#1637824188547'}, {'videoId': 9035052, 'flowPool': '3#11#3#1637824172827'}] # res = get_videos_remain_view_count(4, videos) # print(res) text = '测试 @李倩' send_msg_to_feishu(text)