import { randomUUID } from "node:crypto"; import { mkdir } from "node:fs/promises"; import { join } from "node:path"; import type { PipelineInput, PipelineJob, ExportFile, } from "@pipeline/shared"; import { PLATFORM_PRESETS } from "@pipeline/shared"; import { createLogger } from "@pipeline/shared/node"; import { parseText } from "./stages/parse.js"; import { generateTTS } from "./stages/tts.js"; import { resolveAssets } from "./stages/assets.js"; import { composeProject } from "./stages/compose.js"; import { renderVideo } from "./stages/render.js"; import { exportVideo, type PublishTargetConfig } from "./stages/export.js"; import { cleanupExpiredOutput } from "./cleanup.js"; import { runPublish, notifyGenerationFailure, type PublishConfig, } from "./publish/index.js"; export interface PipelineConfig { branding: { channelName: string; }; llm: { baseURL?: string; apiKey?: string; model: string; }; tts: { provider: string; voiceId?: string; model?: string; format?: "mp3" | "wav" | "pcm"; speed?: number; }; alignment?: { provider: "whisper" | "native"; whisperModel?: string; language?: string; }; output: { dir: string; /** Auto-clean cached outputs older than this many days (0 disables). */ retentionDays?: number; }; /** OSS upload + Feishu notification, resolved by resolvePublishConfig. */ publish?: PublishConfig; /** * Per-platform × per-template publish metadata for the sidecar manifest * (partition tid / category, extra tags). Lookup: publishMeta[platform][template]. * Missing combos simply omit those keys from the manifest. */ publishMeta?: Record>; /** Skip the publish stage entirely (e.g. local debugging). */ skipPublish?: boolean; assets: { root: string; inputDir: string; }; templates: { entryPoint: string; }; skipTts?: boolean; skipLlm?: boolean; } export interface PipelineCallbacks { onStageStart?: (stage: string) => void; onStageComplete?: (stage: string) => void; onError?: (stage: string, error: Error) => void; } /** * Compose a suggested video title for Feishu notifications: the parsed title, * joined with the cover scene's one-line trendSummary (github-trending) via a * fullwidth separator when present; otherwise just the title. Returns undefined * when no title is known (e.g. parse failed) so the line can be omitted. */ function buildTitleSuggestion( title: string | undefined, trendSummary: string | undefined ): string | undefined { if (!title) return undefined; return trendSummary ? `${title}|${trendSummary}` : title; } export async function runPipeline( input: PipelineInput, config: PipelineConfig, callbacks?: PipelineCallbacks ): Promise { const jobId = randomUUID(); const log = createLogger("pipeline"); const t0 = Date.now(); const workDir = join(config.output.dir, "tmp", jobId); await mkdir(workDir, { recursive: true }); // Opportunistic cache sweep: prune outputs older than retentionDays. // Runs on every job (service jobs are low-frequency); never throws. if (config.output.retentionDays && config.output.retentionDays > 0) { const cleaned = await cleanupExpiredOutput(config.output.dir, config.output.retentionDays); log.debug( `cleanup scanned=${cleaned.scanned} removed=${cleaned.removed} freed=${Math.round(cleaned.bytesFreed / 1024 / 1024)}MB` ); } const platforms = input.platforms; const job: PipelineJob = { id: jobId, input, status: "running", createdAt: Date.now(), updatedAt: Date.now(), }; log.info( `job ${jobId} start template=${input.template} platforms=${platforms.join(",")} ` + `chars=${input.text.length} skipLlm=${!!config.skipLlm} skipTts=${!!config.skipTts}` ); try { // Stage 1: Parse (shared across all platforms) callbacks?.onStageStart?.("parse"); const tParse = Date.now(); const parsed = await parseText(input.text, input.template, { llm: config.llm, skipLlm: config.skipLlm, source: input.source, }); job.parsed = parsed; log.info(`parse ok scenes=${parsed.scenes?.length ?? 0} (${Date.now() - tParse}ms)`); callbacks?.onStageComplete?.("parse"); // Stage 2: TTS (shared across all platforms) callbacks?.onStageStart?.("tts"); const tTts = Date.now(); const tts = await generateTTS(parsed, workDir, { provider: config.tts.provider, voiceId: config.tts.voiceId || input.voiceId, model: config.tts.model, format: config.tts.format, speed: config.tts.speed, skip: config.skipTts, alignment: config.alignment, }); job.tts = tts; log.info( `tts ok provider=${config.tts.provider} scenes=${tts.scenes?.length ?? 0} ` + `duration=${(tts.totalDurationSeconds ?? 0).toFixed(1)}s (${Date.now() - tTts}ms)` ); callbacks?.onStageComplete?.("tts"); // Stages 3-6: Per-platform loop const allExportFiles: ExportFile[] = []; job.composed = {}; for (const platform of platforms) { // Stage 3: Assets callbacks?.onStageStart?.(`assets:${platform}`); const assets = await resolveAssets( parsed, workDir, config.assets.root, config.assets.inputDir, { template: input.template, aspect: PLATFORM_PRESETS[platform].aspect } ); callbacks?.onStageComplete?.(`assets:${platform}`); // Stage 4: Compose callbacks?.onStageStart?.(`compose:${platform}`); const composed = composeProject(parsed, tts, assets, platform, input.template, config.branding.channelName); job.composed[platform] = composed; callbacks?.onStageComplete?.(`compose:${platform}`); // Stage 5: Render callbacks?.onStageStart?.(`render:${platform}`); const renderOutput = join(workDir, `render-${platform}.mp4`); const tRender = Date.now(); await renderVideo(composed, renderOutput, config.templates.entryPoint, config.assets.root); log.info(`render ${platform} ok -> ${renderOutput} (${Date.now() - tRender}ms)`); callbacks?.onStageComplete?.(`render:${platform}`); // Stage 6: Export callbacks?.onStageStart?.(`export:${platform}`); const exported = await exportVideo( renderOutput, composed, config.output.dir, jobId, config.publishMeta?.[platform]?.[input.template] ); const expFile = exported.files[0]; log.info( `export ${platform} -> ${expFile?.filePath} ` + `(${((expFile?.fileSizeBytes ?? 0) / 1024 / 1024).toFixed(1)}MB)` ); allExportFiles.push(...exported.files); callbacks?.onStageComplete?.(`export:${platform}`); } job.exported = { jobId, createdAt: Date.now(), files: allExportFiles }; job.status = "completed"; // Stage 7: Publish — upload to OSS + notify Feishu. Opt-in via config. if (!config.skipPublish && config.publish) { log.info( `publish oss=${config.publish.oss ? "on" : "off"} feishu=${config.publish.feishu ? "on" : "off"}` ); callbacks?.onStageStart?.("publish"); const publishCtx = { jobId, template: input.template, platforms, titleSuggestion: buildTitleSuggestion( parsed.title, parsed.scenes[0]?.trendSummary ), }; try { const outcome = await runPublish(allExportFiles, config.publish, publishCtx); const urlByPath = new Map(outcome.uploaded.map((u) => [u.filePath, u.ossUrl])); job.exported = { ...job.exported, files: allExportFiles.map((f) => ({ ...f, ossUrl: urlByPath.get(f.filePath), })), }; if (!outcome.ok && outcome.error) { job.publishError = outcome.error; log.error(`publish failed: ${outcome.error}`); callbacks?.onError?.("publish", new Error(outcome.error)); } callbacks?.onStageComplete?.("publish"); } catch (err) { // Publish must never mask a successful render. const msg = err instanceof Error ? err.message : String(err); job.publishError = msg; log.error(`publish error: ${msg}`); callbacks?.onError?.("publish", err instanceof Error ? err : new Error(msg)); } } } catch (err) { job.status = "failed"; job.error = err instanceof Error ? err.message : String(err); log.error(`job ${jobId} failed: ${job.error}`); callbacks?.onError?.("pipeline", err instanceof Error ? err : new Error(String(err))); // Render failed before any file existed — push failure info to Feishu. if (!config.skipPublish && config.publish) { await notifyGenerationFailure( config.publish, { jobId, template: input.template, platforms, titleSuggestion: buildTitleSuggestion( job.parsed?.title, job.parsed?.scenes?.[0]?.trendSummary ), }, job.error ?? "unknown error" ).catch(() => {}); } } const elapsed = ((Date.now() - t0) / 1000).toFixed(1); log.info( `job ${jobId} ${job.status} in ${elapsed}s` + (job.publishError ? ` publishError=${job.publishError.slice(0, 160)}` : "") ); job.updatedAt = Date.now(); return job; }