# -*- encoding: utf-8 -*- """ @File : DevicePushService.py @Time : 2022/11/23 11:40 @Author : stephen @Email : zhangdongming@asj6.wecom.work @Software: PyCharm """ import hashlib import json import logging import threading import time import boto3 import botocore import requests from AnsjerPush.Config.aiConfig import DEVICE_EVENT_TYPE, ALGORITHM_COMBO_TYPES from AnsjerPush.config import CONFIG_INFO, CONFIG_CN, MULTI_CHANNEL_TYPE_LIST, SYS_EVENT_TYPE_LIST, AWS_ACCESS_KEY_ID, \ AWS_SECRET_ACCESS_KEY, EVENT_DICT, EVENT_DICT_CN, CONFIG_US from AnsjerPush.config import XMPUSH_CONFIG, OPPOPUSH_CONFIG from Model.models import UidPushModel, SysMsgModel, DeviceSharePermission, DeviceChannelUserSet, \ DeviceChannelUserPermission, UidSetModel, Device_Info from Object.ETkObject import ETkObject from Object.RedisObject import RedisObject from Object.UidTokenObject import UidTokenObject from Object.utils import LocalDateTimeUtil from Service.CommonService import CommonService from Service.EquipmentInfoService import EquipmentInfoService, EQUIPMENT_INFO_DICT from Service.HuaweiPushService.HuaweiPushService import HuaweiPushObject from Service.PushService import PushObject LOGGING = logging.getLogger('info') class DevicePushService: @staticmethod def decode_uid(etk, uidToken): """ 解密UID,优先解密etk 否则判断uidToken """ # 解密获取uid if etk: eto = ETkObject(etk) uid = eto.uid else: uto = UidTokenObject(uidToken) uid = uto.UID LOGGING.info('消息推送-当前UID：{}'.format(uid)) return uid @staticmethod def judge_sys_msg(event_type): """ 判断是否属于系统消息 @param event_type: 事件类型 @return: bool """ if event_type in SYS_EVENT_TYPE_LIST: return True return False @staticmethod def get_s3_client(region): """ 根据地区获取S3 client @param region: 地区,1：国外, 2:国内 @return: aws_s3_client """ if int(region) == 1: aws_s3_client = boto3.client( 's3', aws_access_key_id=AWS_ACCESS_KEY_ID[1], aws_secret_access_key=AWS_SECRET_ACCESS_KEY[1], config=botocore.client.Config(signature_version='s3v4'), region_name='us-east-1' ) else: aws_s3_client = boto3.client( 's3', aws_access_key_id=AWS_ACCESS_KEY_ID[0], aws_secret_access_key=AWS_SECRET_ACCESS_KEY[0], config=botocore.client.Config(signature_version='s3v4'), region_name='cn-northwest-1' ) return aws_s3_client @classmethod def query_uid_push(cls, uid, event_type): """ 查询uid_push和uid_set数据 @param uid: uid @param event_type: 事件类型 @return: uid_push_qs """ if event_type not in [606, 607]: uid_push_qs = UidPushModel.objects.filter(uid_set__uid=uid, uid_set__detect_status=1). \ values('token_val', 'app_type', 'appBundleId', 'm_code', 'push_type', 'userID_id', 'userID__NickName', 'lang', 'm_code', 'tz', 'uid_set__nickname', 'uid_set__detect_interval', 'uid_set__detect_group', 'uid_set__channel', 'uid_set__ai_type', 'uid_set__device_type', 'uid_set__new_detect_interval', 'uid_set__msg_notify') else: # 一键通话只推主用户 device_info_qs = Device_Info.objects.filter(UID=uid).values('vodPrimaryUserID') primary_user_id = device_info_qs[0]['vodPrimaryUserID'] uid_push_qs = UidPushModel.objects.filter(uid_set__uid=uid, userID_id=primary_user_id). \ values('token_val', 'app_type', 'appBundleId', 'm_code', 'push_type', 'userID_id', 'userID__NickName', 'lang', 'm_code', 'tz', 'uid_set__nickname', 'uid_set__detect_interval', 'uid_set__detect_group', 'uid_set__channel', 'uid_set__ai_type', 'uid_set__device_type', 'uid_set__new_detect_interval', 'uid_set__msg_notify', 'jg_token_val') return uid_push_qs @staticmethod def qs_to_list(qs): """ qs对象转存列表 @param qs: query set对象 @return: qs_list """ qs_list = [] for i in qs: qs_list.append(i) return qs_list @staticmethod def cache_push_detect_interval(redis_obj, name, detect_interval, new_detect_interval): """ 缓存设置推送消息的时间间隔 @param redis_obj: redis对象 @param name: redis key @param detect_interval: 原推送时间间隔 @param new_detect_interval: 新推送时间间隔 """ if CONFIG_INFO != CONFIG_CN: detect_interval = new_detect_interval if new_detect_interval > 0 else detect_interval detect_interval = 60 if detect_interval < 60 else detect_interval else: # 国内推送兼容问题，有值并且大于旧消息间隔则使用new_detect_interval detect_interval = new_detect_interval if new_detect_interval > detect_interval else detect_interval redis_obj.set_data(key=name, val=1, expire=detect_interval - 5) @classmethod def save_msg_push(cls, **params): """ 保存推送数据和推送消息 @param params: 推送参数 @return: bool """ sys_msg_list = [] saved_user_id_list = [] uid = params['uid'] push_kwargs = params['push_kwargs'] now_time = int(time.time()) redis_obj = RedisObject() try: params['event_tag'] = cls.get_event_tag(params['ai_type'], params['event_type'], params['detection']) is_app_push = True if params['event_type'] in [606, 607] else \ cls.is_send_app_push( params['event_type'], params['event_tag'], params['app_push_config'], params['app_push'], uid) save_equipment_info = False equipment_info_key = EquipmentInfoService.randoms_choice_equipment_info_key() for up in params['uid_set_push_list']: lang = up['lang'] tz = up['tz'] if tz is None or tz == '': tz = 0 # 保存系统消息或推送消息数据 user_id = up['userID_id'] if user_id not in saved_user_id_list: # 防止同一用户重复写入数据 # 系统消息 if params['is_sys_msg']: sys_msg_text = cls.get_msg_text(channel=params['channel'], n_time=params['n_time'], lang=lang, tz=tz, is_sys=1, device_type=params['device_type'], event_type=params['event_type'], electricity=params['electricity']) sys_msg_list.append(SysMsgModel(userID_id=user_id, msg=sys_msg_text, addTime=now_time, updTime=now_time, uid=uid, eventType=params['event_type'])) # 保存推送消息 else: if not save_equipment_info: save_equipment_info = True params['userID_id'] = user_id answer_status = 1 if params['dealings_type'] == 1 else 0 equipment_info_kwargs = { 'device_user_id': params['userID_id'], 'event_time': params['n_time'], 'event_type': params['event_type'], 'device_uid': params['uid'], 'device_nick_name': params['nickname'], 'channel': params['channel'], 'alarm': 'Motion \tChannel:{}'.format(params['channel']), 'is_st': params['is_st'], 'add_time': int(time.time()), 'storage_location': params['storage_location'], 'event_tag': params['event_tag'], 'answer_status': answer_status } # 保存到redis列表 equipment_info_value = json.dumps(equipment_info_kwargs) redis_obj.rpush(equipment_info_key, equipment_info_value) saved_user_id_list.append(user_id) # 推送 if is_app_push: push_type = up['push_type'] if params['event_type'] in [606, 607] and push_type in [5, 6]: push_kwargs['jg_token_val'] = up['jg_token_val'] else: if 'jg_token_val' in push_kwargs: push_kwargs.pop('jg_token_val') appBundleId = up['appBundleId'] token_val = up['token_val'] # 发送标题 msg_title = cls.get_msg_title(nickname=params['nickname']) # 发送内容 msg_text = cls.get_msg_text(channel=params['channel'], n_time=params['n_time'], lang=lang, tz=tz, event_type=params['event_type'], ai_type=params['ai_type'], device_type=params['device_type'], electricity=params['electricity'], dealings_type=params['dealings_type'], event_tag=params['event_tag'] ) # 补齐推送参数 push_kwargs['appBundleId'] = appBundleId push_kwargs['token_val'] = token_val push_kwargs['msg_title'] = msg_title push_kwargs['msg_text'] = msg_text params['push_kwargs'] = push_kwargs params['appBundleId'] = appBundleId params['token_val'] = token_val params['lang'] = lang params['tz'] = tz push_result = cls.send_app_msg_push(push_type, **params) LOGGING.info('{}推送类型:{},推送结果:{}'.format(uid, push_type, push_result)) # 写入系统消息 if sys_msg_list: SysMsgModel.objects.bulk_create(sys_msg_list) if save_equipment_info: equipment_info_list = [] equipment_info_model = EQUIPMENT_INFO_DICT[equipment_info_key] end = 0 equipment_info_len = redis_obj.llen(equipment_info_key) # 一键通话和视频通话需要实时写入数据 # 美洲服通过定时任务写入数据 if params['event_type'] in [606, 607] or CONFIG_INFO != CONFIG_US: # 缓存数据多于100条,批量保存前100条,否则保存全部 equipment_info_len = redis_obj.llen(equipment_info_key) end = 99 if equipment_info_len > 100 else equipment_info_len - 1 if end != 0: equipment_info_redis_list = redis_obj.lrange(equipment_info_key, 0, end) redis_obj.ltrim(equipment_info_key, end+1, -1) for equipment_info in equipment_info_redis_list: equipment_info_data = eval(equipment_info) # 设备昵称存在表情,解码utf-8 if equipment_info_data.get('device_nick_name') is not None: equipment_info_data['device_nick_name'] = equipment_info_data['device_nick_name'].\ encode('UTF-8', 'ignore').decode('UTF-8') equipment_info_list.append(equipment_info_model(**equipment_info_data)) equipment_info_model.objects.bulk_create(equipment_info_list) return True except Exception as e: LOGGING.info('推送消息或存表异常: error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e))) return False @classmethod def get_event_tag(cls, ai_type, event_type, detection=0): """ 获取事件标签 """ algorithm = False if ai_type > 0 and detection == 1: algorithm = True elif (ai_type == 7 and event_type <= 7) or (ai_type == 47 and event_type <= 47) or (detection == 1): algorithm = True if not algorithm: return ',' + str(event_type) + ',' event_res = DEVICE_EVENT_TYPE.get(event_type, 0) if event_res > 0: return ',' + str(event_res) + ',' event_type = cls.dec_to_bin(event_type) types = cls.get_combo_types(event_type) res = ','.join(types) + ',' return ',' + res @classmethod def get_combo_types(cls, event_type): """ 获取设备算法组合类型 51：移动侦测，52：传感器报警，53：影像遗失，54：PIR，55：门磁报警，56：外部发报，57：人型报警（提示：有人出现），58：车型，59：宠物，60：人脸，61：异响， 62：区域闯入，63：区域闯出，64：长时间无人检测，65：长时间无人检测，66：往来检测，67：哭声检测，68：手势检测,69:火焰检测 0：代表空字符，702：摄像头休眠，703：摄像头唤醒，704：电量过低 AWS AI识别 1：人形，2：车型，3：宠物，4：包裹。云端AI类型 """ try: types = [] event_type = str(event_type) len_type = len(event_type) for i in range(len_type): e_type = event_type[len_type - 1 - i] if e_type == '1': types.append(str(ALGORITHM_COMBO_TYPES[i])) LOGGING.info('算法对照打印：{}'.format(ALGORITHM_COMBO_TYPES)) return types except Exception as e: print('推送错误异常,errLine:{}, errMsg:{}'.format(e.__traceback__.tb_lineno, repr(e))) return event_type @staticmethod def dec_to_bin(num): """ 十进制转二进制 """ result = "" while num != 0: ret = num % 2 num //= 2 result = str(ret) + result return result @classmethod def send_app_msg_push(cls, push_type, **param): """ 发送推送 @param push_type: 推送类型 @return push_result: bool """ try: push_kwargs = param['push_kwargs'] push_result = False # is_st为1或3,且推送类型为apns,gcm,华为,异步推送图片 if (param['is_st'] == 1 or param['is_st'] == 3) and \ (push_type == 0 or push_type == 1 or push_type == 3): if param['is_st'] == 1: key = '{}/{}/{}.jpeg'.format(param['uid'], param['channel'], param['n_time']) else: key = '{}/{}/{}_0.jpeg'.format(param['uid'], param['channel'], param['n_time']) # 开始异步推送图片 push_thread = threading.Thread(target=cls.async_send_picture_push, args=( push_type, param['aws_s3_client'], param['bucket'], key, param['uid'], param['appBundleId'], param['token_val'], param['event_type'], param['n_time'], push_kwargs['msg_title'], push_kwargs['msg_text'], param['channel'])) push_thread.start() push_result = True # 不推图 else: if push_type in [0, 1, 2]: kwargs = { 'nickname': param['uid'], 'app_bundle_id': param['appBundleId'], 'token_val': param['token_val'], 'n_time': param['n_time'], 'event_type': param['event_type'], 'msg_title': push_kwargs['msg_title'], 'msg_text': push_kwargs['msg_text'], 'uid': param['uid'], 'channel': param['channel'] } if push_type == 0: # ios apns push_result = PushObject.ios_apns_push(**kwargs) elif push_type == 1: # android gcm push_result = PushObject.android_fcm_push(**kwargs) elif push_type == 2: # android jpush kwargs.pop('uid') push_result = PushObject.android_jpush(**kwargs) elif push_type == 3: huawei_push_object = HuaweiPushObject() huawei_push_object.send_push_notify_message(**push_kwargs) elif push_type == 4: # android xmpush if param['event_type'] in [606, 607]: channel_id = 111934 else: channel_id = 104551 cls.do_xmpush(channel_id=channel_id, **push_kwargs) push_result = True elif push_type == 5: # android vivopush push_result = PushObject.android_vivopush(**push_kwargs) elif push_type == 6: # android oppopush channel_id = 'DEVICE_REMINDER' cls.do_oppopush(channel_id=channel_id, **push_kwargs) push_result = True elif push_type == 7: # android meizupush push_result = PushObject.android_meizupush(**push_kwargs) return push_result except Exception as e: LOGGING.info('发送推送异常,error_line:{},error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e))) return False @staticmethod def get_msg_title(nickname): """ 获取消息标题 """"" return nickname @staticmethod def get_event_type_text(lang, event_type, dealings_type): """ 事件类型文案键值查找 """ if lang == 'cn': if event_type in EVENT_DICT_CN: if isinstance(EVENT_DICT_CN[event_type], dict): msg_type = EVENT_DICT_CN[event_type][dealings_type] else: msg_type = EVENT_DICT_CN[event_type] else: msg_type = '未知事件类型 ' return msg_type else: if event_type in EVENT_DICT: if isinstance(EVENT_DICT[event_type], dict): msg_type = EVENT_DICT[event_type][dealings_type] else: msg_type = EVENT_DICT[event_type] else: msg_type = 'Unknown event type' return msg_type @staticmethod def get_msg_text(channel, n_time, lang, tz, event_type, electricity='', is_sys=0, dealings_type=0, ai_type=0, device_type=0, event_tag=''): """ 获取消息文本 @param: channel 通道号 @param: n_time 触发事件 @param: lang 语言 @param: tz 时区 @param: event_type 事件类型 @param: electricity 电量 @param: is_sys 是否系统消息 @param: dealings_type 往来类型 1 进 1 离开 @param: ai_type 设备本地AI只能算法 事件类型 @param: device_type 设备类型 @param: event_tag 设备算法事件标签 """ msg_type = '' event_type = int(event_type) device_type = int(device_type) event_list = [] if event_tag: event_list = [int(event) for event in event_tag.split(',') if event] if lang == 'cn': if event_type == 51: msg_type = '检测到画面变化' elif event_type == 57: msg_type = '有人出现' elif event_type == 58: msg_type = '有车出现' elif event_type == 59: msg_type = '有宠物出现' elif event_type == 60: msg_type = '发现人脸' elif event_type == 61: msg_type = '有异响' elif event_type == 62: msg_type = '区域闯入' elif event_type == 63: msg_type = '区域闯出' elif event_type == 64: msg_type = '有人徘徊' elif event_type == 65: msg_type = '长时间无人出现' elif event_type == 704: msg_type = '剩余电量 ' + electricity elif event_type == 702: msg_type = '摄像头休眠' elif event_type == 703: msg_type = '摄像头唤醒' elif event_type in [606, 607]: msg_type = '有人呼叫，请点击查看' elif event_type not in [606, 607] and ai_type > 0 and event_list: msg_type = ''.join([DevicePushService.get_event_type_text(lang, item, dealings_type) for item in event_list]) if is_sys: if device_type in MULTI_CHANNEL_TYPE_LIST: send_text = '{} 通道:{}'.format(msg_type, channel) else: send_text = msg_type else: if device_type in MULTI_CHANNEL_TYPE_LIST: send_text = '{} 通道:{}'.format(msg_type, channel) else: send_text = '{}'.format(msg_type) else: if event_type == 51: msg_type = 'Screen change detected' elif event_type == 57: msg_type = 'Person detected' elif event_type == 58: msg_type = 'Vehicle detected' elif event_type == 59: msg_type = 'Pet detected' elif event_type == 60: msg_type = 'Human face detected' elif event_type == 61: msg_type = 'Abnormal sound detected' elif event_type == 62: msg_type = 'Intrusion detected in the area' elif event_type == 63: msg_type = 'Area vacated' elif event_type == 64: msg_type = 'Loitering detected' elif event_type == 65: msg_type = 'No appearance for a long time' elif event_type == 704: msg_type = 'Battery remaining ' + electricity elif event_type == 702: msg_type = 'Camera sleep' elif event_type == 703: msg_type = 'Camera wake' elif event_type in [606, 607]: msg_type = 'Someone is calling, please click to view' elif event_type not in [606, 607] and ai_type > 0 and event_list: msg_type = ''.join([DevicePushService.get_event_type_text(lang, item, dealings_type) for item in event_list]) if is_sys: if device_type in MULTI_CHANNEL_TYPE_LIST: send_text = '{} channel:{}'.format(msg_type, channel) else: send_text = msg_type else: if device_type in MULTI_CHANNEL_TYPE_LIST: send_text = '{} channel:{}'.format(msg_type, channel) else: send_text = '{}'.format(msg_type) return send_text @staticmethod def do_xmpush(channel_id, uid, channel, appBundleId, token_val, event_type, n_time, msg_title, msg_text): """ android 国内小米APP消息提醒推送 """ url = 'https://api.xmpush.xiaomi.com/v3/message/regid' app_secret = XMPUSH_CONFIG[appBundleId] # payload = {'alert': 'Motion', 'msg': '', 'sound': 'sound.aif', 'zpush': '1', # 'received_at': n_time, 'event_time': n_time, 'event_type': event_type, # 'uid': uid, 'channel': channel # } data = { 'title': msg_title, 'description': msg_text, 'payload': 'payload', 'restricted_package_name': appBundleId, 'registration_id': token_val, 'extra.channel_id': channel_id, 'extra.alert': 'Motion', 'extra.msg': '', 'extra.sound': 'sound.aif', 'extra.zpush': '1', 'extra.received_at': n_time, 'extra.event_time': n_time, 'extra.event_type': event_type, 'extra.uid': uid, 'extra.channel': channel, } headers = { 'Authorization': 'key={}'.format(app_secret) } response = requests.post(url, data=data, headers=headers) if response.status_code == 200: LOGGING.info('uid:{},时间:{}小米推送结果：{}'.format(uid, n_time, response.json())) return response.json() @staticmethod def do_oppopush(channel_id, uid, channel, appBundleId, token_val, event_type, n_time, msg_title, msg_text, jg_token_val=''): """ android 国内oppo APP消息提醒推送 """ app_key = OPPOPUSH_CONFIG[appBundleId]['Key'] master_secret = OPPOPUSH_CONFIG[appBundleId]['Secret'] url = 'https://api.push.oppomobile.com/' now_time = str(round(time.time() * 1000)) # 1、实例化一个sha256对象 sha256 = hashlib.sha256() # 2、调用update方法进行加密 sha256.update((app_key + now_time + master_secret).encode('utf-8')) # 3、调用hexdigest方法,获取加密结果 sign = sha256.hexdigest() # 获取auth_token get_token_url = url + 'server/v1/auth' post_data = { 'app_key': app_key, 'sign': sign, 'timestamp': now_time } headers = {'Content-Type': 'application/x-www-form-urlencoded'} response = requests.post(get_token_url, data=post_data, headers=headers) result = response.json() # 发送推送 push_url = url + 'server/v1/message/notification/unicast' extra_data = {'alert': 'Motion', 'msg': '', 'sound': 'sound.aif', 'zpush': '1', 'received_at': n_time, 'event_time': n_time, 'event_type': event_type, 'uid': uid, 'channel': channel} message = { "target_type": 2, "target_value": token_val, "notification": { "title": msg_title, "content": msg_text, 'channel_id': channel_id, 'action_parameters': extra_data, 'click_action_type': 4, 'click_action_activity': 'com.ansjer.zccloud_a.AJ_MainView.AJ_Home.AJMainActivity' } } push_data = { 'auth_token': result['data']['auth_token'], 'message': json.dumps(message) } response = requests.post(push_url, data=push_data, headers=headers) if response.status_code == 200: LOGGING.info("oppo推送返回值:{},uid:{},time:{},event:{}".format(response.json(), uid, now_time, event_type)) if event_type in [606, 607]: PushObject.jpush_transparent_transmission(msg_title, msg_text, appBundleId, jg_token_val, extra_data) return response.json() @classmethod def async_send_picture_push(cls, push_type, aws_s3_client, bucket, key, uid, appBundleId, token_val, event_type, n_time, msg_title, msg_text, channel): """ 异步推送图片 """ try: image_url = aws_s3_client.generate_presigned_url( 'get_object', Params={'Bucket': bucket, 'Key': key}, ExpiresIn=3600) push_result = False if push_type == 0: push_result = PushObject.ios_apns_push( uid, appBundleId, token_val, n_time, event_type, msg_title, msg_text, uid, channel, image_url) elif push_type == 1: push_result = PushObject.android_fcm_push( uid, appBundleId, token_val, n_time, event_type, msg_title, msg_text, uid, channel, image_url) elif push_type == 3: huawei_push_object = HuaweiPushObject() push_result = huawei_push_object.send_push_notify_message( token_val=token_val, msg_title=msg_title, msg_text=msg_text, uid=uid, event_type=event_type, n_time=n_time, image_url=image_url) LOGGING.info('{}推送图片,push_type:{},推送结果:{}'.format(uid, push_type, push_result)) except Exception as e: LOGGING.info('异步推送图片异常,error_line:{},error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e))) @staticmethod def get_res_data(**kwargs): """ 获取响应数据 @return: res_data """ res_data = {'code': 0, 'msg': 'success'} is_st = kwargs['is_st'] if is_st == 0 or is_st == 2: res_data['msg'] = 'success 0 or 2' elif is_st == 1: key_name = '{}/{}/{}.jpeg'.format(kwargs['uid'], kwargs['channel'], kwargs['n_time']) params = {'Key': key_name} if kwargs['region'] == 2: # 2:国内 params['Bucket'] = 'push' else: # 1：国外 params['Bucket'] = 'foreignpush' img_url = DevicePushService.generate_s3_url(kwargs['aws_s3_client'], params) res_data['img_push'] = img_url res_data['msg'] = 'success 1' elif is_st == 3: img_url_list = [] if kwargs['region'] == 2: # 2:国内 params = {'Bucket': 'push'} else: # 1：国外 params = {'Bucket': 'foreignpush'} for i in range(kwargs['is_st']): key_name = '{}/{}/{}_{}.jpeg'.format(kwargs['uid'], kwargs['channel'], kwargs['n_time'], i) params['Key'] = key_name img_url = DevicePushService.generate_s3_url(kwargs['aws_s3_client'], params) img_url_list.append(img_url) res_data['img_url_list'] = img_url_list res_data['msg'] = 'success 3' return res_data @staticmethod def generate_s3_url(aws_s3_client, params): """ 获取S3对象URL """ response_url = aws_s3_client.generate_presigned_url( ClientMethod='put_object', Params=params, ExpiresIn=3600 ) return response_url @staticmethod def check_share_permission(user_id, channel, uid): """ 检查用户是否有权限接收设备报警推送 """ user_permission_qs = DeviceChannelUserSet.objects.filter(user_id=user_id, uid=uid) \ .values('id', 'channels') # 根据当前用户与uid查询是否设置过通道权限，不存在则不是分享设备 if not user_permission_qs.exists(): return True up_id = user_permission_qs[0]['id'] channels = user_permission_qs[0]['channels'] channels_list = [int(val) for val in channels.split(',')] # 当前uid是属于分享设备并且设置了权限 # 判断通道是否设置了权限，不存在则当前通道没有权限接受消息推送 if int(channel) not in channels_list: return False permission_qs = DeviceSharePermission.objects.filter(code='AlarmMessages').values('id') p_id = permission_qs[0]['id'] # 当前通道存在设置则查看是否有 消息推送权限 channel_permission_qs = DeviceChannelUserPermission.objects \ .filter(channel_user_id=up_id, permission_id=p_id) \ .values('permission_id', 'channel_user_id') if not channel_permission_qs.exists(): return False return True @classmethod def is_algorithm_type(cls, uid, event_type): """ 判断是否是算法类型 62、63、64、65、66不限制推送 """ uid_set_qs = UidSetModel.objects.filter(uid=uid).values('ai_type') if not uid_set_qs.exists(): return False if uid_set_qs[0]['ai_type'] == 0: return False event_types = [62, 63, 64, 65, 66] event_res = DEVICE_EVENT_TYPE.get(event_type, 0) if event_res in event_types: return True event_types2 = cls.get_combo_types(event_type) if not event_types2: return False c = [x for x in event_types if x in event_types2] return True if c else False @staticmethod def is_send_app_push(event_type, event_tag, app_push_config, msg_interval=None, uid=None): """ 是否进行APP消息提醒 @return: True|False """ try: if not app_push_config: return True is_push = app_push_config['appPush'] if is_push != 1: # 1：进行APP提醒，其它则不执行APP提醒 return False if msg_interval: # 存在消息间隔数据缓存 不推送APP消息 return False all_day = app_push_config['pushTime']['allDay'] # 允许设备类型APP提醒列表 app_event_types = app_push_config['eventTypes']['device'] if all_day == 0: # 1：全天提醒，0：自定义时间提醒 push_time_config = app_push_config['pushTime'] # 计算当前时间是否在自定义消息提醒范围内 if not DevicePushService.is_push_notify_allowed_now(push_time_config): LOGGING.info('{}APP推送提醒不在自定义时间内：{}'.format(uid, push_time_config)) return False # APP接收提醒，判断识别类型是否勾选提醒 push_result = DevicePushService.is_type_push(event_type, event_tag, app_event_types) LOGGING.info('{}APP推送消息类型提醒是否执行：{}'.format(uid, push_result)) return push_result except Exception as e: LOGGING.info('{}判断是否执行APP推送异常,errLine:{}, errMsg:{}'.format(uid, e.__traceback__.tb_lineno, repr(e))) return True @staticmethod def is_type_push(event_type, event_tag, app_event_types): # 检查事件标签和应用事件类型是否都存在 if event_tag and app_event_types: # 将事件标签按逗号分割成列表，并转换为整数类型 tag_list = [int(event) for event in event_tag.split(',') if event] # 判断是否有任一标签允许应用提醒 return any(item in app_event_types for item in tag_list) # 检查事件类型和用户所选事件类型是否都存在，并判断事件类型在用户所选事件类型列表中 return event_type and app_event_types and event_type in app_event_types @staticmethod def is_push_notify_allowed_now(push_time_config): """ 判断当前时间是否在允许APP推送提醒 """ now_time = int(time.time()) start_time = push_time_config['startTime'] end_time = push_time_config['endTime'] repeat = push_time_config['repeat'] tz = push_time_config['timeZone'] # 获取当前日期和周几 now_date, week = DevicePushService.get_now_date_and_week(now_time, tz) # 判断是否在重复日范围内 if not DevicePushService.is_repeated(week, repeat): return False # 计算当前日期在一天中的秒数 seconds = LocalDateTimeUtil.convert_time_to_seconds(now_date) # 判断是否在APP推送提醒范围内 return DevicePushService.is_in_effect(start_time, end_time, seconds) @staticmethod def is_in_effect(start, end, now_seconds): """ 判断是否在提醒时间范围内 @params: 开始时间秒 @params: 结束时间秒 @params: 当前时间秒 @return: 当前时间是在范围内返回True 否则False """ if start < end: return start <= now_seconds <= end else: return start <= now_seconds or now_seconds <= end @staticmethod def is_repeated(week_day, repeat_day): """ 判断是否重复日 @params: week_day 周几 @params: 重复日1-127 @return: 如果当前日期在重复日则返回True否则False """ # 判断对应位置上的值是否为 1 is_repeat = (repeat_day >> (week_day - 1)) & 1 == 1 return is_repeat @staticmethod def get_now_date_and_week(now_time, tz): now_data = CommonService.get_now_time_str(now_time, tz, 'cn') week = LocalDateTimeUtil.date_to_week(now_data) return now_data, week