ASJServer/Controller/WeChatMiniProgram/UserController.py

# @Author    : Rocky
# @File      : UserController.py
# @Time      : 2024/11/27 16:35
import hashlib
import os
import time

import requests
from Ansjer.config import LOGGER
from django.views.generic.base import View

from Model.models import WeChatMiniProgram
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)
        if operation == 'getPhoneNumber':
            return self.get_phone_number(request_dict, response)
        elif operation == 'login':
            return self.login(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: res
        """
        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 = {
                'pure_phone_number': 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)))

    @classmethod
    def login(cls, request_dict, response):
        """
        小程序登录
        https://developers.weixin.qq.com/miniprogram/dev/OpenApiDoc/user-login/code2Session.html
        https://developers.weixin.qq.com/miniprogram/dev/framework/open-ability/login.html
        @param request_dict:
        @param response:
        @return:
        """
        js_code = request_dict.get('js_code', None)
        if not js_code:
            return response.json(444)
        try:
            params = {
                'js_code': js_code,
                'grant_type': 'authorization_code',
                'appid': WeChatMiniProgramConfigEnum.AppID.value,
                'secret': WeChatMiniProgramConfigEnum.AppSecret.value
            }
            r = requests.get(url=WeChatMiniProgramAPIEnum.code2SessionAPI.value, params=params, timeout=5)
            result = eval(r.content)
            errcode = result.get('errcode')
            assert errcode == 0
            openid = result['openid']
            unionid = result['unionid']
            session_key = result['session_key']
            # 生成登录态
            login_status = cls.generate_login_status(openid, session_key)
            now_time = int(time.time())
            WeChatMiniProgram.objects.create(
                login_status=login_status, openid=openid, unionid=unionid, session_key=session_key,
                created_time=now_time, updated_time=now_time
            )
            res = {
                'login_status': login_status
            }
            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)))

    @staticmethod
    def generate_login_status(openid, session_key):
        """
        生成登录态
        @param openid:
        @param session_key:
        @return: login_status
        """
        salt = os.urandom(16).hex()
        # 将openid, session_key和salt拼接成一个字符串
        to_hash = openid + session_key + salt
        # 使用hashlib生成SHA256哈希值
        hash_object = hashlib.sha256(to_hash.encode())
        # 获取十六进制格式的哈希值
        login_status = hash_object.hexdigest()
        return login_status