From fd5b0560ce7187ba09ffb54b20b70ef72bde97bf Mon Sep 17 00:00:00 2001 From: lzh <526494747@qq.com> Date: Tue, 22 Sep 2026 10:06:54 +0800 Subject: [PATCH] feat(voice): add Tencent Cloud realtime (streaming) speech recognition Dictate with live partial results: the composer mic streams 16 kHz PCM over a WebSocket to Octop, which signs and proxies the audio to Tencent's realtime ASR V2 endpoint. Credentials never leave the server; the browser only learns a `stt_realtime` boolean from `/api/voice/active`. - settings: add a "realtime speech recognition" switch plus an AppID field to the Tencent Cloud voice drawer. Both live in `voice_providers.extra_json`, so no schema migration is needed. - interaction: with realtime on, the mic becomes press-and-hold (start on pointerdown, finish on release/leave/cancel) and stays available while a reply is streaming, so dictation can be queued. With the switch off the previous click-to-record one-shot upload path is unchanged. - protocol: realtime ASR V2 answers with `result` (`slice_type`, `index`, `voice_text_str`) while the docs describe `sentences`; both are normalized in `infra/voice/realtime.py`. - errors: new `VOICE_REALTIME_UNAVAILABLE` code plus localized copy for the Tencent realtime error codes (4002-4009, 5000-5002). Verified end-to-end against Tencent with real speech (edge-tts -> 16 kHz PCM over the live WebSocket) and by `make all`. --- CHANGELOG.md | 1 + dashboard/src/api/modules/voice.ts | 2 + dashboard/src/hooks/useVoiceConfig.ts | 12 +- dashboard/src/hooks/useVoiceInput.ts | 173 ++++++++++++- dashboard/src/locales/en.json | 7 + dashboard/src/locales/zh.json | 7 + .../components/ChatInput.prefill.test.tsx | 3 + .../src/pages/Chat/components/ChatInput.tsx | 28 ++- .../components/ChatInputActionsRow.test.tsx | 92 +++++++ .../Chat/components/ChatInputActionsRow.tsx | 51 +++- dashboard/src/pages/Settings/Voice/index.tsx | 46 +++- dashboard/src/utils/pcmCapture.test.ts | 71 ++++++ dashboard/src/utils/pcmCapture.ts | 71 ++++++ dashboard/src/utils/realtimeStt.test.ts | 187 ++++++++++++++ dashboard/src/utils/realtimeStt.ts | 175 +++++++++++++ src/octop/api/routers/voice.py | 194 ++++++++++++++- src/octop/i18n/domains/voice.py | 25 +- src/octop/i18n/en.json | 13 + src/octop/i18n/zh.json | 13 + src/octop/infra/errors.py | 2 + src/octop/infra/utils/tencent_asr_sign.py | 70 ++++++ src/octop/infra/voice/adapters.py | 8 +- src/octop/infra/voice/manager.py | 31 ++- src/octop/infra/voice/realtime.py | 217 ++++++++++++++++ tests/integration/test_voice_stt_stream_ws.py | 196 +++++++++++++++ tests/support/http.py | 3 + tests/unit/i18n/test_voice.py | 22 +- tests/unit/test_voice_manager.py | 87 +++++++ tests/unit/test_voice_realtime.py | 231 ++++++++++++++++++ tests/unit/utils/test_tencent_asr_sign.py | 84 +++++++ 30 files changed, 2089 insertions(+), 33 deletions(-) create mode 100644 dashboard/src/utils/pcmCapture.test.ts create mode 100644 dashboard/src/utils/pcmCapture.ts create mode 100644 dashboard/src/utils/realtimeStt.test.ts create mode 100644 dashboard/src/utils/realtimeStt.ts create mode 100644 src/octop/infra/utils/tencent_asr_sign.py create mode 100644 src/octop/infra/voice/realtime.py create mode 100644 tests/integration/test_voice_stt_stream_ws.py create mode 100644 tests/unit/test_voice_realtime.py create mode 100644 tests/unit/utils/test_tencent_asr_sign.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d01e44d13..710dec0f3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,7 @@ - 登录验证码新增极验行为验 v4（#870）：设置页配置 captcha_id / captcha_key，登录弹窗完成验证，服务端按官方协议 HMAC-SHA256 签名后到 gcaptcha4 二次校验（form-urlencoded，仅 result=success 放行） - 对话支持默认折叠思考与工具过程（浏览器本地偏好）(#718) - 对话中可隐藏不常用的共享专家（浏览器本地偏好）(#589) +- 腾讯云语音新增实时（流式）识别：开启后输入框麦克风为按住说话，边说边显示中间结果、松开即出全文；凭据仅保存在服务端，浏览器只收到开关状态；配置存于语音 provider 的 extra_json，无数据库迁移 ### 修复 diff --git a/dashboard/src/api/modules/voice.ts b/dashboard/src/api/modules/voice.ts index 8562a6414..530c0b83c 100644 --- a/dashboard/src/api/modules/voice.ts +++ b/dashboard/src/api/modules/voice.ts @@ -26,6 +26,8 @@ export interface VoiceProviderRow { export interface ActiveVoice { stt: string; tts: string; + /** True when the active STT provider streams results over WebSocket. */ + stt_realtime?: boolean; } export interface VoiceProviderInput { diff --git a/dashboard/src/hooks/useVoiceConfig.ts b/dashboard/src/hooks/useVoiceConfig.ts index 78c315b06..b7fec1d29 100644 --- a/dashboard/src/hooks/useVoiceConfig.ts +++ b/dashboard/src/hooks/useVoiceConfig.ts @@ -30,10 +30,14 @@ export function prefetchVoiceConfig(): void { } export function useVoiceConfig() { - const [active, setActive] = useState({ - stt: "browser", - tts: "browser", - }); + const [active, setActive] = useState( + () => + cachedActiveVoice() ?? { + stt: "browser", + tts: "browser", + stt_realtime: false, + }, + ); const [loading, setLoading] = useState(true); const refresh = useCallback(async () => { diff --git a/dashboard/src/hooks/useVoiceInput.ts b/dashboard/src/hooks/useVoiceInput.ts index 76d2614d5..29ad1f84b 100644 --- a/dashboard/src/hooks/useVoiceInput.ts +++ b/dashboard/src/hooks/useVoiceInput.ts @@ -1,8 +1,16 @@ import { useCallback, useRef, useState } from "react"; import { useTranslation } from "react-i18next"; +import { getWsUrl } from "../api/config"; import { voiceApi, type ActiveVoice } from "../api/modules/voice"; +import { getAuthToken } from "../api/request"; import { speechLocaleFromUi } from "../utils/localePrefs"; -import { cachedActiveVoice, fetchActiveVoice } from "./useVoiceConfig"; +import { PcmFrameBuffer, downsampleToPcm16 } from "../utils/pcmCapture"; +import { RealtimeSttClient } from "../utils/realtimeStt"; +import { + cachedActiveVoice, + fetchActiveVoice, + useVoiceConfig, +} from "./useVoiceConfig"; import { message as antMessage } from "@/utils/antdMessage"; @@ -237,18 +245,147 @@ export function isSttAvailable(): boolean { return browserSttAvailable(); // browser STT (Chrome / Edge) } -export function useVoiceInput(onText: (text: string) => void) { +export interface VoiceInputOptions { + /** Called when a capture session starts, before any transcript arrives. */ + onStart?: () => void; + /** Cumulative transcript while the session is still being refined. */ + onInterim?: (text: string) => void; + /** Cumulative transcript of the session; the realtime path calls it per sentence. */ + onText: (text: string) => void; +} + +/** Live 16 kHz capture feeding the realtime STT WebSocket. */ +interface RealtimeCapture { + client: RealtimeSttClient; + stream: MediaStream; + ctx: AudioContext; + source: MediaStreamAudioSourceNode; + processor: ScriptProcessorNode; + sink: GainNode; +} + +function realtimeSttUrl(language: string): string { + const params = new URLSearchParams({ + locale: language.toLowerCase().startsWith("zh") ? "zh" : "en", + }); + const token = getAuthToken(); + if (token) params.set("token", token); + return `${getWsUrl("/voice/stt-stream")}?${params.toString()}`; +} + +function audioContextCtor(): typeof AudioContext | null { + return ( + window.AudioContext ?? + (window as Window & { webkitAudioContext?: typeof AudioContext }) + .webkitAudioContext ?? + null + ); +} + +export function useVoiceInput(options: VoiceInputOptions) { const { t, i18n } = useTranslation(); const [recording, setRecording] = useState(false); const [transcribing, setTranscribing] = useState(false); const mediaRecorderRef = useRef(null); const chunksRef = useRef([]); + const captureRef = useRef(null); + // A press-and-hold release can land while `startRealtime` is still awaiting + // getUserMedia; remember it so the capture is not left running. + const pendingStopRef = useRef(false); + // Keep the latest callbacks without rebuilding every capture closure. + const optionsRef = useRef(options); + optionsRef.current = options; + const { active } = useVoiceConfig(); // Follow dashboard UI locale — not navigator.language (often en-US on // machines used with Chinese UI / Chinese speech). const language = speechLocaleFromUi(i18n.language); + const teardownRealtime = useCallback(() => { + const capture = captureRef.current; + captureRef.current = null; + if (!capture) return; + capture.processor.onaudioprocess = null; + capture.processor.disconnect(); + capture.source.disconnect(); + capture.sink.disconnect(); + capture.stream.getTracks().forEach((track) => track.stop()); + void capture.ctx.close(); + }, []); + + const finishRealtime = useCallback(() => { + teardownRealtime(); + setRecording(false); + setTranscribing(false); + }, [teardownRealtime]); + + const startRealtime = useCallback(async () => { + const stream = await navigator.mediaDevices.getUserMedia({ + audio: { + channelCount: 1, + echoCancellation: true, + noiseSuppression: true, + }, + }); + const Ctor = audioContextCtor(); + if (!Ctor) throw new Error("AudioContext unavailable"); + const ctx = new Ctor(); + const source = ctx.createMediaStreamSource(stream); + const processor = ctx.createScriptProcessor(1024, 1, 1); + const sink = ctx.createGain(); + sink.gain.value = 0; // keep the graph running without echoing the mic + + const client = new RealtimeSttClient(realtimeSttUrl(language), { + onInterim: (text) => optionsRef.current.onInterim?.(text), + onFinal: (text) => { + if (text) optionsRef.current.onText(text); + }, + onDone: finishRealtime, + onError: (message) => { + finishRealtime(); + antMessage.error(message || t("voice.realtimeFailed")); + }, + }); + + captureRef.current = { client, stream, ctx, source, processor, sink }; + const frames = new PcmFrameBuffer(); + processor.onaudioprocess = (event) => { + const pcm = downsampleToPcm16( + event.inputBuffer.getChannelData(0), + ctx.sampleRate, + ); + for (const frame of frames.push(pcm)) client.send(frame); + }; + source.connect(processor); + processor.connect(sink); + sink.connect(ctx.destination); + + // Frames are buffered until the server reports ready; failures surface + // through `onError`, so the connect rejection needs no extra handling. + void client.connect().catch(() => undefined); + setRecording(true); + }, [finishRealtime, language, t]); + + const stopRealtime = useCallback(() => { + const capture = captureRef.current; + if (!capture) return; + capture.processor.onaudioprocess = null; + capture.processor.disconnect(); + capture.source.disconnect(); + capture.sink.disconnect(); + capture.stream.getTracks().forEach((track) => track.stop()); + setRecording(false); + setTranscribing(true); + // `end` makes the server flush its final sentences, then `onDone` clears. + capture.client.end(); + }, []); + const stopRecording = useCallback(async () => { + pendingStopRef.current = true; + if (captureRef.current) { + stopRealtime(); + return; + } const recorder = mediaRecorderRef.current; if (!recorder || recorder.state === "inactive") { setRecording(false); @@ -280,7 +417,7 @@ export function useVoiceInput(onText: (text: string) => void) { const upload = await recordedBlobToWav(blob); const result = await voiceApi.transcribe(upload, language); const text = result.text?.trim() ?? ""; - if (text) onText(text); + if (text) optionsRef.current.onText(text); else antMessage.info(t("voice.sttEmpty")); } catch (err) { const msg = err instanceof Error ? err.message : ""; @@ -292,18 +429,32 @@ export function useVoiceInput(onText: (text: string) => void) { } finally { setTranscribing(false); } - }, [onText, t, language]); + }, [t, language, stopRealtime]); const startRecording = useCallback(async () => { // Read cached config synchronously to stay in the user-gesture stack. const active: ActiveVoice | null = cachedActiveVoice(); + pendingStopRef.current = false; + optionsRef.current.onStart?.(); + + if (active?.stt_realtime && canRecordAudio() && audioContextCtor()) { + // Realtime STT: stream PCM and let results arrive while speaking. + try { + await startRealtime(); + } catch { + antMessage.error(t("voice.micDenied")); + return; + } + if (pendingStopRef.current) stopRealtime(); + return; + } if ((active?.stt ?? "browser") === "browser" && browserSttAvailable()) { // Browser STT: use the native SpeechRecognition API directly. setTranscribing(true); try { const text = await transcribeWithBrowser(language); - if (text) onText(text); + if (text) optionsRef.current.onText(text); else antMessage.info(t("voice.sttEmpty")); } catch (err) { const msg = err instanceof Error ? err.message : ""; @@ -343,7 +494,7 @@ export function useVoiceInput(onText: (text: string) => void) { } catch { antMessage.error(t("voice.micDenied")); } - }, [onText, t, language]); + }, [t, language, startRealtime, stopRealtime]); const toggle = useCallback(() => { if (transcribing) return; @@ -351,5 +502,13 @@ export function useVoiceInput(onText: (text: string) => void) { else void startRecording(); }, [recording, transcribing, startRecording, stopRecording]); - return { recording, transcribing, toggle }; + return { + recording, + transcribing, + /** True when the mic streams results live (press and hold, not click). */ + realtime: active.stt_realtime === true, + start: startRecording, + stop: stopRecording, + toggle, + }; } diff --git a/dashboard/src/locales/en.json b/dashboard/src/locales/en.json index 6a95769ae..aceacc29f 100644 --- a/dashboard/src/locales/en.json +++ b/dashboard/src/locales/en.json @@ -132,6 +132,7 @@ "VOICE_KIND_UNSUPPORTED": "Unsupported voice provider.", "VOICE_CAPABILITY_MISMATCH": "Voice capability does not match.", "VOICE_PROVIDER_DISABLED": "Voice provider is disabled.", + "VOICE_REALTIME_UNAVAILABLE": "Realtime speech recognition is unavailable. Check the Tencent Cloud voice settings.", "INTERNAL_ERROR": "Internal server error.", "TLS_NOT_ELIGIBLE": "TLS issuance is not available in the current server configuration.", "TLS_ISSUE_IN_PROGRESS": "A certificate issuance task is already running.", @@ -2626,6 +2627,12 @@ "probeSuccess": "Voice model probe succeeded", "probeFailed": "Voice model probe failed", "tencentHint": "Create SecretId / SecretKey in Tencent Cloud console. ASR and TTS include free trial quota.", + "realtimeStt": "Realtime speech recognition", + "realtimeSttHint": "When enabled, voice input in the composer streams text as you speak (AppID required).", + "realtimeAppIdRequired": "AppID is required to enable realtime speech recognition.", + "realtimeFailed": "Realtime speech recognition failed", + "holdToTalk": "Hold to talk", + "releaseToStop": "Release to finish", "openaiHint": "Uses OpenAI Whisper for STT and OpenAI TTS for read-aloud.", "mimoHint": "Xiaomi MiMo ASR/TTS. Choose endpoint and enter the corresponding API key.", "mimoEndpoint": "API Endpoint", diff --git a/dashboard/src/locales/zh.json b/dashboard/src/locales/zh.json index 063c06161..b9cb0edcc 100644 --- a/dashboard/src/locales/zh.json +++ b/dashboard/src/locales/zh.json @@ -132,6 +132,7 @@ "VOICE_KIND_UNSUPPORTED": "不支持的语音提供商。", "VOICE_CAPABILITY_MISMATCH": "语音能力不匹配。", "VOICE_PROVIDER_DISABLED": "语音提供商已禁用。", + "VOICE_REALTIME_UNAVAILABLE": "实时语音识别不可用，请检查腾讯云语音配置。", "INTERNAL_ERROR": "服务器内部错误。", "TLS_NOT_ELIGIBLE": "当前服务器配置不支持 TLS 证书申请。", "TLS_ISSUE_IN_PROGRESS": "已有证书申请任务正在进行。", @@ -2626,6 +2627,12 @@ "probeSuccess": "语音模型探测成功", "probeFailed": "语音模型探测失败", "tencentHint": "在腾讯云控制台创建 SecretId / SecretKey，语音识别与合成均有免费试用额度。", + "realtimeStt": "实时语音识别", + "realtimeSttHint": "开启后，聊天输入框的语音输入将边说边出字（需要 AppID）。", + "realtimeAppIdRequired": "开启实时语音识别需要填写 AppID。", + "realtimeFailed": "实时语音识别失败", + "holdToTalk": "按住说话", + "releaseToStop": "松开结束", "openaiHint": "使用 OpenAI Whisper 识别、OpenAI TTS 朗读。", "mimoHint": "小米 Mimo 语音识别与合成。选择接入点并填写对应的 API Key。", "mimoEndpoint": "API 接入点", diff --git a/dashboard/src/pages/Chat/components/ChatInput.prefill.test.tsx b/dashboard/src/pages/Chat/components/ChatInput.prefill.test.tsx index a06d82df9..94bf7fd9f 100644 --- a/dashboard/src/pages/Chat/components/ChatInput.prefill.test.tsx +++ b/dashboard/src/pages/Chat/components/ChatInput.prefill.test.tsx @@ -17,6 +17,9 @@ vi.mock("../../../hooks/useVoiceInput", () => ({ useVoiceInput: () => ({ recording: false, transcribing: false, + realtime: false, + start: vi.fn(), + stop: vi.fn(), toggle: vi.fn(), }), })); diff --git a/dashboard/src/pages/Chat/components/ChatInput.tsx b/dashboard/src/pages/Chat/components/ChatInput.tsx index 79384fc46..da5eb5194 100644 --- a/dashboard/src/pages/Chat/components/ChatInput.tsx +++ b/dashboard/src/pages/Chat/components/ChatInput.tsx @@ -205,9 +205,19 @@ const ChatInput = forwardRef( setText(value); }, []); - const handleVoiceText = useCallback( + const textRef = useRef(text); + useEffect(() => { + textRef.current = text; + }, [text]); + // Text that existed before the current voice session started; the session's + // (possibly still refining) transcript is appended to it on every update. + const voiceBaselineRef = useRef(""); + + const applyVoiceText = useCallback( (spoken: string) => { - setText((prev) => (prev.trim() ? `${prev.trim()} ${spoken}` : spoken)); + const base = voiceBaselineRef.current.trim(); + const next = spoken.trim(); + setText(base ? (next ? `${base} ${next}` : base) : next); userHasEditedRef.current = true; onUserInput?.(); }, @@ -216,8 +226,17 @@ const ChatInput = forwardRef( const { recording, transcribing, + realtime: realtimeVoice, + start: startVoice, + stop: stopVoice, toggle: toggleVoice, - } = useVoiceInput(handleVoiceText); + } = useVoiceInput({ + onStart: () => { + voiceBaselineRef.current = textRef.current; + }, + onInterim: applyVoiceText, + onText: applyVoiceText, + }); const textareaRef = useRef(null); const { attachments, @@ -849,6 +868,9 @@ const ChatInput = forwardRef( uploading={uploading} recording={recording} transcribing={transcribing} + voiceHoldToTalk={realtimeVoice} + onVoiceStart={() => void startVoice()} + onVoiceStop={() => void stopVoice()} browserRecording={browserRecording} browserReplayBusy={browserReplayBusy} browserLastRecordingId={browserLastRecordingId} diff --git a/dashboard/src/pages/Chat/components/ChatInputActionsRow.test.tsx b/dashboard/src/pages/Chat/components/ChatInputActionsRow.test.tsx index a63436dd7..5470844f7 100644 --- a/dashboard/src/pages/Chat/components/ChatInputActionsRow.test.tsx +++ b/dashboard/src/pages/Chat/components/ChatInputActionsRow.test.tsx @@ -4,6 +4,17 @@ import { describe, expect, it, vi } from "vitest"; import type { ResolvedModel } from "../../../api/types"; import ChatInputActionsRow from "./ChatInputActionsRow"; +// `isSttAvailable()` runs at module load, so the browser APIs it probes must +// exist before this module graph is imported. +vi.hoisted(() => { + vi.stubGlobal("MediaRecorder", class {}); + Object.defineProperty(navigator, "mediaDevices", { + value: { getUserMedia: () => Promise.resolve() }, + configurable: true, + }); + return true; +}); + vi.mock("./ContextWindowRing", () => ({ default: () => null, })); @@ -66,3 +77,84 @@ describe("ChatInputActionsRow compact pickers", () => { expect(document.querySelector(".ant-drawer-content")).toBeNull(); }); }); + +function renderRow(overrides: Record = {}) { + const handlers = { + onVoiceStart: vi.fn(), + onVoiceStop: vi.fn(), + onToggleVoice: vi.fn(), + }; + const view = render( + + + , + ); + const mic = view.container + .querySelector("svg.lucide-mic") + ?.closest("button") as HTMLButtonElement; + return { ...view, ...handlers, mic }; +} + +describe("ChatInputActionsRow voice button", () => { + it("dictates while held down when realtime STT is on", () => { + const { mic, onVoiceStart, onVoiceStop, onToggleVoice } = renderRow({ + voiceHoldToTalk: true, + }); + + fireEvent.pointerDown(mic); + expect(onVoiceStart).toHaveBeenCalledTimes(1); + + fireEvent.pointerUp(mic); + expect(onVoiceStop).toHaveBeenCalledTimes(1); + + // A plain click must not toggle the recorder in hold-to-talk mode. + fireEvent.click(mic); + expect(onToggleVoice).not.toHaveBeenCalled(); + expect(onVoiceStart).toHaveBeenCalledTimes(1); + }); + + it("finishes when the pointer slides off the button", () => { + const { mic, onVoiceStop } = renderRow({ voiceHoldToTalk: true }); + + fireEvent.pointerDown(mic); + fireEvent.pointerLeave(mic); + fireEvent.pointerUp(mic); + + expect(onVoiceStop).toHaveBeenCalledTimes(1); + }); + + it("keeps click-to-toggle when realtime STT is off", () => { + const { mic, onToggleVoice, onVoiceStart } = renderRow(); + + fireEvent.click(mic); + expect(onToggleVoice).toHaveBeenCalledTimes(1); + + fireEvent.pointerDown(mic); + expect(onVoiceStart).not.toHaveBeenCalled(); + }); + + it("stays available while a reply is still streaming", () => { + const { mic } = renderRow({ voiceHoldToTalk: true, isStreaming: true }); + + expect(mic).not.toBeDisabled(); + }); +}); diff --git a/dashboard/src/pages/Chat/components/ChatInputActionsRow.tsx b/dashboard/src/pages/Chat/components/ChatInputActionsRow.tsx index bcfa05e1c..755453e32 100644 --- a/dashboard/src/pages/Chat/components/ChatInputActionsRow.tsx +++ b/dashboard/src/pages/Chat/components/ChatInputActionsRow.tsx @@ -96,6 +96,10 @@ interface ChatInputActionsRowProps { uploading: boolean; recording: boolean; transcribing: boolean; + /** Realtime STT streams while held down, so the mic is press-and-hold. */ + voiceHoldToTalk?: boolean; + onVoiceStart?: () => void; + onVoiceStop?: () => void; browserRecording?: boolean; browserReplayBusy?: boolean; browserLastRecordingId?: string | null; @@ -157,6 +161,9 @@ export default function ChatInputActionsRow({ uploading, recording, transcribing, + voiceHoldToTalk = false, + onVoiceStart, + onVoiceStop, browserRecording = false, browserReplayBusy = false, browserLastRecordingId = null, @@ -204,6 +211,7 @@ export default function ChatInputActionsRow({ const navigate = useNavigate(); const skillDisplayName = useSkillDisplayName(); const actionsRowRef = useRef(null); + const voiceHeldRef = useRef(false); const [isCompact, setIsCompact] = useState(false); const [skillPickerOpen, setSkillPickerOpen] = useState(false); const [expertPickerOpen, setExpertPickerOpen] = useState(false); @@ -217,6 +225,20 @@ export default function ChatInputActionsRow({ ); /** Mobile-only bottom drawer for the overflow ("more") menu. */ const [mobileOverflowOpen, setMobileOverflowOpen] = useState(false); + + // Press-and-hold dictation: start on pointerdown, finish on release or when + // the pointer slides off the button (also covers touch cancel). + const handleVoicePressStart = () => { + if (voiceHeldRef.current) return; + voiceHeldRef.current = true; + onVoiceStart?.(); + }; + + const handleVoicePressEnd = () => { + if (!voiceHeldRef.current) return; + voiceHeldRef.current = false; + onVoiceStop?.(); + }; /** Narrow-desktop overflow popover (tools / skills / …). */ const [overflowPopoverOpen, setOverflowPopoverOpen] = useState(false); /** Active sub-picker for compact layouts (drawer on mobile, panel in popover). */ @@ -1233,6 +1255,10 @@ export default function ChatInputActionsRow({ title={ !_sttAvailable ? t("voice.sttNotAvailable", "此设备不支持语音输入（需要 HTTPS）") + : voiceHoldToTalk + ? recording + ? t("voice.releaseToStop", "松开结束") + : t("voice.holdToTalk", "按住说话") : recording ? t("voice.stopRecording", "停止录音") : transcribing @@ -1246,8 +1272,29 @@ export default function ChatInputActionsRow({ recording || transcribing ? styles.secondaryBtnActive : "" }`} type="button" - disabled={disabled || isStreaming || transcribing || !_sttAvailable} - onClick={onToggleVoice} + // Dictation only fills the composer, so it stays available while a + // reply is still streaming. + disabled={disabled || transcribing || !_sttAvailable} + onClick={voiceHoldToTalk ? undefined : onToggleVoice} + onPointerDown={ + voiceHoldToTalk + ? (event) => { + event.preventDefault(); + handleVoicePressStart(); + } + : undefined + } + onPointerUp={voiceHoldToTalk ? handleVoicePressEnd : undefined} + onPointerLeave={voiceHoldToTalk ? handleVoicePressEnd : undefined} + onPointerCancel={voiceHoldToTalk ? handleVoicePressEnd : undefined} + onContextMenu={ + voiceHoldToTalk ? (event) => event.preventDefault() : undefined + } + style={ + voiceHoldToTalk + ? { touchAction: "none", userSelect: "none" } + : undefined + } > diff --git a/dashboard/src/pages/Settings/Voice/index.tsx b/dashboard/src/pages/Settings/Voice/index.tsx index 470a7842f..354bee4e3 100644 --- a/dashboard/src/pages/Settings/Voice/index.tsx +++ b/dashboard/src/pages/Settings/Voice/index.tsx @@ -1,5 +1,14 @@ import { useCallback, useEffect, useMemo, useState } from "react"; -import { Button, Divider, Drawer, Form, Input, Select, Typography } from "antd"; +import { + Button, + Divider, + Drawer, + Form, + Input, + Select, + Switch, + Typography, +} from "antd"; import { message } from "@/utils/antdMessage"; import { Activity, Mic2, Check, Settings2 } from "lucide-react"; @@ -37,6 +46,8 @@ export function VoiceSettingsPanel() { const [apiKey, setApiKey] = useState(""); const [secretId, setSecretId] = useState(""); const [secretKey, setSecretKey] = useState(""); + const [realtimeStt, setRealtimeStt] = useState(false); + const [appId, setAppId] = useState(""); const [mimoEndpoint, setMimoEndpoint] = useState<"payg" | "tokenplan">( "payg", ); @@ -101,6 +112,8 @@ export function VoiceSettingsPanel() { const extra = existing?.extra ?? {}; setSecretId(String(extra.secret_id ?? "")); setSecretKey(String(extra.secret_key ?? "")); + setRealtimeStt(extra.realtime_stt === true); + setAppId(String(extra.app_id ?? "")); setMimoEndpoint(extra.endpoint_type === "tokenplan" ? "tokenplan" : "payg"); setMimoVoiceId(String(extra.voice_id ?? "冰糖")); }; @@ -115,6 +128,8 @@ export function VoiceSettingsPanel() { secret_id: secretId, secret_key: secretKey, region: "ap-guangzhou", + realtime_stt: realtimeStt, + app_id: appId.trim() || undefined, }; } else if (preset.kind === "edge") { extra = { voice_id: "zh-CN-XiaoxiaoNeural" }; @@ -152,6 +167,10 @@ export function VoiceSettingsPanel() { ? Boolean(secretId.trim() && secretKey.trim()) : Boolean(apiKey.trim()); if (!complete) message.warning(t("voice.credentialsRequired")); + if (complete && realtimeStt && !appId.trim()) { + message.warning(t("voice.realtimeAppIdRequired")); + return false; + } return complete; }; @@ -192,6 +211,9 @@ export function VoiceSettingsPanel() { } else { await voiceApi.createProvider(payload); } + // The active payload carries `stt_realtime`, so drop the cache even when + // the active provider itself does not change (e.g. the switch was toggled). + invalidateVoiceConfigCache(); if ( preset.kind !== "browser" && (preset.capability === "stt" || preset.capability === "both") && @@ -199,7 +221,6 @@ export function VoiceSettingsPanel() { ) { const next = await voiceApi.setActive({ stt: preset.id }); setActive(next); - invalidateVoiceConfigCache(); } if ( preset.kind !== "browser" && @@ -208,7 +229,6 @@ export function VoiceSettingsPanel() { ) { const next = await voiceApi.setActive({ tts: preset.id }); setActive(next); - invalidateVoiceConfigCache(); } message.success(t("voice.saved")); setConfigure(null); @@ -419,6 +439,26 @@ export function VoiceSettingsPanel() { onChange={(e) => setSecretKey(e.target.value)} /> + + + + + setAppId(e.target.value)} + /> + + {realtimeStt && ( +
+ {t("voice.realtimeSttHint")} +
+ )} )} {configure?.preset.kind === "openai" && ( diff --git a/dashboard/src/utils/pcmCapture.test.ts b/dashboard/src/utils/pcmCapture.test.ts new file mode 100644 index 000000000..e811459e3 --- /dev/null +++ b/dashboard/src/utils/pcmCapture.test.ts @@ -0,0 +1,71 @@ +import { describe, expect, it } from "vitest"; +import { + FRAME_BYTES, + PcmFrameBuffer, + downsampleToPcm16, + floatToPcm16, +} from "./pcmCapture"; + +describe("floatToPcm16", () => { + it("clamps to the signed 16-bit range", () => { + const out = floatToPcm16(new Float32Array([0, 1, -1, 2, -2])); + expect([...out]).toEqual([0, 32767, -32768, 32767, -32768]); + }); +}); + +describe("downsampleToPcm16", () => { + it("keeps 16 kHz input as is", () => { + const out = downsampleToPcm16(new Float32Array([0, 0.5, -0.5]), 16000); + expect([...out]).toEqual([0, 16383, -16384]); + }); + + it("interpolates 48 kHz input down to a third of the samples", () => { + const input = new Float32Array(480).fill(0.5); + const out = downsampleToPcm16(input, 48000); + expect(out.length).toBe(160); + expect([...out].every((sample) => sample === 16383)).toBe(true); + }); +}); + +describe("PcmFrameBuffer", () => { + it("emits whole frames of 1280 bytes", () => { + const buffer = new PcmFrameBuffer(); + const frames = buffer.push(new Int16Array(640).fill(1)); + + expect(frames).toHaveLength(1); + expect(frames[0].byteLength).toBe(FRAME_BYTES); + expect([...frames[0].slice(0, 2)]).toEqual([1, 0]); + }); + + it("carries the remainder into the next frame", () => { + const buffer = new PcmFrameBuffer(); + + expect(buffer.push(new Int16Array(320).fill(1))).toHaveLength(0); + const frames = buffer.push(new Int16Array(320).fill(2)); + + expect(frames).toHaveLength(1); + expect([...new Int16Array(frames[0].buffer)].slice(0, 320)).toEqual( + new Array(320).fill(1), + ); + expect([...new Int16Array(frames[0].buffer)].slice(320)).toEqual( + new Array(320).fill(2), + ); + }); + + it("emits several frames for a large chunk", () => { + const buffer = new PcmFrameBuffer(); + expect(buffer.push(new Int16Array(1600))).toHaveLength(2); + }); + + it("drops the carry on reset", () => { + const buffer = new PcmFrameBuffer(); + buffer.push(new Int16Array(320).fill(1)); + buffer.reset(); + + expect(buffer.push(new Int16Array(320).fill(2))).toHaveLength(0); + const frames = buffer.push(new Int16Array(320).fill(2)); + expect([...new Int16Array(frames[0].buffer)].slice(0, 320)).toEqual( + new Array(320).fill(2), + ); + }); +}); diff --git a/dashboard/src/utils/pcmCapture.ts b/dashboard/src/utils/pcmCapture.ts new file mode 100644 index 000000000..da42b5395 --- /dev/null +++ b/dashboard/src/utils/pcmCapture.ts @@ -0,0 +1,71 @@ +/** + * 16 kHz mono PCM helpers for realtime speech recognition. + * + * Tencent's realtime ASR expects 16-bit mono PCM at 16 kHz, sent in ~40 ms + * frames (640 samples / 1280 bytes) at roughly 1:1 realtime. + */ + +export const REALTIME_SAMPLE_RATE = 16000; +export const FRAME_SAMPLES = 640; +export const FRAME_BYTES = FRAME_SAMPLES * 2; + +function toPcm16(sample: number): number { + const clamped = Math.max(-1, Math.min(1, sample)); + return clamped < 0 ? clamped * 0x8000 : clamped * 0x7fff; +} + +export function floatToPcm16(input: Float32Array): Int16Array { + const out = new Int16Array(input.length); + for (let i = 0; i < input.length; i += 1) { + out[i] = toPcm16(input[i]); + } + return out; +} + +/** Linearly interpolated resample to 16 kHz PCM16; adequate for speech. */ +export function downsampleToPcm16( + input: Float32Array, + inputRate: number, +): Int16Array { + if (inputRate <= REALTIME_SAMPLE_RATE) return floatToPcm16(input); + const ratio = inputRate / REALTIME_SAMPLE_RATE; + const outLength = Math.floor(input.length / ratio); + const out = new Int16Array(outLength); + for (let i = 0; i < outLength; i += 1) { + const position = i * ratio; + const left = Math.floor(position); + const right = Math.min(left + 1, input.length - 1); + const weight = position - left; + out[i] = toPcm16(input[left] * (1 - weight) + input[right] * weight); + } + return out; +} + +/** Accumulates arbitrary PCM chunks into whole {@link FRAME_BYTES} frames. */ +export class PcmFrameBuffer { + private carry = new Uint8Array(0); + + push(samples: Int16Array): Uint8Array[] { + const bytes = new Uint8Array( + samples.buffer, + samples.byteOffset, + samples.byteLength, + ); + const merged = new Uint8Array(this.carry.length + bytes.length); + merged.set(this.carry, 0); + merged.set(bytes, this.carry.length); + + const frames: Uint8Array[] = []; + let offset = 0; + while (merged.length - offset >= FRAME_BYTES) { + frames.push(merged.slice(offset, offset + FRAME_BYTES)); + offset += FRAME_BYTES; + } + this.carry = merged.slice(offset); + return frames; + } + + reset(): void { + this.carry = new Uint8Array(0); + } +} diff --git a/dashboard/src/utils/realtimeStt.test.ts b/dashboard/src/utils/realtimeStt.test.ts new file mode 100644 index 000000000..4e00a3e08 --- /dev/null +++ b/dashboard/src/utils/realtimeStt.test.ts @@ -0,0 +1,187 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { RealtimeSttClient } from "./realtimeStt"; + +class FakeSocket { + static instances: FakeSocket[] = []; + + binaryType = ""; + onopen: (() => void) | null = null; + onmessage: ((event: { data: unknown }) => void) | null = null; + onerror: (() => void) | null = null; + onclose: (() => void) | null = null; + sent: unknown[] = []; + closed = false; + readonly url: string; + + constructor(url: string) { + this.url = url; + FakeSocket.instances.push(this); + } + + send(data: unknown): void { + this.sent.push(data); + } + + close(): void { + this.closed = true; + this.onclose?.(); + } + + emitMessage(payload: unknown): void { + this.onmessage?.({ data: JSON.stringify(payload) }); + } + + emitClose(): void { + this.onclose?.(); + } + + emitError(): void { + this.onerror?.(); + } +} + +function setup() { + FakeSocket.instances = []; + vi.stubGlobal("WebSocket", FakeSocket); + const handlers = { + onInterim: vi.fn(), + onFinal: vi.fn(), + onDone: vi.fn(), + onError: vi.fn(), + }; + const client = new RealtimeSttClient( + "ws://test/api/voice/stt-stream", + handlers, + ); + return { client, handlers }; +} + +function lastSocket(): FakeSocket { + return FakeSocket.instances[FakeSocket.instances.length - 1]; +} + +describe("RealtimeSttClient", () => { + afterEach(() => { + vi.unstubAllGlobals(); + }); + + it("buffers audio until the server is ready", async () => { + const { client } = setup(); + const connected = client.connect(); + const socket = lastSocket(); + + client.send(new Uint8Array([1, 2])); + expect(socket.sent).toHaveLength(0); + + socket.emitMessage({ type: "ready" }); + await connected; + + expect(socket.sent).toEqual([new Uint8Array([1, 2])]); + }); + + it("reports the cumulative transcript", async () => { + const { client, handlers } = setup(); + const connected = client.connect(); + const socket = lastSocket(); + socket.emitMessage({ type: "ready" }); + await connected; + + socket.emitMessage({ type: "interim", text: "你好", sentence_id: 0 }); + expect(handlers.onInterim).toHaveBeenLastCalledWith("你好"); + + socket.emitMessage({ type: "final", text: "你好世界", sentence_id: 0 }); + expect(handlers.onFinal).toHaveBeenLastCalledWith("你好世界"); + + socket.emitMessage({ type: "interim", text: "再见", sentence_id: 1 }); + expect(handlers.onInterim).toHaveBeenLastCalledWith("你好世界再见"); + }); + + it("does not duplicate a sentence confirmed twice", async () => { + const { client, handlers } = setup(); + const connected = client.connect(); + const socket = lastSocket(); + socket.emitMessage({ type: "ready" }); + await connected; + + socket.emitMessage({ type: "final", text: "你好", sentence_id: 0 }); + socket.emitMessage({ type: "final", text: "你好", sentence_id: 0 }); + + expect(handlers.onFinal).toHaveBeenLastCalledWith("你好"); + }); + + it("commits a trailing interim sentence when the stream ends", async () => { + const { client, handlers } = setup(); + const connected = client.connect(); + const socket = lastSocket(); + socket.emitMessage({ type: "ready" }); + await connected; + + socket.emitMessage({ type: "interim", text: "未确认" }); + socket.emitMessage({ type: "done" }); + + expect(handlers.onFinal).toHaveBeenLastCalledWith("未确认"); + expect(handlers.onDone).toHaveBeenCalledTimes(1); + expect(handlers.onError).not.toHaveBeenCalled(); + expect(socket.closed).toBe(true); + }); + + it("sends the end frame through the socket", async () => { + const { client } = setup(); + const connected = client.connect(); + const socket = lastSocket(); + socket.emitMessage({ type: "ready" }); + await connected; + + client.end(); + + expect(JSON.parse(String(socket.sent[socket.sent.length - 1]))).toEqual({ + type: "end", + }); + }); + + it("surfaces the server error message", async () => { + const { client, handlers } = setup(); + const connected = client.connect(); + const socket = lastSocket(); + socket.emitMessage({ type: "ready" }); + await connected; + + socket.emitMessage({ type: "error", message: "资源包额度已耗尽。" }); + + expect(handlers.onError).toHaveBeenCalledWith("资源包额度已耗尽。"); + expect(handlers.onDone).not.toHaveBeenCalled(); + }); + + it("reports an unexpected close as a transport failure", async () => { + const { client, handlers } = setup(); + const connected = client.connect(); + const socket = lastSocket(); + socket.emitMessage({ type: "ready" }); + await connected; + + socket.emitClose(); + + expect(handlers.onError).toHaveBeenCalledWith(null); + }); + + it("rejects connect when the socket fails before ready", async () => { + const { client } = setup(); + const connected = client.connect(); + lastSocket().emitError(); + + await expect(connected).rejects.toThrow(); + }); + + it("stops sending once closed", async () => { + const { client } = setup(); + const connected = client.connect(); + const socket = lastSocket(); + socket.emitMessage({ type: "ready" }); + await connected; + + client.close(); + client.send(new Uint8Array([1])); + + expect(socket.sent).toHaveLength(0); + }); +}); diff --git a/dashboard/src/utils/realtimeStt.ts b/dashboard/src/utils/realtimeStt.ts new file mode 100644 index 000000000..592eecec5 --- /dev/null +++ b/dashboard/src/utils/realtimeStt.ts @@ -0,0 +1,175 @@ +/** + * Browser client for Octop's realtime STT WebSocket (`/api/voice/stt-stream`). + * + * Tencent refines a sentence (`interim`) before confirming it (`final`), so the + * client tracks confirmed sentences by id and reports the *cumulative* + * transcript of the session. The composer can therefore replace only the tail + * of the input it owns instead of appending duplicates. + */ + +export interface RealtimeSttHandlers { + /** Cumulative transcript while the current sentence is still being refined. */ + onInterim?: (text: string) => void; + /** Cumulative transcript once a sentence is confirmed. */ + onFinal?: (text: string) => void; + /** The audio stream finished server-side. */ + onDone?: () => void; + /** ``message`` comes from the server; ``null`` means a transport failure. */ + onError?: (message: string | null) => void; +} + +interface ServerFrame { + type?: string; + text?: string; + sentence_id?: number; + message?: string; +} + +/** Sentence id used when committing a trailing interim sentence at ``done``. */ +const TRAILING_SENTENCE_ID = Number.MAX_SAFE_INTEGER; + +export class RealtimeSttClient { + private socket: WebSocket | null = null; + private readonly finals = new Map(); + private interim = ""; + private pending: Uint8Array[] = []; + private ready = false; + private finished = false; + private endRequested = false; + private closed = false; + private readonly url: string; + private readonly handlers: RealtimeSttHandlers; + + constructor(url: string, handlers: RealtimeSttHandlers = {}) { + this.url = url; + this.handlers = handlers; + } + + /** Open the socket; resolves once the server reports ``ready``. */ + connect(): Promise { + return new Promise((resolve, reject) => { + const socket = new WebSocket(this.url); + socket.binaryType = "arraybuffer"; + this.socket = socket; + + const failConnect = (error: unknown) => reject(error); + + socket.onopen = () => { + // Nothing to send yet: PCM frames are buffered until `ready`. + }; + socket.onmessage = (event: MessageEvent) => { + const frame = parseFrame(event.data); + if (!frame) return; + if (frame.type === "ready") { + this.ready = true; + this.flushPending(); + resolve(); + return; + } + this.handleFrame(frame); + }; + socket.onerror = () => { + if (!this.ready) failConnect(new Error("realtime STT socket error")); + this.fail(null); + }; + socket.onclose = () => { + if (!this.ready) { + failConnect(new Error("realtime STT socket closed before ready")); + } + if (!this.finished && !this.closed) this.fail(null); + }; + }); + } + + /** Queue or send one PCM frame (16 kHz / 16-bit / mono). */ + send(frame: Uint8Array): void { + if (this.closed) return; + if (!this.ready) { + this.pending.push(frame); + return; + } + this.socket?.send(frame); + } + + /** Tell the server the audio stream ended and wait for its final results. */ + end(): void { + if (this.closed) return; + this.endRequested = true; + if (this.ready) this.socket?.send(JSON.stringify({ type: "end" })); + } + + /** Tear down without reporting an error (user cancelled). */ + close(): void { + this.closed = true; + this.pending = []; + this.socket?.close(); + this.socket = null; + } + + private handleFrame(frame: ServerFrame): void { + if (frame.type === "interim") { + this.interim = frame.text ?? ""; + this.handlers.onInterim?.(this.transcript()); + return; + } + if (frame.type === "final") { + this.finals.set(frame.sentence_id ?? 0, frame.text ?? ""); + this.interim = ""; + this.handlers.onFinal?.(this.transcript()); + return; + } + if (frame.type === "done") { + this.commitTrailingInterim(); + this.finished = true; + this.handlers.onDone?.(); + this.close(); + return; + } + if (frame.type === "error") { + this.fail(frame.message ?? null); + } + } + + private commitTrailingInterim(): void { + if (!this.interim) return; + this.finals.set(TRAILING_SENTENCE_ID, this.interim); + this.interim = ""; + this.handlers.onFinal?.(this.transcript()); + } + + private transcript(): string { + const confirmed = [...this.finals.entries()] + .sort(([a], [b]) => a - b) + .map(([, text]) => text) + .filter(Boolean); + if (this.interim) confirmed.push(this.interim); + return confirmed.join(""); + } + + private flushPending(): void { + if (!this.ready) return; + const queued = this.pending; + this.pending = []; + for (const frame of queued) this.socket?.send(frame); + if (this.endRequested) this.end(); + } + + private fail(message: string | null): void { + if (this.finished || this.closed) return; + this.finished = true; + this.handlers.onError?.(message); + this.close(); + } +} + +function parseFrame(data: unknown): ServerFrame | null { + if (typeof data !== "string") return null; + try { + const parsed: unknown = JSON.parse(data); + return parsed && typeof parsed === "object" + ? (parsed as ServerFrame) + : null; + } catch { + return null; + } +} diff --git a/src/octop/api/routers/voice.py b/src/octop/api/routers/voice.py index a6a811caf..d0095914a 100644 --- a/src/octop/api/routers/voice.py +++ b/src/octop/api/routers/voice.py @@ -2,16 +2,37 @@ from __future__ import annotations +import asyncio +import contextlib +import json from collections.abc import AsyncIterator from typing import Any, Literal -from fastapi import APIRouter, Depends, File, Form, Request, UploadFile +from fastapi import ( + APIRouter, + Depends, + File, + Form, + Query, + Request, + UploadFile, + WebSocket, + WebSocketDisconnect, +) from fastapi.responses import StreamingResponse from pydantic import BaseModel, Field +from starlette.websockets import WebSocketState -from octop.api.deps import current_user, get_server, require_permission +from octop.api.deps import ( + current_user, + get_server, + require_permission, + resolve_user_from_token, +) +from octop.i18n.domains.voice import realtime_error_message from octop.infra.errors import ErrorCode, OctopError -from octop.infra.utils.locale import resolve_request_locale +from octop.infra.utils.locale import Locale, normalize_locale, resolve_request_locale +from octop.infra.voice import realtime from octop.infra.voice.manager import VoiceManager from octop.infra.voice.presets import load_voice_presets @@ -26,6 +47,17 @@ def _voice_manager(server: Any) -> VoiceManager: ) +def _active_payload(mgr: VoiceManager) -> dict[str, Any]: + """Active provider names plus whether STT can stream. + + The dashboard needs ``stt_realtime`` synchronously inside a user gesture to + decide between streaming and the one-shot upload path. + """ + active: dict[str, Any] = dict(mgr.get_active()) + active["stt_realtime"] = mgr.realtime_stt_config() is not None + return active + + def _row_to_dict(r: Any) -> dict[str, Any]: return { "id": r.id, @@ -97,8 +129,8 @@ async def list_voice_providers( async def get_active_voice( _: Any = Depends(current_user), server: Any = Depends(get_server), -) -> dict[str, str]: - return _voice_manager(server).get_active() +) -> dict[str, Any]: + return _active_payload(_voice_manager(server)) @router.put("/active") @@ -106,8 +138,10 @@ async def set_active_voice( body: ActiveVoiceBody, _: Any = Depends(require_permission("voice")), server: Any = Depends(get_server), -) -> dict[str, str]: - return _voice_manager(server).set_active(stt=body.stt, tts=body.tts) +) -> dict[str, Any]: + mgr = _voice_manager(server) + mgr.set_active(stt=body.stt, tts=body.tts) + return _active_payload(mgr) @router.post("/stt") @@ -152,6 +186,152 @@ async def synthesize_speech( return StreamingResponse(_stream(), media_type=mgr.media_type(body.provider)) +async def _ws_send_json(websocket: WebSocket, payload: dict[str, Any]) -> None: + if websocket.application_state != WebSocketState.CONNECTED: + return + with contextlib.suppress(WebSocketDisconnect, RuntimeError): + await websocket.send_text(json.dumps(payload, ensure_ascii=False)) + + +async def _ws_error( + websocket: WebSocket, + code: ErrorCode, + locale: Locale, + *, + tencent_code: int | None = None, +) -> None: + message = ( + realtime_error_message(tencent_code, locale) + if tencent_code is not None + else OctopError(code, "").localized_message(locale) + ) + await _ws_send_json( + websocket, + { + "type": "error", + "code": code.value, + "message": message, + "tencent_code": tencent_code, + }, + ) + + +def _is_end_frame(raw: str) -> bool: + try: + control = json.loads(raw) + except json.JSONDecodeError: + return False + return isinstance(control, dict) and control.get("type") == "end" + + +async def _pump_events( + session: realtime.TencentRealtimeSession, websocket: WebSocket, locale: Locale +) -> None: + """Forward upstream recognition events to the browser until the stream ends.""" + async for event in session.events(): + if event.kind == "error": + await _ws_error( + websocket, + ErrorCode.VOICE_REALTIME_UNAVAILABLE, + locale, + tencent_code=event.error_code, + ) + await _ws_send_json(websocket, {"type": "done"}) + return + + if event.kind == "done": + await _ws_send_json(websocket, {"type": "done"}) + return + + await _ws_send_json( + websocket, + { + "type": event.kind, + "text": event.text, + "sentence_id": event.sentence_id, + }, + ) + + +@router.websocket("/stt-stream") +async def stream_transcription( + websocket: WebSocket, + token: str | None = Query(default=None), + locale: str | None = Query(default=None), +) -> None: + """Proxy one realtime ASR session to Tencent Cloud over WebSocket. + + The browser sends 16 kHz PCM16 mono frames as binary messages and + ``{"type": "end"}`` when it is done; the server replies with ``ready`` / + ``interim`` / ``final`` / ``done`` / ``error`` JSON messages. + """ + server = websocket.app.state.octop_server + if not token: + await websocket.close(code=4001, reason="missing token") + return + try: + resolve_user_from_token(server, token) + except OctopError as exc: + await websocket.close(code=4001, reason=f"auth failed: {exc.code.value}") + return + + await websocket.accept() + lang = normalize_locale(locale) if locale else resolve_request_locale(websocket) + + config = _voice_manager(server).realtime_stt_config() + if config is None: + await _ws_error(websocket, ErrorCode.VOICE_REALTIME_UNAVAILABLE, lang) + await websocket.close(code=4004, reason="realtime STT is not configured") + return + + session = realtime.open_session(config) + reader: asyncio.Task[None] | None = None + try: + try: + await session.connect() + except OctopError as exc: + tencent_code = exc.details.get("tencent_code") + await _ws_error( + websocket, + exc.code, + lang, + tencent_code=tencent_code if isinstance(tencent_code, int) else None, + ) + await websocket.close(code=4004, reason="upstream connect failed") + return + + await _ws_send_json(websocket, {"type": "ready"}) + reader = asyncio.create_task(_pump_events(session, websocket, lang)) + + while websocket.application_state == WebSocketState.CONNECTED: + message = await websocket.receive() + if message.get("type") == "websocket.disconnect": + break + chunk = message.get("bytes") + if chunk is not None: + await session.send_audio(chunk) + continue + raw = message.get("text") + if raw and _is_end_frame(raw): + await session.finish() + break + + if reader is not None: + with contextlib.suppress(TimeoutError): + await asyncio.wait_for(reader, timeout=realtime.READER_TIMEOUT_SECONDS) + except WebSocketDisconnect: + pass + finally: + if reader is not None and not reader.done(): + reader.cancel() + with contextlib.suppress(asyncio.CancelledError): + await reader + await session.close() + if websocket.application_state == WebSocketState.CONNECTED: + with contextlib.suppress(RuntimeError, WebSocketDisconnect): + await websocket.close() + + @admin_router.get("") async def admin_list_voice_providers( _: Any = Depends(require_permission("voice")), diff --git a/src/octop/i18n/domains/voice.py b/src/octop/i18n/domains/voice.py index f2fc84b87..9a7ae1176 100644 --- a/src/octop/i18n/domains/voice.py +++ b/src/octop/i18n/domains/voice.py @@ -1,4 +1,4 @@ -"""``voice.*`` — probe and Tencent provider error copy.""" +"""``voice.*`` — probe, realtime and Tencent provider error copy.""" from __future__ import annotations @@ -36,3 +36,26 @@ def format_voice_probe_error(exc: BaseException, locale: str | Locale) -> str: if sep and lookup(f"voice.tencent.{code}", locale) is not None: return tr(f"voice.tencent.{code}", locale) return raw + + +#: Realtime ASR error codes worth their own copy; anything else falls back to +#: ``voice.realtime.error_unknown``. +_ASR_ERROR_KEYS = { + 4002: "error_4002", + 4003: "error_4003", + 4004: "error_4004", + 4005: "error_4005", + 4006: "error_4006", + 4007: "error_4007", + 4008: "error_4008", + 4009: "error_4009", + 5000: "error_5000", + 5001: "error_5000", + 5002: "error_5000", +} + + +def realtime_error_message(tencent_code: int | None, locale: str | Locale) -> str: + """Localized copy for a realtime ASR failure code (``None`` = unknown).""" + suffix = _ASR_ERROR_KEYS.get(int(tencent_code or 0), "error_unknown") + return tr(f"voice.realtime.{suffix}", locale) diff --git a/src/octop/i18n/en.json b/src/octop/i18n/en.json index b856f3247..bed35efda 100644 --- a/src/octop/i18n/en.json +++ b/src/octop/i18n/en.json @@ -393,6 +393,7 @@ "VOICE_KIND_UNSUPPORTED": "Unsupported voice provider.", "VOICE_CAPABILITY_MISMATCH": "Voice capability does not match.", "VOICE_PROVIDER_DISABLED": "Voice provider is disabled.", + "VOICE_REALTIME_UNAVAILABLE": "Realtime speech recognition is unavailable. Check the Tencent Cloud voice settings.", "INTERNAL_ERROR": "Internal server error.", "TLS_NOT_ELIGIBLE": "TLS issuance is not available in the current server configuration.", "TLS_ISSUE_IN_PROGRESS": "A certificate issuance task is already running.", @@ -743,6 +744,18 @@ "http_status": "provider returned HTTP {status}", "network_error": "network error: {name}" }, + "realtime": { + "error_4002": "Realtime speech recognition authentication failed. Check the SecretId and SecretKey.", + "error_4003": "Realtime speech recognition is not enabled for this Tencent Cloud account.", + "error_4004": "The Tencent Cloud realtime speech recognition resource pack is exhausted.", + "error_4005": "The Tencent Cloud account is in arrears. Top up and retry.", + "error_4006": "Realtime speech recognition concurrency limit reached. Retry in a moment.", + "error_4007": "Audio decoding failed. Check the microphone input.", + "error_4008": "Audio upload timed out. Check the network connection.", + "error_4009": "The realtime speech recognition connection was closed.", + "error_5000": "The Tencent Cloud service is temporarily unavailable. Retry shortly.", + "error_unknown": "Realtime speech recognition failed. Retry in a moment." + }, "tencent": { "AuthFailure": { "SecretIdNotFound": "SecretId was not found. Check that the key is correct.", diff --git a/src/octop/i18n/zh.json b/src/octop/i18n/zh.json index bde66143a..87c8ec698 100644 --- a/src/octop/i18n/zh.json +++ b/src/octop/i18n/zh.json @@ -393,6 +393,7 @@ "VOICE_KIND_UNSUPPORTED": "不支持的语音提供商。", "VOICE_CAPABILITY_MISMATCH": "语音能力不匹配。", "VOICE_PROVIDER_DISABLED": "语音提供商已禁用。", + "VOICE_REALTIME_UNAVAILABLE": "实时语音识别不可用，请检查腾讯云语音配置。", "INTERNAL_ERROR": "服务器内部错误。", "TLS_NOT_ELIGIBLE": "当前服务器配置不支持 TLS 证书申请。", "TLS_ISSUE_IN_PROGRESS": "已有证书申请任务正在进行。", @@ -743,6 +744,18 @@ "http_status": "服务商返回 HTTP {status}", "network_error": "网络错误：{name}" }, + "realtime": { + "error_4002": "实时语音识别鉴权失败，请检查 SecretId / SecretKey。", + "error_4003": "当前腾讯云账号尚未开通实时语音识别服务。", + "error_4004": "腾讯云实时语音识别的资源包额度已耗尽。", + "error_4005": "腾讯云账号已欠费停服，请充值后重试。", + "error_4006": "实时语音识别并发数超限，请稍后再试。", + "error_4007": "音频解码失败，请检查麦克风输入。", + "error_4008": "音频发送超时，请保持网络稳定。", + "error_4009": "实时语音识别连接已断开。", + "error_5000": "腾讯云服务暂时不可用，请重试。", + "error_unknown": "实时语音识别失败，请稍后重试。" + }, "tencent": { "AuthFailure": { "SecretIdNotFound": "SecretId 不存在，请检查密钥是否填写正确。", diff --git a/src/octop/infra/errors.py b/src/octop/infra/errors.py index 29348b25d..95735f3dd 100644 --- a/src/octop/infra/errors.py +++ b/src/octop/infra/errors.py @@ -67,6 +67,7 @@ class ErrorCode(StrEnum): VOICE_KIND_UNSUPPORTED = "VOICE_KIND_UNSUPPORTED" VOICE_CAPABILITY_MISMATCH = "VOICE_CAPABILITY_MISMATCH" VOICE_PROVIDER_DISABLED = "VOICE_PROVIDER_DISABLED" + VOICE_REALTIME_UNAVAILABLE = "VOICE_REALTIME_UNAVAILABLE" INTERNAL_ERROR = "INTERNAL_ERROR" TLS_NOT_ELIGIBLE = "TLS_NOT_ELIGIBLE" TLS_ISSUE_IN_PROGRESS = "TLS_ISSUE_IN_PROGRESS" @@ -177,6 +178,7 @@ _DEFAULT_STATUS: dict[ErrorCode, int] = { ErrorCode.VOICE_KIND_UNSUPPORTED: 400, ErrorCode.VOICE_CAPABILITY_MISMATCH: 400, ErrorCode.VOICE_PROVIDER_DISABLED: 400, + ErrorCode.VOICE_REALTIME_UNAVAILABLE: 400, ErrorCode.INTERNAL_ERROR: 500, ErrorCode.TLS_NOT_ELIGIBLE: 403, ErrorCode.TLS_ISSUE_IN_PROGRESS: 409, diff --git a/src/octop/infra/utils/tencent_asr_sign.py b/src/octop/infra/utils/tencent_asr_sign.py new file mode 100644 index 000000000..5ade9bcd8 --- /dev/null +++ b/src/octop/infra/utils/tencent_asr_sign.py @@ -0,0 +1,70 @@ +"""Tencent Cloud realtime speech recognition (ASR V2) WebSocket URL signing. + +This is a *different* scheme from :mod:`octop.infra.utils.tencent_sign`, which +implements TC3-HMAC-SHA256 for the JSON APIs (``SentenceRecognition`` / +``TextToVoice``). Realtime ASR V2 authenticates through query parameters: + +1. every parameter except ``signature`` is sorted by key and joined into + ``asr.cloud.tencent.com/asr/v2/?k1=v1&k2=v2`` (no ``wss://`` prefix, + and the values are *not* percent-encoded); +2. that string is signed with ``HMAC-SHA1(secret_key)`` and base64-encoded; +3. only the resulting ``signature`` value is percent-encoded when appended to + the final URL. +""" + +from __future__ import annotations + +import base64 +import hashlib +import hmac +import secrets +import time +from urllib.parse import quote +from uuid import uuid4 + +ASR_V2_HOST = "asr.cloud.tencent.com" +#: Large-model engine; supports Mandarin + English + Cantonese and 30 dialects. +DEFAULT_ENGINE = "16k_zh_en_2.0" +#: Tencent's ``voice_format`` token for raw 16-bit mono PCM. +PCM_VOICE_FORMAT = 1 +DEFAULT_EXPIRED_SECONDS = 3600 + + +def asr_v2_signature(origin: str, secret_key: str) -> str: + """Return the base64 HMAC-SHA1 signature for a realtime ASR ``origin``.""" + digest = hmac.new(secret_key.encode("utf-8"), origin.encode("utf-8"), hashlib.sha1).digest() + return base64.b64encode(digest).decode("ascii") + + +def build_asr_v2_url( + *, + app_id: str, + secret_id: str, + secret_key: str, + engine: str = DEFAULT_ENGINE, + voice_id: str | None = None, + timestamp: int | None = None, + nonce: int | None = None, + expired_seconds: int = DEFAULT_EXPIRED_SECONDS, +) -> str: + """Build a signed ``wss://`` URL for one realtime ASR connection. + + ``voice_id`` must be unique per connection (a new one per call by default), + and ``timestamp`` / ``nonce`` are injectable so tests can pin the signature. + """ + issued_at = int(time.time()) if timestamp is None else int(timestamp) + params: dict[str, str] = { + "secretid": secret_id, + "timestamp": str(issued_at), + "expired": str(issued_at + int(expired_seconds)), + "nonce": str(secrets.randbelow(2**31) if nonce is None else int(nonce)), + "engine_model_type": engine, + "voice_id": voice_id or uuid4().hex, + "voice_format": str(PCM_VOICE_FORMAT), + "needvad": "1", + } + ordered = sorted(params) + path = f"{ASR_V2_HOST}/asr/v2/{app_id}" + query = "&".join(f"{key}={params[key]}" for key in ordered) + signature = quote(asr_v2_signature(f"{path}?{query}", secret_key), safe="") + return f"wss://{path}?{query}&signature={signature}" diff --git a/src/octop/infra/voice/adapters.py b/src/octop/infra/voice/adapters.py index aa16a1bf7..3054b3ee3 100644 --- a/src/octop/infra/voice/adapters.py +++ b/src/octop/infra/voice/adapters.py @@ -38,7 +38,7 @@ class BrowserOnlyError(Exception): """Raised when the active provider must run in the browser.""" -def _parse_tencent_credentials(row: VoiceProviderRow) -> tuple[str, str]: +def parse_tencent_credentials(row: VoiceProviderRow) -> tuple[str, str]: extra = row.get_extra() secret_id = extra.get("secret_id") or "" secret_key = extra.get("secret_key") or "" @@ -163,7 +163,7 @@ async def transcribe_tencent( mime: str, language: str, ) -> STTResult: - secret_id, secret_key = _parse_tencent_credentials(row) + secret_id, secret_key = parse_tencent_credentials(row) extra = row.get_extra() eng = str(extra.get("eng_service_type") or "16k_zh") if language.lower().startswith("en"): @@ -209,7 +209,7 @@ async def synthesize_tencent( voice_id: str | None, speed: float, ) -> AsyncIterator[bytes]: - secret_id, secret_key = _parse_tencent_credentials(row) + secret_id, secret_key = parse_tencent_credentials(row) extra = row.get_extra() voice_type = int(voice_id or extra.get("voice_type") or 101001) payload: dict[str, Any] = { @@ -477,7 +477,7 @@ def _missing_credentials(row: VoiceProviderRow, kind: str, *, locale: str = "en" """Probe-time credential check; returns an error message when incomplete.""" if kind == "tencent": try: - _parse_tencent_credentials(row) + parse_tencent_credentials(row) except ValueError: return voice_credentials_error(kind, locale) return None diff --git a/src/octop/infra/voice/manager.py b/src/octop/infra/voice/manager.py index 6f96cae44..455b8a604 100644 --- a/src/octop/infra/voice/manager.py +++ b/src/octop/infra/voice/manager.py @@ -9,7 +9,7 @@ from typing import Any from octop.infra.db.repos.settings import SettingsRepo from octop.infra.db.repos.voice_providers import VoiceProviderRepo, VoiceProviderRow from octop.infra.errors import ErrorCode, OctopError -from octop.infra.voice import adapters +from octop.infra.voice import adapters, realtime from octop.infra.voice.presets import is_builtin_preset @@ -89,6 +89,35 @@ class VoiceManager: kind = self.resolve(name).kind return "audio/wav" if kind == "mimo" else "audio/mpeg" + def realtime_stt_config(self) -> realtime.RealtimeConfig | None: + """Credentials for streaming STT, or ``None`` when it is not available. + + Streaming only applies to Tencent Cloud, requires the ``realtime_stt`` + switch on that provider row plus an ``app_id``, and is ignored unless the + provider is the active STT service. + """ + try: + resolved = self.resolve(self.get_active()["stt"]) + except OctopError: + return None + if resolved.kind != "tencent" or resolved.row is None: + return None + extra = resolved.row.get_extra() + if not extra.get("realtime_stt"): + return None + app_id = str(extra.get("app_id") or "").strip() + if not app_id: + return None + try: + secret_id, secret_key = adapters.parse_tencent_credentials(resolved.row) + except ValueError: + return None + return realtime.RealtimeConfig( + app_id=app_id, + secret_id=secret_id, + secret_key=secret_key, + ) + async def transcribe( self, audio: bytes, diff --git a/src/octop/infra/voice/realtime.py b/src/octop/infra/voice/realtime.py new file mode 100644 index 000000000..383c2cef9 --- /dev/null +++ b/src/octop/infra/voice/realtime.py @@ -0,0 +1,217 @@ +"""Tencent Cloud realtime speech recognition (ASR V2) upstream session. + +One :class:`TencentRealtimeSession` owns a single connection to +``wss://asr.cloud.tencent.com/asr/v2/``. It is deliberately transport +agnostic: it pushes raw PCM frames upstream and yields the parsed JSON events +back, leaving the browser-facing WebSocket, localisation and error mapping to +the HTTP layer. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +from collections.abc import AsyncIterator +from dataclasses import dataclass +from typing import Any, Literal + +from octop.infra.errors import ErrorCode, OctopError +from octop.infra.utils.tencent_asr_sign import DEFAULT_ENGINE, build_asr_v2_url + +logger = logging.getLogger(__name__) + +HANDSHAKE_TIMEOUT_SECONDS = 10.0 +#: How long the router waits for ``final: 1`` after the client says "end". +READER_TIMEOUT_SECONDS = 15.0 +CLOSE_TIMEOUT_SECONDS = 5.0 +MAX_MESSAGE_BYTES = 1 << 20 + + +@dataclass(frozen=True) +class RealtimeConfig: + """Credentials needed to open one realtime ASR connection.""" + + app_id: str + secret_id: str + secret_key: str + engine: str = DEFAULT_ENGINE + + +@dataclass(frozen=True) +class RealtimeEvent: + """One normalized upstream event. + + ``interim`` text is still being refined, ``final`` text replaces that + sentence, ``done`` marks the end of the stream and ``error`` carries a + Tencent error code. + """ + + kind: Literal["interim", "final", "done", "error"] + text: str = "" + sentence_id: int = 0 + error_code: int = 0 + + +def _as_int(value: Any) -> int: + try: + return int(value) + except (TypeError, ValueError): + return 0 + + +def normalize_event(event: dict[str, Any]) -> RealtimeEvent | None: + """Map a raw upstream frame onto a :class:`RealtimeEvent`. + + Tencent returns two shapes for the same stream: ``16k_zh_en_2.0`` sends + ``result`` (``slice_type`` 0 = 中间结果, 1 = 稳定结果, 2 = 该句最终结果, + ``index`` = sentence id, ``voice_text_str`` = text), while the API docs + describe ``sentences``. Both are accepted. + """ + code = _as_int(event.get("code")) + if code != 0: + return RealtimeEvent(kind="error", error_code=code) + + sentences = event.get("sentences") + if isinstance(sentences, dict) and sentences.get("sentence"): + return RealtimeEvent( + kind="final" if _as_int(sentences.get("sentence_type")) == 1 else "interim", + text=str(sentences["sentence"]), + sentence_id=_as_int(sentences.get("sentence_id")), + ) + + result = event.get("result") + if isinstance(result, dict) and result.get("voice_text_str"): + return RealtimeEvent( + kind="interim" if _as_int(result.get("slice_type")) == 0 else "final", + text=str(result["voice_text_str"]), + sentence_id=_as_int(result.get("index")), + ) + + if _as_int(event.get("final")) == 1: + return RealtimeEvent(kind="done") + return None + + +def _parse_control(raw: str | bytes) -> tuple[int, str] | None: + """Return ``(code, message)`` of an upstream control frame, or ``None``.""" + try: + data = json.loads(raw if isinstance(raw, str) else raw.decode("utf-8")) + except (json.JSONDecodeError, UnicodeDecodeError): + return None + if not isinstance(data, dict): + return None + try: + code = int(data.get("code") or 0) + except (TypeError, ValueError): + code = 0 + return code, str(data.get("message") or "") + + +class TencentRealtimeSession: + """A live realtime ASR stream for one browser session.""" + + def __init__(self, config: RealtimeConfig) -> None: + self._config = config + self._ws: Any = None + + @property + def connected(self) -> bool: + return self._ws is not None + + async def connect(self) -> None: + """Open the upstream connection and verify the handshake.""" + from websockets.asyncio.client import connect as ws_connect + from websockets.exceptions import WebSocketException + + url = build_asr_v2_url( + app_id=self._config.app_id, + secret_id=self._config.secret_id, + secret_key=self._config.secret_key, + engine=self._config.engine, + ) + try: + self._ws = await ws_connect( + url, + # websockets >= 14 defaults to proxy=True, which silently routes + # through HTTPS_PROXY / ALL_PROXY when the deployment sets one. + proxy=None, + # Tencent does not answer WebSocket pings; keepalive would abort. + ping_interval=None, + open_timeout=HANDSHAKE_TIMEOUT_SECONDS, + close_timeout=CLOSE_TIMEOUT_SECONDS, + max_size=MAX_MESSAGE_BYTES, + ) + raw = await asyncio.wait_for(self._ws.recv(), timeout=HANDSHAKE_TIMEOUT_SECONDS) + except (WebSocketException, OSError, TimeoutError) as exc: + await self.close() + raise OctopError( + ErrorCode.VOICE_REALTIME_UNAVAILABLE, + f"realtime ASR connect failed: {exc}", + ) from exc + + control = _parse_control(raw) + if control is None: + await self.close() + raise OctopError( + ErrorCode.VOICE_REALTIME_UNAVAILABLE, + "realtime ASR sent an unexpected handshake frame", + ) + code, message = control + if code != 0: + await self.close() + raise OctopError( + ErrorCode.VOICE_REALTIME_UNAVAILABLE, + f"realtime ASR handshake failed: {message or code}", + details={"tencent_code": code}, + ) + + async def send_audio(self, pcm: bytes) -> None: + """Forward one PCM frame (16 kHz / 16-bit / mono) upstream.""" + if self._ws is None or not pcm: + return + await self._ws.send(pcm) + + async def finish(self) -> None: + """Tell Tencent the audio stream ended so it can flush ``final``.""" + if self._ws is None: + return + await self._ws.send(json.dumps({"type": "end"})) + + async def events(self) -> AsyncIterator[RealtimeEvent]: + """Yield normalized events until the connection closes.""" + from websockets.exceptions import ConnectionClosed + + if self._ws is None: + return + while True: + try: + raw = await self._ws.recv() + except ConnectionClosed: + return + if isinstance(raw, (bytes, bytearray)): + continue + try: + frame = json.loads(raw) + except json.JSONDecodeError: + logger.debug("realtime ASR sent a non-JSON text frame") + continue + if not isinstance(frame, dict): + continue + event = normalize_event(frame) + if event is not None: + yield event + + async def close(self) -> None: + ws, self._ws = self._ws, None + if ws is None: + return + try: + await ws.close() + except Exception: # a failed close must not mask the real error + logger.debug("realtime ASR close failed", exc_info=True) + + +def open_session(config: RealtimeConfig) -> TencentRealtimeSession: + """Factory kept separate so tests can monkeypatch the upstream session.""" + return TencentRealtimeSession(config) diff --git a/tests/integration/test_voice_stt_stream_ws.py b/tests/integration/test_voice_stt_stream_ws.py new file mode 100644 index 000000000..5936ace5c --- /dev/null +++ b/tests/integration/test_voice_stt_stream_ws.py @@ -0,0 +1,196 @@ +"""Realtime (streaming) Tencent STT WebSocket.""" + +from __future__ import annotations + +import json +from collections.abc import AsyncIterator +from pathlib import Path +from typing import Any + +import pytest +from starlette.websockets import WebSocketDisconnect + +from octop.infra.errors import ErrorCode, OctopError +from octop.infra.voice import realtime +from octop.infra.voice.manager import VoiceManager +from octop.infra.voice.realtime import RealtimeEvent +from tests.support.app import octop_client +from tests.support.auth import auth_header, bootstrap_admin +from tests.support.http import ws_connect, ws_token + +TENCENT_REALTIME_EXTRA = { + "secret_id": "sid", + "secret_key": "skey", + "region": "ap-guangzhou", + "realtime_stt": True, + "app_id": "1302566622", +} + + +@pytest.fixture +async def env(tmp_octop_home: Path) -> AsyncIterator[Any]: + async with octop_client(tmp_octop_home) as (c, srv): + await bootstrap_admin(c, tmp_octop_home) + yield c, srv, await auth_header(c) + + +def _manager(srv: Any) -> VoiceManager: + return VoiceManager( + settings_repo=srv.services.settings_repo, + voice_provider_repo=srv.services.voice_provider_repo, + ) + + +def _enable_realtime(srv: Any, extra: dict[str, object] | None = None) -> None: + srv.services.voice_provider_repo.create( + name="tencent", + kind="tencent", + capability="both", + api_key="sid:skey", + extra_json=json.dumps(TENCENT_REALTIME_EXTRA if extra is None else extra), + ) + _manager(srv).set_active(stt="tencent") + + +def _stream_ws(c: Any, auth: dict[str, str] | None = None) -> Any: + query = f"?token={ws_token(auth)}&locale=zh" if auth else "?locale=zh" + return ws_connect(c._octop_app, f"/api/voice/stt-stream{query}") + + +class FakeSession: + """Stands in for the upstream Tencent connection.""" + + def __init__( + self, + frames: list[RealtimeEvent], + *, + connect_error: OctopError | None = None, + ) -> None: + self._frames = frames + self._connect_error = connect_error + self.audio: list[bytes] = [] + self.finished = False + self.closed = False + self.opened = False + + async def connect(self) -> None: + if self._connect_error is not None: + raise self._connect_error + self.opened = True + + async def send_audio(self, pcm: bytes) -> None: + self.audio.append(pcm) + + async def finish(self) -> None: + self.finished = True + + async def events(self) -> AsyncIterator[RealtimeEvent]: + for frame in self._frames: + yield frame + + async def close(self) -> None: + self.closed = True + + +def _install_fake_session( + monkeypatch: pytest.MonkeyPatch, + frames: list[RealtimeEvent], + *, + connect_error: OctopError | None = None, +) -> FakeSession: + session = FakeSession(frames, connect_error=connect_error) + monkeypatch.setattr(realtime, "open_session", lambda _config: session) + return session + + +def _sentence(text: str, *, final: bool, sentence_id: int = 0) -> RealtimeEvent: + return RealtimeEvent(kind="final" if final else "interim", text=text, sentence_id=sentence_id) + + +async def test_missing_token_is_rejected(env: Any) -> None: + c, _srv, _auth = env + with pytest.raises(WebSocketDisconnect): + async with _stream_ws(c): + pass + + +async def test_streams_audio_and_relays_results(env: Any, monkeypatch: pytest.MonkeyPatch) -> None: + c, srv, auth = env + _enable_realtime(srv) + session = _install_fake_session( + monkeypatch, + [ + _sentence("你好", final=False), + _sentence("你好世界", final=True), + RealtimeEvent(kind="done"), + ], + ) + + async with _stream_ws(c, auth) as ws: + assert await ws.receive_json() == {"type": "ready"} + await ws.send_bytes(b"\x00" * 1280) + await ws.send_bytes(b"\x01" * 1280) + await ws.send_json({"type": "end"}) + frames = await ws.drain_turn() + + assert frames == [ + {"type": "interim", "text": "你好", "sentence_id": 0}, + {"type": "final", "text": "你好世界", "sentence_id": 0}, + {"type": "done"}, + ] + assert session.audio == [b"\x00" * 1280, b"\x01" * 1280] + assert session.finished is True + assert session.closed is True + + +async def test_realtime_must_be_configured(env: Any, monkeypatch: pytest.MonkeyPatch) -> None: + c, srv, auth = env + _install_fake_session(monkeypatch, []) + + async with _stream_ws(c, auth) as ws: + frame = await ws.receive_json() + assert frame["type"] == "error" + assert frame["code"] == ErrorCode.VOICE_REALTIME_UNAVAILABLE.value + with pytest.raises(WebSocketDisconnect): + await ws.receive_json() + + +async def test_upstream_handshake_failure_is_reported( + env: Any, monkeypatch: pytest.MonkeyPatch +) -> None: + c, srv, auth = env + _enable_realtime(srv) + _install_fake_session( + monkeypatch, + [], + connect_error=OctopError( + ErrorCode.VOICE_REALTIME_UNAVAILABLE, + "handshake failed", + details={"tencent_code": 4003}, + ), + ) + + async with _stream_ws(c, auth) as ws: + frame = await ws.receive_json() + assert frame["type"] == "error" + assert frame["tencent_code"] == 4003 + assert "实时语音识别" in frame["message"] + + +async def test_upstream_stream_error_is_forwarded( + env: Any, monkeypatch: pytest.MonkeyPatch +) -> None: + c, srv, auth = env + _enable_realtime(srv) + _install_fake_session(monkeypatch, [RealtimeEvent(kind="error", error_code=4004)]) + + async with _stream_ws(c, auth) as ws: + assert await ws.receive_json() == {"type": "ready"} + await ws.send_bytes(b"\x00" * 1280) + + error = await ws.receive_json() + assert error["type"] == "error" + assert error["code"] == ErrorCode.VOICE_REALTIME_UNAVAILABLE.value + assert error["tencent_code"] == 4004 + assert "资源包" in error["message"] + assert await ws.receive_json() == {"type": "done"} diff --git a/tests/support/http.py b/tests/support/http.py index 1ade227e8..b3bebdf74 100644 --- a/tests/support/http.py +++ b/tests/support/http.py @@ -95,6 +95,9 @@ class ASGIWebSocketSession: {"type": "websocket.receive", "text": json.dumps(payload, ensure_ascii=False)}, ) + async def send_bytes(self, payload: bytes) -> None: + await self._to_app.put({"type": "websocket.receive", "bytes": payload}) + async def receive_json(self, *, timeout: float = _WS_RECEIVE_TIMEOUT_S) -> dict[str, Any]: event = await self._next_event(timeout) if event["type"] == "websocket.close": diff --git a/tests/unit/i18n/test_voice.py b/tests/unit/i18n/test_voice.py index cc045357d..a662bd8dc 100644 --- a/tests/unit/i18n/test_voice.py +++ b/tests/unit/i18n/test_voice.py @@ -4,7 +4,11 @@ from __future__ import annotations import httpx -from octop.i18n.domains.voice import format_voice_probe_error, tencent_api_language +from octop.i18n.domains.voice import ( + format_voice_probe_error, + realtime_error_message, + tencent_api_language, +) def test_tencent_secret_id_is_localized() -> None: @@ -32,3 +36,19 @@ def test_tencent_api_language_header() -> None: assert tencent_api_language("zh") == "zh-CN" assert tencent_api_language("en") == "en-US" assert tencent_api_language(None) is None + + +def test_realtime_error_codes_are_localized() -> None: + assert realtime_error_message(4003, "zh") == "当前腾讯云账号尚未开通实时语音识别服务。" + assert realtime_error_message(4004, "en") == ( + "The Tencent Cloud realtime speech recognition resource pack is exhausted." + ) + # 5001 / 5002 是同一个偶发故障文案 + assert realtime_error_message(5002, "zh") == "腾讯云服务暂时不可用，请重试。" + + +def test_unknown_realtime_code_falls_back() -> None: + assert realtime_error_message(4999, "zh") == "实时语音识别失败，请稍后重试。" + assert realtime_error_message(None, "en") == ( + "Realtime speech recognition failed. Retry in a moment." + ) diff --git a/tests/unit/test_voice_manager.py b/tests/unit/test_voice_manager.py index 66d239f64..2b39dadc6 100644 --- a/tests/unit/test_voice_manager.py +++ b/tests/unit/test_voice_manager.py @@ -2,6 +2,7 @@ from __future__ import annotations +import json from pathlib import Path import pytest @@ -11,6 +12,7 @@ from octop.infra.db.pool import SqlitePool from octop.infra.db.repos.settings import SettingsRepo from octop.infra.db.repos.voice_providers import VoiceProviderRepo, VoiceProviderRow from octop.infra.errors import ErrorCode, OctopError +from octop.infra.utils.tencent_asr_sign import DEFAULT_ENGINE from octop.infra.voice import adapters from octop.infra.voice.manager import VoiceManager @@ -24,6 +26,34 @@ def voice_mgr(tmp_path: Path) -> VoiceManager: return VoiceManager(settings_repo=settings, voice_provider_repo=repo) +@pytest.fixture +def tencent_mgr(tmp_path: Path) -> tuple[VoiceManager, VoiceProviderRepo]: + db = SqlitePool(tmp_path / "octop.db") + run_migrations(db) + repo = VoiceProviderRepo(db) + manager = VoiceManager(settings_repo=SettingsRepo(db), voice_provider_repo=repo) + return manager, repo + + +def _add_tencent(repo: VoiceProviderRepo, extra: dict[str, object]) -> None: + repo.create( + name="tencent", + kind="tencent", + capability="both", + api_key="sid:skey", + extra_json=json.dumps(extra), + ) + + +TENCENT_REALTIME_EXTRA: dict[str, object] = { + "secret_id": "sid", + "secret_key": "skey", + "region": "ap-guangzhou", + "realtime_stt": True, + "app_id": "1302566622", +} + + def test_default_active_is_browser(voice_mgr: VoiceManager) -> None: assert voice_mgr.get_active() == {"stt": "browser", "tts": "browser"} @@ -88,3 +118,60 @@ async def test_configuration_probe_uses_unsaved_values( assert captured_row.name == "draft" assert captured_row.api_key == "sk-draft" assert captured_row.base_url == "https://example.test/v1" + + +def test_realtime_config_is_none_by_default(voice_mgr: VoiceManager) -> None: + assert voice_mgr.realtime_stt_config() is None + + +def test_realtime_config_needs_tencent_as_active_stt( + tencent_mgr: tuple[VoiceManager, VoiceProviderRepo], +) -> None: + manager, repo = tencent_mgr + _add_tencent(repo, TENCENT_REALTIME_EXTRA) + + assert manager.realtime_stt_config() is None + + manager.set_active(stt="tencent") + config = manager.realtime_stt_config() + assert config is not None + assert config.app_id == "1302566622" + assert config.secret_id == "sid" + assert config.secret_key == "skey" + assert config.engine == DEFAULT_ENGINE + + +def test_realtime_config_needs_the_switch_on( + tencent_mgr: tuple[VoiceManager, VoiceProviderRepo], +) -> None: + manager, repo = tencent_mgr + _add_tencent(repo, {**TENCENT_REALTIME_EXTRA, "realtime_stt": False}) + manager.set_active(stt="tencent") + + assert manager.realtime_stt_config() is None + + +@pytest.mark.parametrize("app_id", ["", " ", None]) +def test_realtime_config_needs_an_app_id( + tencent_mgr: tuple[VoiceManager, VoiceProviderRepo], app_id: str | None +) -> None: + manager, repo = tencent_mgr + _add_tencent(repo, {**TENCENT_REALTIME_EXTRA, "app_id": app_id}) + manager.set_active(stt="tencent") + + assert manager.realtime_stt_config() is None + + +def test_realtime_config_needs_complete_credentials( + tencent_mgr: tuple[VoiceManager, VoiceProviderRepo], +) -> None: + manager, repo = tencent_mgr + repo.create( + name="tencent", + kind="tencent", + capability="both", + extra_json=json.dumps({"realtime_stt": True, "app_id": "1302566622"}), + ) + manager.set_active(stt="tencent") + + assert manager.realtime_stt_config() is None diff --git a/tests/unit/test_voice_realtime.py b/tests/unit/test_voice_realtime.py new file mode 100644 index 000000000..5840d0dac --- /dev/null +++ b/tests/unit/test_voice_realtime.py @@ -0,0 +1,231 @@ +"""Unit tests for the Tencent realtime ASR upstream session.""" + +from __future__ import annotations + +import json +from typing import Any + +import pytest +import websockets.asyncio.client as ws_client +from websockets.exceptions import ConnectionClosed + +from octop.infra.errors import ErrorCode, OctopError +from octop.infra.voice import realtime +from octop.infra.voice.realtime import RealtimeEvent + +CONFIG = realtime.RealtimeConfig(app_id="1302566622", secret_id="sid", secret_key="skey") + + +class FakeConnection: + def __init__(self, frames: list[Any]) -> None: + self._frames = list(frames) + self.sent: list[Any] = [] + self.closed = False + + async def recv(self) -> Any: + if not self._frames: + raise ConnectionClosed(None, None) + return self._frames.pop(0) + + async def send(self, payload: Any) -> None: + self.sent.append(payload) + + async def close(self) -> None: + self.closed = True + + +def _patch_connect( + monkeypatch: pytest.MonkeyPatch, + connection: FakeConnection | None, + *, + error: Exception | None = None, + capture: list[tuple[str, dict[str, Any]]] | None = None, +) -> None: + async def fake_connect(url: str, **kwargs: Any) -> FakeConnection: + if capture is not None: + capture.append((url, kwargs)) + if error is not None: + raise error + assert connection is not None + return connection + + monkeypatch.setattr(ws_client, "connect", fake_connect) + + +def _handshake(code: int = 0, message: str = "success") -> str: + return json.dumps({"code": code, "message": message, "voice_id": "v"}) + + +@pytest.mark.asyncio +async def test_connect_signs_the_url_and_accepts_the_handshake( + monkeypatch: pytest.MonkeyPatch, +) -> None: + conn = FakeConnection([_handshake()]) + capture: list[tuple[str, dict[str, Any]]] = [] + _patch_connect(monkeypatch, conn, capture=capture) + + session = realtime.open_session(CONFIG) + await session.connect() + + url, kwargs = capture[0] + assert url.startswith("wss://asr.cloud.tencent.com/asr/v2/1302566622?") + assert "engine_model_type=16k_zh_en_2.0" in url + assert "voice_format=1" in url + assert "&signature=" in url + assert kwargs["proxy"] is None + assert kwargs["ping_interval"] is None + assert session.connected is True + + +@pytest.mark.asyncio +async def test_connect_maps_a_tencent_error_code( + monkeypatch: pytest.MonkeyPatch, +) -> None: + conn = FakeConnection([_handshake(4003, "service is not open")]) + _patch_connect(monkeypatch, conn) + + session = realtime.open_session(CONFIG) + with pytest.raises(OctopError) as excinfo: + await session.connect() + + assert excinfo.value.code == ErrorCode.VOICE_REALTIME_UNAVAILABLE + assert excinfo.value.details["tencent_code"] == 4003 + assert conn.closed is True + assert session.connected is False + + +@pytest.mark.asyncio +async def test_connect_rejects_an_unexpected_handshake_frame( + monkeypatch: pytest.MonkeyPatch, +) -> None: + conn = FakeConnection(["definitely not json"]) + _patch_connect(monkeypatch, conn) + + session = realtime.open_session(CONFIG) + with pytest.raises(OctopError) as excinfo: + await session.connect() + + assert excinfo.value.code == ErrorCode.VOICE_REALTIME_UNAVAILABLE + assert "unexpected handshake" in excinfo.value.message + + +@pytest.mark.asyncio +async def test_connect_wraps_transport_failures( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _patch_connect(monkeypatch, None, error=OSError("network down")) + + session = realtime.open_session(CONFIG) + with pytest.raises(OctopError) as excinfo: + await session.connect() + + assert excinfo.value.code == ErrorCode.VOICE_REALTIME_UNAVAILABLE + assert "network down" in excinfo.value.message + + +@pytest.mark.asyncio +async def test_audio_and_end_frames_are_forwarded( + monkeypatch: pytest.MonkeyPatch, +) -> None: + conn = FakeConnection([_handshake()]) + _patch_connect(monkeypatch, conn) + + session = realtime.open_session(CONFIG) + await session.connect() + await session.send_audio(b"\x00" * 1280) + await session.finish() + + assert conn.sent[0] == b"\x00" * 1280 + assert json.loads(conn.sent[1]) == {"type": "end"} + + +@pytest.mark.asyncio +async def test_send_before_connect_is_a_noop() -> None: + session = realtime.open_session(CONFIG) + + await session.send_audio(b"\x00" * 1280) + await session.finish() + await session.close() + + +@pytest.mark.asyncio +async def test_events_yields_only_recognized_frames( + monkeypatch: pytest.MonkeyPatch, +) -> None: + conn = FakeConnection( + [ + _handshake(), # consumed by connect() + b"\x00\x01\x02", # binary frames are not control events + "not json", + json.dumps({"code": 0, "unrelated": {}}), + json.dumps(_result_frame("你好", slice_type=0)), + json.dumps({"code": 0, "final": 1}), + ] + ) + _patch_connect(monkeypatch, conn) + + session = realtime.open_session(CONFIG) + await session.connect() + events = [event async for event in session.events()] + + assert events == [ + RealtimeEvent(kind="interim", text="你好", sentence_id=0), + RealtimeEvent(kind="done"), + ] + + +def _result_frame(text: str, *, slice_type: int, index: int = 0) -> dict[str, Any]: + return { + "code": 0, + "result": { + "slice_type": slice_type, + "index": index, + "voice_text_str": text, + "start_time": 0, + "end_time": 500, + }, + } + + +def test_normalize_result_frames() -> None: + """`16k_zh_en_2.0` answers with `result`, not the documented `sentences`.""" + interim = realtime.normalize_event(_result_frame("今天", slice_type=0)) + stable = realtime.normalize_event(_result_frame("今天天气不错", slice_type=1)) + ultimate = realtime.normalize_event(_result_frame("今天天气不错。", slice_type=2, index=1)) + + assert interim == RealtimeEvent(kind="interim", text="今天", sentence_id=0) + assert stable == RealtimeEvent(kind="final", text="今天天气不错", sentence_id=0) + assert ultimate == RealtimeEvent(kind="final", text="今天天气不错。", sentence_id=1) + + +def test_normalize_sentences_frames() -> None: + """The documented `sentences` shape is still accepted.""" + frame = { + "code": 0, + "sentences": {"sentence": "你好", "sentence_type": 1, "sentence_id": 3}, + } + assert realtime.normalize_event(frame) == RealtimeEvent( + kind="final", text="你好", sentence_id=3 + ) + + +def test_normalize_terminal_and_error_frames() -> None: + assert realtime.normalize_event({"code": 0, "final": 1}) == RealtimeEvent(kind="done") + assert realtime.normalize_event({"code": 4004, "message": "exhausted"}) == RealtimeEvent( + kind="error", error_code=4004 + ) + assert realtime.normalize_event({"code": 0, "result": {"slice_type": 1}}) is None + + +@pytest.mark.asyncio +async def test_close_is_idempotent(monkeypatch: pytest.MonkeyPatch) -> None: + conn = FakeConnection([_handshake()]) + _patch_connect(monkeypatch, conn) + + session = realtime.open_session(CONFIG) + await session.connect() + await session.close() + await session.close() + + assert conn.closed is True + assert session.connected is False diff --git a/tests/unit/utils/test_tencent_asr_sign.py b/tests/unit/utils/test_tencent_asr_sign.py new file mode 100644 index 000000000..53b3a5ea4 --- /dev/null +++ b/tests/unit/utils/test_tencent_asr_sign.py @@ -0,0 +1,84 @@ +"""Tests for Tencent realtime ASR V2 WebSocket URL signing.""" + +from __future__ import annotations + +import base64 +import hashlib +import hmac +from urllib.parse import quote + +from octop.infra.utils.tencent_asr_sign import ( + ASR_V2_HOST, + DEFAULT_ENGINE, + DEFAULT_EXPIRED_SECONDS, + asr_v2_signature, + build_asr_v2_url, +) + +_ORIGIN = ( + f"{ASR_V2_HOST}/asr/v2/1302566622" + "?engine_model_type=16k_zh_en_2.0&expired=1700003600&needvad=1&nonce=12345" + "&secretid=AKIDSID×tamp=1700000000&voice_format=1&voice_id=vtest" +) +_SIGNATURE = "rhoZKciBwud6h4DGanqKV1YVWLw=" + + +def _fixed_url(**overrides: object) -> str: + kwargs: dict[str, object] = { + "app_id": "1302566622", + "secret_id": "AKIDSID", + "secret_key": "SKEY", + "voice_id": "vtest", + "timestamp": 1700000000, + "nonce": 12345, + } + kwargs.update(overrides) + return build_asr_v2_url(**kwargs) # type: ignore[arg-type] + + +def _query_of(url: str) -> str: + return url.split("?", 1)[1].rsplit("&signature=", 1)[0] + + +def test_signature_is_hmac_sha1_base64() -> None: + expected = base64.b64encode( + hmac.new(b"SKEY", _ORIGIN.encode("utf-8"), hashlib.sha1).digest() + ).decode("ascii") + assert expected == _SIGNATURE + assert asr_v2_signature(_ORIGIN, "SKEY") == _SIGNATURE + + +def test_build_url_matches_locked_origin_and_encoded_signature() -> None: + assert _fixed_url() == f"wss://{_ORIGIN}&signature={quote(_SIGNATURE, safe='')}" + + +def test_signature_value_is_percent_encoded() -> None: + tail = _fixed_url().split("&signature=", 1)[1] + assert tail == "rhoZKciBwud6h4DGanqKV1YVWLw%3D" + assert not any(ch in tail for ch in "=+/") + + +def test_params_are_sorted_and_pcm_is_selected() -> None: + query = _query_of(_fixed_url()) + keys = [part.split("=", 1)[0] for part in query.split("&")] + assert keys == sorted(keys) + assert "voice_format=1" in query + assert "needvad=1" in query + assert f"engine_model_type={DEFAULT_ENGINE}" in query + assert f"expired={1700000000 + DEFAULT_EXPIRED_SECONDS}" in query + + +def test_origin_values_are_not_percent_encoded() -> None: + url = _fixed_url(secret_id="sid+slash/eq=") + assert "secretid=sid+slash/eq=&" in url + + +def test_engine_is_overridable() -> None: + assert f"engine_model_type={'16k_zh'}&" in _fixed_url(engine="16k_zh") + "&" + + +def test_voice_id_is_regenerated_per_call() -> None: + first = build_asr_v2_url(app_id="a", secret_id="s", secret_key="k") + second = build_asr_v2_url(app_id="a", secret_id="s", secret_key="k") + assert first != second + assert "voice_id=" in first