双引擎语音识别

export WILDFLOW_API_KEY="替换为控制台生成的密钥"
export INPUT_MEDIA="/绝对路径/音视频文件.mp4"

ffmpeg -hide_banner -y \
  -i "$INPUT_MEDIA" \
  -vn -ac 1 -ar 16000 -sample_fmt s16 -c:a flac \
  input.flac

export INPUT_SHA256="$(shasum -a 256 input.flac | awk '{print $1}')"

curl --fail-with-body --silent --show-error \
  -X POST https://api.wildflow.cn/v1/input-artifacts \
  -H "Authorization: Bearer $WILDFLOW_API_KEY" \
  -H "Content-Type: audio/flac" \
  -H "X-WildFlow-Content-SHA256: $INPUT_SHA256" \
  --data-binary @input.flac \
  --output input-artifact.json

export INPUT_ARTIFACT_ID="$(python3 -c \
  'import json; print(json.load(open("input-artifact.json"))["artifact"]["id"])')"

export WILDFLOW_IDEMPOTENCY_KEY="$(python3 -c 'import uuid; print(uuid.uuid4())')"

python3 - "$INPUT_ARTIFACT_ID" > asr-request.json <<'PY'
import json
import sys

print(json.dumps({
    "model": "wildflow/dual-asr-v1",
    "input_artifact_ids": [sys.argv[1]],
    "parameters": {
        "language": "zh",
        "context": "行业访谈与产品讨论",
        "hotwords": ["在这里填写品牌名", "在这里填写专业术语"]
    }
}, ensure_ascii=False))
PY

curl --fail-with-body --silent --show-error \
  -D asr-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 @asr-request.json \
  --output operation.json

export RESULT_FILE="dual-asr-result.json"
export RESULT_MEDIA_TYPE="application/json"

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

{
  "schema_version": 1,
  "source_artifact_id": "输入 Artifact ID",
  "timeline": {"source_offset_seconds": 0},
  "engines": {
    "segment_text": {
      "model_id": "microsoft/VibeVoice-ASR",
      "model_revision": "固定 revision",
      "segments": []
    },
    "word_timestamps": {
      "model_id": "Systran/faster-whisper-large-v3",
      "model_revision": "固定 revision",
      "segments": []
    }
  }
}