API Reference¶

class WebsiteConfig(BaseModel):
    """Parsed website configuration."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    name: str
    start_urls: Annotated[list[HttpUrl], Field(min_length=1)]
    extract_type: Literal["html", "json"]
    extract_config: HtmlExtractorConfig | JsonExtractorConfig

    # Custom parser modules (loaded before extraction)
    parsers: list[str] = Field(default_factory=list)

    # Optional settings
    output_dir: Path = Field(default_factory=lambda: Path("data"))
    headers: dict[str, str] = Field(default_factory=dict)

    # Fetcher settings
    fetch_type: Literal["httpx", "pydoll"] = "httpx"
    browser_settings: dict[str, Any] = Field(default_factory=dict)

    # Item deduplication
    id_field: str | None = None  # Field to use as item ID (e.g., "property_id")

    # Policy (single source of truth for all crawl behavior)
    policy: CrawlPolicy = Field(default_factory=CrawlPolicy)

@dataclass
class CrawlComponents:
    """All components needed for a crawl."""

    store: StateStore
    fetcher: Fetcher
    extractor: HtmlExtractor | JsonExtractor
    policy: CrawlPolicy

class Orchestrator:
    """Coordinates fetching, extraction, and storage.

    The orchestrator:
    1. Pulls item URLs from the store queue
    2. Fetches content concurrently
    3. Extracts items and links
    4. Saves items to the store (with dedup by item_id or URL)
    5. Adds pagination links to in-memory set
    6. Adds new item links to queue (if not already scraped)
    7. Handles retry logic based on policy
    8. Checks stopping conditions

    Example:
    ```python
    store = StateStore("data/mysite/state.db", id_field="property_id")
    store.add_item_url("https://example.com/listing")

    orchestrator = Orchestrator(
        store=store,
        fetcher=HttpxFetcher(),
        extractor=HtmlExtractor(config),
        policy=CrawlPolicy(concurrency=10),
    )

    result = await orchestrator.run()
    print(f"Extracted {result.stats.items_extracted} items")
    store.export_jsonl("output.jsonl")
    ```
    """

    def __init__(
        self,
        store: StateStore,
        fetcher: Fetcher,
        extractor: Extractor,
        policy: CrawlPolicy | None = None,
        middleware: list[Middleware] | None = None,
        on_item: OnItemCallback | None = None,
        on_url_complete: OnUrlCallback | None = None,
        on_error: OnErrorCallback | None = None,
        full_crawl: bool = False,
    ):
        """Initialize the orchestrator.

        Args:
            store: Unified state store for URLs and items
            fetcher: Fetcher for retrieving content
            extractor: Extractor for parsing content
            policy: Crawl policy (defaults to CrawlPolicy())
            middleware: List of middleware to apply (in order)
            on_item: Callback for each extracted item (item, is_new)
            on_url_complete: Callback when a URL is processed
            on_error: Callback for errors (url, message)
            full_crawl: Disable caught-up stopping, crawl all pagination
        """
        self.store = store
        self.fetcher = fetcher
        self.extractor = extractor
        self.policy = policy or CrawlPolicy()
        self.middleware = MiddlewareChain(middleware)

        self.on_item = on_item
        self.on_url_complete = on_url_complete
        self.on_error = on_error

        self.stats = CrawlStats()
        self._semaphore: asyncio.Semaphore | None = None
        self._full_crawl = full_crawl
        self._is_incremental = False  # Set in run() based on existing items
        self._is_first_batch = True  # Track first batch to skip initial delay

    async def run(self) -> CrawlResult:
        """Run the crawl with concurrent fetching.

        Processes URLs from the store until:
        - Queue is empty
        - Stopping condition is met
        - Error threshold exceeded

        Returns:
            CrawlResult with stats and store reference
        """
        # Check if this is an incremental crawl (re-run with existing data)
        # Only enable "caught up" stopping for incremental crawls
        # full_crawl mode disables caught-up stopping for complete traversal
        self._is_incremental = self.store.item_count() > 0 and not self._full_crawl

        # Reset any crashed in-progress URLs
        reset_count = self.store.reset_in_progress()
        if reset_count > 0:
            logger.info(f"Reset {reset_count} in-progress URLs from previous run")

        # Retry failed item URLs from previous runs
        reset_failed, perm_failed = self.store.reset_failed_items()
        if reset_failed > 0:
            logger.info(f"Retrying {reset_failed} previously failed item URLs")
        if perm_failed > 0:
            logger.info(f"Marked {perm_failed} item URLs as permanently failed")

        # Clear pagination for fresh run
        self.store.reset_pagination()

        logger.info(
            f"Starting crawl with {self.store.url_pending_count()} URLs "
            f"(concurrency: {self.policy.concurrency})"
        )
        stopped_reason = None
        self._semaphore = asyncio.Semaphore(self.policy.concurrency)
        self._is_first_batch = True

        try:
            while True:
                # Check stopping conditions
                should_stop, reason = self.policy.should_stop(
                    requests_completed=self.stats.urls_processed,
                    items_extracted=self.stats.items_extracted,
                    consecutive_failures=self.stats.consecutive_failures,
                    total_failures=self.stats.urls_failed,
                    consecutive_caught_up=self.stats.consecutive_caught_up,
                )
                if should_stop:
                    logger.info(f"Stopping: {reason}")
                    stopped_reason = reason
                    break

                # Get batch of URLs to process
                tasks = self._get_batch()
                if not tasks:
                    logger.info("Queue empty")
                    break

                # Process batch concurrently
                await self._process_batch(tasks)

                # Apply delay between batches (skip first batch)
                if not self._is_first_batch and self.policy.delay > 0:
                    await asyncio.sleep(self.policy.delay)

                self._is_first_batch = False

        except KeyboardInterrupt:
            logger.info("Interrupted by user")
            stopped_reason = "User interrupted"

        return CrawlResult(
            stats=self.stats,
            store=self.store,
            stopped_reason=stopped_reason,
        )

    def _get_batch(self) -> list[UrlTask]:
        """Get a batch of URLs to process concurrently."""
        batch = []
        remaining = self.policy.concurrency

        # Check if we're near the max_requests limit
        if self.policy.max_requests:
            remaining = min(remaining, self.policy.max_requests - self.stats.urls_processed)

        for _ in range(remaining):
            task = self.store.get_next_url()
            if task is None:
                break
            batch.append(task)

        return batch

    async def _process_batch(self, tasks: list[UrlTask]) -> None:
        """Process a batch of URLs concurrently."""
        coroutines = [self._process_url_with_semaphore(task) for task in tasks]
        await asyncio.gather(*coroutines, return_exceptions=True)

    async def _process_url_with_semaphore(self, task: UrlTask) -> None:
        """Process URL with semaphore for rate limiting."""
        async with self._semaphore:
            try:
                await self._process_url(task)
            except Exception as e:
                logger.error(f"Unhandled error processing {task.url}: {e}")
                error = CrawlError.extraction_error(f"Unhandled error: {e}")
                self._handle_failure(task.url, error, task.attempts or 0)

    async def _process_url(self, task: UrlTask) -> None:
        """Process a single URL with middleware hooks."""
        url = task.url
        logger.debug(f"Processing: {url} (attempt {task.attempts + 1})")

        # Create middleware context
        ctx = MiddlewareContext(url=url, url_type=task.url_type)

        # Pre-fetch middleware
        ctx = await self.middleware.run_pre_fetch(ctx)
        if ctx.skip:
            logger.debug(f"Skipped by middleware: {url}")
            self.store.mark_url_done(url)
            return

        # Fetch (with headers from middleware)
        fetch_result = await self.fetcher.fetch(ctx.url, headers=ctx.headers or None)

        if not fetch_result.success:
            ctx.error = fetch_result.error
            ctx = await self.middleware.run_on_error(ctx)
            if ctx.error:  # Error not cleared by middleware
                self._handle_failure(url, ctx.error, task.attempts)
            return

        # Post-fetch middleware
        ctx.content = fetch_result.content
        ctx = await self.middleware.run_post_fetch(ctx)

        # Check if middleware triggered an error (e.g., CAPTCHA detection)
        if ctx.error:
            ctx = await self.middleware.run_on_error(ctx)
            if ctx.error:
                self._handle_failure(url, ctx.error, task.attempts)
            return

        # Determine if we should extract items from this URL type
        skip_items = self.policy.items_from != "all" and task.url_type != self.policy.items_from

        # Extract
        extract_result = self.extractor.extract(ctx.content, skip_items=skip_items)

        if not extract_result.success:
            ctx.error = extract_result.error
            ctx = await self.middleware.run_on_error(ctx)
            if ctx.error:
                self._handle_failure(url, ctx.error, task.attempts)
            return

        # Post-extract middleware
        ctx.extract_result = extract_result
        ctx = await self.middleware.run_post_extract(ctx)
        extract_result = ctx.extract_result

        # Save items
        new_items = 0
        for item in extract_result.items:
            is_new, item_key = self.store.save_item(item, url)
            if is_new:
                new_items += 1
            if self.on_item:
                self.on_item(item, is_new)

        # Check item links against store before adding to queue
        new_item_links = []
        for link in extract_result.item_links:
            if not self.store.has_item_for_url(link):
                new_item_links.append(link)

        added_items = self.store.add_item_urls(new_item_links)

        # Determine if this pagination page is "caught up" (has item links but all already scraped)
        # Only for incremental crawls - fresh crawls should follow all pagination
        # (related items discovered on detail pages can cause false "caught up" on fresh crawls)
        is_pagination_page = task.url_type == "pagination"
        has_item_links = len(extract_result.item_links) > 0
        all_already_scraped = len(new_item_links) == 0
        caught_up = (
            self._is_incremental and is_pagination_page and has_item_links and all_already_scraped
        )

        # Only follow pagination if this page found new items (per-branch stopping)
        # This naturally handles multi-seed crawls: each branch stops independently
        if caught_up:
            added_pagination = 0
        else:
            added_pagination = self.store.add_pagination_urls(extract_result.pagination_links)

        self.stats.record_queued(added_pagination + added_items)

        # Log meaningful events only (pagination pages)
        if caught_up:
            logger.info(f"Caught up | {url}")
        elif is_pagination_page:
            logger.info(f"Found {added_items} new items | {url}")

        # Mark URL done
        self.store.mark_url_done(url)
        self.stats.record_success(new_items, caught_up=caught_up)

        if self.on_url_complete:
            self.on_url_complete(url)

        logger.debug(
            f"Extracted {len(extract_result.items)} items ({new_items} new), "
            f"found {len(extract_result.pagination_links)} pagination + "
            f"{len(extract_result.item_links)} item links from {url}"
        )

    def _handle_failure(self, url: str, error, attempts: int) -> None:
        """Handle a failed URL."""
        # Ensure we have a CrawlError
        if not isinstance(error, CrawlError):
            error = CrawlError.extraction_error(str(error))

        logger.warning(f"Failed: {url} - {error.message}")

        if self.on_error:
            self.on_error(url, error.message)

        # Check if we should retry
        if self.policy.should_retry(error, attempts + 1):
            delay = self.policy.get_retry_delay(attempts + 1)
            self.store.schedule_url_retry(url, delay, error.message)
            self.stats.record_failure(will_retry=True)
            logger.debug(f"Will retry {url} in {delay:.1f}s")
        else:
            self.store.mark_url_failed(url, error.message)
            self.stats.record_failure(will_retry=False)

class CrawlPolicy(StrictModel):
    """All behavioral rules in one place.

    This centralizes decisions about:
    - When to retry a failed request
    - When to stop crawling
    - When to checkpoint
    """

    # Retry behavior
    max_retries: NonNegativeInt = 3
    """Maximum retry attempts per URL."""

    retry_delay: PositiveFloat = 1.0
    """Initial delay between retries (seconds)."""

    backoff_factor: Annotated[float, Field(ge=1.0)] = 2.0
    """Multiply delay by this factor after each retry."""

    max_retry_delay: PositiveFloat = 60.0
    """Maximum delay between retries (seconds)."""

    retryable_categories: set[ErrorCategory] = Field(
        default_factory=lambda: {
            ErrorCategory.NETWORK,
            ErrorCategory.SERVER,
            ErrorCategory.RATE_LIMITED,
        }
    )
    """Error categories that should be retried."""

    # Stopping rules
    max_requests: PositiveInt | None = None
    """Maximum requests to process (None = unlimited). Retries don't count."""

    max_consecutive_failures: PositiveInt = 10
    """Stop after this many consecutive failures."""

    max_error_rate: UnitInterval = 0.5
    """Stop if error rate exceeds this (0.0 to 1.0)."""

    min_requests_for_error_rate: PositiveInt = 20
    """Minimum requests before error rate check applies."""

    stop_on_empty: bool = True
    """Stop pagination when a page yields no items or links."""

    stop_on_caught_up: bool = False
    """Stop globally when reaching already-scraped items.

    Note: Per-branch stopping is now handled automatically in the orchestrator.
    Each pagination branch stops independently when it encounters a page where
    all items are already scraped. This global setting is disabled by default
    but can be enabled as an additional stop condition."""

    caught_up_threshold: PositiveInt = 3
    """Global stop threshold: stop after this many consecutive caught-up pages.

    Only applies when stop_on_caught_up=True. For multi-seed crawls, prefer
    the default per-branch stopping which handles each seed independently."""

    # Concurrency
    concurrency: PositiveInt = 5
    """Number of concurrent requests."""

    # Request pacing
    delay: NonNegativeFloat = 0.0
    """Delay after each batch (seconds)."""

    jitter: NonNegativeFloat = 0.1
    """Small random delay (0 to jitter seconds) before each request."""

    # Item filtering
    items_from: Literal["item", "pagination", "all"] = "item"
    """Which URL types to save items from: "item", "pagination", or "all"."""

    # Checkpointing
    checkpoint_every: PositiveInt = 100
    """Checkpoint after this many items extracted."""

    def should_retry(self, error: CrawlError, attempts: int) -> bool:
        """Determine if a failed request should be retried.

        Args:
            error: The error that occurred
            attempts: Number of attempts already made (including the failed one)

        Returns:
            True if the request should be retried
        """
        # Never retry non-retryable errors
        if not error.retryable:
            return False

        # Check if error category is retryable
        if error.category not in self.retryable_categories:
            return False

        # Check max retries
        if attempts >= self.max_retries:
            return False

        return True

    def get_retry_delay(self, attempts: int) -> float:
        """Calculate delay before next retry.

        Uses exponential backoff: delay * (backoff_factor ^ attempts)

        Args:
            attempts: Number of attempts already made

        Returns:
            Delay in seconds before next attempt
        """
        delay = self.retry_delay * (self.backoff_factor ** (attempts - 1))
        return min(delay, self.max_retry_delay)

    def should_stop(
        self,
        requests_completed: int,
        items_extracted: int,
        consecutive_failures: int,
        total_failures: int,
        consecutive_caught_up: int = 0,
    ) -> tuple[bool, str]:
        """Determine if crawling should stop.

        Args:
            requests_completed: Successful requests so far (retries don't count)
            items_extracted: Total items extracted so far (for logging)
            consecutive_failures: Current streak of consecutive failures
            total_failures: Total failures so far
            consecutive_caught_up: Pages where all items were already scraped

        Returns:
            Tuple of (should_stop, reason)
        """
        # Check max requests
        if self.max_requests and requests_completed >= self.max_requests:
            return True, f"Reached max requests limit ({self.max_requests})"

        # Check consecutive failures
        if consecutive_failures >= self.max_consecutive_failures:
            return True, f"Too many consecutive failures ({consecutive_failures})"

        # Check error rate (only after minimum requests)
        if requests_completed >= self.min_requests_for_error_rate:
            error_rate = total_failures / requests_completed
            if error_rate > self.max_error_rate:
                return True, f"Error rate too high ({error_rate:.1%} > {self.max_error_rate:.1%})"

        # Check caught-up (incremental crawling)
        if self.stop_on_caught_up and consecutive_caught_up >= self.caught_up_threshold:
            return True, "Caught up with existing data"

        return False, ""

    def should_checkpoint(self, items_since_last: int) -> bool:
        """Determine if a checkpoint should be saved.

        Args:
            items_since_last: Items extracted since last checkpoint

        Returns:
            True if a checkpoint should be saved
        """
        return items_since_last >= self.checkpoint_every

