rust_serverless_trading_view_processor

rust_serverless_trading_view_processor.rs

impl Processor {
    pub async fn run(&mut self) -> Result<usize, AppError> {
        let html_data = self.get_data_from_url().await?;
        self.process_data(html_data).await?;
        let count = self.insert().await?;
        Ok(count)
    }

    async fn get_data_from_url(&mut self) -> Result<HtmlData, AppError> {
        // http request to get the page
        let resp = reqwest::get(&self.url).await?;
        assert!(resp.status().is_success());

        // http response body
        let body = resp.text().await?;

        // get the data id from the html body
        // NOTE. This data id is to used when we process the data in the html page
        let data_id = text_between(
            body.clone(),
            "data-props-id=\"".to_string(),
            "\"".to_string(),
        );

        // get the data from the html body
        // NOTE. the data have been loaded and save in the html body for this webpage
        // which we DO NOT need to process the HTML tags
        // save sometimes
        let formatted_string = text_between(
            body,
            "</div><script type=\"application/prs.init-data+json\">".to_string(),
            "</script>".to_string(),
        );

        Ok(HtmlData {
            data_id: data_id,
            data: formatted_string,
        })
    }

    async fn process_data(&mut self, html_data: HtmlData) -> Result<(), AppError> {
        // transform the data from string to serde_json Value
        // This is NOT a json format
        let v: Value = serde_json::from_str(&html_data.data)?;

        // getting the response data in the those Value element convert to str
        let data_str = v[html_data.data_id]["response_json"]["data"].to_string();

        self.data = serde_json::from_str(&data_str)?;

        Ok(())
    }

    async fn insert(&self) -> Result<usize, AppError> {
        // insert to DB
        let mut count = 0;
        if let Some(data) = &self.data {
            count = data.iter().count();
            stream::iter(data)
                .for_each_concurrent(30, |data| async move {
                    let tv = TradingView::from(data);
                    self.db.insert(&tv).await.unwrap();
                })
                .await;
            let msg = format!("total process record {}", count);
            self.alert.producer(msg.as_str()).await?;
        }
        Ok(count)
    }

    pub async fn new() -> Result<Self, AppError> {
        let trading_view_alert_slack = AppAlert::new(Box::new(Slack {
            webhook_url: SLACK_WEBHOOK_URL.to_string(),
            client: reqwest::Client::new(),
        }));
        let new_rds = AwsRds::new().await?;
        let aws_rds = AppDB::new(Box::new(new_rds));
        Ok(Proccessor {
            url: TRADING_VIEW_URL.to_owned(),
            data: None,
            alert: trading_view_alert_slack,
            db: aws_rds,
        })
    }
}

#[cfg(test)]
mod tests {

    use super::*;
    use dotenv::dotenv;

    #[tokio::test]
    async fn test_process_success() {
        dotenv().ok();
        let mut new_proccessor = Proccessor::new().await.unwrap();
        new_proccessor.run().await.unwrap();
    }
}