ASJServer/Controller/IotCoreController.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import hashlib
import logging
import threading
import time
import uuid
from collections import OrderedDict

import requests
from django.views import View

from Ansjer.config import AWS_IOT_GETS3_PULL_CHINA_ID, AWS_IOT_GETS3_PULL_CHINA_SECRET, \
    AWS_IOT_GETS3_PULL_FOREIGN_ID, AWS_IOT_GETS3_PULL_FOREIGN_SECRET, AWS_ARN, AWS_IOT_SES_ACCESS_CHINA_REGION, \
    AWS_IOT_SES_ACCESS_FOREIGN_REGION_ASIA, AWS_IOT_SES_ACCESS_FOREIGN_REGION_EUROPE, \
    AWS_IOT_SES_ACCESS_FOREIGN_REGION_AMERICA, REGION_ID_LIST, ALEXA_DOMAIN
from Model.models import Device_Info, iotdeviceInfoModel, SerialNumberModel, UidSetModel
from Object.IOTCore.IotObject import IOTClient
from Object.ResponseObject import ResponseObject
from Object.TokenObject import TokenObject
from Service.CommonService import CommonService


class IotCoreView(View):

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

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

    def validate(self, operation, request_dict, request):
        response = ResponseObject()
        lang = request_dict.get('lang', 'en')
        response.lang = lang
        if operation == 'createKeysAndCertificate':  # 设备注册到IoT core
            return self.create_key_and_certificate(request_dict, response)
        elif operation == 'requestPublishMessage':
            return self.request_publish_message(request_dict, response)
        elif operation == 'getS3PullKey':
            return self.get_s3_pull_key(request_dict, response, request)
        elif operation == 'thingRegroup':  # OTA升级成功重新分组
            return self.thing_regroup(request_dict, response)
        elif operation == 'pcGetIotInfo':
            return self.pcGetIotInfo(request_dict, response)
        else:
            token = TokenObject(request_dict.get('token', None))
            if token.code != 0:
                return response.json(token.code)
            response.lang = token.lang

            if operation == 'clearIotCerm':
                return self.clear_Iot_Cerm(request_dict, response)
            elif operation == 'getIotInfo':
                return self.getIotInfo(request_dict, response)
            else:
                return response.json(404)

    # 设备注册到aws iot core
    @staticmethod
    def create_key_and_certificate(request_dict, response):
        logger = logging.getLogger('info')
        logger.info('设备注册到aws iot core请求参数:{}'.format(request_dict))
        token = request_dict.get('token', None)
        language = request_dict.get('language', None)
        time_stamp = request_dict.get('time_stamp', None)
        device_version = request_dict.get('device_version', None)

        if not all([token, language, time_stamp, device_version]):
            return response.json(444, {'param': 'token, language, time_stamp, device_version'})

        device_version = device_version.replace('.', '_')  # 物品组命名不能包含'.'
        try:
            # 时间戳token校验
            no_rtc = request_dict.get('no_rtc', None)
            if no_rtc:
                if not CommonService.check_time_stamp_token_without_distance(token, time_stamp):
                    return response.json(13)
            else:
                if not CommonService.check_time_stamp_token(token, time_stamp):
                    return response.json(13)

            uid = request_dict.get('uid', '')
            uid_code = request_dict.get('uid_code', None)
            company_mark = '11A'
            if not uid:  # 传序列号
                serial_number = request_dict.get('serial_number', None)
                serial_number_code = request_dict.get('serial_number_code', None)
                if not all([serial_number, serial_number_code]):
                    return response.json(444, {'param': 'serial_number, serial_number_code'})

                # 序列号编码解码校验
                serial_number_code = CommonService.decode_data(serial_number_code)
                if serial_number != serial_number_code:
                    return response.json(404)

                serial = serial_number[0:6]
                company_mark = serial_number[-3:]
                try:
                    SerialNumberModel.objects.get(serial_number=serial)
                except:
                    return response.json(173)

                thing_name_suffix = serial_number  # 物品名后缀
                iot_device_info_qs = iotdeviceInfoModel.objects.filter(serial_number=serial)
            else:  # 传uid
                # uid编码解码校验
                uid_code = CommonService.decode_data(uid_code)
                if uid != uid_code:
                    return response.json(404)

                serial = ''  # iot_deviceInfo表写入serial_number为''
                thing_name_suffix = uid  # 物品名后缀
                iot_device_info_qs = iotdeviceInfoModel.objects.filter(uid=uid)
            # 判断设备是否已注册过
            if iot_device_info_qs.exists():
                iot = iot_device_info_qs[0]
                res = {
                    'certificateId': iot.certificate_id,
                    'certificatePem': iot.certificate_pem,
                    'publicKey': iot.public_key,
                    'privateKey': iot.private_key,
                    'endpoint': iot.endpoint
                }
                return response.json(0, {'res': res})
            else:
                # 获取并判断region_id是否有效
                region_id = CommonService.confirm_region_id()
                if region_id not in REGION_ID_LIST:
                    return response.json(444, {'invalid region_id': region_id})

                iotClient = IOTClient(region_id)
                # 拼接物品名
                thingName = CommonService.get_thing_name(company_mark, thing_name_suffix)
                thingGroup = device_version + '_' + language
                res = iotClient.register_to_iot_core(thingName, thingGroup, response)

                token_iot_number = hashlib.md5((str(uuid.uuid1()) + str(int(time.time()))).encode('utf-8')).hexdigest()
                
                data = {
                    'uid': uid,
                    'serial_number': serial,
                    'endpoint': res[0]['endpoint'],
                    'certificate_id': res[0]['certificateId'],
                    'certificate_pem': res[0]['certificatePem'],
                    'public_key': res[0]['publicKey'],
                    'private_key': res[0]['privateKey'],
                    'thing_name': res[1]['ThingName'],
                    'thing_groups': res[1]['thingGroupName'],
                    'token_iot_number': token_iot_number
                }
                # 创建线程调用Alexa接口
                threading.Thread(target=call_alexa_api, kwargs=data).start()
                
                iotdeviceInfoModel.objects.create(**data)
                res = {
                    'certificateId': res[0]['certificateId'],
                    'certificatePem': res[0]['certificatePem'],
                    'publicKey': res[0]['publicKey'],
                    'privateKey': res[0]['privateKey'],
                    'endpoint': res[0]['endpoint']
                }
                return response.json(0, {'res': res})
        except Exception as e:
            print(e)
            return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))

    @staticmethod
    def thing_regroup(request_dict, response):
        """
        OTA升级成功重新分组
        @param request_dict: 请求参数
        @request_dict uid: 设备uid
        @request_dict region_id: 地区id
        @request_dict time_stamp: 时间戳
        @request_dict time_stamp_token: 时间戳token
        @request_dict device_version: 设备版本
        @request_dict language: 版本语言
        @param response: 响应对象
        @return: response
        """
        uid = request_dict.get('uid', '')
        time_stamp = request_dict.get('time_stamp', None)
        time_stamp_token = request_dict.get('time_stamp_token', None)
        device_version = request_dict.get('device_version', None)
        language = request_dict.get('language', None)

        if not all([time_stamp, time_stamp_token, device_version, language]):
            return response.json(444)

        # 时间戳token校验
        no_rtc = request_dict.get('no_rtc', None)
        if no_rtc:
            if not CommonService.check_time_stamp_token_without_distance(time_stamp_token, time_stamp):
                return response.json(13)
        else:
            if not CommonService.check_time_stamp_token(time_stamp_token, time_stamp):
                return response.json(13)

        # 获取并判断region_id是否有效
        region_id = CommonService.confirm_region_id()
        if region_id not in REGION_ID_LIST:
            return response.json(444, {'invalid region_id': region_id})

        company_mark = '11A'
        thing_name_suffix = uid
        if not uid:
            # 使用序列号
            serial_number = request_dict.get('serial_number', None)
            if not serial_number:
                return response.json(444, {{'error param': 'uid and serial_number'}})
            company_mark = serial_number[-3:]
            thing_name_suffix = serial_number
            uid = CommonService.query_uid_with_serial(serial_number)

        uid_set_qs = UidSetModel.objects.filter(uid=uid)
        thingName = CommonService.get_thing_name(company_mark, thing_name_suffix)
        new_thingGroupName = (device_version + '_' + language).replace('.', '_')  # 物品组命名不能包含'.'

        try:
            iotClient = IOTClient(int(region_id))
            # 获取旧物品组
            list_groups_res = iotClient.client.list_thing_groups_for_thing(thingName=thingName, maxResults=1)
            old_thingGroupName = list_groups_res['thingGroups'][0]['groupName']
            # 没有新物品组则创建
            list_thing_groups_res = iotClient.client.list_thing_groups(namePrefixFilter=new_thingGroupName
                                                                       , maxResults=1, recursive=False)
            if not list_thing_groups_res['thingGroups']:
                attributes = {
                    "update_time": "0"
                }
                thingGroupProperties = {
                    "thingGroupDescription": "OTA",
                    "attributePayload": {
                        "attributes": attributes,
                        "merge": False  # 更新时覆盖掉而不是合并
                    }
                }
                iotClient.client.create_thing_group(thingGroupName=new_thingGroupName
                                                    , thingGroupProperties=thingGroupProperties)
            iotClient.client.update_thing_groups_for_thing(thingName=thingName
                                                           , thingGroupsToAdd=[new_thingGroupName]
                                                           , thingGroupsToRemove=[old_thingGroupName])

            # 更新设备版本信息
            uid_set_qs.update(version=device_version)
            return response.json(0)
        except Exception as e:
            print(e)
            return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))

    def clear_Iot_Cerm(self, request_dict, response):
        serial_number = request_dict.get('serial_number', None)

        if serial_number:

            iot = iotdeviceInfoModel.objects.filter(thing_name="Ansjer_Device_" + serial_number)
            if iot.exists():
                iot.delete()
            return response.json(0)

        else:
            return response.json(444)

    # Alexa请求IoT Core下发MQTT消息通知设备开始或停止推流,或唤醒设备
    @staticmethod
    def request_publish_message(request_dict, response):
        UID = request_dict.get('UID', None)
        rtsp = request_dict.get('rtsp', None)
        enable = request_dict.get('enable', '1')
        if not all([UID, rtsp]):
            return response.json(444)

        try:
            thing_name = CommonService.query_serial_with_uid(UID)  # 存在序列号则为使用序列号作为物品名
            topic_name = 'ansjer/generic/{}'.format(thing_name)
            msg = OrderedDict(
                [
                    ('alexaRtspCommand', rtsp),
                    ('enable', int(enable)),
                ]
            )
            if not CommonService.req_publish_mqtt_msg(thing_name, topic_name, msg):
                return response.json(10044)
            return response.json(0)
        except Exception as e:
            return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))

    def get_s3_pull_key(self, request_dict, response, request):
        # 通用发布主题通知
        UID = request_dict.get('UID', None)
        if not all([UID]):
            return response.json(444)

        try:
            # 获取检查uid的序列号,如果没有序列号,不使用MQTT下发消息
            device_info_qs = Device_Info.objects.filter(UID=UID).values('UID', 'serial_number')
            if not device_info_qs.exists():
                return response.json(10043)
            uid = device_info_qs[0]['UID']
            serial_number = device_info_qs[0]['serial_number']
            # 如果device_info表的serial_number不为空,物品名为'Ansjer_Device_序列号'
            thing_name_suffix = serial_number if serial_number != '' else uid
            # 获取数据组织将要请求的url
            iot = iotdeviceInfoModel.objects.filter(thing_name__contains=thing_name_suffix).values('thing_name',
                                                                                                   'endpoint',
                                                                                                   'token_iot_number')
            if not iot.exists():
                return response.json(10043)
            endpoint = iot[0]['endpoint']
            MSG = self.get_s3_key_return_msg(endpoint)

            return response.json(0, MSG)
        except Exception as e:
            # print(e)
            return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))

    def get_s3_key_return_msg(self, endpoint):
        MSG = {}
        if 'cn-northwest-1' in endpoint:
            key = AWS_IOT_GETS3_PULL_CHINA_ID
            secret = AWS_IOT_GETS3_PULL_CHINA_SECRET
            arn = AWS_ARN[0]
            region_name = AWS_IOT_SES_ACCESS_CHINA_REGION
        else:
            key = AWS_IOT_GETS3_PULL_FOREIGN_ID
            secret = AWS_IOT_GETS3_PULL_FOREIGN_SECRET
            arn = AWS_ARN[1]
            if 'ap-southeast-1' in endpoint:
                region_name = AWS_IOT_SES_ACCESS_FOREIGN_REGION_ASIA
            if 'eu-west-1' in endpoint:
                region_name = AWS_IOT_SES_ACCESS_FOREIGN_REGION_EUROPE
            if 'us-east-1' in endpoint:
                region_name = AWS_IOT_SES_ACCESS_FOREIGN_REGION_AMERICA

        MSG['AccessKeyId'] = key
        MSG['AccessKeySecret'] = secret
        MSG['bucket_name'] = 'asj-log'
        MSG['arn'] = arn
        MSG['region_name'] = region_name
        return MSG

    def getIotInfo(self, request_dict, response):
        # 获取IoT数据
        serial_number = request_dict.get('serial_number', None)
        uid = request_dict.get('uid', None)
        if not uid and not serial_number:
            return response.json(444)

        try:
            if serial_number:
                serial_number = serial_number[0:6]
                iot_info_qs = iotdeviceInfoModel.objects.filter(serial_number=serial_number). \
                    values('endpoint', 'token_iot_number')
            else:
                iot_info_qs = iotdeviceInfoModel.objects.filter(uid=uid). \
                    values('endpoint', 'token_iot_number')

            if not iot_info_qs.exists():
                return response.json(173)

            endpoint = iot_info_qs[0]['endpoint']
            token_iot_number = iot_info_qs[0]['token_iot_number']
            res = {'endpoint': endpoint, 'token_iot_number': token_iot_number}
            return response.json(0, res)
        except Exception as e:
            return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))

    def pcGetIotInfo(self, request_dict, response):
        # PC工具获取IoT数据
        serial_number = request_dict.get('serial_number', None)
        uid = request_dict.get('uid', None)
        if not uid and not serial_number:
            return response.json(444)

        try:
            if serial_number:
                serial_number = serial_number[0:6]
                iot_info_qs = iotdeviceInfoModel.objects.filter(serial_number=serial_number). \
                    values('endpoint', 'token_iot_number')
            else:
                iot_info_qs = iotdeviceInfoModel.objects.filter(uid=uid). \
                    values('endpoint', 'token_iot_number')

            if not iot_info_qs.exists():
                return response.json(173)

            endpoint = iot_info_qs[0]['endpoint']
            token_iot_number = iot_info_qs[0]['token_iot_number']
            res = {'endpoint': endpoint, 'token_iot_number': token_iot_number}
            return response.json(0, res)
        except Exception as e:
            return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))


def call_alexa_api(**kwargs):
    url = ALEXA_DOMAIN + 'device/creatOrUpdateIotInfo'
    requests.post(url=url, data=kwargs, timeout=30)