Merge branch 'async_media_upload' into 'master'

Async media upload

See merge request veretcle/scootaloo!24

Cargo.lock

@@ -2016,7 +2016,7 @@ dependencies = [
 
 [[package]]
 name = "scootaloo"
-version = "0.7.0"
+version = "0.8.0"
 dependencies = [
  "chrono",
  "clap",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "scootaloo"
-version = "0.7.0"
+version = "0.8.0"
 authors = ["VC <veretcle+framagit@mateu.be>"]
 edition = "2021"
 

src/lib.rs

@@ -13,152 +13,149 @@ mod twitter;
 use twitter::*;
 
 mod util;
+use crate::util::generate_media_ids;
 
 mod state;
 pub use state::{init_db, migrate_db};
 use state::{read_state, write_state, TweetToToot};
 
 use elefren::{prelude::*, status_builder::StatusBuilder};
-use log::{debug, error, info, warn};
+use log::info;
 use rusqlite::Connection;
-use std::borrow::Cow;
-use tokio::fs::remove_file;
+use std::sync::Arc;
+use tokio::{spawn, sync::Mutex};
 
 /// This is where the magic happens
 #[tokio::main]
 pub async fn run(config: Config) {
-    // get OAuth2 token
-    let token = get_oauth2_token(&config.twitter);
+    // create the task vector for handling multiple accounts
+    let mut mtask = vec![];
 
-    for mastodon_config in config.mastodon.values() {
-        // open the SQLite connection
-        let conn = Connection::open(&config.scootaloo.db_path).unwrap_or_else(|e| {
+    // open the SQLite connection
+    let conn = Arc::new(Mutex::new(
+        Connection::open(&config.scootaloo.db_path).unwrap_or_else(|e| {
             panic!(
                 "Something went wrong when opening the DB {}: {}",
                 &config.scootaloo.db_path, e
             )
+        }),
+    ));
+
+    for mastodon_config in config.mastodon.into_values() {
+        // create temporary value for each task
+        let scootaloo_cache_path = config.scootaloo.cache_path.clone();
+        let token = get_oauth2_token(&config.twitter);
+        let task_conn = conn.clone();
+
+        let task = spawn(async move {
+            info!("Starting treating {}", &mastodon_config.twitter_screen_name);
+
+            // retrieve the last tweet ID for the username
+            let lconn = task_conn.lock().await;
+            let last_tweet_id = read_state(&lconn, &mastodon_config.twitter_screen_name, None)
+                .unwrap_or_else(|e| panic!("Cannot retrieve last_tweet_id: {}", e))
+                .map(|s| s.tweet_id);
+            drop(lconn);
+
+            // get Mastodon instance
+            let mastodon = get_mastodon_token(&mastodon_config);
+
+            // get user timeline feed (Vec<tweet>)
+            let mut feed =
+                get_user_timeline(&mastodon_config.twitter_screen_name, &token, last_tweet_id)
+                    .await
+                    .unwrap_or_else(|e| {
+                        panic!(
+                            "Something went wrong when trying to retrieve {}’s timeline: {}",
+                            &mastodon_config.twitter_screen_name, e
+                        )
+                    });
+
+            // empty feed -> exiting
+            if feed.is_empty() {
+                info!("Nothing to retrieve since last time, exiting…");
+                return;
+            }
+
+            // order needs to be chronological
+            feed.reverse();
+
+            for tweet in &feed {
+                info!("Treating Tweet {} inside feed", tweet.id);
+                // initiate the toot_reply_id var
+                let mut toot_reply_id: Option<String> = None;
+                // determine if the tweet is part of a thread (response to self) or a standard response
+                if let Some(r) = &tweet.in_reply_to_screen_name {
+                    if r.to_lowercase() != mastodon_config.twitter_screen_name.to_lowercase() {
+                        // we are responding not threading
+                        info!("Tweet is a direct response, skipping");
+                        continue;
+                    }
+                    info!("Tweet is a thread");
+                    // get the corresponding toot id
+                    let lconn = task_conn.lock().await;
+                    toot_reply_id = read_state(
+                        &lconn,
+                        &mastodon_config.twitter_screen_name,
+                        tweet.in_reply_to_status_id,
+                    )
+                    .unwrap_or(None)
+                    .map(|s| s.toot_id);
+                    drop(lconn);
+                };
+
+                // build basic status by just yielding text and dereferencing contained urls
+                let mut status_text = build_basic_status(tweet);
+
+                // building associative media list
+                let (media_url, status_medias) =
+                    generate_media_ids(tweet, &scootaloo_cache_path, &mastodon).await;
+
+                status_text = status_text.replace(&media_url, "");
+
+                info!("Building corresponding Mastodon status");
+
+                let mut status_builder = StatusBuilder::new();
+
+                status_builder.status(&status_text).media_ids(status_medias);
+
+                if let Some(i) = toot_reply_id {
+                    status_builder.in_reply_to(&i);
+                }
+
+                // can be activated for test purposes
+                // status_builder.visibility(elefren::status_builder::Visibility::Private);
+
+                let status = status_builder
+                    .build()
+                    .unwrap_or_else(|_| panic!("Cannot build status with text {}", &status_text));
+
+                // publish status
+                // again unwrap is safe here as we are in the main thread
+                let published_status = mastodon.new_status(status).unwrap();
+                // this will panic if it cannot publish the status, which is a good thing, it allows the
+                // last_tweet gathered not to be written
+
+                let ttt_towrite = TweetToToot {
+                    twitter_screen_name: mastodon_config.twitter_screen_name.clone(),
+                    tweet_id: tweet.id,
+                    toot_id: published_status.id,
+                };
+
+                // write the current state (tweet ID and toot ID) to avoid copying it another time
+                let lconn = task_conn.lock().await;
+                write_state(&lconn, ttt_towrite)
+                    .unwrap_or_else(|e| panic!("Can’t write the last tweet retrieved: {}", e));
+                drop(lconn);
+            }
         });
-        // retrieve the last tweet ID for the username
-        let last_tweet_id = read_state(&conn, &mastodon_config.twitter_screen_name, None)
-            .unwrap_or_else(|e| panic!("Cannot retrieve last_tweet_id: {}", e))
-            .map(|s| s.tweet_id);
 
-        // get Mastodon instance
-        let mastodon = get_mastodon_token(mastodon_config);
+        // push each task into the vec task
+        mtask.push(task);
+    }
 
-        // get user timeline feed (Vec<tweet>)
-        let mut feed = get_user_timeline(mastodon_config, &token, last_tweet_id)
-            .await
-            .unwrap_or_else(|e| {
-                panic!(
-                    "Something went wrong when trying to retrieve {}’s timeline: {}",
-                    &mastodon_config.twitter_screen_name, e
-                )
-            });
-
-        // empty feed -> exiting
-        if feed.is_empty() {
-            info!("Nothing to retrieve since last time, exiting…");
-            return;
-        }
-
-        // order needs to be chronological
-        feed.reverse();
-
-        for tweet in &feed {
-            debug!("Treating Tweet {} inside feed", tweet.id);
-            // initiate the toot_reply_id var
-            let mut toot_reply_id: Option<String> = None;
-            // determine if the tweet is part of a thread (response to self) or a standard response
-            if let Some(r) = &tweet.in_reply_to_screen_name {
-                if r.to_lowercase() != mastodon_config.twitter_screen_name.to_lowercase() {
-                    // we are responding not threadin
-                    info!("Tweet is a direct response, skipping");
-                    continue;
-                }
-                info!("Tweet is a thread");
-                toot_reply_id = read_state(
-                    &conn,
-                    &mastodon_config.twitter_screen_name,
-                    tweet.in_reply_to_status_id,
-                )
-                .unwrap_or(None)
-                .map(|s| s.toot_id);
-            };
-
-            // build basic status by just yielding text and dereferencing contained urls
-            let mut status_text = build_basic_status(tweet);
-
-            let mut status_medias: Vec<String> = vec![];
-            // reupload the attachments if any
-            if let Some(m) = &tweet.extended_entities {
-                for media in &m.media {
-                    let local_tweet_media_path =
-                        match get_tweet_media(media, &config.scootaloo.cache_path).await {
-                            Ok(m) => m,
-                            Err(e) => {
-                                error!("Cannot get tweet media for {}: {}", &media.url, e);
-                                continue;
-                            }
-                        };
-
-                    let mastodon_media_ids = match mastodon
-                        .media(Cow::from(local_tweet_media_path.to_owned()))
-                    {
-                        Ok(m) => {
-                            remove_file(&local_tweet_media_path)
-                            .await
-                            .unwrap_or_else(|e|
-                                warn!("Attachment for {} has been uploaded, but I’m unable to remove the existing file: {}", &local_tweet_media_path, e)
-                            );
-                            m.id
-                        }
-                        Err(e) => {
-                            error!(
-                                "Attachment {} cannot be uploaded to Mastodon Instance: {}",
-                                &local_tweet_media_path, e
-                            );
-                            continue;
-                        }
-                    };
-
-                    status_medias.push(mastodon_media_ids);
-
-                    // last step, removing the reference to the media from with the toot’s text
-                    status_text = status_text.replace(&media.url, "");
-                }
-            }
-            // finished reuploading attachments, now let’s do the toot baby!
-
-            debug!("Building corresponding Mastodon status");
-
-            let mut status_builder = StatusBuilder::new();
-
-            status_builder.status(&status_text).media_ids(status_medias);
-
-            if let Some(i) = toot_reply_id {
-                status_builder.in_reply_to(&i);
-            }
-
-            let status = status_builder
-                .build()
-                .unwrap_or_else(|_| panic!("Cannot build status with text {}", &status_text));
-
-            // publish status
-            // again unwrap is safe here as we are in the main thread
-            let published_status = mastodon.new_status(status).unwrap();
-            // this will panic if it cannot publish the status, which is a good thing, it allows the
-            // last_tweet gathered not to be written
-
-            let ttt_towrite = TweetToToot {
-                twitter_screen_name: mastodon_config.twitter_screen_name.clone(),
-                tweet_id: tweet.id,
-                toot_id: published_status.id,
-            };
-
-            // write the current state (tweet ID and toot ID) to avoid copying it another time
-            write_state(&conn, ttt_towrite)
-                .unwrap_or_else(|e| panic!("Can’t write the last tweet retrieved: {}", e));
-        }
+    // launch and wait for every handle
+    for handle in mtask {
+        handle.await.unwrap();
     }
 }

src/mastodon.rs

@@ -101,9 +101,9 @@ pub fn register(host: &str, screen_name: &str) {
 
     println!(
         "Please insert the following block at the end of your configuration file:
-        \n[mastodon.{}]
-        \ntwitter_screen_name = \"{}\"
-        \n{}",
+[mastodon.{}]
+twitter_screen_name = \"{}\"
+{}",
         screen_name.to_lowercase(),
         screen_name,
         toml

src/twitter.rs

@@ -1,4 +1,3 @@
-use crate::config::MastodonConfig;
 use crate::config::TwitterConfig;
 use crate::util::cache_media;
 use crate::ScootalooError;
@@ -30,20 +29,15 @@ pub fn get_oauth2_token(config: &TwitterConfig) -> Token {
 
 /// Gets Twitter user timeline
 pub async fn get_user_timeline(
-    config: &MastodonConfig,
+    screen_name: &str,
     token: &Token,
     lid: Option<u64>,
 ) -> Result<Vec<Tweet>, Box<dyn Error>> {
     // fix the page size to 200 as it is the maximum Twitter authorizes
-    let (_, feed) = user_timeline(
-        UserID::from(config.twitter_screen_name.to_owned()),
-        true,
-        false,
-        token,
-    )
-    .with_page_size(200)
-    .older(lid)
-    .await?;
+    let (_, feed) = user_timeline(UserID::from(screen_name.to_owned()), true, false, token)
+        .with_page_size(200)
+        .older(lid)
+        .await?;
 
     Ok(feed.to_vec())
 }

src/util.rs

@@ -1,11 +1,100 @@
-use crate::ScootalooError;
+use crate::{twitter::get_tweet_media, ScootalooError};
+
+use std::{borrow::Cow, error::Error};
+
+use egg_mode::tweet::Tweet;
+
+use elefren::prelude::*;
+
+use log::{error, info, warn};
+
 use reqwest::Url;
-use std::error::Error;
+
 use tokio::{
-    fs::{create_dir_all, File},
+    fs::{create_dir_all, remove_file, File},
     io::copy,
 };
 
+/// Generate associative table between media ids and tweet extended entities
+pub async fn generate_media_ids(
+    tweet: &Tweet,
+    cache_path: &str,
+    mastodon: &Mastodon,
+) -> (String, Vec<String>) {
+    let mut media_url = "".to_string();
+    let mut media_ids: Vec<String> = vec![];
+
+    if let Some(m) = &tweet.extended_entities {
+        // create tasks list
+        let mut tasks = vec![];
+
+        // size of media_ids vector, should be equal to the media vector
+        media_ids.resize(m.media.len(), String::new());
+
+        info!("{} medias in tweet", m.media.len());
+
+        for (i, media) in m.media.iter().enumerate() {
+            // attribute media url
+            media_url = media.url.clone();
+
+            // clone everything we need
+            let cache_path = String::from(cache_path);
+            let media = media.clone();
+            let mastodon = mastodon.clone();
+
+            let task = tokio::task::spawn(async move {
+                info!("Start treating {}", media.media_url_https);
+                // get the tweet embedded media
+                let local_tweet_media_path = match get_tweet_media(&media, &cache_path).await {
+                    Ok(l) => l,
+                    Err(e) => {
+                        return Err(ScootalooError::new(&format!(
+                            "Cannot get tweet media for {}: {}",
+                            &media.url, e
+                        )))
+                    }
+                };
+
+                // upload media to Mastodon
+                let mastodon_media = mastodon.media(Cow::from(local_tweet_media_path.to_owned()));
+                // at this point, we can safely erase the original file
+                // it doesn’t matter if we can’t remove, cache_media fn is idempotent
+                remove_file(&local_tweet_media_path).await.ok();
+
+                let mastodon_media = match mastodon_media {
+                    Ok(m) => m,
+                    Err(e) => {
+                        return Err(ScootalooError::new(&format!(
+                            "Attachment {} cannot be uploaded to Mastodon Instance: {}",
+                            &local_tweet_media_path, e
+                        )))
+                    }
+                };
+
+                Ok((i, mastodon_media.id))
+            });
+
+            tasks.push(task);
+        }
+
+        for task in tasks {
+            match task.await {
+                // insert the media at the right place
+                Ok(Ok((i, v))) => media_ids[i] = v,
+                Ok(Err(e)) => warn!("{}", e),
+                Err(e) => error!("Something went wrong when joining the main thread: {}", e),
+            }
+        }
+    } else {
+        info!("No media in tweet");
+    }
+
+    // in case some media_ids slot remained empty due to errors, remove them
+    media_ids.retain(|x| !x.is_empty());
+
+    (media_url, media_ids)
+}
+
 /// Gets and caches Twitter Media inside the determined temp dir
 pub async fn cache_media(u: &str, t: &str) -> Result<String, Box<dyn Error>> {
     // create dir