diff --git a/docs/public-api/crawlee-utils.api.md b/docs/public-api/crawlee-utils.api.md index e9ab3d9c4057..f0f3601000d1 100644 --- a/docs/public-api/crawlee-utils.api.md +++ b/docs/public-api/crawlee-utils.api.md @@ -65,6 +65,12 @@ export enum EnqueueStrategy { SameOrigin = "same-origin" } +// @public +export function extractMicrodata(raw: string): Promise; + +// @public (undocumented) +export function extractMicrodata($: CheerioAPI): Promise; + // @public export function extractUrls(options: ExtractUrlsOptions): string[]; @@ -95,6 +101,16 @@ const LINKEDIN_REGEX: RegExp; // @public const LINKEDIN_REGEX_GLOBAL: RegExp; +// @public +export interface MicrodataItem { + id?: string; + properties: Record; + type?: string[]; +} + +// @public +export type MicrodataValue = string | MicrodataItem; + // Not exported by the entry point; reachable only as a referenced type. // @public (undocumented) interface NestedSitemap { diff --git a/docs/public-api/crawlee.api.md b/docs/public-api/crawlee.api.md index 86806b621e5d..0f03126c3686 100644 --- a/docs/public-api/crawlee.api.md +++ b/docs/public-api/crawlee.api.md @@ -5,6 +5,7 @@ ```ts import { downloadListOfUrls } from '@crawlee/utils'; +import { extractMicrodata } from '@crawlee/utils'; import { Log } from '@apify/log'; import { parseOpenGraph } from '@crawlee/utils'; import { playwrightUtils } from '@crawlee/playwright'; @@ -21,6 +22,7 @@ export const utils: { sleep: typeof sleep; downloadListOfUrls: typeof downloadListOfUrls; parseOpenGraph: typeof parseOpenGraph; + extractMicrodata: typeof extractMicrodata; }; diff --git a/packages/core/src/memory-storage/resource-clients/request-queue.ts b/packages/core/src/memory-storage/resource-clients/request-queue.ts index 5290d21b6a02..6dbd04218c47 100644 --- a/packages/core/src/memory-storage/resource-clients/request-queue.ts +++ b/packages/core/src/memory-storage/resource-clients/request-queue.ts @@ -46,11 +46,11 @@ export class RequestQueueBackend extends BaseClient implements storage.RequestQu pendingRequestCount = 0; /** * Serializes every operation that reads-then-writes this backend's shared queue state — the - * `requests` map, the `forefrontRequestIds` array, the `inProgressRequestIds` set and the request - * counts. Those mutations span `await` points, so without this mutex a concurrent operation could - * interleave and corrupt them (e.g. a head scan pruning `forefrontRequestIds` while - * `addBatchOfRequests` pushes to it). Held by every mutating method as well as by `isEmpty`/ - * `isFinished`, whose head scan also prunes `forefrontRequestIds`. + * `requests` map, the `pendingRequestIds` set, the `forefrontRequestIds` array, the + * `inProgressRequestIds` set and the request counts. Those mutations span `await` points, so + * without this mutex a concurrent operation could interleave and corrupt them (e.g. a head scan + * pruning `forefrontRequestIds` while `addBatchOfRequests` pushes to it). Held by every mutating + * method as well as by `isEmpty`/`isFinished`, whose head scan also prunes `forefrontRequestIds`. */ readonly #queueStateMutex = new AsyncQueue(); #forefrontRequestIds: string[] = []; @@ -66,6 +66,13 @@ export class RequestQueueBackend extends BaseClient implements storage.RequestQu readonly #inProgressRequestIds = new Set(); readonly #requests = new Map(); + + /** + * IDs of requests that are not yet handled (pending or in progress), in insertion order. Handled + * requests stay in `requests` for deduplication but are removed from here, so head scans only + * ever walk the unhandled tail instead of every request the queue has ever seen. + */ + readonly #pendingRequestIds = new Set(); // kept as TS-private: storage-backend tests read this field at runtime private readonly storageBackend: MemoryStorageBackend; @@ -94,6 +101,7 @@ export class RequestQueueBackend extends BaseClient implements storage.RequestQu // leave dangling ids in `forefrontRequestIds`/`inProgressRequestIds`, which a later head // scan would resolve to a missing request and dereference. this.#requests.clear(); + this.#pendingRequestIds.clear(); this.#forefrontRequestIds = []; this.#inProgressRequestIds.clear(); } @@ -110,6 +118,7 @@ export class RequestQueueBackend extends BaseClient implements storage.RequestQu try { // Clear all in-memory state this.#requests.clear(); + this.#pendingRequestIds.clear(); this.#forefrontRequestIds = []; this.#inProgressRequestIds.clear(); this.handledRequestCount = 0; @@ -126,9 +135,7 @@ export class RequestQueueBackend extends BaseClient implements storage.RequestQu yield this.#forefrontRequestIds[i]; } - for (const key of this.#requests.keys()) { - yield key; - } + yield* this.#pendingRequestIds; } /** @@ -144,7 +151,7 @@ export class RequestQueueBackend extends BaseClient implements storage.RequestQu * Computing the flag is expensive: because an in-progress request may sit anywhere in the queue, it * forces a scan of every pending entry even when only `limit` items are wanted. Callers that only * need the head (e.g. {@link fetchNextRequest}, {@link isEmpty}) leave it off so the scan can stop as - * soon as the page is filled, keeping those calls O(head) instead of O(N). + * soon as the page is filled, keeping those calls O(head) instead of O(pending). */ private async listPendingHead( limit: number, @@ -173,11 +180,10 @@ export class RequestQueueBackend extends BaseClient implements storage.RequestQu const request = this.#requests.get(requestId)!; - // Permanently-handled requests (`orderNo === null`) are in a terminal state and can be skipped. + // Only `forefrontRequestIds` can still reference a handled request (`orderNo === null`); + // `pendingRequestIds` drops them on handling. Remember the id so the list gets pruned below. if (request.orderNo === null) { - if (this.#forefrontRequestIds.includes(requestId)) { - handledForefrontIds.add(requestId); - } + handledForefrontIds.add(requestId); continue; } @@ -262,6 +268,7 @@ export class RequestQueueBackend extends BaseClient implements storage.RequestQu this.#requests.set(requestModel.id, requestModel); if (requestModel.orderNo) { + this.#pendingRequestIds.add(requestModel.id); this.pendingRequestCount += 1; } else { this.handledRequestCount += 1; @@ -325,6 +332,7 @@ export class RequestQueueBackend extends BaseClient implements storage.RequestQu const requestModel = this.createInternalRequest({ ...request, handledAt }, false); this.#requests.set(id, requestModel); + this.#pendingRequestIds.delete(id); // The request is no longer in progress for this client. this.#inProgressRequestIds.delete(id); diff --git a/packages/crawlee/src/index.ts b/packages/crawlee/src/index.ts index 83940ac1d5e0..54c490ccf719 100644 --- a/packages/crawlee/src/index.ts +++ b/packages/crawlee/src/index.ts @@ -1,7 +1,7 @@ import { log } from '@crawlee/core'; import { playwrightUtils } from '@crawlee/playwright'; import { puppeteerUtils } from '@crawlee/puppeteer'; -import { downloadListOfUrls, parseOpenGraph, sleep, social } from '@crawlee/utils'; +import { downloadListOfUrls, extractMicrodata, parseOpenGraph, sleep, social } from '@crawlee/utils'; export * from '@crawlee/core'; export * from '@crawlee/utils'; @@ -24,4 +24,5 @@ export const utils = { sleep, downloadListOfUrls, parseOpenGraph, + extractMicrodata, }; diff --git a/packages/utils/src/index.ts b/packages/utils/src/index.ts index 0a81aadf99dd..70f212653f6a 100644 --- a/packages/utils/src/index.ts +++ b/packages/utils/src/index.ts @@ -4,6 +4,7 @@ export { EnqueueStrategy } from './internals/url.js'; export type { DownloadListOfUrlsOptions, ExtractUrlsOptions } from './internals/extract-urls.js'; export { sleep, expandShadowRoots } from './internals/general.js'; export * as social from './internals/social.js'; +export * from './internals/extract-microdata.js'; export * from './internals/open_graph_parser.js'; export * from './internals/robots.js'; export * from './internals/sitemap.js'; diff --git a/packages/utils/src/internals/extract-microdata.ts b/packages/utils/src/internals/extract-microdata.ts new file mode 100644 index 000000000000..a9b222196104 --- /dev/null +++ b/packages/utils/src/internals/extract-microdata.ts @@ -0,0 +1,182 @@ +import type { CheerioAPI } from 'cheerio'; +import type { AnyNode, Element } from 'domhandler'; +import { isTag } from 'domhandler'; + +/** The value of a microdata property: either text or a nested item. */ +export type MicrodataValue = string | MicrodataItem; + +/** A single schema.org item extracted from a document. */ +export interface MicrodataItem { + /** Tokens of the item's `itemtype` attribute. */ + type?: string[]; + /** The item's `itemid` attribute. */ + id?: string; + /** Values keyed by `itemprop` name, an array where the property repeats. */ + properties: Record; +} + +interface ExtractionContext { + $: CheerioAPI; + /** Built on first `itemref` lookup, which most documents never trigger. */ + idIndex?: Map; +} + +/** + * Easily parse all schema.org microdata from a page with just a `CheerioAPI` object or raw HTML, + * following the [microdata processing model](https://html.spec.whatwg.org/multipage/microdata.html#microdata). + * + * Text values are trimmed and their inner whitespace collapsed. URL-valued attributes are returned + * verbatim rather than resolved against the document's base URL. + * + * @param htmlOrCheerioElement A `CheerioAPI` object, or a string of raw HTML. + * @returns The document's top-level items. Nested items are the property values of their parent. + */ +export async function extractMicrodata(raw: string): Promise; +export async function extractMicrodata($: CheerioAPI): Promise; +export async function extractMicrodata(htmlOrCheerioElement: string | CheerioAPI): Promise { + // Dynamic so that importing `@crawlee/utils` does not pull in cheerio - see #3836. + const { load } = await import('cheerio'); + const $ = typeof htmlOrCheerioElement === 'string' ? load(htmlOrCheerioElement) : htmlOrCheerioElement; + const context: ExtractionContext = { $ }; + + return $('[itemscope]') + .toArray() + .filter((element) => !('itemprop' in element.attribs)) + .map((element) => parseItem(context, element, new Set())); +} + +function parseItem(context: ExtractionContext, element: Element, ancestors: Set): MicrodataItem { + const item: MicrodataItem = { properties: {} }; + const type = uniqueTokens(element.attribs.itemtype); + const id = element.attribs.itemid; + + if (type.length > 0) { + item.type = type; + } + + if (id) { + item.id = id.trim(); + } + + ancestors.add(element); + + for (const propertyElement of collectPropertyElements(context, element)) { + const value = getPropertyValue(context, propertyElement, ancestors); + + for (const name of uniqueTokens(propertyElement.attribs.itemprop)) { + addProperty(item.properties, name, value); + } + } + + ancestors.delete(element); + + return item; +} + +function collectPropertyElements(context: ExtractionContext, scope: Element): Element[] { + const elements: Element[] = []; + + collectFromNodes(scope.children, elements); + + for (const id of uniqueTokens(scope.attribs.itemref)) { + const referenced = (context.idIndex ??= indexIds(context.$)).get(id); + + if (referenced) { + collectFromNodes([referenced], elements); + } + } + + return elements; +} + +function collectFromNodes(nodes: AnyNode[], elements: Element[]): void { + for (const node of nodes) { + if (!isTag(node)) { + continue; + } + + if ('itemprop' in node.attribs) { + elements.push(node); + } + + // A nested item owns everything below it, so its subtree is not part of the enclosing item. + if (!('itemscope' in node.attribs)) { + collectFromNodes(node.children, elements); + } + } +} + +function indexIds($: CheerioAPI): Map { + const index = new Map(); + + for (const element of $('[id]').toArray()) { + // Duplicate ids are invalid HTML; `getElementById` resolves them to the first element. + if (!index.has(element.attribs.id)) { + index.set(element.attribs.id, element); + } + } + + return index; +} + +function getPropertyValue(context: ExtractionContext, element: Element, ancestors: Set): MicrodataValue { + if ('itemscope' in element.attribs) { + // `itemref` can point back at an enclosing item, which the spec treats as an error. + return ancestors.has(element) ? { properties: {} } : parseItem(context, element, ancestors); + } + + const { attribs } = element; + + switch (element.tagName.toLowerCase()) { + case 'meta': + return attribs.content ?? ''; + case 'audio': + case 'embed': + case 'iframe': + case 'img': + case 'source': + case 'track': + case 'video': + return attribs.src ?? ''; + case 'a': + case 'area': + case 'link': + return attribs.href ?? ''; + case 'object': + return attribs.data ?? ''; + case 'data': + case 'meter': + return attribs.value ?? ''; + case 'time': + return attribs.datetime ?? context.$(element).text().replace(/\s+/g, ' ').trim(); + default: + return context.$(element).text().replace(/\s+/g, ' ').trim(); + } +} + +function addProperty( + properties: Record, + name: string, + value: MicrodataValue, +): void { + const existing = properties[name]; + + if (existing === undefined) { + properties[name] = value; + } else if (Array.isArray(existing)) { + existing.push(value); + } else { + properties[name] = [existing, value]; + } +} + +/** `itemtype`, `itemprop` and `itemref` are all unordered sets of unique space-separated tokens. */ +function uniqueTokens(value: string | undefined): string[] { + if (!value) { + return []; + } + + const tokens = value.split(/\s+/).filter(Boolean); + + return tokens.length > 1 ? [...new Set(tokens)] : tokens; +} diff --git a/test/core/storages/request_queue.test.ts b/test/core/storages/request_queue.test.ts index ec0f82db163b..6d8a336d7249 100644 --- a/test/core/storages/request_queue.test.ts +++ b/test/core/storages/request_queue.test.ts @@ -703,3 +703,39 @@ describe('RequestQueue background batches', () => { expect((await queue.checkReadiness()).status).toBe('finished'); }, 10_000); }); + +describe('MemoryStorageBackend request queue', () => { + test('head operations do not scan already-handled requests', async () => { + const backend = new MemoryStorageBackend(); + const queue = await backend.createRequestQueueBackend({ name: 'handled-scan' }); + + // Simulate a crawl nearing its end: a large number of handled requests and few pending ones. + const handledAt = new Date().toISOString(); + const handledCount = 200_000; + for (let i = 0; i < handledCount; i += 1_000) { + await queue.addBatchOfRequests( + Array.from({ length: 1_000 }, (_, j) => ({ + url: `http://example.com/${i + j}`, + uniqueKey: `handled-${i + j}`, + handledAt, + })), + ); + } + expect((await queue.getMetadata()).handledRequestCount).toBe(handledCount); + + // Each call used to walk the whole map (O(handled)): ~8s for this loop vs ~5ms once only pending + // requests are scanned. The loose budget only separates those two regimes, well above CI noise. + const start = performance.now(); + for (let i = 0; i < 200; i++) { + await queue.addBatchOfRequests([{ url: `http://example.com/new-${i}`, uniqueKey: `new-${i}` }]); + expect(await queue.isEmpty()).toBe(false); + const request = await queue.fetchNextRequest(); + expect(request!.uniqueKey).toBe(`new-${i}`); + expect(await queue.isFinished()).toBe(false); + await queue.markRequestAsHandled({ ...request!, handledAt }); + } + expect(await queue.isEmpty()).toBe(true); + expect(await queue.isFinished()).toBe(true); + expect(performance.now() - start).toBeLessThan(500); + }, 60_000); +}); diff --git a/test/utils/extract-microdata.test.ts b/test/utils/extract-microdata.test.ts new file mode 100644 index 000000000000..37491f9f8ff8 --- /dev/null +++ b/test/utils/extract-microdata.test.ts @@ -0,0 +1,157 @@ +import { extractMicrodata } from '@crawlee/utils'; +import { load } from 'cheerio'; + +describe('extractMicrodata', () => { + it('returns no items for a document without microdata', async () => { + expect(await extractMicrodata('

