import os
import time
import base64
import requests
def _ensure_dir(path: str) -> None:
directory = os.path.dirname(path)
if directory and not os.path.exists(directory):
os.makedirs(directory, exist_ok=True)
def _save_bytes_to_file(data: bytes, save_path: str) -> str:
_ensure_dir(save_path)
with open(save_path, "wb") as f:
f.write(data)
return os.path.abspath(save_path)
def call_text_to_clone_voice_v3(
base_url: str,
key: str,
source_audio_url: str,
target_text: str,
save_dir: str = "testdemo/outputs",
speed: str = "1.0",
pitch: str = "1.0",
ref_text: str = "我的声音将用于平台克隆,并合法使用,为自己的行为负责",
req_type: int = 2,
notify_url: str = None,
speakerid: str = None,
audio_format: str = "mp3",
) -> str:
"""调用 /api/texttoclonevoicev3/index 并将结果保存到本地或返回任务ID。
req_type:
- 2: 同步返回音频的base64,直接落盘
- 1: 异步模式,必须提供 notify_url,本函数返回任务ID
"""
audio_format = (audio_format or "mp3").lower()
if audio_format not in ("mp3", "wav"):
audio_format = "mp3"
endpoint = f"{base_url.rstrip('/')}/api/texttoclonevoicev3/index"
payload = {
"key": key,
"source_audio_url": source_audio_url,
"target_text": target_text,
"speed": str(speed),
"pitch": str(pitch),
"ref_text": ref_text,
"type": str(req_type),
}
if speakerid:
payload["speakerid"] = speakerid
if req_type == 1:
if not notify_url:
raise ValueError("notify_url is required when type=1 (async mode)")
payload["notify_url"] = notify_url
resp = requests.post(endpoint, data=payload, timeout=120)
resp.raise_for_status()
result = resp.json()
if result.get("code") != 200:
raise RuntimeError(f"API error: {result}")
data = result.get("data") or {}
timestamp = int(time.time())
if req_type == 2:
audio_b64 = data.get("audio_base64")
if not audio_b64:
raise RuntimeError("Missing audio_base64 in response for type=2")
audio_bytes = base64.b64decode(audio_b64)
save_path = os.path.join(save_dir, f"texttoclonevoicev3_{timestamp}.{audio_format}")
return _save_bytes_to_file(audio_bytes, save_path)
# type == 1
taskid = data.get("taskid")
if not taskid:
raise RuntimeError("Missing taskid in response for type=1")
return str(taskid)
if __name__ == "__main__":
# 仅保留:调用 texttoclonevoicev3 接口(同步返回base64并落盘)
BASE_URL_V3 = "https://api.hihookeji.com"
API_KEY_V3 = "平台申请"
REF_TEXT_V3 = "我的声音将用于平台克隆,并合法使用,为自己的行为负责"
EXAMPLE_SOURCE_AUDIO_URL = "https://example.mp3"
TARGET_TEXT_3 = "这是声音克隆3同步合成示例。"
try:
out3 = call_text_to_clone_voice_v3(
base_url=BASE_URL_V3,
key=API_KEY_V3,
source_audio_url=EXAMPLE_SOURCE_AUDIO_URL,
target_text=TARGET_TEXT_3,
speed="1.0",
pitch="1.0",
ref_text=REF_TEXT_V3,
req_type=2
)
print(f"/api/texttoclonevoicev3/index 已保存到: {out3}")
except Exception as e:
print("/api/texttoclonevoicev3/index 调用失败:", e)
客服微信
请打开手机微信,扫一扫联系我们

返回顶部