kumo 0.3.15

An async web crawling framework for Rust - Scrapy for Rust
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

use async_trait::async_trait;

use crate::{
    error::KumoError,
    logging::{event, target},
};

use super::Pipeline;

/// Drops items that are missing any of the required fields.
///
/// # Example
/// ```rust,ignore
/// .pipeline(RequireFields::new(&["title", "price"]))
/// ```
pub struct RequireFields {
    fields: Vec<String>,
}

impl RequireFields {
    pub fn new(fields: &[&str]) -> Self {
        Self {
            fields: fields.iter().map(|s| s.to_string()).collect(),
        }
    }
}

#[async_trait]
impl Pipeline for RequireFields {
    async fn process(
        &self,
        item: serde_json::Value,
    ) -> Result<Option<serde_json::Value>, KumoError> {
        for field in &self.fields {
            if item.get(field).is_none() {
                tracing::debug!(
                    target: target::ITEM,
                    event = event::ITEM_DROP,
                    missing_field = %field,
                    reason = "missing_field",
                    "item.drop"
                );
                return Ok(None);
            }
        }
        Ok(Some(item))
    }
}

/// Drops items that do not satisfy a synchronous predicate.
///
/// # Example
/// ```rust,ignore
/// .pipeline(FilterPipeline::new(|item| {
///     item["price"].as_f64().map(|p| p > 0.0).unwrap_or(false)
/// }))
/// ```
pub struct FilterPipeline<F>
where
    F: Fn(&serde_json::Value) -> bool + Send + Sync,
{
    predicate: F,
}

impl<F> FilterPipeline<F>
where
    F: Fn(&serde_json::Value) -> bool + Send + Sync,
{
    pub fn new(predicate: F) -> Self {
        Self { predicate }
    }
}

#[async_trait]
impl<F> Pipeline for FilterPipeline<F>
where
    F: Fn(&serde_json::Value) -> bool + Send + Sync,
{
    async fn process(
        &self,
        item: serde_json::Value,
    ) -> Result<Option<serde_json::Value>, KumoError> {
        if (self.predicate)(&item) {
            Ok(Some(item))
        } else {
            tracing::debug!(
                target: target::ITEM,
                event = event::ITEM_DROP,
                reason = "filter",
                "item.drop"
            );
            Ok(None)
        }
    }
}