Add durable remote audio inference and verified offline speech models

This commit is contained in:
134 2026-09-18 14:12:28 +08:00
parent 229f4d72cb
commit 4f9453b8b2
22 changed files with 1458 additions and 49 deletions

3
.gitignore vendored
View File

@ -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__/

View File

@ -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 }
);
}
}

View File

@ -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。

View File

@ -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<typeof audioJobInputSchema>;
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(),
});

View File

@ -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<string>,
track: "video" | "audio" = "video",
keep: "both" | "vocals" | "background" = "both",
requestKey: string = userId
): Promise<EditPlan> {
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({

View File

@ -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] };

View File

@ -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<string, unknown>,
dir: string,
requestKey: string = dir,
progress?: (stage: string) => Promise<void>
): Promise<LocalAnalysis> {
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");

View File

@ -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<void> = 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;
}

View File

@ -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/,

View File

@ -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']<s['end']<=job['result']['durationSeconds']*1e6+100000 for s in subs)
report['subtitles']=subs
else:
original,sr=sf.read(file,dtype='float32',always_2d=True)
stems=[]
for stem in ('vocals','background'):
data=call('/v1/audio-jobs/'+jid+'/'+stem);meta=job['result']['stems'][stem]
assert len(data)==meta['bytes'] and hashlib.sha256(data).hexdigest()==meta['sha256']
p=root/(stem+'.wav');p.write_bytes(data);wave,rate=sf.read(p,dtype='float32',always_2d=True)
assert rate==sr and wave.shape==original.shape
stems.append(wave)
error=float(np.max(np.abs(stems[0]+stems[1]-original)));assert error<1e-6,error
assert not np.allclose(stems[0],original)
report['reconstructionMaxError']=error
report[operation]={'jobId':jid,'attempts':job['attempts'],'wallSeconds':round(time.monotonic()-start,2),'durationSeconds':job['result']['durationSeconds']}
(root/'report.json').write_text(json.dumps(report,ensure_ascii=False,indent=2))
print('REAL_AUDIO_API_PASS',json.dumps(report,ensure_ascii=False))

View File

