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
109
110
111
112
113
114
115
116
117
118
119
120
121

use anyhow::Result;
use tokio_cron_scheduler::{Job, JobScheduler};
use serde::{Deserialize, Serialize};
use surrealdb::engine::remote::http::Client;
use surrealdb::Surreal;
use crate::run_crawler;

#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct CrawlerJob {
    pub name: String,
    pub cron: String, // 6-field cron, e.g. "0 * * * * *" for every minute
    pub url: String,
    pub deep_scan: bool,
    pub active: bool,
    pub client_id: String,
    pub project_id: String,
    pub domain: String,
}

pub struct SchedulerManager {
    db: Surreal<Client>,
    scheduler: JobScheduler,
}

impl SchedulerManager {
    pub async fn new() -> Result<Self> {
        let db = Surreal::new::<surrealdb::engine::remote::http::Http>("localhost:8000").await?;
        db.signin(surrealdb::opt::auth::Root {
            username: "root".into(),
            password: "root".into(),
        }).await?;
        db.use_ns("bgust").use_db("ecosystem").await?;

        let scheduler = JobScheduler::new().await?;

        Ok(Self { db, scheduler })
    }

    pub async fn seed_default_job_if_needed(&self) -> Result<()> {
        let jobs_val: Vec<serde_json::Value> = self.db.select("crawler_jobs").await?;
        if jobs_val.is_empty() {
            println!("⏰ [SCHEDULER] No jobs found in SurrealDB. Seeding default BCV task...");
            let default_job = CrawlerJob {
                name: "tasas_bcv".to_string(),
                cron: "0 0 9 * * Mon-Fri *".to_string(), // Runs at 09:00 on weekdays (6-field format)
                url: "https://www.bcv.org.ve/".to_string(),
                deep_scan: false,
                active: true,
                client_id: "client_default".to_string(),
                project_id: "project_default".to_string(),
                domain: "finance".to_string(),
            };
            
            // Insert job into SurrealDB using a serialized JSON value
            let _: Option<serde_json::Value> = self.db.create(("crawler_jobs", default_job.name.as_str()))
                .content(serde_json::to_value(&default_job)?)
                .await?;
            println!("⏰ [SCHEDULER] Seeded job: tasas_bcv");
        }
        Ok(())
    }

    pub async fn start(&self) -> Result<()> {
        self.seed_default_job_if_needed().await?;

        // Query active jobs
        let mut query_res = self.db.query("SELECT * FROM crawler_jobs WHERE active = true").await?;
        let active_jobs_val: Vec<serde_json::Value> = query_res.take(0)?;
        
        let mut active_jobs = Vec::new();
        for val in active_jobs_val {
            if let Ok(job) = serde_json::from_value::<CrawlerJob>(val) {
                active_jobs.push(job);
            }
        }

        println!("⏰ [SCHEDULER] Found {} active jobs in SurrealDB.", active_jobs.len());

        for job in active_jobs {
            let name = job.name.clone();
            let url = job.url.clone();
            let deep_scan = job.deep_scan;
            let cron_expr = job.cron.clone();
            let client_id = job.client_id.clone();
            let project_id = job.project_id.clone();
            let domain = job.domain.clone();

            println!("⏰ [SCHEDULER] Scheduling job: {} with cron: {}", name, cron_expr);

            let name_err = name.clone();
            let job_runner = Job::new_async(cron_expr.as_str(), move |_uuid, _l| {
                let url = url.clone();
                let name = name.clone();
                let client_id = client_id.clone();
                let project_id = project_id.clone();
                let domain = domain.clone();
                Box::pin(async move {
                    println!("⏰ [SCHEDULER] Running scheduled job '{}' for url '{}'...", name, url);
                    match run_crawler(&url, deep_scan, Some(&client_id), Some(&project_id), Some(&domain)).await {
                        Ok(ptr) => println!("⏰ [SCHEDULER] Job '{}' completed successfully. Saved to: {}", name, ptr),
                        Err(e) => eprintln!("⚠️ [SCHEDULER] Job '{}' failed: {:?}", name, e),
                    }
                })
            });

            match job_runner {
                Ok(j) => {
                    self.scheduler.add(j).await?;
                }
                Err(e) => {
                    eprintln!("⚠️ [SCHEDULER] Failed to create job structure for '{}': {:?}", name_err, e);
                }
            }
        }

        self.scheduler.start().await?;
        println!("⏰ [SCHEDULER] Job Scheduler Daemon started successfully.");
        
        Ok(())
    }
}