api_get.py

import requests
import json
import hmac
import hashlib
import time
import base64
import urllib
import urllib3
import gradio as gr
# 加密函数
def hmac_sha256_encrypt(message, key):
    message = message.encode('utf-8')
    key = key.encode('utf-8')
    # 计算HMAC值
    hashed = hmac.new(key, message, hashlib.sha256).digest()
    return hashed
    
# // 声明accessToken
appkey = '4e8f2e8492c947da9c3738b74b4ac2fa'
accessToken = '6b4c98203218457fbe276e13990f3550';
reqTimestamp = "";
reqSignature = '';
def  pre_request():
#   // 获取当前时间戳


  timestamp = str (round(time.time()));
  # timestamp = str (1691034072);
  
#   // 设置本次请求的签名时使用的时间戳，单位：秒，时间要求在 UTC/GMT 零时区时间的前后 5 分钟内
  global reqTimestamp
  reqTimestamp = timestamp;
#   // 获取待计算签名的内容
  signingContent = "appkey="+ appkey +"&timestamp="+timestamp;
  
#   // 计算 HMAC-SHA256 值
  hash = hmac_sha256_encrypt(signingContent, accessToken);
  
#   // 将签名内容转换16进制
  hashInBase64 = base64.b64encode(hash).decode('utf-8');
# url encode  签名串  用于拼接请求地址        
  hashInBase64.encode('utf-8')
  encodeSign = urllib.parse.quote_plus(hashInBase64)

#   // 设置本次请求的签名签名串
  global reqSignature
  reqSignature = encodeSign;
  # print(reqTimestamp, reqSignature, appkey)

def req_audioMake(text):
  pre_request()
  url = "https://gw.tvs.qq.com/v2/ivh/videomaker/broadcastservice/tts?timestamp="+reqTimestamp+"&signature="+reqSignature+"&appkey="+appkey

  payload = json.dumps({
    "Header": {},
    "Payload": {
      "TimbreKey": "",
      "VirtualmanKey": "36b88108f6de6da255534eaf4363c7d7",
      "InputSsml": text,
      "Speed": 1
    }
  })
  headers = {
    'Content-Type': 'application/json'
  }

  response = requests.request("POST", url, headers=headers, data=payload)
  
  res = json.loads(response.text)
  if res["Header"]["Code"] != 0:
    print(response.text)
    return 

  taskID = (res["Payload"]["TaskId"])
  print(taskID)
  return taskID

def req_videoMake(text = "你好"):
    pre_request()
    taskID = ""

    # 调用视颯合户接口，获取taskID
    # 调用视颯合户接口，获取taskID
    url = "https://gw.tvs.qq.com/v2/ivh/videomaker/broadcastservice/videomake?timestamp="+reqTimestamp+"&signature="+reqSignature+"&appkey="+appkey
    payload = json.dumps({
    "Header": {},
    "Payload": {
        "VirtualmanKey": "36b88108f6de6da255534eaf4363c7d7",
        "InputSsml": text,
        "SpeechParam": {
        "Speed": 1,
        "TimbreKey": "female_1"
        },
        "VideoParam": {
        "Format": "GreenScreenMp4"
        }
    }
    })
    

    headers = {
    'Content-Type': 'application/json'
    }

    response = requests.request("POST", url, headers=headers, data=payload)

    res = json.loads(response.text)
    if res["Header"]["Code"] != 0:
      print(response.text)
      return 

    taskID = (res["Payload"]["TaskId"])
    print(taskID)
    return taskID



def get_media(taskID):

  pre_request()
  url = "https://gw.tvs.qq.com/v2/ivh/videomaker/broadcastservice/getprogress?timestamp="+reqTimestamp+"&signature="+reqSignature+"&appkey="+appkey

  payload = json.dumps({
    "Header": {},
    "Payload": {
      "TaskId": taskID
    }
  })
  headers = {
    'Content-Type': 'application/json'
  }

  response = requests.request("POST", url, headers=headers, data=payload)

  # print(response.text)
  return response.text


def generate_video(text = "你好"):
    taskID = req_videoMake(text)
    # print(taskID)

    sleepSecond = 5
    time.sleep(sleepSecond)
    videoRep = json.loads(get_media(taskID))
 
    status = videoRep["Payload"]["Status"]

    while status == "COMMIT" or status == "MAKING":
        time.sleep(1)
        videoRep = json.loads(get_media(taskID))
        status = videoRep["Payload"]["Status"]
        print(status)

    if status == "FAIL":
        print(videoRep)
        return
 
    mediaUrl = videoRep["Payload"]["MediaUrl"]

    return mediaUrl


def generate_audio(text):
    taskID = req_audioMake(text)
    # print(taskID)

    sleepSecond = 5
    time.sleep(sleepSecond)
    videoRep = json.loads(get_media(taskID))
 
    status = videoRep["Payload"]["Status"]

    while status == "COMMIT" or status == "MAKING":
        time.sleep(1)
        videoRep = json.loads(get_media(taskID))
        status = videoRep["Payload"]["Status"]
        print(status)

    if status == "FAIL":
        print(videoRep)
        return
 
    mediaUrl = videoRep["Payload"]["MediaUrl"]

    return mediaUrl