小程序获取设备票据接口

Ansjer/server_urls/wechat_mini_program_url.py

@@ -3,8 +3,9 @@
 # @Time      : 2024/11/27 16:30
 from django.urls import re_path
 
-from Controller.WeChatMiniProgram import UserController
+from Controller.WeChatMiniProgram import UserController, DeviceController
 
 urlpatterns = [
     re_path(r'^login/(?P<operation>.*)$', UserController.UserView.as_view()),
+    re_path(r'^device/(?P<operation>.*)$', DeviceController.DeviceView.as_view()),
 ]

Controller/WeChatMiniProgram/CommonController.py

@@ -0,0 +1,44 @@
+# @Author    : Rocky
+# @File      : CommonController.py
+# @Time      : 2025/2/19 13:54
+import requests
+
+from Ansjer.config import LOGGER
+from Object.Enums.WeChatEnum import WeChatMiniProgramConfigEnum, WeChatMiniProgramAPIEnum
+from Object.RedisObject import RedisObject
+
+
+class CommonObj:
+    @staticmethod
+    def get_access_token():
+        """
+        获取小程序全局唯一后台接口调用凭据，token有效期为7200s
+        https://developers.weixin.qq.com/miniprogram/dev/OpenApiDoc/mp-access-token/getAccessToken.html
+        @return: access_token
+        """
+        try:
+            redis_obj = RedisObject()
+            access_token_key = WeChatMiniProgramConfigEnum.AccessTokenKey.value
+            expires_time = redis_obj.get_ttl(access_token_key)
+            # 如果有效时间大于十分钟,返回token,否则刷新token
+            if expires_time > 10 * 60:
+                access_token = redis_obj.get_data(access_token_key)
+                return access_token
+            else:
+                params = {
+                    'grant_type': 'client_credential',
+                    'appid': WeChatMiniProgramConfigEnum.AppID.value,
+                    'secret': WeChatMiniProgramConfigEnum.AppSecret.value
+                }
+                r = requests.get(url=WeChatMiniProgramAPIEnum.getAccessTokenAPI.value, params=params, timeout=5)
+                result = eval(r.content)
+                access_token = result.get('access_token')
+                assert access_token
+                expires_in = result.get('expires_in')
+                # 保存到Redis
+                redis_obj.set_ex_data(key=access_token_key, val=access_token, expire=expires_in)
+                return access_token
+        except Exception as e:
+            LOGGER.info('微信小程序获取token异常:error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))
+            return None
+

Controller/WeChatMiniProgram/DeviceController.py

@@ -0,0 +1,60 @@
+# @Author    : Rocky
+# @File      : DeviceController.py.py
+# @Time      : 2025/2/19 11:22
+import requests
+from django.views.generic.base import View
+
+from Controller.WeChatMiniProgram.CommonController import CommonObj
+from Object.Enums.WeChatEnum import WeChatMiniProgramAPIEnum
+from Object.ResponseObject import ResponseObject
+
+
+class DeviceView(View):
+
+    def get(self, request, *args, **kwargs):
+        request.encoding = 'utf-8'
+        operation = kwargs.get('operation')
+        return self.validation(request.GET, request, operation)
+
+    def post(self, request, *args, **kwargs):
+        request.encoding = 'utf-8'
+        operation = kwargs.get('operation')
+        return self.validation(request.POST, request, operation)
+
+    def validation(self, request_dict, request, operation):
+        lang = request_dict.get('lang', 'cn')
+        response = ResponseObject(lang)
+        if operation == 'getSnTicket':
+            return self.get_sn_ticket(request_dict, response)
+
+    @staticmethod
+    def get_sn_ticket(request_dict, response):
+        """
+        获取设备票据
+        https://developers.weixin.qq.com/miniprogram/dev/OpenApiDoc/hardware-device/getSnTicket.html
+        @param request_dict:
+        @param response:
+        @return: res
+        """
+        uid = request_dict.get('uid', None)
+        model_id = request_dict.get('model_id', None)
+        if not all([uid, model_id]):
+            return response.json(444)
+        try:
+            access_token = CommonObj.get_access_token()
+            url = WeChatMiniProgramAPIEnum.getSnTicketAPI.value.format(access_token)
+            data = {
+                'sn': uid,
+                'model_id': model_id
+            }
+            r = requests.post(url=url, json=data, timeout=5)
+            result = eval(r.content)
+            errcode = result.get('errcode')
+            assert errcode == 0
+            sn_ticket = result['sn_ticket']
+            res = {
+                'sn_ticket': sn_ticket
+            }
+            return response.json(0, res)
+        except Exception as e:
+            return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))

