initial commit
This commit is contained in:
Pierre
@@ -0,0 +1,73 @@
|
||||
/**
|
||||
* MetaMesh Plugin: ffmpeg
|
||||
* Entry point
|
||||
*/
|
||||
|
||||
import Fastify from 'fastify';
|
||||
import type {
|
||||
HealthResponse,
|
||||
ProcessRequest,
|
||||
ProcessResponse,
|
||||
CallbackPayload,
|
||||
ConfigureRequest,
|
||||
ConfigureResponse,
|
||||
} from './types.js';
|
||||
import { manifest, process as processFile } from './plugin.js';
|
||||
|
||||
const app = Fastify({ logger: true });
|
||||
let ready = false;
|
||||
let pluginConfig: Record<string, unknown> = {};
|
||||
|
||||
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 {
|
||||
pluginConfig = request.body.config || {};
|
||||
console.log('[ffmpeg] Configuration updated');
|
||||
return { status: 'ok' };
|
||||
} catch (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);
|
||||
}
|
||||
|
||||
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('[ffmpeg] Callback error:', error);
|
||||
}
|
||||
}).catch(console.error);
|
||||
|
||||
return reply.send({ status: 'accepted' } as ProcessResponse);
|
||||
});
|
||||
|
||||
const port = parseInt(process.env.PORT || '8080', 10);
|
||||
const host = process.env.HOST || '0.0.0.0';
|
||||
|
||||
app.listen({ port, host }).then(() => {
|
||||
ready = true;
|
||||
console.log(`[ffmpeg] Plugin listening on http://${host}:${port}`);
|
||||
});
|
||||
|
||||
process.on('SIGTERM', async () => {
|
||||
ready = false;
|
||||
await app.close();
|
||||
process.exit(0);
|
||||
});
|
||||
@@ -0,0 +1,73 @@
|
||||
/**
|
||||
* 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 ?? {};
|
||||
}
|
||||
}
|
||||
+201
@@ -0,0 +1,201 @@
|
||||
/**
|
||||
* FFmpeg Plugin
|
||||
*
|
||||
* Extracts video/audio/subtitle stream metadata using FFprobe.
|
||||
*/
|
||||
|
||||
import ffmpeg from 'fluent-ffmpeg';
|
||||
import type { PluginManifest, ProcessRequest, CallbackPayload } from './types.js';
|
||||
import { MetaCoreClient } from './meta-core-client.js';
|
||||
|
||||
export const manifest: PluginManifest = {
|
||||
id: 'ffmpeg',
|
||||
name: 'FFmpeg Metadata',
|
||||
version: '1.0.0',
|
||||
description: 'Extracts video/audio/subtitle stream metadata using FFprobe',
|
||||
author: 'MetaMesh',
|
||||
dependencies: ['file-info'],
|
||||
priority: 15,
|
||||
color: '#4CAF50',
|
||||
defaultQueue: 'fast',
|
||||
timeout: 60000,
|
||||
schema: {
|
||||
'fileinfo/duration': { label: 'Duration', type: 'string', readonly: true },
|
||||
'fileinfo/formatName': { label: 'Format Name', type: 'string', readonly: true },
|
||||
'fileinfo/streamdetails/video': {
|
||||
label: 'Video Streams',
|
||||
type: 'json',
|
||||
readonly: true,
|
||||
hint: 'Video codec, resolution, bitrate, frame rate',
|
||||
},
|
||||
'fileinfo/streamdetails/audio': {
|
||||
label: 'Audio Streams',
|
||||
type: 'json',
|
||||
readonly: true,
|
||||
hint: 'Audio codec, language, sample rate, channels',
|
||||
},
|
||||
'fileinfo/streamdetails/subtitle': {
|
||||
label: 'Subtitle Streams',
|
||||
type: 'json',
|
||||
readonly: true,
|
||||
hint: 'Subtitle codec, language, title',
|
||||
},
|
||||
},
|
||||
config: {},
|
||||
};
|
||||
|
||||
interface FFprobeData {
|
||||
format?: {
|
||||
duration?: number;
|
||||
format_name?: string;
|
||||
};
|
||||
streams?: Array<{
|
||||
codec_type?: string;
|
||||
codec_name?: string;
|
||||
index?: number;
|
||||
duration?: string;
|
||||
width?: number;
|
||||
height?: number;
|
||||
bit_rate?: string;
|
||||
avg_frame_rate?: string;
|
||||
sample_rate?: string;
|
||||
channel_layout?: string;
|
||||
disposition?: {
|
||||
forced?: number;
|
||||
default?: number;
|
||||
};
|
||||
tags?: {
|
||||
language?: string;
|
||||
title?: string;
|
||||
rotate?: string;
|
||||
};
|
||||
}>;
|
||||
}
|
||||
|
||||
function runFFprobe(filePath: string): Promise<FFprobeData> {
|
||||
return new Promise((resolve, reject) => {
|
||||
ffmpeg(filePath).ffprobe((err, data) => {
|
||||
if (err) reject(err);
|
||||
else resolve(data as FFprobeData);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
export async function process(
|
||||
request: ProcessRequest,
|
||||
sendCallback: (payload: CallbackPayload) => Promise<void>
|
||||
): Promise<void> {
|
||||
const startTime = Date.now();
|
||||
const metaCore = new MetaCoreClient(request.metaCoreUrl);
|
||||
|
||||
try {
|
||||
const { cid, filePath, existingMeta } = request;
|
||||
|
||||
// Only process video files
|
||||
const fileType = existingMeta?.fileType;
|
||||
if (fileType !== 'video') {
|
||||
await sendCallback({
|
||||
taskId: request.taskId,
|
||||
status: 'skipped',
|
||||
duration: Date.now() - startTime,
|
||||
reason: 'Not a video file',
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
// Check if already processed
|
||||
if (existingMeta?.['fileinfo/duration']) {
|
||||
await sendCallback({
|
||||
taskId: request.taskId,
|
||||
status: 'skipped',
|
||||
duration: Date.now() - startTime,
|
||||
reason: 'Already processed',
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
// Run FFprobe
|
||||
const data = await runFFprobe(filePath);
|
||||
const metadata: Record<string, string> = {};
|
||||
|
||||
// Format metadata
|
||||
if (data.format?.duration != null) {
|
||||
metadata['fileinfo/duration'] = String(data.format.duration);
|
||||
}
|
||||
if (data.format?.format_name) {
|
||||
metadata['fileinfo/formatName'] = data.format.format_name;
|
||||
}
|
||||
|
||||
// Stream counters
|
||||
let videoIdx = 0, audioIdx = 0, subtitleIdx = 0;
|
||||
|
||||
for (const stream of data.streams || []) {
|
||||
if (!stream) continue;
|
||||
|
||||
const codecType = stream.codec_type || '';
|
||||
let basePath: string;
|
||||
|
||||
if (codecType === 'video') {
|
||||
basePath = `fileinfo/streamdetails/video/${videoIdx++}`;
|
||||
} else if (codecType === 'audio') {
|
||||
basePath = `fileinfo/streamdetails/audio/${audioIdx++}`;
|
||||
} else if (codecType === 'subtitle') {
|
||||
basePath = `fileinfo/streamdetails/subtitle/${subtitleIdx++}`;
|
||||
} else {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Common fields
|
||||
if (stream.codec_name) metadata[`${basePath}/codec`] = stream.codec_name;
|
||||
if (stream.index != null) metadata[`${basePath}/index`] = String(stream.index);
|
||||
if (stream.duration) metadata[`${basePath}/duration`] = stream.duration;
|
||||
if (stream.codec_type) metadata[`${basePath}/codecType`] = stream.codec_type;
|
||||
if (stream.disposition?.forced != null) {
|
||||
metadata[`${basePath}/forced`] = stream.disposition.forced ? 'true' : 'false';
|
||||
}
|
||||
if (stream.disposition?.default != null) {
|
||||
metadata[`${basePath}/default`] = stream.disposition.default ? 'true' : 'false';
|
||||
}
|
||||
if (stream.tags?.language) metadata[`${basePath}/language`] = stream.tags.language;
|
||||
if (stream.tags?.title) metadata[`${basePath}/title`] = stream.tags.title;
|
||||
|
||||
// Video-specific
|
||||
if (stream.width) metadata[`${basePath}/width`] = String(stream.width);
|
||||
if (stream.height) metadata[`${basePath}/height`] = String(stream.height);
|
||||
if (stream.bit_rate && stream.bit_rate !== 'N/A') {
|
||||
metadata[`${basePath}/bitrate`] = stream.bit_rate;
|
||||
}
|
||||
if (stream.avg_frame_rate && stream.avg_frame_rate !== '0/0') {
|
||||
metadata[`${basePath}/frameRate`] = stream.avg_frame_rate;
|
||||
}
|
||||
if (stream.tags?.rotate) metadata[`${basePath}/rotation`] = stream.tags.rotate;
|
||||
|
||||
// Audio-specific
|
||||
if (stream.sample_rate) metadata[`${basePath}/sampleRate`] = stream.sample_rate;
|
||||
if (stream.channel_layout) metadata[`${basePath}/channelLayout`] = stream.channel_layout;
|
||||
}
|
||||
|
||||
// Write to meta-core
|
||||
await metaCore.mergeMetadata(cid, metadata);
|
||||
|
||||
const duration = Date.now() - startTime;
|
||||
console.log(`[ffmpeg] Processed in ${duration}ms`);
|
||||
|
||||
await sendCallback({
|
||||
taskId: request.taskId,
|
||||
status: 'completed',
|
||||
duration,
|
||||
});
|
||||
} catch (error) {
|
||||
const duration = Date.now() - startTime;
|
||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||
console.error(`[ffmpeg] Error:`, errorMessage);
|
||||
|
||||
await sendCallback({
|
||||
taskId: request.taskId,
|
||||
status: 'failed',
|
||||
duration,
|
||||
error: errorMessage,
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
Reference in New Issue
Block a user