oca_conductor/data_set/
csv_data_set.rs

1use crate::data_set::DataSet;
2use crate::errors::GenericError;
3use serde::{Serialize, Serializer};
4use serde_json::Value;
5use std::collections::BTreeMap;
6
7#[cfg(feature = "transformer")]
8use crate::transformer::data_set_transformer::Operation;
9#[cfg(feature = "transformer")]
10use oca_rs::state::oca::OCA;
11#[cfg(feature = "transformer")]
12use serde_json::Map;
13
14#[derive(Clone)]
15pub struct CSVDataSet {
16    pub raw: String,
17    delimiter: char,
18}
19
20impl Serialize for CSVDataSet {
21    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
22    where
23        S: Serializer,
24    {
25        serializer.serialize_str(&self.raw)
26    }
27}
28
29impl DataSet for CSVDataSet {
30    fn new(raw: String) -> Box<Self> {
31        Box::new(Self {
32            raw,
33            delimiter: ';',
34        })
35    }
36
37    fn get_raw(&self) -> String {
38        self.raw.clone()
39    }
40
41    fn load(
42        &self,
43        attribute_types: BTreeMap<String, String>,
44    ) -> Result<Vec<Value>, Vec<GenericError>> {
45        let mut errors = vec![];
46        let mut rows_value = vec![];
47        for line in self.raw.lines() {
48            rows_value.push(Value::Array(
49                line.split(self.delimiter)
50                    .map(|el| Value::String(el.to_string()))
51                    .collect(),
52            ))
53        }
54        let mut result = vec![];
55
56        let header_row_v = rows_value.get(0).unwrap().as_array().unwrap();
57        let header_row = header_row_v
58            .iter()
59            .map(|v| v.as_str().unwrap())
60            .collect::<Vec<&str>>();
61
62        for row_value in &rows_value[1..rows_value.len()] {
63            if let Value::Array(row_v) = row_value {
64                let row: Vec<Value> = row_v
65                    .iter()
66                    .enumerate()
67                    .map(|(i, v)| {
68                        if let Value::String(v_str) = v {
69                            if v_str.trim().is_empty() {
70                                return Value::Null;
71                            }
72                        }
73                        let attribute_name = header_row.get(i).unwrap().to_string();
74                        let attribute_type_op = attribute_types.get(&attribute_name);
75                        match attribute_type_op {
76                            Some(attribute_type) => match Self::parse_value(v, attribute_type) {
77                                Ok(parsed) => parsed,
78                                Err(e) => {
79                                    errors
80                                        .push(GenericError::from(format!("{attribute_name}: {e}")));
81                                    Value::Null
82                                }
83                            },
84                            None => v.clone(),
85                        }
86                    })
87                    .collect();
88                result.push(header_row.iter().cloned().zip(row.clone()).collect());
89            }
90        }
91        if !errors.is_empty() {
92            return Err(errors);
93        }
94        Ok(result)
95    }
96
97    #[cfg(feature = "transformer")]
98    fn transform_schema(
99        &self,
100        mappings: BTreeMap<String, String>,
101        subset_attributes_op: Option<Vec<String>>,
102    ) -> Result<Box<dyn DataSet + Sync + Send>, GenericError> {
103        let mut transformed_raw = self.raw.clone();
104        if let Some(header_line) = transformed_raw.lines().take(1).next() {
105            let mut headers = header_line
106                .split(self.delimiter)
107                .map(|header| match mappings.get(header) {
108                    Some(mapping) => mapping,
109                    None => header,
110                })
111                .collect::<Vec<&str>>();
112            let mut records = transformed_raw
113                .lines()
114                .skip(1)
115                .map(|s| s.to_string())
116                .collect::<Vec<String>>();
117
118            if let Some(subset_attributes) = subset_attributes_op {
119                let keep: Vec<bool> = headers
120                    .clone()
121                    .iter()
122                    .map(|attr_name| subset_attributes.contains(&attr_name.to_string()))
123                    .collect();
124                let mut iter = keep.iter();
125                headers.retain(|_| *iter.next().unwrap());
126                records = records
127                    .iter()
128                    .map(|record| {
129                        let mut iter = keep.iter();
130                        let mut r = record.split(self.delimiter).collect::<Vec<&str>>();
131                        r.retain(|_| *iter.next().unwrap());
132                        r.join(&self.delimiter.to_string())
133                    })
134                    .collect::<Vec<String>>();
135            }
136            transformed_raw =
137                headers.join(&self.delimiter.to_string()) + "\n" + &records.join("\n");
138        }
139        Ok(Box::new(
140            Self::new(transformed_raw).delimiter(self.delimiter),
141        ))
142    }
143
144    #[cfg(feature = "transformer")]
145    fn transform_data(
146        &self,
147        oca: &OCA,
148        entry_code_mappings: BTreeMap<String, BTreeMap<String, String>>,
149        unit_transformation_operations: BTreeMap<String, Vec<Operation>>,
150    ) -> Result<Box<dyn DataSet + Sync + Send>, Vec<GenericError>> {
151        let mut transformed_data_set = vec![];
152
153        for record in self.load(oca.capture_base.attributes.clone())? {
154            let record_map = record.as_object().unwrap();
155            let mut transformed_data = Map::new();
156            for (k, v) in record_map {
157                let key = k.to_string();
158                let mut value = v.clone();
159                if let Some(mapped_entries) = entry_code_mappings.get(k) {
160                    match value {
161                        Value::Array(ref values_vec) => {
162                            let mut mapped_values = vec![];
163                            for v in values_vec {
164                                match mapped_entries.get(v.as_str().unwrap()) {
165                                    Some(mapped_entry) => {
166                                        mapped_values.push(Value::String(mapped_entry.to_string()));
167                                    }
168                                    None => {
169                                        mapped_values.push(v.clone());
170                                    }
171                                }
172                            }
173                            value = Value::Array(mapped_values);
174                        }
175                        Value::String(_) => {
176                            if let Some(mapped_entry) = mapped_entries.get(value.as_str().unwrap())
177                            {
178                                value = Value::String(mapped_entry.clone());
179                            };
180                        }
181                        _ => (),
182                    }
183                }
184                if let Some(operations) = unit_transformation_operations.get(k) {
185                    if let Value::Number(num) = &value {
186                        value = Value::Number(
187                            serde_json::value::Number::from_f64(
188                                self.calculate_value_units(num.as_f64().unwrap(), operations),
189                            )
190                            .unwrap(),
191                        );
192                    }
193                }
194                transformed_data.insert(key, value);
195            }
196
197            transformed_data_set.push(Value::Object(transformed_data));
198        }
199
200        let mut data = String::new();
201        for (i, record_val) in transformed_data_set.iter().enumerate() {
202            if let Value::Object(record) = record_val {
203                if i == 0 {
204                    data.push_str(
205                        &record
206                            .keys()
207                            .map(|v| v.to_string())
208                            .collect::<Vec<String>>()
209                            .join(&self.delimiter.to_string()),
210                    );
211                }
212                let line = String::from("\n")
213                    + &record
214                        .values()
215                        .map(|v| {
216                            if v.is_string() {
217                                v.as_str().unwrap().to_string()
218                            } else {
219                                v.to_string()
220                            }
221                        })
222                        .collect::<Vec<String>>()
223                        .join(&self.delimiter.to_string());
224                data.push_str(&line);
225            }
226        }
227
228        Ok(Box::new(Self::new(data).delimiter(self.delimiter)))
229    }
230}
231
232impl CSVDataSet {
233    pub fn delimiter(&mut self, d: char) -> Self {
234        self.delimiter = d;
235        self.clone()
236    }
237
238    fn parse_value(value: &Value, attribute_type: &str) -> Result<Value, GenericError> {
239        if value.is_string() {
240            let value_str = value.as_str().unwrap();
241            let parsed_value = match attribute_type {
242                "Text" => value.clone(),
243                "Array[Text]" => {
244                    let mut parsed = vec![];
245                    for v in serde_json::from_str::<Value>(value_str)
246                        .unwrap_or_else(|_| Value::String(value_str.to_string()))
247                        .as_array()
248                        .ok_or_else(|| {
249                            GenericError::from(format!("\"{value_str}\" value is not an array"))
250                        })?
251                    {
252                        parsed.push(Self::parse_value(v, "Text")?)
253                    }
254                    Value::Array(parsed)
255                }
256                "Numeric" => Value::Number(value_str.parse()?),
257                "Array[Numeric]" => {
258                    let mut parsed = vec![];
259                    for v in serde_json::from_str::<Value>(value_str)
260                        .unwrap_or_else(|_| Value::String(value_str.to_string()))
261                        .as_array()
262                        .ok_or_else(|| {
263                            GenericError::from(format!("\"{value_str}\" value is not an array"))
264                        })?
265                    {
266                        parsed.push(Self::parse_value(v, "Numeric")?)
267                    }
268                    Value::Array(parsed)
269                }
270                "Boolean" => Value::Bool(value_str.parse()?),
271                "Array[Boolean]" => {
272                    let mut parsed = vec![];
273                    for v in serde_json::from_str::<Value>(value_str)
274                        .unwrap_or_else(|_| Value::String(value_str.to_string()))
275                        .as_array()
276                        .ok_or_else(|| {
277                            GenericError::from(format!("\"{value_str}\" value is not an array"))
278                        })?
279                    {
280                        parsed.push(Self::parse_value(v, "Boolean")?)
281                    }
282                    Value::Array(parsed)
283                }
284                "DateTime" => value.clone(),
285                "Array[DateTime]" => {
286                    let mut parsed = vec![];
287                    for v in serde_json::from_str::<Value>(value_str)
288                        .unwrap_or_else(|_| Value::String(value_str.to_string()))
289                        .as_array()
290                        .ok_or_else(|| {
291                            GenericError::from(format!("\"{value_str}\" value is not an array"))
292                        })?
293                    {
294                        parsed.push(Self::parse_value(v, "DateTime")?)
295                    }
296                    Value::Array(parsed)
297                }
298                _ => Value::Null,
299                // TODO add parsing Binary and SAI types
300            };
301
302            return Ok(parsed_value);
303        }
304
305        Ok(value.clone())
306    }
307}
308
309#[cfg(test)]
310mod tests {
311    use super::*;
312    use oca_rs::state::oca::OCA;
313
314    fn setup_oca() -> OCA {
315        let common_assets_dir_path = format!("{}/../assets", env!("CARGO_MANIFEST_DIR"));
316        let oca_result = oca_zip_resolver::resolve_from_zip(
317            format!("{common_assets_dir_path}/oca_bundle.zip").as_str(),
318        );
319        assert!(oca_result.is_ok());
320        oca_result.unwrap()
321    }
322
323    #[test]
324    fn load_csv_data_set() {
325        let oca = setup_oca();
326        let result = CSVDataSet::new(
327            r#"asd
328test@example.com"#
329                .to_string(),
330        )
331        .load(oca.capture_base.attributes);
332        assert!(result.is_ok());
333    }
334
335    #[test]
336    fn load_csv_data_set_with_custom_delimiter() {
337        let oca = setup_oca();
338        let result = CSVDataSet::new(
339            r#"first,second
3401,2"#
341                .to_string(),
342        )
343        .delimiter(',')
344        .load(oca.capture_base.attributes);
345
346        assert!(result.is_ok());
347        if let Value::Object(record) = result.unwrap().first().unwrap() {
348            assert_eq!(record.len(), 2);
349        }
350    }
351
352    #[test]
353    fn parse_vaules_with_invalid_array() {
354        let array_types = vec![
355            "Array[Text]",
356            "Array[Numeric]",
357            "Array[DateTime]",
358            "Array[Boolean]",
359        ];
360        for array_type in array_types {
361            let result = CSVDataSet::parse_value(&Value::String("asd".to_string()), array_type);
362            assert!(result.is_err());
363        }
364    }
365}
oca_conductor/data_set/csv_data_set.rs

oca_conductor/data_set/
csv_data_set.rs
