测试推送

AnsjerPush/urls.py

@@ -15,7 +15,7 @@ Including another URLconf
 """
 from django.contrib import admin
 from django.urls import path
-from Controller import DetectController, ShadowController, TestDetectController, EquipmentInfo, DetectControllerV2, AiController
+from Controller import DetectController, ShadowController, TestDetectController, EquipmentInfo, DetectControllerV2, AiController, TestController
 from django.conf.urls import url
 
 urlpatterns = [
@@ -31,4 +31,7 @@ urlpatterns = [
     path('notifyV2/push', DetectControllerV2.NotificationView.as_view()),
     url(r'^notifyV2/(?P<operation>.*)$', DetectControllerV2.NotificationView.as_view()),
     url(r'^AiService/(?P<operation>.*)$', AiController.AiView.as_view()),
+
+    url('testDetect/testJpush', TestController.testDetect),
+    url('testDetect/testApns', TestController.testDetectApns),
 ]

Controller/TestController.py

@@ -632,3 +632,64 @@ class NotificationView(View):
                 llt = redisObject.get_ttl(tmpDKey)
                 if llt > detect_interval:
                     redisObject.set_data(key=tmpDKey, val=1, expire=detect_interval)
+
+def testDetect(request):
+    app_key = JPUSH_CONFIG['com.ansjer.loocamdcloud_ab']['Key']
+    master_secret = JPUSH_CONFIG['com.ansjer.loocamdcloud_ab']['Secret']
+    # 此处换成各自的app_key和master_secre
+    _jpush = jpush.JPush(app_key, master_secret)
+    push = _jpush.create_push()
+    # if you set the logging level to "DEBUG",it will show the debug logging.
+    # _jpush.set_logging("DEBUG")
+    # push.audience = jpush.all_
+    push.audience = jpush.registration_id('190e35f7e0b60afd5af')
+    push_data = {"alert": "Motion ", "event_time": int(time.time()), "event_type": 1, "msg": "",
+                 "received_at": int(time.time()), "sound": "sound.aif", "uid": 'test', "zpush": "1", "channel": 1}
+    android = jpush.android(alert='test', priority=1, style=1, alert_type=7,
+                            big_text='test', title='test',
+                            extras={'hello':'test'})
+    push.notification = jpush.notification(android=android)
+    push.platform = jpush.all_
+    res = push.send()
+    print(res)
+    return res.status_code
+
+def testDetectApns(request):
+    logger = logging.getLogger('info')
+    logger.info("进来do_apns函数了")
+    logger.info(os.path.join(BASE_DIR, APNS_CONFIG['com.ansjer.customizede']['pem_path']))
+    try:
+        cli = apns2.APNSClient(mode=APNS_MODE,
+                               client_cert=os.path.join(BASE_DIR, APNS_CONFIG['com.ansjer.customizede']['pem_path']))
+
+        push_data = {"alert": "Motion ", "event_time": int(time.time()), "event_type": 1, "msg": "",
+                     "received_at": int(time.time()), "sound": "sound.aif", "uid": 111, "zpush": "1", "channel": 1}
+        alert = apns2.PayloadAlert(body='hello', title='title')
+        payload = apns2.Payload(alert=alert, custom=push_data)
+
+        # return uid, channel, 'com.ansjer.customizede', 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='a0bff8dbd4fb50b888236c3088e4653d79c4e69385716357b2849e8e674072ab', topic='com.ansjer.customizede')
+        print(res.status_code)
+        logger.info("推送状态：")
+        logger.info(res.status_code)
+
+        #     200, 推送成功。
+        # 　　400, 请求有问题。
+        # 　　403, 证书或Token有问题。
+        # 　　405, 请求方式不正确, 只支持POST请求
+        # 　　410, 设备的Token与证书不一致
+        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))
+        logger.info(repr(e))
+        return repr(e)