Merge branch '14-do-not-notify-subscribers-for-shorts' into 'master'

Resolve "Do not notify subscribers for Shorts"

Closes #14

See merge request veretcle/tootube!29

Cargo.toml

@@ -1,21 +1,23 @@
 [package]
 name = "tootube"
 authors = ["VC <veretcle+framagit@mateu.be>"]
-version = "0.4.1"
+version = "0.10.0"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+rpassword = "^7.3"
 reqwest = { version = "^0.11", features = ["json", "stream", "multipart"] }
 tokio = { version = "^1", features = ["full"] }
 clap = "^4"
-serde = { version = "1.0", features = ["derive"] }
+serde = { version = "^1", features = ["derive"] }
 toml = "^0.5"
 log = "^0.4"
 env_logger = "^0.10"
-futures-core = "0.3.28"
+indicatif = "^0.17"
-bytes = "1.5.0"
+async-stream = "^0.3"
+futures-util = "^0.3"
 
 [profile.release]
 strip = true

README.md

@@ -1,6 +1,6 @@
 # What is it?
 
-This program takes the very last video published on a PeerTube instance and pushes it to a corresponding YouTube channel.
+This program takes the very last or a specified video published on a PeerTube instance and pushes it to a corresponding YouTube channel.
 
 # What the status of this?
 
@@ -12,7 +12,8 @@ So consider this a work in progress that will slowly get better with time.
 
 # What does it do exactly?
 
-* it retrieves the latest PeerTube video download URL from an instance
+* it retrieves the latest (or a specified) PeerTube video download URL from an instance
+* if you register the app into your PeerTube instance (optional), you can retrieve the latest video source instead of the highest quality encoded video
 * it creates a resumable upload into the target YouTube account
 * it downloads/uploads the latest PeerTube video to YouTube without using a cache (stream-to-stream)
 
@@ -46,8 +47,22 @@ Create your `tootube.toml` config file:
 ```toml
 [peertube]
 base_url="https://p.nintendojo.fr"
+# optional
+# allows you to delete the original video source file once it’s uploaded to YouTube
+delete_video_source_after_transfer=true # this option is only available if you have Administrator privileges
+# optional
+# everything below is given by the register command with --peertube option
+[peertube.oauth2]
+client_id="<YOUR CLIENT_ID>"
+client_secret="<YOUR CLIENT_SECRET>"
+refresh_token="<YOUR CLIENT TOKEN>"
 
 [youtube]
+notify_subscribers_on_shorts=false # will you notify subscribers for shorts?
+# optional
+# allows you to notify subscribers when transferring shorts, defaults to false
+
+[youtube.oauth2]
 refresh_token="" # leave empty for now
 client_id="<YOUR CLIENT_ID>"
 client_secret="<YOUR CLIENT_SECRET>"
@@ -56,7 +71,15 @@ client_secret="<YOUR CLIENT_SECRET>"
 Then run:
 
 ```bash
