Files
MangaCircus/peering/adapter.js
T

111 lines
7.2 KiB
JavaScript

'use strict';
const fs = require('node:fs').promises;
const path = require('node:path');
const crypto = require('node:crypto');
const { PeeringState, requirePeer } = require('./state');
const client = require('./client');
const { ArchiveControlError } = require('../scraper/lib/control');
const { selectByScope } = require('../scraper/lib/scope');
const { scanLibrary, findManga, shouldDownload } = require('../scraper/lib/library-inventory');
const { publishManifest } = require('../scraper/lib/publisher');
const { writeJsonAtomic } = require('../scraper/lib/json-file');
const { sleep } = require('../scraper/lib/delay');
const dataRoot = () => path.resolve(process.env.PEER_DATA || path.join(process.env.DATA || 'data', 'peering'));
const store = () => new PeeringState(dataRoot());
const infoFor = value => {
try {
const u = new URL(String(value));
if (u.protocol !== 'peer:' || u.username || u.password || u.search || u.hash || !/^p[0-9a-f]{24}$/.test(u.hostname)
|| !/^\/m[0-9a-f]{24}$/.test(u.pathname)) return null;
const peer = requirePeer(store().load(), u.hostname);
return { source: `peer-${peer.id}`, id: `${peer.id}-${u.pathname.slice(1)}`, peerId: peer.id,
shareId: u.pathname.slice(1), title: peer.alias, url: `peer://${peer.id}${u.pathname}` };
} catch { return null; }
};
function refsOf(result) {
return result.chapters.map(ref => ({ id: ref.id, number: ref.number, volume: ref.volume,
book: { number: Number(ref.volume) || 1 }, title: ref.title, pages: ref.pageCount }));
}
async function discover(info) {
const result = await client.manga(store(), info.peerId, info.shareId);
return { title: result.title, chapters: refsOf(result), availability: { selectedChapters: result.chapters.length, coverage: 'shared-selection' } };
}
async function run(info, item, options) {
const stateStore = store(), peer = requirePeer(stateStore.load(), info.peerId);
if (!peer.paired || !peer.receive) throw new Error('Downloads from this peer are disabled');
const stop = () => { const reason = options.shouldStop?.(); if (reason) throw new ArchiveControlError(reason === true ? 'paused' : String(reason)); };
stop();
const remote = await client.manga(stateStore, info.peerId, info.shareId);
const catalog = { title: remote.title, chapters: refsOf(remote), availability: { selectedChapters: remote.chapters.length, coverage: 'shared-selection' } };
if (options.onCatalog) await options.onCatalog(catalog);
const sourceUrl = info.url, source = info.source;
const workRoot = path.join(options.libraryRoot, `.mcscraper-work-${info.id}`);
const stateFile = path.join(workRoot, 'job.json');
await fs.mkdir(workRoot, { recursive: true });
let job = { version: 1, publicationMode: 'incremental', source, seriesUrl: sourceUrl, completed: [], jobCompleted: 0,
status: 'running', started: new Date().toISOString(), updated: new Date().toISOString(), total: 0, current: null };
try { const prior = JSON.parse(await fs.readFile(stateFile, 'utf8')); if (prior.seriesUrl === sourceUrl) job = { ...prior, status: 'running' }; }
catch (e) { if (e.code !== 'ENOENT') throw e; }
const record = async () => { job.updated = new Date().toISOString(); await writeJsonAtomic(stateFile, job); };
const inventory = await scanLibrary(options.libraryRoot);
let manga = findManga(inventory, { source, seriesId: info.id, sourceUrl, libraryMangaId: item.libraryMangaId });
if (item.libraryMangaId && !manga) throw new Error(`Library manga target not found: ${item.libraryMangaId}`);
const selected = selectByScope(catalog.chapters.filter(ref => shouldDownload(manga, ref, source, item.existingPolicy)), item.scope);
job.total = selected.length; job.jobCompleted = 0; job.current = null; await record();
let publishedCount = 0;
const series = { title: remote.title, direction: ['rtl', 'ltr', 'ttb'].includes(remote.direction) ? remote.direction : 'rtl',
sources: [{ name: `Peer ${peer.alias}`, url: sourceUrl }],
metadata: { title: remote.title }, scraper: { ids: { [source]: info.id } } };
for (let index = 0; index < selected.length; index++) {
stop();
const ref = selected[index];
if (!shouldDownload(manga, ref, source, item.existingPolicy)) { job.jobCompleted++; await record(); continue; }
const chapter = await client.chapter(stateStore, info.peerId, info.shareId, ref.id);
if (String(chapter.number) !== String(ref.number) || String(chapter.volume) !== String(ref.volume)) throw new Error('Peer chapter identity changed');
const digest = crypto.createHash('sha256').update(JSON.stringify(chapter.pages.map(p => [p.id, p.sha256]))).digest('hex').slice(0, 20);
const stage = path.join(workRoot, 'pages', `${ref.id}-${digest}`);
await fs.mkdir(stage, { recursive: true });
job.current = { number: ref.number, sourceId: ref.id, position: index + 1, total: selected.length,
downloadedPages: 0, expectedPages: chapter.pages.length };
await record();
const pages = chapter.pages.map((page, i) => ({ ...page, path: path.join(stage, `${String(i + 1).padStart(4, '0')}.${page.mime.split('/')[1] === 'jpeg' ? 'jpg' : page.mime.split('/')[1]}`) }));
const concurrency = Math.min(4, Math.max(1, Number(requirePeer(stateStore.load(), info.peerId).download?.concurrent || 2)));
for (let offset = 0; offset < pages.length; offset += concurrency) {
stop();
const batch = pages.slice(offset, offset + concurrency);
await Promise.all(batch.map(page => client.downloadPage(stateStore, info.peerId, page, page.path)));
job.current.downloadedPages = Math.min(pages.length, offset + batch.length);
await record();
stop();
}
const publication = await publishManifest({ version: 1, series, chapters: [{
number: ref.number, title: ref.title, book: { number: Number(ref.volume) || 1 },
source: { name: `Peer ${peer.alias} ${peer.id.slice(1, 7)}`, url: sourceUrl },
pages: pages.map(page => ({ path: page.path })),
}] }, { libraryRoot: options.libraryRoot, ...(manga ? { mangaDir: manga.path } : {}) });
if (publication.results[0].status === 'published') {
publishedCount++;
if (!manga) manga = { id: path.basename(publication.mangaDir), path: publication.mangaDir, chapters: [], meta: {} };
manga.chapters.push({ book: String(Number(ref.volume) || 1), number: String(ref.number), sourceKey: source });
if (options.onPublished) await options.onPublished({ mangaDir: publication.mangaDir, chapter: ref.number });
}
job.jobCompleted++; job.lastChapter = ref.number; job.current = null; await record();
// Only a fully verified and published chapter loses its resumable page cache.
await fs.rm(stage, { recursive: true, force: true });
if (index < selected.length - 1) { stop(); await sleep(Math.max(500, Number(options.settings.chapterMinMs || 500))); }
}
job.status = 'complete'; job.current = null; await record();
return { finalPath: manga?.path || null, publishedCount, selectedCount: selected.length };
}
function adapterForPeer(value) {
const info = infoFor(value);
if (!info) return null;
const adapter = { id: info.source, label: `Peer ${info.peerId.slice(1, 7)}`,
async discover() { return discover(info); },
async run(item, options) { return run(info, item, options); } };
return { adapter, info };
}
module.exports = { adapterForPeer, infoFor, discover, run };