diff --git a/package-lock.json b/package-lock.json index 4c291625..4cbf8dc9 100644 --- a/package-lock.json +++ b/package-lock.json @@ -49,7 +49,7 @@ "source-map-support": "^0.5.21", "template-format": "^2.0.0", "undici": "^8.12.4", - "unprint": "^1.0.0-beta.0", + "unprint": "^1.0.0-beta.1", "why-is-node-running": "^3.2.3", "winston": "^3.11.0", "winston-daily-rotate-file": "^5.0.0", @@ -8829,9 +8829,9 @@ } }, "node_modules/unprint": { - "version": "1.0.0-beta.0", - "resolved": "https://registry.npmjs.org/unprint/-/unprint-1.0.0-beta.0.tgz", - "integrity": "sha512-OOATeXAMAgZzmtDlrc8KOD6hSCWYvqBY+4ASszkxYTTKgPD2rC5M+OM3VFO0VeO/LcIM/BaIJuzi1Eg9rBWPAQ==", + "version": "1.0.0-beta.1", + "resolved": "https://registry.npmjs.org/unprint/-/unprint-1.0.0-beta.1.tgz", + "integrity": "sha512-Q4tw3kVVagKqheiA59Q3KYNlOLQjd5RO5NdE86JA+W1eeFelVwYdGSk1T3NANje2+NwdrSqGw9kRBhH0HIUhzQ==", "license": "ISC", "dependencies": { "bottleneck": "^2.19.5", diff --git a/package.json b/package.json index 2c3f29f0..2ebc4857 100755 --- a/package.json +++ b/package.json @@ -81,7 +81,7 @@ "source-map-support": "^0.5.21", "template-format": "^2.0.0", "undici": "^8.12.4", - "unprint": "^1.0.0-beta.0", + "unprint": "^1.0.0-beta.1", "why-is-node-running": "^3.2.3", "winston": "^3.11.0", "winston-daily-rotate-file": "^5.0.0", diff --git a/src/app.js b/src/app.js index c6ec1f38..9b64cb6d 100755 --- a/src/app.js +++ b/src/app.js @@ -11,7 +11,6 @@ const fs = require('fs').promises; const { format, intervalToDuration } = require('date-fns'); const argv = require('./argv'); -const http = require('./utils/http'); const logger = require('./logger')(__filename); const knex = require('./knex'); @@ -210,7 +209,7 @@ async function init() { } if (argv.request) { - const res = await http[argv.requestMethod](argv.request); + const res = await unprint.request(argv.request, { method: argv.requestMethod, interface: 'request' }); console.log(res.status, res.body); } @@ -270,6 +269,7 @@ async function init() { } await unprint.closeAllBrowsers(); + await unprint.closeAgents(); knex.destroy(); redis.destroy(); diff --git a/src/deep.js b/src/deep.js index 7df131c8..d6af6a9d 100755 --- a/src/deep.js +++ b/src/deep.js @@ -10,7 +10,6 @@ const include = require('./utils/argv-include')(argv); const { fetchReleaseEntities, urlToHostname } = require('./entities'); const logger = require('./logger')(__filename); const getRecursiveParameters = require('./utils/get-recursive-parameters'); -const windows = require('./utils/http-windows'); const waitImmediate = util.promisify(setImmediate); @@ -142,19 +141,10 @@ async function scrapeRelease(baseRelease, entities, type = 'scene') { ? await fetchScene(layoutScraper, baseRelease.url, entity, baseRelease, options) : await fetchMovie(layoutScraper, baseRelease.url, entity, baseRelease, options); - const pathname = baseRelease.path || (baseRelease.url && new URL(baseRelease.url).pathname.replace(/\//g, '_')); - if (rawScrapedRelease) { delete rawScrapedRelease.query; // some scrapers pass the qu-wrapped window instance to parent scrapers, filling up memory } - if (windows.has(pathname)) { - logger.debug(`Closing window for ${pathname}`); - - windows.get(pathname).close(); - windows.delete(pathname); - } - await waitImmediate; // logger.debug(`Memory usage after: ${process.memoryUsage.rss() / 1000000} MB (${baseRelease.url})`); diff --git a/src/media.js b/src/media.js index 2b096891..23037fb3 100755 --- a/src/media.js +++ b/src/media.js @@ -23,13 +23,15 @@ const { S3Client, DeleteObjectsCommand } = require('@aws-sdk/client-s3'); const logger = require('./logger')(__filename); const argv = require('./argv'); const knex = require('./knex'); -const http = require('./utils/http'); const batchInsert = require('./utils/batch-insert'); const chunk = require('./utils/chunk'); const { fetchEntityReleaseIds } = require('./entity-releases'); const streamQueue = taskQueue(); +// carried over from the legacy HTTP module, unprint's raw request interface would identify as traxxx +const mediaUserAgent = 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36'; + const s3 = new S3Client({ region: 'eu-central-1', endpoint: 'https://s3.eu-central-1.wasabisys.com', @@ -637,24 +639,28 @@ async function storeFile(media, options) { } async function fetchHttpSource(source, tempFileTarget, hashStream) { - const res = await http.get(source.src, { - limits: 'media', + const res = await unprint.get(source.src, { + interface: 'request', + limiter: 'media', headers: { + 'user-agent': mediaUserAgent, // explicit host not allowed in HTTP/2 // host: new URL(source.src).hostname, // ...(source.host && { host: source.host }), ...(source.referer && { referer: source.referer }), }, - stream: true, // sources are fetched in parallel, don't gobble up memory - followRedirects: source.followRedirects, + tls: { rejectUnauthorized: false }, + // sources are fetched in parallel, stream to file instead of gobbling up memory transforms: [hashStream], destination: tempFileTarget, - ...(source.interval && { interval: source.interval }), - ...(source.concurrency && { concurrency: source.concurrency }), + // undefined would override the defaults + ...(typeof source.followRedirects === 'boolean' && { followRedirects: source.followRedirects }), + ...((argv.mediaInterval ?? source.interval) && { interval: argv.mediaInterval ?? source.interval }), + ...((argv.mediaConcurrency ?? source.concurrency) && { concurrency: argv.mediaConcurrency ?? source.concurrency }), }); if (!res.ok) { - throw new Error(`Response ${res.status} not OK`); + throw new Error(`Response ${res.status} not OK${res.statusText ? ` (${res.statusText})` : ''}`); // status 200 if stream was interrupted } return { @@ -772,6 +778,8 @@ async function fetchSource(source, baseMedia) { hashStream.destroy(); tempFileTarget?.destroy(); + await fsPromises.unlink(tempFilePath).catch(() => { /* probably never created */ }); + if (error.code !== 'VERIFY_TYPE') { logger.warn(`Failed attempt ${attempts}/${maxAttempts} to fetch ${source.src}: ${error.message}`); @@ -914,6 +922,9 @@ async function storeMedias(baseMedias, options) { async (baseMedia) => storeFile(baseMedia, options), { concurrency: 100 }, // don't overload disk ); + } else { + // duplicates aren't stored, clean up their fetched temp files + await Promise.all(existingHashMedias.map(async (media) => media.file?.path && fsPromises.unlink(media.file.path).catch(() => { /* probably already removed */ }))); } const newMediaWithEntries = savedMedias.filter(Boolean).map((media, index) => curateMediaEntry(media, index)); diff --git a/src/utils/http-windows.js b/src/utils/http-windows.js deleted file mode 100755 index 4fd928aa..00000000 --- a/src/utils/http-windows.js +++ /dev/null @@ -1,3 +0,0 @@ -'use strict'; - -module.exports = new Map(); diff --git a/src/utils/http.js b/src/utils/http.js deleted file mode 100755 index 4a41f613..00000000 --- a/src/utils/http.js +++ /dev/null @@ -1,245 +0,0 @@ -'use strict'; - -const config = require('config'); -const Promise = require('bluebird'); -const undici = require('undici'); -const fs = require('fs').promises; -// const util = require('util'); -// const stream = require('stream'); -const { pipeline } = require('stream/promises'); -const Bottleneck = require('bottleneck'); -const { JSDOM } = require('jsdom'); - -const windows = require('./http-windows'); - -const logger = require('../logger')(__filename); -const virtualConsole = require('./virtual-console')(__filename); -const argv = require('../argv'); - -// const pipeline = util.promisify(stream.pipeline); - -const limiters = {}; - -Promise.config({ - cancellation: true, -}); - -const defaultOptions = { - timeout: argv.requestTimeout, - encodeJSON: true, - parse: false, - headers: { - 'user-agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36', - }, -}; - -function useProxy(url) { - if (!config.proxy.enable) { - return false; - } - - const { hostname } = new URL(url); - - return config.proxy.hostnames.includes(hostname); -} - -const propMap = { - media: { - interval: 'mediaInterval', - concurrency: 'mediaConcurrency', - }, -}; - -function getLimiterValue(prop, options, hostname) { - const mappedProp = propMap[options.limits]?.[prop] || prop; - - if (typeof argv[mappedProp] !== 'undefined') { - return argv[mappedProp]; - } - - if (typeof options[prop] !== 'undefined') { - return options[prop]; - } - - if (config.limits[hostname]?.enable !== false && typeof config.limits[hostname]?.[prop] !== 'undefined') { - return config.limits[hostname][prop]; - } - - return config.limits[options.limits || 'default'][prop]; -} - -function getLimiter(options = {}, url) { - const { hostname } = new URL(url); - - const interval = getLimiterValue('interval', options, hostname); - const concurrency = getLimiterValue('concurrency', options, hostname); - - if (!limiters[interval]?.[concurrency]) { - limiters[interval] = limiters[interval] || {}; - - limiters[interval][concurrency] = new Bottleneck({ - minTime: interval, - maxConcurrent: concurrency, - timeout: (options.timeout || defaultOptions.timeout) + 10000, // timeout 10 seconds after bhttp should - }); - } - - return limiters[interval][concurrency]; -} - -const defaultAgent = new undici.Agent({ - allowH2: true, - connect: { - rejectUnauthorized: false, - }, -}); - -const proxyAgent = new undici.ProxyAgent({ - uri: `http://${config.proxy.host}:${config.proxy.port}`, -}); - -async function request(method = 'get', url, body, requestOptions = {}, limiter, redirects = 0) { - const withProxy = useProxy(url); - - logger.debug(`${redirects > 0 ? 'REDIRECT ' : ''}${method.toUpperCase()} (${limiter._store.storeOptions.minTime}ms/${limiter._store.storeOptions.maxConcurrent}p${withProxy ? ' proxy' : ''}) ${url}`); - - const headers = { - ...requestOptions.headers, - }; - - const res = await undici.request(url, { - method: method.toUpperCase(), - headers, - body: body ?? null, - dispatcher: withProxy - ? proxyAgent - : defaultAgent, - maxRedirections: 0, // handle manually - }); - - if (res.headers.location && redirects < 3) { - // Drain the body to free the socket before redirecting - await res.body.dump(); - const nextUrl = new URL(res.headers.location, url).href; - return request(method, nextUrl, body, requestOptions, limiter, redirects + 1); - } - - return res; -} - -async function finalizeResult(res, url, options) { - if (options.destination) { - await pipeline( - res.body, - ...(options.transforms || []), - options.destination, - ); - - return { - statusCode: res.statusCode, - headers: res.headers, - status: res.statusCode, - ok: res.statusCode >= 200 && res.statusCode <= 299, - }; - } - - const buffer = await res.body.arrayBuffer(); - const html = Buffer.from(buffer).toString(); - const window = options?.parse ? new JSDOM(html, { virtualConsole, ...options.extract }).window : null; - const pathname = new URL(url).pathname.replace(/\//g, '_'); - - if (window && /fetchScene|fetchMovie/.test(new Error().stack)) { - windows.set(pathname, window); - } - - if (argv.saveHtml) { - await fs.writeFile(`./html/${pathname}.html`, html); - } - - return { - statusCode: res.statusCode, - status: res.statusCode, - headers: res.headers, - body: html, - html, - document: window?.document || null, - window, - ok: res.statusCode >= 200 && res.statusCode <= 299, - }; -} - -function getTimeout(options, url) { - return new Promise((_resolve, reject, onCancel) => { - const timeout = setTimeout(() => { - logger.debug(`Canceled timed out request to ${url}`); - reject(new Error(`URL ${url} timed out`)); - }, (options?.timeout || defaultOptions.timeout) + 10000); - - onCancel(() => { - clearTimeout(timeout); - }); - }); -} - -async function scheduleRequest(method = 'get', url, body, requestOptions = {}) { - if (typeof url !== 'string') { - console.trace(`Bad URL: ${JSON.stringify(url)}`); - } - - const options = { - ...defaultOptions, - ...requestOptions, - headers: { - ...Object.fromEntries(requestOptions.includeDefaultHeaders === false ? {} : Object.entries(defaultOptions.headers).map(([key, value]) => [key.toLowerCase(), value])), - ...Object.fromEntries(Object.entries(requestOptions.headers || {}).map(([key, value]) => [key.toLowerCase(), value])), - }, - responseTimeout: requestOptions.responseTimeout || requestOptions.timeout || defaultOptions.timeout, - stream: !!requestOptions.destination, - }; - - const limiter = getLimiter(options, url); - const timeout = getTimeout(options, url); - - const result = await limiter.schedule(async () => Promise.race([request(method, url, body, options, limiter), timeout])); - - timeout.cancel(); - - const curatedResult = await finalizeResult(result, url, options); - - logger.silly(`Response ${curatedResult.status} for ${method.toUpperCase()} ${url}`); - - return curatedResult; -} - -async function get(url, options) { - return scheduleRequest('get', url, null, options); -} - -async function post(url, body, options) { - return scheduleRequest('post', url, body, options); -} - -async function put(url, body, options) { - return scheduleRequest('put', url, body, options); -} - -async function patch(url, body, options) { - return scheduleRequest('patch', url, body, options); -} - -async function del(url, options) { - return scheduleRequest('delete', url, null, options); -} - -async function head(url, options) { - return scheduleRequest('head', url, null, options); -} - -module.exports = { - get, - head, - post, - delete: del, - put, - patch, -}; diff --git a/src/utils/virtual-console.js b/src/utils/virtual-console.js deleted file mode 100755 index 639ddae9..00000000 --- a/src/utils/virtual-console.js +++ /dev/null @@ -1,18 +0,0 @@ -'use strict'; - -const { VirtualConsole } = require('jsdom'); -const path = require('path'); - -const logger = require('../logger')(__filename); - -function getVirtualConsole(filepath) { - const virtualConsole = new VirtualConsole(); - const context = path.basename(filepath).replace(path.extname(filepath), ''); - - virtualConsole.on('error', (message) => logger.warn(`Error from JSDOM in ${context}: ${message}`)); - virtualConsole.on('jsdomError', (message) => logger.warn(`Error from JSDOM in ${context}: ${message}`)); - - return virtualConsole; -} - -module.exports = getVirtualConsole;