Spaces:
Running
Running
File size: 3,653 Bytes
6784ca6 eaeab3e 6784ca6 ef926c5 6784ca6 ef926c5 6784ca6 8da761a 6784ca6 ef926c5 6784ca6 ef926c5 6784ca6 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 |
import requests
import time
import uuid
import os
from datetime import datetime
from supabase import create_client, Client
try:
OneKey = os.environ['OneKey'].strip()
OneKey = OneKey.split("#")
TrumpAiUrl = OneKey[0]
ApiKey = OneKey[1]
SUPABASE_URL = OneKey[2]
UserUuid = OneKey[3]
BackendUrl = OneKey[4]
BackendApiKey = OneKey[5]
SUPABASE_KEY = OneKey[6]
# GRADIO_ALLOWED_HOSTNAMES = OneKey[7]
# os.environ["GRADIO_ALLOWED_HOSTNAMES"] = GRADIO_ALLOWED_HOSTNAMES
except Exception as e:
print(f"OneKey: {e}")
# exit(1)
# 创建Supabase客户端
supabase: Client = create_client(SUPABASE_URL, SUPABASE_KEY)
# 任务状态枚举
class TaskStatus:
Created = "created"
Processing = "processing"
TextRewrited = "text_rewrited"
TextFormated = "text_formated"
VoiceCompleted = "voice_completed"
VoiceError = "voice_error"
VideoCompleted = "video_completed"
VideoPublished = "video_published"
VideoError = "video_error"
Completed = "completed"
Failed = "failed"
Cancelled = "cancelled"
NoCredits = "no_credits"
# 视频模板枚举
class VideoTemplate:
Outdoor = "outdoor"
UsFlag = "us-flag"
WhiteHouse = "white-house"
def create_task_v3(task_type, text, word_num, is_rewrite, language="en"):
import json
is_rewrite = False
url = f"{BackendUrl}/trump_process_ctx_api_v3"
headers = {
"Content-Type": "application/json"
}
print(url)
data = {
"video_template": "",
"speaker_template": "https://www.trumpaivoice.net/SelfitAssert/Heygem/Trump/trump_shot01.MP3",
"text": text,
"word_num": word_num,
"is_rewrite": False,
"language": language,
"watermark": True,
"type": "voice",
"cost_credits": 2,
"user_uuid": UserUuid,
"secret_key": "219ngu"
}
try:
resp = requests.post(url, headers=headers, data=json.dumps(data), timeout=60)
if not resp.ok:
print(f"调用trump_process_ctx_api失败: {resp.status_code} {resp.text}")
return None
try:
ctx_json = resp.json()
except Exception as e:
print(f"解析trump_process_ctx_api返回异常: {e}")
return None
if not ctx_json or ctx_json.get("code") != 0 or not ctx_json.get("data") or not ctx_json["data"].get("task_id"):
print(f"trump_process_ctx_api返回异常: {ctx_json}")
return None
return {
"task_id": ctx_json["data"]["task_id"],
"uuid": ctx_json["data"]["task_uuid"],
"status": "created",
"message": "任务创建成功，后台处理中"
}
except Exception as err:
print(f"create_task_v3异常: {err}")
return None
def get_task_result(task_id):
# Poll for task status and result
url = f"{TrumpAiUrl}/api/task-status/uuid/{task_id}"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {ApiKey}"
}
print(url)
try:
resp = requests.get(url, headers=headers, timeout=30)
resp.raise_for_status()
result = resp.json()
return result
except Exception as e:
return {}
if __name__ == "__main__":
task_type = "voice"
text = "Hello, this is a test message for Trump AI Voice."
word_num = 10
is_rewrite = True
task_result = create_task_v3(task_type, text, word_num, is_rewrite)
print(f"task_result: {task_result}")
|