|
|
@@ -0,0 +1,190 @@
|
|
|
+import { createJob, updateJob } from "@/lib/job-store";
|
|
|
+import { loadConfig } from "@/lib/config";
|
|
|
+import { resolveOutputDir } from "@pipeline/shared/node";
|
|
|
+import { spawn } from "node:child_process";
|
|
|
+import { writeFileSync, mkdirSync, statSync } from "node:fs";
|
|
|
+import { join, resolve } from "node:path";
|
|
|
+import { tmpdir } from "node:os";
|
|
|
+
|
|
|
+export interface RenderJobParams {
|
|
|
+ template: string;
|
|
|
+ platforms: string[];
|
|
|
+ ttsProvider: string;
|
|
|
+ source?: string;
|
|
|
+ sourceArgs?: Record<string, string>;
|
|
|
+ /** Raw input text (JSON / markdown / plain). Required when `source` is absent. */
|
|
|
+ text?: string;
|
|
|
+ voiceId?: string;
|
|
|
+ noTts?: boolean;
|
|
|
+ /** Skip OSS upload + Feishu notification. Default: publish when configured. */
|
|
|
+ noPublish?: boolean;
|
|
|
+}
|
|
|
+
|
|
|
+/**
|
|
|
+ * Spawn the compiled CLI to render a video, tracking it in the job store so it
|
|
|
+ * shows up in the WebUI / jobs list exactly like a WebUI/HTTP-triggered render.
|
|
|
+ * Shared by `POST /api/render` and the in-process scheduler.
|
|
|
+ *
|
|
|
+ * Returns immediately with the `jobId`; `done` resolves when the child process
|
|
|
+ * exits (success or failure) so callers that need ordering (e.g. the scheduler's
|
|
|
+ * single-flight guard) can await it.
|
|
|
+ */
|
|
|
+export function startRenderJob(params: RenderJobParams): {
|
|
|
+ jobId: string;
|
|
|
+ done: Promise<void>;
|
|
|
+} {
|
|
|
+ const {
|
|
|
+ template, platforms, ttsProvider, source, sourceArgs,
|
|
|
+ text, voiceId, noTts, noPublish,
|
|
|
+ } = params;
|
|
|
+
|
|
|
+ const job = createJob({ text: text ?? "", template, platforms, ttsProvider, voiceId });
|
|
|
+
|
|
|
+ const jobDir = join(tmpdir(), "pipeline-jobs", job.id);
|
|
|
+ mkdirSync(jobDir, { recursive: true });
|
|
|
+
|
|
|
+ // Same unified output directory the CLI resolves (OUTPUT_DIR > config > ./output).
|
|
|
+ const config = loadConfig();
|
|
|
+ const outputDir = resolveOutputDir(config?.output?.dir);
|
|
|
+
|
|
|
+ const cliArgs = [
|
|
|
+ "render",
|
|
|
+ "-t", template,
|
|
|
+ "-p", platforms.join(","),
|
|
|
+ "--tts-provider", ttsProvider,
|
|
|
+ "--output", outputDir,
|
|
|
+ ];
|
|
|
+
|
|
|
+ if (source) {
|
|
|
+ cliArgs.push("--source", source);
|
|
|
+ if (sourceArgs?.owner) cliArgs.push("--source-owner", sourceArgs.owner);
|
|
|
+ if (sourceArgs?.repo) cliArgs.push("--source-repo", sourceArgs.repo);
|
|
|
+ } else {
|
|
|
+ const inputFile = join(jobDir, "input.json");
|
|
|
+ writeFileSync(inputFile, text!, "utf-8");
|
|
|
+ cliArgs.splice(1, 0, inputFile);
|
|
|
+ }
|
|
|
+ if (voiceId) cliArgs.push("--voice", voiceId);
|
|
|
+ if (noTts) cliArgs.push("--no-tts");
|
|
|
+ if (noPublish) cliArgs.push("--no-publish");
|
|
|
+
|
|
|
+ const cliPath = resolve(process.cwd(), "../cli/dist/index.js");
|
|
|
+ const projectRoot = resolve(process.cwd(), "..");
|
|
|
+
|
|
|
+ updateJob(job.id, { status: "running" });
|
|
|
+
|
|
|
+ const child = spawn("node", [cliPath, ...cliArgs], {
|
|
|
+ cwd: projectRoot,
|
|
|
+ stdio: ["ignore", "pipe", "pipe"],
|
|
|
+ env: { ...process.env },
|
|
|
+ });
|
|
|
+
|
|
|
+ const jobIdShort = job.id.slice(0, 8);
|
|
|
+ const linePrefix = `[job ${jobIdShort}] `;
|
|
|
+ const fwdOut = makePrefixer(linePrefix, (s) => process.stdout.write(s));
|
|
|
+ const fwdErr = makePrefixer(linePrefix, (s) => process.stderr.write(s));
|
|
|
+
|
|
|
+ let stdout = "";
|
|
|
+ let stderr = "";
|
|
|
+ let logBuf = "";
|
|
|
+
|
|
|
+ const done = new Promise<void>((resolveDone) => {
|
|
|
+ // Capture for parsing/UI AND mirror to the container's stdout/stderr so
|
|
|
+ // `docker compose logs` shows every stage / TTS retry / publish event live.
|
|
|
+ child.stdout.on("data", (data: Buffer) => {
|
|
|
+ const s = data.toString();
|
|
|
+ stdout += s;
|
|
|
+ logBuf += s;
|
|
|
+ fwdOut.push(s);
|
|
|
+ });
|
|
|
+ child.stderr.on("data", (data: Buffer) => {
|
|
|
+ const s = data.toString();
|
|
|
+ stderr += s;
|
|
|
+ logBuf += s;
|
|
|
+ fwdErr.push(s);
|
|
|
+ });
|
|
|
+
|
|
|
+ child.on("close", (code: number) => {
|
|
|
+ fwdOut.flush();
|
|
|
+ fwdErr.flush();
|
|
|
+ const log = capLog(logBuf);
|
|
|
+ if (code === 0) {
|
|
|
+ const mp4Matches = [...stdout.matchAll(/-> (.+\.mp4)/g)];
|
|
|
+ const filePaths = mp4Matches.map((m) => m[1].trim());
|
|
|
+
|
|
|
+ const outputFiles = filePaths.map((filePath) => {
|
|
|
+ let fileSizeBytes = 0;
|
|
|
+ try { fileSizeBytes = statSync(filePath).size; } catch {}
|
|
|
+ const isPortrait = filePath.includes("douyin") || filePath.includes("9x16");
|
|
|
+ return {
|
|
|
+ filePath,
|
|
|
+ fileSizeBytes,
|
|
|
+ width: isPortrait ? 1080 : 1920,
|
|
|
+ height: isPortrait ? 1920 : 1080,
|
|
|
+ durationSeconds: 0,
|
|
|
+ };
|
|
|
+ });
|
|
|
+
|
|
|
+ const ossUrls = [...stdout.matchAll(/^[ \t]*oss: (\S+)/gm)].map((m) => m[1].trim());
|
|
|
+ const publishWarn = stdout.match(/^Publish warning: (.+)$/m);
|
|
|
+ const publishError = publishWarn ? publishWarn[1].trim() : undefined;
|
|
|
+
|
|
|
+ updateJob(job.id, {
|
|
|
+ status: "completed",
|
|
|
+ outputFiles: outputFiles.length > 0 ? outputFiles : undefined,
|
|
|
+ ossUrls: ossUrls.length > 0 ? ossUrls : undefined,
|
|
|
+ publishError,
|
|
|
+ log,
|
|
|
+ });
|
|
|
+ } else {
|
|
|
+ updateJob(job.id, {
|
|
|
+ status: "failed",
|
|
|
+ error: stderr.trim() || `Process exited with code ${code}`,
|
|
|
+ log,
|
|
|
+ });
|
|
|
+ }
|
|
|
+ resolveDone();
|
|
|
+ });
|
|
|
+
|
|
|
+ child.on("error", (err: Error) => {
|
|
|
+ fwdOut.flush();
|
|
|
+ fwdErr.flush();
|
|
|
+ updateJob(job.id, {
|
|
|
+ status: "failed",
|
|
|
+ error: err.message,
|
|
|
+ log: capLog(logBuf),
|
|
|
+ });
|
|
|
+ resolveDone();
|
|
|
+ });
|
|
|
+ });
|
|
|
+
|
|
|
+ return { jobId: job.id, done };
|
|
|
+}
|
|
|
+
|
|
|
+// Keep the per-job log bounded so jobs.json doesn't grow without limit.
|
|
|
+const LOG_CAP = 200_000;
|
|
|
+function capLog(buf: string): string {
|
|
|
+ if (buf.length <= LOG_CAP) return buf;
|
|
|
+ return `...(truncated ${buf.length - LOG_CAP} bytes from the top)\n` + buf.slice(-LOG_CAP);
|
|
|
+}
|
|
|
+
|
|
|
+// Line-buffered forwarder: writes each complete line (prefixed with the job id)
|
|
|
+// to the container's stdout/stderr, holding back a trailing partial line until
|
|
|
+// the next chunk or flush. Lets `docker compose logs` attribute lines to jobs.
|
|
|
+function makePrefixer(prefix: string, write: (s: string) => void) {
|
|
|
+ let pending = "";
|
|
|
+ return {
|
|
|
+ push(chunk: string) {
|
|
|
+ pending += chunk;
|
|
|
+ const lines = pending.split("\n");
|
|
|
+ pending = lines.pop() ?? "";
|
|
|
+ for (const line of lines) write(prefix + line + "\n");
|
|
|
+ },
|
|
|
+ flush() {
|
|
|
+ if (pending.length > 0) {
|
|
|
+ write(prefix + pending + "\n");
|
|
|
+ pending = "";
|
|
|
+ }
|
|
|
+ },
|
|
|
+ };
|
|
|
+}
|
lkatzey