Merge branch '6-refactor-peertube-and-youtube-to-impl-struct-instead-of-functional-style' into 'master'

♻️: refactor src/peertube.rs code to be more efficient regarding reqwest management

Closes #6

See merge request veretcle/tootube!20
This commit is contained in:
VC
2024-05-14 18:11:27 +00:00
5 changed files with 370 additions and 344 deletions

2
Cargo.lock generated
View File

@@ -1192,7 +1192,7 @@ dependencies = [
[[package]] [[package]]
name = "tootube" name = "tootube"
version = "0.6.0" version = "0.6.1"
dependencies = [ dependencies = [
"async-stream", "async-stream",
"clap", "clap",

View File

@@ -1,7 +1,7 @@
[package] [package]
name = "tootube" name = "tootube"
authors = ["VC <veretcle+framagit@mateu.be>"] authors = ["VC <veretcle+framagit@mateu.be>"]
version = "0.6.0" version = "0.6.1"
edition = "2021" edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

View File

@@ -8,24 +8,31 @@ use config::Config;
mod peertube; mod peertube;
pub use peertube::register as register_peertube; pub use peertube::register as register_peertube;
use peertube::{delete_original_video_source, get_latest_video, get_original_video_source}; use peertube::PeerTube;
mod youtube; mod youtube;
pub use youtube::register as register_youtube; pub use youtube::register as register_youtube;
use youtube::{add_video_to_playlists, create_resumable_upload, now_kiss}; use youtube::YouTube;
#[tokio::main] #[tokio::main]
pub async fn run(config: Config, pl: Vec<String>) { pub async fn run(config: Config, pl: Vec<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 {
Some(s) => PeerTube::new(&config.peertube.base_url)
.with_client(&s.client_id, &s.client_secret, &s.refresh_token)
.await .await
.unwrap_or_else(|e| { .unwrap_or_else(|e| panic!("Cannot instantiate PeerTube struct: {}", e)),
None => PeerTube::new(&config.peertube.base_url),
};
// Get the latest video object
let latest_vid = peertube.get_latest_video().await.unwrap_or_else(|e| {
panic!("Cannot retrieve the latest video, something must have gone terribly wrong: {e}") panic!("Cannot retrieve the latest video, something must have gone terribly wrong: {e}")
}); });
// We have a refresh_token, try to use it // We have a refresh_token, try to use it
let source_url = match &config.peertube.oauth2 { let source_url = match &config.peertube.oauth2 {
Some(_) => get_original_video_source(&latest_vid.uuid, &config.peertube) Some(_) => peertube
.get_original_video_source(&latest_vid.uuid)
.await .await
.ok(), .ok(),
None => None, None => None,
@@ -48,30 +55,37 @@ pub async fn run(config: Config, pl: Vec<String>) {
debug!("PT download URL: {}", &dl_url); debug!("PT download URL: {}", &dl_url);
let resumable_upload_url = create_resumable_upload(&config.youtube, &latest_vid) let youtube = YouTube::new(
&config.youtube.client_id,
&config.youtube.client_secret,
&config.youtube.refresh_token,
)
.await
.unwrap_or_else(|e| panic!("Cannot instantiate YouTube struct: {}", e));
let resumable_upload_url = youtube
.create_resumable_upload(&latest_vid)
.await .await
.unwrap_or_else(|e| panic!("Cannot retrieve the uploads resumable id: {e}")); .unwrap_or_else(|e| panic!("Cannot retrieve the uploads resumable id: {e}"));
debug!("YT upload URL: {}", &resumable_upload_url); debug!("YT upload URL: {}", &resumable_upload_url);
let yt_video_id = now_kiss( let yt_video_id = youtube
&dl_url, .now_kiss(&dl_url, &resumable_upload_url, &config.tootube)
&resumable_upload_url,
&config.youtube,
&config.tootube,
)
.await .await
.unwrap_or_else(|e| panic!("Cannot resume upload!: {e}")); .unwrap_or_else(|e| panic!("Cannot resume upload!: {e}"));
debug!("YT video ID: {}", &yt_video_id); debug!("YT video ID: {}", &yt_video_id);
if !pl.is_empty() { if !pl.is_empty() {
add_video_to_playlists(&config.youtube, &yt_video_id, &pl) youtube
.add_video_to_playlists(&yt_video_id, &pl)
.await .await
.unwrap_or_else(|e| panic!("Cannot add video to playlist(s): {e}")); .unwrap_or_else(|e| panic!("Cannot add video to playlist(s): {e}"));
} }
// delete the source video if requested (it wont be used anymore) // delete the source video if requested (it wont be used anymore)
if config.peertube.delete_video_source_after_transfer { if config.peertube.delete_video_source_after_transfer {
delete_original_video_source(&latest_vid.uuid, &config.peertube) peertube
.delete_original_video_source(&latest_vid.uuid)
.await .await
.unwrap_or_else(|e| panic!("Cannot delete source video: {e}")); .unwrap_or_else(|e| panic!("Cannot delete source video: {e}"));

View File

@@ -1,6 +1,9 @@
use crate::{config::PeertubeConfig, error::TootubeError}; use crate::{config::PeertubeConfig, error::TootubeError};
use log::debug; use log::debug;
use reqwest::Client; use reqwest::{
header::{HeaderMap, HeaderValue},
Client,
};
use rpassword::prompt_password; use rpassword::prompt_password;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::{boxed::Box, cmp::Ordering, error::Error, io::stdin}; use std::{boxed::Box, cmp::Ordering, error::Error, io::stdin};
@@ -96,28 +99,6 @@ pub struct PeerTubeVideoStreamingPlaylistsFilesResolution {
pub id: u16, pub id: u16,
} }
/// This gets the last video uploaded to the PeerTube server
pub async fn get_latest_video(u: &str) -> Result<PeerTubeVideo, Box<dyn Error>> {
let body = reqwest::get(format!("{}/api/v1/videos?count=1&sort=-publishedAt", u))
.await?
.json::<PeerTubeVideos>()
.await?;
let vid = get_video_detail(u, &body.data[0].uuid).await?;
Ok(vid)
}
/// This gets all the crispy details about one particular video
async fn get_video_detail(u: &str, v: &str) -> Result<PeerTubeVideo, Box<dyn Error>> {
let body = reqwest::get(format!("{}/api/v1/videos/{}", u, v))
.await?
.json::<PeerTubeVideo>()
.await?;
Ok(body)
}
/// This function makes the registration process a little bit easier /// This function makes the registration process a little bit easier
#[tokio::main] #[tokio::main]
pub async fn register(config: &PeertubeConfig) -> Result<(), Box<dyn Error>> { pub async fn register(config: &PeertubeConfig) -> Result<(), Box<dyn Error>> {
@@ -173,59 +154,100 @@ pub async fn register(config: &PeertubeConfig) -> Result<(), Box<dyn Error>> {
Ok(()) Ok(())
} }
async fn get_refresh_token(config: &PeertubeConfig) -> Result<String, Box<dyn Error>> { #[derive(Debug)]
// unwrap is safe here, this function is only called when oauth2 is present pub struct PeerTube {
let oauth2_config = config.oauth2.as_ref().unwrap().clone(); base_url: String,
// retrieve the refresh_token from the file client: Client,
let refresh_token = read_to_string(&oauth2_config.refresh_token).await?; }
debug!( impl PeerTube {
"Opened file {} to retrieve Token", /// Create a new PeerTube struct with a basic embedded reqwest::Client
&oauth2_config.refresh_token pub fn new(base_url: &str) -> Self {
); PeerTube {
base_url: format!("{}/api/v1", base_url),
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,
client_id: &str,
client_secret: &str,
refresh_token_path: &str,
) -> Result<Self, Box<dyn Error>> {
let refresh_token = read_to_string(refresh_token_path).await?;
let params = PeerTubeUsersToken { let params = PeerTubeUsersToken {
client_id: oauth2_config.client_id.clone(), client_id: client_id.to_string(),
client_secret: oauth2_config.client_secret.clone(), client_secret: client_secret.to_string(),
grant_type: "refresh_token".to_string(), grant_type: "refresh_token".to_string(),
refresh_token: Some(refresh_token), refresh_token: Some(refresh_token),
username: None, username: None,
password: None, password: None,
}; };
let client = Client::new(); let req = self
let oauth2_token = client .client
.post(format!("{}/api/v1/users/token", config.base_url)) .post(&format!("{}/users/token", self.base_url))
.form(&params) .form(&params)
.send() .send()
.await? .await?
.json::<PeerTubeUsersTokenResponse>() .json::<PeerTubeUsersTokenResponse>()
.await?; .await?;
debug!("Retrieved access_token: {}", &oauth2_token.access_token); write(refresh_token_path, req.refresh_token).await?;
// write the new refresh token to the file let mut headers = HeaderMap::new();
write(&oauth2_config.refresh_token, oauth2_token.refresh_token).await?; headers.insert(
"Authorization",
debug!( HeaderValue::from_str(&format!("Bearer {}", req.access_token))?,
"Written refresh_token to file {}",
&oauth2_config.refresh_token
); );
Ok(oauth2_token.access_token) self.client = reqwest::Client::builder()
} .default_headers(headers)
.build()?;
pub async fn get_original_video_source( Ok(self)
uuid: &str, }
config: &PeertubeConfig,
) -> Result<String, Box<dyn Error>> {
let access_token = get_refresh_token(config).await?;
let client = Client::new(); /// 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 source_vid = client let vid = self.get_video_detail(&body.data[0].uuid).await?;
.get(format!("{}/api/v1/videos/{}/source", config.base_url, uuid))
.header("Authorization", format!("Bearer {}", &access_token)) Ok(vid)
}
/// This gets all the crispy details about one particular video
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() .send()
.await? .await?
.json::<PeerTubeVideoSourceResponse>() .json::<PeerTubeVideoSourceResponse>()
@@ -233,9 +255,9 @@ pub async fn get_original_video_source(
debug!("Got the Source Vid URL: {}", &source_vid.file_download_url); debug!("Got the Source Vid URL: {}", &source_vid.file_download_url);
let video_file_token = client let video_file_token = self
.post(format!("{}/api/v1/videos/{}/token", config.base_url, uuid)) .client
.header("Authorization", format!("Bearer {}", &access_token)) .post(format!("{}/videos/{}/token", self.base_url, uuid))
.send() .send()
.await? .await?
.json::<PeerTubeVideoTokenResponse>() .json::<PeerTubeVideoTokenResponse>()
@@ -247,22 +269,13 @@ pub async fn get_original_video_source(
"{}?videoFileToken={}", "{}?videoFileToken={}",
source_vid.file_download_url, video_file_token.files.token source_vid.file_download_url, video_file_token.files.token
)) ))
} }
pub async fn delete_original_video_source( /// Delete the original video source
uuid: &str, pub async fn delete_original_video_source(&self, uuid: &str) -> Result<(), Box<dyn Error>> {
config: &PeertubeConfig, let res = self
) -> Result<(), Box<dyn Error>> { .client
let access_token = get_refresh_token(config).await?; .delete(format!("{}/videos/{}/source/file", self.base_url, uuid))
let client = Client::new();
let res = client
.delete(format!(
"{}/api/v1/videos/{}/source/file",
config.base_url, uuid
))
.header("Authorization", format!("Bearer {}", &access_token))
.send() .send()
.await?; .await?;
@@ -276,4 +289,5 @@ pub async fn delete_original_video_source(
} }
Ok(()) Ok(())
}
} }

View File

@@ -7,12 +7,13 @@ use async_stream::stream;
use futures_util::StreamExt; use futures_util::StreamExt;
use indicatif::{ProgressBar, ProgressStyle}; use indicatif::{ProgressBar, ProgressStyle};
use log::{debug, warn}; use log::{debug, warn};
use reqwest::{multipart::Form, Body, Client}; use reqwest::{
header::{HeaderMap, HeaderValue},
multipart::Form,
Body, Client,
};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::{cmp::min, 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)] #[derive(Serialize, Debug)]
struct RefreshTokenRequest { struct RefreshTokenRequest {
@@ -193,50 +194,58 @@ pub async fn register(config: &YoutubeConfig) -> Result<(), Box<dyn Error>> {
Ok(()) Ok(())
} }
/// 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 {
impl YouTube {
pub async fn new(
client_id: &str,
client_secret: &str,
refresh_token: &str,
) -> Result<Self, Box<dyn Error>> {
let mut youtube = YouTube {
client: Client::new(),
};
let refresh_token = RefreshTokenRequest { let refresh_token = RefreshTokenRequest {
refresh_token: config.refresh_token.clone(), refresh_token: refresh_token.to_string(),
client_id: config.client_id.clone(), client_id: client_id.to_string(),
client_secret: config.client_secret.clone(), client_secret: client_secret.to_string(),
..Default::default() ..Default::default()
}; };
let client = Client::new(); let access_token = youtube
let res = client .client
.post("https://accounts.google.com/o/oauth2/token") .post("https://accounts.google.com/o/oauth2/token")
.json(&refresh_token) .json(&refresh_token)
.send() .send()
.await?
.json::<AccessTokenResponse>()
.await?; .await?;
let access_token: AccessTokenResponse = res.json().await?; let mut headers = HeaderMap::new();
headers.insert(
"Authorization",
HeaderValue::from_str(&format!("Bearer {}", access_token.access_token))?,
);
debug!("YT Access Token: {}", &access_token.access_token); youtube.client = reqwest::Client::builder()
Ok(access_token.access_token) .default_headers(headers)
}) .build()?;
.await
.cloned()
}
/// This function takes a list of playlists keyword and returns a list of playlist ID Ok(youtube)
async fn get_playlist_ids( }
config: &YoutubeConfig, /// This function takes a list of playlists keyword and returns a list of playlist ID
pl: &[String], async fn get_playlist_ids(&self, pl: &[String]) -> Result<Vec<String>, Box<dyn Error>> {
) -> Result<Vec<String>, Box<dyn Error>> {
let mut page_token = String::new(); let mut page_token = String::new();
let mut playlists: Vec<String> = vec![]; let mut playlists: Vec<String> = vec![];
let access_token = refresh_token(config).await?; while let Ok(local_pl) = self.client
let client = Client::new();
while let Ok(local_pl) = client
.get(&format!( .get(&format!(
"https://www.googleapis.com/youtube/v3/playlists?part=snippet&mine=true&pageToken={}", "https://www.googleapis.com/youtube/v3/playlists?part=snippet&mine=true&pageToken={}",
page_token page_token
)) ))
.header("Authorization", format!("Bearer {}", access_token))
.send() .send()
.await? .await?
.json::<YoutubePlaylistListResponse>() .json::<YoutubePlaylistListResponse>()
@@ -249,7 +258,6 @@ async fn get_playlist_ids(
.filter_map(|s| pl.contains(&s.snippet.title).then_some(s.id.clone())) .filter_map(|s| pl.contains(&s.snippet.title).then_some(s.id.clone()))
.collect(), .collect(),
); );
// if nextPageToken is present, continue the loop // if nextPageToken is present, continue the loop
match local_pl.next_page_token { match local_pl.next_page_token {
None => break, None => break,
@@ -259,16 +267,15 @@ async fn get_playlist_ids(
debug!("Playlists IDs: {:?}", &playlists); debug!("Playlists IDs: {:?}", &playlists);
Ok(playlists) Ok(playlists)
} }
/// This function adds the video id to the corresponding named playlist(s) /// This function adds the video id to the corresponding named playlist(s)
pub async fn add_video_to_playlists( pub async fn add_video_to_playlists(
config: &YoutubeConfig, &self,
v: &str, v: &str,
pl: &[String], pl: &[String],
) -> Result<(), Box<dyn Error>> { ) -> Result<(), Box<dyn Error>> {
let access_token = refresh_token(config).await?; let playlists_ids = self.get_playlist_ids(pl).await?;
let playlists_ids = get_playlist_ids(config, pl).await?;
for pl_id in playlists_ids { for pl_id in playlists_ids {
let yt_pl_upload_params = YoutubePlaylistItemsParams { let yt_pl_upload_params = YoutubePlaylistItemsParams {
@@ -282,10 +289,9 @@ pub async fn add_video_to_playlists(
}, },
}; };
let client = Client::new(); let res = self
let res = client .client
.post("https://youtube.googleapis.com/youtube/v3/playlistItems?part=snippet") .post("https://youtube.googleapis.com/youtube/v3/playlistItems?part=snippet")
.header("Authorization", format!("Bearer {}", access_token))
.json(&yt_pl_upload_params) .json(&yt_pl_upload_params)
.send() .send()
.await?; .await?;
@@ -300,15 +306,13 @@ pub async fn add_video_to_playlists(
} }
Ok(()) Ok(())
} }
/// This function creates a resumable YT upload, putting all the parameters in /// This function creates a resumable YT upload, putting all the parameters in
pub async fn create_resumable_upload( pub async fn create_resumable_upload(
config: &YoutubeConfig, &self,
vid: &PeerTubeVideo, vid: &PeerTubeVideo,
) -> Result<String, Box<dyn Error>> { ) -> Result<String, Box<dyn Error>> {
let access_token = refresh_token(config).await?;
if vid.name.chars().count() > 100 { if vid.name.chars().count() > 100 {
warn!( warn!(
"PT Video Title ({}) is too long, it will be truncated", "PT Video Title ({}) is too long, it will be truncated",
@@ -333,9 +337,7 @@ pub async fn create_resumable_upload(
}; };
debug!("YT upload params: {:?}", &upload_params); debug!("YT upload params: {:?}", &upload_params);
let client = Client::new(); let res = self.client.post("https://www.googleapis.com/upload/youtube/v3/videos?uploadType=resumable&part=snippet%2Cstatus")
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) .json(&upload_params)
.send().await?; .send().await?;
@@ -349,18 +351,14 @@ pub async fn create_resumable_upload(
} else { } else {
Err(TootubeError::new("Cannot create resumable upload!").into()) Err(TootubeError::new("Cannot create resumable upload!").into())
} }
} }
/// This takes the PT stream for download, connects it to YT stream for upload
/// This takes the PT stream for download, connects it to YT stream for upload pub async fn now_kiss(
pub async fn now_kiss( &self,
dl_url: &str, dl_url: &str,
r_url: &str, r_url: &str,
config: &YoutubeConfig,
pg_conf: &TootubeConfig, pg_conf: &TootubeConfig,
) -> Result<String, Box<dyn Error>> { ) -> Result<String, Box<dyn Error>> {
// Get access token
let access_token = refresh_token(config).await?;
// Get the upstream bytes stream // Get the upstream bytes stream
let res = reqwest::get(dl_url).await?; let res = reqwest::get(dl_url).await?;
let content_lengh = res let content_lengh = res
@@ -394,10 +392,9 @@ pub async fn now_kiss(
}; };
// Create client // Create client
let client = Client::new(); let res = self
let res = client .client
.put(r_url) .put(r_url)
.header("Authorization", format!("Bearer {}", access_token))
.body(Body::wrap_stream(async_stream)) .body(Body::wrap_stream(async_stream))
.send() .send()
.await?; .await?;
@@ -408,4 +405,5 @@ pub async fn now_kiss(
} else { } else {
Err(TootubeError::new(&format!("Cannot upload video: {:?}", res.text().await?)).into()) Err(TootubeError::new(&format!("Cannot upload video: {:?}", res.text().await?)).into())
} }
}
} }