@dataclass
class CrawlResult:
    """Result of a crawl run."""

    stats: CrawlStats
    store: StateStore
    stopped_reason: str | None = None

@dataclass
class CrawlStats:
    """Runtime statistics for a crawl.

    Tracks counts needed for policy decisions (stopping rules)
    and progress reporting.
    """

    # URL counts
    urls_queued: int = 0
    urls_processed: int = 0
    urls_succeeded: int = 0
    urls_failed: int = 0
    urls_retried: int = 0

    # Item counts
    items_extracted: int = 0
    items_since_checkpoint: int = 0

    # Failure tracking
    consecutive_failures: int = 0

    # Incremental crawling tracking
    consecutive_caught_up: int = 0
    """Consecutive listing pages where all items were already scraped."""

    # Timing
    started_at: datetime = field(default_factory=datetime.now)
    last_success_at: datetime | None = None
    last_checkpoint_at: datetime | None = None

    def record_success(self, items_count: int = 0, caught_up: bool = False) -> None:
        """Record a successful URL processing.

        Args:
            items_count: Number of items extracted from this URL
            caught_up: True if this page had items but all were already scraped
        """
        self.urls_processed += 1
        self.urls_succeeded += 1
        self.consecutive_failures = 0
        self.items_extracted += items_count
        self.items_since_checkpoint += items_count
        self.last_success_at = datetime.now()

        # Track caught-up status for incremental crawling
        if caught_up:
            self.consecutive_caught_up += 1
        else:
            self.consecutive_caught_up = 0

    def record_failure(self, will_retry: bool = False) -> None:
        """Record a failed URL processing.

        Args:
            will_retry: If True, URL will be retried (not a terminal failure)
        """
        self.urls_processed += 1
        self.consecutive_failures += 1
        if will_retry:
            self.urls_retried += 1
        else:
            self.urls_failed += 1  # Only count terminal failures

    def record_queued(self, count: int = 1) -> None:
        """Record URLs added to queue."""
        self.urls_queued += count

    def record_checkpoint(self) -> None:
        """Record that a checkpoint was saved."""
        self.items_since_checkpoint = 0
        self.last_checkpoint_at = datetime.now()

    @property
    def error_rate(self) -> float:
        """Calculate current error rate."""
        if self.urls_processed == 0:
            return 0.0
        return self.urls_failed / self.urls_processed

    @property
    def success_rate(self) -> float:
        """Calculate current success rate."""
        return 1.0 - self.error_rate

    @property
    def elapsed_seconds(self) -> float:
        """Seconds since crawl started."""
        return (datetime.now() - self.started_at).total_seconds()

    @property
    def urls_per_second(self) -> float:
        """Average URLs processed per second."""
        elapsed = self.elapsed_seconds
        if elapsed == 0:
            return 0.0
        return self.urls_processed / elapsed

    @property
    def items_per_second(self) -> float:
        """Average items extracted per second."""
        elapsed = self.elapsed_seconds
        if elapsed == 0:
            return 0.0
        return self.items_extracted / elapsed

    def summary(self) -> dict:
        """Return a summary dict for logging/reporting."""
        return {
            "urls_processed": self.urls_processed,
            "urls_succeeded": self.urls_succeeded,
            "urls_failed": self.urls_failed,
            "items_extracted": self.items_extracted,
            "error_rate": f"{self.error_rate:.1%}",
            "elapsed": f"{self.elapsed_seconds:.1f}s",
            "urls_per_sec": f"{self.urls_per_second:.1f}",
        }