This commit is contained in:
Worph
@@ -0,0 +1,81 @@
|
||||
/**
|
||||
* MetaMesh Plugin: still-extractor
|
||||
*
|
||||
* ============================================================================
|
||||
* PLUGIN FILE ACCESS ARCHITECTURE - WebDAV
|
||||
* ============================================================================
|
||||
*
|
||||
* File access via WebDAV, resolved per request from the calling meta-core's
|
||||
* /urls -> webdavUrlInternal (WEBDAV_URL overrides):
|
||||
* - Read media files: GET /webdav/watch/... or /webdav/test/...
|
||||
* - Write output: PUT /webdav/plugin/still-extractor/...
|
||||
* - Temp: Local /cache/temp mount (frame written before WebDAV upload)
|
||||
*
|
||||
* SECURITY: Plugins use WebDAV for all file access. No /files mount needed.
|
||||
*
|
||||
* ============================================================================
|
||||
*/
|
||||
|
||||
import Fastify from 'fastify';
|
||||
import type { HealthResponse, ProcessRequest, ProcessResponse, CallbackPayload, ConfigureRequest, ConfigureResponse } from './types.js';
|
||||
import { manifest, process as processFile, configure } from './plugin.js';
|
||||
|
||||
const app = Fastify({ logger: true });
|
||||
let ready = false;
|
||||
|
||||
app.get('/health', async (): Promise<HealthResponse> => ({
|
||||
status: 'healthy',
|
||||
ready,
|
||||
version: manifest.version
|
||||
}));
|
||||
|
||||
app.get('/manifest', async () => manifest);
|
||||
|
||||
app.post<{ Body: ConfigureRequest }>('/configure', async (request): Promise<ConfigureResponse> => {
|
||||
try {
|
||||
configure(request.body.config || {});
|
||||
console.log(`[${manifest.id}] Configuration updated`);
|
||||
return { status: 'ok' };
|
||||
} catch (error) {
|
||||
console.error(`[${manifest.id}] Configuration error:`, error);
|
||||
return { status: 'error', error: error instanceof Error ? error.message : String(error) };
|
||||
}
|
||||
});
|
||||
|
||||
app.post<{ Body: ProcessRequest }>('/process', async (request, reply) => {
|
||||
const { taskId, cid, filePath, callbackUrl, metaCoreUrl } = request.body;
|
||||
|
||||
if (!taskId || !cid || !filePath || !callbackUrl || !metaCoreUrl) {
|
||||
return reply.send({ status: 'rejected', error: 'Missing required fields' } as ProcessResponse);
|
||||
}
|
||||
|
||||
// Process asynchronously and send callback when done
|
||||
processFile(request.body, async (payload: CallbackPayload) => {
|
||||
try {
|
||||
await fetch(callbackUrl, {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify(payload)
|
||||
});
|
||||
} catch (error) {
|
||||
console.error(`[${manifest.id}] Callback error:`, error);
|
||||
}
|
||||
}).catch((error) => {
|
||||
console.error(`[${manifest.id}] Process error:`, error);
|
||||
});
|
||||
|
||||
return reply.send({ status: 'accepted' } as ProcessResponse);
|
||||
});
|
||||
|
||||
const port = parseInt(process.env.PORT || '8080', 10);
|
||||
|
||||
app.listen({ port, host: '0.0.0.0' }).then(() => {
|
||||
ready = true;
|
||||
console.log(`[${manifest.id}] Listening on port ${port}`);
|
||||
});
|
||||
|
||||
process.on('SIGTERM', async () => {
|
||||
ready = false;
|
||||
await app.close();
|
||||
process.exit(0);
|
||||
});
|
||||
@@ -0,0 +1,93 @@
|
||||
/**
|
||||
* meta-core API Client for writing metadata
|
||||
* Fails gracefully if meta-core is unavailable (logs warning, returns without throwing)
|
||||
*/
|
||||
|
||||
export class MetaCoreClient {
|
||||
constructor(private baseUrl: string) {}
|
||||
|
||||
private async safeFetch(url: string, options: RequestInit): Promise<Response | null> {
|
||||
try {
|
||||
return await fetch(url, { ...options, signal: AbortSignal.timeout(5000) });
|
||||
} catch (error) {
|
||||
console.warn(`[MetaCoreClient] Warning: meta-core unavailable at ${this.baseUrl}`);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
async setProperty(hashId: string, key: string, value: string): Promise<void> {
|
||||
const response = await this.safeFetch(`${this.baseUrl}/meta/${hashId}/${key}`, {
|
||||
method: 'PUT',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ value }),
|
||||
});
|
||||
if (response && !response.ok) {
|
||||
console.warn(`[MetaCoreClient] Failed to set property: ${response.status}`);
|
||||
}
|
||||
}
|
||||
|
||||
async getProperty(hashId: string, key: string): Promise<string | null> {
|
||||
const response = await this.safeFetch(`${this.baseUrl}/meta/${hashId}/${key}`, { method: 'GET' });
|
||||
if (!response || response.status === 404) return null;
|
||||
if (!response.ok) return null;
|
||||
const data = await response.json() as { value?: string };
|
||||
return data.value ?? null;
|
||||
}
|
||||
|
||||
async mergeMetadata(hashId: string, metadata: Record<string, string>): Promise<void> {
|
||||
const response = await this.safeFetch(`${this.baseUrl}/meta/${hashId}`, {
|
||||
method: 'PATCH',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify(metadata),
|
||||
});
|
||||
if (response && !response.ok) {
|
||||
console.warn(`[MetaCoreClient] Failed to merge metadata: ${response.status}`);
|
||||
}
|
||||
}
|
||||
|
||||
async deleteProperty(hashId: string, key: string): Promise<void> {
|
||||
const response = await this.safeFetch(`${this.baseUrl}/meta/${hashId}/${key}`, { method: 'DELETE' });
|
||||
if (response && !response.ok && response.status !== 404) {
|
||||
console.warn(`[MetaCoreClient] Failed to delete property: ${response.status}`);
|
||||
}
|
||||
}
|
||||
|
||||
async addToSet(hashId: string, key: string, value: string): Promise<void> {
|
||||
const response = await this.safeFetch(`${this.baseUrl}/meta/${hashId}/_add/${key}`, {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ value }),
|
||||
});
|
||||
if (response && !response.ok) {
|
||||
console.warn(`[MetaCoreClient] Failed to add to set: ${response.status}`);
|
||||
}
|
||||
}
|
||||
|
||||
async getMetadata(hashId: string): Promise<Record<string, string>> {
|
||||
const response = await this.safeFetch(`${this.baseUrl}/meta/${hashId}`, { method: 'GET' });
|
||||
if (!response || response.status === 404) return {};
|
||||
if (!response.ok) return {};
|
||||
const data = await response.json() as { metadata?: Record<string, string> };
|
||||
return data.metadata ?? {};
|
||||
}
|
||||
|
||||
/**
|
||||
* Compute the CID (Content Identifier) for a file
|
||||
* Uses the meta-core /file/cid API endpoint
|
||||
*/
|
||||
async computeFileCID(filePath: string): Promise<string | null> {
|
||||
try {
|
||||
const response = await this.safeFetch(`${this.baseUrl}/file/cid`, {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ path: filePath }),
|
||||
});
|
||||
if (!response || !response.ok) return null;
|
||||
const data = await response.json() as { cid?: string };
|
||||
return data.cid ?? null;
|
||||
} catch (error) {
|
||||
console.debug(`[MetaCoreClient] Failed to compute CID for ${filePath}:`, error);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
}
|
||||
+559
@@ -0,0 +1,559 @@
|
||||
/**
|
||||
* Still Extractor Plugin
|
||||
*
|
||||
* Grabs a representative frame out of a local video file and stores it as a
|
||||
* content-addressed JPEG, referenced from the video record as `still`.
|
||||
*
|
||||
* ============================================================================
|
||||
* WHY THIS EXISTS
|
||||
* ============================================================================
|
||||
* Artwork on a local file otherwise comes from TMDB, and only when a title
|
||||
* match succeeds. `still` (METADATA_KEYS.md section 6.1) is the episode-level
|
||||
* thumbnail slot, reserved for a future TMDB season-detail fetch and until now
|
||||
* written by nobody. This plugin gives it a first writer from the one source
|
||||
* that never 404s and never depends on a match: the file itself.
|
||||
*
|
||||
* A still is NOT a poster. Nothing gates on it and its absence is not an error
|
||||
* — hence `skipped` rather than `failed` on every "couldn't get a usable
|
||||
* frame" path. We never write `poster`: a frame grab masquerading as a real
|
||||
* poster would propagate to every peer that replicates the record.
|
||||
*
|
||||
* ============================================================================
|
||||
* PLUGIN FILE ACCESS ARCHITECTURE - WebDAV
|
||||
* ============================================================================
|
||||
* File access via WebDAV, on the meta-core named by each /process request
|
||||
* (its /urls -> webdavUrlInternal; WEBDAV_URL overrides):
|
||||
* - Read media files: GET /webdav/watch/... or /webdav/test/...
|
||||
* - Write output: PUT /webdav/plugin/still-extractor/...
|
||||
* - Temp: /cache/temp (frame written by ffmpeg before upload)
|
||||
*
|
||||
* We deliberately do NOT write a `stillPath` sibling. METADATA_KEYS.md names
|
||||
* one, but that is the gateway/peer_proxy mirroring convention: on the
|
||||
* meta-sort path `/files/plugin` is a watcher root
|
||||
* (meta-core/internal/config/config.go DefaultWatcherPaths), so the JPEG gets
|
||||
* its own record and midhash alias and `LookupPathByCID` resolves it through
|
||||
* the reverse index — "There is no path-companion sidecar field"
|
||||
* (meta-core/internal/storage/client.go). The tmdb plugin writes `poster`
|
||||
* without `posterPath` for the same reason.
|
||||
* ============================================================================
|
||||
*/
|
||||
|
||||
import { spawn } from 'child_process';
|
||||
import { createHash, randomBytes } from 'crypto';
|
||||
import { mkdirSync, readFileSync, existsSync, unlinkSync } from 'fs';
|
||||
import * as path from 'path';
|
||||
import type { PluginManifest, ProcessRequest, CallbackPayload } from './types.js';
|
||||
import { MetaCoreClient } from './meta-core-client.js';
|
||||
import { getWebDAVClient, WebDAVClient } from './webdav-client.js';
|
||||
|
||||
const PLUGIN_OUTPUT_WEBDAV_PATH = '/files/plugin/still-extractor';
|
||||
const TEMP_PATH = '/cache/temp';
|
||||
|
||||
export const manifest: PluginManifest = {
|
||||
id: 'still-extractor',
|
||||
name: 'Still Extractor',
|
||||
version: '1.0.0',
|
||||
description: 'Extracts a representative frame from a video file and stores it as the record\'s `still`',
|
||||
author: 'MetaMesh',
|
||||
// file-info gives us `fileType`; ffmpeg gives us `fileinfo/duration` and
|
||||
// the `stream/{n}` table we pick the primary video stream out of.
|
||||
dependencies: ['file-info', 'ffmpeg'],
|
||||
priority: 55,
|
||||
color: '#9C27B0',
|
||||
// One seek + one decode over HTTP. Never the fast queue.
|
||||
defaultQueue: 'background',
|
||||
timeout: 300000,
|
||||
schema: {
|
||||
still: {
|
||||
label: 'Still',
|
||||
type: 'cid',
|
||||
readonly: true,
|
||||
hint: 'CID of a frame extracted from the video itself (best-effort episode thumbnail)',
|
||||
},
|
||||
},
|
||||
config: {
|
||||
forceRecompute: { type: 'boolean', label: 'Force Recompute', default: false },
|
||||
seekPercents: { type: 'string', label: 'Seek Percents', default: '20,45,70' },
|
||||
maxWidth: { type: 'number', label: 'Max Width (px)', default: 640 },
|
||||
jpegQuality: { type: 'number', label: 'JPEG Quality (ffmpeg -q:v, lower is better)', default: 4 },
|
||||
frameTimeoutMs: { type: 'number', label: 'Per-Frame ffmpeg Timeout (ms)', default: 120000 },
|
||||
},
|
||||
};
|
||||
|
||||
// Configuration
|
||||
let forceRecompute = false;
|
||||
let seekPercents = [20, 45, 70];
|
||||
let maxWidth = 640;
|
||||
let jpegQuality = 4;
|
||||
let frameTimeoutMs = 120000;
|
||||
|
||||
export function configure(config: Record<string, unknown>): void {
|
||||
forceRecompute = config.forceRecompute === true;
|
||||
if (typeof config.seekPercents === 'string') {
|
||||
const parsed = config.seekPercents
|
||||
.split(',')
|
||||
.map((p) => Number(p.trim()))
|
||||
.filter((p) => Number.isFinite(p) && p > 0 && p < 100);
|
||||
if (parsed.length > 0) seekPercents = parsed;
|
||||
}
|
||||
if (Number.isFinite(Number(config.maxWidth))) maxWidth = Number(config.maxWidth);
|
||||
if (Number.isFinite(Number(config.jpegQuality))) jpegQuality = Number(config.jpegQuality);
|
||||
if (Number.isFinite(Number(config.frameTimeoutMs))) frameTimeoutMs = Number(config.frameTimeoutMs);
|
||||
console.log(
|
||||
`[still-extractor] Config: forceRecompute=${forceRecompute}, seekPercents=${seekPercents.join(',')}, ` +
|
||||
`maxWidth=${maxWidth}, jpegQuality=${jpegQuality}, frameTimeoutMs=${frameTimeoutMs}`
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Compute midhash256 CID from a Buffer (matches meta-hash algorithm).
|
||||
* Copied verbatim from metamesh-plugin-subtitle-extractor.
|
||||
*/
|
||||
export function computeMidHash256FromBuffer(data: Buffer): string {
|
||||
const SAMPLE_SIZE = 1024 * 1024; // 1MB
|
||||
const MIDHASH_VARINT = Buffer.from([0x80, 0x20]);
|
||||
|
||||
const fileSize = data.length;
|
||||
|
||||
const sizeBuffer = Buffer.allocUnsafe(8);
|
||||
sizeBuffer.writeBigUInt64BE(BigInt(fileSize), 0);
|
||||
|
||||
let sampleData: Buffer;
|
||||
if (fileSize <= SAMPLE_SIZE) {
|
||||
sampleData = data;
|
||||
} else {
|
||||
const middleOffset = Math.floor((fileSize - SAMPLE_SIZE) / 2);
|
||||
sampleData = data.subarray(middleOffset, middleOffset + SAMPLE_SIZE);
|
||||
}
|
||||
|
||||
const hashInput = Buffer.concat([sizeBuffer, sampleData]);
|
||||
const hashBuffer = createHash('sha256').update(hashInput).digest();
|
||||
|
||||
const cidBytes = Buffer.concat([
|
||||
Buffer.from([0x01]),
|
||||
MIDHASH_VARINT,
|
||||
MIDHASH_VARINT,
|
||||
Buffer.from([0x20]),
|
||||
hashBuffer
|
||||
]);
|
||||
|
||||
const base32Chars = 'abcdefghijklmnopqrstuvwxyz234567';
|
||||
let cid = 'b';
|
||||
let bits = 0;
|
||||
let value = 0;
|
||||
for (const byte of cidBytes) {
|
||||
value = (value << 8) | byte;
|
||||
bits += 8;
|
||||
while (bits >= 5) {
|
||||
bits -= 5;
|
||||
cid += base32Chars[(value >> bits) & 0x1f];
|
||||
}
|
||||
}
|
||||
if (bits > 0) {
|
||||
cid += base32Chars[(value << (5 - bits)) & 0x1f];
|
||||
}
|
||||
|
||||
return cid;
|
||||
}
|
||||
|
||||
/** Sanitize filename by removing invalid characters. */
|
||||
export function sanitizeFilename(name: string): string {
|
||||
return name
|
||||
.replace(/[<>:"/\\|?*]/g, '')
|
||||
.replace(/\s+/g, ' ')
|
||||
.trim();
|
||||
}
|
||||
|
||||
/**
|
||||
* Output filename for a still.
|
||||
*
|
||||
* The source CID is part of the name — unlike the tmdb plugin, which
|
||||
* deliberately leaves the file CID out so every release of one movie shares a
|
||||
* poster. A still belongs to exactly one file, so two files with the same
|
||||
* basename must not collide.
|
||||
*/
|
||||
export function stillFilename(filePath: string, sourceCid: string): string {
|
||||
const base = sanitizeFilename(path.basename(filePath, path.extname(filePath)));
|
||||
const shortCid = sourceCid.slice(-12);
|
||||
return `${base}[${shortCid}]_still.jpg`;
|
||||
}
|
||||
|
||||
export interface VideoStream {
|
||||
/** Absolute ffmpeg stream index — what `-map 0:<index>` wants. */
|
||||
index: number;
|
||||
width?: number;
|
||||
height?: number;
|
||||
codec?: string;
|
||||
frameRate?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Codecs that are a single picture, not a moving image. An `attached_pic`
|
||||
* stream is almost always one of these.
|
||||
*/
|
||||
const STILL_IMAGE_CODECS = new Set(['mjpeg', 'png', 'bmp', 'gif', 'webp', 'tiff']);
|
||||
|
||||
/**
|
||||
* Pick the primary video stream out of the ffmpeg plugin's `stream/{n}` table.
|
||||
*
|
||||
* ⚠ NOT `0:v:0`, and ⚠ not simply the largest. ffprobe reports attached cover
|
||||
* art (`attached_pic`) as `codec_type=video`, and the ffmpeg plugin's
|
||||
* `embeddedimage` branch never fires for it, so an embedded poster lands here
|
||||
* as a perfectly ordinary `type:'video'` stream — which `0:v:0` would happily
|
||||
* extract instead of the film. Nor can we just take the biggest: a 600x900
|
||||
* cover out-measures a 640x480 video.
|
||||
*
|
||||
* The stream table does not carry dispositions, so we use what it does carry.
|
||||
* The ffmpeg plugin drops `avg_frame_rate` when it is `0/0` (plugin.ts), and an
|
||||
* attached picture has no frame rate — so "has a frameRate and is not a
|
||||
* still-image codec" separates moving pictures from covers. Largest area wins
|
||||
* within the better class; if nothing looks like moving video we fall back to
|
||||
* the largest of whatever is there rather than giving up.
|
||||
*
|
||||
* Accepts both shapes meta-core hands back: the namespaced `stream/{n}` keys
|
||||
* the ffmpeg plugin writes, and the flattened `stream` array.
|
||||
*/
|
||||
export function selectPrimaryVideoStream(existingMeta: Record<string, string> | undefined): VideoStream | null {
|
||||
if (!existingMeta) return null;
|
||||
|
||||
const raw: unknown[] = [];
|
||||
|
||||
const flattened = existingMeta['stream'] as unknown;
|
||||
if (flattened) {
|
||||
try {
|
||||
const arr = Array.isArray(flattened) ? flattened : JSON.parse(String(flattened));
|
||||
if (Array.isArray(arr)) raw.push(...arr);
|
||||
} catch {
|
||||
// fall through to the namespaced form
|
||||
}
|
||||
}
|
||||
|
||||
if (raw.length === 0) {
|
||||
for (const [key, value] of Object.entries(existingMeta)) {
|
||||
if (key.startsWith('stream/')) raw.push(value);
|
||||
}
|
||||
}
|
||||
|
||||
const moving: VideoStream[] = [];
|
||||
const other: VideoStream[] = [];
|
||||
|
||||
for (const entry of raw) {
|
||||
let stream: {
|
||||
type?: string; index?: number; width?: number; height?: number;
|
||||
codec?: string; frameRate?: string;
|
||||
};
|
||||
try {
|
||||
stream = typeof entry === 'string' ? JSON.parse(entry) : (entry as typeof stream);
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
if (!stream || stream.type !== 'video' || stream.index == null) continue;
|
||||
|
||||
const candidate: VideoStream = {
|
||||
index: Number(stream.index),
|
||||
width: stream.width != null ? Number(stream.width) : undefined,
|
||||
height: stream.height != null ? Number(stream.height) : undefined,
|
||||
codec: stream.codec,
|
||||
frameRate: stream.frameRate,
|
||||
};
|
||||
|
||||
const looksMoving =
|
||||
!!candidate.frameRate && !STILL_IMAGE_CODECS.has((candidate.codec ?? '').toLowerCase());
|
||||
(looksMoving ? moving : other).push(candidate);
|
||||
}
|
||||
|
||||
const area = (s: VideoStream) => (s.width ?? 0) * (s.height ?? 0);
|
||||
const pool = moving.length > 0 ? moving : other;
|
||||
return pool.reduce<VideoStream | null>((best, s) => (!best || area(s) > area(best) ? s : best), null);
|
||||
}
|
||||
|
||||
/**
|
||||
* Seek offsets to try, in order.
|
||||
*
|
||||
* Clamped away from both ends: the head is titles/logos/black and the tail is
|
||||
* credits. A file with no usable duration gets a blind 10s attempt and then
|
||||
* the very start, which is all we can honestly do.
|
||||
*/
|
||||
export function candidateOffsets(durationSec: number | undefined, percents: number[] = seekPercents): number[] {
|
||||
if (!durationSec || !Number.isFinite(durationSec) || durationSec <= 0) {
|
||||
return [10, 0];
|
||||
}
|
||||
if (durationSec <= 4) {
|
||||
return [Math.max(0, durationSec / 2)];
|
||||
}
|
||||
const lo = 2;
|
||||
const hi = durationSec - 2;
|
||||
const offsets = percents
|
||||
.map((p) => (durationSec * p) / 100)
|
||||
.map((t) => Math.min(hi, Math.max(lo, t)));
|
||||
return Array.from(new Set(offsets.map((t) => Math.round(t * 1000) / 1000)));
|
||||
}
|
||||
|
||||
export interface FrameStats {
|
||||
mean: number;
|
||||
stddev: number;
|
||||
}
|
||||
|
||||
/** Mean + standard deviation of an 8-bit grayscale buffer, both 0..255. */
|
||||
export function frameStats(gray: Buffer): FrameStats {
|
||||
if (gray.length === 0) return { mean: 0, stddev: 0 };
|
||||
let sum = 0;
|
||||
for (const v of gray) sum += v;
|
||||
const mean = sum / gray.length;
|
||||
let variance = 0;
|
||||
for (const v of gray) variance += (v - mean) ** 2;
|
||||
return { mean, stddev: Math.sqrt(variance / gray.length) };
|
||||
}
|
||||
|
||||
/**
|
||||
* Is this frame worth keeping?
|
||||
*
|
||||
* Rejects the three things a blind seek lands on far more often than you would
|
||||
* expect: a black frame, a white/blown card, and a flat single-colour fade.
|
||||
* `thumbnail=N` already avoids most of them; this is the backstop.
|
||||
*/
|
||||
export function isUsableFrame(stats: FrameStats): boolean {
|
||||
if (stats.stddev < 6) return false; // flat: fade, solid colour, blank card
|
||||
if (stats.mean < 12 || stats.mean > 243) return false; // black / blown out
|
||||
return true;
|
||||
}
|
||||
|
||||
/** Run a command, resolving with its exit code, stdout and stderr. Killed hard on timeout. */
|
||||
function run(
|
||||
cmd: string,
|
||||
args: string[],
|
||||
timeoutMs: number
|
||||
): Promise<{ code: number | null; stdout: Buffer; stderr: string; timedOut: boolean }> {
|
||||
return new Promise((resolve) => {
|
||||
const child = spawn(cmd, args, { stdio: ['ignore', 'pipe', 'pipe'] });
|
||||
const stdout: Buffer[] = [];
|
||||
let stderr = '';
|
||||
let timedOut = false;
|
||||
|
||||
const timer = setTimeout(() => {
|
||||
timedOut = true;
|
||||
child.kill('SIGKILL');
|
||||
}, timeoutMs);
|
||||
|
||||
child.stdout.on('data', (d: Buffer) => stdout.push(d));
|
||||
child.stderr.on('data', (d: Buffer) => {
|
||||
// Keep the tail only — a broken input can spew megabytes.
|
||||
stderr = (stderr + d.toString()).slice(-4000);
|
||||
});
|
||||
child.on('error', (err) => {
|
||||
clearTimeout(timer);
|
||||
resolve({ code: null, stdout: Buffer.concat(stdout), stderr: stderr + String(err), timedOut });
|
||||
});
|
||||
child.on('close', (code) => {
|
||||
clearTimeout(timer);
|
||||
resolve({ code, stdout: Buffer.concat(stdout), stderr, timedOut });
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Grab one frame at `offsetSec` and return the encoded JPEG.
|
||||
*
|
||||
* ⚠ `-ss` goes BEFORE `-i`. Input seeking makes ffmpeg jump via HTTP Range;
|
||||
* output seeking (`-ss` after `-i`) decodes every frame from zero, which over
|
||||
* WebDAV means pulling the whole file to reach the 20% mark. On MKV expect a
|
||||
* tail read regardless — the cue index lives at EOF.
|
||||
*/
|
||||
export async function grabFrame(
|
||||
input: string,
|
||||
streamIndex: number,
|
||||
offsetSec: number,
|
||||
outputPath: string
|
||||
): Promise<Buffer | null> {
|
||||
const args = ['-y', '-hide_banner', '-loglevel', 'error'];
|
||||
|
||||
// rw_timeout is a protocol option; only http/tcp accept it, and passing it
|
||||
// to the file protocol makes ffmpeg bail before it starts.
|
||||
if (/^https?:\/\//i.test(input)) {
|
||||
args.push('-rw_timeout', '30000000'); // microseconds
|
||||
}
|
||||
|
||||
args.push(
|
||||
'-ss', String(offsetSec),
|
||||
'-i', input,
|
||||
'-map', `0:${streamIndex}`,
|
||||
'-an', '-sn', '-dn',
|
||||
'-frames:v', '1',
|
||||
// thumbnail=N picks the most representative of the next N frames rather
|
||||
// than whatever the seek happened to land on.
|
||||
'-vf', `thumbnail=60,scale='min(iw,${maxWidth})':-2`,
|
||||
'-q:v', String(jpegQuality),
|
||||
'-f', 'image2',
|
||||
outputPath
|
||||
);
|
||||
|
||||
const result = await run('ffmpeg', args, frameTimeoutMs);
|
||||
if (result.timedOut) {
|
||||
console.warn(`[still-extractor] ffmpeg timed out at ${offsetSec}s`);
|
||||
return null;
|
||||
}
|
||||
if (result.code !== 0 || !existsSync(outputPath)) {
|
||||
console.warn(`[still-extractor] ffmpeg failed at ${offsetSec}s: ${result.stderr.trim() || `exit ${result.code}`}`);
|
||||
return null;
|
||||
}
|
||||
|
||||
const jpeg = readFileSync(outputPath);
|
||||
return jpeg.length > 0 ? jpeg : null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Measure a JPEG we already have on disk.
|
||||
*
|
||||
* Deliberately a second ffmpeg pass on the local temp file rather than a second
|
||||
* output on the grab command: two filter chains off one input would each run
|
||||
* their own `thumbnail` instance, and there is no guarantee the frame we
|
||||
* measured is the frame we encoded. This decodes a ~40 KB local file — no
|
||||
* seek, no network.
|
||||
*/
|
||||
export async function statsForJpeg(jpegPath: string): Promise<FrameStats | null> {
|
||||
const result = await run(
|
||||
'ffmpeg',
|
||||
['-hide_banner', '-loglevel', 'error', '-i', jpegPath,
|
||||
'-vf', 'scale=32:32,format=gray', '-f', 'rawvideo', '-'],
|
||||
15000
|
||||
);
|
||||
if (result.code !== 0 || result.stdout.length === 0) return null;
|
||||
return frameStats(result.stdout);
|
||||
}
|
||||
|
||||
export interface ExtractResult {
|
||||
jpeg: Buffer;
|
||||
offsetSec: number;
|
||||
stats: FrameStats;
|
||||
}
|
||||
|
||||
/**
|
||||
* Walk the candidate offsets until one yields a usable frame.
|
||||
*
|
||||
* Returns null when every candidate was black, flat, or unreadable — that is a
|
||||
* `skipped`, not a failure.
|
||||
*/
|
||||
export async function extractStill(
|
||||
input: string,
|
||||
streamIndex: number,
|
||||
durationSec: number | undefined,
|
||||
tempPath: string = TEMP_PATH
|
||||
): Promise<ExtractResult | null> {
|
||||
mkdirSync(tempPath, { recursive: true });
|
||||
// NB: `process` is this module's own export, not node's global — hence a
|
||||
// random suffix rather than a pid for uniqueness between concurrent tasks.
|
||||
const outputPath = path.join(tempPath, `still-${randomBytes(6).toString('hex')}.jpg`);
|
||||
|
||||
try {
|
||||
for (const offsetSec of candidateOffsets(durationSec)) {
|
||||
const jpeg = await grabFrame(input, streamIndex, offsetSec, outputPath);
|
||||
if (!jpeg) continue;
|
||||
|
||||
const stats = await statsForJpeg(outputPath);
|
||||
if (!stats) {
|
||||
console.warn(`[still-extractor] Could not measure frame at ${offsetSec}s`);
|
||||
continue;
|
||||
}
|
||||
if (!isUsableFrame(stats)) {
|
||||
console.log(
|
||||
`[still-extractor] Rejected frame at ${offsetSec}s ` +
|
||||
`(mean=${stats.mean.toFixed(1)}, stddev=${stats.stddev.toFixed(1)})`
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
console.log(
|
||||
`[still-extractor] Accepted frame at ${offsetSec}s ` +
|
||||
`(mean=${stats.mean.toFixed(1)}, stddev=${stats.stddev.toFixed(1)}, ${jpeg.length} bytes)`
|
||||
);
|
||||
return { jpeg, offsetSec, stats };
|
||||
}
|
||||
return null;
|
||||
} finally {
|
||||
if (existsSync(outputPath)) {
|
||||
try { unlinkSync(outputPath); } catch { /* best effort */ }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Upload the frame and return its CID. */
|
||||
async function storeStill(
|
||||
client: WebDAVClient,
|
||||
jpeg: Buffer,
|
||||
filePath: string,
|
||||
sourceCid: string
|
||||
): Promise<string> {
|
||||
const webdavPath = `${PLUGIN_OUTPUT_WEBDAV_PATH}/${stillFilename(filePath, sourceCid)}`;
|
||||
await client.writeFile(webdavPath, jpeg);
|
||||
console.log(`[still-extractor] Uploaded still to WebDAV: ${webdavPath}`);
|
||||
return computeMidHash256FromBuffer(jpeg);
|
||||
}
|
||||
|
||||
export async function process(
|
||||
request: ProcessRequest,
|
||||
sendCallback: (payload: CallbackPayload) => Promise<void>
|
||||
): Promise<void> {
|
||||
const startTime = Date.now();
|
||||
const { taskId, cid, filePath, existingMeta } = request;
|
||||
const metaCore = new MetaCoreClient(request.metaCoreUrl);
|
||||
|
||||
const skip = (reason: string) =>
|
||||
sendCallback({ taskId, status: 'skipped', duration: Date.now() - startTime, reason });
|
||||
|
||||
try {
|
||||
if (existingMeta?.fileType !== 'video') {
|
||||
await skip('Not a video file');
|
||||
return;
|
||||
}
|
||||
|
||||
if (existingMeta?.still && !forceRecompute) {
|
||||
// Also the mechanism that keeps this plugin from ever fighting a
|
||||
// future TMDB episode-still writer: authoritative artwork wins by
|
||||
// arriving first and never being overwritten from here.
|
||||
await skip('Still already present');
|
||||
return;
|
||||
}
|
||||
|
||||
const stream = selectPrimaryVideoStream(existingMeta);
|
||||
if (!stream) {
|
||||
await skip('No video stream');
|
||||
return;
|
||||
}
|
||||
|
||||
// WebDAV of the very core we're enriching, so the frame bytes and the
|
||||
// CID we write can never point at different cores.
|
||||
const webdavClient = await getWebDAVClient(request.metaCoreUrl);
|
||||
if (!webdavClient) {
|
||||
await sendCallback({
|
||||
taskId,
|
||||
status: 'failed',
|
||||
duration: Date.now() - startTime,
|
||||
error: 'No WebDAV endpoint available',
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
const duration = Number(existingMeta?.['fileinfo/duration']);
|
||||
const input = webdavClient.toWebDAVUrl(filePath);
|
||||
|
||||
const extracted = await extractStill(input, stream.index, Number.isFinite(duration) ? duration : undefined);
|
||||
if (!extracted) {
|
||||
await skip('No usable frame');
|
||||
return;
|
||||
}
|
||||
|
||||
const stillCid = await storeStill(webdavClient, extracted.jpeg, filePath, cid);
|
||||
await metaCore.setProperty(cid, 'still', stillCid);
|
||||
console.log(`[still-extractor] Set still CID: ${stillCid}`);
|
||||
|
||||
await sendCallback({ taskId, status: 'completed', duration: Date.now() - startTime });
|
||||
} catch (error) {
|
||||
console.error(`[still-extractor] Error processing ${filePath}:`, error);
|
||||
await sendCallback({
|
||||
taskId,
|
||||
status: 'failed',
|
||||
duration: Date.now() - startTime,
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
/**
|
||||
* MetaMesh Plugin Types
|
||||
*/
|
||||
|
||||
export interface PluginManifest {
|
||||
id: string;
|
||||
name: string;
|
||||
version: string;
|
||||
description?: string;
|
||||
author?: string;
|
||||
dependencies?: string[];
|
||||
schema?: Record<string, SchemaField>;
|
||||
config?: Record<string, ConfigField>;
|
||||
priority?: number;
|
||||
color?: string;
|
||||
defaultQueue?: 'fast' | 'background';
|
||||
timeout?: number;
|
||||
}
|
||||
|
||||
export interface SchemaField {
|
||||
label: string;
|
||||
type?: 'string' | 'number' | 'boolean' | 'select' | 'text' | 'array' | 'json' | 'cid';
|
||||
readonly?: boolean;
|
||||
hint?: string;
|
||||
}
|
||||
|
||||
export interface ConfigField {
|
||||
type: 'string' | 'number' | 'boolean' | 'select';
|
||||
label?: string;
|
||||
required?: boolean;
|
||||
default?: unknown;
|
||||
secret?: boolean;
|
||||
}
|
||||
|
||||
export interface HealthResponse {
|
||||
status: 'healthy' | 'unhealthy';
|
||||
ready: boolean;
|
||||
version: string;
|
||||
}
|
||||
|
||||
export interface ProcessRequest {
|
||||
taskId: string;
|
||||
cid: string;
|
||||
filePath: string;
|
||||
callbackUrl: string;
|
||||
metaCoreUrl: string;
|
||||
existingMeta?: Record<string, string>;
|
||||
}
|
||||
|
||||
export interface ProcessResponse {
|
||||
status: 'accepted' | 'rejected';
|
||||
error?: string;
|
||||
}
|
||||
|
||||
export interface CallbackPayload {
|
||||
taskId: string;
|
||||
status: 'completed' | 'failed' | 'skipped';
|
||||
duration: number;
|
||||
error?: string;
|
||||
reason?: string;
|
||||
}
|
||||
|
||||
export interface ConfigureRequest {
|
||||
config: Record<string, unknown>;
|
||||
}
|
||||
|
||||
export interface ConfigureResponse {
|
||||
status: 'ok' | 'error';
|
||||
error?: string;
|
||||
}
|
||||
@@ -0,0 +1,199 @@
|
||||
/**
|
||||
* WebDAV Client for Plugin File Access
|
||||
*
|
||||
* Provides HTTP-based file access to meta-sort's WebDAV endpoint.
|
||||
* Replaces direct filesystem access for containerized plugins.
|
||||
*/
|
||||
|
||||
export interface FileStats {
|
||||
size: number;
|
||||
mtime?: Date;
|
||||
}
|
||||
|
||||
export class WebDAVClient {
|
||||
private baseUrl: string;
|
||||
|
||||
constructor(baseUrl: string) {
|
||||
// Remove trailing slash if present
|
||||
this.baseUrl = baseUrl.replace(/\/$/, '');
|
||||
}
|
||||
|
||||
/**
|
||||
* Convert absolute file path to WebDAV URL
|
||||
* Strips /files prefix and appends to base URL
|
||||
*/
|
||||
toWebDAVUrl(filePath: string): string {
|
||||
// filePath is like "/files/watch/movie.mkv"
|
||||
// We need to convert to "http://meta-sort-dev/webdav/watch/movie.mkv"
|
||||
let relativePath = filePath;
|
||||
|
||||
// Strip /files prefix if present
|
||||
if (relativePath.startsWith('/files/')) {
|
||||
relativePath = relativePath.substring(6); // Remove "/files"
|
||||
} else if (relativePath.startsWith('/files')) {
|
||||
relativePath = relativePath.substring(6);
|
||||
}
|
||||
|
||||
// Ensure path starts with /
|
||||
if (!relativePath.startsWith('/')) {
|
||||
relativePath = '/' + relativePath;
|
||||
}
|
||||
|
||||
return this.baseUrl + relativePath;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get file stats (size, mtime) via HTTP HEAD request
|
||||
*/
|
||||
async stat(filePath: string): Promise<FileStats> {
|
||||
const url = this.toWebDAVUrl(filePath);
|
||||
|
||||
const response = await fetch(url, { method: 'HEAD' });
|
||||
|
||||
if (!response.ok) {
|
||||
throw new Error(`WebDAV HEAD failed for ${filePath}: ${response.status} ${response.statusText}`);
|
||||
}
|
||||
|
||||
const contentLength = response.headers.get('content-length');
|
||||
const lastModified = response.headers.get('last-modified');
|
||||
|
||||
return {
|
||||
size: contentLength ? parseInt(contentLength, 10) : 0,
|
||||
mtime: lastModified ? new Date(lastModified) : undefined,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Read first N bytes of a file (for magic byte detection)
|
||||
* Uses HTTP Range request
|
||||
*/
|
||||
async readBytes(filePath: string, start: number, end: number): Promise<Buffer> {
|
||||
const url = this.toWebDAVUrl(filePath);
|
||||
|
||||
const response = await fetch(url, {
|
||||
method: 'GET',
|
||||
headers: {
|
||||
'Range': `bytes=${start}-${end}`,
|
||||
},
|
||||
});
|
||||
|
||||
if (!response.ok && response.status !== 206) {
|
||||
throw new Error(`WebDAV Range GET failed for ${filePath}: ${response.status} ${response.statusText}`);
|
||||
}
|
||||
|
||||
const arrayBuffer = await response.arrayBuffer();
|
||||
return Buffer.from(arrayBuffer);
|
||||
}
|
||||
|
||||
/**
|
||||
* Read entire file as buffer (for small files only)
|
||||
*/
|
||||
async readFile(filePath: string): Promise<Buffer> {
|
||||
const url = this.toWebDAVUrl(filePath);
|
||||
|
||||
const response = await fetch(url, { method: 'GET' });
|
||||
|
||||
if (!response.ok) {
|
||||
throw new Error(`WebDAV GET failed for ${filePath}: ${response.status} ${response.statusText}`);
|
||||
}
|
||||
|
||||
const arrayBuffer = await response.arrayBuffer();
|
||||
return Buffer.from(arrayBuffer);
|
||||
}
|
||||
|
||||
/**
|
||||
* Write file via HTTP PUT
|
||||
*/
|
||||
async writeFile(filePath: string, data: Buffer | string): Promise<void> {
|
||||
const url = this.toWebDAVUrl(filePath);
|
||||
|
||||
const response = await fetch(url, {
|
||||
method: 'PUT',
|
||||
body: data,
|
||||
headers: {
|
||||
'Content-Type': 'application/octet-stream',
|
||||
},
|
||||
});
|
||||
|
||||
if (!response.ok) {
|
||||
throw new Error(`WebDAV PUT failed for ${filePath}: ${response.status} ${response.statusText}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if file exists via HTTP HEAD
|
||||
*/
|
||||
async exists(filePath: string): Promise<boolean> {
|
||||
const url = this.toWebDAVUrl(filePath);
|
||||
|
||||
try {
|
||||
const response = await fetch(url, { method: 'HEAD' });
|
||||
return response.ok;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* WebDAV client for the meta-core that drove a /process call.
|
||||
*
|
||||
* The endpoint is read from that core's own /urls (`webdavUrlInternal`),
|
||||
* so blobs always land in the /files of the core we write metadata to. A feeder
|
||||
* whose plugins run beside it but enrich a remote hub core needs no per-stack
|
||||
* wiring, and it is impossible to write posters into one core while pointing
|
||||
* CIDs at another.
|
||||
*
|
||||
* WEBDAV_URL still wins when set — an explicit override for tests and for
|
||||
* topologies where the core's advertised internal URL isn't reachable from here.
|
||||
*
|
||||
* Resolutions are cached per meta-core URL; failures are not, so a core that is
|
||||
* still starting up is retried on the next task.
|
||||
*/
|
||||
const clientCache = new Map<string, Promise<WebDAVClient | null>>();
|
||||
|
||||
async function resolveWebDAVUrl(metaCoreUrl: string): Promise<string | null> {
|
||||
// /urls, not /api/urls — metaCoreUrl points at the meta-core backend, whose
|
||||
// discovery routes are unprefixed (same base the /meta/* writes use).
|
||||
const base = metaCoreUrl.replace(/\/$/, '');
|
||||
const response = await fetch(`${base}/urls`, { signal: AbortSignal.timeout(5000) });
|
||||
|
||||
if (!response.ok) {
|
||||
throw new Error(`GET ${base}/urls failed: ${response.status} ${response.statusText}`);
|
||||
}
|
||||
|
||||
const urls = await response.json() as { webdavUrlInternal?: string; webdavUrl?: string };
|
||||
return urls.webdavUrlInternal ?? urls.webdavUrl ?? null;
|
||||
}
|
||||
|
||||
export async function getWebDAVClient(metaCoreUrl: string): Promise<WebDAVClient | null> {
|
||||
const override = process.env.WEBDAV_URL;
|
||||
if (override) {
|
||||
return new WebDAVClient(override);
|
||||
}
|
||||
|
||||
let pending = clientCache.get(metaCoreUrl);
|
||||
if (!pending) {
|
||||
pending = resolveWebDAVUrl(metaCoreUrl).then((webdavUrl) => {
|
||||
if (!webdavUrl) {
|
||||
console.warn(`[webdav-client] meta-core ${metaCoreUrl} advertises no WebDAV URL`);
|
||||
return null;
|
||||
}
|
||||
console.log(`[webdav-client] Using WebDAV endpoint ${webdavUrl} (from ${metaCoreUrl})`);
|
||||
return new WebDAVClient(webdavUrl);
|
||||
});
|
||||
clientCache.set(metaCoreUrl, pending);
|
||||
}
|
||||
|
||||
try {
|
||||
const client = await pending;
|
||||
if (!client) {
|
||||
clientCache.delete(metaCoreUrl);
|
||||
}
|
||||
return client;
|
||||
} catch (error) {
|
||||
clientCache.delete(metaCoreUrl);
|
||||
console.warn(`[webdav-client] Failed to resolve WebDAV URL from ${metaCoreUrl}: ${error instanceof Error ? error.message : String(error)}`);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user