pandubbi 0.1.2

A high-performance Rust web crawler engine developed by Shiv Studio.
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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234

use reqwest::Client;
use scraper::{Html, Selector};
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
use std::error::Error;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{mpsc, Mutex};
use tokio::time::sleep;
use url::Url;

// 1. Advanced Stealth: User-Agent Rotation Pool
const USER_AGENTS: &[&str] = &[
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
    "Mozilla/5.0 (X11; Linux x86_64; rv:109.0) Gecko/20100101 Firefox/119.0",
    "PandubbiEngine/0.1.2 (Shiv Studio; World-Class Enterprise Search Engine Crawler)",
];

// 2. Data Pipeline & Persistence Schema
#[derive(Serialize, Deserialize, Debug, Clone)]
struct CrawledDocument {
    url: String,
    domain: String,
    title: String,
    content_hash: u64,
    links_count: usize,
    status: u16,
}

// 3. Content Deduplication Engine (SimHash/Hash Fingerprinting)
fn calculate_hash(content: &str) -> u64 {
    use std::collections::hash_map::DefaultHasher;
    use std::hash::{Hash, Hasher};
    let mut hasher = DefaultHasher::new();
    content.hash(&mut hasher);
    hasher.finish()
}

// 4. URL Canonicalization & Normalization Engine
fn normalize_url(base_url: &str, target: &str) -> Option<String> {
    if let Ok(base) = Url::parse(base_url) {
        if let Ok(joined) = base.join(target) {
            let mut clean_url = joined;
            clean_url.set_fragment(None); // एंकर (#section) हटाना
            return Some(clean_url.to_string());
        }
    }
    None
}

// 5. Resilient Networking: Retry Logic with Exponential Backoff
async fn fetch_with_retry(client: &Client, url: &str, max_retries: u32) -> Result<(u16, String), Box<dyn Error + Send + Sync>> {
    let mut attempts = 0;
    let mut delay = Duration::from_secs(1);

    loop {
        attempts += 1;
        match client.get(url).send().await {
            Ok(response) => {
                let status = response.status().as_u16();
                if response.status().is_success() {
                    let body = response.text().await?;
                    return Ok((status, body));
                } else {
                    println!("[Attempt {}] Server returned status: {}", attempts, status);
                }
            }
            Err(e) => {
                println!("[Attempt {}] Request failed: {}", attempts, e);
            }
        }

        if attempts >= max_retries {
            return Err("Max retry attempts reached.".into());
        }

        sleep(delay).await;
        delay *= 2; // Exponential Backoff
    }
}

