即時轉錄

import json
import sys
import threading
import websocket

url = "wss://api.agtcloud.cn/v1/asr/realtime"
stop = threading.Event()
sender = None
ws = websocket.create_connection(url, timeout=40, enable_multithread=True)

def send_audio():
    try:
        with open(sys.argv[1], "rb") as audio:
            while not stop.is_set():
                chunk = audio.read(4800)
                if not chunk:
                    break
                if len(chunk) % 2:
                    raise ValueError("PCM 位元組數必須是偶數")
                ws.send_binary(chunk)
                if stop.wait(len(chunk) / 48000):
                    return
        if not stop.is_set():
            silence = bytes(4800)
            for _ in range(10):
                ws.send_binary(silence)
                if stop.wait(0.1):
                    return
            ws.send(json.dumps({"type": "endStream"}))
    except Exception as exc:
        print("傳送中斷：", exc)
        stop.set()
        ws.close()

try:
    ws.send(json.dumps({
        "authorization": {"accessToken": "Bearer your-api-key"},
        "model": "muse-voice-transcribe-1.0",
        "audioEncoding": "PCM_24KHZ",
        "mode": "ENDPOINTING",
        "partialMode": "CUMULATIVE",
        "languageBias": ["Mandarin Chinese"],
        "emitAudioProgress": True
    }))
    ack = json.loads(ws.recv())
    if ack.get("type") == "error" or not ack.get("sessionId"):
        raise RuntimeError("連線初始化失敗：" + json.dumps(ack, ensure_ascii=False))
    print("sessionId:", ack["sessionId"])
    sender = threading.Thread(target=send_audio, daemon=True)
    sender.start()
    while not stop.is_set():
        message = ws.recv()
        if not message:
            break
        event = json.loads(message)
        print(json.dumps(event, ensure_ascii=False))
        if event.get("type") == "error":
            break
finally:
    stop.set()
    ws.close()
    if sender is not None:
        sender.join(timeout=2)