Function _process_sitemap_item

src/crawlee/_utils/sitemap.py:229–284

Process a sitemap item and yield appropriate results.

(
    item: _SitemapItem,
    source: SitemapSource,
    depth: int,
    visited_sitemap_urls: set[str],
    sources: list[SitemapSource],
    *,
    emit_nested_sitemaps: bool,
    enqueue_strategy: EnqueueStrategy,
)

async def _process_sitemap_item(
    item: _SitemapItem,
    source: SitemapSource,
    depth: int,
    visited_sitemap_urls: set[str],
    sources: list[SitemapSource],
    *,
    emit_nested_sitemaps: bool,
    enqueue_strategy: EnqueueStrategy,
) -> AsyncGenerator[SitemapUrl | NestedSitemap | None, None]:
    """Process a sitemap item and yield appropriate results."""
    item_copy = item.copy()  # Work with a copy to avoid modifying the original

    if 'type' not in item_copy:
        return

    item_type = item_copy.pop('type')

    # Handle sitemap URL references (nested sitemaps)
    if item_type == 'sitemap_url' and 'url' in item_copy:
        sitemap_url = item_copy['url']
        if sitemap_url and sitemap_url not in visited_sitemap_urls:
            if parent_url := source.get('url'):
                ok, reason = filter_url(target=sitemap_url, strategy=enqueue_strategy, origin=parent_url)
                if not ok:
                    logger.warning(f'Skipping nested sitemap {sitemap_url!r} (parent {parent_url!r}): {reason}.')
                    return

            # Add to processing queue
            sources.append(SitemapSource(type='url', url=sitemap_url, depth=depth + 1))

            # Output the nested sitemap reference if requested
            if emit_nested_sitemaps:
                yield NestedSitemap(loc=sitemap_url, origin_sitemap_url=parent_url)

    # Handle individual URL entries
    elif item_type == 'url' and 'loc' in item_copy:
        # Determine the origin sitemap URL for tracking purposes
        origin_url = _get_origin_url(source)

        loc = item_copy['loc']
        parent_url = source.get('url')
        if parent_url and loc:
            ok, reason = filter_url(target=loc, strategy=enqueue_strategy, origin=parent_url)
            if not ok:
                logger.warning(f'Skipping sitemap URL {loc!r} (parent {parent_url!r}): {reason}.')
                return

        # Create and yield the sitemap URL object
        yield SitemapUrl(
            loc=loc,
            lastmod=item_copy.get('lastmod'),
            changefreq=item_copy.get('changefreq'),
            priority=item_copy.get('priority'),
            origin_sitemap_url=origin_url,
        )