388 lines
14 KiB
JavaScript
388 lines
14 KiB
JavaScript
 
|
import { spawn } from 'node:child_process';
|
|||
|
|
import fs from 'node:fs/promises';
|
|||
|
|
import path from 'node:path';
|
|||
|
|
import crypto from 'node:crypto';
|
|||
|
|
import { EventEmitter } from 'node:events';
|
|||
|
|
import { ffprobe } from './probe.js';
|
|||
|
|
import { injectSphericalV1 } from './spherical.js';
|
|||
|
|
|
|||
|
|
export const MODE_LABEL = {
|
|||
|
|
cuda: 'NVDEC + CUDA 縮放 + NVENC',
|
|||
|
|
nvenc: 'CPU 解碼/縮放 + NVENC',
|
|||
|
|
vaapi: 'VAAPI 硬體編碼（Intel/AMD，/dev/dri）',
|
|||
|
|
cpu: 'libx264（CPU）',
|
|||
|
|
};
|
|||
|
|
|
|||
|
|
const VAAPI_DEVICE = process.env.VAAPI_DEVICE || '/dev/dri/renderD128';
|
|||
|
|
const X264_PRESET = process.env.X264_PRESET || 'veryfast';
|
|||
|
|
|
|||
|
|
function tryFfmpeg(args, timeoutMs = 20000) {
|
|||
|
|
return new Promise(resolve => {
|
|||
|
|
const p = spawn('ffmpeg', ['-hide_banner', '-v', 'error', '-y', ...args], { windowsHide: true });
|
|||
|
|
const t = setTimeout(() => { p.kill(); resolve(false); }, timeoutMs);
|
|||
|
|
p.on('error', () => { clearTimeout(t); resolve(false); });
|
|||
|
|
p.on('close', code => { clearTimeout(t); resolve(code === 0); });
|
|||
|
|
});
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/** Figure out which encode pipelines work on this machine, best first. */
|
|||
|
|
export async function detectModes() {
|
|||
|
|
const modes = [];
|
|||
|
|
const src = ['-f', 'lavfi', '-i', 'testsrc=s=256x128:d=0.1:r=30'];
|
|||
|
|
if (await tryFfmpeg([...src, '-vf', 'hwupload_cuda,scale_cuda=128:64', '-c:v', 'h264_nvenc', '-f', 'null', '-'])) {
|
|||
|
|
modes.push('cuda', 'nvenc');
|
|||
|
|
} else if (await tryFfmpeg([...src, '-c:v', 'h264_nvenc', '-f', 'null', '-'])) {
|
|||
|
|
modes.push('nvenc');
|
|||
|
|
}
|
|||
|
|
// Intel/AMD iGPU via VAAPI (typical for a NAS with /dev/dri passed into the container).
|
|||
|
|
if (await fs.access(VAAPI_DEVICE).then(() => true, () => false) &&
|
|||
|
|
await tryFfmpeg(['-vaapi_device', VAAPI_DEVICE, ...src,
|
|||
|
|
'-vf', 'format=nv12,hwupload,scale_vaapi=128:64', '-c:v', 'h264_vaapi', '-f', 'null', '-'])) {
|
|||
|
|
modes.push('vaapi');
|
|||
|
|
}
|
|||
|
|
modes.push('cpu');
|
|||
|
|
return modes;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/** "7M" * 2 -> "14M" (keeps the unit suffix). */
|
|||
|
|
function scaleRate(rate, factor) {
|
|||
|
|
const m = /^([\d.]+)\s*([kKmM]?)$/.exec(String(rate));
|
|||
|
|
if (!m) return rate;
|
|||
|
|
return `${Math.round(parseFloat(m[1]) * factor)}${m[2]}`;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/**
|
|||
|
|
* One ffmpeg run, one decode of the source, N scaled outputs. Reading the
|
|||
|
|
* source is usually the slow part (network drive), so every extra quality is
|
|||
|
|
* nearly free when produced in the same pass.
|
|||
|
|
*/
|
|||
|
|
function buildArgs({ mode, src, targets, audioCodec }) {
|
|||
|
|
const a = ['-y', '-hide_banner', '-loglevel', 'error', '-nostats'];
|
|||
|
|
if (mode === 'cuda') a.push('-hwaccel', 'cuda', '-hwaccel_output_format', 'cuda');
|
|||
|
|
if (mode === 'vaapi') a.push('-vaapi_device', VAAPI_DEVICE, '-hwaccel', 'vaapi', '-hwaccel_output_format', 'vaapi');
|
|||
|
|
a.push('-i', src);
|
|||
|
|
|
|||
|
|
for (const t of targets) {
|
|||
|
|
a.push('-map', '0:v:0', '-map', '0:a?', '-sn', '-dn');
|
|||
|
|
if (mode === 'cuda') a.push('-vf', `scale_cuda=${t.width}:${t.height}`);
|
|||
|
|
// format=nv12|vaapi,hwupload: works whether the decoder ran on the GPU or fell back to software.
|
|||
|
|
else if (mode === 'vaapi') a.push('-vf', `format=nv12|vaapi,hwupload,scale_vaapi=${t.width}:${t.height}:format=nv12`);
|
|||
|
|
else if (mode === 'nvenc') a.push('-vf', `scale=${t.width}:${t.height}:flags=bicubic,format=nv12`);
|
|||
|
|
else a.push('-vf', `scale=${t.width}:${t.height}:flags=bicubic,format=yuv420p`);
|
|||
|
|
|
|||
|
|
const bufsize = scaleRate(t.maxrate, 2);
|
|||
|
|
if (mode === 'cpu') {
|
|||
|
|
a.push('-c:v', 'libx264', '-preset', X264_PRESET, '-crf', '23',
|
|||
|
|
'-maxrate', t.maxrate, '-bufsize', bufsize, '-profile:v', 'high', '-g', '60');
|
|||
|
|
} else if (mode === 'vaapi') {
|
|||
|
|
a.push('-c:v', 'h264_vaapi', '-rc_mode', 'VBR', '-b:v', t.bitrate, '-maxrate', t.maxrate, '-bufsize', bufsize,
|
|||
|
|
'-profile:v', 'high', '-g', '60', '-bf', '2');
|
|||
|
|
} else {
|
|||
|
|
a.push('-c:v', 'h264_nvenc', '-preset', t.nvencPreset || 'p4', '-rc', 'vbr',
|
|||
|
|
'-b:v', t.bitrate, '-maxrate', t.maxrate, '-bufsize', bufsize,
|
|||
|
|
'-profile:v', 'high', '-g', '60', '-bf', '2', '-spatial-aq', '1');
|
|||
|
|
}
|
|||
|
|
if (audioCodec === 'aac') a.push('-c:a', 'copy');
|
|||
|
|
else a.push('-c:a', 'aac', '-b:a', '160k');
|
|||
|
|
a.push('-movflags', '+faststart', t.part);
|
|||
|
|
}
|
|||
|
|
a.push('-progress', 'pipe:1');
|
|||
|
|
return a;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
export class Transcoder extends EventEmitter {
|
|||
|
|
/**
|
|||
|
|
* @param {object} o
|
|||
|
|
* @param {string} o.cacheDir
|
|||
|
|
* @param {Array} o.qualities config.qualities
|
|||
|
|
* @param {(name:string)=>Promise<{file:string,stat:import('fs').Stats,info:object}>} o.resolveSource
|
|||
|
|
*/
|
|||
|
|
constructor({ cacheDir, qualities, resolveSource }) {
|
|||
|
|
super();
|
|||
|
|
this.cacheDir = cacheDir;
|
|||
|
|
this.qualities = new Map(qualities.map(q => [q.id, q]));
|
|||
|
|
this.resolveSource = resolveSource;
|
|||
|
|
this.modes = ['cpu'];
|
|||
|
|
this.jobs = new Map(); // id -> job
|
|||
|
|
this.queue = []; // job ids waiting
|
|||
|
|
this.running = null; // job currently encoding
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
async init() {
|
|||
|
|
await fs.mkdir(this.cacheDir, { recursive: true });
|
|||
|
|
this.modes = await detectModes();
|
|||
|
|
// Remove leftovers from a previous crash.
|
|||
|
|
for (const f of await fs.readdir(this.cacheDir)) {
|
|||
|
|
if (f.endsWith('.part.mp4') || f.endsWith('.sph.tmp')) await fs.rm(path.join(this.cacheDir, f), { force: true });
|
|||
|
|
}
|
|||
|
|
return this.modes;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
variantPath(name, qid) {
|
|||
|
|
return path.join(this.cacheDir, `${name}__${qid}.mp4`);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/** Is a finished, up-to-date variant on disk? Stale ones (source changed) are deleted. */
|
|||
|
|
async variantStatus(name, qid, sourceStat) {
|
|||
|
|
const out = this.variantPath(name, qid);
|
|||
|
|
try {
|
|||
|
|
const meta = JSON.parse(await fs.readFile(out + '.json', 'utf8'));
|
|||
|
|
const st = await fs.stat(out);
|
|||
|
|
// mtime tolerance of 2 s: SMB/WebDAV/FAT round timestamps, and the cache may be copied
|
|||
|
|
// between the PC (GPU transcode) and the NAS (serving) that see the same source file.
|
|||
|
|
if (meta.sourceSize === sourceStat.size && Math.abs(meta.sourceMtime - sourceStat.mtimeMs) < 2000) {
|
|||
|
|
return { ready: true, size: st.size, mode: meta.mode, createdAt: meta.createdAt, spherical: !!meta.spherical };
|
|||
|
|
}
|
|||
|
|
await this.deleteVariant(name, qid);
|
|||
|
|
} catch { /* not there */ }
|
|||
|
|
return { ready: false };
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
async deleteVariant(name, qid) {
|
|||
|
|
const job = this.findJob(name, qid);
|
|||
|
|
if (job) this.cancel(job.id);
|
|||
|
|
const out = this.variantPath(name, qid);
|
|||
|
|
await fs.rm(out, { force: true });
|
|||
|
|
await fs.rm(out + '.json', { force: true });
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/** Active (queued/running) job that will produce this variant, if any. */
|
|||
|
|
findJob(name, qid) {
|
|||
|
|
for (const j of this.jobs.values()) {
|
|||
|
|
if (j.name === name && (j.status === 'queued' || j.status === 'running') && j.targets.some(t => t.quality === qid)) return j;
|
|||
|
|
}
|
|||
|
|
return null;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
list() {
|
|||
|
|
return [...this.jobs.values()].map(publicJob);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/**
|
|||
|
|
* Queue one job that produces every quality in `qids` from a single read of
|
|||
|
|
* the source. Qualities that already exist or are already in flight are skipped.
|
|||
|
|
*/
|
|||
|
|
async enqueue(name, qids) {
|
|||
|
|
const { file, stat, info } = await this.resolveSource(name);
|
|||
|
|
if (!info?.video) throw httpError(400, '無法讀取影片資訊');
|
|||
|
|
|
|||
|
|
const targets = [];
|
|||
|
|
const skipped = [];
|
|||
|
|
for (const qid of qids) {
|
|||
|
|
const q = this.qualities.get(qid);
|
|||
|
|
if (!q) throw httpError(400, `未知的畫質 ${qid}`);
|
|||
|
|
if (q.width >= info.video.width) { skipped.push(`${q.label}：不小於原始解析度`); continue; }
|
|||
|
|
if (this.findJob(name, qid)) { skipped.push(`${q.label}：已在佇列中`); continue; }
|
|||
|
|
if ((await this.variantStatus(name, qid, stat)).ready) { skipped.push(`${q.label}：已存在`); continue; }
|
|||
|
|
const height = Math.round((q.width * info.video.height) / info.video.width / 2) * 2;
|
|||
|
|
targets.push({
|
|||
|
|
quality: qid, label: q.label, width: q.width, height,
|
|||
|
|
bitrate: q.bitrate, maxrate: q.maxrate, nvencPreset: q.nvencPreset,
|
|||
|
|
final: this.variantPath(name, qid),
|
|||
|
|
part: this.variantPath(name, qid).replace(/\.mp4$/, '.part.mp4'),
|
|||
|
|
done: false,
|
|||
|
|
});
|
|||
|
|
}
|
|||
|
|
if (!targets.length) throw httpError(409, skipped.length ? `沒有需要轉檔的畫質（${skipped.join('；')}）` : '沒有需要轉檔的畫質');
|
|||
|
|
targets.sort((a, b) => b.width - a.width);
|
|||
|
|
|
|||
|
|
const job = {
|
|||
|
|
id: crypto.randomBytes(6).toString('hex'),
|
|||
|
|
name, targets, skipped,
|
|||
|
|
src: file, sourceStat: { size: stat.size, mtime: Math.floor(stat.mtimeMs) },
|
|||
|
|
duration: info.duration || 0,
|
|||
|
|
audioCodec: info.audio?.codec || null,
|
|||
|
|
spherical: info.projection === 'equirectangular',
|
|||
|
|
status: 'queued', progress: 0, fps: 0, speed: 0, eta: null, outTime: 0,
|
|||
|
|
mode: null, attempt: 0, note: null, error: null,
|
|||
|
|
createdAt: Date.now(), startedAt: null, finishedAt: null,
|
|||
|
|
_proc: null,
|
|||
|
|
};
|
|||
|
|
this.jobs.set(job.id, job);
|
|||
|
|
this.queue.push(job.id);
|
|||
|
|
this.emit('update');
|
|||
|
|
this.pump();
|
|||
|
|
return publicJob(job);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
cancel(id) {
|
|||
|
|
const job = this.jobs.get(id);
|
|||
|
|
if (!job) throw httpError(404, '找不到工作');
|
|||
|
|
if (job.status === 'queued') {
|
|||
|
|
this.queue = this.queue.filter(x => x !== id);
|
|||
|
|
job.status = 'cancelled';
|
|||
|
|
job.finishedAt = Date.now();
|
|||
|
|
this.emit('update');
|
|||
|
|
} else if (job.status === 'running') {
|
|||
|
|
job.status = 'cancelled';
|
|||
|
|
job._proc?.kill();
|
|||
|
|
}
|
|||
|
|
return publicJob(job);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/** Drop finished entries from the list. */
|
|||
|
|
clearFinished() {
|
|||
|
|
for (const [id, j] of this.jobs) {
|
|||
|
|
if (j.status === 'done' || j.status === 'error' || j.status === 'cancelled') this.jobs.delete(id);
|
|||
|
|
}
|
|||
|
|
this.emit('update');
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
async pump() {
|
|||
|
|
if (this.running || !this.queue.length) return;
|
|||
|
|
const job = this.jobs.get(this.queue.shift());
|
|||
|
|
if (!job || job.status !== 'queued') return this.pump();
|
|||
|
|
this.running = job;
|
|||
|
|
job.status = 'running';
|
|||
|
|
job.startedAt = Date.now();
|
|||
|
|
this.emit('update');
|
|||
|
|
try {
|
|||
|
|
await this.run(job);
|
|||
|
|
} catch (e) {
|
|||
|
|
job.status = 'error';
|
|||
|
|
job.error = e.message;
|
|||
|
|
}
|
|||
|
|
job.finishedAt = Date.now();
|
|||
|
|
this.running = null;
|
|||
|
|
this.emit('update');
|
|||
|
|
this.emit('finished', publicJob(job));
|
|||
|
|
this.pump();
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
async removeParts(job) {
|
|||
|
|
for (const t of job.targets) await fs.rm(t.part, { force: true });
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
run(job) {
|
|||
|
|
return new Promise(resolve => {
|
|||
|
|
const attempt = (i) => {
|
|||
|
|
const mode = this.modes[i];
|
|||
|
|
job.mode = mode;
|
|||
|
|
job.attempt = i + 1;
|
|||
|
|
job.progress = 0; job.outTime = 0; job.fps = 0; job.speed = 0; job.eta = null;
|
|||
|
|
const args = buildArgs({ mode, src: job.src, targets: job.targets, audioCodec: job.audioCodec });
|
|||
|
|
const p = spawn('ffmpeg', args, { windowsHide: true });
|
|||
|
|
job._proc = p;
|
|||
|
|
let buf = '';
|
|||
|
|
const stderr = [];
|
|||
|
|
|
|||
|
|
p.stdout.on('data', d => {
|
|||
|
|
buf += d;
|
|||
|
|
let nl;
|
|||
|
|
while ((nl = buf.indexOf('\n')) >= 0) {
|
|||
|
|
const line = buf.slice(0, nl).trim();
|
|||
|
|
buf = buf.slice(nl + 1);
|
|||
|
|
this.progressLine(job, line);
|
|||
|
|
}
|
|||
|
|
});
|
|||
|
|
p.stderr.on('data', d => {
|
|||
|
|
stderr.push(String(d));
|
|||
|
|
if (stderr.length > 40) stderr.shift();
|
|||
|
|
});
|
|||
|
|
p.on('error', e => stderr.push(e.message));
|
|||
|
|
p.on('close', async code => {
|
|||
|
|
job._proc = null;
|
|||
|
|
if (job.status === 'cancelled') {
|
|||
|
|
await this.removeParts(job);
|
|||
|
|
return resolve();
|
|||
|
|
}
|
|||
|
|
if (code === 0) {
|
|||
|
|
try {
|
|||
|
|
for (const t of job.targets) await this.finalize(job, t, mode);
|
|||
|
|
job.status = 'done';
|
|||
|
|
job.progress = 1;
|
|||
|
|
job.eta = 0;
|
|||
|
|
job.note = null;
|
|||
|
|
} catch (e) {
|
|||
|
|
job.status = 'error';
|
|||
|
|
job.error = `寫入輸出失敗：${e.message}`;
|
|||
|
|
await this.removeParts(job);
|
|||
|
|
}
|
|||
|
|
return resolve();
|
|||
|
|
}
|
|||
|
|
const errText = stderr.join('').trim().split(/\r?\n/).filter(Boolean).slice(-3).join(' | ');
|
|||
|
|
await this.removeParts(job);
|
|||
|
|
if (i + 1 < this.modes.length) {
|
|||
|
|
job.note = `${MODE_LABEL[mode]} 失敗（${errText || 'exit ' + code}），改用 ${MODE_LABEL[this.modes[i + 1]]}`;
|
|||
|
|
console.warn(`[transcode] ${job.name}: ${job.note}`);
|
|||
|
|
this.emit('update');
|
|||
|
|
return attempt(i + 1);
|
|||
|
|
}
|
|||
|
|
job.status = 'error';
|
|||
|
|
job.error = errText || `ffmpeg exited ${code}`;
|
|||
|
|
resolve();
|
|||
|
|
});
|
|||
|
|
};
|
|||
|
|
attempt(0);
|
|||
|
|
});
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/** Re-attach 360 metadata (ffmpeg drops it on re-encode), verify, then publish the variant. */
|
|||
|
|
async finalize(job, t, mode) {
|
|||
|
|
let spherical = false;
|
|||
|
|
if (job.spherical) {
|
|||
|
|
job.note = `寫入 360 metadata（${t.label}）…`;
|
|||
|
|
this.emit('update');
|
|||
|
|
const tmp = t.final.replace(/\.mp4$/, '.sph.tmp');
|
|||
|
|
try {
|
|||
|
|
const r = await injectSphericalV1(t.part, tmp);
|
|||
|
|
if (r.injected) {
|
|||
|
|
const check = await ffprobe(tmp);
|
|||
|
|
if (check.projectionSource === 'metadata' && check.duration > 0) {
|
|||
|
|
await fs.rm(t.part, { force: true });
|
|||
|
|
await fs.rename(tmp, t.part);
|
|||
|
|
spherical = true;
|
|||
|
|
} else {
|
|||
|
|
console.warn(`[transcode] spherical 注入後驗證失敗，保留未注入版本 (${job.name} ${t.quality})`);
|
|||
|
|
await fs.rm(tmp, { force: true });
|
|||
|
|
}
|
|||
|
|
} else {
|
|||
|
|
console.warn(`[transcode] 未注入 spherical metadata：${r.reason}`);
|
|||
|
|
}
|
|||
|
|
} catch (e) {
|
|||
|
|
console.warn(`[transcode] spherical 注入失敗：${e.message}`);
|
|||
|
|
await fs.rm(tmp, { force: true });
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
await fs.writeFile(t.final + '.json', JSON.stringify({
|
|||
|
|
sourceSize: job.sourceStat.size, sourceMtime: job.sourceStat.mtime,
|
|||
|
|
width: t.width, height: t.height, mode, spherical, createdAt: Date.now(),
|
|||
|
|
}));
|
|||
|
|
await fs.rename(t.part, t.final);
|
|||
|
|
t.done = true;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
progressLine(job, line) {
|
|||
|
|
const eq = line.indexOf('=');
|
|||
|
|
if (eq < 0) return;
|
|||
|
|
const key = line.slice(0, eq);
|
|||
|
|
const val = line.slice(eq + 1).trim();
|
|||
|
|
switch (key) {
|
|||
|
|
case 'out_time_us': job.outTime = Number(val) / 1e6; break;
|
|||
|
|
case 'out_time_ms': if (!job.outTime) job.outTime = Number(val) / 1e6; break; // older ffmpeg: actually µs
|
|||
|
|
case 'fps': job.fps = parseFloat(val) || 0; break;
|
|||
|
|
case 'speed': job.speed = parseFloat(val) || 0; break;
|
|||
|
|
case 'progress': {
|
|||
|
|
if (job.duration > 0) {
|
|||
|
|
job.progress = Math.min(0.999, job.outTime / job.duration);
|
|||
|
|
job.eta = job.speed > 0 ? Math.max(0, (job.duration - job.outTime) / job.speed) : null;
|
|||
|
|
}
|
|||
|
|
this.emit('update');
|
|||
|
|
break;
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
function publicJob(j) {
|
|||
|
|
const { _proc, src, sourceStat, ...rest } = j;
|
|||
|
|
rest.targets = j.targets.map(({ quality, label, width, height, done }) => ({ quality, label, width, height, done }));
|
|||
|
|
return rest;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
function httpError(status, message) {
|
|||
|
|
const e = new Error(message);
|
|||
|
|
e.status = status;
|
|||
|
|
return e;
|
|||
|
|
}
|