API Reference¶

async def start(self):
    """Starts the browser and initializes the configured scraping strategy.

    Raises:
        ValueError: If ``scraping_strategy`` is ``"genai"`` but
            ``settings.genai`` is not configured, or an unknown strategy
            is requested.
    """
    self._closed = False

    scraping_strategy = getattr(
        self.settings, "scraping_strategy", ScrapingStrategy.HEURISTIC
    )
    if not isinstance(scraping_strategy, str):
        scraping_strategy = ScrapingStrategy.HEURISTIC

    if scraping_strategy == ScrapingStrategy.GENAI:
        if not getattr(self.settings, "genai", None):
            raise ValueError("GenAI settings is required for GenAI strategy")
    elif scraping_strategy not in (
        ScrapingStrategy.HEURISTIC,
        ScrapingStrategy.MARKDOWNIFY,
    ):
        raise ValueError(f"Unknown strategy: {scraping_strategy}")

    self.browser = GoogleChrome(
        self.settings.browser_settings,
        proxy_pool=self.settings.create_proxy_pool(),
    )
    await self.browser.start()

    if scraping_strategy == ScrapingStrategy.HEURISTIC:
        strategy = HeuristicStrategy(
            settings=self.settings,
            browser=self.browser,
        )
    elif scraping_strategy == ScrapingStrategy.MARKDOWNIFY:
        strategy = MarkdownifyStrategy(
            settings=self.settings,
            browser=self.browser,
        )
    else:
        strategy = LLMStrategy(
            provider=self.settings.genai.provider,
            model_name=self.settings.genai.model_name,
            max_retries=self.settings.max_retries,
            api_key=self.settings.genai.api_key,
            base_url=self.settings.genai.base_url,
            output_schema=self.settings.genai.output_schema,
            provider_kwargs=self.settings.genai.provider_kwargs,
            timeout=self.settings.genai.timeout,
            think=self.settings.genai.think,
            exclude_selectors=self.settings.exclude_selectors,
            browser=self.browser,
        )
        await strategy.initialize()

    self.strategy = strategy

async def close(self):
    """Closes the scraping strategy and the browser."""
    if self.strategy:
        await self.strategy.close()

    if self.browser:
        await self.browser.close()

async def run(self, url: str, filters=None) -> list[dict]:
    """Crawls ``url`` and collects all extracted content.

    Args:
        url (str): The starting URL; the crawl stays within its origin.
        filters (Optional[Callable[[dict], bool]]): Predicate applied to
            each extracted content dict post-extraction, pre-collection;
            items failing it are dropped. See ``onecrawler.filters``.

    Returns:
        list[dict]: Extracted content dicts, each including its source
        ``url``.
    """
    self._ensure_open()
    runtime, pool = await self._create_runtime(url, content_filter=filters)
    try:
        return await runtime.run()
    finally:
        await pool.close()

async def stream(self, url: str, filters=None) -> AsyncGenerator[dict, None]:
    """Crawls ``url``, yielding extracted content dicts as they arrive.

    Args:
        url (str): The starting URL; the crawl stays within its origin.
        filters (Optional[Callable[[dict], bool]]): Predicate applied to
            each extracted content dict post-extraction, pre-yield; items
            failing it are dropped. See ``onecrawler.filters``.

    Yields:
        dict: An extracted content dict, including its source ``url``.
    """
    self._ensure_open()
    runtime, pool = await self._create_runtime(
        url, streaming=True, content_filter=filters
    )
    try:
        async for item in runtime.stream():
            yield item
    finally:
        await pool.close()

async def start(self):
    """Starts the engine and initializes the browser."""
    self._closed = False
    self.browser = GoogleChrome(
        self.settings.browser_settings,
        proxy_pool=self.settings.create_proxy_pool(),
    )
    await self.browser.start()

async def close(self):
    """Closes the engine and releases browser resources."""
    if hasattr(self, "browser") and self.browser:
        await self.browser.close()