@ -0,0 +1,36 @@
# 独立自动转写 / 人声分离服务
算力服务器 CPU 单队列运行 FunASR Paraformer(含 FSMN VAD / CT-Transformer 标点)与 HTDemucs;48 仅进行有界素材准备、状态管理、对象存储归档。CPU 推理不占用千问或视频高清的显存。systemd 限制 4 核 / 12GiB;Python 推理线程 2;一次一个作业。
## 安装
Python 3.12 独立环境,先从官方 `https://download.pytorch.org/whl/cpu` 安装 torch/torchaudio 2.6.0 CPU wheel,再安装 `requirements.txt`。运行 `prepare_models.py` 下载 `models.json` 中固定 SHA-256 的官方语音模型运行文件;模型说明文档可变化,不参与运行文件校验。运行 `services/smart-edit/prepare_separation.py` 安装固定 SHA-256 的官方 HTDemucs。权重不进入 Git。
`node services/audio-processing/build.cjs` 产出 bundle 和 Python 脚本,安装到 `/opt/audio-processing/current/`。设置仅 root 可读的 `/etc/audio-processing.env`:
- `AUDIO_API_TOKEN`:独立随机密钥,至少 32 字符。
- `AUDIO_PORT=18342`,`AUDIO_HOST=127.0.0.1`。
- `AUDIO_PYTHON=/opt/audio-processing/venv/bin/python`。
- `AUDIO_DATA_DIR=/var/lib/audio-processing`,由专用 `audio-processing` 用户持有。
- `SMART_EDIT_ASR_MODEL`、`SMART_EDIT_VAD_MODEL`、`SMART_EDIT_PUNC_MODEL`:三个完整本地模型目录。
- `SMART_EDIT_SEPARATION_MODEL_DIR`:HTDemucs 目录。
- `HF_HOME`、`MODELSCOPE_CACHE`、`XDG_CACHE_HOME`:服务数据目录下的可写缓存。
部署 `audio-processing.service` 和私有反向 SSH 隧道,沿用专用受限隧道账号,仅增加 localhost 18342 监听权限。主项目配置 `SMART_EDIT_AUDIO_URL=http://127.0.0.1:18342` 与匹配的 `SMART_EDIT_AUDIO_TOKEN`。模型或服务不可用时能力接口返回 false,不自动切回主机重推理。
## 协议与恢复
全部接口需要 Bearer 认证,禁止面向公网暴露。
1. `POST /v1/audio-jobs`:`{id,tenant,plan:{operation,sha256,bytes}}`,operation 为 transcribe / separate;id 与 tenant 是 64 位十六进制。
2. 新任务为 awaiting-input;`PUT /v1/audio-jobs/:id/input` 上传原始 WAV。流式校验大小和 SHA-256,通过后才入队。
3. `GET /v1/audio-jobs/:id` 返回真实状态、字幕时间戳或两音轨校验值。分离结果通过 `/vocals`、`/background` 下载。
4. `GET /health` 认证后返回能力与队列状态;启动时检查模型和依赖是否存在。
客户端编号取决于用户隔离标识、持久主任务编号、操作和输入内容;重试复用同一任务,参数冲突 409。状态先 fsync 再确认成功;先持久化结果再清理输入,避免重启丢失结果。正在执行的任务重启后继续,最多三次。网络中断和队列繁忙保存为可恢复状态,失败不会冒充生成结果。
仅允许 180 秒以内 PCM WAV、1/2 声道、固定采样率;分离输入必须 44.1kHz 双声道。单文件 ≤70MiB;30 个待处理/待上传作业、200 个记录;结果 7 天清理;剩余磁盘不足10GiB停止接新处理。每作业10分钟处理超时。请求不能指定模型、命令或本地路径;推理关闭联网更新,模型代码禁止 trust_remote_code。
模型来源:[FunASR](https://github.com/modelscope/FunASR)、[官方 Paraformer](https://modelscope.cn/models/iic/speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch)、[Demucs](https://github.com/facebookresearch/demucs)。保留安装包许可证和模型说明。
真实 API 验证脚本:`AUDIO_QA_ENV=/etc/audio-processing.env <venv-python> scripts/qa/audio-processing-service.py speech.wav mixed.wav output-dir --restart`。仅在专用测试或空闲服务部署窗口使用 `--restart`;使用合成测试语音,脚本会重启该服务验证恢复。凭据、模型、输入和输出不提交 Git。

View File

@ -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

View File

@ -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

View File

@ -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));

View File

@ -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<void>
) {
await progress(
plan.operation === "transcribe"
? "算力服务器正在自动转写"
: "算力服务器正在分离人声与背景声"
);
await new Promise<void>((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)!;
}

View File

@ -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"
}
]

View File

@ -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)

View File

@ -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

View File

@ -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

View File

@ -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 0<meta.duration<=180 or meta.channels not in (1,2) or meta.samplerate not in (16000,22050,44100,48000):
raise ValueError('仅接受 180 秒内、1/2 声道的 PCM WAV 音频')
if meta.subtype not in ('PCM_16','PCM_24','PCM_32','FLOAT'):
raise ValueError('音频编码不支持')
root=Path(destination)
result={'durationSeconds':meta.duration}
if operation=='separate':
separate(source,destination)
import hashlib
result['stems']={}
for stem in ['vocals','background']:
f=root/(stem+'.wav')
result['stems'][stem]={'bytes':f.stat().st_size,'sha256':hashlib.sha256(f.read_bytes()).hexdigest()}
elif operation=='transcribe':
result['subtitles']=analyze({'audio':source,'transcribe':True})['subtitles']
if len(result['subtitles'])>500: 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])

View File

@ -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<string, Job>();
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<string, any>;
};
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<string>();
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;
});

View File

@ -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<string>();
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 });
}
});