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

use aws_sdk_s3 as s3;
use anyhow::Result;
use surrealdb::engine::remote::http::Client;
use surrealdb::Surreal;
use serde::{Serialize, Deserialize};

pub struct StorageManager {
    s3_client: s3::Client,
    db: Surreal<Client>,
}

#[derive(Serialize, Deserialize)]
struct DataPointer {
    id: String,
    bucket: String,
    key: String,
    mime_type: String,
}

impl StorageManager {
    pub async fn new() -> Result<Self> {
        let config = aws_config::defaults(aws_config::BehaviorVersion::latest())
            .endpoint_url("http://localhost:9000")
            .load().await;
        let s3_client = s3::Client::new(&config);
        
        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?;

        Ok(Self { s3_client, db })
    }

    pub async fn save_object(&self, id: &str, data: Vec<u8>, mime: &str) -> Result<String> {
        let bucket = "bgust-data";
        let key = format!("raw/{}", id);
        
        self.s3_client.put_object()
            .bucket(bucket)
            .key(&key)
            .body(data.into())
            .send().await?;

        let pointer = DataPointer {
            id: id.to_string(),
            bucket: bucket.to_string(),
            key: key.clone(),
            mime_type: mime.to_string(),
        };

        let _: Option<serde_json::Value> = self.db.create(("pointers", id))
            .content(serde_json::to_value(&pointer)?)
            .await?;
        
        Ok(format!("s3://{}/{}", bucket, key))
    }
}