微信小程序接口

Ansjer/server_urls/wechat_mini_program_url.py

@@ -0,0 +1,10 @@
+# @Author    : Rocky
+# @File      : wechat_mini_program_url.py
+# @Time      : 2024/11/27 16:30
+from django.urls import re_path
+
+from Controller.WeChatMiniProgram import UserController
+
+urlpatterns = [
+    re_path(r'^login/(?P<operation>.*)$', UserController.UserView.as_view()),
+]

Controller/WeChatMiniProgram/UserController.py

@@ -0,0 +1,97 @@
+# @Author    : Rocky
+# @File      : UserController.py
+# @Time      : 2024/11/27 16:35
+import requests
+from Ansjer.config import LOGGER
+from django.views.generic.base import View
+
+from Object.Enums.WeChatEnum import WeChatMiniProgramAPIEnum, WeChatMiniProgramConfigEnum
+from Object.RedisObject import RedisObject
+from Object.ResponseObject import ResponseObject
+
+
+class EquipmentFamilyView(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)
+        access_token = self.get_access_token()
+        if operation == 'getPhoneNumber':
+            return self.get_phone_number(request_dict, response)
+        else:
+            return response.json(414)
+
+    @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):
+        """
+        获取手机号码
+        @param request_dict:
+        @param response:
+        @return: phone_number
+        """
+        code = request_dict.get('code', None)
+        if not code:
+            return response.json(444)
+        try:
+            access_token = cls.get_access_token()
+            url = WeChatMiniProgramAPIEnum.getPhoneNumberAPI.value.format(access_token)
+            data = {
+                'code': code
+            }
+            r = requests.post(url=url, data=data, timeout=5)
+            result = eval(r.content)
+            errcode = result.get('errcode')
+            assert errcode == 0
+            pure_phone_number = result['phone_info']['purePhoneNumber']
+            res = {
+                'purePhoneNumber': pure_phone_number
+            }
+            return response.json(0, res)
+        except Exception as e:
+            LOGGER.info('微信小程序获取手机号码异常:error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))
+            return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))
+

Object/Enums/WeChatEnum.py

@@ -0,0 +1,18 @@
+# @Author    : Rocky
+# @File      : WeChatEnum.py
+# @Time      : 2024/11/27 17:06
+from enum import IntEnum, unique
+
+
+@unique
+class WeChatMiniProgramConfigEnum(IntEnum):
+    AppID = 'wx21134afc92f1c08c'
+    AppSecret = '4e3758b1f0b2b5514e6f651616ece824'
+    AccessTokenKey = 'WeChatMiniProgramAccessToken'
+
+
+@unique
+class WeChatMiniProgramAPIEnum(IntEnum):
+    getAccessTokenAPI = 'https://api.weixin.qq.com/cgi-bin/token'
+    getPhoneNumberAPI = 'https://api.weixin.qq.com/wxa/business/getuserphonenumber?access_token={}'
+