ASJServer/Controller/WeChatMiniProgram/CommonController.py

# @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