spider_lib/crawler/

core.rs

//! The core Crawler implementation for the `spider-lib` framework.
//!
//! This module defines the `Crawler` struct, which acts as the central orchestrator
//! for the web scraping process. It ties together the scheduler, downloader,
//! middlewares, spiders, and item pipelines to execute a crawl. The crawler
//! manages the lifecycle of requests and items, handles concurrency, supports
//! checkpointing for fault tolerance, and collects statistics for monitoring.
//!
//! It utilizes a task-based asynchronous model, spawning distinct tasks for
//! handling initial requests, downloading web pages, parsing responses, and
//! processing scraped items.

use crate::downloader::Downloader;
use crate::error::SpiderError;
use crate::item::ScrapedItem;
use crate::middleware::Middleware;
use crate::pipeline::Pipeline;
use crate::request::Request;
use crate::scheduler::Scheduler;
use crate::spider::Spider;
use crate::state::CrawlerState;
use crate::stats::StatCollector;
use anyhow::Result;
use futures_util::future::join_all;
use kanal::{AsyncReceiver, bounded_async};
use tracing::{error, info};

#[cfg(feature = "checkpoint")]
use crate::checkpoint::save_checkpoint;
#[cfg(feature = "checkpoint")]
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;

#[cfg(feature = "middleware-cookies")]
use cookie_store::CookieStore;

/// The central orchestrator for the web scraping process, handling requests, responses, items, concurrency, checkpointing, and statistics collection.
pub struct Crawler<S: Spider, C> {
    scheduler: Arc<Scheduler>,
    req_rx: AsyncReceiver<Request>,
    stats: Arc<StatCollector>,
    downloader: Arc<dyn Downloader<Client = C> + Send + Sync>,
    middlewares: Vec<Box<dyn Middleware<C> + Send + Sync>>,
    spider: Arc<Mutex<S>>,
    item_pipelines: Vec<Box<dyn Pipeline<S::Item>>>,
    max_concurrent_downloads: usize,
    parser_workers: usize,
    max_concurrent_pipelines: usize,
    #[cfg(feature = "checkpoint")]
    checkpoint_path: Option<PathBuf>,
    #[cfg(feature = "checkpoint")]
    checkpoint_interval: Option<Duration>,
    #[cfg(feature = "middleware-cookies")]
    pub cookie_store: Arc<Mutex<CookieStore>>,
}

