Spaces:
Runtime error
Runtime error
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 +"×tamp="+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 | |