async def run(self, url: str) -> List[str]:
    """Runs the link extraction for the given URL.

    Args:
        url (str): The starting URL.

    Returns:
        List[str]: A list of absolute URLs discovered.

    Raises:
        ValueError: If an unknown strategy is configured.
    """
    self._ensure_open()

    strategy = self.settings.link_extraction_strategy

    self.logger.info(
        "Running link extraction on %s with strategy: %s",
        url,
        strategy,
    )

    if strategy == LinkExtractionStrategy.SHALLOW:
        return await extract_url_from_current_page(
            url=url,
            browser=self.browser,
            include_link_patterns=self.settings.include_link_patterns,
            exclude_link_patterns=self.settings.exclude_link_patterns,
            max_links=self.settings.link_extraction_limit,
        )

    if strategy != LinkExtractionStrategy.DEEP:
        raise ValueError(f"Unknown strategy: {strategy}")

    results = []
    async for link in self.stream(url):
        results.append(link)

    return results

async def stream(self, url: str) -> AsyncGenerator[str, None]:
    """Streams discovered links incrementally (deep strategy only).

    Args:
        url (str): The starting URL.

    Yields:
        str: Discovered absolute URL.

    Raises:
        AssertionError: If strategy is 'shallow'.
        ValueError: If an unknown strategy is configured.
    """
    self._ensure_open()

    strategy = self.settings.link_extraction_strategy

    assert (
        strategy != LinkExtractionStrategy.SHALLOW
    ), "Shallow link extraction does not support stream"

    self.logger.info(
        "Running link extraction stream on %s with strategy: %s",
        url,
        strategy,
    )

    if strategy != LinkExtractionStrategy.DEEP:
        raise ValueError(f"Unknown strategy: {strategy}")

    parsed = urlparse(url)
    base_prefix = f"{parsed.scheme}://{parsed.netloc}"

    scheduler = BFScheduler(url)
    spider = LinkSpider(base_prefix)
    pool = BrowserPool(
        self.browser,
        self.settings.concurrency,
    )

    await pool.init()

    runtime = BFSRuntime(
        scheduler=scheduler,
        pool=pool,
        spider=spider,
        base_prefix=base_prefix,
        max_links=self.settings.link_extraction_limit,
        include_pattern=self.settings.include_link_patterns,
        exclude_pattern=self.settings.exclude_link_patterns,
        enable_human_behaviors=self.settings.human_behavior_settings is not None,
        human_behavior_settings=self.settings.human_behavior_settings,
        concurrency=self.settings.concurrency,
        streaming=True,
        wait_until=self.settings.browser_settings.wait_until,
        timeout=self.settings.browser_settings.timeout,
        show_progress=self.settings.show_progress,
    )

    try:
        async for link in runtime.stream():
            yield link

    finally:
        await pool.close()

async def start(self):
    """Starts the scraper engine."""
    self._closed = False

    if self.settings.browser_settings:
        self.browser = GoogleChrome(
            settings=self.settings.browser_settings,
            proxy_pool=self.settings.create_proxy_pool(),
        )
        await self.browser.start()

    if self.settings.scraping_strategy == ScrapingStrategy.HEURISTIC:
        self.strategy = HeuristicStrategy(
            settings=self.settings,
            browser=self.browser,
        )

    elif self.settings.scraping_strategy == ScrapingStrategy.MARKDOWNIFY:
        self.strategy = MarkdownifyStrategy(
            settings=self.settings,
            browser=self.browser,
        )

    elif self.settings.scraping_strategy == ScrapingStrategy.GENAI:
        if not self.settings.genai:
            raise ValueError("GenAI settings are required for GenAI strategy")

        self.strategy = LLMStrategy(
            provider=self.settings.genai.provider,
            model_name=self.settings.genai.model_name,
            max_retries=self.settings.max_retries,
            api_key=self.settings.genai.api_key,
            base_url=self.settings.genai.base_url,
            output_schema=self.settings.genai.output_schema,
            provider_kwargs=self.settings.genai.provider_kwargs,
            timeout=self.settings.genai.timeout,
            think=self.settings.genai.think,
            exclude_selectors=self.settings.exclude_selectors,
            browser=self.browser,
        )

        await self.strategy.initialize()

    else:
        raise ValueError(
            f"Unknown scraping strategy: {self.settings.scraping_strategy}"
        )

    self.logger.info(
        f"Scraper started using {self.settings.scraping_strategy} strategy"
    )

