# @Author : Rocky # @File : AlexaController.py # @Time : 2023/12/25 10:46 import time import requests from django.views import View from Model.models import AlexaOauth from Object.ResponseObject import ResponseObject from Object.TokenObject import TokenObject from Ansjer.config import CONFIG_INFO, CONFIG_EUR class AppToAppView(View): 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() token = TokenObject(request.META.get('HTTP_AUTHORIZATION')) if token.code != 0: return response.json(token.code) user_id = token.userID if operation == 'getAlexaAppURLAndLWAFallbackURL': # 获取Alexa App和LWA fallback URL return self.get_alexa_app_url_and_lwa_fallback_url(response) elif operation == 'accountLinkWithAmazonAuthorizationCode': # 通过亚马逊授权码连接账号 return self.account_link_with_amazon_authorization_code(user_id, request_dict, response) elif operation == 'disableSkillAndUnlinkAccount': # 取消连接skill和账号 return self.disable_skill_and_unlink_account(user_id, response) elif operation == 'getSkillPageURL': # 获取skill页面URL(取消链接) return self.get_skill_page_url(response) else: return response.json(414) @staticmethod def get_alexa_app_url_and_lwa_fallback_url(response): client_id = 'amzn1.application-oa2-client.98a01914518743e481d51115144dafb0' # Alexa Client Id skill_stage = 'development' # 开发中: development, 已上线: live redirect_uri = 'https://smart.loocam2.com' alexa_app_url = 'https://alexa.amazon.com/spa/skill-account-linking-consent?' \ 'fragment=skill-account-linking-consent&client_id={}&' \ 'scope=alexa::skills:account_linking&skill_stage={}&response_type=code&' \ 'redirect_uri={}'.format(client_id, skill_stage, redirect_uri) lwa_fallback_url = 'https://www.amazon.com/ap/oa?' \ 'client_id={}&scope=alexa::skills:account_linking&response_type=code&redirect_uri={}&'.\ format(client_id, redirect_uri) res = { 'alexa_app_url': alexa_app_url, 'lwa_fallback_url': lwa_fallback_url } return response.json(0, res) @staticmethod def account_link_with_amazon_authorization_code(user_id, request_dict, response): amazon_authorization_code = request_dict.get('amazon_authorization_code', None) if not amazon_authorization_code: return response.json(444) now_time = int(time.time()) # 获取亚马逊访问令牌 # https://developer.amazon.com/zh/docs/login-with-amazon/authorization-code-grant.html#access-token-request amazon_base_uri = 'https://api.amazon.com' url = amazon_base_uri + '/auth/o2/token' redirect_uri = 'https://smart.loocam2.com' data = { 'grant_type': 'authorization_code', 'code': amazon_authorization_code, 'client_id': 'amzn1.application-oa2-client.98a01914518743e481d51115144dafb0', 'client_secret': '43353cac67670aefd64a5f95309754ddd6bcfe8a087cc3cad1348b626f64b132', 'redirect_uri': redirect_uri } try: r = requests.post(url=url, data=data, timeout=10) assert r.status_code == 200 res_data = eval(r.content) assert res_data.get('access_token') assert res_data.get('refresh_token') amazon_access_token = res_data['access_token'] amazon_refresh_token = res_data['refresh_token'] # 保存令牌数据 alexa_oauth_qs = AlexaOauth.objects.filter(user_id=user_id) if alexa_oauth_qs.exists(): alexa_oauth_qs.update(amazon_access_token=amazon_access_token, amazon_refresh_token=amazon_refresh_token, update_time=now_time) else: AlexaOauth.objects.create(user_id=user_id, amazon_access_token=amazon_access_token, amazon_refresh_token=amazon_refresh_token, create_time=now_time, update_time=now_time) # 获取用户授权码 url = redirect_uri + '/appToApp/oa2/getAuthCode' region_code = 'EU' if CONFIG_INFO != CONFIG_EUR: region_code = 'US' params = { 'user_id': user_id, 'region_code': region_code } r = requests.get(url=url, params=params, timeout=10) assert r.status_code == 200 res = eval(r.content) user_authorization_code = res['res']['user_authorization_code'] data = { "stage": "development", "accountLinkRequest": { "redirectUri": redirect_uri, "authCode": user_authorization_code, "type": "AUTH_CODE" } } # 请求连接skill # https://developer.amazon.com/en-US/docs/alexa/smapi/skill-enablement.html headers = { 'Content-Type': 'application/json', 'Authorization': 'Bearer {}'.format(amazon_access_token) } alexa_api_endpoint_list = ['api.amazonalexa.com', 'api.eu.amazonalexa.com', 'api.fe.amazonalexa.com'] skill_id = 'amzn1.ask.skill.ff5a5074-7ec7-442b-979b-cb57095f7a94' for alexa_api_endpoint in alexa_api_endpoint_list: url = 'https://{}/v1/users/~current/skills/{}/enablement'.format(alexa_api_endpoint, skill_id) r = requests.post(headers=headers, url=url, json=data, timeout=30) if r.status_code == 201: AlexaOauth.objects.filter(user_id=user_id).update(alexa_api_endpoint=alexa_api_endpoint) res_data = eval(r.content) return response.json(0, res_data) return response.json(0) except Exception as e: return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e))) @staticmethod def disable_skill_and_unlink_account(user_id, response): alexa_oauth_qs = AlexaOauth.objects.filter(user_id=user_id).values('alexa_api_endpoint', 'amazon_refresh_token') if not alexa_oauth_qs: return response.json(173) now_time = int(time.time()) try: # 使用刷新令牌获取新的访问令牌 # https://developer.amazon.com/zh/docs/login-with-amazon/authorization-code-grant.html#using-refresh-tokens alexa_api_endpoint = alexa_oauth_qs[0]['alexa_api_endpoint'] amazon_refresh_token = alexa_oauth_qs[0]['amazon_refresh_token'] amazon_base_uri = 'https://api.amazon.com' url = amazon_base_uri + '/auth/o2/token' data = { 'grant_type': 'refresh_token', 'refresh_token': amazon_refresh_token, 'client_id': 'amzn1.application-oa2-client.98a01914518743e481d51115144dafb0', 'client_secret': '43353cac67670aefd64a5f95309754ddd6bcfe8a087cc3cad1348b626f64b132' } r = requests.post(url=url, data=data, timeout=10) assert r.status_code == 200 res_data = eval(r.content) assert res_data.get('access_token') assert res_data.get('refresh_token') new_access_token = res_data['access_token'] new_refresh_token = res_data['refresh_token'] alexa_oauth_qs.update(amazon_access_token=new_access_token, amazon_refresh_token=new_refresh_token, update_time=now_time) headers = { 'Content-Type': 'application/json', 'Authorization': 'Bearer {}'.format(new_access_token) } skill_id = 'amzn1.ask.skill.ff5a5074-7ec7-442b-979b-cb57095f7a94' url = 'https://{}/v1/users/~current/skills/{}/enablement'.format(alexa_api_endpoint, skill_id) r = requests.delete(headers=headers, url=url, timeout=30) # 2xx响应状态码为成功 assert str(r.status_code)[:1] == '2' return response.json(0) except Exception as e: return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e))) @staticmethod def get_skill_page_url(response): skill_asin = 'B0C94Q7H1L' skill_page_url = 'https://alexa.amazon.com/spa/index.html#skills/dp/{}'.format(skill_asin) lwa_page_url = 'https://www.amazon.com/dp/{}'.format(skill_asin) res = { 'skill_page_url': skill_page_url, 'lwa_page_url': lwa_page_url } return response.json(0, res)