Nothing to see here

')).toEqual([]); + }); + + it('accepts a CheerioAPI object', async () => { + const $ = load('
Existing DOM
'); + + expect(await extractMicrodata($)).toEqual([{ properties: { name: 'Existing DOM' } }]); + }); + + it('splits itemtype and itemprop token sets', async () => { + const items = await extractMicrodata(` +
+ orange +
`); + + expect(items).toEqual([ + { + type: ['https://schema.org/Product', 'https://schema.org/Thing'], + id: 'urn:isbn:1', + properties: { 'favorite-color': 'orange', 'favorite-fruit': 'orange' }, + }, + ]); + }); + + it('reads the value from the attribute belonging to each element type', async () => { + const items = await extractMicrodata(` +
+ + A lamp + + + $19.99 + Great + A 3D model +

Collapses any + inner whitespace

+ Not a link +
`); + + expect(items[0].properties).toEqual({ + sku: 'LAMP-1', + image: '/lamp.jpg', + url: '/products/lamp', + releaseDate: '2026-08-29', + price: '19.99', + rating: '4.5', + model: '/lamp.glb', + description: 'Collapses any inner whitespace', + permalink: '', + }); + }); + + it('groups a repeated property into an array', async () => { + const items = await extractMicrodata(` +
+ first + second + third +
`); + + expect(items[0].properties).toEqual({ tag: ['first', 'second', 'third'] }); + }); + + it('nests an item used as a property value', async () => { + const items = await extractMicrodata(` +
+ Coffee +
+ + 12 +
+
`); + + expect(items).toEqual([ + { + type: ['https://schema.org/Product'], + properties: { + name: 'Coffee', + offers: { + type: ['https://schema.org/Offer'], + properties: { priceCurrency: 'USD', price: '12' }, + }, + }, + }, + ]); + }); + + it('treats a nested itemscope without itemprop as a separate top-level item', async () => { + const items = await extractMicrodata(` +
+ Outer +
+ Inner +
+
`); + + expect(items).toEqual([ + { type: ['https://schema.org/Article'], properties: { headline: 'Outer' } }, + { type: ['https://schema.org/Comment'], properties: { text: 'Inner' } }, + ]); + }); + + it('collects properties referenced with itemref', async () => { + const items = await extractMicrodata(` +
+ Engineer +
+

Ada Lovelace

+
+ +
`); + + expect(items).toEqual([ + { + type: ['https://schema.org/Person'], + properties: { + jobTitle: 'Engineer', + name: 'Ada Lovelace', + url: 'https://example.com/ada', + }, + }, + ]); + }); + + it('stops at an itemref cycle instead of recursing forever', async () => { + // `loop-a` and `loop-b` reference each other, so the second visit of `loop-a` must be cut short. + const items = await extractMicrodata(` +
+ root +
+
+ a +
+
+ b +
`); + + expect(items).toEqual([ + { + properties: { + name: 'root', + self: { + properties: { + name: 'a', + self: { properties: { name: 'b', self: { properties: {} } } }, + }, + }, + }, + }, + ]); + }); +});