-tootube register --config <PATH TO YOUR TOOTUBE.TOML FILE>
+tootube register --youtube --config <PATH TO YOUR TOOTUBE.TOML FILE>
 ```
 
-You’ll be then prompted with all the necessary information to register `tootube`. You’ll end with a `refresh_token` that you can now paste inside your tootube configuration.
+You’ll be then prompted with all the necessary information to register `tootube`. You’ll end with a `refresh_token` that you will be written back to the config file.
+
+If you wish to register `tootube` on PeerTube, you can do so by using:
+
+```bash
+tootube register --peertube --config <PATH TO YOUR TOOTUBE.TOML FILE>
+```
+
+It will require your username/password (beware that 2FA is not supported for this feature as of now) and generate a first `refresh_token` that will be written back to the config file.

src/config.rs

@@ -1,32 +1,118 @@
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
-use std::fs::read_to_string;
+use std::{
+    error::Error,
+    fs::{read_to_string, write},
+};
 
 // General configuration Struct
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Deserialize, Serialize)]
 pub struct Config {
     pub peertube: PeertubeConfig,
     pub youtube: YoutubeConfig,
+    #[serde(default)]
+    pub tootube: TootubeConfig,
 }
 
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Deserialize, Serialize)]
 pub struct PeertubeConfig {
     pub base_url: String,
+    #[serde(default)]
+    pub delete_video_source_after_transfer: bool,
+    pub oauth2: Option<PeertubeConfigOauth2>,
 }
 
-#[derive(Debug, Deserialize)]
+impl Default for PeertubeConfig {
+    fn default() -> Self {
+        PeertubeConfig {
+            base_url: "".to_string(),
+            delete_video_source_after_transfer: false,
+            oauth2: None,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct PeertubeConfigOauth2 {
+    pub client_id: String,
+    pub client_secret: String,
+    pub refresh_token: String,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
 pub struct YoutubeConfig {
+    #[serde(default)]
+    pub notify_subscribers_on_shorts: bool,
+    pub oauth2: YoutubeConfigOauth2,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct YoutubeConfigOauth2 {
     pub refresh_token: String,
     pub client_id: String,
     pub client_secret: String,
 }
 
-/// Parses the TOML file into a Config struct
+#[derive(Debug, Deserialize, Serialize)]
-pub fn parse_toml(toml_file: &str) -> Config {
+pub struct TootubeConfig {
-    let toml_config =
+    pub progress_bar: String,
-        read_to_string(toml_file).unwrap_or_else(|e| panic!("Cannot open file {toml_file}: {e}"));
+    pub progress_chars: String,
-
+}
-    let config: Config = toml::from_str(&toml_config)
+
-        .unwrap_or_else(|e| panic!("Cannot parse TOML file {toml_file}: {e}"));
+impl Default for TootubeConfig {
-
+    fn default() -> Self {
-    config
+        TootubeConfig {
+            progress_bar: "{msg}\n{spinner:.green} [{elapsed_precise}] [{wide_bar:.cyan/blue}] {bytes}/{total_bytes} ({bytes_per_sec}, {eta})".to_string(),
+            progress_chars: "#>-".to_string(),
+        }
+    }
+}
+
+impl Config {
+    /// Parses the TOML file into a Config struct
+    pub fn new(toml_file: &str) -> Self {
+        let toml_config = read_to_string(toml_file)
+            .unwrap_or_else(|e| panic!("Cannot open file {toml_file}: {e}"));
+
+        let config: Config = toml::from_str(&toml_config)
+            .unwrap_or_else(|e| panic!("Cannot parse TOML file {toml_file}: {e}"));
+
+        config
+    }
+
+    pub fn dump(&mut self, toml_file: &str) -> Result<(), Box<dyn Error>> {
+        // bring back the default for progress bar
+        self.tootube = TootubeConfig::default();
+        write(toml_file, toml::to_string_pretty(self)?)?;
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::fs::{remove_file, write};
+
+    #[test]
+    fn test_minimal_conf() {
+        let file_name = "/tmp/test_minimal_conf.toml";
+        let data = r#"
+[peertube]
+base_url = 'https://p.nintendojo.fr'
+
+[youtube.oauth2]
+refresh_token = 'rt_N/A'
+client_id = 'ci_N/A'
+client_secret = 'cs_N/A'
+"#;
+        write(file_name, data).unwrap();
+        let config = Config::new(file_name);
+
+        assert_eq!(config.youtube.notify_subscribers_on_shorts, false);
+        assert_eq!(config.tootube.progress_chars, "#>-");
+        assert_eq!(config.youtube.oauth2.refresh_token, "rt_N/A");
+        assert_eq!(config.youtube.oauth2.client_id, "ci_N/A");
+        assert_eq!(config.youtube.oauth2.client_secret, "cs_N/A");
+
+        remove_file(file_name).unwrap();
+    }
 }

src/lib.rs

@@ -1,56 +1,137 @@
-use bytes::Bytes;
-use futures_core::stream::Stream;
 use log::debug;
-use std::error::Error;
 
 mod error;
 
 mod config;
-pub use config::parse_toml;
+pub use config::Config;
-use config::Config;
+pub use config::PeertubeConfigOauth2;
 
 mod peertube;
-use peertube::{get_latest_video, get_max_resolution_dl};
+pub use peertube::register as register_peertube;
+use peertube::{get_playlists_to_be_added_to, PeerTube};
 
 mod youtube;
-pub use youtube::register;
+pub use youtube::register as register_youtube;
-use youtube::{add_video_to_playlists, create_resumable_upload, now_kiss};
+use youtube::YouTube;
-
-async fn get_dl_video_stream(
-    u: &str,
-) -> Result<impl Stream<Item = Result<Bytes, reqwest::Error>>, Box<dyn Error>> {
-    Ok(reqwest::get(u).await?.bytes_stream())
-}
 
+/// This is where the magic happens
+/// This takes the main options from config & command line args
+/// and sends back the updated PeerTube refresh_token if any
 #[tokio::main]
-pub async fn run(config: Config, pl: Vec<String>) {
+pub async fn run(config: &Config, pl: Vec<String>, pt_video_id: Option<&str>) -> Option<String> {
-    // Get the latest video object
+    // Create PeerTube struct
-    let latest_vid = get_latest_video(&config.peertube.base_url)
+    let peertube = match &config.peertube.oauth2 {
-        .await
+        Some(s) => PeerTube::new(&config.peertube.base_url)
-        .unwrap_or_else(|e| {
+            .with_client(s)
+            .await
+            .unwrap_or_else(|e| panic!("Cannot instantiate PeerTube struct: {}", e)),
+        None => PeerTube::new(&config.peertube.base_url),
+    };
+    // Get the latest video object or the targeted pt_video_id
+    let latest_vid = match pt_video_id {
+        None => peertube.get_latest_video().await.unwrap_or_else(|e| {
             panic!("Cannot retrieve the latest video, something must have gone terribly wrong: {e}")
-        });
+        }),
+        Some(v) => peertube.get_video_detail(v).await.unwrap_or_else(|e| {
+            panic!(
+                "Cannot retrieve the specified video, something must have gone terribly wrong: {e}"
+            )
+        }),
+    };
+
+    // We have a refresh_token, try to use it
+    let source_url = match &config.peertube.oauth2 {
+        Some(_) => peertube
+            .get_original_video_source(&latest_vid.uuid)
+            .await
+            .ok(),
+        None => None,
+    };
+
+    // Whatever happens, collect the highest quality possible
+    let high_quality_url = latest_vid.streaming_playlists.as_ref().unwrap()[0]
+        .files
+        .iter()
+        .max()
+        .unwrap()
+        .file_download_url
+        .clone();
+
+    // dl_url corresponds to source url if available, best quality if not
+    let dl_url = match source_url {
+        Some(s) => s,
+        None => high_quality_url,
+    };
 
-    let dl_url = get_max_resolution_dl(latest_vid.streaming_playlists.as_ref().unwrap());
     debug!("PT download URL: {}", &dl_url);
 
-    let pt_stream = get_dl_video_stream(&dl_url)
+    let youtube = YouTube::new(
-        .await
+        &config.youtube.oauth2.client_id,
-        .unwrap_or_else(|e| panic!("Cannot download video at URL {}: {}", dl_url, e));
+        &config.youtube.oauth2.client_secret,
+        &config.youtube.oauth2.refresh_token,
+    )
+    .await
+    .unwrap_or_else(|e| panic!("Cannot instantiate YouTube struct: {}", e));
 
-    let resumable_upload_url = create_resumable_upload(&config.youtube, &latest_vid)
+    // Do not notify when notify_subscribers_on_shorts = False and latest_id is a short
+    debug!(
+        "Will user get notified? {}",
+        !(!config.youtube.notify_subscribers_on_shorts & latest_vid.is_short())
+    );
+
+    let resumable_upload_url = youtube
+        .create_resumable_upload(
+            &latest_vid,
+            !(!config.youtube.notify_subscribers_on_shorts & latest_vid.is_short()),
+        )
         .await
         .unwrap_or_else(|e| panic!("Cannot retrieve the upload’s resumable id: {e}"));
     debug!("YT upload URL: {}", &resumable_upload_url);
 
-    let yt_video_id = now_kiss(pt_stream, &resumable_upload_url, &config.youtube)
+    let yt_video_id = youtube
+        .now_kiss(&dl_url, &resumable_upload_url, &config.tootube)
         .await
         .unwrap_or_else(|e| panic!("Cannot resume upload!: {e}"));
     debug!("YT video ID: {}", &yt_video_id);
 
+    youtube
+        .set_thumbnail(
+            &yt_video_id,
+            &format!("{}{}", &config.peertube.base_url, latest_vid.preview_path),
+        )
+        .await
+        .unwrap_or_else(|e| panic!("Cannot upload the thumbnail: {}", e));
+
     if !pl.is_empty() {
-        add_video_to_playlists(&config.youtube, &yt_video_id, &pl)
+        youtube
+            .add_video_to_playlists(&yt_video_id, &pl)
             .await
             .unwrap_or_else(|e| panic!("Cannot add video to playlist(s): {e}"));
     }
+
+    // delete the source video if requested (it won’t be used anymore)
+    if config.peertube.delete_video_source_after_transfer {
+        peertube
+            .delete_original_video_source(&latest_vid.uuid)
+            .await
+            .unwrap_or_else(|e| panic!("Cannot delete source video: {e}"));
+
+        debug!("Original Video {} has been deleted", &latest_vid.uuid);
+    }
+
+    // Updates the playlist of PeerTube if necessary
+    if config.peertube.oauth2.is_some() && !pl.is_empty() {
+        debug!("Updating playlists on PeerTube");
+
+        if let Ok(pl_to_add_to) =
+            get_playlists_to_be_added_to(&peertube, &latest_vid.uuid, &pl, latest_vid.channel.id)
+                .await
+        {
+            for p in pl_to_add_to {
+                let _ = peertube.add_video_to_playlist(&latest_vid.uuid, &p).await;
+            }
+        }
+    }
+
+    peertube.refresh_token
 }

src/main.rs

@@ -10,6 +10,7 @@ fn main() {
         .arg(
             Arg::new("config")
                 .short('c')
+                .global(true)
                 .long("config")
                 .value_name("CONFIG_FILE")
                 .help("TOML config file for tootube")
@@ -26,31 +27,74 @@ fn main() {
                 .num_args(0..)
                 .display_order(2),
         )
+        .arg(
+            Arg::new("id")
+                .short('i')
+                .long("id")
+                .value_name("ID")
+                .help("Specify the PeerTube Video ID")
+                .num_args(0..=1)
+                .display_order(3),
+        )
+        .arg(
+            Arg::new("vice")
+                .long("vice")
+                .aliases(["chybrare", "coquinou"])
+                .action(clap::ArgAction::SetTrue)
+                .display_order(4),
+        )
         .subcommand(
             Command::new("register")
                 .version(env!("CARGO_PKG_VERSION"))
-                .about("Command to register to YouTube OAuth2.0")
+                .about("Command to register to YouTube or PeerTube OAuth2.0")
                 .arg(
-                    Arg::new("config")
+                    Arg::new("youtube")
-                        .short('c')
+                        .long("youtube")
-                        .long("config")
+                        .short('y')
-                        .value_name("CONFIG_FILE")
+                        .required(true)
-                        .help("TOML config file for tootube")
+                        .conflicts_with("peertube")
-                        .num_args(1)
+                        .action(clap::ArgAction::SetTrue),
-                        .default_value(DEFAULT_CONFIG_PATH)
+                )
-                        .display_order(1),
+                .arg(
+                    Arg::new("peertube")
+                        .long("peertube")
+                        .short('p')
+                        .required(true)
+                        .action(clap::ArgAction::SetTrue),
                 ),
         )
         .get_matches();
 
     if let Some(("register", sub_m)) = matches.subcommand() {
-        let config = parse_toml(sub_m.get_one::<String>("config").unwrap());
+        let mut config = Config::new(sub_m.get_one::<String>("config").unwrap());
-        register(&config.youtube)
+
-            .unwrap_or_else(|e| panic!("Cannot register to YouTube API: {}", e));
+        if sub_m.get_flag("youtube") {
+            let yt_refresh_token = register_youtube(&config.youtube.oauth2)
+                .unwrap_or_else(|e| panic!("Cannot register to YouTube API: {}", e));
+
+            config.youtube.oauth2.refresh_token = yt_refresh_token;
+        }
+
+        if sub_m.get_flag("peertube") {
+            let pt_oauth2 = register_peertube(&config.peertube)
+                .unwrap_or_else(|e| panic!("Cannot register to PeerTube API: {}", e));
+
+            config.peertube.oauth2 = Some(pt_oauth2);
+        }
+
+        config
+            .dump(sub_m.get_one::<String>("config").unwrap())
+            .unwrap_or_else(|e| panic!("Cannot write back to Tootube Config file: {}", e));
+
         return;
     }
 
-    let config = parse_toml(matches.get_one::<String>("config").unwrap());
+    let mut config = Config::new(matches.get_one::<String>("config").unwrap());
+
+    if matches.get_flag("vice") {
+        config.tootube.progress_bar = "{msg}\n[{elapsed_precise}] 8{wide_bar:.magenta}ᗺ {bytes}/{total_bytes} ({bytes_per_sec}, {eta})".to_string();
+        config.tootube.progress_chars = "=D ".to_string();
+    }
 
     let playlists: Vec<String> = matches
         .get_many::<String>("playlists")
@@ -58,7 +102,18 @@ fn main() {
         .map(|v| v.to_string())
         .collect();
 
+    let pt_video_id = matches.get_one::<String>("id").map(|s| s.as_str());
+
     env_logger::init();
 
-    run(config, playlists);
+    // runs the main program logic & retrieves the updated PeerTube refresh_token
+    if let Some(x) = run(&config, playlists, pt_video_id) {
+        config.peertube.oauth2 = config.peertube.oauth2.map(|y| PeertubeConfigOauth2 {
+            client_id: y.client_id,
+            client_secret: y.client_secret,
+            refresh_token: x,
+        });
+
+        let _ = config.dump(matches.get_one::<String>("config").unwrap());
+    };
 }

src/peertube.rs

@@ -1,9 +1,16 @@
-use serde::Deserialize;
+use crate::{config::PeertubeConfig, config::PeertubeConfigOauth2, error::TootubeError};
-use std::{boxed::Box, cmp::Ordering, error::Error};
+use log::debug;
+use reqwest::{
+    header::{HeaderMap, HeaderValue},
+    multipart::Form,
+    Client,
+};
+use rpassword::prompt_password;
+use serde::{Deserialize, Serialize};
+use std::{boxed::Box, cmp::Ordering, error::Error, io::stdin};
 
 #[derive(Debug, Deserialize)]
 pub struct PeerTubeVideos {
-    pub total: u64,
     pub data: Vec<PeerTubeVideo>,
 }
 
@@ -12,9 +19,29 @@ pub struct PeerTubeVideo {
     pub name: String,
     pub uuid: String,
     pub description: String,
+    pub duration: u64,
+    #[serde(rename = "aspectRatio")]
+    pub aspect_ratio: Option<f32>,
+    #[serde(rename = "previewPath")]
+    pub preview_path: String,
     #[serde(rename = "streamingPlaylists")]
     pub streaming_playlists: Option<Vec<PeerTubeVideoStreamingPlaylists>>,
     pub tags: Option<Vec<String>>,
+    pub channel: PeerTubeVideoChannel,
+}
+
+impl PeerTubeVideo {
+    pub fn is_short(&self) -> bool {
+        if self.duration < 60 && self.aspect_ratio.is_some_and(|x| x == 0.5625) {
+            return true;
+        }
+        false
+    }
+}
+
+#[derive(Debug, Deserialize)]
+pub struct PeerTubeVideoChannel {
+    pub id: u8,
 }
 
 #[derive(Debug, Deserialize)]
@@ -22,9 +49,46 @@ pub struct PeerTubeVideoStreamingPlaylists {
     pub files: Vec<PeerTubeVideoStreamingPlaylistsFiles>,
 }
 
+#[derive(Debug, Deserialize)]
+struct PeerTubeOauthClientsLocalResponse {
+    client_id: String,
+    client_secret: String,
+}
+
+#[derive(Debug, Serialize)]
+struct PeerTubeUsersToken {
+    client_id: String,
+    client_secret: String,
+    grant_type: String,
+    password: Option<String>,
+    username: Option<String>,
+    refresh_token: Option<String>,
+}
+
+#[derive(Debug, Deserialize)]
+struct PeerTubeUsersTokenResponse {
+    refresh_token: String,
+    access_token: String,
+}
+
+#[derive(Debug, Deserialize)]
+struct PeerTubeVideoSourceResponse {
+    #[serde(rename = "fileDownloadUrl")]
+    pub file_download_url: String,
+}
+
+#[derive(Debug, Deserialize)]
+struct PeerTubeVideoTokenResponse {
+    files: PeerTubeVideoTokenResponseFiles,
+}
+
+#[derive(Debug, Deserialize)]
+struct PeerTubeVideoTokenResponseFiles {
+    token: String,
+}
+
 #[derive(Eq, Debug, Deserialize)]
 pub struct PeerTubeVideoStreamingPlaylistsFiles {
-    pub id: u64,
     pub resolution: PeerTubeVideoStreamingPlaylistsFilesResolution,
     #[serde(rename = "fileDownloadUrl")]
     pub file_download_url: String,
@@ -53,60 +117,441 @@ pub struct PeerTubeVideoStreamingPlaylistsFilesResolution {
     pub id: u16,
 }
 
-/// This gets the last video uploaded to the PeerTube server
+#[derive(Debug, Deserialize)]
-pub async fn get_latest_video(u: &str) -> Result<PeerTubeVideo, Box<dyn Error>> {
+pub struct PeerTubeVideoPlaylists {
-    let body = reqwest::get(format!("{}/api/v1/videos?count=1&sort=-publishedAt", u))
+    pub total: u16,
-        .await?
+    pub data: Vec<PeerTubeVideoPlaylist>,
-        .json::<PeerTubeVideos>()
-        .await?;
-
-    let vid = get_video_detail(u, &body.data[0].uuid).await?;
-
-    Ok(vid)
 }
 
-/// This returns the direct download URL for with the maximum resolution
+#[derive(Debug, Deserialize)]
-pub fn get_max_resolution_dl(p: &[PeerTubeVideoStreamingPlaylists]) -> String {
+pub struct PeerTubeVideoPlaylist {
-    p[0].files.iter().max().unwrap().file_download_url.clone()
+    pub uuid: String,
+    #[serde(rename = "displayName")]
+    pub display_name: String,
 }
 
-/// This gets all the crispy details about one particular video
+#[derive(Debug, Serialize)]
-async fn get_video_detail(u: &str, v: &str) -> Result<PeerTubeVideo, Box<dyn Error>> {
+struct PeerTubeVideoPlaylistsVideos {
-    let body = reqwest::get(format!("{}/api/v1/videos/{}", u, v))
+    #[serde(rename = "videoId")]
+    video_id: String,
+}
+
+#[derive(Debug, Deserialize)]
+struct PeerTubeVideoPlaylistResponse {
+    #[serde(rename = "videoPlaylist")]
+    video_playlist: PeerTubeVideoPlaylistResponseVideoPlaylist,
+}
+
+#[derive(Debug, Deserialize)]
+struct PeerTubeVideoPlaylistResponseVideoPlaylist {
+    uuid: String,
+}
+
+#[derive(Debug, Deserialize)]
+struct PeerTubeVideoPlaylistsPlaylistIdVideos {
+    total: u32,
+    data: Vec<PeerTubeVideoPlaylistsPlaylistIdVideosData>,
+}
+
+#[derive(Debug, Deserialize)]
+struct PeerTubeVideoPlaylistsPlaylistIdVideosData {
+    video: PeerTubeVideo,
+}
+
+/// This function makes the registration process a little bit easier
+#[tokio::main]
+pub async fn register(config: &PeertubeConfig) -> Result<PeertubeConfigOauth2, Box<dyn Error>> {
+    // Get client ID/secret
+    let oauth2_client = reqwest::get(format!("{}/api/v1/oauth-clients/local", config.base_url))
         .await?
-        .json::<PeerTubeVideo>()
+        .json::<PeerTubeOauthClientsLocalResponse>()
         .await?;
 
-    Ok(body)
+    println!(
+        "Please type your PeerTube username for instance {}:",
+        config.base_url
+    );
+    let mut username = String::new();
+    stdin()
+        .read_line(&mut username)
+        .expect("Unable to read back the username!");
+
+    let password = prompt_password("Your password: ").expect("Unable to read back the password!");
+
+    let params = PeerTubeUsersToken {
+        client_id: oauth2_client.client_id.clone(),
+        client_secret: oauth2_client.client_secret.clone(),
+        grant_type: "password".to_string(),
+        username: Some(username.trim().to_string()),
+        password: Some(password.clone()),
+        refresh_token: None,
+    };
+
+    let client = Client::new();
+    let oauth2_token = client
+        .post(format!("{}/api/v1/users/token", config.base_url))
+        .form(&params)
+        .send()
+        .await?
+        .json::<PeerTubeUsersTokenResponse>()
+        .await?;
+
+    println!(
+        "The following lines will be written to the `peertube` section of your tootube.toml file:"
+    );
+    println!("[peertube.oauth2]");
+    println!("client_id=\"{}\"", oauth2_client.client_id);
+    println!("client_secret=\"{}\"", oauth2_client.client_secret);
+    println!("refresh_token=\"{}\"", oauth2_token.refresh_token);
+
+    Ok(PeertubeConfigOauth2 {
+        client_id: oauth2_client.client_id,
+        client_secret: oauth2_client.client_secret,
+        refresh_token: oauth2_token.refresh_token,
+    })
+}
+
+#[derive(Debug)]
+pub struct PeerTube {
+    base_url: String,
+    pub refresh_token: Option<String>,
+    client: Client,
+}
+
+impl PeerTube {
+    /// Create a new PeerTube struct with a basic embedded reqwest::Client
+    pub fn new(base_url: &str) -> Self {
+        PeerTube {
+            base_url: format!("{}/api/v1", base_url),
+            refresh_token: None,
+            client: Client::new(),
+        }
+    }
+
+    /// Retrieve the refresh_token and access_token and update the embedded reqwest::Client to have
+    /// the default required header
+    pub async fn with_client(
+        mut self,
+        pt_oauth2: &PeertubeConfigOauth2,
+    ) -> Result<Self, Box<dyn Error>> {
+        let params = PeerTubeUsersToken {
+            client_id: pt_oauth2.client_id.to_owned(),
+            client_secret: pt_oauth2.client_secret.to_owned(),
+            grant_type: "refresh_token".to_string(),
+            refresh_token: Some(pt_oauth2.refresh_token.to_owned()),
+            username: None,
+            password: None,
+        };
+
+        let req = self
+            .client
+            .post(format!("{}/users/token", self.base_url))
+            .form(&params)
+            .send()
+            .await?
+            .json::<PeerTubeUsersTokenResponse>()
+            .await?;
+
+        let mut headers = HeaderMap::new();
+        headers.insert(
+            "Authorization",
+            HeaderValue::from_str(&format!("Bearer {}", req.access_token))?,
+        );
+
+        self.client = reqwest::Client::builder()
+            .default_headers(headers)
+            .build()?;
+
+        self.refresh_token = Some(req.refresh_token);
+
+        Ok(self)
+    }
+
+    /// This gets the last video uploaded to the PeerTube server
+    pub async fn get_latest_video(&self) -> Result<PeerTubeVideo, Box<dyn Error>> {
+        let body = self
+            .client
+            .get(format!(
+                "{}/videos?count=1&sort=-publishedAt",
+                self.base_url
+            ))
+            .send()
+            .await?
+            .json::<PeerTubeVideos>()
+            .await?;
+
+        let vid = self.get_video_detail(&body.data[0].uuid).await?;
+
+        Ok(vid)
+    }
+
+    /// This gets all the crispy details about one particular video
+    pub async fn get_video_detail(&self, v: &str) -> Result<PeerTubeVideo, Box<dyn Error>> {
+        let body = self
+            .client
+            .get(format!("{}/videos/{}", self.base_url, v))
+            .send()
+            .await?
+            .json::<PeerTubeVideo>()
+            .await?;
+
+        Ok(body)
+    }
+
+    /// Get the original video source
+    pub async fn get_original_video_source(&self, uuid: &str) -> Result<String, Box<dyn Error>> {
+        let source_vid = self
+            .client
+            .get(format!("{}/videos/{}/source", self.base_url, uuid))
+            .send()
+            .await?
+            .json::<PeerTubeVideoSourceResponse>()
+            .await?;
+
+        debug!("Got the Source Vid URL: {}", &source_vid.file_download_url);
+
+        let video_file_token = self
+            .client
+            .post(format!("{}/videos/{}/token", self.base_url, uuid))
+            .send()
+            .await?
+            .json::<PeerTubeVideoTokenResponse>()
+            .await?;
+
+        debug!("Got the File Token: {}", &video_file_token.files.token);
+
+        Ok(format!(
+            "{}?videoFileToken={}",
+            source_vid.file_download_url, video_file_token.files.token
+        ))
+    }
+
+    /// Delete the original video source
+    pub async fn delete_original_video_source(&self, uuid: &str) -> Result<(), Box<dyn Error>> {
+        let res = self
+            .client
+            .delete(format!("{}/videos/{}/source/file", self.base_url, uuid))
+            .send()
+            .await?;
+
+        if !res.status().is_success() {
+            return Err(TootubeError::new(&format!(
+                "Cannot delete source video file {}: {}",
+                uuid,
+                res.text().await?
+            ))
+            .into());
+        }
+
+        Ok(())
+    }
+
+    /// List every playlists on PeerTube
+    pub async fn list_video_playlists(&self) -> Result<Vec<PeerTubeVideoPlaylist>, Box<dyn Error>> {
+        let mut playlists: Vec<PeerTubeVideoPlaylist> = vec![];
+
+        let mut start = 0;
+        let inc = 15;
+
+        while let Ok(mut local_pl) = self
+            .client
+            .get(format!(
+                "{}/video-playlists?count={}&playlistType=1&start={}",
+                self.base_url, inc, start
+            ))
+            .send()
+            .await?
+            .json::<PeerTubeVideoPlaylists>()
+            .await
+        {
+            start += inc;
+            playlists.append(&mut local_pl.data);
+            if start >= local_pl.total {
+                break;
+            }
+        }
+
+        Ok(playlists)
+    }
+
+    /// Add a public playlist
+    pub async fn create_video_playlist(
+        &self,
+        c_id: u8,
+        display_name: &str,
+    ) -> Result<String, Box<dyn Error>> {
+        let form = Form::new()
+            .text("displayName", display_name.to_string())
+            .text("privacy", "1")
+            .text("videoChannelId", format!("{}", c_id));
+
+        let pl_created = self
+            .client
+            .post(format!("{}/video-playlists", self.base_url))
+            .multipart(form)
+            .send()
+            .await?
+            .json::<PeerTubeVideoPlaylistResponse>()
+            .await?;
+
+        Ok(pl_created.video_playlist.uuid)
+    }
+
+    /// Add a video into a playlist
+    pub async fn add_video_to_playlist(
+        &self,
+        vid_uuid: &str,
+        pl_uuid: &str,
+    ) -> Result<(), Box<dyn Error>> {
+        let video_to_add = PeerTubeVideoPlaylistsVideos {
+            video_id: vid_uuid.to_string(),
+        };
+
+        let res = self
+            .client
+            .post(format!(
+                "{}/video-playlists/{}/videos",
+                self.base_url, pl_uuid
+            ))
+            .json(&video_to_add)
+            .send()
+            .await?;
+
+        if !res.status().is_success() {
+            return Err(TootubeError::new(&format!(
+                "Cannot add video {} to playlist {}: {}",
+                vid_uuid,
+                pl_uuid,
+                res.text().await?
+            ))
+            .into());
+        };
+
+        Ok(())
+    }
+
+    /// List all videos of a playlist
+    pub async fn list_videos_playlist(&self, uuid: &str) -> Result<Vec<String>, Box<dyn Error>> {
+        let mut videos: Vec<PeerTubeVideo> = vec![];
+
+        let mut start = 0;
+        let inc = 15;
+
+        while let Ok(l_vid) = self
+            .client
+            .get(format!(
+                "{}/video-playlists/{}/videos?start={}&count={}",
+                &self.base_url, &uuid, start, inc
+            ))
+            .send()
+            .await?
+            .json::<PeerTubeVideoPlaylistsPlaylistIdVideos>()
+            .await
+        {
+            start += inc;
+            videos.append(&mut l_vid.data.into_iter().map(|x| x.video).collect());
+            if start >= l_vid.total {
+                break;
+            }
+        }
+
+        Ok(videos.into_iter().map(|x| x.uuid).collect())
+    }
+}
+
+/// Given a PeerTube instance, video UUID, list of named playlists and channel ID, this function:
+/// * adds playlists if they do not exists
+/// * adds the video to said playlists if they’re not in it already
+pub async fn get_playlists_to_be_added_to(
+    peertube: &PeerTube,
+    vid_uuid: &str,
+    pl: &[String],
+    c_id: u8,
+) -> Result<Vec<String>, Box<dyn Error>> {
+    let mut playlist_to_be_added_to: Vec<String> = vec![];
+
+    if let Ok(local_pl) = peertube.list_video_playlists().await {
+        // list the displayNames of each playlist
+        let current_playlist: Vec<String> =
+            local_pl.iter().map(|s| s.display_name.clone()).collect();
+        // get the playlist whose displayName does not exist yet
+        let pl_to_create: Vec<_> = pl
+            .iter()
+            .filter(|x| !current_playlist.contains(x))
+            .collect();
+
+        debug!("Playlists to be added: {:?}", &pl_to_create);
+
+        playlist_to_be_added_to = local_pl
+            .into_iter()
+            .filter(|x| pl.contains(&x.display_name))
+            .map(|x| x.uuid.clone())
+            .collect();
+
+        // create the missing playlists
+        for p in pl_to_create {
+            if let Ok(s) = peertube.create_video_playlist(c_id, p).await {
+                playlist_to_be_added_to.push(s);
+            }
+        }
+
+        for p in playlist_to_be_added_to.clone().iter() {
+            if let Ok(s) = peertube.list_videos_playlist(p).await {
+                // if a video already exists inside a playlist, drop it
+                if s.contains(&vid_uuid.to_string()) {
+                    playlist_to_be_added_to.retain(|i| *i != *p)
+                }
+            }
+        }
+
+        debug!("Playlists to be added to: {:?}", &playlist_to_be_added_to);
+    };
+
+    Ok(playlist_to_be_added_to)
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
 
-    #[test]
+    #[tokio::test]
-    fn test_get_max_resolution_dl() {
+    async fn test_get_latest_video() {
-        let str_plist: Vec<PeerTubeVideoStreamingPlaylists> =
+        let peertube = PeerTube::new("https://peertube.cpy.re");
-            vec![PeerTubeVideoStreamingPlaylists {
-                files: vec![
-                    PeerTubeVideoStreamingPlaylistsFiles {
-                        id: 1025,
-                        resolution: PeerTubeVideoStreamingPlaylistsFilesResolution { id: 990 },
-                        file_download_url: "meh!".to_string(),
-                    },
-                    PeerTubeVideoStreamingPlaylistsFiles {
-                        id: 1027,
-                        resolution: PeerTubeVideoStreamingPlaylistsFilesResolution { id: 1720 },
-                        file_download_url: "blah".to_string(),
-                    },
-                    PeerTubeVideoStreamingPlaylistsFiles {
-                        id: 1026,
-                        resolution: PeerTubeVideoStreamingPlaylistsFilesResolution { id: 360 },
-                        file_download_url: "nope".to_string(),
-                    },
-                ],
-            }];
 
-        assert_eq!("blah".to_string(), get_max_resolution_dl(&str_plist));
+        let vid = peertube.get_latest_video().await.unwrap();
+
+        assert_eq!(vid.uuid.len(), 36);
+    }
+
+    #[tokio::test]
+    #[should_panic]
+    async fn test_get_original_video_source() {
+        let peertube = PeerTube::new("https://peertube.cpy.re");
+
+        let _vid = peertube
+            .get_original_video_source("t")
+            .await
+            .expect("Should panic!");
+    }
+
+    #[tokio::test]
+    async fn test_list_video_playlists() {
+        let peertube = PeerTube::new("https://peertube.cpy.re");
+
+        let pl = peertube.list_video_playlists().await.unwrap();
+
+        assert!(!pl.is_empty());
+    }
+
+    #[tokio::test]
+    async fn test_list_videos_playlist() {
+        let peertube = PeerTube::new("https://peertube.cpy.re");
+
+        let pl_videos = peertube
+            .list_videos_playlist("73a5c1fa-64c5-462d-81e5-b120781c2d72")
+            .await
+            .unwrap();
+
+        assert!(!pl_videos.is_empty());
     }
 }

src/youtube.rs

@@ -1,13 +1,19 @@
-use crate::{config::YoutubeConfig, error::TootubeError, peertube::PeerTubeVideo};
+use crate::{
-use bytes::Bytes;
+    config::{TootubeConfig, YoutubeConfigOauth2},
-use futures_core::stream::Stream;
+    error::TootubeError,
-use log::debug;
+    peertube::PeerTubeVideo,
-use reqwest::{multipart::Form, Body, Client};
+};
+use async_stream::stream;
+use futures_util::StreamExt;
+use indicatif::{ProgressBar, ProgressStyle};
+use log::{debug, warn};
+use reqwest::{
+    header::{HeaderMap, HeaderValue},
+    multipart::Form,
+    Body, Client,
+};
 use serde::{Deserialize, Serialize};
-use std::{error::Error, io::stdin};
+use std::{cmp::min, error::Error, io::stdin};
-use tokio::sync::OnceCell;
-
-static ACCESS_TOKEN: OnceCell<String> = OnceCell::const_new();
 
 #[derive(Serialize, Debug)]
 struct RefreshTokenRequest {
@@ -154,8 +160,9 @@ struct YoutubePlaylistListResponseItemSnippet {
 }
 
 /// This function makes the registration process a little bit easier
+/// It returns the expected refresh_token so that it can be written back to the file
 #[tokio::main]
-pub async fn register(config: &YoutubeConfig) -> Result<(), Box<dyn Error>> {
+pub async fn register(config: &YoutubeConfigOauth2) -> Result<String, Box<dyn Error>> {
     println!("Click on the link below to authorize {} to upload to YouTube and deal with your playlists:", env!("CARGO_PKG_NAME"));
     println!("https://accounts.google.com/o/oauth2/v2/auth?client_id={}&redirect_uri=urn:ietf:wg:oauth:2.0:oob&scope=https://www.googleapis.com/auth/youtube%20https://www.googleapis.com/auth/youtube.upload&response_type=code", config.client_id);
     println!("Paste the returned authorization code:");
@@ -181,189 +188,254 @@ pub async fn register(config: &YoutubeConfig) -> Result<(), Box<dyn Error>> {
 
     let refresh_token: RegistrationAccessTokenResponse = res.json().await?;
 
-    println!("You can now paste the following line inside the `youtube` section of your tootube.toml file:");
+    println!(
-
+        "The following line will be written to the `youtube` section of your tootube.toml file:"
+    );
     println!("refresh_token=\"{}\"", refresh_token.refresh_token);
 
-    Ok(())
+    Ok(refresh_token.refresh_token)
 }
 
-/// Ensures that Token has been refreshed and that it is unique
+pub struct YouTube {
-async fn refresh_token(config: &YoutubeConfig) -> Result<String, reqwest::Error> {
+    client: Client,
-    ACCESS_TOKEN
-        .get_or_try_init(|| async {
-            let refresh_token = RefreshTokenRequest {
-                refresh_token: config.refresh_token.clone(),
-                client_id: config.client_id.clone(),
-                client_secret: config.client_secret.clone(),
-                ..Default::default()
-            };
-
-            let client = Client::new();
-            let res = client
-                .post("https://accounts.google.com/o/oauth2/token")
-                .json(&refresh_token)
-                .send()
-                .await?;
-
-            let access_token: AccessTokenResponse = res.json().await?;
-
-            debug!("YT Access Token: {}", &access_token.access_token);
-            Ok(access_token.access_token)
-        })
-        .await
-        .cloned()
 }
 
-/// This function takes a list of playlists keyword and returns a list of playlist ID
+impl YouTube {
-async fn get_playlist_ids(
+    pub async fn new(
-    config: &YoutubeConfig,
+        client_id: &str,
-    pl: &[String],
+        client_secret: &str,
-) -> Result<Vec<String>, Box<dyn Error>> {
+        refresh_token: &str,
-    let mut page_token = String::new();
+    ) -> Result<Self, Box<dyn Error>> {
-    let mut playlists: Vec<String> = vec![];
+        let mut youtube = YouTube {
-
+            client: Client::new(),
-    let access_token = refresh_token(config).await?;
-    let client = Client::new();
-
-    while let Ok(local_pl) = client
-        .get(&format!(
-            "https://www.googleapis.com/youtube/v3/playlists?part=snippet&mine=true&pageToken={}",
-            page_token
-        ))
-        .header("Authorization", format!("Bearer {}", access_token))
-        .send()
-        .await?
-        .json::<YoutubePlaylistListResponse>()
-        .await
-    {
-        playlists.append(
-            &mut local_pl
-                .items
-                .iter()
-                .filter_map(|s| pl.contains(&s.snippet.title).then_some(s.id.clone()))
-                .collect(),
-        );
-
-        // if nextPageToken is present, continue the loop
-        match local_pl.next_page_token {
-            None => break,
-            Some(a) => page_token = a.clone(),
-        }
-    }
-
-    debug!("Playlists IDs: {:?}", &playlists);
-    Ok(playlists)
-}
-
-/// This function adds the video id to the corresponding named playlist(s)
-pub async fn add_video_to_playlists(
-    config: &YoutubeConfig,
-    v: &str,
-    pl: &[String],
-) -> Result<(), Box<dyn Error>> {
-    let access_token = refresh_token(config).await?;
-    let playlists_ids = get_playlist_ids(config, pl).await?;
-
-    for pl_id in playlists_ids {
-        let yt_pl_upload_params = YoutubePlaylistItemsParams {
-            snippet: YoutubePlaylistItemsParamsSnippet {
-                playlist_id: pl_id.clone(),
-                resource_id: YoutubePlaylistItemsParamsSnippetResourceId {
-                    video_id: v.to_string(),
-                    ..Default::default()
-                },
-                ..Default::default()
-            },
         };
 
-        let client = Client::new();
+        let refresh_token = RefreshTokenRequest {
-        let res = client
+            refresh_token: refresh_token.to_string(),
-            .post("https://youtube.googleapis.com/youtube/v3/playlistItems?part=snippet")
+            client_id: client_id.to_string(),
-            .header("Authorization", format!("Bearer {}", access_token))
+            client_secret: client_secret.to_string(),
-            .json(&yt_pl_upload_params)
+            ..Default::default()
+        };
+
+        let access_token = youtube
+            .client
+            .post("https://accounts.google.com/o/oauth2/token")
+            .json(&refresh_token)
+            .send()
+            .await?
+            .json::<AccessTokenResponse>()
+            .await?;
+
+        let mut headers = HeaderMap::new();
+        headers.insert(
+            "Authorization",
+            HeaderValue::from_str(&format!("Bearer {}", access_token.access_token))?,
+        );
+
+        youtube.client = reqwest::Client::builder()
+            .default_headers(headers)
+            .build()?;
+
+        Ok(youtube)
+    }
+
+    /// This function uploads the thumbnail from PeerTube
+    pub async fn set_thumbnail(
+        &self,
+        video_id: &str,
+        preview_url: &str,
+    ) -> Result<(), Box<dyn Error>> {
+        let res = reqwest::get(preview_url).await?;
+        let stream = res.bytes_stream();
+
+        let res = self
+            .client
+            .post(format!(
+                "https://www.googleapis.com/upload/youtube/v3/thumbnails/set?videoId={}&uploadType=media",
+                video_id
+            ))
+            .header("Content-Type", "application/octet-stream")
+            .body(Body::wrap_stream(stream))
             .send()
             .await?;
 
-        if !res.status().is_success() {
+        res.status().is_success().then_some(()).ok_or(
-            return Err(TootubeError::new(&format!(
+            TootubeError::new(&format!("Thumbnail not uploaded: {}", res.text().await?)).into(),
-                "Something went wrong when trying to add the video to a playlist: {}",
+        )
-                res.text().await?
+    }
+
+    /// This function takes a list of playlists keyword and returns a list of playlist ID
+    async fn get_playlist_ids(&self, pl: &[String]) -> Result<Vec<String>, Box<dyn Error>> {
+        let mut page_token = String::new();
+        let mut playlists: Vec<String> = vec![];
+
+        while let Ok(local_pl) = self.client
+            .get(format!(
+                "https://www.googleapis.com/youtube/v3/playlists?part=snippet&mine=true&pageToken={}",
+                page_token
             ))
-            .into());
+            .send()
+            .await?
+            .json::<YoutubePlaylistListResponse>()
+            .await
+        {
+            playlists.append(
+                &mut local_pl
+                    .items
+                    .iter()
+                    .filter_map(|s| pl.contains(&s.snippet.title).then_some(s.id.clone()))
+                    .collect(),
+            );
+            // if nextPageToken is present, continue the loop
+            match local_pl.next_page_token {
+                None => break,
+                Some(a) => page_token.clone_from(&a),
+            }
+        }
+
+        debug!("Playlists IDs: {:?}", &playlists);
+        Ok(playlists)
+    }
+
+    /// This function adds the video id to the corresponding named playlist(s)
+    pub async fn add_video_to_playlists(
+        &self,
+        v: &str,
+        pl: &[String],
+    ) -> Result<(), Box<dyn Error>> {
+        let playlists_ids = self.get_playlist_ids(pl).await?;
+
+        for pl_id in playlists_ids {
+            let yt_pl_upload_params = YoutubePlaylistItemsParams {
+                snippet: YoutubePlaylistItemsParamsSnippet {
+                    playlist_id: pl_id.clone(),
+                    resource_id: YoutubePlaylistItemsParamsSnippetResourceId {
+                        video_id: v.to_string(),
+                        ..Default::default()
+                    },
+                    ..Default::default()
+                },
+            };
+
+            let res = self
+                .client
+                .post("https://youtube.googleapis.com/youtube/v3/playlistItems?part=snippet")
+                .json(&yt_pl_upload_params)
+                .send()
+                .await?;
+
+            if !res.status().is_success() {
+                return Err(TootubeError::new(&format!(
+                    "Something went wrong when trying to add the video to a playlist: {}",
+                    res.text().await?
+                ))
+                .into());
+            }
+        }
+
+        Ok(())
+    }
+
+    /// This function creates a resumable YT upload, putting all the parameters in
+    pub async fn create_resumable_upload(
+        &self,
+        vid: &PeerTubeVideo,
+        notify: bool,
+    ) -> Result<String, Box<dyn Error>> {
+        if vid.name.chars().count() > 100 {
+            warn!(
+                "PT Video Title ({}) is too long, it will be truncated",
+                &vid.name
+            );
+        }
+
+        let upload_params = YoutubeUploadParams {
+            snippet: {
+                YoutubeUploadParamsSnippet {
+                    title: vid.name.chars().take(100).collect::<String>(),
+                    description: vid.description.clone(),
+                    tags: vid.tags.clone(),
+                    ..Default::default()
+                }
+            },
+            status: {
+                YoutubeUploadParamsStatus {
+                    ..Default::default()
+                }
+            },
+        };
+        debug!("YT upload params: {:?}", &upload_params);
+
+        let notify_subscriber = notify.then_some(()).map_or("False", |_| "True");
+
+        let res = self.client.post(format!("https://www.googleapis.com/upload/youtube/v3/videos?uploadType=resumable&part=snippet%2Cstatus&notifySubscribers={}", notify_subscriber))
+            .json(&upload_params)
+            .send().await?;
+
+        res.status()
+            .is_success()
+            .then_some(
+                res.headers()
+                    .get("location")
+                    .ok_or("Cannot find suitable header")?
+                    .to_str()?
+                    .to_string(),
+            )
+            .ok_or(TootubeError::new("Cannot create resumable upload!").into())
+    }
+
+    /// This takes the PT stream for download, connects it to YT stream for upload
+    pub async fn now_kiss(
+        &self,
+        dl_url: &str,
+        r_url: &str,
+        pg_conf: &TootubeConfig,
+    ) -> Result<String, Box<dyn Error>> {
+        // Get the upstream bytes stream
+        let res = reqwest::get(dl_url).await?;
+        let content_lengh = res
+            .content_length()
+            .ok_or(format!("Cannot get content length from {}", dl_url))?;
+        let mut stream = res.bytes_stream();
+
+        // Create the progress bar
+        let pb = ProgressBar::new(content_lengh);
+        pb.set_style(
+            ProgressStyle::default_bar()
+                .template(&pg_conf.progress_bar)?
+                .progress_chars(&pg_conf.progress_chars),
+        );
+        pb.set_message("Transferring…");
+        let mut transferring: u64 = 0;
+
+        // yields the stream chunk by chunk, updating the progress bar at the same time
+        let async_stream = stream! {
+            while let Some(chunk) = stream.next().await {
+                if let Ok(chunk) = &chunk {
+                    let new = min(transferring + (chunk.len() as u64), content_lengh);
+                    transferring = new;
+                    pb.set_position(new);
+                    if transferring >= content_lengh {
+                        pb.finish();
+                    }
+                }
+                yield chunk;
+            }
+        };
+
+        // Create client
+        let res = self
+            .client
+            .put(r_url)
+            .body(Body::wrap_stream(async_stream))
+            .send()
+            .await?;
+
+        if res.status().is_success() {
+            let yt_videos: YoutubeVideos = res.json().await?;
+            Ok(yt_videos.id)
+        } else {
+            Err(TootubeError::new(&format!("Cannot upload video: {:?}", res.text().await?)).into())
         }
     }
-
-    Ok(())
-}
-
-/// This function creates a resumable YT upload, putting all the parameters in
-pub async fn create_resumable_upload(
-    config: &YoutubeConfig,
-    vid: &PeerTubeVideo,
-) -> Result<String, Box<dyn Error>> {
-    let access_token = refresh_token(config).await?;
-
-    let upload_params = YoutubeUploadParams {
-        snippet: {
-            YoutubeUploadParamsSnippet {
-                title: vid.name.clone(),
-                description: vid.description.clone(),
-                tags: vid.tags.clone(),
-                ..Default::default()
-            }
-        },
-        status: {
-            YoutubeUploadParamsStatus {
-                ..Default::default()
-            }
-        },
-    };
-    debug!("YT upload params: {:?}", &upload_params);
-
-    let client = Client::new();
-    let res = client.post("https://www.googleapis.com/upload/youtube/v3/videos?uploadType=resumable&part=snippet%2Cstatus")
-        .header("Authorization", format!("Bearer {}", access_token))
-        .json(&upload_params)
-        .send().await?;
-
-    if res.status().is_success() {
-        Ok(res
-            .headers()
-            .get("location")
-            .ok_or("Cannot find suitable header")?
-            .to_str()?
-            .to_string())
-    } else {
-        Err(TootubeError::new("Cannot create resumable upload!").into())
-    }
-}
-
-/// This takes the PT stream for download, connects it to YT stream for upload
-pub async fn now_kiss<'a>(
-    stream: impl Stream<Item = Result<Bytes, reqwest::Error>>
-        + std::marker::Send
-        + std::marker::Sync
-        + 'a + 'static,
-    r_url: &'a str,
-    config: &'a YoutubeConfig,
-) -> Result<String, Box<dyn Error>> {
-    // Get access token
-    let access_token = refresh_token(config).await?;
-
-    // Create client
-    let client = Client::new();
-    let res = client
-        .put(r_url)
-        .header("Authorization", format!("Bearer {}", access_token))
-        .body(Body::wrap_stream(stream))
-        .send()
-        .await?;
-
-    if res.status().is_success() {
-        let yt_videos: YoutubeVideos = res.json().await?;
-        Ok(yt_videos.id)
-    } else {
-        Err(TootubeError::new(&format!("Cannot upload video: {:?}", res.text().await?)).into())
-    }
 }