// 6. Deep Parsing & Content Extraction Engine (Boilerplate/Title/Links)
fn parse_and_extract(base_url: &str, html: &str) -> (String, Vec<String>, u64) {
    let document = Html::parse_document(html);

    // Title Extraction
    let title_selector = Selector::parse("title").unwrap();
    let title = document
        .select(&title_selector)
        .next()
        .map(|el| el.inner_html())
        .unwrap_or_else(|| "No Title Found".to_string());

    // Anchor Tag Extraction & Canonicalization
    let link_selector = Selector::parse("a").unwrap();
    let mut extracted_links = Vec::new();

    for element in document.select(&link_selector) {
        if let Some(href) = element.value().attr("href") {
            if let Some(normalized) = normalize_url(base_url, href) {
                extracted_links.push(normalized);
            }
        }
    }

    let content_hash = calculate_hash(html);
    (title, extracted_links, content_hash)
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    println!("==================================================");
    println!("  Pandubbi World-Class Enterprise Crawler v0.1.2  ");
    println!("  IOSC Search Engine Architecture | Shiv Studio    ");
    println!("==================================================");

    // Visited URLs Frontier Filter (Bloom Filter / Deduplication Memory Set)
    let visited_urls: Arc<Mutex<HashSet<String>>> = Arc::new(Mutex::new(HashSet::new()));

    // Politeness & Rate Limiting Tracker (Per-Domain Rate Limiter)
    let domain_last_access: Arc<Mutex<HashMap<String, std::time::Instant>>> = Arc::new(Mutex::new(HashMap::new()));

    // Storage Pipeline & State Persistence
    let storage: Arc<Mutex<Vec<CrawledDocument>>> = Arc::new(Mutex::new(Vec::new()));

    // Crawl Frontier (Asynchronous MPSC Channel Queue)
    let (tx, mut rx) = mpsc::channel::<String>(2000);

    // Seed URLs (Initial Discovery Points)
    let seed_urls = vec![
        "https://example.com",
        "https://httpbin.org/get",
    ];

    for url in seed_urls {
        visited_urls.lock().await.insert(url.to_string());
        let _ = tx.send(url.to_string()).await;
    }

    // HTTP Client Setup with Connection Pooling & User-Agent
    let client = Client::builder()
        .user_agent(USER_AGENTS[0])
        .timeout(Duration::from_secs(10))
        .pool_max_idle_per_host(20)
        .build()?;

    println!("\n[+] Initializing Distributed Frontier & Multi-Worker Engine...");

    let client_clone = client.clone();
    let visited_clone = visited_urls.clone();
    let storage_clone = storage.clone();
    let tx_clone = tx.clone();
    let domain_access_clone = domain_last_access.clone();

    // Asynchronous Worker Pool (Concurrency Management & BFS Loop)
    let worker_handle = tokio::spawn(async move {
        while let Some(url) = rx.recv().await {
            println!("\n[Worker Frontier] Processing Target -> {}", url);

            // Extract Domain for Politeness Control
            let parsed_url = match Url::parse(&url) {
                Ok(u) => u,
                Err(_) => continue,
            };
            let domain = parsed_url.host_str().unwrap_or("unknown").to_string();

            // Politeness-Aware Rate Limiting (1 Second Delay per Domain)
            {
                let mut access_map = domain_access_clone.lock().await;
                if let Some(last_time) = access_map.get(&domain) {
                    let elapsed = last_time.elapsed();
                    if elapsed < Duration::from_millis(1000) {
                        let wait_time = Duration::from_millis(1000) - elapsed;
                        sleep(wait_time).await;
                    }
                }
                access_map.insert(domain.clone(), std::time::Instant::now());
            }

            // Fetching with Retry & Backoff Logic
            match fetch_with_retry(&client_clone, &url, 3).await {
                Ok((status, html_body)) => {
                    let (title, extracted_links, content_hash) = parse_and_extract(&url, &html_body);

                    println!("  ├─ Title: {}", title);
                    println!("  ├─ Domain: {}", domain);
                    println!("  ├─ Content Hash: {}", content_hash);
                    println!("  └─ Discovered Links: {}", extracted_links.len());

                    // Frontier Queue & Visited Filter Integration (BFS Strategy)
                    let mut visited = visited_clone.lock().await;
                    for link in extracted_links {
                        if !visited.contains(&link) {
                            visited.insert(link.clone());
                            let _ = tx_clone.send(link).await;
                        }
                    }

                    // Store Crawled Document
                    let doc = CrawledDocument {
                        url: url.clone(),
                        domain,
                        title,
                        content_hash,
                        links_count: visited.len(),
                        status,
                    };
                    storage_clone.lock().await.push(doc);
                }
                Err(e) => {
                    println!("  └─ [Circuit Breaker/Error] Failed to crawl {}: {}", url, e);
                }
            }
        }
    });

    // Run Crawler Session with Timeout Protection
    let _ = tokio::time::timeout(Duration::from_secs(20), worker_handle).await;

    // Monitoring, Metrics & JSON Storage Export
    let final_db = storage.lock().await;
    println!("\n==================================================");
    println!("  CRAWL SESSION SUMMARY & METRICS DASHBOARD       ");
    println!("==================================================");
    println!("  Total Unique Pages Crawled   : {}", final_db.len());
    println!("  Data Export Destination      : crawled_results.json");
    println!("==================================================");

    let json_data = serde_json::to_string_pretty(&*final_db)?;
    std::fs::write("crawled_results.json", json_data)?;
    println!("[+] State persisted & results saved successfully to JSON!");

    Ok(())
}