[pull] master from apify:master

docs/public-api/crawlee-utils.api.md

@@ -65,6 +65,12 @@ export enum EnqueueStrategy {
     SameOrigin = "same-origin"
 }
 
+// @public
+export function extractMicrodata(raw: string): Promise<MicrodataItem[]>;
+
+// @public (undocumented)
+export function extractMicrodata($: CheerioAPI): Promise<MicrodataItem[]>;
+
 // @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<string, MicrodataValue | MicrodataValue[]>;
+    type?: string[];
+}
+
+// @public
+export type MicrodataValue = string | MicrodataItem;
+
 // Not exported by the entry point; reachable only as a referenced type.
 // @public (undocumented)
 interface NestedSitemap {

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;
 };
 
 

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<string>();
 
     readonly #requests = new Map<string, InternalRequest>();
+
+    /**
+     * 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<string>();
     // 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);

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,
 };

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';

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<string, MicrodataValue | MicrodataValue[]>;
+}
+
+interface ExtractionContext {
+    $: CheerioAPI;
+    /** Built on first `itemref` lookup, which most documents never trigger. */
+    idIndex?: Map<string, Element>;
+}
+
+/**
+ * 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<MicrodataItem[]>;
+export async function extractMicrodata($: CheerioAPI): Promise<MicrodataItem[]>;
+export async function extractMicrodata(htmlOrCheerioElement: string | CheerioAPI): Promise<MicrodataItem[]> {
+    // 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<Element>): 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<string, Element> {
+    const index = new Map<string, Element>();
+
+    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<Element>): 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<string, MicrodataValue | MicrodataValue[]>,
+    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;
+}

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);
+});