bgustcrawler 0.2.0

Distributed web crawler and scheduler integrated with Kafka, SurrealDB, and S3 storage
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108

mod consumer;
mod storage;
mod bloom;
mod producer;
pub mod search;
pub mod legal;
pub mod scheduler;

use anyhow::Result;
use bgustscraper::worker::{PlaywrightWorker, ScrapingEngine, ScrapeOptions};
use crate::storage::StorageManager;
use crate::bloom::ThreadSafeBloomFilter;
use crate::producer::Producer;
use uuid::Uuid;
use sha2::{Sha256, Digest};
use std::sync::OnceLock;
use serde_json::json;

fn bloom_filter() -> &'static ThreadSafeBloomFilter {
    static BLOOM: OnceLock<ThreadSafeBloomFilter> = OnceLock::new();
    BLOOM.get_or_init(|| ThreadSafeBloomFilter::new(100_000, 5))
}

pub async fn run_crawler(
    url: &str,
    deep_scan: bool,
    client_id: Option<&str>,
    project_id: Option<&str>,
    domain: Option<&str>,
) -> Result<String> {
    // 1. Bloom Filter Deduplication
    let bf = bloom_filter();
    if bf.contains(&url.to_string()) {
        println!("🚫 [CRAWLER] URL already visited (deduplicated by Bloom Filter): {}", url);
        return Ok("DUPLICATE_SKIPPED".to_string());
    }
    bf.insert(&url.to_string());

    println!("🚀 [CRAWLER] Crawling URL: {}", url);
    let mut worker = PlaywrightWorker::new(true, ScrapingEngine::Playwright)?;
    let storage = StorageManager::new().await?;

    let response = worker.scrape(url, Some(ScrapeOptions {
        deep_scan,
        semantic_embeddings: None,
        text: None,
        prompt: None,
    }))?;
    
    if response.status == "success" {
        let markdown = response.markdown.unwrap_or_else(|| "No markdown available".to_string());
        let id = Uuid::new_v4().to_string();
        
        // 2. Data Lineage: Compute SHA-256 hash of original text
        let mut hasher = Sha256::new();
        hasher.update(markdown.as_bytes());
        let sha256_bytes = hasher.finalize();
        let sha256_hash = sha256_bytes.iter().map(|b| format!("{:02x}", b)).collect::<String>();

        // 3. Save raw file to Object Storage (MinIO/S3) and SurrealDB metadata
        let pointer = storage.save_object(&id, markdown.as_bytes().to_vec(), "text/markdown").await?;
        println!("📥 [CRAWLER] Saved document to object store at {}", pointer);

        // 4. Publish message to Redpanda/Kafka under "crawl_results"
        match Producer::new("localhost:9094") {
            Ok(producer) => {
                let payload = json!({
                    "id": id,
                    "client_id": client_id.unwrap_or("client_default"),
                    "project_id": project_id.unwrap_or("project_default"),
                    "domain": domain.unwrap_or("general"),
                    "url": url,
                    "pointer": pointer,
                    "sha256": sha256_hash,
                    "content": markdown
                });
                
                match producer.send_message("crawl_results", &id, &payload.to_string()).await {
                    Ok(_) => println!("📢 [CRAWLER] Published crawl results to Redpanda for URL: {}", url),
                    Err(e) => eprintln!("⚠️ [CRAWLER] Failed to publish message to Redpanda: {:?}", e),
                }
            }
            Err(e) => {
                eprintln!("⚠️ [CRAWLER] Failed to initialize Kafka producer: {:?}", e);
            }
        }

        Ok(pointer)
    } else {
        Err(anyhow::anyhow!("Crawler failed: {}", response.message.unwrap_or_default()))
    }
}

pub async fn start_background_consumer(brokers: &str) -> Result<()> {
    let consumer = consumer::Consumer::new(brokers, "crawler_group", &["crawl_tasks"]);
    consumer.listen(|payload| async move {
        println!("Received background task: {}", payload);
        if let Ok(req) = serde_json::from_str::<serde_json::Value>(&payload) {
            if let Some(url) = req["url"].as_str() {
                let client_id = req["client_id"].as_str();
                let project_id = req["project_id"].as_str();
                let domain = req["domain"].as_str();
                let deep_scan = req["deep_scan"].as_bool().unwrap_or(false);
                let _ = run_crawler(url, deep_scan, client_id, project_id, domain).await;
            }
        }
    }).await
}