Retired http utility in favor of unprint.
This commit is contained in:
DebaucheryLibrarian
+2
-2
@@ -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();
|
||||
|
||||
-10
@@ -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})`);
|
||||
|
||||
+19
-8
@@ -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));
|
||||
|
||||
@@ -1,3 +0,0 @@
|
||||
'use strict';
|
||||
|
||||
module.exports = new Map();
|
||||
@@ -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,
|
||||
};
|
||||
@@ -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;
|
||||
Reference in New Issue
Block a user