align plugin with new architecture
Build and Push Docker Image / build (push) Failing after 15s

This commit is contained in:
Worph
2026-09-13 11:52:46 +02:00
parent da127f5185
commit 4d51289acc
15 changed files with 2643 additions and 272 deletions
+170
View File
@@ -0,0 +1,170 @@
/**
* Language code normalisation — the `lang3` vocabulary meta-watch reads.
*
* Port of meta-watch's `MW_LANG.toLang3` (packages/meta-watch/ui/js/lang-codes.js),
* the same normaliser the `language` plugin copied. Every spelling folds to
* **ISO 639-2/B** — `fre` not `fra`, `ger` not `deu`, `chi` not `zho` — because
* that is what meta-watch writes (`subtitles/fre/<cid>`, routes/subtitles.rs)
* and what its readers compare against:
*
* ISO 639-1 "fr" -> fre
* ISO 639-2/B "fre" -> fre
* ISO 639-2/T "fra" -> fre (MP4 `mdhd` speaks T, Matroska speaks B)
* BCP-47 "pt-BR" -> por (primary subtag only)
*
* One deliberate difference: "no language" is `und`, not "" — the code becomes a
* key segment (`subtitles/und/<cid>`, METADATA_KEYS.md §8) and `und` is the
* registry's spelling for undetermined.
*
* Copied, not imported: plugins are independent submodules. This file is
* byte-identical in metamesh-plugin-subtitle and metamesh-plugin-subtitle-extractor.
*/
// Canonical table keyed by 639-2/B: display name (lowercased for filename
// matching) and the 639-1 code that folds into it.
const LANGS: Record<string, { name: string; a1?: string }> = {
eng: { name: 'english', a1: 'en' },
fre: { name: 'french', a1: 'fr' },
ger: { name: 'german', a1: 'de' },
spa: { name: 'spanish', a1: 'es' },
ita: { name: 'italian', a1: 'it' },
por: { name: 'portuguese', a1: 'pt' },
dut: { name: 'dutch', a1: 'nl' },
swe: { name: 'swedish', a1: 'sv' },
nor: { name: 'norwegian', a1: 'no' },
dan: { name: 'danish', a1: 'da' },
fin: { name: 'finnish', a1: 'fi' },
ice: { name: 'icelandic', a1: 'is' },
pol: { name: 'polish', a1: 'pl' },
cze: { name: 'czech', a1: 'cs' },
slo: { name: 'slovak', a1: 'sk' },
slv: { name: 'slovenian', a1: 'sl' },
hun: { name: 'hungarian', a1: 'hu' },
rum: { name: 'romanian', a1: 'ro' },
bul: { name: 'bulgarian', a1: 'bg' },
rus: { name: 'russian', a1: 'ru' },
ukr: { name: 'ukrainian', a1: 'uk' },
bel: { name: 'belarusian', a1: 'be' },
srp: { name: 'serbian', a1: 'sr' },
hrv: { name: 'croatian', a1: 'hr' },
bos: { name: 'bosnian', a1: 'bs' },
mac: { name: 'macedonian', a1: 'mk' },
alb: { name: 'albanian', a1: 'sq' },
gre: { name: 'greek', a1: 'el' },
tur: { name: 'turkish', a1: 'tr' },
heb: { name: 'hebrew', a1: 'he' },
ara: { name: 'arabic', a1: 'ar' },
per: { name: 'persian', a1: 'fa' },
urd: { name: 'urdu', a1: 'ur' },
hin: { name: 'hindi', a1: 'hi' },
ben: { name: 'bengali', a1: 'bn' },
tam: { name: 'tamil', a1: 'ta' },
tel: { name: 'telugu', a1: 'te' },
mal: { name: 'malayalam', a1: 'ml' },
kan: { name: 'kannada', a1: 'kn' },
mar: { name: 'marathi', a1: 'mr' },
guj: { name: 'gujarati', a1: 'gu' },
pan: { name: 'punjabi', a1: 'pa' },
nep: { name: 'nepali', a1: 'ne' },
sin: { name: 'sinhala', a1: 'si' },
tha: { name: 'thai', a1: 'th' },
lao: { name: 'lao', a1: 'lo' },
khm: { name: 'khmer', a1: 'km' },
bur: { name: 'burmese', a1: 'my' },
vie: { name: 'vietnamese', a1: 'vi' },
ind: { name: 'indonesian', a1: 'id' },
may: { name: 'malay', a1: 'ms' },
tgl: { name: 'tagalog', a1: 'tl' },
fil: { name: 'filipino' },
jpn: { name: 'japanese', a1: 'ja' },
kor: { name: 'korean', a1: 'ko' },
chi: { name: 'chinese', a1: 'zh' },
mon: { name: 'mongolian', a1: 'mn' },
kaz: { name: 'kazakh', a1: 'kk' },
uzb: { name: 'uzbek', a1: 'uz' },
aze: { name: 'azerbaijani', a1: 'az' },
geo: { name: 'georgian', a1: 'ka' },
arm: { name: 'armenian', a1: 'hy' },
est: { name: 'estonian', a1: 'et' },
lav: { name: 'latvian', a1: 'lv' },
lit: { name: 'lithuanian', a1: 'lt' },
cat: { name: 'catalan', a1: 'ca' },
glg: { name: 'galician', a1: 'gl' },
baq: { name: 'basque', a1: 'eu' },
wel: { name: 'welsh', a1: 'cy' },
gle: { name: 'irish', a1: 'ga' },
afr: { name: 'afrikaans', a1: 'af' },
swa: { name: 'swahili', a1: 'sw' },
amh: { name: 'amharic', a1: 'am' },
zul: { name: 'zulu', a1: 'zu' },
lat: { name: 'latin', a1: 'la' },
epo: { name: 'esperanto', a1: 'eo' },
yid: { name: 'yiddish', a1: 'yi' },
jav: { name: 'javanese', a1: 'jv' },
mul: { name: 'multiple' },
};
// ISO 639-2/T -> 639-2/B: the only codes where the two standards disagree.
const T2B: Record<string, string> = {
sqi: 'alb', hye: 'arm', eus: 'baq', mya: 'bur', ces: 'cze',
zho: 'chi', cym: 'wel', deu: 'ger', nld: 'dut', ell: 'gre', fas: 'per',
fra: 'fre', kat: 'geo', isl: 'ice', mkd: 'mac', msa: 'may',
ron: 'rum', slk: 'slo',
};
// "This track has no language" spellings.
const NONE = new Set(['und', 'zxx', 'mis', 'unknown', 'none']);
// 639-1 -> 639-2/B, from the table plus legacy two-letter codes.
const A1: Record<string, string> = { nb: 'nor', nn: 'nor', iw: 'heb', in: 'ind', ji: 'yid', jw: 'jav' };
const NAMES: Record<string, string> = {};
for (const [b, entry] of Object.entries(LANGS)) {
if (entry.a1) A1[entry.a1] = b;
NAMES[entry.name] = b;
}
/**
* franc-min answers in ISO 639-3 individual-language codes; a few of them are
* not 639-2 codes at all (Mandarin `cmn` is 639-2 `chi`). Folded before toLang3.
*/
const FRANC_TO_639_2: Record<string, string> = {
cmn: 'chi', arb: 'ara', pes: 'per', swh: 'swa', uzn: 'uzb', zlm: 'may',
npi: 'nep', azj: 'aze', plt: 'mlg', pbu: 'pus', ckb: 'kur', fuv: 'ful', qug: 'que',
};
/** Canonical lang3 (ISO 639-2/B) for any code; `und` when there is none. */
export function toLang3(code: unknown): string {
const s = String(code ?? '').trim().toLowerCase();
if (!s) return 'und';
if (NAMES[s]) return NAMES[s];
const p = s.split(/[-_]/)[0];
if (NONE.has(p)) return 'und';
if (p.length === 2) return A1[p] ?? 'und';
if (p.length === 3 && /^[a-z]{3}$/.test(p)) return T2B[p] ?? p;
return 'und';
}
/** lang3 for a franc-min result (`und` stays `und`). */
export function francToLang3(code: string): string {
return toLang3(FRANC_TO_639_2[code] ?? code);
}
/**
* Strict variant for filename tokens (`video.fr.srt`, `video.English.srt`):
* only a token that is recognisably a language resolves, so `S01`, `x264` or
* `Extended` never read as one. Returns null for anything else.
*/
export function languageToken(token: string): string | null {
const t = token.trim().toLowerCase();
if (!t) return null;
if (NAMES[t]) return NAMES[t];
if (t === 'und') return 'und';
const p = t.split(/[-_]/)[0];
if (p !== t && !/^[a-z]{2,3}[-_][a-z0-9]{2,8}$/.test(t)) return null;
if (p.length === 2) return A1[p] ?? null;
if (p.length === 3) {
if (LANGS[p]) return p;
if (T2B[p]) return T2B[p];
}
return null;
}
+39 -64
View File
@@ -1,6 +1,10 @@
/**
* meta-core API Client for writing metadata
* Fails gracefully if meta-core is unavailable (logs warning, returns without throwing)
* meta-core API client for writing metadata.
*
* Writes report success so a failed write fails the task instead of being
* swallowed. Key-sets are written as individual `prefix/member = "true"`
* fields through PATCH — never via `_add`, which stores one comma-joined
* string and can never produce the key-set shape.
*/
export class MetaCoreClient {
@@ -8,86 +12,57 @@ export class MetaCoreClient {
private async safeFetch(url: string, options: RequestInit): Promise<Response | null> {
try {
return await fetch(url, { ...options, signal: AbortSignal.timeout(5000) });
return await fetch(url, { ...options, signal: AbortSignal.timeout(10000) });
} catch (error) {
console.warn(`[MetaCoreClient] Warning: meta-core unavailable at ${this.baseUrl}`);
console.warn(`[MetaCoreClient] Warning: meta-core unavailable at ${this.baseUrl}: ${error instanceof Error ? error.message : String(error)}`);
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;
if (!response || !response.ok) return null;
const data = await response.json() as { value?: unknown };
return data.value === undefined || data.value === null ? null : String(data.value);
}
async mergeMetadata(hashId: string, metadata: Record<string, string>): Promise<void> {
/**
* Whole record, in whatever form the endpoint serves (meta-sort's /meta is
* nested, meta-core's is flat) — callers flatten. `null` when the call
* itself failed, `{}` when the record does not exist.
*/
async getMetadata(hashId: string): Promise<Record<string, unknown> | null> {
const response = await this.safeFetch(`${this.baseUrl}/meta/${hashId}`, { method: 'GET' });
if (!response) return null;
if (response.status === 404) return {};
if (!response.ok) return null;
const data = await response.json() as { metadata?: Record<string, unknown> };
return data.metadata ?? {};
}
/** PATCH — merge `metadata` into the record. True on success. */
async mergeMetadata(hashId: string, metadata: Record<string, string>): Promise<boolean> {
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}`);
if (!response) return false;
if (!response.ok) {
console.warn(`[MetaCoreClient] Failed to merge metadata into ${hashId}: ${response.status}`);
return false;
}
return true;
}
async deleteProperty(hashId: string, key: string): Promise<void> {
/** DELETE one field. A missing field counts as deleted. */
async deleteProperty(hashId: string, key: string): Promise<boolean> {
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;
if (!response) return false;
if (!response.ok && response.status !== 404) {
console.warn(`[MetaCoreClient] Failed to delete ${hashId}/${key}: ${response.status}`);
return false;
}
return true;
}
}
+65
View File
@@ -0,0 +1,65 @@
/**
* Reading a record regardless of the shape it arrives in, and writing only what
* is new.
*
* meta-sort's scheduler fills `existingMeta` from meta-sort's own GET /meta/{hash},
* which returns the NESTED document (`fileinfo: {duration: 1433.089}`,
* `stream: [...]`, a key-set as `subtitles: {fre: {<cid>: true}}`). meta-core's
* backend serves the FLAT form (`fileinfo/duration`, `stream/0`,
* `subtitles/fre/<cid>`). A plugin that reads `existingMeta['x/y']` silently sees
* nothing in production — so everything here goes through `flattenMeta`.
*
* Byte-identical in metamesh-plugin-subtitle and metamesh-plugin-subtitle-extractor.
*/
/** Flatten either form into `a/b/c -> string` (meta-sort's flattenMetadata, plus array indices). */
export function flattenMeta(meta: Record<string, unknown> | undefined | null): Record<string, string> {
const out: Record<string, string> = {};
const walk = (value: unknown, key: string) => {
if (value === null || value === undefined) return;
if (Array.isArray(value)) {
value.forEach((v, i) => walk(v, key ? `${key}/${i}` : String(i)));
} else if (typeof value === 'object') {
for (const [k, v] of Object.entries(value as Record<string, unknown>)) {
walk(v, key ? `${key}/${k}` : k);
}
} else if (key) {
out[key] = String(value);
}
};
walk(meta ?? {}, '');
return out;
}
/** The subset of `wanted` the record does not already hold with that exact value. */
export function onlyNewKeys(existingFlat: Record<string, string>, wanted: Record<string, string>): Record<string, string> {
const out: Record<string, string> = {};
for (const [k, v] of Object.entries(wanted)) {
if (existingFlat[k] !== v) out[k] = v;
}
return out;
}
/**
* Members of a LEGACY scalar csv-set (`subtitles = "cidA,cidB"`, written by the
* old `_add` path), or null when the field is absent or already a key-set.
*
* Such a scalar must be deleted before any `field/...` key is written: meta-sort
* rebuilds its nested document by walking the `/`-segments, and a string sitting
* where an object is needed either throws or silently hides the key-set.
*/
export function legacyCsvMembers(meta: Record<string, unknown> | undefined | null, field: string): string[] | null {
const v = meta?.[field];
if (typeof v !== 'string' && typeof v !== 'number') return null;
return String(v).split(/[,|]/).map((s) => s.trim()).filter(Boolean);
}
/** Every `<prefix>/<lang>/<cid>` leaf for `cid` whose language is not `lang3`. */
export function staleLanguageLeaves(existingFlat: Record<string, string>, prefix: string, cid: string, lang3: string): string[] {
const stale: string[] = [];
for (const key of Object.keys(existingFlat)) {
const parts = key.split('/');
if (parts.length === 3 && parts[0] === prefix && parts[2] === cid && parts[1] !== lang3) stale.push(key);
}
return stale;
}
+433 -135
View File
@@ -1,37 +1,78 @@
/**
* Subtitle Plugin
*
* For subtitle files: finds sibling video files and detects language.
* For video files: finds sibling subtitle files.
* Links subtitle SIDECAR files to the video they sit next to, in both
* directions, and determines each sidecar's language.
*
* Matches old SubtitleProcessor output:
* - For subtitle files: videos (add CID), subtitleLanguage
* - For video files: subtitles (add CID)
* ============================================================================
* WHAT IT WRITES (METADATA_KEYS.md §8)
* ============================================================================
* On the video record, per sidecar `<subCid>` (midhash256) in `<lang3>`
* (ISO 639-2/B, `und` when undetermined):
* subtitles/<lang3>/<subCid> = "true" what meta-watch lists
* subtitleLanguages/<lang3> = "true" (not for `und`)
* On the sidecar's own record:
* videos/<videoCid> = "true"
* subtitleLanguage = <lang3> (not for `und`)
*
* Whichever of the two records is processed first, the result is the same: each
* run writes its own record in full, and the other record only when it already
* exists (no stub records carrying nothing but subtitle keys). Writes are one
* PATCH of only-new keys; a failed write fails the task.
*
* ============================================================================
* WHAT COUNTS AS A SIDECAR
* ============================================================================
* `<video basename>.<ext>` or `<video basename>.<qualifiers>.<ext>`, compared
* case-insensitively, `<ext>` in srt/ass/ssa/vtt, where every qualifier token is
* a language (`en`, `eng`, `fre`, `pt-BR`, `English`) or a flag (`forced`, `sdh`,
* `cc`, `hi`, `default`, `full`, `signs`, `songs`). So `Movie.fr.forced.srt`
* belongs to `Movie.mkv`, but `Movie.Extended.eng.srt` does not.
* Language: the qualifier token when there is one, else a franc-min sniff of the
* subtitle text (timecodes and ASS markup stripped), else `und`. `hi` is read as
* the hearing-impaired flag, not Hindi (use `hin`).
*
* ============================================================================
* FILE ACCESS - WebDAV only
* ============================================================================
* The container has no /files mount. Directory listings, size and bytes all come
* over WebDAV from the meta-core that drove the /process call (its /urls ->
* webdavUrlInternal; WEBDAV_URL overrides).
*/
import { readFile, access, open } from 'fs/promises';
import { dirname, extname } from 'path';
import { posix } from 'path';
import { createHash } from 'crypto';
import { franc } from 'franc-min';
import { FileType, getSiblingFiles } from '@metazla/filename-tools';
import { FileType } from '@metazla/filename-tools';
import type { PluginManifest, ProcessRequest, CallbackPayload } from './types.js';
import { MetaCoreClient } from './meta-core-client.js';
import { createWebDAVClient, WebDAVClient } from './webdav-client.js';
import { getWebDAVClient, WebDAVClient, DirectoryEntry } from './webdav-client.js';
import { languageToken, francToLang3 } from './langs.js';
import { flattenMeta, onlyNewKeys, legacyCsvMembers, staleLanguageLeaves } from './meta-shape.js';
// Initialize WebDAV client if WEBDAV_URL is set
const webdavClient = createWebDAVClient();
if (webdavClient) {
console.log('[subtitle] Using WebDAV for file access');
} else {
console.log('[subtitle] Using direct filesystem access');
}
const fileTypeTool = new FileType();
const fileType = new FileType();
export const SIDECAR_EXTENSIONS = new Set(['srt', 'ass', 'ssa', 'vtt']);
/** Bytes of subtitle text read for language detection (ASS style headers can run several KB). */
const DETECT_BYTES = 64 * 1024;
const FLAG_TOKENS: Record<string, 'forced' | 'hearingImpaired' | 'other'> = {
forced: 'forced',
sdh: 'hearingImpaired',
cc: 'hearingImpaired',
hi: 'hearingImpaired',
default: 'other',
full: 'other',
signs: 'other',
songs: 'other',
};
export const manifest: PluginManifest = {
id: 'subtitle',
name: 'Subtitle Processor',
version: '1.0.0',
description: 'Processes subtitle files and links them to videos',
version: '1.1.0',
description: 'Links subtitle sidecar files to their videos and detects their language',
author: 'MetaMesh',
dependencies: ['file-info'],
priority: 45,
@@ -40,40 +81,362 @@ export const manifest: PluginManifest = {
timeout: 30000,
schema: {
subtitleLanguage: { label: 'Subtitle Language', type: 'string', readonly: true },
videos: { label: 'Linked Videos', type: 'array', readonly: true },
subtitles: { label: 'Linked Subtitles', type: 'array', readonly: true },
videos: { label: 'Linked Videos', type: 'json', readonly: true, hint: 'videos/<cid>' },
subtitles: { label: 'Linked Subtitles', type: 'json', readonly: true, hint: 'subtitles/<lang3>/<cid>' },
},
config: {},
};
async function fileExists(path: string): Promise<boolean> {
if (webdavClient) {
return webdavClient.exists(path);
// ----------------------------------------------------------------------------
// Names
// ----------------------------------------------------------------------------
export function splitExtension(fileName: string): { base: string; ext: string } {
const dot = fileName.lastIndexOf('.');
if (dot <= 0) return { base: fileName, ext: '' };
return { base: fileName.slice(0, dot), ext: fileName.slice(dot + 1).toLowerCase() };
}
export interface SubtitleQualifiers {
lang3?: string;
forced: boolean;
hearingImpaired: boolean;
}
/** Qualifier tokens (`fr`, `forced`, …); null as soon as one is neither a language nor a flag. */
export function parseQualifierTokens(tokens: string[]): SubtitleQualifiers | null {
const q: SubtitleQualifiers = { forced: false, hearingImpaired: false };
for (const raw of tokens) {
const token = raw.trim().toLowerCase();
if (!token) continue;
const flag = FLAG_TOKENS[token];
if (flag === 'forced') q.forced = true;
else if (flag === 'hearingImpaired') q.hearingImpaired = true;
else if (flag !== 'other') {
const lang = languageToken(token);
if (!lang) return null;
q.lang3 ??= lang;
}
}
try {
await access(path);
return true;
} catch {
return false;
return q;
}
/** Is `candidate` a sidecar of `videoFileName`, and what do its qualifiers say? */
export function parseSidecarName(videoFileName: string, candidate: string): (SubtitleQualifiers & { ext: string }) | null {
const { ext } = splitExtension(candidate);
if (!SIDECAR_EXTENSIONS.has(ext)) return null;
const videoBase = splitExtension(videoFileName).base.toLowerCase();
const base = splitExtension(candidate).base;
if (!videoBase) return null;
if (base.toLowerCase() === videoBase) return { ext, forced: false, hearingImpaired: false };
if (!base.toLowerCase().startsWith(`${videoBase}.`)) return null;
const q = parseQualifierTokens(base.slice(videoBase.length + 1).split('.'));
return q ? { ...q, ext } : null;
}
/** Qualifiers read off the end of a subtitle's own name, when no sibling video anchors it. */
export function trailingQualifiers(fileName: string): SubtitleQualifiers {
const tokens = splitExtension(fileName).base.split('.');
const q: SubtitleQualifiers = { forced: false, hearingImpaired: false };
// Keep at least one token as the name proper.
for (let i = tokens.length - 1; i >= 1; i--) {
const one = parseQualifierTokens([tokens[i]]);
if (!one) break;
q.forced ||= one.forced;
q.hearingImpaired ||= one.hearingImpaired;
// Walking backwards: the token nearest the name wins, as in parseQualifierTokens.
if (one.lang3) q.lang3 = one.lang3;
}
return q;
}
// ----------------------------------------------------------------------------
// Language sniffing
// ----------------------------------------------------------------------------
/** The spoken text of a subtitle file: no counters, timecodes, headers or markup. */
export function cleanSubtitleText(raw: string, ext: string): string {
const lines = raw.replace(/^\uFEFF/, '').split(/\r?\n/);
let text: string[];
if (ext === 'ass' || ext === 'ssa') {
text = lines
.filter((l) => l.startsWith('Dialogue:'))
.map((l) => l.split(',').slice(9).join(','))
.map((l) => l.replace(/\{[^}]*\}/g, '').replace(/\\[Nnh]/g, ' '));
} else {
text = lines.filter((l) => {
const t = l.trim();
return t && !/^\d+$/.test(t) && !t.includes('-->') && !/^WEBVTT/.test(t) && !/^(NOTE|STYLE|REGION)\b/.test(t);
});
}
return text
.join(' ')
.replace(/<[^>]+>/g, ' ')
.replace(/\{[^}]*\}/g, ' ')
.replace(/\s+/g, ' ')
.trim();
}
/** lang3 from subtitle text, `und` when there is too little to tell. */
export function detectLanguageFromText(text: string): string {
if (text.length < 20) return 'und';
const code = franc(text);
return code === 'und' ? 'und' : francToLang3(code);
}
// ----------------------------------------------------------------------------
// CIDs
// ----------------------------------------------------------------------------
const MIDHASH_SAMPLE_SIZE = 1024 * 1024;
/** midhash256 CID from a file's size and its middle sample (matches meta-hash). */
export function computeMidHash256(fileSize: number, sampleData: Buffer): string {
const MIDHASH_VARINT = Buffer.from([0x80, 0x20]);
const sizeBuffer = Buffer.allocUnsafe(8);
sizeBuffer.writeBigUInt64BE(BigInt(fileSize), 0);
const hashBuffer = createHash('sha256').update(Buffer.concat([sizeBuffer, sampleData])).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;
}
export function midhashSampleRange(fileSize: number): { start: number; end: number } {
if (fileSize <= MIDHASH_SAMPLE_SIZE) return { start: 0, end: fileSize - 1 };
const start = Math.floor((fileSize - MIDHASH_SAMPLE_SIZE) / 2);
return { start, end: start + MIDHASH_SAMPLE_SIZE - 1 };
}
export function computeMidHash256FromBuffer(data: Buffer): string {
const { start, end } = midhashSampleRange(data.length);
return computeMidHash256(data.length, data.subarray(start, end + 1));
}
/** CID of a file over WebDAV: one Range read of at most 1 MiB, however large the file. */
export async function midhashOverWebDAV(client: WebDAVClient, filePath: string, size?: number): Promise<string> {
const fileSize = size ?? (await client.stat(filePath)).size;
const { start, end } = midhashSampleRange(fileSize);
const sample = fileSize === 0 ? Buffer.alloc(0) : await client.readBytes(filePath, start, end);
if (sample.length !== end - start + 1) {
throw new Error(`short read for ${filePath}: wanted ${end - start + 1} bytes, got ${sample.length}`);
}
return computeMidHash256(fileSize, sample);
}
// ----------------------------------------------------------------------------
// Record keys
// ----------------------------------------------------------------------------
export function videoSubtitleKeys(subCid: string, lang3: string): Record<string, string> {
const keys: Record<string, string> = { [`subtitles/${lang3}/${subCid}`]: 'true' };
if (lang3 !== 'und') keys[`subtitleLanguages/${lang3}`] = 'true';
return keys;
}
export function subtitleRecordKeys(lang3: string, videoCids: string[]): Record<string, string> {
const keys: Record<string, string> = {};
for (const v of videoCids) keys[`videos/${v}`] = 'true';
if (lang3 !== 'und') keys.subtitleLanguage = lang3;
return keys;
}
async function detectFromFile(client: WebDAVClient, filePath: string, ext: string, size?: number): Promise<string> {
if (size === 0) return 'und';
const end = size !== undefined ? Math.min(size, DETECT_BYTES) - 1 : DETECT_BYTES - 1;
const bytes = await client.readBytes(filePath, 0, end);
return detectLanguageFromText(cleanSubtitleText(bytes.toString('utf8'), ext));
}
/**
* Read first N bytes of a file
* Link `subCid`/`lang3` onto a video record that is not the one being processed:
* only if it exists, only new keys, and with any other-language leaf for the
* same CID removed (one file, one language).
*/
async function readFirstData(filePath: string, encoding: BufferEncoding = 'utf8', size = 1024): Promise<string> {
if (webdavClient) {
const buffer = await webdavClient.readBytes(filePath, 0, size - 1);
return buffer.toString(encoding);
async function linkOntoVideo(metaCore: MetaCoreClient, videoCid: string, subCid: string, lang3: string): Promise<boolean> {
const current = await metaCore.getMetadata(videoCid);
if (current === null) return false;
const flat = flattenMeta(current);
if (!flat.fileType) return true; // not indexed yet — its own run links it
return writeVideoLinks(metaCore, videoCid, current, flat, [{ subCid, lang3 }]);
}
async function writeVideoLinks(
metaCore: MetaCoreClient,
videoCid: string,
meta: Record<string, unknown>,
flat: Record<string, string>,
links: Array<{ subCid: string; lang3: string }>,
): Promise<boolean> {
const deletes: string[] = [];
const sets: Record<string, string> = {};
// Legacy scalar csv-sets from the `_add` era. Members still found on disk are
// re-linked with their language below; the rest are kept as `und` leaves.
const legacySubs = legacyCsvMembers(meta, 'subtitles');
if (legacySubs) {
deletes.push('subtitles');
for (const cid of legacySubs) {
if (!links.some((l) => l.subCid === cid)) sets[`subtitles/und/${cid}`] = 'true';
}
}
const handle = await open(filePath, 'r');
try {
const buffer = Buffer.alloc(size);
await handle.read(buffer, 0, size, 0);
return buffer.toString(encoding);
} finally {
await handle.close();
const legacyLangs = legacyCsvMembers(meta, 'subtitleLanguages');
if (legacyLangs) {
deletes.push('subtitleLanguages');
for (const code of legacyLangs) {
const l = languageToken(code);
if (l && l !== 'und') sets[`subtitleLanguages/${l}`] = 'true';
}
}
for (const { subCid, lang3 } of links) deletes.push(...staleLanguageLeaves(flat, 'subtitles', subCid, lang3));
for (const key of deletes) {
if (!(await metaCore.deleteProperty(videoCid, key))) return false;
delete flat[key];
}
let wanted: Record<string, string> = { ...sets };
for (const { subCid, lang3 } of links) wanted = { ...wanted, ...videoSubtitleKeys(subCid, lang3) };
const fresh = onlyNewKeys(flat, wanted);
return Object.keys(fresh).length === 0 || metaCore.mergeMetadata(videoCid, fresh);
}
async function writeSubtitleRecord(
metaCore: MetaCoreClient,
subCid: string,
meta: Record<string, unknown>,
lang3: string,
videoCids: string[],
): Promise<boolean> {
const flat = flattenMeta(meta);
const sets: Record<string, string> = {};
const legacyVideos = legacyCsvMembers(meta, 'videos');
if (legacyVideos) {
if (!(await metaCore.deleteProperty(subCid, 'videos'))) return false;
delete flat.videos;
for (const v of legacyVideos) sets[`videos/${v}`] = 'true';
}
const fresh = onlyNewKeys(flat, { ...sets, ...subtitleRecordKeys(lang3, videoCids) });
return Object.keys(fresh).length === 0 || metaCore.mergeMetadata(subCid, fresh);
}
// ----------------------------------------------------------------------------
// Task
// ----------------------------------------------------------------------------
type Outcome = { ok: true; summary: string } | { ok: false; error: string };
/** Video record: find its sidecars, link them. */
export async function processVideoFile(
client: WebDAVClient,
metaCore: MetaCoreClient,
videoCid: string,
filePath: string,
meta: Record<string, unknown>,
): Promise<Outcome> {
const dir = posix.dirname(filePath);
const videoName = posix.basename(filePath);
const entries = await client.list(dir);
const links: Array<{ subCid: string; lang3: string; entry: DirectoryEntry }> = [];
for (const entry of entries) {
if (entry.type !== 'file' || entry.size === 0) continue;
const q = parseSidecarName(videoName, entry.name);
if (!q) continue;
const subPath = `${dir}/${entry.name}`;
const subCid = await midhashOverWebDAV(client, subPath, entry.size);
const lang3 = q.lang3 ?? (await detectFromFile(client, subPath, q.ext, entry.size));
links.push({ subCid, lang3, entry });
}
const flat = flattenMeta(meta);
if (!(await writeVideoLinks(metaCore, videoCid, meta, flat, links))) {
return { ok: false, error: 'meta-core write failed (video record)' };
}
for (const { subCid, lang3 } of links) {
const subMeta = await metaCore.getMetadata(subCid);
if (subMeta === null) return { ok: false, error: `meta-core read failed (${subCid})` };
if (!flattenMeta(subMeta).fileType) continue; // not indexed yet — its own run links it
if (!(await writeSubtitleRecord(metaCore, subCid, subMeta, lang3, [videoCid]))) {
return { ok: false, error: `meta-core write failed (${subCid})` };
}
}
return { ok: true, summary: `${links.length} sidecar(s) [${links.map((l) => l.lang3).join(', ')}]` };
}
/** Subtitle record: find the video(s) it belongs to, determine its language, link. */
export async function processSubtitleFile(
client: WebDAVClient,
metaCore: MetaCoreClient,
subCid: string,
filePath: string,
meta: Record<string, unknown>,
): Promise<Outcome> {
const dir = posix.dirname(filePath);
const subName = posix.basename(filePath);
const ext = splitExtension(subName).ext;
// Plugin output folders (subtitle-extractor, opensubtitles, meta-watch uploads)
// hold no videos; their names already end in `.<lang3>.<ext>`.
const videos: DirectoryEntry[] = [];
if (!dir.startsWith('/files/plugin/') && SIDECAR_EXTENSIONS.has(ext)) {
for (const entry of await client.list(dir)) {
if (entry.type !== 'file') continue;
if (fileTypeTool.getFileTypeFromExtension(entry.name) !== 'video') continue;
if (parseSidecarName(entry.name, subName)) videos.push(entry);
}
}
// Same qualifiers the video-side run derives, so both runs agree on the language.
const q = videos.length > 0 ? parseSidecarName(videos[0].name, subName)! : trailingQualifiers(subName);
const size = meta.sizeByte !== undefined && Number.isFinite(Number(meta.sizeByte)) ? Number(meta.sizeByte) : undefined;
const lang3 = q.lang3 ?? (await detectFromFile(client, filePath, ext, size));
const videoCids: string[] = [];
for (const v of videos) videoCids.push(await midhashOverWebDAV(client, `${dir}/${v.name}`, v.size));
if (!(await writeSubtitleRecord(metaCore, subCid, meta, lang3, videoCids))) {
return { ok: false, error: 'meta-core write failed (subtitle record)' };
}
for (const videoCid of videoCids) {
if (!(await linkOntoVideo(metaCore, videoCid, subCid, lang3))) {
return { ok: false, error: `meta-core write failed (${videoCid})` };
}
}
return { ok: true, summary: `lang=${lang3}, ${videoCids.length} video(s)` };
}
/**
* A subtitle file the byte classifier could not place. file-info returns
* `undefined` for a `text/plain` sniff (METADATA_KEYS.md `fileType`: "could be a
* document, a subtitle or anything") and has been seen writing `other` for
* extracted `.ass` files — either way the `fileType === 'subtitle'` gate alone
* would never let this plugin see the file. The extension decides then.
*/
export function isUnclassifiedSubtitle(meta: Record<string, unknown>, filePath: string): boolean {
const fileType = meta.fileType;
if (fileType !== undefined && fileType !== null && fileType !== '' && fileType !== 'other' && fileType !== 'undefined') {
return false;
}
const ext = typeof meta.extension === 'string' && meta.extension
? meta.extension.toLowerCase()
: splitExtension(posix.basename(filePath)).ext;
return SIDECAR_EXTENSIONS.has(ext);
}
export async function process(
@@ -81,104 +444,39 @@ export async function process(
sendCallback: (payload: CallbackPayload) => Promise<void>
): Promise<void> {
const startTime = Date.now();
const metaCore = new MetaCoreClient(request.metaCoreUrl);
const { taskId, cid, filePath } = request;
const meta = (request.existingMeta ?? {}) as Record<string, unknown>;
const finish = (status: CallbackPayload['status'], extra: { reason?: string; error?: string } = {}) =>
sendCallback({ taskId, status, duration: Date.now() - startTime, ...extra });
try {
const { cid, filePath, existingMeta } = request;
const currentFileType = existingMeta?.fileType;
if (currentFileType === 'subtitle') {
await processSubtitleFile(metaCore, cid, filePath);
} else if (currentFileType === 'video') {
await processVideoFile(metaCore, cid, filePath, existingMeta);
const fileType = meta.fileType === 'subtitle' || isUnclassifiedSubtitle(meta, filePath) ? 'subtitle' : meta.fileType;
if (fileType !== 'video' && fileType !== 'subtitle') {
await finish('skipped', { reason: 'Not a video or subtitle file' });
return;
}
await sendCallback({
taskId: request.taskId,
status: 'completed',
duration: Date.now() - startTime,
});
const client = await getWebDAVClient(request.metaCoreUrl);
if (!client) {
console.error(`[subtitle] ${filePath}: no WebDAV endpoint available`);
await finish('failed', { error: 'No WebDAV endpoint available' });
return;
}
const metaCore = new MetaCoreClient(request.metaCoreUrl);
const outcome = fileType === 'subtitle'
? await processSubtitleFile(client, metaCore, cid, filePath, meta)
: await processVideoFile(client, metaCore, cid, filePath, meta);
if (!outcome.ok) {
console.error(`[subtitle] ${filePath}: ${outcome.error}`);
await finish('failed', { error: outcome.error });
return;
}
console.log(`[subtitle] ${filePath}: ${outcome.summary}`);
await finish('completed');
} catch (error) {
await sendCallback({
taskId: request.taskId,
status: 'failed',
duration: Date.now() - startTime,
error: error instanceof Error ? error.message : String(error),
});
}
}
/**
* Process subtitle file: find sibling videos and detect language
*/
async function processSubtitleFile(
metaCore: MetaCoreClient,
cid: string,
filePath: string
): Promise<void> {
try {
// Find sibling video files
const siblings = await getSiblingFiles(filePath);
for (const siblingPath of siblings) {
const siblingType = fileType.getFileTypeFromExtension(siblingPath);
if (siblingType === 'video') {
try {
const videoCid = await metaCore.computeFileCID(siblingPath);
if (videoCid) {
await metaCore.addToSet(cid, 'videos', videoCid);
}
} catch (e) {
console.debug(`[subtitle] Could not get CID for video: ${siblingPath}`);
}
}
}
// Detect subtitle language by reading first 4KB
try {
const content = await readFirstData(filePath, 'utf8', 4096);
const detectedLang = franc(content);
if (detectedLang && detectedLang !== 'und') {
await metaCore.setProperty(cid, 'subtitleLanguage', detectedLang);
}
} catch (e) {
console.debug(`[subtitle] Could not detect language for: ${filePath}`);
}
const mode = webdavClient ? 'WebDAV' : 'filesystem';
console.log(`[subtitle] Processed subtitle file: ${filePath} (${mode})`);
} catch (error) {
console.error(`[subtitle] Error processing subtitle file ${filePath}:`, error);
}
}
/**
* Process video file: find sibling subtitle files
*/
async function processVideoFile(
metaCore: MetaCoreClient,
cid: string,
filePath: string,
existingMeta?: Record<string, string>
): Promise<void> {
try {
const extension = existingMeta?.extension || '';
const srtPath = filePath.replace(new RegExp(`\\.${extension}$`, 'i'), '.srt');
// Check for .srt subtitle
if (await fileExists(srtPath)) {
try {
const subtitleCid = await metaCore.computeFileCID(srtPath);
if (subtitleCid) {
await metaCore.addToSet(cid, 'subtitles', subtitleCid);
}
} catch (e) {
console.debug(`[subtitle] Could not get CID for subtitle: ${srtPath}`);
}
}
console.log(`[subtitle] Checked subtitles for video: ${filePath}`);
} catch (error) {
console.error(`[subtitle] Error finding subtitles for ${filePath}:`, error);
console.error(`[subtitle] ${filePath}:`, error);
await finish('failed', { error: error instanceof Error ? error.message : String(error) });
}
}
+3 -1
View File
@@ -44,7 +44,9 @@ export interface ProcessRequest {
filePath: string;
callbackUrl: string;
metaCoreUrl: string;
existingMeta?: Record<string, string>;
// meta-sort sends its NESTED /meta document (`fileinfo: {...}`, `stream: [...]`);
// meta-core serves the flat `a/b` form. Top-level scalars match in both.
existingMeta?: Record<string, unknown>;
}
export interface ProcessResponse {
+97 -20
View File
@@ -64,7 +64,7 @@ export class WebDAVClient {
}
/**
* Read first N bytes of a file (for language detection)
* 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> {
@@ -86,7 +86,7 @@ export class WebDAVClient {
}
/**
* Read entire file as buffer (for small files)
* Read entire file as buffer (for small files only)
*/
async readFile(filePath: string): Promise<Buffer> {
const url = this.toWebDAVUrl(filePath);
@@ -101,6 +101,25 @@ export class WebDAVClient {
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
*/
@@ -116,38 +135,96 @@ export class WebDAVClient {
}
/**
* List directory contents via HTTP GET (returns JSON from nginx autoindex)
* List a directory: GET <dir>/ with Accept: application/json, which both
* meta-core's WebDAV and meta-sort's nginx autoindex answer with
* [{name, type: "file"|"directory", size, mtime}].
*
* (Addition to the tmdb/still-extractor copy of this file: the subtitle
* plugin has no /files mount, so sibling discovery must go over WebDAV.)
*/
async readdir(dirPath: string): Promise<string[]> {
async list(dirPath: string): Promise<DirectoryEntry[]> {
const url = this.toWebDAVUrl(dirPath);
const finalUrl = url.endsWith('/') ? url : url + '/';
const response = await fetch(finalUrl, {
method: 'GET',
headers: {
'Accept': 'application/json',
},
});
const response = await fetch(finalUrl, { method: 'GET', headers: { 'Accept': 'application/json' } });
if (!response.ok) {
throw new Error(`WebDAV GET (dir) failed for ${dirPath}: ${response.status} ${response.statusText}`);
}
const entries = await response.json() as Array<{ name: string; type: string }>;
return entries.map(e => e.name);
const entries = await response.json() as Array<{ name: string; type: string; size?: number }>;
return entries.map((e) => ({
name: e.name,
type: e.type === 'directory' ? 'directory' : 'file',
size: typeof e.size === 'number' ? e.size : undefined,
}));
}
}
export interface DirectoryEntry {
name: string;
type: 'file' | 'directory';
size?: number;
}
/**
* Create a WebDAV client from environment variables
* 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.
*/
export function createWebDAVClient(): WebDAVClient | null {
const webdavUrl = process.env.WEBDAV_URL;
const clientCache = new Map<string, Promise<WebDAVClient | null>>();
if (!webdavUrl) {
console.warn('[webdav-client] WEBDAV_URL not set, WebDAV client unavailable');
return 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}`);
}
return new WebDAVClient(webdavUrl);
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;
}
}