uiuifree_text_data/
csv_tokenizer.rs

1use std::error::Error;
2use std::sync::Mutex;
3
4use rayon::prelude::*;
5use uiuifree_normalize::{free_text, remove_html};
6
7pub struct CsvTokenizer {
8    path: String,
9    // callable_test: fn(String) -> String,
10}
11
12impl CsvTokenizer {
13    pub fn new(path: &str) -> CsvTokenizer {
14        CsvTokenizer {
15            path: path.to_string()
16            // callable_test: |x| {
17            //     x
18            // },
19        }
20    }
21    pub fn trim_csv(input: &str, output: &str) {
22        // 読み込むファイルパスの設定
23        match CsvTokenizer::new(input).execute(output) {
24            Ok(_) => {}
25            Err(e) => {
26                println!("{}", e)
27            }
28        };
29    }
30
31    // fn set_fn(&mut self, callable_test: fn(String) -> String) {
32    //     self.callable_test = callable_test
33    // }
34
35    pub fn headers(&self) -> Result<Vec<String>, Box<dyn Error>> {
36        let path = String::from(self.path.as_str());
37        let mut rdr = csv::Reader::from_path(path)?;
38
39        let mut res = Vec::new();
40        for i in rdr.headers()?.iter() {
41            res.push(i.to_string());
42        }
43        Ok(res)
44    }
45    pub fn records(&self) -> Result<Vec<Vec<String>>, Box<dyn Error>> {
46        let path = String::from(self.path.as_str());
47        let mut rdr = csv::Reader::from_path(path)?;
48
49        let mut rows: Vec<Vec<String>> = Vec::new();
50        for result in rdr.records() {
51            let record = result?;
52            let mut row: Vec<String> = Vec::new();
53            for s in record.iter() {
54                row.push(s.to_string());
55            }
56            rows.push(row);
57        }
58        Ok(rows)
59    }
60
61    pub async fn process<F, T, U>(&mut self, callable: F) -> Result<Vec<U>, Box<dyn Error>>
62        where
63            F: Fn(Vec<String>) -> T,
64            T: std::future::Future<Output=U> + std::marker::Send + 'static,
65            U: std::marker::Send + 'static
66    {
67        let path = String::from(self.path.as_str());
68        let mut rdr = csv::ReaderBuilder::new().flexible(true).from_path(path)?;
69
70        let mut handles = vec![];
71        let mut response = vec![];
72
73        for result in rdr.records() {
74            let record = result?;
75            let mut row: Vec<String> = Vec::new();
76            for s in record.iter() {
77                row.push(free_text(s.to_string()));
78                // row.push(remove_html(free_text(s.to_string())))
79                // row.push(remove_html(free_text(s.to_string())))
80            }
81            handles.push(tokio::spawn(callable(row)));
82        }
83
84        for handle in handles {
85            response.push(handle.await.unwrap())
86        }
87
88        return Ok(response);
89    }
90    pub async fn process2<'a ,F, T, U>(&'a mut self, callable: F) -> Result<Vec<U>, Box<dyn Error>>
91        where
92            F: Fn(Vec<String>) -> T  + std::marker::Sync + std::marker::Send +'a+ 'static + Copy,
93            T: std::future::Future<Output=U> + std::marker::Send + 'static,
94            U: std::marker::Send + 'static
95    {
96        let path = String::from(self.path.as_str());
97        let mut rdr = csv::ReaderBuilder::new().flexible(true).from_path(path)?;
98        let mut response = vec![];
99        let mut tmp_records = vec![];
100        for result in rdr.records() {
101            let record = result?;
102            let mut row: Vec<String> = Vec::new();
103            for s in record.iter() {
104                row.push(free_text(s.to_string()));
105                // row.push(remove_html(free_text(s.to_string())))
106                // row.push(remove_html(free_text(s.to_string())))
107
108            }
109            tmp_records.push(row)
110        }
111        let (tx,mut rx) = tokio::sync::mpsc::channel(tmp_records.len());
112        for row in tmp_records {
113            let tx = tx.clone();
114            tokio::spawn(
115                async move {
116                    tx.send(callable(row).await).await.unwrap();
117                }
118            );
119        }
120        drop(tx);
121        while let Some(handle) = rx.recv().await {
122            response.push(handle)
123        }
124        return Ok(response);
125    }
126    pub fn execute(&self, out_path: &str) -> Result<(), Box<dyn Error>> {
127        let path = String::from(self.path.as_str());
128        // Build the CSV reader and iterate over each record.
129        let mut rdr = csv::Reader::from_path(path)?;
130        let mut w = csv::Writer::from_path(out_path)?;
131        let headers = rdr.headers()?;
132
133        w.write_record(headers)?;
134        let mut rows: Vec<Vec<String>> = Vec::new();
135
136        for record in rdr.records() {
137            if record.is_err() {
138                break;
139            }
140            let record = record.unwrap();
141            let mut row: Vec<String> = Vec::new();
142            for s in record.iter() {
143                row.push(s.to_string());
144            }
145            rows.push(row);
146        }
147
148        let vec: Mutex<Vec<Vec<String>>> = Mutex::new(Vec::new());
149        rows.par_iter_mut().for_each(|row| {
150            let mut new_rows = Vec::new();
151            for v in row {
152                new_rows.push(remove_html(free_text(v.to_string())))
153                // HW 落ちるので書き換え
154                // new_rows.push(remove_html(space(v.to_string())))
155            }
156            vec.lock().unwrap().push(new_rows);
157        });
158        let rows = vec.lock().unwrap().to_vec();
159        for row in rows {
160            w.write_record(row)?;
161        }
162        Ok(())
163    }
164}
165
166
167
uiuifree_text_data/csv_tokenizer.rs

uiuifree_text_data/
csv_tokenizer.rs
