import json import logging import subprocess import threading import time from collections import OrderedDict from datetime import datetime from gc import enable import requests from django.http import JsonResponse from django.shortcuts import render from django.views import View from django.views.generic import TemplateView from azoauth.config import * from model.models import UserModel, UidRtspModel, AlexaAuthModel, UserCountModel, SwitchModel from object.RedisObject import RedisObject from object.ResObject import ResObject from object.ResponseObject import ResponseObject from object.tkObject import tkObject from service.CommonService import CommonService import uuid class authView(TemplateView): def post(self, request, *args, **kwargs): request.encoding = 'utf-8' request_dict = json.loads(request.body.decode('utf-8')) return self.validate(request_dict) def get(self, request, *args, **kwargs): request.encoding = 'utf-8' request_dict = request.GET return self.validate(request_dict) def validate(self, request_dict): logger = logging.getLogger('django') logger.info('------oa2/auth请求参数: {}------'.format(request_dict)) state = request_dict.get("state", '') client_id = request_dict.get("client_id", '') response_type = request_dict.get("response_type", '') scope = request_dict.get("scope", '') redirect_uri = request_dict.get("redirect_uri", '') context = { 'state': state, 'client_id': client_id, 'response_type': response_type, 'scope': scope, 'redirect_uri': redirect_uri, 'skill_name': 'zosi smart' } return render(request=None, template_name="login.html", context=context) # Anlapus登录 class authAnlapusView(TemplateView): def post(self, request, *args, **kwargs): request.encoding = 'utf-8' request_dict = json.loads(request.body.decode('utf-8')) return self.validate(request_dict) def get(self, request, *args, **kwargs): request.encoding = 'utf-8' request_dict = request.GET return self.validate(request_dict) def validate(self, request_dict): state = request_dict.get("state", '') client_id = request_dict.get("client_id", '') response_type = request_dict.get("response_type", '') scope = request_dict.get("scope", '') redirect_uri = request_dict.get("redirect_uri", '') context = { 'state': state, 'client_id': client_id, 'response_type': response_type, 'scope': scope, 'redirect_uri': redirect_uri, 'skill_name': 'Anlapus' } return render(request=None, template_name="login_anlapus.html", context=context) # loocam登录 class authLoocamView(TemplateView): def post(self, request, *args, **kwargs): request.encoding = 'utf-8' request_dict = json.loads(request.body.decode('utf-8')) return self.validate(request_dict) def get(self, request, *args, **kwargs): request.encoding = 'utf-8' request_dict = request.GET return self.validate(request_dict) def validate(self, request_dict): state = request_dict.get("state", '') client_id = request_dict.get("client_id", '') response_type = request_dict.get("response_type", '') scope = request_dict.get("scope", '') redirect_uri = request_dict.get("redirect_uri", '') context = { 'state': state, 'client_id': client_id, 'response_type': response_type, 'scope': scope, 'redirect_uri': redirect_uri, 'skill_name': 'loocam' } logger = logging.getLogger('django') logger.info('loocam请求登录网页，参数为{}'.format(context)) return render(request=None, template_name="login_loocam.html", context=context) # VSees登录 class authVSeesView(TemplateView): def post(self, request, *args, **kwargs): request.encoding = 'utf-8' request_dict = json.loads(request.body.decode('utf-8')) return self.validate(request_dict) def get(self, request, *args, **kwargs): request.encoding = 'utf-8' request_dict = request.GET return self.validate(request_dict) def validate(self, request_dict): state = request_dict.get("state", '') client_id = request_dict.get("client_id", '') response_type = request_dict.get("response_type", '') scope = request_dict.get("scope", '') redirect_uri = request_dict.get("redirect_uri", '') context = { 'state': state, 'client_id': client_id, 'response_type': response_type, 'scope': scope, 'redirect_uri': redirect_uri, 'skill_name': 'VSees' } logger = logging.getLogger('django') logger.info('VSees请求登录网页，参数为{}'.format(context)) return render(request=None, template_name="login_VSees.html", context=context) class loginHandleView(TemplateView): def post(self, request, *args, **kwargs): request.encoding = 'utf-8' request_dict = json.loads(request.body.decode('utf-8')) return self.validate(request_dict) def get(self, request, *args, **kwargs): request.encoding = 'utf-8' request_dict = request.GET return self.validate(request_dict) def validate(self, request_dict): response = ResObject() user = request_dict.get("user", '') pwd = request_dict.get("pwd", '') state = request_dict.get("state", '') client_id = request_dict.get("client_id", '') response_type = request_dict.get("response_type", '') scope = request_dict.get("scope", '') redirect_uri = request_dict.get("redirect_uri", '') skill_name = request_dict.get("skill_name", '') # 返回code logger = logging.getLogger('django') logger.info('------开始认证登录------参数:{}'.format(request_dict)) # 请求美洲服登录接口 region_code = 'US' auth_request_url = '{SERVER_PREFIX}/oalexa/auth'.format(SERVER_PREFIX=SERVER_PREFIX) requests_data = {'userName': user, 'userPwd': pwd} res = requests.post(url=auth_request_url, data=requests_data) if res.status_code != 200: return response.json(10, res={'错误': '请求响应异常'}) res_json = res.json() logger.info('美洲服务器响应: {}'.format(res_json)) # 账号不存在/密码错误/账号存在但用户地区为欧洲时请求欧洲服 result_code = res_json['result_code'] if result_code == 104 or result_code == 111 or (result_code == 0 and res_json['result']['region_code'] == 'EU'): auth_request_url = '{SERVER_PREFIX}/oalexa/auth'.format(SERVER_PREFIX=SERVER_PREFIX_EU) res = requests.post(url=auth_request_url, data=requests_data) if res.status_code != 200: return response.json(10, res={'错误': '请求响应异常'}) res_json = res.json() logger.info('欧洲服务器响应: {}'.format(res_json)) if res_json['result_code'] == 0: region_code = 'EU' if res_json['result_code'] != 0: return response.json(10, res={'msg': 'error'}, extra={'msg': res_json['reason']}) nowTime = int(time.time()) code = CommonService.encrypt_data(32) userID = res_json['result']['userID'] user_qs = UserModel.objects.filter(userID=userID) if user_qs.exists(): user_qs.update(region_code=region_code, code=code, updTime=nowTime) else: UserModel.objects.create(userID=userID, region_code=region_code, code=code, addTime=nowTime, updTime=nowTime) year_month = str(time.strftime('%Y%m', time.localtime(nowTime))) # 获取当前年月 user_count_qs = UserCountModel.objects.filter(skill_name=skill_name, year_month=year_month).values('amount') if not user_count_qs.exists(): UserCountModel.objects.create(skill_name=skill_name, year_month=year_month, amount=1) else: # 用户数量+1 amount = user_count_qs[0]['amount'] + 1 user_count_qs.update(amount=amount) redirect_uri += '?code=' + code + '&state=' + state logger.info('------认证登录响应------:{}'.format(redirect_uri)) return response.json(0, res=redirect_uri) class oa2TokenView(TemplateView): def post(self, request, *args, **kwargs): request.encoding = 'utf-8' # request_dict = json.loads(request.body.decode('utf-8')) request_dict = request.POST return self.validate(request_dict) def get(self, request, *args, **kwargs): request.encoding = 'utf-8' request_dict = request.GET return self.validate(request_dict) def validate(self, request_dict): # 增加对code和client_id的校验代码，返回access_token和refresh_token code = request_dict.get("code", None) grant_type = request_dict.get("grant_type", None) # refresh_token, authorization_code client_id = request_dict.get("client_id", None) refresh_token = request_dict.get("refresh_token", None) logger = logging.getLogger('django') logger.info('请求获取令牌接口参数:{}'.format(request_dict)) # 根据用户授权码获取令牌 user_qs = UserModel.objects.filter(code=code) if not user_qs.exists(): if grant_type == 'authorization_code': user_qs = UserModel.objects.filter(user_authorization_code=code) else: user_qs = UserModel.objects.filter(refresh_token=refresh_token) if not user_qs.exists(): res_json = {'msg': 'code not exists'} logger.info('请求获取令牌接口响应:{}'.format(res_json)) return JsonResponse(res_json) access_token = CommonService.encrypt_data(randomlength=32) refresh_token = CommonService.encrypt_data(randomlength=32) is_update = user_qs.update(access_token=access_token, refresh_token=refresh_token) if is_update: res_json = { "access_token": access_token, "token_type": "bearer", "expires_in": 3600, "refresh_token": refresh_token, } logger.info('请求获取令牌接口响应:{}'.format(res_json)) return JsonResponse(res_json) else: logger.info({'msg': 'error'}) return JsonResponse({'msg': 'error'}) def runSendRtspMsg_thread(UID, PWD, MSG): command = "./pushtool {UID} {PWD} {MSG} 1".format(UID=UID, PWD=PWD, MSG=MSG) print('command=>{command}'.format(command=command)) try: back = subprocess. \ Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE). \ communicate(timeout=10) except Exception as e: return False else: print("back0----", back[0].decode()) # 注意需要进行解码操作，默认输出的是字节 print("back1----", back[1].decode()) # back是一个元祖，可以通过元祖取值的方式获取结果 return True class oa2RtspStartView(TemplateView): def post(self, request, *args, **kwargs): request.encoding = 'utf-8' request_dict = request.POST return self.validate(request_dict) def get(self, request, *args, **kwargs): request.encoding = 'utf-8' request_dict = request.GET return self.validate(request_dict) def validate(self, request_dict): st = request_dict.get("st", 0) uid = request_dict.get("id", '') access_token = request_dict.get("access_token", '') skill_name = request_dict.get("skill_name", 'zosi smart') user_qs = UserModel.objects.filter(access_token=access_token) if not user_qs.exists(): return JsonResponse({'错误': '用户数据不存在'}) ur_qs = UidRtspModel.objects.filter(uid=uid).values('user_id', 'uid', 'nick', 'rtsp_url', 'password') if not ur_qs.exists(): return JsonResponse({'错误': 'uid数据不存在'}) # 确认设备的用户地区 region = 'US' user_id = ur_qs[0]['user_id'] user_qs = UserModel.objects.filter(userID=user_id).values('region_code') if user_qs.exists(): region = user_qs[0]['region_code'] UID = ur_qs[0]['uid'] nick = ur_qs[0]['nick'] PWD = ur_qs[0]['password'] stream_name = ur_qs[0]['rtsp_url'] channel = '0' if '_' in UID: # 多通道设备 channel = UID[-1:] UID = UID[:-2] # 根据用户地区确认域名 domain_name = SERVER_PREFIX_EU if region == 'EU' else SERVER_PREFIX RESP_SERVER_DOMAIN = RESP_SERVER_DOMAIN_DATA[region] MSG = '{}://{}:8554/{}'.format(RTSP_PREFIX, RESP_SERVER_DOMAIN, stream_name) # 测试往go2rtc推流 if UID == '3VCV5HZD9DRPA6EU111A': MSG = 'rtsp://144.24.9.228:8554/camera1' logger = logging.getLogger('django') logger.info('------{} 开始向设备下发推流指令------'.format(skill_name)) # 此处用线程异步发送指令 if int(st) == 1: send_flag = self.runSendStop(UID, PWD, MSG) logger.info('----------send_flag---st=1-----------------') if send_flag: return JsonResponse({'msg': 'stop yes', 'code': 0}) else: return JsonResponse({'msg': 'stop no', 'code': 0}) # 创建并启动线程 thread = threading.Thread( target=self.send_rtsp_command_thread, args=(domain_name, UID, PWD, MSG, channel, logger, region) ) thread.start() # 拉流地址 rtsp_uri = '{}://{}:443/{}'.format(RTSP_PREFIX, RESP_SERVER_DOMAIN, stream_name) stop_time = int(time.time()) + 2 * 60 expirationTime = time.strftime('%Y-%m-%dT%H:%MZ', time.localtime(stop_time)) res_json = { 'uid': UID, 'pwd': PWD, 'msg': MSG, 'uri': rtsp_uri, 'endpointId': uid, 'friendlyName': nick, 'manufacturerName': skill_name, 'expirationTime': expirationTime, 'description': 'Camera connected via {}'.format(skill_name), 'audioCodecs': 'ACC', 'videoCodecs': 'H264', 'protocols': ['RTSP'], 'idleTimeoutSeconds': 5, 'modelName': 'P1425-LE', 'authorizationTypes': ['NONE'], 'manufacturerId': 'zosi-ACCC8E5E7513', 'resolutions': {'width': 640, 'height': 360}, } logger.info('------------返回控制摄像头的信息---------------: {}'.format(res_json)) return JsonResponse(res_json, safe=False) def send_rtsp_command_thread(self, domain_name, UID, PWD, MSG, channel, logger, region): """异步发送RTSP命令的线程函数""" thing_name = CommonService.query_serial_with_uid(UID) topic_name = 'ansjer/generic/{}'.format(thing_name) msg = OrderedDict( [ ('alexaRtspCommand', MSG), ('enable', 1), ] ) publish_result = CommonService.req_publish_mqtt_msg(thing_name, topic_name, msg) logger.info('{}下发指令结果:{}'.format(UID, publish_result)) command = "./pushtool {UID} {PWD} {MSG} 1 {channel}".format( UID=UID, PWD=PWD, MSG=MSG, channel=channel) # 请求MQTT发布消息 url = '{}/iot/requestPublishMessage'.format(domain_name) requests_data = {'UID': UID, 'rtsp': MSG, 'enable': '1'} try: r = requests.post(url, requests_data) if r.status_code == 200: res = r.json() logger.info('请求MQTT发布消息参数:{},result_code: {}'.format(requests_data, res['result_code'])) if res['result_code'] == 0: logger.info('请求MQTT下发指令成功---正式服') elif res['result_code'] == 10044: url = '{}/iot/requestPublishMessage'.format(SERVER_PREFIX_TEST) r = requests.post(url, requests_data) if r.status_code == 200: res = r.json() if res['result_code'] == 0: logger.info('请求MQTT下发指令成功---测试服') else: self.runSendRtspMsg(logger, region, command) else: self.runSendRtspMsg(logger, region, command) else: self.runSendRtspMsg(logger, region, command) else: self.runSendRtspMsg(logger, region, command) except Exception as e: logger.error('发送RTSP命令异常: {}'.format(str(e))) self.runSendRtspMsg(logger, region, command) def runReqRtspMsg(self, UID, PWD, MSG): request_url = 'http://localhost:5000/?UID={UID}&MSG={MSG}&CMD=1&PWD={PWD}'. \ format(UID=UID, PWD=PWD, MSG=MSG) res = requests.get(url=request_url) print(res) return True # 触发此方法，让摄像头推流到MSG流地址 def runSendRtspMsg(self, logger, region, command): logger.info('------------推流指令: {}---------------'.format(command)) if region == 'CN': logger.info('------------国内发送推流指令---------------') url = "http://52.83.252.41:7880/alexa/command?command={command}".format(command=command) # 请求国内服务器调用pushtool try: requests.get(url=url, timeout=10) except Exception as e: logger.info('请求国内服务器调用pushtool异常: {}'.format(repr(e))) else: logger.info('------------国外发送推流指令---------------') try: back = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE). \ communicate(timeout=10) logger.info('back: {}'.format(str(back[0].decode()) + str(back[1].decode()))) except Exception as e: self.runSendRtspMsg(logger, 'CN', command) # 调用失败时尝试用国内的发送 logger.info('调用pushtool异常: {}'.format(repr(e))) def runSendStop(self, UID, PWD, MSG): command = "./pushtool {UID} {PWD} {MSG} 0".format(UID=UID, PWD=PWD, MSG=MSG) print('command=>{command}'.format(command=command)) try: back = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE). \ communicate(timeout=10) except Exception as e: return False else: print("back0----", back[0].decode()) # 注意需要进行解码操作，默认输出的是字节 print("back1----", back[1].decode()) # back是一个元祖，可以通过元祖取值的方式获取结果 return True class oa2DiscoveryDevice(TemplateView): def post(self, request, *args, **kwargs): request.encoding = 'utf-8' request_dict = request.POST return self.validate(request_dict) def get(self, request, *args, **kwargs): request.encoding = 'utf-8' request_dict = request.GET return self.validate(request_dict) def validate(self, request_dict): # 增加对code和client_id的校验代码，返回access_token和refresh_token skill_name = request_dict.get("skill_name", 'zosi smart') access_token = request_dict.get("access_token", None) logger = logging.getLogger('django') logger.info('--------{} 开始搜索设备--------'.format(skill_name)) response = ResObject() user_qs = UserModel.objects.filter(access_token=access_token) if not user_qs.exists(): return response.json(500, res={'msg': '用户数据不存在'}) user = user_qs[0] userID = user.userID region_code = user.region_code logger.info('搜索设备的用户id:{},地区:{}'.format(userID, region_code)) # AlexaAuthModel表数据 alexAuth = AlexaAuthModel.objects.filter(token=access_token, skill_name=skill_name).order_by('-addTime') if alexAuth.exists(): auth_res = alexAuth.values() event_access_token = auth_res[0]['access_token'] event_refresh_token = auth_res[0]['refresh_token'] event_token = auth_res[0]['token'] event_expiresTime = auth_res[0]['expiresTime'] event_addTime = auth_res[0]['addTime'] event_updTime = auth_res[0]['updTime'] event_alexa_region = auth_res[0]['alexa_region'] AlexaAuthModel.objects.filter(userID=userID).delete() alexAuth.delete() AlexaAuthModel.objects.create( userID=userID, access_token=event_access_token, refresh_token=event_refresh_token, token=event_token, expiresTime=event_expiresTime, addTime=event_addTime, updTime=event_updTime, alexa_region=event_alexa_region, skill_name=skill_name, ) # 请求搜索设备 domain_name = SERVER_PREFIX_EU if region_code == 'EU' else SERVER_PREFIX auth_request_url = '{}/oalexa/discoveryuid'.format(domain_name) requests_data = {'sid': 'admin', 'sst': 'admin', 'alexa_user_id': userID} res = requests.post(url=auth_request_url, data=requests_data) if res.status_code != 200: return response.json(500, res={'msg': 'discover device error'}) res_json = res.json() logger.info('{}服务器搜索设备响应:{}'.format(region_code, res_json)) if res_json['result_code'] != 0: return response.json(500, res={'msg': 'discover device error'}) uid_arr = res_json['result']['uid_arr'] rtko = tkObject(rank=1) now_time = int(time.time()) user.uid_rtsp.clear() res_json = [] uid_rtsp_id_list = [] for uid_a in uid_arr: uid = uid_a['uid'] serial_number = uid_a['serial_number'] nick = uid_a['nick'] rtsp_url = rtko.encrypt(data=uid) try: uid_rtsp_qs = UidRtspModel.objects.get(uid=uid) except UidRtspModel.DoesNotExist: uid_rtsp_qs = UidRtspModel.objects.create( user_id=userID, uid=uid, serial_number=serial_number, nick=nick, region=region_code, password=uid_a['password'], rtsp_url=rtsp_url, addTime=now_time, updTime=now_time) else: uid_rtsp_qs.nick = nick uid_rtsp_qs.region = region_code uid_rtsp_qs.password = uid_a['password'] uid_rtsp_qs.user_id = userID uid_rtsp_qs.serial_number = serial_number uid_rtsp_qs.save() uid_rtsp_id_list.append(uid_rtsp_qs.id) rtsp_domain = RESP_SERVER_DOMAIN_DATA[region_code] rtsp_uri = '{}://{}:443/{}'.format(RTSP_PREFIX, rtsp_domain, rtsp_url) ur_data = { 'uri': rtsp_uri, 'endpointId': uid, 'friendlyName': nick, 'manufacturerName': skill_name, 'description': 'Camera connected via {}'.format(skill_name), 'protocols': ['RTSP'], 'audioCodecs': ['ACC'], 'videoCodecs': ['H264'], 'modelName': 'P1425-LE', 'authorizationTypes': ['NONE'], 'manufacturerId': 'zosi-ACCC8E5E7513', 'resolutions': [{'width': 640, 'height': 360}], } res_json.append(ur_data) user.uid_rtsp.add(*uid_rtsp_id_list) logger.info('搜索设备返回值: {}'.format(res_json)) return JsonResponse(res_json, safe=False) class oa2DiscoverySwitch(TemplateView): def post(self, request, *args, **kwargs): request.encoding = 'utf-8' request_dict = request.POST return self.validate(request_dict) def get(self, request, *args, **kwargs): request.encoding = 'utf-8' request_dict = request.GET return self.validate(request_dict) def validate(self, request_dict): # 增加对code和client_id的校验代码，返回access_token和refresh_token skill_name = request_dict.get("skill_name", 'loocam') access_token = request_dict.get("access_token", None) logger = logging.getLogger('django') logger.info('--------{} 开始搜索设备--------'.format(skill_name)) user_qs = UserModel.objects.filter(access_token=access_token) if not user_qs.exists(): return JsonResponse({'错误': '用户数据不存在'}) response = ResObject() user = user_qs.first() userID = user.userID region_code = user.region_code logger.info('userID: {},region_code:{}'.format(userID, region_code)) # 更新事件网关接口 alexAuth = AlexaAuthModel.objects.filter(token=access_token, skill_name=skill_name).order_by('-addTime') if alexAuth.exists(): auth_res = alexAuth.values() event_access_token = auth_res[0]['access_token'] event_refresh_token = auth_res[0]['refresh_token'] event_token = auth_res[0]['token'] event_expiresTime = auth_res[0]['expiresTime'] event_addTime = auth_res[0]['addTime'] event_updTime = auth_res[0]['updTime'] event_alexa_region = auth_res[0]['alexa_region'] AlexaAuthModel.objects.filter(userID=userID).delete() alexAuth.delete() logger.info('update_event_access_token') logger.info(event_token) AlexaAuthModel.objects.create( userID=userID, access_token=event_access_token, refresh_token=event_refresh_token, token=event_token, expiresTime=event_expiresTime, addTime=event_addTime, updTime=event_updTime, alexa_region=event_alexa_region, skill_name=skill_name, ) domain_name = SERVER_PREFIX_EU if region_code == 'EU' else SERVER_PREFIX auth_request_url = '{}/oalexa/discoveryswitch'.format(domain_name) requests_data = {'sid': 'admin', 'sst': 'admin', 'alexa_user_id': userID} res = requests.post(url=auth_request_url, data=requests_data) res_json = res.json() logger.info('{}正式服务器响应: {}'.format(region_code, res_json)) # 添加测试服务器测试 if res_json['result_code'] != 0: auth_request_url = '{}/oalexa/discoveryswitch'.format(SERVER_PREFIX_TEST) res = requests.post(url=auth_request_url, data=requests_data) res_json = res.json() logger.info('请求服务器url: {}'.format(auth_request_url)) logger.info('服务器响应: {}'.format(res_json)) if res_json['result_code'] != 0: return response.json(0, res={'msg': 'error'}) switch_list = res_json['result']['switch_list'] now_time = int(time.time()) res_json = [] for switch in switch_list: serial_number = switch['uid'] nick = switch['nick'] switch_info_qs = SwitchModel.objects.filter(serial_number=serial_number, userID=userID) if switch_info_qs.exists(): switch_info_qs.update(nick=nick, region=switch['region'], updTime=now_time) else: SwitchModel.objects.create(serial_number=serial_number, nick=nick, region=switch['region'], addTime=now_time, updTime=now_time, userID=userID) ur_data = { 'endpointId': serial_number, 'friendlyName': nick, 'manufacturerName': skill_name, 'description': 'Plug connected via {}'.format(skill_name), 'modelName': 'P1425-LE', 'manufacturerId': 'zosi-ACCC8E5E7513', } res_json.append(ur_data) logger.info('搜索设备返回值: {}'.format(res_json)) return JsonResponse(res_json, safe=False) class powerController(TemplateView): def post(self, request, *args, **kwargs): request.encoding = 'utf-8' request_dict = request.POST return self.power_controller(request_dict) def get(self, request, *args, **kwargs): request.encoding = 'utf-8' request_dict = request.GET return self.power_controller(request_dict) def power_controller(self, request_dict): serial_number = request_dict.get('serial_number', '') access_token = request_dict.get('access_token', '') skill_name = request_dict.get('skill_name', 'loocam') power_controller = request_dict.get('power_controller', '') if not all([serial_number, access_token, skill_name]): return JsonResponse({'result_code': '500', '错误': '缺少参数'}) user_qs = UserModel.objects.filter(access_token=access_token) if not user_qs.exists(): return JsonResponse({'result_code': '500', '错误': '用户数据不存在'}) userID = user_qs.first().userID switch_qs = SwitchModel.objects.filter(serial_number=serial_number, userID=userID).values('nick', 'region') if not switch_qs.exists(): return JsonResponse({'result_code': '500', '错误': 'serial_number数据不存在'}) logger = logging.getLogger('django') logger.info('{} 控制插座 {}'.format(skill_name, serial_number)) # 请求MQTT发布消息 region = switch_qs[0]['region'] if region == 'EU': domain_name = SERVER_PREFIX_EU elif region == 'CN': domain_name = SERVER_PREFIX_TEST else: domain_name = SERVER_PREFIX url = '{}/api/loocam/open/socket/alexa-socket-switch'.format(domain_name) requests_data = {'serial_number': serial_number, 'power_controller': power_controller} # TurnOn, TurnOff r = requests.post(url, requests_data) if r.status_code != 200: return JsonResponse({'result_code': '500', '错误': '请求服务器响应异常'}) res = r.json() logger.info('{}服务器返回状态: {}'.format(region, res)) if res['result_code'] != 0: return JsonResponse({'result_code': '500', '错误': '请求MQTT下发指令失败'}) else: logger.info('请求MQTT下发指令成功') return JsonResponse({'result_code': '0'}) class RtcController(TemplateView): 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): if operation == 'rtc': return self.rtc(request_dict) @classmethod def rtc(cls, request_dict): uid = request_dict.get("uid") access_token = request_dict.get("access_token") skill_name = request_dict.get("skill_name") offer_sdp = request_dict.get("offer_sdp") if not all([uid, access_token, skill_name, offer_sdp]): return JsonResponse({'错误': '缺少参数'}) user_qs = UserModel.objects.filter(access_token=access_token) if not user_qs.exists(): return JsonResponse({'错误': '用户数据不存在'}) ur_qs = UidRtspModel.objects.filter(uid=uid).values('user_id', 'uid', 'nick', 'rtsp_url', 'password') if not ur_qs.exists(): return JsonResponse({'错误': 'uid数据不存在'}) logger = logging.getLogger('django') try: # 请求go2rtc创建流 rtsp = GO2RTC_RTSP.format(uid) rtc_url = GO2RTC_API.format('api/streams') params = { 'src': rtsp, 'name': uid, } r = requests.put(url=rtc_url, data=params, timeout=30) assert r.status_code == 200 # mqtt下发推流指令 logger.info('------rtc开始向设备下发推流指令:{}------'.format(uid)) # 确认设备的用户地区 region = 'US' user_id = ur_qs[0]['user_id'] user_qs = UserModel.objects.filter(userID=user_id).values('region_code') if user_qs.exists(): region = user_qs[0]['region_code'] # 根据用户地区确认域名 domain_name = SERVER_PREFIX_EU if region == 'EU' else SERVER_PREFIX # 请求MQTT发布消息 url = '{}/iot/requestPublishMessage'.format(domain_name) requests_data = {'UID': uid, 'rtsp': rtsp, 'enable': '1'} # 1: 开始推流,0: 停止推流; channel: 推流通道 r = requests.post(url, requests_data) assert r.status_code == 200 res = r.json() logger.info('rtc请求MQTT发布消息参数:{},result_code: {}'.format(requests_data, res['result_code'])) if res['result_code'] == 0: logger.info('rtc请求MQTT下发指令成功,地区:{}'.format(region)) elif res['result_code'] == 10044: url = '{}/iot/requestPublishMessage'.format(SERVER_PREFIX_TEST) # 测试服务器 r = requests.post(url, requests_data) assert r.status_code == 200 res = r.json() if res['result_code'] == 0: logger.info('请求MQTT下发指令成功---测试服') # 获取SDP sdp = cls.handle_alexa_offer(offer_sdp, rtsp) # params = { # 'src': uid, # } # r = requests.get(url=rtc_url, params=params, timeout=30) # assert r.status_code == 200 # res = r.json() # # 遍历producers数组，查找包含sdp字段的对象 # sdp = '' # for producer in res['producers']: # if 'sdp' in producer: # sdp = producer['sdp'] # break res_json = { 'SDP': sdp } logger.info('------------rtc响应数据---------------: {}'.format(res_json)) return JsonResponse(res_json, safe=False) except Exception as e: logger.info('rtc异常,error_line:{},error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e))) return JsonResponse({'错误': 'rtc异常'}) class VesseTest(TemplateView): def get(self, request, *args, **kwargs): request.encoding = 'utf-8' operation = kwargs.get('operation') return self.validation(request.GET, operation, request) def post(self, request, *args, **kwargs): request.encoding = 'utf-8' operation = kwargs.get('operation') return self.validation(request.POST, operation, request) def validation(self, request_dict, operation, request): response = ResponseObject() if operation == 'get-token': # 获取token return self.get_token(response) elif operation == 'get-user-id': # 获取userid return self.get_user_id(request_dict, response, request) elif operation == 'get-user-info': # 获取用户信息 return self.get_user_info(request_dict, response, request) elif operation == 'rtc': # rtc测试 return self.rtc(request_dict, response, request) def get_token(self, response): data = { 'corpid': 'ww467ec1685e8262e6', 'corpsecret': 'IeUoaQ-0hEhEduCQq1zyfVXjfeZpMsThK1nklszRzUY' } try: token_url = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken' token_response = requests.get(token_url, params=data) if token_response.status_code == 200: data = token_response.json() access_token = data.get("access_token") if access_token: return response.json(0, {'access_token': access_token}) else: return response.json(0, "Failed to get AccessToken.") else: code = token_response.status_code content = token_response.content return response.json(code, content) except requests.exceptions.RequestException as e: print(e) return response.json(500, repr(e)) def get_user_id(self, request_dict, response, request): access_token = request_dict.get('access_token', None) code = request_dict.get('code', None) if not all([access_token, code]): return response.json(444, 'error: access_token, code') data = { 'access_token': access_token, 'code': code } try: token_url = 'https://qyapi.weixin.qq.com/cgi-bin/auth/getuserinfo' token_response = requests.get(token_url, params=data) if token_response.status_code == 200: data = token_response.json() userid = data.get("userid") if userid: return response.json(0, {'userid': userid}) else: code = token_response.status_code content = token_response.content return response.json(code, content) except Exception as e: print(e) return response.json(500, repr(e)) def get_user_info(self, request_dict, response, request): access_token = request_dict.get('access_token', None) userid = request_dict.get('userid', None) if not all([access_token, userid]): return response.json(444, 'error: access_token, userid') data = { 'access_token': access_token, 'userid': userid } try: token_url = 'https://qyapi.weixin.qq.com/cgi-bin/user/get' token_response = requests.get(token_url, params=data) if token_response.status_code == 200: data = token_response.json() userid = data.get("userid") name = data.get('name') position = data.get('position') status = data.get('status') if userid: url_data = { 'userid': userid, 'name': name, 'position': position, 'status': status } return response.json(0, {'data': url_data}) else: code = token_response.status_code content = token_response.content return response.json(code, content) except Exception as e: print(e) return response.json(500, repr(e)) @staticmethod def rtc(request_dict, response, request): uid = request_dict.get('uid', None) try: # 请求go2rtc创建流 # rtsp = GO2RTC_RTSP.format(uid) rtc_url = GO2RTC_API.format('api/streams') # params = { # 'src': rtsp, # 'name': uid, # } # r = requests.put(url=rtc_url, params=params, timeout=30) # assert r.status_code == 200 # 获取SDP params = { 'src': uid, } r = requests.get(url=rtc_url, params=params, timeout=30) assert r.status_code == 200 res = r.json() # 遍历producers数组，查找包含sdp字段的对象 sdp = '' for producer in res['producers']: if 'sdp' in producer: sdp = producer['sdp'] break res_json = { 'SDP': sdp } return response.json(0) except Exception as e: print(e) return response.json(500, repr(e)) logger = logging.getLogger('django') def get_uuid(): return str(uuid.uuid4())