ASJServer/Controller/WeChatMiniProgram/UserController.py

# @Author    : Rocky
# @File      : UserController.py
# @Time      : 2024/11/27 16:35
import hashlib
import hmac
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:
            valid = self.check_session_key(request)
            if not valid:
                return response.json(309)

    @classmethod
    def check_session_key(cls, request):
        """
        校验 session_key 是否有效
        https://developers.weixin.qq.com/miniprogram/dev/OpenApiDoc/user-login/checkSessionKey.html
        @return: bool
        """
        try:
            login_status = request.META.get('HTTP_AUTHORIZATION')
            if not login_status:
                return False
            wechat_mini_program_qs = WeChatMiniProgram.objects.filter(login_status=login_status). \
                values('openid', 'session_key')
            if not wechat_mini_program_qs.exists():
                return False
            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()
            params = {
                'openid': openid,
                'access_token': access_token,
                'signature': signature,
                'sig_method': 'hmac_sha256'
            }
            r = requests.get(url=WeChatMiniProgramAPIEnum.checkSessionKeyAPI.value, params=params, timeout=5)
            result = eval(r.content)
            errcode = result.get('errcode')
            assert errcode == 0
            return True
        except Exception as e:
            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):
        """
        获取手机号码
        @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())
            wechat_mini_program_qs = WeChatMiniProgram.objects.filter(openid=openid)
            if wechat_mini_program_qs.exists():
                wechat_mini_program_qs.update(
                    login_status=login_status, session_key=session_key, updated_time=now_time
                )
            else:
                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

    @staticmethod
    def generate_signature(session_key):
        """
        用户登录态签名，用session_key对空字符串签名得到的结果。即 signature = hmac_sha256(session_key, "")
        @param session_key:
        @return: signature
        """
        # 将session_key和要签名的数据（这里是空字符串）转换为字节类型
        session_key_bytes = session_key.encode('utf-8')
        data = b""  # 空字符串转换为字节类型
        # 创建一个新的hmac对象，使用sha256作为hash函数
        hmac_object = hmac.new(session_key_bytes, data, hashlib.sha256)
        # 获取十六进制格式的签名结果
        signature = hmac_object.hexdigest()
        return signature