VoxCPM2 语音合成

export WILDFLOW_API_KEY="替换为控制台生成的密钥"
export WILDFLOW_IDEMPOTENCY_KEY="$(python3 -c 'import uuid; print(uuid.uuid4())')"

curl --fail-with-body --silent --show-error -D response-headers.txt -X POST https://api.wildflow.cn/v1/jobs \
  -H "Authorization: Bearer $WILDFLOW_API_KEY" \
  -H "Idempotency-Key: $WILDFLOW_IDEMPOTENCY_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "VoxCPM2",
    "parameters": {
      "input": "你好，欢迎使用野生流动。",
      "voice": "default",
      "speed": 1.0
    }
  }' \
  --output operation.json

{
  "model": "VoxCPM2",
  "parameters": {
    "input": "你好，欢迎使用野生流动。",
    "voice": "default",
    "speed": 1.0
  }
}

set "WILDFLOW_API_KEY=替换为控制台新生成的密钥"
set "WILDFLOW_IDEMPOTENCY_KEY=voxcpm2-request-001"

curl.exe --fail-with-body -sS -D response-headers.txt -X POST "https://api.wildflow.cn/v1/jobs" ^
  -H "Authorization: Bearer %WILDFLOW_API_KEY%" ^
  -H "Idempotency-Key: %WILDFLOW_IDEMPOTENCY_KEY%" ^
  -H "Content-Type: application/json" ^
  --data-binary "@request.json" ^
  --output operation.json

set "OPERATION_ID=把 operation.json 中的 id 粘贴到这里"

curl.exe --fail-with-body -sS ^
  -H "Authorization: Bearer %WILDFLOW_API_KEY%" ^
  "https://api.wildflow.cn/v1/jobs/%OPERATION_ID%" ^
  --output status.json

set "ARTIFACT_ID=把 status.json 中 artifacts[0].id 粘贴到这里"

curl.exe --fail-with-body -sS ^
  -H "Authorization: Bearer %WILDFLOW_API_KEY%" ^
  "https://api.wildflow.cn/v1/artifacts/%ARTIFACT_ID%/content" ^
  --output result.mp3

export RESULT_FILE="result.mp3"
export RESULT_MEDIA_TYPE="audio/mpeg"

python3 - <<'PY'
import hashlib
import json
import os
import time
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from pathlib import Path
from urllib.error import HTTPError
from urllib.parse import quote
from urllib.request import Request, urlopen

base = "https://api.wildflow.cn"
headers = {"Authorization": "Bearer " + os.environ["WILDFLOW_API_KEY"]}
operation = json.loads(Path("operation.json").read_text())
operation_id = operation["id"]
status_url = base + "/v1/jobs/" + quote(operation_id, safe="")
output = Path(os.environ["RESULT_FILE"])
expected_media_type = os.environ["RESULT_MEDIA_TYPE"]
deadline = time.monotonic() + 7200

def pause(value):
    try:
        seconds = float(value)
    except (TypeError, ValueError):
        try:
            seconds = (parsedate_to_datetime(value) - datetime.now(timezone.utc)).total_seconds()
        except (TypeError, ValueError, OverflowError):
            seconds = 3
    seconds = max(1, seconds)
    if time.monotonic() + seconds >= deadline:
        raise SystemExit("等待超时。保留 operation.json，稍后重新执行本段继续查询。")
    time.sleep(seconds)

while time.monotonic() < deadline:
    try:
        with urlopen(Request(status_url, headers=headers), timeout=60) as response:
            status = json.load(response)
            retry_after = response.headers.get("Retry-After", "3")
    except HTTPError as error:
        if error.code in (429, 503):
            pause(error.headers.get("Retry-After", "3"))
            continue
        raise SystemExit(f"查询失败：HTTP {error.code}。保留原任务，检查错误后继续查询。")

    state = status["state"]
    print("任务状态：", state, flush=True)
    if state == "succeeded":
        artifacts = status.get("artifacts", [])
        if not artifacts:
            raise SystemExit("任务成功但缺少结果文件，请保留任务 ID 并联系支持。")
        artifact = artifacts[0]
        if artifact["media_type"] != expected_media_type:
            raise SystemExit("结果格式与预期不同，请检查任务和模型。")
        download_url = base + "/v1/artifacts/" + quote(artifact["id"], safe="") + "/content"
        temporary = output.with_name(output.name + ".part")
        digest = hashlib.sha256()
        size = 0
        with urlopen(Request(download_url, headers=headers), timeout=60) as response:
            with temporary.open("wb") as file:
                while chunk := response.read(1024 * 1024):
                    file.write(chunk)
                    digest.update(chunk)
                    size += len(chunk)
        if size != artifact["size_bytes"] or digest.hexdigest() != artifact["sha256"]:
            raise SystemExit("文件校验失败，保留原任务并重新执行本段下载。")
        temporary.replace(output)
        print("已保存：", output)
        break
    if state not in ("queued", "running", "submitting"):
        raise SystemExit(f"任务已停止：{state}。请保留任务 ID；不要换幂等键重复提交。")
    pause(retry_after)
else:
    raise SystemExit("等待超时。保留 operation.json，稍后继续查询原任务。")
PY

ffprobe -v error \
  -show_entries stream=codec_name,sample_rate,channels,bit_rate:format=duration \
  -of json result.mp3