QingLong/AI/WxMini/Utils/TokenUtil.py

import json
from datetime import datetime

from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.request import CommonRequest

from WxMini.Milvus.Config.MulvusConfig import *
from WxMini.Utils.RedisUtil import set_tts_token, get_tts_token


def getToken():
    # 从Redis获取 TTS Token
    retrieved_token = get_tts_token()
    if retrieved_token:
        #print("使用Redis中的Token:", retrieved_token)
        return retrieved_token
    # 创建AcsClient实例
    client = AcsClient(
        ACCESS_KEY_ID, ACCESS_KEY_SECRET, "cn-shanghai"
    )

    # 创建request，并设置参数
    request = CommonRequest()
    request.set_method('POST')
    request.set_domain('nls-meta.cn-shanghai.aliyuncs.com')
    request.set_version('2019-02-28')
    request.set_action_name('CreateToken')

    try:
        response = client.do_action_with_exception(request)
        jss = json.loads(response)
        if 'Token' in jss and 'Id' in jss['Token']:
            token = jss['Token']['Id']
            expireTime = jss['Token']['ExpireTime']
            # 转换为本地时间
            expire_date = datetime.fromtimestamp(expireTime)
            # 格式化输出
            formatted_date = expire_date.strftime("%Y-%m-%d %H:%M:%S")
            print("过期时间:", formatted_date)
            # 计算时间差（秒数）
            now = datetime.now()
            time_diff = (expire_date - now).total_seconds()
            print("距离过期还有（秒）:", time_diff)
            # 设置 TTS Token
            set_tts_token(token)
            return token
    except Exception as e:
        print(e)