impl<S, C> Crawler<S, C>
where
    S: Spider + 'static,
    S::Item: ScrapedItem,
    C: Send + Sync + Clone + 'static,
{
    /// Creates a new `Crawler` instance with the given components and configuration.
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn new(
        scheduler: Arc<Scheduler>,
        req_rx: AsyncReceiver<Request>,
        downloader: Arc<dyn Downloader<Client = C> + Send + Sync>,
        middlewares: Vec<Box<dyn Middleware<C> + Send + Sync>>,
        spider: S,
        item_pipelines: Vec<Box<dyn Pipeline<S::Item>>>,
        max_concurrent_downloads: usize,
        parser_workers: usize,
        max_concurrent_pipelines: usize,
        #[cfg(feature = "checkpoint")] checkpoint_path: Option<PathBuf>,
        #[cfg(feature = "checkpoint")] checkpoint_interval: Option<Duration>,
        stats: Arc<StatCollector>,
        #[cfg(feature = "middleware-cookies")] cookie_store: Arc<Mutex<CookieStore>>,
    ) -> Self {
        Crawler {
            scheduler,
            req_rx,
            stats,
            downloader,
            middlewares,
            spider: Arc::new(Mutex::new(spider)),
            item_pipelines,
            max_concurrent_downloads,
            parser_workers,
            max_concurrent_pipelines,
            #[cfg(feature = "checkpoint")]
            checkpoint_path,
            #[cfg(feature = "checkpoint")]
            checkpoint_interval,
            #[cfg(feature = "middleware-cookies")]
            cookie_store,
        }
    }

    /// Starts the crawl, orchestrating the scraping process, managing tasks, handling shutdown, checkpointing, and logging statistics.
    pub async fn start_crawl(self) -> Result<(), SpiderError> {
        info!("Crawler starting crawl");

        let Crawler {
            scheduler,
            req_rx,
            stats,
            downloader,
            middlewares,
            spider,
            item_pipelines,
            max_concurrent_downloads,
            parser_workers,
            max_concurrent_pipelines,
            #[cfg(feature = "checkpoint")]
            checkpoint_path,
            #[cfg(feature = "checkpoint")]
            checkpoint_interval,
            #[cfg(feature = "middleware-cookies")]
            cookie_store,
        } = self;

        let state = CrawlerState::new();
        let pipelines = Arc::new(item_pipelines);
        // Calculate channel capacity based on concurrency settings
        let channel_capacity = std::cmp::max(
            max_concurrent_downloads * 2,
            parser_workers * max_concurrent_pipelines,
        )
        .max(100);

        let (res_tx, res_rx) = bounded_async(channel_capacity);
        let (item_tx, item_rx) = bounded_async(channel_capacity);

        let initial_requests_task =
            spawn_initial_requests_task::<S>(scheduler.clone(), spider.clone(), stats.clone());

        let middlewares_manager = super::SharedMiddlewareManager::new(middlewares);

        let downloader_task = super::spawn_downloader_task::<S, C>(
            scheduler.clone(),
            req_rx,
            downloader,
            middlewares_manager,
            state.clone(),
            res_tx.clone(),
            max_concurrent_downloads,
            stats.clone(),
        );

        let parser_task = super::spawn_parser_task::<S>(
            scheduler.clone(),
            spider.clone(),
            state.clone(),
            res_rx,
            item_tx.clone(),
            parser_workers,
            stats.clone(),
        );

        let item_processor_task = super::spawn_item_processor_task::<S>(
            state.clone(),
            item_rx,
            pipelines.clone(),
            max_concurrent_pipelines,
            stats.clone(),
        );

        #[cfg(feature = "checkpoint")]
        if let (Some(path), Some(interval)) = (&checkpoint_path, checkpoint_interval) {
            let scheduler_clone = scheduler.clone();
            let pipelines_clone = pipelines.clone();
            let path_clone = path.clone();
            #[cfg(feature = "middleware-cookies")]
            let cookie_store_clone = cookie_store.clone();

            tokio::spawn(async move {
                let mut interval_timer = tokio::time::interval(interval);
                interval_timer.tick().await;
                loop {
                    tokio::select! {
                        _ = interval_timer.tick() => {
                            if let Ok(scheduler_checkpoint) = scheduler_clone.snapshot().await {
                                #[cfg(not(feature = "middleware-cookies"))]
                                let save_result = save_checkpoint::<S>(&path_clone, scheduler_checkpoint, &pipelines_clone).await;
                                #[cfg(feature = "middleware-cookies")]
                                let save_result = save_checkpoint::<S>(&path_clone, scheduler_checkpoint, &pipelines_clone, &cookie_store_clone).await;

                                if let Err(e) = save_result {
                                    error!("Periodic checkpoint save failed: {}", e);
                                }
                            }
                        }
                    }
                }
            });
        }

        tokio::select! {
            _ = tokio::signal::ctrl_c() => {
                info!("Ctrl-C received, initiating graceful shutdown.");
            }
            _ = async {
                loop {
                    if scheduler.is_idle() && state.is_idle() {
                        tokio::time::sleep(Duration::from_millis(50)).await;
                        if scheduler.is_idle() && state.is_idle() {
                            break;
                        }
                    }
                    tokio::time::sleep(Duration::from_millis(100)).await;
                }
            } => {
                info!("Crawl has become idle, initiating shutdown.");
            }
        }

        info!("Initiating actor shutdowns.");

        drop(res_tx);
        drop(item_tx);

        scheduler.shutdown().await?;

        item_processor_task
            .await
            .map_err(|e| SpiderError::GeneralError(format!("Item processor task failed: {}", e)))?;

        parser_task
            .await
            .map_err(|e| SpiderError::GeneralError(format!("Parser task failed: {}", e)))?;

        downloader_task
            .await
            .map_err(|e| SpiderError::GeneralError(format!("Downloader task failed: {}", e)))?;

        initial_requests_task.await.map_err(|e| {
            SpiderError::GeneralError(format!("Initial requests task failed: {}", e))
        })?;

        #[cfg(feature = "checkpoint")]
        if let Some(path) = &checkpoint_path {
            let scheduler_checkpoint = scheduler.snapshot().await?;
            #[cfg(not(feature = "middleware-cookies"))]
            let result = save_checkpoint::<S>(path, scheduler_checkpoint, &pipelines).await;
            #[cfg(feature = "middleware-cookies")]
            let result =
                save_checkpoint::<S>(path, scheduler_checkpoint, &pipelines, &cookie_store).await;

            if let Err(e) = result {
                error!("Final checkpoint save failed: {}", e);
            }
        }

        info!("Closing item pipelines...");
        let closing_futures: Vec<_> = pipelines.iter().map(|p| p.close()).collect();
        join_all(closing_futures).await;

        info!("Crawl finished successfully.");
        Ok(())
    }

    /// Returns a cloned Arc to the `StatCollector` instance used by this crawler.
    ///
    /// This allows programmatic access to the collected statistics at any time during or after the crawl.
    pub fn get_stats(&self) -> Arc<StatCollector> {
        Arc::clone(&self.stats)
    }
}

fn spawn_initial_requests_task<S>(
    scheduler: Arc<Scheduler>,
    spider: Arc<Mutex<S>>,
    stats: Arc<StatCollector>,
) -> tokio::task::JoinHandle<()>
where
    S: Spider + 'static,
    S::Item: ScrapedItem,
{
    tokio::spawn(async move {
        match spider.lock().await.start_requests() {
            Ok(requests) => {
                for mut req in requests {
                    req.url.set_fragment(None);
                    match scheduler.enqueue_request(req).await {
                        Ok(_) => {
                            stats.increment_requests_enqueued();
                        }
                        Err(e) => {
                            error!("Failed to enqueue initial request: {}", e);
                        }
                    }
                }
            }
            Err(e) => error!("Failed to create start requests: {}", e),
        }
    })
}