From 237530321e51ae3389f283fc940d52f240449498 Mon Sep 17 00:00:00 2001 From: MasutaYunikon Date: Wed, 11 Mar 2026 00:35:41 +0100 Subject: [PATCH] feat: make the organizer threaded to not consumme api main thread --- server/organize-worker.ts | 237 +++++++++++++++++++++++++++ server/organize.ts | 331 +++++++++++--------------------------- 2 files changed, 327 insertions(+), 241 deletions(-) create mode 100644 server/organize-worker.ts diff --git a/server/organize-worker.ts b/server/organize-worker.ts new file mode 100644 index 0000000..c32c1e1 --- /dev/null +++ b/server/organize-worker.ts @@ -0,0 +1,237 @@ +/** + * organize-worker.ts + * ================== + * Worker thread dédié à l'organisation des fichiers. + * Tourne dans un thread séparé pour ne pas bloquer le event loop Express. + * + * Messages reçus (parentPort.on) : { type: 'run', torrents: TorrentInfo[] } + * Messages envoyés (parentPort.postMessage) : + * { type: 'log', level: 'info'|'warn'|'error', msg: string } + * { type: 'result', hash: string, name: string, done, skipped, errors } + * { type: 'done' } + */ + +import fs from 'fs' +import fsp from 'fs/promises' +import path from 'path' +import { parentPort } from 'worker_threads' + +const TORRENTS_PATH = path.join(process.cwd(), 'data', 'torrent_final.json') +const ORGANIZED_PATH = path.join(process.cwd(), 'data', 'organized.json') + +// ─── Utils log ──────────────────────────────────────────────── + +function log(msg: string) { parentPort?.postMessage({ type: 'log', level: 'info', msg }) } +function warn(msg: string) { parentPort?.postMessage({ type: 'log', level: 'warn', msg }) } +function error(msg: string) { parentPort?.postMessage({ type: 'log', level: 'error', msg }) } + +// ─── Settings (lu directement, pas d'import circulaire) ────── + +function readSettings(): { mediaPath: string; completePath: string; organizeMode: string } { + try { + const p = path.join(process.cwd(), 'data', 'settings.json') + if (!fs.existsSync(p)) return { mediaPath: '/media/Kai', completePath: '/downloads/complete', organizeMode: 'hardlink' } + return JSON.parse(fs.readFileSync(p, 'utf-8')) + } catch { return { mediaPath: '/media/Kai', completePath: '/downloads/complete', organizeMode: 'hardlink' } } +} + +// ─── Organized log ──────────────────────────────────────────── + +function loadOrganized(): Record> { + try { + if (!fs.existsSync(ORGANIZED_PATH)) return {} + return JSON.parse(fs.readFileSync(ORGANIZED_PATH, 'utf-8')) + } catch { return {} } +} + +function markOrganized(hash: string, filename: string) { + const data = loadOrganized() + if (!data[hash]) data[hash] = {} + data[hash][filename] = new Date().toISOString() + fs.writeFileSync(ORGANIZED_PATH, JSON.stringify(data, null, 2), 'utf-8') +} + +function isOrganized(hash: string, filename: string): boolean { + return !!loadOrganized()[hash]?.[filename] +} + +// ─── Torrent lookup ─────────────────────────────────────────── + +function findTorrentByHash(hash: string): any | null { + try { + if (!fs.existsSync(TORRENTS_PATH)) return null + const torrents = JSON.parse(fs.readFileSync(TORRENTS_PATH, 'utf-8')) + return torrents.find((t: any) => t.infohash?.toLowerCase() === hash.toLowerCase()) ?? null + } catch { return null } +} + +// ─── Filesystem ops ─────────────────────────────────────────── + +function seasonFolder(n: number): string { + return `Saison ${String(n).padStart(3, '0')}` +} + +function tryHardlink(src: string, dest: string): boolean { + try { fs.linkSync(src, dest); return true } + catch (err) { + warn(`[organize] hardlink échoué (${err instanceof Error ? err.message : err})`) + return false + } +} + +// ─── Organise un torrent ────────────────────────────────────── + +async function organizeTorrent(hash: string, name: string, savePath: string) { + const { mediaPath, completePath, organizeMode } = readSettings() + const result = { total: 0, skipped: 0, done: 0, errors: [] as { file: string; error: string }[] } + + log(`[organize] ── ${name} ──`) + log(`[organize] hash=${hash} save_path=${savePath} mode=${organizeMode}`) + + const torrent = findTorrentByHash(hash) + if (!torrent) { + error(`[organize] ✗ Hash ${hash} introuvable dans torrent_final.json`) + throw new Error(`Torrent introuvable : ${hash}`) + } + log(`[organize] ✓ Trouvé : ${torrent.raw}`) + + const serieTitle = torrent.serie_title || torrent.show_title || name + const torrentFiles: any[] = torrent.torrent_files ?? [] + log(`[organize] serie=${serieTitle} fichiers=${torrentFiles.length}`) + + // ── Cas fichier unique ──────────────────────────────────── + if (torrentFiles.length === 0) { + log(`[organize] torrent_files vide → tentative fichier unique`) + const candidates = [path.join(savePath, name), path.join(completePath, name)] + let src: string | null = null + for (const c of candidates) { + log(`[organize] candidate: ${c} → ${fs.existsSync(c) ? '✓' : '✗'}`) + if (fs.existsSync(c)) { src = c; break } + } + if (!src) throw new Error(`Source introuvable pour "${name}"`) + + const seasonNum = torrent.resolved_episodes?.[0]?.season_number ?? 1 + const destDir = path.join(mediaPath, serieTitle, seasonFolder(seasonNum)) + const dest = path.join(destDir, name) + log(`[organize] dest: ${dest}`) + + if (!fs.existsSync(dest)) { + fs.mkdirSync(destDir, { recursive: true }) + if (organizeMode === 'hardlink') { + if (!tryHardlink(src, dest)) await fsp.copyFile(src, dest) + } else { + await fsp.copyFile(src, dest) + await fsp.unlink(src) + } + log(`[organize] ✓ ${organizeMode} OK`) + } else { + log(`[organize] → dest existe déjà, skip`) + } + markOrganized(hash, name) + return { ...result, total: 1, done: 1 } + } + + // ── Map filename → season ───────────────────────────────── + const filenameSeasonMap = new Map() + for (const ep of torrent.resolved_episodes ?? []) { + if (ep.filename && ep.season_number !== undefined) + filenameSeasonMap.set(ep.filename, ep.season_number) + } + + result.total = torrentFiles.length + + for (const tf of torrentFiles) { + const filename = tf.filename + const filePath: string[] = tf.path + + if (isOrganized(hash, filename)) { result.skipped++; continue } + + const candidates = [ + path.join(savePath, name, ...filePath.slice(0, -1), filename), + path.join(savePath, ...filePath.slice(0, -1), filename), + path.join(savePath, name, filename), + path.join(savePath, filename), + path.join(completePath, name, ...filePath.slice(0, -1), filename), + path.join(completePath, name, filename), + path.join(completePath, filename), + ] + + let src: string | null = null + for (const c of candidates) { + if (fs.existsSync(c)) { src = c; break } + } + + if (!src) { + const tried = candidates.map(c => ` • ${c}`).join('\n') + error(`[organize] ✗ ${filename} introuvable:\n${tried}`) + result.errors.push({ file: filename, error: 'Source introuvable' }) + continue + } + + const seasonNum = filenameSeasonMap.get(filename) ?? 1 + const destDir = path.join(mediaPath, serieTitle, seasonFolder(seasonNum)) + const dest = path.join(destDir, filename) + + if (fs.existsSync(dest)) { + markOrganized(hash, filename) + result.skipped++ + continue + } + + try { + fs.mkdirSync(destDir, { recursive: true }) + if (organizeMode === 'hardlink') { + if (!tryHardlink(src, dest)) await fsp.copyFile(src, dest) + } else { + await fsp.copyFile(src, dest) + await fsp.unlink(src) + } + markOrganized(hash, filename) + result.done++ + log(`[organize] ✓ ${filename} → Saison ${String(seasonNum).padStart(3,'0')}`) + } catch (err) { + const msg = err instanceof Error ? err.message : 'Erreur inconnue' + error(`[organize] ✗ ${filename}: ${msg}`) + result.errors.push({ file: filename, error: msg }) + } + } + + log(`[organize] ── Résultat ${name}: ${result.done} OK, ${result.skipped} skippés, ${result.errors.length} erreurs`) + return result +} + +// ─── Boucle principale du worker ───────────────────────────── + +parentPort?.on('message', async (msg: any) => { + if (msg.type !== 'run') return + + const torrents: any[] = msg.torrents + const organized = loadOrganized() + + for (const t of torrents) { + if (t.state !== 'seeding') continue + + const torrent = findTorrentByHash(t.hash) + if (!torrent) { + warn(`[organize] hash ${t.hash} (${t.name}) non trouvé`) + continue + } + + const files: any[] = torrent.torrent_files ?? [] + // Pour fichier unique : vérifier via le name directement + const allDone = files.length > 0 + ? files.every((f: any) => organized[t.hash]?.[f.filename]) + : !!organized[t.hash]?.[t.name] + + if (allDone) continue + + try { + const result = await organizeTorrent(t.hash, t.name, t.save_path) + parentPort?.postMessage({ type: 'result', hash: t.hash, name: t.name, ...result }) + } catch (err) { + error(`[organize] Erreur ${t.name}: ${err instanceof Error ? err.message : err}`) + } + } + + parentPort?.postMessage({ type: 'done' }) +}) \ No newline at end of file diff --git a/server/organize.ts b/server/organize.ts index 9c72943..bc918d6 100644 --- a/server/organize.ts +++ b/server/organize.ts @@ -1,84 +1,15 @@ /** - * Organizer - * ========= - * Déplace ou hardlinke les fichiers téléchargés vers la médiathèque Jellyfin. - * Incrémental : skip les fichiers déjà organisés. - * - * Structure dest : - * mediaPath / serie_title / Saison 001 / filename.mkv + * organize.ts + * =========== + * Lance l'organizer dans un Worker Thread dédié pour ne pas bloquer Express. + * Le worker gère tout le filesystem (copy, hardlink, move). */ -import fs from 'fs' -import path from 'path' -import { readSettings } from './settings.js' +import path from 'path' +import { Worker } from 'worker_threads' -const TORRENTS_PATH = path.join(process.cwd(), 'data', 'torrent_final.json') -const ORGANIZED_PATH = path.join(process.cwd(), 'data', 'organized.json') - -// ─── Organized log ──────────────────────────────────────────── - -function loadOrganized(): Record> { - try { - if (!fs.existsSync(ORGANIZED_PATH)) return {} - return JSON.parse(fs.readFileSync(ORGANIZED_PATH, 'utf-8')) - } catch { return {} } -} - -function saveOrganized(data: Record>) { - fs.writeFileSync(ORGANIZED_PATH, JSON.stringify(data, null, 2), 'utf-8') -} - -function markOrganized(hash: string, filename: string) { - const data = loadOrganized() - if (!data[hash]) data[hash] = {} - data[hash][filename] = new Date().toISOString() - saveOrganized(data) -} - -function isOrganized(hash: string, filename: string): boolean { - const data = loadOrganized() - return !!data[hash]?.[filename] -} - -// ─── Torrent lookup ─────────────────────────────────────────── - -function readTorrents(): any[] { - try { - if (!fs.existsSync(TORRENTS_PATH)) return [] - return JSON.parse(fs.readFileSync(TORRENTS_PATH, 'utf-8')) - } catch { return [] } -} - -function findTorrentByHash(hash: string): any | null { - const torrents = readTorrents() - return torrents.find(t => t.infohash?.toLowerCase() === hash.toLowerCase()) ?? null -} - -// ─── Filesystem ops ─────────────────────────────────────────── - -function ensureDir(dir: string) { - fs.mkdirSync(dir, { recursive: true }) -} - -function seasonFolder(n: number): string { - return `Saison ${String(n).padStart(3, '0')}` -} - -function tryHardlink(src: string, dest: string): boolean { - try { - fs.linkSync(src, dest) - return true - } catch (err) { - console.warn(`[organize] hardlink échoué (${err instanceof Error ? err.message : err})`) - return false - } -} - -function copyFile(src: string, dest: string) { - fs.copyFileSync(src, dest) -} - -// ─── Résultat d'organisation ────────────────────────────────── +// Un seul worker à la fois +let workerRunning = false export interface OrganizeResult { total : number @@ -87,185 +18,103 @@ export interface OrganizeResult { errors : { file: string; error: string }[] } -// ─── Organise un torrent ────────────────────────────────────── - -export async function organizeTorrent( - hash : string, - name : string, // nom du dossier torrent (retourné par qB) - savePath: string // save_path retourné par qBittorrent -): Promise { - const settings = readSettings() - const { mediaPath, completePath, organizeMode } = settings - - console.log(`[organize] ── Début organisation ──────────────────────`) - console.log(`[organize] hash : ${hash}`) - console.log(`[organize] name : ${name}`) - console.log(`[organize] save_path : ${savePath}`) - console.log(`[organize] mediaPath : ${mediaPath}`) - console.log(`[organize] completePath: ${completePath}`) - console.log(`[organize] mode : ${organizeMode}`) - - const result: OrganizeResult = { total: 0, skipped: 0, done: 0, errors: [] } - - // Trouver le torrent dans torrent_final.json via infohash - const torrent = findTorrentByHash(hash) - if (!torrent) { - console.error(`[organize] ✗ Aucun torrent avec infohash="${hash}" dans torrent_final.json`) - throw new Error(`Torrent introuvable pour le hash : "${hash}"`) - } - console.log(`[organize] ✓ Torrent trouvé : ${torrent.raw}`) - - const serieTitle = torrent.serie_title || torrent.show_title || name - const torrentFiles: any[] = torrent.torrent_files ?? [] - console.log(`[organize] serie_title : ${serieTitle}`) - console.log(`[organize] fichiers : ${torrentFiles.length}`) - - if (torrentFiles.length === 0) { - throw new Error(`Aucun fichier trouvé dans le torrent "${name}"`) - } - - // Map filename → season_number depuis resolved_episodes - const filenameSeasonMap = new Map() - for (const ep of torrent.resolved_episodes ?? []) { - if (ep.filename && ep.season_number !== undefined) { - filenameSeasonMap.set(ep.filename, ep.season_number) - } - } - - result.total = torrentFiles.length - - for (const tf of torrentFiles) { - const filename = tf.filename - const filePath: string[] = tf.path - - console.log(`[organize] ── Fichier : ${filename}`) - console.log(`[organize] path[] : ${JSON.stringify(filePath)}`) - - // Skip si déjà organisé - if (isOrganized(hash, filename)) { - console.log(`[organize] → déjà organisé, skip`) - result.skipped++ - continue - } - - // Essayer plusieurs chemins sources possibles - const candidates = [ - path.join(savePath, name, ...filePath.slice(0, -1), filename), - path.join(savePath, ...filePath.slice(0, -1), filename), - path.join(savePath, name, filename), - path.join(savePath, filename), - path.join(completePath, name, ...filePath.slice(0, -1), filename), - path.join(completePath, name, filename), - path.join(completePath, filename), - ] - - let src: string | null = null - for (const candidate of candidates) { - console.log(`[organize] candidate: ${candidate} → ${fs.existsSync(candidate) ? '✓ EXISTS' : '✗ absent'}`) - if (fs.existsSync(candidate)) { - src = candidate - break - } - } - - if (!src) { - const err = `Source introuvable (essayé ${candidates.length} chemins)` - console.error(`[organize] ✗ ${err}`) - result.errors.push({ file: filename, error: err }) - continue - } - - console.log(`[organize] ✓ source : ${src}`) - - // Déterminer la saison - const seasonNum = filenameSeasonMap.get(filename) ?? 1 - console.log(`[organize] saison : ${seasonNum}`) - - // Chemin destination - const destDir = path.join(mediaPath, serieTitle, seasonFolder(seasonNum)) - const dest = path.join(destDir, filename) - console.log(`[organize] dest : ${dest}`) - - // Skip si dest existe déjà - if (fs.existsSync(dest)) { - console.log(`[organize] → dest existe déjà, skip`) - markOrganized(hash, filename) - result.skipped++ - continue - } - - try { - ensureDir(destDir) - - if (organizeMode === 'hardlink') { - const ok = tryHardlink(src, dest) - if (!ok) { - console.warn(`[organize] hardlink impossible → fallback copy`) - copyFile(src, dest) - } - } else { - copyFile(src, dest) - fs.unlinkSync(src) - } - - markOrganized(hash, filename) - result.done++ - console.log(`[organize] ✓ ${organizeMode} OK`) - } catch (err) { - const msg = err instanceof Error ? err.message : 'Erreur inconnue' - console.error(`[organize] ✗ ${msg}`) - result.errors.push({ file: filename, error: msg }) - } - } - - console.log(`[organize] ── Résultat : ${result.done} OK, ${result.skipped} skippés, ${result.errors.length} erreurs`) - return result -} - -// ─── Auto-organise polling ──────────────────────────────────── - +/** + * Lance l'auto-organisation dans un worker thread. + * listFn est appelée dans le thread principal (accès qBittorrent), + * puis les données sont passées au worker. + */ export async function autoOrganizeAll( listFn: () => Promise ): Promise { + if (workerRunning) { + console.log('[organize] Worker déjà en cours, skip') + return + } + let torrents: any[] try { torrents = await listFn() } catch (err) { - console.error('[organize] autoOrganizeAll: impossible de récupérer la liste des torrents:', err) + console.error('[organize] Impossible de récupérer la liste des torrents:', err) return } - console.log(`[organize] autoOrganizeAll: ${torrents.length} torrents récupérés`) + if (torrents.length === 0) return - const organized = loadOrganized() + const seedingCount = torrents.filter(t => t.state === 'seeding').length + if (seedingCount === 0) return - for (const t of torrents) { - if (t.state !== 'seeding') { - console.log(`[organize] skip ${t.name} (state=${t.state})`) - continue + console.log(`[organize] Lancement worker (${seedingCount} torrents en seeding)`) + workerRunning = true + + const workerPath = path.join(path.dirname(new URL(import.meta.url).pathname), 'organize-worker.js') + + const worker = new Worker(workerPath) + + worker.on('message', (msg: any) => { + if (msg.type === 'log') { + if (msg.level === 'error') console.error(msg.msg) + else if (msg.level === 'warn') console.warn(msg.msg) + else console.log(msg.msg) + } else if (msg.type === 'result') { + console.log(`[organize] ${msg.name} → ${msg.done} OK, ${msg.skipped} skippés, ${msg.errors.length} erreurs`) + } else if (msg.type === 'done') { + console.log('[organize] Worker terminé') + workerRunning = false + worker.terminate() } + }) - const torrent = findTorrentByHash(t.hash) - if (!torrent) { - console.warn(`[organize] hash ${t.hash} (${t.name}) non trouvé dans torrent_final.json`) - continue - } + worker.on('error', (err) => { + console.error('[organize] Worker erreur:', err) + workerRunning = false + }) - const files: any[] = torrent.torrent_files ?? [] - if (files.length === 0) continue + worker.on('exit', (code) => { + if (code !== 0) console.error(`[organize] Worker exit code ${code}`) + workerRunning = false + }) - const allDone = files.every((f: any) => organized[t.hash]?.[f.filename]) - if (allDone) { - console.log(`[organize] ${t.name} → déjà entièrement organisé`) - continue - } + // Envoyer les torrents au worker + worker.postMessage({ type: 'run', torrents }) +} - console.log(`[organize] Auto-organisation : ${t.name}`) - try { - const result = await organizeTorrent(t.hash, t.name, t.save_path) - console.log(`[organize] ${t.name} → ${result.done} fichiers, ${result.skipped} skippés, ${result.errors.length} erreurs`) - } catch (err) { - console.error(`[organize] Erreur pour ${t.name}:`, err) - } - } +/** + * Organisation manuelle d'un torrent (appelée depuis l'API /organize). + * Lance aussi dans un worker thread. + */ +export async function organizeTorrent( + hash : string, + name : string, + savePath: string +): Promise { + return new Promise((resolve, reject) => { + const workerPath = path.join(path.dirname(new URL(import.meta.url).pathname), 'organize-worker.js') + const worker = new Worker(workerPath) + + // On envoie un seul torrent "fictif seeding" + const fakeTorrent = { hash, name, save_path: savePath, state: 'seeding' } + const result: OrganizeResult = { total: 0, skipped: 0, done: 0, errors: [] } + + worker.on('message', (msg: any) => { + if (msg.type === 'log') { + if (msg.level === 'error') console.error(msg.msg) + else if (msg.level === 'warn') console.warn(msg.msg) + else console.log(msg.msg) + } else if (msg.type === 'result') { + result.total = msg.total + result.skipped = msg.skipped + result.done = msg.done + result.errors = msg.errors + } else if (msg.type === 'done') { + worker.terminate() + resolve(result) + } + }) + + worker.on('error', (err) => { worker.terminate(); reject(err) }) + worker.on('exit', (code) => { if (code !== 0) reject(new Error(`Worker exit ${code}`)) }) + + worker.postMessage({ type: 'run', torrents: [fakeTorrent] }) + }) } \ No newline at end of file