Controller/WeChatMiniProgram/UserController.py

@@ -11,6 +11,7 @@ from Ansjer.config import LOGGER
 from Ansjer.config import SERVER_DOMAIN_SSL
 from django.views.generic.base import View
 
+from Controller.WeChatMiniProgram.CommonController import CommonObj
 from Model.models import WeChatMiniProgram, Device_User, Device_Info, DeviceNameLanguage
 from Object.Enums.WeChatEnum import WeChatMiniProgramAPIEnum, WeChatMiniProgramConfigEnum
 from Object.RedisObject import RedisObject
@@ -61,7 +62,7 @@ class UserView(View):
             openid = wechat_mini_program_qs[0]['openid']
             session_key = wechat_mini_program_qs[0]['session_key']
             signature = cls.generate_signature(session_key)
-            access_token = cls.get_access_token()
+            access_token = CommonObj.get_access_token()
             params = {
                 'openid': openid,
                 'access_token': access_token,
@@ -83,42 +84,10 @@ class UserView(View):
             return False
 
     @staticmethod
-    def get_access_token():
-        """
-        获取小程序全局唯一后台接口调用凭据，token有效期为7200s
-        https://developers.weixin.qq.com/miniprogram/dev/OpenApiDoc/mp-access-token/getAccessToken.html
-        @return: access_token
-        """
-        try:
-            redis_obj = RedisObject()
-            access_token_key = WeChatMiniProgramConfigEnum.AccessTokenKey.value
-            expires_time = redis_obj.get_ttl(access_token_key)
-            # 如果有效时间大于十分钟,返回token,否则刷新token
-            if expires_time > 10 * 60:
-                access_token = redis_obj.get_data(access_token_key)
-                return access_token
-            else:
-                params = {
-                    'grant_type': 'client_credential',
-                    'appid': WeChatMiniProgramConfigEnum.AppID.value,
-                    'secret': WeChatMiniProgramConfigEnum.AppSecret.value
-                }
-                r = requests.get(url=WeChatMiniProgramAPIEnum.getAccessTokenAPI.value, params=params, timeout=5)
-                result = eval(r.content)
-                access_token = result.get('access_token')
-                assert access_token
-                expires_in = result.get('expires_in')
-                # 保存到Redis
-                redis_obj.set_ex_data(key=access_token_key, val=access_token, expire=expires_in)
-                return access_token
-        except Exception as e:
-            LOGGER.info('微信小程序获取token异常:error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))
-            return None
-
-    @classmethod
-    def get_phone_number(cls, request_dict, response):
+    def get_phone_number(request_dict, response):
         """
         获取手机号码
+        https://developers.weixin.qq.com/miniprogram/dev/OpenApiDoc/user-info/phone-number/getPhoneNumber.html
         @param request_dict:
         @param response:
         @return: res
@@ -127,7 +96,7 @@ class UserView(View):
         if not code:
             return response.json(444)
         try:
-            access_token = cls.get_access_token()
+            access_token = CommonObj.get_access_token()
             url = WeChatMiniProgramAPIEnum.getPhoneNumberAPI.value.format(access_token)
             data = {
                 'code': code

Object/Enums/WeChatEnum.py

@@ -17,3 +17,4 @@ class WeChatMiniProgramAPIEnum(Enum):
     getPhoneNumberAPI = 'https://api.weixin.qq.com/wxa/business/getuserphonenumber?access_token={}'
     code2SessionAPI = 'https://api.weixin.qq.com/sns/jscode2session'
     checkSessionKeyAPI = 'https://api.weixin.qq.com/wxa/checksession'
+    getSnTicketAPI = 'POST https://api.weixin.qq.com/wxa/getsnticket?access_token={}'