import json import logging import os import threading import time import apns2 import boto3 import botocore import jpush as jpush from botocore import client from django.http import JsonResponse from django.views.generic.base import View from pyfcm import FCMNotification from AnsjerPush.config import AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY from AnsjerPush.config import JPUSH_CONFIG, FCM_CONFIG, APNS_CONFIG, BASE_DIR, APNS_MODE from Model.models import UidPushModel, SysMsgModel 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 from Service.GatewayService import GatewayPushService # 移动侦测V2接口 class NotificationV2View(View): def get(self, request, *args, **kwargs): request.encoding = 'utf-8' return self.validation(request.GET) def post(self, request, *args, **kwargs): request.encoding = 'utf-8' return self.validation(request.POST) def validation(self, request_dict): logger = logging.getLogger('info') logger.info("移动侦测V2接口参数:{}".format(request_dict)) uidToken = request_dict.get('uidToken', None) etk = request_dict.get('etk', None) channel = request_dict.get('channel', '1') n_time = request_dict.get('n_time', None) event_type = request_dict.get('event_type', None) is_st = request_dict.get('is_st', None) region = request_dict.get('region', None) electricity = request_dict.get('electricity', '') if not all([channel, n_time]): return JsonResponse(status=200, data={ 'code': 444, 'msg': 'param is wrong'}) if not region or not is_st: return JsonResponse(status=200, data={'code': 404, 'msg': 'no region or is_st'}) try: is_st = int(is_st) region = int(region) # 解密获取uid if etk: eto = ETkObject(etk) uid = eto.uid else: uto = UidTokenObject(uidToken) uid = uto.UID # uid = request_dict.get('uid', None) # 调试 # 判断uid长度 if len(uid) != 20 and len(uid) != 14: return JsonResponse(status=200, data={'code': 404, 'msg': 'wrong uid'}) logger.info('调用推送接口的uid：{}'.format(uid)) pkey = '{uid}_{channel}_{event_type}_ptl'.format(uid=uid, channel=channel, event_type=event_type) ykey = '{uid}_redis_qs'.format(uid=uid) is_sys_msg = self.is_sys_msg(int(event_type)) if is_sys_msg: dkey = '{uid}_{channel}_{event_type}_flag'.format(uid=uid, channel=channel, event_type=event_type) else: dkey = '{uid}_{channel}_flag'.format(uid=uid, channel=channel) redisObj = RedisObject(db=6) have_ykey = redisObj.get_data(key=ykey) # uid_set 数据库缓存 have_pkey = redisObj.get_data(key=pkey) # 一分钟限制key have_dkey = redisObj.get_data(key=dkey) # 推送消息时间间隔 logger.info('ykey：{}, pkey: {}, dkey: {}'.format(have_ykey, have_pkey, have_dkey)) # 一分钟内不推送 if have_pkey: return JsonResponse(status=200, data={'code': 0, 'msg': 'Push again in one minute'}) redisObj.set_data(key=pkey, val=1, expire=60) # 查询推送数据 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__new_detect_interval') if not uid_push_qs.exists(): logger.info('uid_push 数据不存在') return JsonResponse(status=200, data={'code': 176, 'msg': 'no uid_push data'}) ai_type = uid_push_qs.first()['uid_set__ai_type'] event_type = self.get_combo_msg_type(ai_type, int(event_type)) redis_list = [] for qs in uid_push_qs: redis_list.append(qs) # 修改redis数据，并设置过期时间为10分钟 redisObj.set_data(key=ykey, val=str(redis_list), expire=600) nickname = redis_list[0]['uid_set__nickname'] detect_interval = redis_list[0]['uid_set__detect_interval'] if not nickname: nickname = uid if not have_dkey: # 设置推送消息的时间间隔 new_detect_interval = redis_list[0]['uid_set__new_detect_interval'] detect_interval = new_detect_interval if new_detect_interval > 0 else detect_interval detect_interval = 60 if detect_interval < 60 else detect_interval redisObj.set_data(key=dkey, val=1, expire=detect_interval-5) logger.info('APP消息推送间隔：{}s'.format(detect_interval)) if is_st == 1 or is_st == 3: # 使用aws s3 aws_s3_client = s3_client(region=region) bucket = 'foreignpush' if region == 1 else 'push' kwag_args = { 'uid': uid, 'channel': channel, 'event_type': event_type, 'n_time': n_time, } sys_msg_list = [] userID_ids = [] do_apns_code = '' do_fcm_code = '' do_jpush_code = '' logger.info('进入手机推送------') logger.info('uid={}'.format(uid)) logger.info(redis_list) new_device_info_list = [] local_date_time = '' for up in redis_list: push_type = up['push_type'] appBundleId = up['appBundleId'] token_val = up['token_val'] lang = up['lang'] tz = up['tz'] if tz is None or tz == '': tz = 0 # 发送标题 msg_title = self.get_msg_title(appBundleId=appBundleId, nickname=nickname) # 发送内容 msg_text = self.get_msg_text(channel=channel, n_time=n_time, lang=lang, tz=tz, event_type=event_type, electricity=electricity) kwag_args['appBundleId'] = appBundleId kwag_args['token_val'] = token_val kwag_args['msg_title'] = msg_title kwag_args['msg_text'] = msg_text logger.info('推送要的数据: {}'.format(kwag_args)) local_date_time = CommonService.get_now_time_str(n_time=n_time, tz=tz, lang='cn') logger.info('<<<<<根据时区计算后日期={}，时区={}'.format(local_date_time, tz)) local_date_time = local_date_time[0:10] logger.info('<<<<<切片后的日期={}'.format(local_date_time)) # 以下是存库 userID_id = up["userID_id"] if userID_id not in userID_ids: now_time = int(time.time()) if is_sys_msg: sys_msg_text = self.get_msg_text(channel=channel, n_time=n_time, lang=lang, tz=tz, event_type=event_type, electricity=electricity, is_sys=1) sys_msg_list.append(SysMsgModel( userID_id=userID_id, msg=sys_msg_text, addTime=now_time, updTime=now_time, uid=uid, eventType=event_type)) else: # start 根据设备侦测时间为准进行分表存储数据 logger.info('分表存数据start------') new_device_info_list.append(EquipmentInfoService.get_equipment_info_obj( local_date_time, device_user_id=userID_id, event_time=n_time, event_type=event_type, device_uid=uid, device_nick_name=nickname, channel=channel, alarm='Motion \tChannel:{channel}'.format(channel=channel), is_st=is_st, receive_time=n_time, add_time=now_time, storage_location=2, border_coords='', )) # end userID_ids.append(userID_id) try: # 推送消息 if not have_dkey: logger.info('APP准备推送:{}, {}'.format(uid, request_dict)) if (is_st == 1 or is_st == 3) and (push_type == 0 or push_type == 1): # 推送显示图片 if is_st == 1: key = '{}/{}/{}.jpeg'.format(uid, channel, n_time) else: key = '{}/{}/{}_0.jpeg'.format(uid, channel, n_time) push_thread = threading.Thread(target=self.push_thread_test, args=( push_type, aws_s3_client, bucket, key, uid, appBundleId, token_val, event_type, n_time, msg_title, msg_text, channel)) push_thread.start() else: if push_type == 0: # ios apns do_apns_code = self.do_apns(**kwag_args) elif push_type == 1: # android gcm do_fcm_code = self.do_fcm(**kwag_args) elif push_type == 2: # android jpush do_jpush_code = self.do_jpush(**kwag_args) except Exception as e: logger.info( "errLine={errLine}, errMsg={errMsg}".format(errLine=e.__traceback__.tb_lineno, errMsg=repr(e))) continue if is_sys_msg: SysMsgModel.objects.bulk_create(sys_msg_list) else: # new 分表批量存储 设备信息 if new_device_info_list and len(new_device_info_list) > 0: # 根据日期获得星期几 week = LocalDateTimeUtil.date_to_week(local_date_time) EquipmentInfoService.equipment_info_bulk_create(week, new_device_info_list) logger.info('设备信息分表批量保存end------') if is_st == 0 or is_st == 2: for up in redis_list: if up['push_type'] == 0: # ios apns up['do_apns_code'] = do_apns_code elif up['push_type'] == 1: # android gcm up['do_fcm_code'] = do_fcm_code elif up['push_type'] == 2: # android jpush up['do_jpush_code'] = do_jpush_code del up['push_type'] del up['userID_id'] del up['userID__NickName'] del up['lang'] del up['tz'] del up['uid_set__nickname'] del up['uid_set__detect_interval'] del up['uid_set__detect_group'] return JsonResponse(status=200, data={'code': 0, 'msg': 'success 0 or 2', 're_list': redis_list}) elif is_st == 1: thumbspng = '{uid}/{channel}/{filename}.jpeg'.format(uid=uid, channel=channel, filename=n_time) Params = {'Key': thumbspng} if region == 2: # 2:国内 Params['Bucket'] = 'push' else: # 1：国外 Params['Bucket'] = 'foreignpush' response_url = generate_s3_url(aws_s3_client, Params) for up in redis_list: up['do_apns_code'] = do_apns_code up['do_fcm_code'] = do_fcm_code up['do_jpush_code'] = do_jpush_code del up['push_type'] del up['userID_id'] del up['userID__NickName'] del up['lang'] del up['tz'] del up['uid_set__nickname'] del up['uid_set__detect_interval'] del up['uid_set__detect_group'] res_data = {'code': 0, 'img_push': response_url, 'msg': 'success'} return JsonResponse(status=200, data=res_data) elif is_st == 3: img_url_list = [] if region == 2: # 2:国内 Params = {'Bucket': 'push'} else: # 1：国外 Params = {'Bucket': 'foreignpush'} for i in range(is_st): thumbspng = '{uid}/{channel}/{filename}_{st}.jpeg'. \ format(uid=uid, channel=channel, filename=n_time, st=i) Params['Key'] = thumbspng response_url = generate_s3_url(aws_s3_client, Params) img_url_list.append(response_url) for up in redis_list: up['do_apns_code'] = do_apns_code up['do_fcm_code'] = do_fcm_code up['do_jpush_code'] = do_jpush_code del up['push_type'] del up['userID_id'] del up['userID__NickName'] del up['lang'] del up['tz'] del up['uid_set__nickname'] del up['uid_set__detect_interval'] del up['uid_set__detect_group'] res_data = {'code': 0, 'img_url_list': img_url_list, 'msg': 'success 3'} return JsonResponse(status=200, data=res_data) except Exception as e: logger.info('移动侦测接口异常: {}'.format(e)) logger.info('错误文件', e.__traceback__.tb_frame.f_globals['__file__']) logger.info('错误行号', e.__traceback__.tb_lineno) data = { 'errLine': e.__traceback__.tb_lineno, 'errMsg': repr(e), } return JsonResponse(status=200, data=json.dumps(data), safe=False) @classmethod def get_combo_msg_type(cls, ai_type, event_type): """ 获取组合类型，ai_type == 47 支持算法小店，需判断组合类型 """ logger = logging.getLogger('info') try: if ai_type != 47: return event_type logger.info('LOG------算法小店组合类型十进制值：{}'.format(event_type)) # 如触发一个事件，则匹配已用类型 1替换后变成51代表移动侦测 1：移动侦测,2：人形,4：车型,8：人脸 event_dict = { 1: 51, 2: 57, 4: 58, 16: 59, 8: 60, 32: 61 } event_val = event_dict.get(event_type, 0) # event_val == 0 没有匹配到单个值则认为组合类型 # 如是3，则转为二进制11，代表（1+2）触发了移动侦测+人形侦测 if event_val == 0: val = cls.dec_to_bin(event_type) return int(val) else: return int(event_val) except Exception as e: logger.info('推送错误异常,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 def push_thread_test(self, push_type, aws_s3_client, bucket, key, uid, appBundleId, token_val, event_type, n_time, msg_title, msg_text, channel): logger = logging.getLogger('info') logger.info('推送图片测试:{} {} {} {} {} {} {} {}'.format(push_type, uid, appBundleId, token_val, event_type, n_time, msg_title, msg_text)) try: image_url = aws_s3_client.generate_presigned_url('get_object', Params={'Bucket': bucket, 'Key': key}, ExpiresIn=300) logger.info('推送图片url:{}'.format(image_url)) if push_type == 0: GatewayPushService.ios_apns_push(uid, appBundleId, token_val, n_time, event_type, msg_title, msg_text, uid, channel, image_url) elif push_type == 1: GatewayPushService.android_fcm_push(uid, appBundleId, token_val, n_time, event_type, msg_title, msg_text, uid, channel, image_url) except Exception as e: logger.info('推送图片测试异常:{}'.format(e)) def get_msg_title(self, appBundleId, nickname): package_title_config = { 'com.ansjer.customizedd_a': 'DVS', 'com.ansjer.zccloud_a': 'ZosiSmart', 'com.ansjer.zccloud_ab': '周视', 'com.ansjer.adcloud_a': 'ADCloud', 'com.ansjer.adcloud_ab': 'ADCloud', 'com.ansjer.accloud_a': 'ACCloud', 'com.ansjer.loocamccloud_a': 'Loocam', 'com.ansjer.loocamdcloud_a': 'Anlapus', 'com.ansjer.customizedb_a': 'COCOONHD', 'com.ansjer.customizeda_a': 'Guardian365', 'com.ansjer.customizedc_a': 'PatrolSecure', } if appBundleId in package_title_config.keys(): return package_title_config[appBundleId] + '(' + nickname + ')' else: return nickname def is_sys_msg(self, event_type): event_type_list = [702, 703, 704] if event_type in event_type_list: return True return False def get_msg_text(self, channel, n_time, lang, tz, event_type, electricity='', is_sys=0): n_date = CommonService.get_now_time_str(n_time=n_time, tz=tz, lang=lang) etype = int(event_type) if lang == 'cn': if etype == 704: msg_type = '剩余电量:' + electricity elif etype == 702: msg_type = '摄像头休眠' elif etype == 703: msg_type = '摄像头唤醒' else: msg_type = '' if is_sys: send_text = '{msg_type} 通道:{channel}'.format(msg_type=msg_type, channel=channel) else: send_text = '{msg_type} 通道:{channel} 日期:{date}'.format(msg_type=msg_type, channel=channel, date=n_date) else: if etype == 704: msg_type = 'Battery remaining:' + electricity elif etype == 702: msg_type = 'Camera sleep' elif etype == 703: msg_type = 'Camera wake' else: msg_type = '' if is_sys: send_text = '{msg_type} channel:{channel}'. \ format(msg_type=msg_type, channel=channel) else: send_text = '{msg_type} channel:{channel} date:{date}'. \ format(msg_type=msg_type, channel=channel, date=n_date) return send_text def do_jpush(self, uid, channel, appBundleId, token_val, event_type, n_time, msg_title, msg_text): app_key = JPUSH_CONFIG[appBundleId]['Key'] master_secret = JPUSH_CONFIG[appBundleId]['Secret'] _jpush = jpush.JPush(app_key, master_secret) push = _jpush.create_push() push.audience = jpush.registration_id(token_val) push_data = {"alert": "Motion ", "event_time": n_time, "event_type": event_type, "msg": "", "received_at": n_time, "sound": "sound.aif", "uid": uid, "zpush": "1", "channel": channel} android = jpush.android(alert=msg_text, priority=1, style=1, alert_type=7, big_text=msg_text, title=msg_title, extras=push_data) push.notification = jpush.notification(android=android) push.platform = jpush.all_ res = push.send() print(res) return res.status_code def do_fcm(self, uid, channel, appBundleId, token_val, event_type, n_time, msg_title, msg_text): logger = logging.getLogger('info') try: serverKey = FCM_CONFIG[appBundleId] except Exception as e: logger.info('------fcm_error:{}'.format(repr(e))) return 'serverKey abnormal' push_service = FCMNotification(api_key=serverKey) data = {"alert": "Motion ", "event_time": n_time, "event_type": event_type, "msg": "", "received_at": n_time, "sound": "sound.aif", "uid": uid, "zpush": "1", "channel": channel} result = push_service.notify_single_device(registration_id=token_val, message_title=msg_title, message_body=msg_text, data_message=data, extra_kwargs={ 'default_vibrate_timings': True, 'default_sound': True, 'default_light_settings': True }) return result def do_apns(self, uid, channel, appBundleId, token_val, event_type, n_time, msg_title, msg_text): logger = logging.getLogger('info') logger.info("进来do_apns函数了") logger.info(token_val) logger.info(APNS_MODE) logger.info(os.path.join(BASE_DIR, APNS_CONFIG[appBundleId]['pem_path'])) try: cli = apns2.APNSClient(mode=APNS_MODE, client_cert=os.path.join(BASE_DIR, APNS_CONFIG[appBundleId]['pem_path'])) push_data = {"alert": "Motion ", "event_time": n_time, "event_type": event_type, "msg": "", "received_at": n_time, "sound": "", "uid": uid, "zpush": "1", "channel": channel} alert = apns2.PayloadAlert(body=msg_text, title=msg_title) payload = apns2.Payload(alert=alert, custom=push_data, sound="default") # return uid, channel, appBundleId, str(token_val), event_type, n_time, msg_title,msg_text n = apns2.Notification(payload=payload, priority=apns2.PRIORITY_LOW) res = cli.push(n=n, device_token=token_val, topic=appBundleId) print(res.status_code) logger.info("apns_推送状态：") logger.info(res.status_code) if res.status_code == 200: return res.status_code else: print('apns push fail') print(res.reason) logger.info('apns push fail') logger.info(res.reason) return res.status_code except (ValueError, ArithmeticError): return 'The program has a numeric format exception, one of the arithmetic exceptions' except Exception as e: print(repr(e)) print('do_apns函数错误行号', e.__traceback__.tb_lineno) logger.info('do_apns错误：{}'.format(repr(e))) return repr(e) def s3_client(region): if region == 2: # 国内 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' ) else: # 国外 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' ) return aws_s3_client def generate_s3_url(aws_s3_client, Params): response_url = aws_s3_client.generate_presigned_url( ClientMethod='put_object', Params=Params, ExpiresIn=3600 ) return response_url