async def close(self):
    """Closes the scraper engine."""
    if self.strategy:
        await self.strategy.close()

    if self.browser:
        await self.browser.close()

    self.logger.info("Scraper closed")

async def stream(self, link: Union[str, List[str]]) -> AsyncGenerator[Any, None]:
    """Streams extracted results as they complete, in completion order.

    Args:
        link (Union[str, List[str]]): One URL or a list of URLs to scrape.

    Yields:
        dict: ``{"url": str, "result": Any}``, where ``result``'s shape
        depends on ``settings.scraping_strategy``/``scraping_output_format``
        (a dict, plain text, or a GenAI ``output_schema`` model instance).
        URLs that fail extraction after retries are silently skipped.
    """
    self._ensure_open()
    links = link if isinstance(link, list) else [link]

    self.logger.info(
        f"Streaming scraper on {len(links)} link(s) "
        f"using {self.settings.scraping_strategy}"
    )

    tasks = [asyncio.create_task(self._process(url)) for url in links]
    show_progress = getattr(
        self.settings,
        "show_progress",
        True,
    )

    with make_progress_bar(
        total=len(tasks),
        desc="Scraping",
        unit="url",
        show_progress=show_progress,
    ) as pbar:
        for task in asyncio.as_completed(tasks):
            try:
                result = await task

                if result is not None:
                    yield result

            except asyncio.CancelledError:
                raise

            except Exception as e:
                self.logger.warning(
                    f"Streaming task failed: [{type(e).__name__}] {e}"
                )

            finally:
                pbar.update(1)

    self.logger.info("Streaming scrape completed")

async def run(self, link: Union[str, List[str]]) -> Union[dict, List[dict], None]:
    """Runs the scraper and collects all results.

    Args:
        link (Union[str, List[str]]): One URL or a list of URLs to scrape.

    Returns:
        Union[dict, List[dict], None]: A list of ``{"url": str, "result":
        Any}`` dicts when ``link`` is a list (possibly shorter than the
        input if some URLs failed extraction); a single such dict (or
        ``None``) when ``link`` is a single URL.
    """
    results = []
    async for result in self.stream(link):
        results.append(result)

    return results if isinstance(link, list) else (results[0] if results else None)

