Function parse_sitemap

src/crawlee/_utils/sitemap.py:465–535

Parse sitemap(s) and yield URLs found in them. This function coordinates the process of fetching and parsing sitemaps, handling both URL-based and raw content sources. It follows nested sitemaps up to the specified maximum depth. Default `ParseSitemapOptions.enqueue_strategy` is `s

(
    initial_sources: list[SitemapSource],
    http_client: HttpClient | None = None,
    proxy_info: ProxyInfo | None = None,
    options: ParseSitemapOptions | None = None,
)

async def parse_sitemap(
    initial_sources: list[SitemapSource],
    http_client: HttpClient | None = None,
    proxy_info: ProxyInfo | None = None,
    options: ParseSitemapOptions | None = None,
) -> AsyncGenerator[SitemapUrl | NestedSitemap, None]:
    """Parse sitemap(s) and yield URLs found in them.

    This function coordinates the process of fetching and parsing sitemaps,
    handling both URL-based and raw content sources. It follows nested sitemaps
    up to the specified maximum depth.

    Default `ParseSitemapOptions.enqueue_strategy` is `same-hostname` which will skip cross-host URLs.
    Use strategy `all` to process all links.
    """
    # Set default options
    options = options or {}
    emit_nested_sitemaps = options.get('emit_nested_sitemaps', False)
    max_depth = options.get('max_depth', float('inf'))
    sitemap_retries = options.get('sitemap_retries', 3)
    timeout = options.get('timeout', timedelta(seconds=30))
    enqueue_strategy = options.get('enqueue_strategy', 'same-hostname')

    # Setup working state
    sources = list(initial_sources)
    visited_sitemap_urls: set[str] = set()

    # Process sources until the queue is empty
    while sources:
        source = sources.pop(0)
        depth = source.get('depth', 0)

        # Skip if we've reached max depth
        if depth > max_depth:
            logger.debug(f'Skipping sitemap {source.get("url", "")} - exceeded max depth {max_depth}')
            continue

        # Process based on source type
        if source['type'] == 'raw':
            async for result in _process_raw_source(
                source,
                depth,
                visited_sitemap_urls,
                sources,
                emit_nested_sitemaps=emit_nested_sitemaps,
                enqueue_strategy=enqueue_strategy,
            ):
                yield result

        elif source['type'] == 'url' and 'url' in source:
            # Add to visited set before processing to avoid duplicates
            if http_client is None:
                raise RuntimeError('HttpClient must be provided for URL-based sitemap sources.')

            visited_sitemap_urls.add(source['url'])

            async for result in _fetch_and_process_sitemap(
                http_client,