ASJPush/Controller/TransparentTransmissionPushController.py

# @Author    : peng
# @File      : TransparentTransmissionPushController.py
# @Time      : 2024年7月9日17:22:19
import json
import threading
import time
import requests
from django.http import HttpResponse
from django.views import View

from Model.models import UidPushModel
from Object.ResponseObject import ResponseObject
from Service.PushService import PushObject
from AnsjerPush.config import XM_PUSH_CHANNEL_ID, HONORPUSH_CONFIG, LOGGER
from Service.HuaweiPushService.push_admin import messaging


class TransparentTransmissionPushView(View):

    def get(self, request, *args, **kwargs):
        request.encoding = 'utf-8'
        operation = kwargs.get('operation')
        return self.validation(request.GET, operation)

    def post(self, request, *args, **kwargs):
        request.encoding = 'utf-8'
        operation = kwargs.get('operation')
        return self.validation(request.POST, operation)

    def validation(self, request_dict, operation):
        response = ResponseObject()
        if operation == 'logout-push':  # 强制退出推送
            return self.logout_push(request_dict, response)
        elif operation == 'del-device-push':  # 删除设备推送
            return self.del_device_push(request_dict, response)

    @staticmethod
    def logout_push(request_dict, response):
        LOGGER.info('进入登出推送接口，参数：{}'.format(request_dict))
        push_token = request_dict.get('push_token', None)
        user_id = request_dict.get('user_id', None)
        if not all([push_token, user_id]):
            return response.json(444)
        try:
            uid_push_qs = UidPushModel.objects.filter(userID=user_id).exclude(token_val=push_token).values(
                'token_val', 'appBundleId', 'push_type', 'jg_token_val').distinct()
            if not uid_push_qs.exists():
                return response.json(173)
            LOGGER.info('推送列表:{}'.format(uid_push_qs))
            push_thread = threading.Thread(target=TransparentTransmissionPushView.thread_push,
                                           args=(uid_push_qs, user_id, '强制登出', '强制登出', 705))
            push_thread.start()
            # 删除推送信息
            # UidPushModel.objects.filter(userID=user_id).exclude(token_val=push_token).delete()
            return response.json(0)
        except Exception as e:
            return HttpResponse(repr(e), status=500)

    @staticmethod
    def del_device_push(request_dict, response):
        uid = request_dict.get('uid', None)
        if not all([uid]):
            return response.json(444)
        try:
            uid_push_qs = UidPushModel.objects.filter(uid_set_uid=uid).values(
                'token_val', 'appBundleId', 'push_type', 'jg_token_val')
            if not uid_push_qs.exists():
                return response.json(173)
            push_thread = threading.Thread(target=TransparentTransmissionPushView.thread_push,
                                           args=(uid_push_qs, uid, '删除设备', '删除设备'))
            push_thread.start()
            return response.json(0)
        except Exception as e:
            return HttpResponse(repr(e), status=500)

    @staticmethod
    def thread_push(qs, user_id, title, content, event_type):
        now_time = int(time.time())
        for push_item in qs:
            kwargs = {
                'nickname': user_id,
                'app_bundle_id': push_item['appBundleId'],
                'token_val': push_item['token_val'],
                'n_time': now_time,
                'event_type': event_type,
                'msg_title': title,
                'msg_text': content,
            }
            if push_item['push_type'] == 0:  # 苹果
                PushObject.ios_apns_push(**kwargs)
            elif push_item['push_type'] == 1:  # 谷歌
                PushObject.android_fcm_push_v1(**kwargs)
            elif push_item['push_type'] == 2:  # 极光
                PushObject.jpush_transparent_transmission(title, content, push_item['appBundleId'],
                                                          push_item['token_val'], kwargs)

            elif push_item['push_type'] == 3:  # 华为
                TransparentTransmissionPushView.huawei_transparent_transmission(user_id=user_id, **kwargs)
            elif push_item['push_type'] == 4:  # 小米
                channel_id = XM_PUSH_CHANNEL_ID['device_reminder']
                PushObject.android_xmpush(channel_id=channel_id, **kwargs)
            elif push_item['push_type'] in [5, 6]:  # vivo, oppo
                kwargs['token_val'] = push_item['jg_token_val']
                PushObject.jpush_transparent_transmission(title, content, push_item['appBundleId'],
                                                          push_item['jg_token_val'], kwargs)

            elif push_item['push_type'] == 7:  # 魅族
                PushObject.android_meizupush(**kwargs)
            elif push_item['push_type'] == 8:  # 荣耀
                TransparentTransmissionPushView.honor_transparent_transmission(**kwargs)

    @staticmethod
    def huawei_transparent_transmission(nickname, app_bundle_id, token_val, n_time, event_type, msg_title,
                                        msg_text, user_id):
        """
        发送透传推送
        @param nickname:
        @param app_bundle_id:
        @param event_type:
        @param n_time:
        @param token_val:
        @param msg_title:
        @param msg_text:
        @param user_id:
        @return: None
        """
        data = {
            'nickname': nickname, 'event_type': event_type, 'event_time': n_time, 'msg_title': msg_title,
            'msg_text': msg_text
        }
        data = json.dumps(data)
        android = messaging.AndroidConfig(
            collapse_key=-1,
            urgency=messaging.AndroidConfig.HIGH_PRIORITY,
            ttl='10000s',
            bi_tag='the_sample_bi_tag_for_receipt_service'
        )

        message = messaging.Message(
            data=data,
            android=android,
            token=[token_val]
        )

        try:
            import certifi
            response = messaging.send_message(message, verify_peer=certifi.where())
            LOGGER.info('{}退出登录，华为透传推送响应: {}'.format(user_id, json.dumps(vars(response))))
            assert (response.code == '80000000')
        except Exception as e:
            LOGGER.info('退出登录，华为透传推送异常: {}'.format(repr(e)))

    @staticmethod
    def honor_transparent_transmission(token_val, n_time, event_type, msg_title, msg_text, app_bundle_id, nickname=''):
        """
        android honor 推送
        @param nickname: 设备昵称
        @param app_bundle_id: app包id
        @param token_val: 推送token
        @param n_time: 当前时间
        @param event_type: 事件类型
        @param msg_title: 推送标题
        @param msg_text: 推送内容
        @return: bool
        """
        try:
            client_id = HONORPUSH_CONFIG[app_bundle_id]['client_id']
            client_secret = HONORPUSH_CONFIG[app_bundle_id]['client_secret']
            app_id = HONORPUSH_CONFIG[app_bundle_id]['app_id']
            get_access_token_url = 'https://iam.developer.hihonor.com/auth/token'
            post_data = {
                'grant_type': 'client_credentials',
                'client_id': client_id,
                'client_secret': client_secret
            }
            headers = {'Content-Type': 'application/x-www-form-urlencoded'}
            access_token_response = requests.post(get_access_token_url, data=post_data, headers=headers)
            access_result = access_token_response.json()
            authorization_token = 'Bearer ' + access_result['access_token']
            # 发送推送
            push_url = 'https://push-api.cloud.hihonor.com/api/v1/{}/sendMessage'.format(app_id)
            headers = {'Content-Type': 'application/json', 'Authorization': authorization_token,
                       'timestamp': str(int(time.time()) * 1000)}
            extra_data = {'alert': 'Motion', 'msg': '', 'sound': 'sound.aif', 'zpush': '1',
                          'received_at': n_time, 'event_time': n_time, 'event_type': str(event_type),
                          'nickname': nickname, 'title': msg_title, 'body': msg_text}
            # 透传推送
            push_data = {
                "data": json.dumps(extra_data),
                "token": [token_val]
            }
            response = requests.post(push_url, json=push_data, headers=headers)
            LOGGER.info("用户:{},时间:{}，荣耀透传推送返回值:{}".format(nickname, n_time, response.json()))
            return True
        except Exception as e:
            LOGGER.info("荣耀推送异常:error_line:{},error_msg:{}".format(e.__traceback__.tb_lineno, repr(e)))
            return False