async def run(self, url: str) -> List[str]:
    """Runs the universal sitemap discovery process.

    Args:
        url (str): The starting URL.

    Returns:
        List[str]: A list of discovered absolute internal URLs.
    """
    base_url = self._normalize_base(url)

    strategies_used: list[str] = []
    all_records: list[URLRecord] = []

    pbar = make_progress_bar(
        total=self.settings.link_extraction_limit,
        desc="Sitemap Discovery",
        unit="url",
        show_progress=self.settings.show_progress,
    )

    def _bump(n: int = 1) -> None:
        remaining = pbar.total - pbar.n
        if remaining > 0:
            pbar.update(min(n, remaining))

    try:
        async with HTTPClient(
            self.settings.concurrency,
            self.settings.request_timeout,
            self.settings.sitemap.user_agent,
            self.settings.max_retries,
            self.settings.retry_delay,
            self.settings.create_proxy_pool(),
        ) as client:
            robots = RobotsParser(client)
            sitemap_parser = SitemapParser(
                client,
                self.settings.concurrency,
                follow_index=self.settings.sitemap.follow_index,
            )

            sitemap_urls = await robots.fetch_sitemaps(base_url)
            if sitemap_urls:
                strategies_used.append("robots.txt")

            probe_tasks = [
                self._probe_url(client, urljoin(base_url, path))
                for path in COMMON_SITEMAP_PATHS
            ]
            probe_results = await asyncio.gather(*probe_tasks)
            found_common = [
                urljoin(base_url, path)
                for path, ok in zip(COMMON_SITEMAP_PATHS, probe_results)
                if ok
            ]
            new_common = [u for u in found_common if u not in sitemap_urls]
            if new_common:
                strategies_used.append("common_paths")
                sitemap_urls.extend(new_common)

            if sitemap_urls:
                strategies_used.append("sitemap_xml")

                # Only let parse_all() stop early when no downstream filter
                # can unpredictably drop a large fraction of records — those
                # filters run after this point and could otherwise leave us
                # under the requested limit despite more URLs being available.
                # robots.txt disallow rules are excluded from this check: they
                # typically remove a small, roughly uniform slice of a site's
                # URL space (unlike a narrow date range or include pattern,
                # which can trivially drop the vast majority of records), so
                # the overfetch margin below is expected to absorb them too.
                filters_may_shrink_results = (
                    self.settings.sitemap.start_date is not None
                    or self.settings.sitemap.end_date is not None
                    or bool(self.settings.include_link_patterns)
                )
                target_count = (
                    None
                    if filters_may_shrink_results
                    else self.settings.link_extraction_limit
                    * _SITEMAP_LIMIT_OVERFETCH_FACTOR
                )

                records, _ = await sitemap_parser.parse_all(
                    sitemap_urls, target_count=target_count, on_records=_bump
                )
                all_records.extend(records)

            if not all_records and self.settings.sitemap.html_fallback:
                strategies_used.append("html_crawl")
                crawler = HTMLCrawler(
                    client,
                    self.settings.concurrency,
                    self.settings.sitemap.max_pages,
                    self.settings.sitemap.max_depth,
                    robots=robots if self.settings.sitemap.respect_robots else None,
                )
                crawl_records = await crawler.crawl(base_url, on_record=_bump)
                all_records.extend(crawl_records)

            # Respect robots.txt for sitemap-sourced URLs too (the HTML crawl
            # fallback above already gates each fetch as it happens).
            if self.settings.sitemap.respect_robots and all_records:
                allowed_flags = await asyncio.gather(
                    *(robots.is_allowed(rec.url, base_url) for rec in all_records)
                )
                all_records = [
                    rec for rec, ok in zip(all_records, allowed_flags) if ok
                ]
    finally:
        pbar.close()

    logging.info(f"Strategies used: {strategies_used}")

    start_date = self.settings.sitemap.start_date
    end_date = self.settings.sitemap.end_date
    strict_date_filter = self.settings.sitemap.strict_date_filter
    if start_date is not None or end_date is not None:
        filtered: list[URLRecord] = []
        for rec in all_records:
            lm = self._parse_lastmod(rec.lastmod)
            if lm is None:
                if not strict_date_filter:
                    filtered.append(rec)
                continue
            if start_date is not None and lm < start_date:
                continue
            if end_date is not None and lm > end_date:
                continue
            filtered.append(rec)
        all_records = filtered

    all_records = [r for r in all_records if not is_xml_url(r.url)]

    if self.settings.include_link_patterns:
        all_records = [
            r
            for r in all_records
            if wildcard_link_match(
                r.url, base_url, self.settings.include_link_patterns
            )
        ]

    if self.settings.sitemap.deduplicate:
        seen: set[str] = set()
        deduped: list[URLRecord] = []
        for rec in all_records:
            norm = normalize_url(rec.url)
            if norm not in seen:
                seen.add(norm)
                deduped.append(rec)
        all_records = deduped

    if len(all_records) > self.settings.link_extraction_limit:
        all_records = all_records[: self.settings.link_extraction_limit]

    return [rec.url for rec in all_records]

def create_proxy_pool(self) -> ProxyPool:
    """Creates a ProxyPool based on the provided proxy settings.

    Returns:
        ProxyPool: An initialized proxy pool.
    """
    if self.proxies:
        proxies = self.proxies
    elif self.browser_settings.proxy:
        proxies = [self.browser_settings.proxy]
    else:
        proxies = []

    return ProxyPool(proxies=proxies, strategy=self.proxy_rotation_method)