sascha/n8n
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request 'fix: wait for every file in season packs' (#2) from fix/media-handoff-season-packs-20260905 into main

Browse source
This commit is contained in:
sascha 2026-09-05 08:20:57 +02:00
commit 187dc1eb16
1 changed files with 2 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

4
workflows/media-handoff-orchestrator.json
View file

@ -20,7 +20,7 @@
},
{
"parameters": {
"jsCode": "const body = $json.body ?? $json;\nconst action = String(body.action ?? '').toLowerCase();\nconst routes = {\n serien4k: { service: 'sonarr', kind: 'episode' },\n serien: { service: 'sonarr1080p', kind: 'episode' },\n video4k: { service: 'radarr', kind: 'movie' },\n video: { service: 'radarr1080p', kind: 'movie' },\n};\nconst now = Date.now();\nconst state = $getWorkflowStaticData('global');\nstate.jobs = state.jobs || {};\nconst jobs = state.jobs;\nconst norm = (v) => String(v ?? '').replace(/\\.nzb$/i, '').toLowerCase().replace(/[^a-z0-9]+/g, '');\nconst active = new Set(['registered', 'processing']);\nfor (const [id, job] of Object.entries(jobs)) {\n const terminalAt = Date.parse(job.completedAt || job.failedAt || job.timeoutAt || 0);\n if (!active.has(job.state) && terminalAt && now - terminalAt > 7 * 86400000) delete jobs[id];\n}\nconst fail = (error, extra = {}) => [{ json: { ok: false, error, ...extra } }];\nif (action === 'start') {\n const category = String(body.category ?? '').toLowerCase();\n const route = routes[category];\n const directory = String(body.directory ?? '');\n const release = String(body.release ?? '');\n const cleanName = String(body.cleanName ?? '');\n if (!route) return fail('unsupported_category', { category });\n if (!directory.startsWith('/usenet/complete/' + category + '/')) return fail('invalid_directory');\n if (!release || release.length > 500 || cleanName.length > 500) return fail('invalid_release');\n const dedupeKey = category + ':' + norm(release || cleanName);\n const prior = Object.values(jobs).find(j => active.has(j.state) && j.dedupeKey === dedupeKey);\n if (prior) return [{ json: { ok: true, idempotent: true, jobId: prior.jobId, state: prior.state, service: prior.service } }];\n const jobId = now.toString(36) + '-' + Math.random().toString(36).slice(2, 10);\n jobs[jobId] = {\n jobId, category, service: route.service, kind: route.kind, directory,\n release, cleanName, dedupeKey, state: 'registered',\n startedAt: new Date(now).toISOString(), updatedAt: new Date(now).toISOString(),\n };\n return [{ json: { ok: true, jobId, state: 'registered', service: route.service, kind: route.kind } }];\n}\nconst jobId = String(body.jobId ?? '');\nconst job = jobs[jobId];\nif (!job) return fail('job_not_found', { jobId });\nif (action === 'moved') {\n if (job.state === 'registered') {\n job.state = 'processing'; job.movedAt = new Date(now).toISOString(); job.updatedAt = job.movedAt;\n }\n return [{ json: { ok: true, jobId, state: job.state, service: job.service } }];\n}\nif (action === 'fail') {\n if (active.has(job.state)) {\n job.state = 'failed'; job.reason = String(body.reason ?? 'handoff_failed').slice(0, 300);\n job.failedAt = new Date(now).toISOString(); job.updatedAt = job.failedAt;\n }\n return [{ json: { ok: true, jobId, state: job.state } }];\n}\nif (action === 'status') {\n return [{ json: { ok: true, jobId, state: job.state, category: job.category, service: job.service,\n reason: job.reason ?? null, startedAt: job.startedAt, updatedAt: job.updatedAt } }];\n}\nreturn fail('unsupported_action', { action });"
"jsCode": "const body = $json.body ?? $json;\nconst action = String(body.action ?? '').toLowerCase();\nconst routes = {\n serien4k: { service: 'sonarr', kind: 'episode' },\n serien: { service: 'sonarr1080p', kind: 'episode' },\n video4k: { service: 'radarr', kind: 'movie' },\n video: { service: 'radarr1080p', kind: 'movie' },\n};\nconst now = Date.now();\nconst state = $getWorkflowStaticData('global');\nstate.jobs = state.jobs || {};\nconst jobs = state.jobs;\nconst norm = (v) => String(v ?? '').replace(/\\.nzb$/i, '').toLowerCase().replace(/[^a-z0-9]+/g, '');\nconst active = new Set(['registered', 'processing']);\nfor (const [id, job] of Object.entries(jobs)) {\n const terminalAt = Date.parse(job.completedAt || job.failedAt || job.timeoutAt || 0);\n if (!active.has(job.state) && terminalAt && now - terminalAt > 7 * 86400000) delete jobs[id];\n}\nconst fail = (error, extra = {}) => [{ json: { ok: false, error, ...extra } }];\nif (action === 'start') {\n const category = String(body.category ?? '').toLowerCase();\n const route = routes[category];\n const directory = String(body.directory ?? '');\n const release = String(body.release ?? '');\n const cleanName = String(body.cleanName ?? '');\n const expectedFiles = Number(body.expectedFiles ?? 1);\n if (!route) return fail('unsupported_category', { category });\n if (!directory.startsWith('/usenet/complete/' + category + '/')) return fail('invalid_directory');\n if (!release || release.length > 500 || cleanName.length > 500) return fail('invalid_release');\n if (!Number.isInteger(expectedFiles) || expectedFiles < 1 || expectedFiles > 100) return fail('invalid_expected_files');\n const dedupeKey = category + ':' + norm(release || cleanName);\n const prior = Object.values(jobs).find(j => active.has(j.state) && j.dedupeKey === dedupeKey);\n if (prior) return [{ json: { ok: true, idempotent: true, jobId: prior.jobId, state: prior.state, service: prior.service } }];\n const jobId = now.toString(36) + '-' + Math.random().toString(36).slice(2, 10);\n jobs[jobId] = {\n jobId, category, service: route.service, kind: route.kind, directory,\n release, cleanName, expectedFiles, importedFiles: 0, dedupeKey, state: 'registered',\n startedAt: new Date(now).toISOString(), updatedAt: new Date(now).toISOString(),\n };\n return [{ json: { ok: true, jobId, state: 'registered', service: route.service, kind: route.kind, expectedFiles } }];\n}\nconst jobId = String(body.jobId ?? '');\nconst job = jobs[jobId];\nif (!job) return fail('job_not_found', { jobId });\nif (action === 'moved') {\n if (job.state === 'registered') {\n job.state = 'processing'; job.movedAt = new Date(now).toISOString(); job.updatedAt = job.movedAt;\n }\n return [{ json: { ok: true, jobId, state: job.state, service: job.service } }];\n}\nif (action === 'fail') {\n if (active.has(job.state)) {\n job.state = 'failed'; job.reason = String(body.reason ?? 'handoff_failed').slice(0, 300);\n job.failedAt = new Date(now).toISOString(); job.updatedAt = job.failedAt;\n }\n return [{ json: { ok: true, jobId, state: job.state } }];\n}\nif (action === 'status') {\n return [{ json: { ok: true, jobId, state: job.state, category: job.category, service: job.service,\n reason: job.reason ?? null, expectedFiles: job.expectedFiles ?? 1, importedFiles: job.importedFiles ?? 0,\n startedAt: job.startedAt, updatedAt: job.updatedAt } }];\n}\nreturn fail('unsupported_action', { action });"
},
"id": "81546df2-b2c1-4753-b45c-8f8a6d223cf1",
"name": "Handle API Request",
@ -240,7 +240,7 @@
},
{
"parameters": {
"jsCode": "const state = $getWorkflowStaticData('global');\nstate.jobs = state.jobs || {};\nconst jobs = state.jobs;\nconst now = Date.now();\nconst norm = (v) => String(v ?? '').replace(/\\.nzb$/i, '').toLowerCase().replace(/[^a-z0-9]+/g, '');\nconst flatten = (nodeName) => {\n const out = [];\n for (const item of $(nodeName).all()) {\n const value = item.json;\n if (Array.isArray(value)) out.push(...value);\n else if (Array.isArray(value?.records)) out.push(...value.records);\n else if (value?.eventType) out.push(value);\n }\n return out;\n};\nconst sources = {\n sonarr: flatten('History Sonarr UHD'),\n sonarr1080p: flatten('History Sonarr FHD'),\n radarr: flatten('History Radarr UHD'),\n radarr1080p: flatten('History Radarr FHD'),\n};\nconst imported = new Set(['downloadFolderImported']);\nconst failed = new Set(['downloadFailed', 'downloadIgnored']);\nconst changed = [];\nfor (const job of Object.values(jobs)) {\n if (!['registered', 'processing'].includes(job.state)) continue;\n const start = Date.parse(job.startedAt);\n const wanted = new Set([norm(job.release), norm(job.cleanName)].filter(Boolean));\n const events = (sources[job.service] || []).filter(e => {\n const at = Date.parse(e.date || 0);\n return at >= start && wanted.has(norm(e.sourceTitle));\n }).sort((a,b) => Date.parse(a.date) - Date.parse(b.date));\n const terminals = events.filter(e => imported.has(e.eventType) || failed.has(e.eventType));\n const latest = terminals.at(-1);\n if (latest && imported.has(latest.eventType)) {\n job.state = 'success'; job.completedAt = latest.date; job.updatedAt = new Date(now).toISOString();\n job.importEventId = latest.id ?? null; changed.push({ jobId: job.jobId, state: job.state }); continue;\n }\n if (latest && failed.has(latest.eventType)) {\n job.state = 'failed'; job.failedAt = latest.date; job.updatedAt = new Date(now).toISOString();\n job.reason = latest.eventType; changed.push({ jobId: job.jobId, state: job.state }); continue;\n }\n const wasRegistered = job.state === 'registered';\n const maxAge = wasRegistered ? 15 * 60000 : 24 * 3600000;\n if (now - start > maxAge) {\n job.state = 'timeout'; job.timeoutAt = new Date(now).toISOString(); job.updatedAt = job.timeoutAt;\n job.reason = wasRegistered ? 'move_not_confirmed' : 'import_timeout';\n changed.push({ jobId: job.jobId, state: job.state });\n }\n}\nreturn [{ json: { ok: true, activeJobs: Object.values(jobs).filter(j => ['registered','processing'].includes(j.state)).length, changed } }];"
"jsCode": "const state = $getWorkflowStaticData('global');\nstate.jobs = state.jobs || {};\nconst jobs = state.jobs;\nconst now = Date.now();\nconst norm = (v) => String(v ?? '').replace(/\\.nzb$/i, '').toLowerCase().replace(/[^a-z0-9]+/g, '');\nconst flatten = (nodeName) => {\n const out = [];\n for (const item of $(nodeName).all()) {\n const value = item.json;\n if (Array.isArray(value)) out.push(...value);\n else if (Array.isArray(value?.records)) out.push(...value.records);\n else if (value?.eventType) out.push(value);\n }\n return out;\n};\nconst sources = {\n sonarr: flatten('History Sonarr UHD'),\n sonarr1080p: flatten('History Sonarr FHD'),\n radarr: flatten('History Radarr UHD'),\n radarr1080p: flatten('History Radarr FHD'),\n};\nconst imported = new Set(['downloadFolderImported']);\nconst failed = new Set(['downloadFailed', 'downloadIgnored']);\nconst changed = [];\nfor (const job of Object.values(jobs)) {\n if (!['registered', 'processing'].includes(job.state)) continue;\n const start = Date.parse(job.startedAt);\n const wanted = new Set([norm(job.release), norm(job.cleanName)].filter(Boolean));\n const events = (sources[job.service] || []).filter(e => {\n const at = Date.parse(e.date || 0);\n return at >= start && wanted.has(norm(e.sourceTitle));\n }).sort((a,b) => Date.parse(a.date) - Date.parse(b.date));\n const terminals = events.filter(e => imported.has(e.eventType) || failed.has(e.eventType));\n const latest = terminals.at(-1);\n const importEvents = events.filter(e => imported.has(e.eventType));\n const importedUnits = new Set(importEvents.map(e => String(\n job.kind === 'episode' ? (e.episodeId ?? e.data?.fileId ?? e.id) : (e.movieId ?? e.data?.fileId ?? e.id)\n )));\n job.importedFiles = importedUnits.size;\n const expectedFiles = Number(job.expectedFiles ?? 1);\n if (importedUnits.size >= expectedFiles) {\n const completed = importEvents.at(-1);\n job.state = 'success'; job.completedAt = completed.date; job.updatedAt = new Date(now).toISOString();\n job.importEventId = completed.id ?? null;\n changed.push({ jobId: job.jobId, state: job.state, importedFiles: importedUnits.size }); continue;\n }\n if (latest && failed.has(latest.eventType)) {\n job.state = 'failed'; job.failedAt = latest.date; job.updatedAt = new Date(now).toISOString();\n job.reason = latest.eventType; changed.push({ jobId: job.jobId, state: job.state }); continue;\n }\n const wasRegistered = job.state === 'registered';\n const maxAge = wasRegistered ? 15 * 60000 : 24 * 3600000;\n if (now - start > maxAge) {\n job.state = 'timeout'; job.timeoutAt = new Date(now).toISOString(); job.updatedAt = job.timeoutAt;\n job.reason = wasRegistered ? 'move_not_confirmed' : 'import_timeout';\n changed.push({ jobId: job.jobId, state: job.state });\n }\n}\nreturn [{ json: { ok: true, activeJobs: Object.values(jobs).filter(j => ['registered','processing'].includes(j.state)).length, changed } }];"
},
"id": "54c60ec7-6854-4916-a4e3-9a2b7d13b503",
"name": "Reconcile Jobs",