From 4f9453b8b2f595a1a3a115dd730e22f2acc44026 Mon Sep 17 00:00:00 2001 From: 134 Date: Fri, 18 Sep 2026 14:12:28 +0800 Subject: [PATCH] Add durable remote audio inference and verified offline speech models --- .gitignore | 3 + .../canvas/smart-edit/capabilities/route.ts | 26 +- .../audio-processing-deployment-2026-09-18.md | 24 ++ lib/video-canvas/smart-edit/audio-contract.ts | 52 +++ lib/video-canvas/smart-edit/audio-process.ts | 98 +++-- lib/video-canvas/smart-edit/execute.ts | 3 + lib/video-canvas/smart-edit/local.ts | 90 ++++- lib/video-canvas/smart-edit/remote-audio.ts | 201 ++++++++++ playwright.helpers.config.ts | 1 + scripts/qa/audio-processing-service.py | 61 +++ services/audio-processing/README.md | 36 ++ .../audio-processing-tunnel.service | 17 + .../audio-processing/audio-processing.service | 28 ++ services/audio-processing/build.cjs | 25 ++ services/audio-processing/job.ts | 88 +++++ services/audio-processing/models.json | 86 ++++ services/audio-processing/prepare_models.py | 26 ++ .../audio-processing/requirements-lock.txt | 93 +++++ services/audio-processing/requirements.txt | 10 + services/audio-processing/run.py | 37 ++ services/audio-processing/server.ts | 372 ++++++++++++++++++ .../routes/video-canvas-audio-remote.test.ts | 130 ++++++ 22 files changed, 1458 insertions(+), 49 deletions(-) create mode 100644 docs/audio-processing-deployment-2026-09-18.md create mode 100644 lib/video-canvas/smart-edit/audio-contract.ts create mode 100644 lib/video-canvas/smart-edit/remote-audio.ts create mode 100644 scripts/qa/audio-processing-service.py create mode 100644 services/audio-processing/README.md create mode 100644 services/audio-processing/audio-processing-tunnel.service create mode 100644 services/audio-processing/audio-processing.service create mode 100644 services/audio-processing/build.cjs create mode 100644 services/audio-processing/job.ts create mode 100644 services/audio-processing/models.json create mode 100644 services/audio-processing/prepare_models.py create mode 100644 services/audio-processing/requirements-lock.txt create mode 100644 services/audio-processing/requirements.txt create mode 100644 services/audio-processing/run.py create mode 100644 services/audio-processing/server.ts create mode 100644 tests/routes/video-canvas-audio-remote.test.ts diff --git a/.gitignore b/.gitignore index 91e1bcc4c..ab3901e3c 100644 --- a/.gitignore +++ b/.gitignore @@ -124,3 +124,6 @@ services/smart-edit/__pycache__/ services/video-upscale/dist/ services/video-upscale/__pycache__/ services/video-upscale/vendor/__pycache__/ + +/services/audio-processing/dist/ +/services/audio-processing/__pycache__/ diff --git a/app/api/canvas/smart-edit/capabilities/route.ts b/app/api/canvas/smart-edit/capabilities/route.ts index 7f3899d42..38cb6e797 100644 --- a/app/api/canvas/smart-edit/capabilities/route.ts +++ b/app/api/canvas/smart-edit/capabilities/route.ts @@ -1,3 +1,4 @@ +import { remoteAudioCapabilities } from "@/lib/video-canvas/smart-edit/remote-audio"; import { auth } from "@/app/(auth)/auth"; import { NextResponse } from "next/server"; import { execFile } from "node:child_process"; @@ -8,7 +9,8 @@ import { isCanvasVisionModel } from "@/lib/video-canvas/model-matching"; import { remoteRenderConfig } from "@/lib/video-canvas/smart-edit/remote-render"; export async function GET() { const session = await auth(); - if (!session?.user?.id) return NextResponse.json({ error: "未登录" }, { status: 401 }); + if (!session?.user?.id) + return NextResponse.json({ error: "未登录" }, { status: 401 }); try { let capabilities = { separation: false, @@ -22,10 +24,12 @@ export async function GET() { const { stdout } = await promisify(execFile)( process.env.SMART_EDIT_PYTHON || "python3", [path.resolve("services/smart-edit/analyze.py"), "--check"], - { timeout: 10000, maxBuffer: 65536 }, + { timeout: 10000, maxBuffer: 65536 } ); capabilities = JSON.parse(stdout); } catch {} + const audio = await remoteAudioCapabilities(); + if (audio) capabilities = { ...capabilities, ...audio }; let renderAvailable = capabilities.ffmpeg; const remote = remoteRenderConfig(); if (remote) { @@ -36,21 +40,29 @@ export async function GET() { redirect: "error", cache: "no-store", }); - renderAvailable = response.ok && (await response.json()).status === "ok"; + renderAvailable = + response.ok && (await response.json()).status === "ok"; } catch { renderAvailable = false; } } const models = (await getEnabledAiModelsWithProvider(false)) .filter((item) => - isCanvasVisionModel({ ...item.model, providerType: item.provider.type }), + isCanvasVisionModel({ ...item.model, providerType: item.provider.type }) ) - .map(({ model }) => ({ id: model.id, name: model.displayName, modelId: model.modelId })); + .map(({ model }) => ({ + id: model.id, + name: model.displayName, + modelId: model.modelId, + })); return NextResponse.json( { capabilities: { ...capabilities, render: renderAvailable }, models }, - { headers: { "Cache-Control": "no-store" } }, + { headers: { "Cache-Control": "no-store" } } ); } catch { - return NextResponse.json({ error: "剪辑服务配置读取失败" }, { status: 503 }); + return NextResponse.json( + { error: "剪辑服务配置读取失败" }, + { status: 503 } + ); } } diff --git a/docs/audio-processing-deployment-2026-09-18.md b/docs/audio-processing-deployment-2026-09-18.md new file mode 100644 index 000000000..cee9387ca --- /dev/null +++ b/docs/audio-processing-deployment-2026-09-18.md @@ -0,0 +1,24 @@ +# 自动转写与人声分离独立服务 — 2026-09-18 + +本轮完成线上模型安装、算力服务器私有音频 API、主项目接入。服务部署说明见 `services/audio-processing/README.md`。 + +## 架构 + +- FunASR Paraformer + FSMN VAD + CT-Transformer 自动转写，HTDemucs 人声分离；全部使用已校验本地权重，推理不下载模型或执行远程模型代码。 +- 模型运行在算力服务器的独立 CPU 单队列；systemd 4 核 / 12GiB 限额，Torch 2线程。千问 GPU 0/1 和视频高清 GPU 3 配置未改。 +- 48 只负责轻量素材准备、持久化主任务和结果归档。既有智能剪辑准备锁保留，避免并发抽取素材压垮主站；推理不在48运行。 +- 私有 API 18342，独立随机 Bearer 密钥，经专用受限 SSH 隧道连接48，未开放公网端口。 +- 上传流校验长度及 SHA-256；服务端记录和结果持久化，刷新重试复用任务编号。输入未上传完不执行；服务重启可恢复；主站下载结果再次校验。 +- 功能能力取算力服务器真实 health；配置了远程服务后遇到网络错误不回退主机推理。错误保留原字幕/原音轨。 + +## 已验证 + +- 70 项定向回归通过，含原有剪辑/任务/编组和新增音频断线、身份隔离、损坏音轨拦截。 +- TypeScript 检查通过。 +- 真实音频 API：缺少认证401、参数冲突409、损坏上传400、相同任务复用、实际中文识别、实际两轨分离、分离中重启恢复。 +- 8.3935秒中文语音识别三句：0–1.349秒“欢迎使用智能剪辑，”；1.349–4.710秒“我们正在测试人声分离和自动字幕，”；4.710–8.205秒“请保留原始视频的画面和声音。” ASR单次实测14.03秒。 +- 人声与背景声合成重构原样例的最大误差 `7.45e-9`；波形确实发生分离，非原音轨复制。含重启恢复的分离验证24.20秒，attempts=2。这是短样例，不是任意素材的质量/性能承诺。 +- 在48执行实际主项目处理函数，经真实私有API调用算力服务器：转写保留手动字幕并新增3句；both/vocals/background三个分支分别生成2/1/1条独立音轨，保留原裁剪、时长、增益并关闭视频原声。测试隔离素材缓存与回传文件，不触碰用户画布；上传回调验证文件交付，不将此宣称为真实用户对象存储端到端验收。 +- 算力服务器及48的认证 health 均返回 asr=true、separation=true；服务 bundle SHA-256 为 `22c20736634cbed32c3a3b227e71ce773f5da5763d52c021cd6a162130b54d22`。 + +测试输入为合成语音/背景音，无付费生成调用。主站发布版本与构建结果在上线后补记。私有配置、模型权重、测试凭据、输入输出不提交 Git。 diff --git a/lib/video-canvas/smart-edit/audio-contract.ts b/lib/video-canvas/smart-edit/audio-contract.ts new file mode 100644 index 000000000..fd7ef4a99 --- /dev/null +++ b/lib/video-canvas/smart-edit/audio-contract.ts @@ -0,0 +1,52 @@ +import { z } from "zod"; +import { subtitleSchema } from "./contract"; +export const audioJobInputSchema = z + .object({ + operation: z.enum(["transcribe", "separate"]), + sha256: z.string().regex(/^[a-f0-9]{64}$/), + bytes: z + .number() + .int() + .positive() + .max(70 * 1024 * 1024), + }) + .strict(); +export type AudioJobInput = z.infer; +export const audioJobStatusSchema = z.object({ + id: z.string().regex(/^[a-f0-9]{64}$/), + status: z.enum([ + "awaiting-input", + "queued", + "running", + "succeeded", + "failed", + ]), + stage: z.string().max(300), + error: z.string().max(500).optional(), + result: z + .object({ + subtitles: z.array(subtitleSchema).max(500).optional(), + durationSeconds: z.number().positive().max(180.1), + stems: z + .object({ + vocals: z.object({ + bytes: z + .number() + .int() + .positive() + .max(70 * 1024 * 1024), + sha256: z.string().regex(/^[a-f0-9]{64}$/), + }), + background: z.object({ + bytes: z + .number() + .int() + .positive() + .max(70 * 1024 * 1024), + sha256: z.string().regex(/^[a-f0-9]{64}$/), + }), + }) + .optional(), + }) + .optional(), +}); diff --git a/lib/video-canvas/smart-edit/audio-process.ts b/lib/video-canvas/smart-edit/audio-process.ts index 181950b05..f7465f7ad 100644 --- a/lib/video-canvas/smart-edit/audio-process.ts +++ b/lib/video-canvas/smart-edit/audio-process.ts @@ -1,3 +1,4 @@ +import { remoteAudio, remoteAudioConfig } from "./remote-audio"; import { execFile } from "node:child_process"; import { promisify } from "node:util"; import { createHash, randomUUID } from "node:crypto"; @@ -31,15 +32,24 @@ export async function processEditAudio( upload: (file: string, suffix: string) => Promise, track: "video" | "audio" = "video", keep: "both" | "vocals" | "background" = "both", + requestKey: string = userId ): Promise { const timeline = planTimeline(plan); if (operation === "separate") { const index = plan.clips.findIndex((c) => c.id === clipId); - const audio = track === "audio" ? plan.audioClips?.find((c) => c.id === clipId) : undefined; + const audio = + track === "audio" + ? plan.audioClips?.find((c) => c.id === clipId) + : undefined; const clip = track === "audio" ? audio : plan.clips[index]; if (!clip || (track === "video" && plan.clips[index].audioDetached)) throw Error("请选择可分离的音视频片段"); - if ((plan.audioClips?.length || 0) + (keep === "both" ? 2 : 1) - (audio ? 1 : 0) > 24) + if ( + (plan.audioClips?.length || 0) + + (keep === "both" ? 2 : 1) - + (audio ? 1 : 0) > + 24 + ) throw Error("分离需要空闲音轨，最多支持 24 个音频片段"); const asset = plan.assets.find((a) => a.id === clip.assetId)!; if (!asset.hasAudio) throw Error("当前视频没有音轨"); @@ -70,22 +80,34 @@ export async function processEditAudio( inputAudio, ]); await progress("正在分离人声与背景声"); - try { - await promisify(execFile)( - process.env.SMART_EDIT_PYTHON || "python3", - [path.resolve("services/smart-edit/separate.py"), inputAudio, dir], - { timeout: 600000, maxBuffer: 2 * 1024 * 1024 }, + if (remoteAudioConfig()) { + await remoteAudio( + "separate", + inputAudio, + userId, + requestKey, + dir, + progress ); - } catch (e) { - const detail = String((e as { stderr?: string }).stderr || ""); - throw Error( - detail.includes("尚未配置") - ? "人声分离模型尚未配置" - : "人声分离失败或超时，原音轨已保留", - ); - } + } else + try { + await promisify(execFile)( + process.env.SMART_EDIT_PYTHON || "python3", + [path.resolve("services/smart-edit/separate.py"), inputAudio, dir], + { timeout: 600000, maxBuffer: 2 * 1024 * 1024 } + ); + } catch (e) { + const detail = String((e as { stderr?: string }).stderr || ""); + throw Error( + detail.includes("尚未配置") + ? "人声分离模型尚未配置" + : "人声分离失败或超时，原音轨已保留" + ); + } await progress("保存人声与背景声音轨"); - const audioClips = (plan.audioClips || []).filter((c) => !audio || c.id !== audio.id); + const audioClips = (plan.audioClips || []).filter( + (c) => !audio || c.id !== audio.id + ); const used = new Set([...plan.clips, ...audioClips].map((c) => c.assetId)); const assets = plan.assets.filter((a) => used.has(a.id)); const t = timeline.clips[index], @@ -97,7 +119,9 @@ export async function processEditAudio( if (keep !== "both" && keep !== stem) continue; const file = path.join(dir, stem + ".wav"), metadata = await probeMedia(file); - if (Math.abs(metadata.duration - (clip.sourceOut - clip.sourceIn)) > 50000) + if ( + Math.abs(metadata.duration - (clip.sourceOut - clip.sourceIn)) > 50000 + ) throw Error("分离音轨时长异常，原音轨已保留"); const id = randomUUID(), bytes = await readFile(file); @@ -119,9 +143,21 @@ export async function processEditAudio( gain: clip.gain, muted: clip.muted, fadeIn: audio ? audio.fadeIn : (t.overlap - t.padBefore) / SECOND, - fadeInStart: audio ? audio.fadeInStart : t.overlap ? t.padBefore / t.overlap : 0, - fadeOut: audio ? audio.fadeOut : next ? (next.overlap - t.padAfter) / SECOND : 0, - fadeOutEnd: audio ? audio.fadeOutEnd : next?.overlap ? t.padAfter / next.overlap : 0, + fadeInStart: audio + ? audio.fadeInStart + : t.overlap + ? t.padBefore / t.overlap + : 0, + fadeOut: audio + ? audio.fadeOut + : next + ? (next.overlap - t.padAfter) / SECOND + : 0, + fadeOutEnd: audio + ? audio.fadeOutEnd + : next?.overlap + ? t.padAfter / next.overlap + : 0, }); } return editPlanSchema.parse({ @@ -129,12 +165,12 @@ export async function processEditAudio( assets, audioClips, clips: plan.clips.map((c) => - !audio && c.id === clip.id ? { ...c, audioDetached: true } : c, + !audio && c.id === clip.id ? { ...c, audioDetached: true } : c ), }); } const captions: TimelineCaption[] = editableCaptions(plan).filter( - (c) => !c.id.startsWith("asr-"), + (c) => !c.id.startsWith("asr-") ); let recognized = 0; for (const [index, clip] of timeline.clips.entries()) { @@ -146,16 +182,26 @@ export async function processEditAudio( const actual = await probeMedia(media.file); if (!actual.hasAudio || Math.abs(actual.duration - asset.duration) > 100000) throw Error("原音轨参数已变化，请重新解析"); - const audio = await prepareAudio(media.file, path.join(media.dir, media.hash + ".wav")); - const analysis = await pythonAnalysis({ version: 1, audio, transcribe: true }, media.dir); + const audio = await prepareAudio( + media.file, + path.join(media.dir, media.hash + ".wav") + ); + const analysis = await pythonAnalysis( + { version: 1, audio, transcribe: true }, + media.dir, + requestKey, + progress + ); for (const [j, s] of analysis.subtitles.entries()) { const from = Math.max(s.start, clip.sourceIn), to = Math.min(s.end, clip.sourceOut); if (to <= from) continue; - const start = Math.round(clip.start + (from - clip.sourceIn) / clip.speed); + const start = Math.round( + clip.start + (from - clip.sourceIn) / clip.speed + ); const end = Math.min( timeline.duration, - Math.round(clip.start + (to - clip.sourceIn) / clip.speed), + Math.round(clip.start + (to - clip.sourceIn) / clip.speed) ); if (end <= start) continue; captions.push({ diff --git a/lib/video-canvas/smart-edit/execute.ts b/lib/video-canvas/smart-edit/execute.ts index 2067d100e..2d3cf593a 100644 --- a/lib/video-canvas/smart-edit/execute.ts +++ b/lib/video-canvas/smart-edit/execute.ts @@ -163,6 +163,8 @@ export async function executeSmartEdit( transcribe: params.transcribe && !input.subtitles.length, }, media.dir, + lifecycle.taskId, + (stage) => save({stage}), ); })(); const subtitles = input.subtitles.length ? input.subtitles : analysis.subtitles; @@ -310,6 +312,7 @@ export async function executeSmartEdit( upload, params.operation === "separate" ? params.track : "video", params.operation === "separate" ? params.keep : "both", + lifecycle.taskId, ); await save({ phase: "selected", stage: "音频处理完成", plans: [plan] }); return { plans: [plan] }; diff --git a/lib/video-canvas/smart-edit/local.ts b/lib/video-canvas/smart-edit/local.ts index 6c2da25c4..88cd6b741 100644 --- a/lib/video-canvas/smart-edit/local.ts +++ b/lib/video-canvas/smart-edit/local.ts @@ -1,6 +1,15 @@ +import { remoteAudio, remoteAudioConfig } from "./remote-audio"; import { execFile } from "node:child_process"; import { promisify } from "node:util"; -import { readFile, writeFile, mkdir, stat, rename, readdir, rm } from "node:fs/promises"; +import { + readFile, + writeFile, + mkdir, + stat, + rename, + readdir, + rm, +} from "node:fs/promises"; import path from "node:path"; import { createHash, randomUUID } from "node:crypto"; import { downloadCanvasMedia } from "../download-media"; @@ -28,19 +37,34 @@ export async function ffmpeg(args: string[], cwd?: string, timeout = 240000) { "1", ...args, ], - { cwd, timeout, maxBuffer: 2 * 1024 * 1024 }, + { cwd, timeout, maxBuffer: 2 * 1024 * 1024 } ); } export async function probeMedia(file: string, maxSeconds = 180) { const { stdout } = await exec( "ffprobe", - ["-v", "error", ...mediaSafety, "-show_streams", "-show_format", "-of", "json", file], - { timeout: 15000, maxBuffer: 1024 * 1024 }, + [ + "-v", + "error", + ...mediaSafety, + "-show_streams", + "-show_format", + "-of", + "json", + file, + ], + { timeout: 15000, maxBuffer: 1024 * 1024 } ); const info = JSON.parse(stdout), v = info.streams?.find((s: any) => s.codec_type === "video"); - const duration = Math.round(Number(info.format?.duration || v?.duration) * SECOND); - if (!Number.isFinite(duration) || duration <= 0 || duration > maxSeconds * SECOND) + const duration = Math.round( + Number(info.format?.duration || v?.duration) * SECOND + ); + if ( + !Number.isFinite(duration) || + duration <= 0 || + duration > maxSeconds * SECOND + ) throw Error(`媒体时长需在 0–${maxSeconds} 秒以内`); if (v && (!v.width || !v.height || v.width * v.height > 33554432)) throw Error("素材画面尺寸超过处理上限"); @@ -64,7 +88,10 @@ export async function materialize(url: string, userId: string) { const mapFile = path.join(dir, key + ".json"); try { const map = JSON.parse(await readFile(mapFile, "utf8")); - if (/^[a-f0-9]{64}$/.test(map.hash) && Date.now() - map.created < 86400000) { + if ( + /^[a-f0-9]{64}$/.test(map.hash) && + Date.now() - map.created < 86400000 + ) { const file = path.join(dir, map.hash); await stat(file); return { file, hash: map.hash as string, dir }; @@ -77,7 +104,9 @@ export async function materialize(url: string, userId: string) { const hash = createHash("sha256").update(bytes).digest("hex"), file = path.join(dir, hash); await rename(temp, file); - await writeFile(mapFile, JSON.stringify({ hash, created: Date.now() }), { mode: 0o600 }); + await writeFile(mapFile, JSON.stringify({ hash, created: Date.now() }), { + mode: 0o600, + }); return { file, hash, dir }; } finally { await rm(temp, { force: true }); @@ -113,18 +142,30 @@ export type LocalAnalysis = { export async function pythonAnalysis( data: Record, dir: string, + requestKey: string = dir, + progress?: (stage: string) => Promise ): Promise { + const remote = data.transcribe && data.audio ? remoteAudioConfig() : null; const key = createHash("sha256") - .update("analysis-v3:") + .update("analysis-v4:" + (remote?.base || "local") + ":") .update(JSON.stringify(data)) .digest("hex"), input = path.join(dir, key + ".analysis-input.json"), output = path.join(dir, key + ".analysis.json"); try { const value = JSON.parse(await readFile(output, "utf8")); - if (!value.error && Array.isArray(value.scenes)) return value; + if ( + !value.error && + Array.isArray(value.scenes) && + (!remote || value.remoteAudioDone === true) + ) + return value; } catch {} - await writeFile(input, JSON.stringify(data), { mode: 0o600 }); + await writeFile( + input, + JSON.stringify(remote ? { ...data, transcribe: false } : data), + { mode: 0o600 } + ); try { await exec( process.env.SMART_EDIT_PYTHON || "python3", @@ -132,19 +173,36 @@ export async function pythonAnalysis( { timeout: 300000, maxBuffer: 4 * 1024 * 1024, - env: { ...process.env, OMP_NUM_THREADS: "2", OPENBLAS_NUM_THREADS: "2" }, - }, + env: { + ...process.env, + OMP_NUM_THREADS: "2", + OPENBLAS_NUM_THREADS: "2", + }, + } ); } catch { const value = JSON.parse( await readFile(output, "utf8").catch( - () => '{"error":"分析运行环境不可用，请安装 services/smart-edit 依赖"}', - ), + () => '{"error":"分析运行环境不可用，请安装 services/smart-edit 依赖"}' + ) ); throw Error(value.error || "素材分析失败或超时"); } const value = JSON.parse(await readFile(output, "utf8")); if (value.error) throw Error(value.error); + if (remote) { + const result = await remoteAudio( + "transcribe", + String(data.audio), + dir, + requestKey, + dir, + progress + ); + value.subtitles = result.subtitles; + value.remoteAudioDone = true; + await writeFile(output, JSON.stringify(value), { mode: 0o600 }); + } value.subtitles = (value.subtitles || []) .map((s: unknown) => subtitleSchema.parse(s)) .slice(0, 500); @@ -154,7 +212,7 @@ export async function prepareAnalysis( file: string, hash: string, dir: string, - hasAudio: boolean, + hasAudio: boolean ) { const proxy = path.join(dir, hash + ".proxy.mp4"), audio = path.join(dir, hash + ".wav"); diff --git a/lib/video-canvas/smart-edit/remote-audio.ts b/lib/video-canvas/smart-edit/remote-audio.ts new file mode 100644 index 000000000..51e92a5cf --- /dev/null +++ b/lib/video-canvas/smart-edit/remote-audio.ts @@ -0,0 +1,201 @@ +import { createHash } from "node:crypto"; +import { createReadStream, createWriteStream } from "node:fs"; +import { stat, rm } from "node:fs/promises"; +import { Transform, Readable } from "node:stream"; +import { pipeline } from "node:stream/promises"; +import path from "node:path"; +import { ProviderTaskUncertainError } from "../video-task-lifecycle"; +import { audioJobInputSchema, audioJobStatusSchema } from "./audio-contract"; +const sha = (s: string) => createHash("sha256").update(s).digest("hex"); +export function remoteAudioConfig() { + const value = process.env.SMART_EDIT_AUDIO_URL; + if (!value) return null; + const url = new URL(value), + token = process.env.SMART_EDIT_AUDIO_TOKEN; + if ( + !["http:", "https:"].includes(url.protocol) || + url.username || + url.password || + url.search || + url.hash || + !token || + token.length < 32 + ) + throw Error("远程音频服务配置无效"); + return { base: value.replace(/\/$/, ""), token }; +} +export async function remoteAudioCapabilities() { + const config = remoteAudioConfig(); + if (!config) return null; + try { + const r = await fetch(config.base + "/health", { + headers: { authorization: "Bearer " + config.token }, + signal: AbortSignal.timeout(4000), + redirect: "error", + cache: "no-store", + }); + if (!r.ok) throw Error(); + const data = await r.json(); + return { + asr: data.status === "ok" && data.capabilities?.asr === true, + separation: + data.status === "ok" && data.capabilities?.separation === true, + }; + } catch { + return { asr: false, separation: false }; + } +} +export async function audioFileHash(file: string) { + const h = createHash("sha256"); + for await (const chunk of createReadStream(file)) h.update(chunk); + return h.digest("hex"); +} +export async function remoteAudio( + operation: "transcribe" | "separate", + file: string, + ownerKey: string, + requestKey: string, + dir: string, + progress: (stage: string) => Promise = async () => {} +) { + const config = remoteAudioConfig(); + if (!config) throw Error("远程音频服务未配置"); + const plan = audioJobInputSchema.parse({ + operation, + bytes: (await stat(file)).size, + sha256: await audioFileHash(file), + }); + const tenant = sha(ownerKey), + id = sha( + "audio-v1:" + tenant + ":" + requestKey + ":" + JSON.stringify(plan) + ); + const headers = { authorization: "Bearer " + config.token }; + async function request( + route: string, + init: RequestInit = {}, + timeoutMs?: number + ) { + let r: Response; + try { + r = await fetch(config!.base + route, { + ...init, + headers: { ...headers, ...init.headers }, + signal: AbortSignal.timeout( + timeoutMs || (init.method === "PUT" ? 180000 : 20000) + ), + redirect: "error", + cache: "no-store", + }); + } catch { + throw new ProviderTaskUncertainError( + "音频服务连接中断，任务已保留，可继续处理同一音频" + ); + } + if ( + (init.method === "PUT" && r.status === 409) || + [429, 500, 502, 503, 504].includes(r.status) + ) + throw new ProviderTaskUncertainError( + "音频服务繁忙，等待继续处理同一任务" + ); + if (!r.ok) throw Error(`音频服务请求失败（${r.status}）`); + return r; + } + async function status(r: Response) { + let value: unknown; + try { + value = await r.json(); + } catch { + throw new ProviderTaskUncertainError( + "音频服务响应中断，将继续查询同一任务" + ); + } + const s = audioJobStatusSchema.parse(value); + if (s.id !== id) throw Error("音频任务编号不匹配"); + return s; + } + let current = await status( + await request("/v1/audio-jobs", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ id, tenant, plan }), + }) + ); + if (current.status === "awaiting-input") { + await progress("传送音频到算力服务器"); + current = await status( + await request("/v1/audio-jobs/" + id + "/input", { + method: "PUT", + headers: { + "content-type": "audio/wav", + "content-length": String(plan.bytes), + }, + body: createReadStream(file) as any, + duplex: "half", + } as RequestInit) + ); + } + const deadline = Date.now() + 8 * 60000; + while (current.status === "queued" || current.status === "running") { + await progress(current.stage); + if (Date.now() > deadline) + throw new ProviderTaskUncertainError( + "远程音频仍在处理中，任务已保存，稍后继续查询" + ); + await new Promise((r) => setTimeout(r, 2000)); + current = await status(await request("/v1/audio-jobs/" + id)); + } + if (current.status === "failed") throw Error(current.error || "音频处理失败"); + if (current.status !== "succeeded" || !current.result) + throw Error("音频服务没有返回有效结果"); + const result = current.result; + if (operation === "transcribe") { + if (!result.subtitles) throw Error("音频服务缺少转写结果"); + if ( + result.subtitles.some( + (s) => s.end > result.durationSeconds * 1e6 + 100000 + ) + ) + throw Error("转写时间戳超出音频范围"); + } else { + if (!result.stems) throw Error("音频服务缺少分离结果"); + for (const stem of ["vocals", "background"] as const) { + await progress(stem === "vocals" ? "读取人声音轨" : "读取背景声音轨"); + const destination = path.join(dir, stem + ".wav"), + temp = destination + ".part"; + try { + const r = await request( + "/v1/audio-jobs/" + id + "/" + stem, + {}, + 180000 + ); + if (!r.body) throw Error(); + let bytes = 0; + const hash = createHash("sha256"), + expected = result.stems[stem]; + await pipeline( + Readable.fromWeb(r.body as any), + new Transform({ + transform(chunk, _enc, cb) { + bytes += chunk.length; + hash.update(chunk); + cb(bytes > expected.bytes ? Error("音轨超过上限") : null, chunk); + }, + }), + createWriteStream(temp, { mode: 0o600 }) + ); + if (bytes !== expected.bytes || hash.digest("hex") !== expected.sha256) + throw Error("分离音轨校验失败"); + const { rename } = await import("node:fs/promises"); + await rename(temp, destination); + } catch { + throw new ProviderTaskUncertainError( + "分离音轨读取中断，将继续读取同一结果" + ); + } finally { + await rm(temp, { force: true }); + } + } + } + return result; +} diff --git a/playwright.helpers.config.ts b/playwright.helpers.config.ts index 7cb34ef3d..f0928caf6 100644 --- a/playwright.helpers.config.ts +++ b/playwright.helpers.config.ts @@ -3,6 +3,7 @@ import { defineConfig } from "@playwright/test"; export default defineConfig({ testDir: "./tests", testMatch: [ + /routes\/video-canvas-audio-remote\.test\.ts/, /routes\/video-canvas-groups-upscale\.test\.ts/, /routes\/video-canvas-connections\.test\.ts/, /routes\/video-canvas-smart-edit\.test\.ts/, diff --git a/scripts/qa/audio-processing-service.py b/scripts/qa/audio-processing-service.py new file mode 100644 index 000000000..b47be18d8 --- /dev/null +++ b/scripts/qa/audio-processing-service.py @@ -0,0 +1,61 @@ +"""Real private API regression. Run on the service host with synthetic speech/mix WAVs. +Environment AUDIO_QA_ENV names a private service env file. --restart exercises recovery. +""" +import sys,os,json,time,hashlib,urllib.request,urllib.error,subprocess +from pathlib import Path +import numpy as np +import soundfile as sf +env=dict(l.split('=',1) for l in Path(os.environ['AUDIO_QA_ENV']).read_text().splitlines() if '=' in l and not l.startswith('#')) +base=os.environ.get('AUDIO_QA_URL','http://127.0.0.1:18342');token=env['AUDIO_API_TOKEN'] +root=Path(sys.argv[3]);root.mkdir(parents=True,exist_ok=True) +def call(route,body=None,method=None): + data=json.dumps(body).encode() if isinstance(body,dict) else body + r=urllib.request.Request(base+route,data=data,method=method,headers={'Authorization':'Bearer '+token,'Content-Type':'application/json' if isinstance(body,dict) else 'audio/wav'}) + with urllib.request.urlopen(r,timeout=60) as response:return response.read() +def state(route,body=None,method=None):return json.loads(call(route,body,method)) +def rejects(code,route,body=None,method=None): + try:call(route,body,method) + except urllib.error.HTTPError as e:assert e.code==code,(e.code,code);return + raise AssertionError('Expected HTTP '+str(code)) +try:urllib.request.urlopen(base+'/health',timeout=10);raise AssertionError('Missing authentication') +except urllib.error.HTTPError as e:assert e.code==401 +report={} +for operation,file in [('transcribe',Path(sys.argv[1])),('separate',Path(sys.argv[2]))]: + data=file.read_bytes();digest=hashlib.sha256(data).hexdigest() + jid=hashlib.sha256(('real-audio-qa-v1:'+operation+digest).encode()).hexdigest() + body={'id':jid,'tenant':hashlib.sha256(b'audio-qa').hexdigest(),'plan':{'operation':operation,'sha256':digest,'bytes':len(data)}} + job=state('/v1/audio-jobs',body) + rejects(409,'/v1/audio-jobs',{**body,'plan':{**body['plan'],'bytes':len(data)+1}}) + if job['status']=='awaiting-input': + rejects(400,'/v1/audio-jobs/'+jid+'/input',b'wrong','PUT') + job=state('/v1/audio-jobs/'+jid+'/input',data,'PUT') + restarted=False;start=time.monotonic() + for i in range(600): + job=state('/v1/audio-jobs/'+jid) + if operation=='separate' and '--restart' in sys.argv and job['status']=='running' and not restarted: + subprocess.run(['systemctl','restart','audio-processing'],check=True);time.sleep(3);restarted=True;continue + if job['status'] in ('succeeded','failed'):break + time.sleep(1) + assert job['status']=='succeeded',job + assert state('/v1/audio-jobs',body)['attempts']==job['attempts'] + if restarted:assert job['attempts']==2,job + if operation=='transcribe': + subs=job['result']['subtitles'];text=''.join(s['text'] for s in subs) + assert '智能剪辑' in text and '人声分离' in text,text + assert len(subs)>=2 and all(0<=s['start'] scripts/qa/audio-processing-service.py speech.wav mixed.wav output-dir --restart`。仅在专用测试或空闲服务部署窗口使用 `--restart`；使用合成测试语音，脚本会重启该服务验证恢复。凭据、模型、输入和输出不提交 Git。 diff --git a/services/audio-processing/audio-processing-tunnel.service b/services/audio-processing/audio-processing-tunnel.service new file mode 100644 index 000000000..aa7477bf7 --- /dev/null +++ b/services/audio-processing/audio-processing-tunnel.service @@ -0,0 +1,17 @@ +[Unit] +Description=Private audio-processing API tunnel to main application +After=network-online.target video-audio-processing.service +Wants=network-online.target + +[Service] +Type=simple +ExecStart=/usr/bin/ssh -NT -o BatchMode=yes -o ExitOnForwardFailure=yes -o ServerAliveInterval=20 -o ServerAliveCountMax=3 -o StrictHostKeyChecking=yes -o UserKnownHostsFile=/etc/video-render-tunnel/known_hosts -i /etc/video-render-tunnel/id_ed25519 -p 64192 -R 127.0.0.1:18342:127.0.0.1:18342 video-render-tunnel@45.192.99.48 +Restart=always +RestartSec=5 +NoNewPrivileges=true +ProtectSystem=strict +ProtectHome=true +PrivateTmp=true + +[Install] +WantedBy=multi-user.target diff --git a/services/audio-processing/audio-processing.service b/services/audio-processing/audio-processing.service new file mode 100644 index 000000000..556cfd2e7 --- /dev/null +++ b/services/audio-processing/audio-processing.service @@ -0,0 +1,28 @@ +[Unit] +Description=Durable private audio inference (FunASR and HTDemucs) +After=network-online.target +[Service] +User=audio-processing +Group=audio-processing +WorkingDirectory=/opt/audio-processing +EnvironmentFile=/etc/audio-processing.env +Environment=CUDA_VISIBLE_DEVICES= +Environment=OMP_NUM_THREADS=2 +Environment=OPENBLAS_NUM_THREADS=2 +Environment=MKL_NUM_THREADS=2 +ExecStart=/usr/bin/node /opt/audio-processing/current/server.cjs +Restart=on-failure +RestartSec=3 +KillMode=mixed +TimeoutStopSec=10 +MemoryMax=12G +CPUQuota=400% +TasksMax=128 +UMask=0077 +NoNewPrivileges=true +PrivateTmp=true +ProtectSystem=strict +ProtectHome=true +ReadWritePaths=/var/lib/audio-processing +[Install] +WantedBy=multi-user.target diff --git a/services/audio-processing/build.cjs b/services/audio-processing/build.cjs new file mode 100644 index 000000000..7c5a6c29c --- /dev/null +++ b/services/audio-processing/build.cjs @@ -0,0 +1,25 @@ +const fs = require("node:fs"), + path = require("node:path"); +const esbuild = require( + path.resolve( + "node_modules/.pnpm", + fs + .readdirSync("node_modules/.pnpm") + .find((n) => n.startsWith("esbuild@0.25.")), + "node_modules/esbuild" + ) +); +esbuild + .build({ + entryPoints: ["services/audio-processing/server.ts"], + outfile: "services/audio-processing/dist/server.cjs", + bundle: true, + platform: "node", + target: "node22", + format: "cjs", + }) + .then(() => { + for(const name of ["analyze.py","separate.py"]) fs.copyFileSync("services/smart-edit/"+name,"services/audio-processing/dist/"+name); + fs.copyFileSync("services/audio-processing/run.py","services/audio-processing/dist/run.py"); + }) + .catch(() => process.exit(1)); diff --git a/services/audio-processing/job.ts b/services/audio-processing/job.ts new file mode 100644 index 000000000..2646274d3 --- /dev/null +++ b/services/audio-processing/job.ts @@ -0,0 +1,88 @@ +import { spawn, execFile } from "node:child_process"; +import { promisify } from "node:util"; +import { readFile } from "node:fs/promises"; +import path from "node:path"; +import { + audioJobStatusSchema, + type AudioJobInput, +} from "../../lib/video-canvas/smart-edit/audio-contract"; +const python = () => process.env.AUDIO_PYTHON || "python3"; +export async function audioHealth() { + const { stdout } = await promisify(execFile)( + python(), + [path.join(__dirname, "run.py"), "--check"], + { timeout: 30000, maxBuffer: 65536 } + ); + const value = JSON.parse(stdout); + if (value.asr !== true || value.separation !== true) + throw Error("音频模型未准备就绪"); + return { asr: true, separation: true }; +} +export async function audioJob( + plan: AudioJobInput, + _tenant: string, + dir: string, + progress: (stage: string) => Promise +) { + await progress( + plan.operation === "transcribe" + ? "算力服务器正在自动转写" + : "算力服务器正在分离人声与背景声" + ); + await new Promise((resolve, reject) => { + const child = spawn( + python(), + [ + path.join(__dirname, "run.py"), + plan.operation, + path.join(dir, "input.wav"), + dir, + ], + { + stdio: ["ignore", "ignore", "pipe"], + detached: true, + env: { ...process.env, CUDA_VISIBLE_DEVICES: "", OMP_NUM_THREADS: "2" }, + } + ); + let tail = "", + timedOut = false; + const timer = setTimeout(() => { + timedOut = true; + try { + process.kill(-child.pid!, "SIGKILL"); + } catch {} + }, 600000); + child.stderr.on("data", (b) => { + tail = (tail + b.toString()).slice(-2000); + }); + child.once("error", () => { + clearTimeout(timer); + reject(Error("音频处理器无法启动")); + }); + child.once("close", (code) => { + clearTimeout(timer); + if (code === 0) resolve(); + else { + console.error( + JSON.stringify({ + event: "audio-inference-failed", + code, + timedOut, + detail: tail.replace(/https?:\S+/g, "[url]"), + }) + ); + reject( + Error( + timedOut + ? "音频处理超时，请缩短素材后重试" + : "音频模型处理失败，请检查素材或服务日志" + ) + ); + } + }); + }); + const result = JSON.parse( + await readFile(path.join(dir, "result.json"), "utf8") + ); + return audioJobStatusSchema.shape.result.parse(result)!; +} diff --git a/services/audio-processing/models.json b/services/audio-processing/models.json new file mode 100644 index 000000000..1eef8e92e --- /dev/null +++ b/services/audio-processing/models.json @@ -0,0 +1,86 @@ +[ + { + "file": "punc_ct-transformer_zh-cn-common-vocab272727-pytorch/model.pt", + "url": "https://modelscope.cn/models/iic/punc_ct-transformer_zh-cn-common-vocab272727-pytorch/resolve/master/model.pt", + "bytes": 291979892, + "sha256": "a5818bb9d933805a916eebe41eb41648f7f9caad30b4bd59d56f3ca135421916" + }, + { + "file": "punc_ct-transformer_zh-cn-common-vocab272727-pytorch/config.yaml", + "url": "https://modelscope.cn/models/iic/punc_ct-transformer_zh-cn-common-vocab272727-pytorch/resolve/master/config.yaml", + "bytes": 810, + "sha256": "a56ec10925b06fa976ad51af373396be2b13e1eb8dc62a5426b5adebaba7071d" + }, + { + "file": "punc_ct-transformer_zh-cn-common-vocab272727-pytorch/configuration.json", + "url": "https://modelscope.cn/models/iic/punc_ct-transformer_zh-cn-common-vocab272727-pytorch/resolve/master/configuration.json", + "bytes": 373, + "sha256": "ebca5ff883e03585b2eec84ce80573f851658bdc61f2887db7d076afe6facd91" + }, + { + "file": "punc_ct-transformer_zh-cn-common-vocab272727-pytorch/tokens.json", + "url": "https://modelscope.cn/models/iic/punc_ct-transformer_zh-cn-common-vocab272727-pytorch/resolve/master/tokens.json", + "bytes": 4207480, + "sha256": "c960ab87bccea4aa15cf49a59f71973c2c330b46668048cd8da253749ec71ee3" + }, + { + "file": "speech_fsmn_vad_zh-cn-16k-common-pytorch/model.pt", + "url": "https://modelscope.cn/models/iic/speech_fsmn_vad_zh-cn-16k-common-pytorch/resolve/master/model.pt", + "bytes": 1721366, + "sha256": "b3be75be477f0780277f3bae0fe489f48718f585f3a6e45d7dd1fbb1a4255fc5" + }, + { + "file": "speech_fsmn_vad_zh-cn-16k-common-pytorch/config.yaml", + "url": "https://modelscope.cn/models/iic/speech_fsmn_vad_zh-cn-16k-common-pytorch/resolve/master/config.yaml", + "bytes": 1215, + "sha256": "486861ca26ddb79081663b6179cb204c6bfae71c52f04aafc48a9e9d8dde1e93" + }, + { + "file": "speech_fsmn_vad_zh-cn-16k-common-pytorch/configuration.json", + "url": "https://modelscope.cn/models/iic/speech_fsmn_vad_zh-cn-16k-common-pytorch/resolve/master/configuration.json", + "bytes": 365, + "sha256": "7bce8867e37d55c3dd8f672695ced18077a2be199ea529a5d432d5350fc0acba" + }, + { + "file": "speech_fsmn_vad_zh-cn-16k-common-pytorch/am.mvn", + "url": "https://modelscope.cn/models/iic/speech_fsmn_vad_zh-cn-16k-common-pytorch/resolve/master/am.mvn", + "bytes": 8040, + "sha256": "6820fef9687708c4fc3fab2530179c8fcea6262daa25514380056cd8f6eb1754" + }, + { + "file": "speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch/model.pt", + "url": "https://modelscope.cn/models/iic/speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch/resolve/master/model.pt", + "bytes": 880502012, + "sha256": "5bba782a5e9196166233b9ab12ba04cadff9ef9212b4ff6153ed9290ff679025" + }, + { + "file": "speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch/config.yaml", + "url": "https://modelscope.cn/models/iic/speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch/resolve/master/config.yaml", + "bytes": 2509, + "sha256": "bfd424b9f61846aac63591e96fe24d66b35f4643bddcb4b9bb759db8f987ab79" + }, + { + "file": "speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch/configuration.json", + "url": "https://modelscope.cn/models/iic/speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch/resolve/master/configuration.json", + "bytes": 472, + "sha256": "b7d2917954fc547e43ad61486a5248596738cf5909907eade11e71c56f446f34" + }, + { + "file": "speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch/tokens.json", + "url": "https://modelscope.cn/models/iic/speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch/resolve/master/tokens.json", + "bytes": 93676, + "sha256": "2b20c2b12572d682afff84ce1c8d560f67b8b32a4c1f21567411d141ed352127" + }, + { + "file": "speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch/am.mvn", + "url": "https://modelscope.cn/models/iic/speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch/resolve/master/am.mvn", + "bytes": 11203, + "sha256": "29b3c740a2c0cfc6b308126d31d7f265fa2be74f3bb095cd2f143ea970896ae5" + }, + { + "file": "speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch/seg_dict", + "url": "https://modelscope.cn/models/iic/speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch/resolve/master/seg_dict", + "bytes": 8287834, + "sha256": "59a2ef803a3f1648ad03a2e1480db1c1ee0c0d7dc4ef4dbd16cea33944329022" + } +] diff --git a/services/audio-processing/prepare_models.py b/services/audio-processing/prepare_models.py new file mode 100644 index 000000000..09a7f6f77 --- /dev/null +++ b/services/audio-processing/prepare_models.py @@ -0,0 +1,26 @@ +"""Install checksum-pinned official Paraformer/VAD/punctuation artifacts.""" +import hashlib,json,os,urllib.request +from pathlib import Path +from concurrent.futures import ThreadPoolExecutor +root=Path(os.environ.get('AUDIO_MODEL_ROOT','/opt/audio-processing/models')) +manifest=json.loads(Path(__file__).with_name('models.json').read_text()) +def checksum(p): + h=hashlib.sha256() + with p.open('rb') as f: + for b in iter(lambda:f.read(1024*1024),b''):h.update(b) + return h.hexdigest() +def install(item): + p=root/item['file'];p.parent.mkdir(parents=True,exist_ok=True) + if p.exists() and p.stat().st_size==item['bytes'] and checksum(p)==item['sha256']:return + temp=p.with_suffix(p.suffix+'.part');size=0 + try: + with urllib.request.urlopen(item['url'],timeout=60) as r,temp.open('wb') as f: + while b:=r.read(1024*1024): + size+=len(b) + if size>item['bytes']:raise RuntimeError('Model size mismatch: '+item['file']) + f.write(b) + if size!=item['bytes'] or checksum(temp)!=item['sha256']:raise RuntimeError('Model checksum mismatch: '+item['file']) + temp.replace(p);print('verified '+item['file'],flush=True) + finally:temp.unlink(missing_ok=True) +with ThreadPoolExecutor(max_workers=4) as pool:list(pool.map(install,manifest)) +print('ALL_SPEECH_MODELS_VERIFIED',flush=True) diff --git a/services/audio-processing/requirements-lock.txt b/services/audio-processing/requirements-lock.txt new file mode 100644 index 000000000..66c81da9a --- /dev/null +++ b/services/audio-processing/requirements-lock.txt @@ -0,0 +1,93 @@ +# Verified Python 3.12 Linux CPU environment; PyTorch CPU index is required. +aliyun-python-sdk-core==2.16.1 +aliyun-python-sdk-kms==2.16.5 +annotated-doc==0.0.5 +antlr4-python3-runtime==4.9.3 +anyio==4.15.1 +audioread==3.1.0 +certifi==2026.7.22 +cffi==2.1.1 +charset-normalizer==3.5.1 +click==8.5.0 +cloudpickle==3.1.2 +crcmod==1.7 +cryptography==50.0.1 +decorator==5.3.1 +demucs==4.0.1 +dora_search==0.1.13 +einops==0.8.2 +filelock==4.0.0 +fsspec==2026.7.0 +funasr==1.4.15 +h11==0.16.0 +hf-xet==1.6.0 +httpcore==1.0.9 +httpx==0.28.1 +huggingface_hub==1.32.0 +hydra-core==1.3.7 +idna==3.20 +jaconv==0.5.0 +jamo==0.4.1 +jieba==0.42.1 +Jinja2==3.1.6 +jmespath==1.1.0 +joblib==1.6.0 +julius==0.2.8 +kaldiio==2.18.1 +lameenc==1.8.4 +lazy-loader==0.5 +librosa==0.11.0 +llvmlite==0.49.0 +markdown-it-py==4.2.0 +MarkupSafe==3.0.3 +mdurl==0.1.2 +modelscope==1.30.0 +mpmath==1.3.0 +msgpack==1.2.2 +narwhals==2.26.0 +networkx==3.6.1 +numba==0.67.0 +numpy==2.2.6 +omegaconf==2.3.1 +openunmix==1.3.0 +oss2==2.19.1 +packaging==26.3 +platformdirs==4.11.10 +pooch==1.9.0 +protobuf==7.36.2 +pycparser==3.0 +pycryptodome==3.23.0 +Pygments==2.21.0 +pynndescent==0.6.0 +PyYAML==6.0.3 +RapidFuzz==3.14.6 +regex==2026.9.10 +requests==2.34.2 +retrying==1.4.2 +rich==15.0.0 +safetensors==0.8.0 +scikit-learn==1.9.1 +scipy==1.18.1 +sentencepiece==0.2.2 +setuptools==84.0.0 +shellingham==1.5.4 +six==1.17.0 +soundfile==0.13.1 +soxr==1.1.0 +submitit==1.5.4 +sympy==1.13.1 +tensorboardX==2.6.5 +threadpoolctl==3.7.0 +tiktoken==0.14.0 +tokenizers==0.23.2 +torch==2.6.0+cpu +torch-complex==0.4.4 +torchaudio==2.6.0+cpu +tqdm==4.70.1 +transformers==5.17.0 +treetable==0.2.6 +typer==0.27.2 +typing_extensions==4.16.0 +umap-learn==0.5.12 +urllib3==2.8.0 +websockets==17.1 diff --git a/services/audio-processing/requirements.txt b/services/audio-processing/requirements.txt new file mode 100644 index 000000000..dfe7b8b78 --- /dev/null +++ b/services/audio-processing/requirements.txt @@ -0,0 +1,10 @@ +# Install torch/torchaudio from https://download.pytorch.org/whl/cpu first. +torch==2.6.0 +torchaudio==2.6.0 +funasr==1.4.15 +modelscope==1.30.0 +transformers==5.17.0 +demucs==4.0.1 +numpy==2.2.6 +soundfile==0.13.1 +librosa==0.11.0 diff --git a/services/audio-processing/run.py b/services/audio-processing/run.py new file mode 100644 index 000000000..af6a4e41e --- /dev/null +++ b/services/audio-processing/run.py @@ -0,0 +1,37 @@ +"""Offline CPU inference with fixed models; WAV input only, no model code downloads.""" +import os, sys, json, contextlib +from pathlib import Path +for key in ('OMP_NUM_THREADS','OPENBLAS_NUM_THREADS','MKL_NUM_THREADS','NUMBA_NUM_THREADS'): + os.environ[key]='2' +os.environ['HF_HUB_OFFLINE']='1' +os.environ['TRANSFORMERS_OFFLINE']='1' +from analyze import capabilities, analyze +from separate import separate + +def run(operation, source, destination): + import soundfile as sf + import numpy as np + meta=sf.info(source) + if meta.format not in ('WAV','WAVEX') or not 0500: raise ValueError('字幕数量超过 500 条') + else: raise ValueError('未知音频操作') + (root/'result.json').write_text(json.dumps(result,ensure_ascii=False)) +if __name__=='__main__': + if sys.argv[1:]==['--check']: + c=capabilities(); print(json.dumps({'asr':c['asr'],'separation':c['separation']})) + else: + with contextlib.redirect_stdout(sys.stderr):run(*sys.argv[1:4]) diff --git a/services/audio-processing/server.ts b/services/audio-processing/server.ts new file mode 100644 index 000000000..22480f98f --- /dev/null +++ b/services/audio-processing/server.ts @@ -0,0 +1,372 @@ +import http from "node:http"; +import { createReadStream, createWriteStream } from "node:fs"; +import { + mkdir, + readFile, + rename, + readdir, + rm, + stat, + statfs, + open, +} from "node:fs/promises"; +import path from "node:path"; +import { createHash, timingSafeEqual } from "node:crypto"; +import { Transform } from "node:stream"; +import { pipeline } from "node:stream/promises"; +import { z } from "zod"; +import { + audioJobInputSchema, + type AudioJobInput, +} from "../../lib/video-canvas/smart-edit/audio-contract"; +import { audioJob, audioHealth } from "./job"; + +const root = path.resolve( + process.env.AUDIO_DATA_DIR || "/var/lib/audio-processing" +); +const token = process.env.AUDIO_API_TOKEN || ""; +if (token.length < 32) + throw Error("AUDIO_API_TOKEN must have at least 32 characters"); +const jobs = new Map(); +type Job = { + id: string; + digest: string; + tenant: string; + plan: AudioJobInput; + status: "awaiting-input" | "queued" | "running" | "succeeded" | "failed"; + stage: string; + createdAt: string; + updatedAt: string; + attempts: number; + error?: string; + result?: Record; +}; +const inputSchema = z.object({ + id: z.string().regex(/^[a-f0-9]{64}$/), + tenant: z.string().regex(/^[a-f0-9]{64}$/), + plan: audioJobInputSchema, +}); +let busy = false, + stopping = false, + accepting = false; +const uploading = new Set(); +let capabilities = { asr: false, separation: false }; +const file = (id: string) => path.join(root, id + ".json"); +async function persist(job: Job) { + job.updatedAt = new Date().toISOString(); + const temp = file(job.id) + ".tmp"; + const handle = await open(temp, "w", 0o600); + try { + await handle.writeFile(JSON.stringify(job)); + await handle.sync(); + } finally { + await handle.close(); + } + await rename(temp, file(job.id)); + const directory = await open(root, "r"); + try { + await directory.sync(); + } finally { + await directory.close(); + } + jobs.set(job.id, job); +} +function publicJob(job: Job) { + return { + id: job.id, + status: job.status, + stage: job.stage, + attempts: job.attempts, + createdAt: job.createdAt, + updatedAt: job.updatedAt, + error: job.error, + result: job.result, + }; +} +async function prune() { + for (const job of jobs.values()) + if ( + ["awaiting-input", "succeeded", "failed"].includes(job.status) && + !uploading.has(job.id) && + Date.now() - Date.parse(job.updatedAt) > 7 * 86400000 + ) { + await rm(path.join(root, job.id), { recursive: true, force: true }); + await rm(file(job.id), { force: true }); + jobs.delete(job.id); + } +} +async function work() { + if (busy || stopping) return; + const job = [...jobs.values()].find((j) => j.status === "queued"); + if (!job) return; + busy = true; + const dir = path.join(root, job.id); + try { + await prune(); + const disk = await statfs(root); + if (disk.bavail * disk.bsize < 10 * 1024 ** 3) + throw Error("音频磁盘剩余空间不足 10 GiB"); + job.status = "running"; + job.attempts++; + job.stage = "准备素材"; + await persist(job); + await mkdir(dir, { recursive: true, mode: 0o700 }); + const result = await audioJob(job.plan, job.tenant, dir, async (stage) => { + job.stage = stage; + await persist(job); + }); + job.result = result; + job.status = "succeeded"; + job.stage = "音频完成"; + await persist(job); + // Keep only the deliverables. Intermediate files must not accumulate per job. + for (const name of await readdir(dir)) + if (!["vocals.wav", "background.wav"].includes(name)) + await rm(path.join(dir, name), { recursive: true, force: true }).catch( + () => { + console.warn( + JSON.stringify({ event: "audio-cleanup-deferred", id: job.id }) + ); + } + ); + } catch (error) { + if (stopping) return; // Preserve running journal; restart will resume this same job. + job.status = "failed"; + job.stage = "音频失败"; + // Do not expose ffmpeg arguments, private cache paths or signed source URLs. + job.error = + error instanceof Error && + !/Command failed|https?:|\/var\/|\/tmp\//.test(error.message) + ? error.message.slice(0, 300) + : "素材处理失败，请检查格式、参数或服务日志"; + await rm(dir, { recursive: true, force: true }).catch(() => {}); + await persist(job); + console.error( + JSON.stringify({ event: "audio-failed", id: job.id, error: job.error }) + ); + } finally { + busy = false; + } + setImmediate( + () => + void work().catch((e) => { + console.error("Audio queue persistence failure"); + process.exit(1); + }) + ); +} +function json(res: http.ServerResponse, status: number, value: unknown) { + res.writeHead(status, { + "content-type": "application/json; charset=utf-8", + "cache-control": "no-store", + }); + res.end(JSON.stringify(value)); +} +async function body(req: http.IncomingMessage) { + let size = 0; + const chunks: Buffer[] = []; + for await (const chunk of req) { + size += chunk.length; + if (size > 1024 * 1024) throw Error("请求超过 1 MiB"); + chunks.push(chunk); + } + return JSON.parse(Buffer.concat(chunks).toString("utf8")); +} +function authenticated(req: http.IncomingMessage) { + const expected = Buffer.from("Bearer " + token), + actual = Buffer.from(req.headers.authorization || ""); + return expected.length === actual.length && timingSafeEqual(expected, actual); +} +async function main() { + capabilities = await audioHealth(); + await mkdir(root, { recursive: true, mode: 0o700 }); + // This service runs as a single systemd process. A restart requeues local work at most twice. + for (const name of await readdir(root)) + if (/^[a-f0-9]{64}\.json$/.test(name)) { + const job: Job = JSON.parse( + await readFile(path.join(root, name), "utf8") + ); + if (job.id + ".json" !== name) throw Error("Invalid audio record"); + if (job.status === "running") { + job.status = job.attempts < 3 ? "queued" : "failed"; + job.stage = + job.status === "queued" + ? "服务重启，等待恢复音频" + : "多次中断，已停止"; + if (job.status === "failed") job.error = job.stage; + await persist(job); + } + jobs.set(job.id, job); + } + await prune(); + const server = http.createServer(async (req, res) => { + if (!authenticated(req)) return json(res, 401, { error: "Unauthorized" }); + const url = new URL(req.url || "/", "http://localhost"); + try { + if (req.method === "GET" && url.pathname === "/health") + return json(res, 200, { + status: "ok", + version: 1, + active: busy ? 1 : 0, + queued: [...jobs.values()].filter((j) => j.status === "queued") + .length, + capabilities, + engine: "FunASR-Paraformer / HTDemucs", + device: "cpu", + }); + if (req.method === "POST" && url.pathname === "/v1/audio-jobs") { + const input = inputSchema.parse(await body(req)); + // Serialize admission across the async filesystem write to preserve idempotency. + if (accepting) return json(res, 503, { error: "请重试提交" }); + accepting = true; + try { + const digest = createHash("sha256") + .update(JSON.stringify({ tenant: input.tenant, plan: input.plan })) + .digest("hex"); + const existing = jobs.get(input.id); + if (existing) + return json( + res, + existing.digest === digest ? 200 : 409, + existing.digest === digest + ? publicJob(existing) + : { error: "任务 ID 已用于不同参数" } + ); + if ( + jobs.size >= 200 || + [...jobs.values()].filter((j) => + ["awaiting-input", "queued", "running"].includes(j.status) + ).length >= 30 + ) + return json(res, 429, { error: "音频队列已满，请稍后重试" }); + const job: Job = { + ...input, + digest, + status: "awaiting-input", + stage: "排队中", + attempts: 0, + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + }; + await persist(job); + json(res, 202, publicJob(job)); + void work().catch(() => { + console.error("Audio queue failed"); + process.exit(1); + }); + return; + } finally { + accepting = false; + } + } + const match = url.pathname.match( + /^\/v1\/audio-jobs\/([a-f0-9]{64})(?:\/(input|vocals|background))?$/ + ); + if (req.method === "PUT" && match?.[2] === "input") { + const job = jobs.get(match[1]); + if (!job) return json(res, 404, { error: "任务不存在" }); + if (job.status !== "awaiting-input") + return json(res, 200, publicJob(job)); + if (uploading.has(job.id)) + return json(res, 409, { error: "同一音频正在上传" }); + uploading.add(job.id); + const dir = path.join(root, job.id), + temp = path.join(dir, "input.part"); + try { + await mkdir(dir, { recursive: true, mode: 0o700 }); + let size = 0; + const hash = createHash("sha256"); + const limit = new Transform({ + transform(chunk, _encoding, cb) { + size += chunk.length; + hash.update(chunk); + cb( + size > job.plan.bytes ? Error("音频超过声明大小") : null, + chunk + ); + }, + }); + await pipeline(req, limit, createWriteStream(temp, { mode: 0o600 })); + if (size !== job.plan.bytes || hash.digest("hex") !== job.plan.sha256) + throw Error("音频大小或校验和不匹配"); + await rename(temp, path.join(dir, "input.wav")); + job.status = "queued"; + job.stage = "等待音频处理"; + await persist(job); + json(res, 202, publicJob(job)); + void work().catch(() => process.exit(1)); + } finally { + uploading.delete(job.id); + await rm(temp, { force: true }); + } + return; + } + if (req.method === "GET" && match) { + const job = jobs.get(match[1]); + if (!job) return json(res, 404, { error: "任务不存在或已过期" }); + if (!match[2]) return json(res, 200, publicJob(job)); + if (job.status !== "succeeded") + return json(res, 409, { error: "音轨尚未就绪" }); + if ( + !["vocals", "background"].includes(match[2]) || + job.plan.operation !== "separate" + ) + return json(res, 404, { error: "结果不存在" }); + const output = path.join(root, job.id, match[2] + ".wav"); + const info = await stat(output); + res.writeHead(200, { + "content-type": "audio/wav", + "content-length": info.size, + "cache-control": "no-store", + }); + await pipeline(createReadStream(output), res); + return; + } + json(res, 404, { error: "Not found" }); + } catch (error) { + if (res.headersSent) { + res.destroy(); + return; + } + json(res, 400, { + error: + error instanceof z.ZodError + ? "音频参数无效" + : error instanceof Error && !/https?:|\/var\//.test(error.message) + ? error.message.slice(0, 300) + : "请求处理失败", + }); + } + }); + server.requestTimeout = 180000; + server.headersTimeout = 10000; + server.listen( + Number(process.env.AUDIO_PORT || 18342), + process.env.AUDIO_HOST || "127.0.0.1" + ); + const fatal = () => { + console.error("Audio queue persistence failure"); + process.exit(1); + }; + const timer = setInterval(() => { + void work().catch(fatal); + }, 3000); + const cleanupTimer = setInterval(() => { + if (!busy && !accepting) void prune().catch(fatal); + }, 3600000); + cleanupTimer.unref(); + for (const signal of ["SIGTERM", "SIGINT"]) + process.on(signal, () => { + stopping = true; + clearInterval(timer); + clearInterval(cleanupTimer); + server.close(); + // systemd KillMode=mixed also terminates FFmpeg; persisted running jobs resume on restart. + setTimeout(() => process.exit(0), 1000).unref(); + }); + console.log("Audio processing service ready"); +} +void main().catch(() => { + console.error("Audio processing service startup failed"); + process.exitCode = 1; +}); diff --git a/tests/routes/video-canvas-audio-remote.test.ts b/tests/routes/video-canvas-audio-remote.test.ts new file mode 100644 index 000000000..8a6e55be4 --- /dev/null +++ b/tests/routes/video-canvas-audio-remote.test.ts @@ -0,0 +1,130 @@ +import { test, expect } from "@playwright/test"; +import { mkdtemp, writeFile, readFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { createHash } from "node:crypto"; +import { + remoteAudio, + remoteAudioCapabilities, +} from "@/lib/video-canvas/smart-edit/remote-audio"; +import { ProviderTaskUncertainError } from "@/lib/video-canvas/video-task-lifecycle"; +test.describe.configure({ mode: "serial" }); +const originalFetch = global.fetch, + oldUrl = process.env.SMART_EDIT_AUDIO_URL, + oldToken = process.env.SMART_EDIT_AUDIO_TOKEN; +test.beforeEach(() => { + process.env.SMART_EDIT_AUDIO_URL = "http://127.0.0.1:19342"; + process.env.SMART_EDIT_AUDIO_TOKEN = "test-".repeat(8); +}); +test.afterEach(() => { + global.fetch = originalFetch; + if (oldUrl === undefined) delete process.env.SMART_EDIT_AUDIO_URL; + else process.env.SMART_EDIT_AUDIO_URL = oldUrl; + if (oldToken === undefined) delete process.env.SMART_EDIT_AUDIO_TOKEN; + else process.env.SMART_EDIT_AUDIO_TOKEN = oldToken; +}); +test("audio uploads once, resumes same identity, and isolates tenant/task/content", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "audio-identity-")), + file = path.join(dir, "input.wav"); + await writeFile(file, "fixture bytes"); + const ids: string[] = [], + jobs = new Set(); + let uploads = 0; + global.fetch = (async (_url: any, init: any) => { + if (init.method === "POST") { + const input = JSON.parse(init.body); + ids.push(input.id); + return Response.json({ + id: input.id, + status: jobs.has(input.id) ? "succeeded" : "awaiting-input", + stage: "ready", + ...(jobs.has(input.id) + ? { + result: { + durationSeconds: 1, + subtitles: [{ start: 0, end: 800000, text: "测试" }], + }, + } + : {}), + }); + } + const id = String(_url).split("/").at(-2)!; + let data = ""; + for await (const c of init.body) data += c; + expect(data).toBe("fixture bytes"); + uploads++; + jobs.add(id); + return Response.json({ + id, + status: "succeeded", + stage: "done", + result: { + durationSeconds: 1, + subtitles: [{ start: 0, end: 800000, text: "测试" }], + }, + }); + }) as any; + try { + await remoteAudio("transcribe", file, "owner", "task", dir); + await remoteAudio("transcribe", file, "owner", "task", dir); + expect(uploads).toBe(1); + expect(ids[0]).toBe(ids[1]); + await remoteAudio("transcribe", file, "other", "task", dir); + await remoteAudio("transcribe", file, "owner", "other-task", dir); + expect(new Set(ids).size).toBe(3); + } finally { + await rm(dir, { recursive: true, force: true }); + } +}); +test("unavailable remote audio remains recoverable and does not advertise local fallback", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "audio-offline-")), + file = path.join(dir, "input.wav"); + await writeFile(file, "audio"); + global.fetch = (async () => { + throw Error("offline"); + }) as any; + try { + expect(await remoteAudioCapabilities()).toEqual({ + asr: false, + separation: false, + }); + await expect( + remoteAudio("transcribe", file, "u", "t", dir) + ).rejects.toBeInstanceOf(ProviderTaskUncertainError); + } finally { + await rm(dir, { recursive: true, force: true }); + } +}); +test("corrupt separated download never replaces saved stem", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "audio-integrity-")), + file = path.join(dir, "input.wav"); + await writeFile(file, "audio"); + await writeFile(path.join(dir, "vocals.wav"), "previous"); + const bytes = Buffer.from("expected"), + sha256 = createHash("sha256").update(bytes).digest("hex"); + global.fetch = (async (_url: any, init: any) => + init.method === "POST" + ? Response.json({ + id: JSON.parse(init.body).id, + status: "succeeded", + stage: "done", + result: { + durationSeconds: 1, + stems: { + vocals: { bytes: bytes.length, sha256 }, + background: { bytes: bytes.length, sha256 }, + }, + }, + }) + : new Response("corrupt!")) as any; + try { + await expect( + remoteAudio("separate", file, "u", "t", dir) + ).rejects.toBeInstanceOf(ProviderTaskUncertainError); + expect(await readFile(path.join(dir, "vocals.wav"), "utf8")).toBe( + "previous" + ); + } finally { + await rm(dir, { recursive: true, force: true }); + } +});