06.03.2024

* FFmpeg feature implemented (Only non-live videos)

Author: Mithronn

Cargo.toml

@@ -49,15 +49,17 @@ hex = "0.4.3"
 unicode-segmentation = "1.11.0"
 boa_engine = "0.17.3"
 mime = "0.3.17"
+bytes = "1.5.0"
 
 [dev-dependencies]
 tokio = { version = "1.36.0", features = ["full"] }
 
 [features]
-default = ["search", "live", "default-tls"]
+default = ["search", "live", "default-tls", "ffmpeg"]
 live = ["tokio/time", "tokio/process"]
 blocking = ["tokio/rt", "tokio/rt-multi-thread"]
 search = []
+ffmpeg = ["tokio/process"]
 default-tls = ["reqwest/default-tls"]
 native-tls = ["reqwest/native-tls"]
 rustls-tls = ["reqwest/rustls-tls"]

src/blocking/info.rs

@@ -10,6 +10,9 @@ use crate::Video as AsyncVideo;
 use super::stream::LiveStreamOptions;
 use super::stream::{NonLiveStreamOptions, Stream};
 
+#[cfg(feature = "ffmpeg")]
+use crate::structs::FFmpegArgs;
+
 #[derive(Clone, Debug, derive_more::Display, PartialEq, Eq)]
 pub struct Video(AsyncVideo);
 
@@ -53,7 +56,10 @@ impl Video {
     ///           println!("{:#?}", chunk);
     ///     }
     /// ```
-    pub fn stream(&self) -> Result<Box<dyn Stream + Send + Sync>, VideoError> {
+    pub fn stream(
+        &self,
+        #[cfg(feature = "ffmpeg")] ffmpeg_args: Option<FFmpegArgs>,
+    ) -> Result<Box<dyn Stream + Send + Sync>, VideoError> {
         let client = self.0.get_client();
 
         let options = self.0.get_options();
@@ -124,14 +130,28 @@ impl Video {
             dl_chunk_size,
             start,
             end,
+            #[cfg(feature = "ffmpeg")]
+            ffmpeg_args,
         })?;
 
         Ok(Box::new(stream))
     }
 
     /// Download video directly to the file
-    pub fn download<P: AsRef<std::path::Path>>(&self, path: P) -> Result<(), VideoError> {
-        Ok(block_async!(self.0.download(path))?)
+    pub fn download<P: AsRef<std::path::Path>>(
+        &self,
+        path: P,
+        #[cfg(feature = "ffmpeg")] ffmpeg_args: Option<FFmpegArgs>,
+    ) -> Result<(), VideoError> {
+        #[cfg(feature = "ffmpeg")]
+        {
+            Ok(block_async!(self.0.download(path, ffmpeg_args))?)
+        }
+
+        #[cfg(not(feature = "ffmpeg"))]
+        {
+            Ok(block_async!(self.0.download(path))?)
+        }
     }
 
     /// Get video URL

src/info.rs

@@ -15,6 +15,9 @@ use crate::stream::{NonLiveStream, NonLiveStreamOptions, Stream};
 
 use crate::structs::{VideoError, VideoFormat, VideoInfo, VideoOptions};
 
+#[cfg(feature = "ffmpeg")]
+use crate::structs::FFmpegArgs;
+
 use crate::utils::{
     add_format_meta, between, choose_format, clean_video_details, get_functions, get_html,
     get_html5player, get_random_v6_ip, get_video_id, is_not_yet_broadcasted, is_play_error,
@@ -330,7 +333,10 @@ impl Video {
     ///           println!("{:#?}", chunk);
     ///     }
     /// ```
-    pub async fn stream(&self) -> Result<Box<dyn Stream + Send + Sync>, VideoError> {
+    pub async fn stream(
+        &self,
+        #[cfg(feature = "ffmpeg")] ffmpeg_args: Option<FFmpegArgs>,
+    ) -> Result<Box<dyn Stream + Send + Sync>, VideoError> {
         let client = &self.client;
 
         let info = self.get_info().await?;
@@ -364,11 +370,12 @@ impl Video {
             }
         }
 
-        let dl_chunk_size = if self.options.download_options.dl_chunk_size.is_some() {
-            self.options.download_options.dl_chunk_size.unwrap()
-        } else {
-            1024 * 1024 * 10_u64 // -> Default is 10MB to avoid Youtube throttle (Bigger than this value can be throttle by Youtube)
-        };
+        let dl_chunk_size = self
+            .options
+            .download_options
+            .dl_chunk_size
+            // 1024 * 1024 * 10_u64 -> Default is 10MB to avoid Youtube throttle (Bigger than this value can be throttle by Youtube)
+            .unwrap_or(1024 * 1024 * 10_u64);
 
         let start = 0;
         let end = start + dl_chunk_size;
@@ -402,15 +409,32 @@ impl Video {
             dl_chunk_size,
             start,
             end,
+            #[cfg(feature = "ffmpeg")]
+            ffmpeg_args,
         })?;
 
         Ok(Box::new(stream))
     }
 
     /// Download video directly to the file
-    pub async fn download<P: AsRef<std::path::Path>>(&self, path: P) -> Result<(), VideoError> {
+    pub async fn download<P: AsRef<std::path::Path>>(
+        &self,
+        path: P,
+        #[cfg(feature = "ffmpeg")] ffmpeg_args: Option<FFmpegArgs>,
+    ) -> Result<(), VideoError> {
         use std::io::Write;
-        let stream = self.stream().await.unwrap();
+
+        let stream: Box<dyn Stream + Send + Sync>;
+
+        #[cfg(not(feature = "ffmpeg"))]
+        {
+            stream = self.stream().await.unwrap();
+        }
+
+        #[cfg(feature = "ffmpeg")]
+        {
+            stream = self.stream(ffmpeg_args).await.unwrap();
+        }
 
         let mut file =
             std::fs::File::create(path).map_err(|e| VideoError::DownloadError(e.to_string()))?;

src/lib.rs

@@ -23,6 +23,10 @@ pub use structs::{
     RequestOptions, StoryBoard, Thumbnail, VideoDetails, VideoError, VideoFormat, VideoInfo,
     VideoOptions, VideoQuality, VideoSearchOptions,
 };
+
+#[cfg(feature = "ffmpeg")]
+pub use structs::FFmpegArgs;
+
 pub use utils::{choose_format, get_random_v6_ip, get_video_id};
 // export to access proxy feature
 pub use reqwest;

src/stream/streams/non_live.rs

@@ -1,8 +1,12 @@
+use async_trait::async_trait;
+use tokio::sync::RwLock;
+
 use crate::constants::DEFAULT_HEADERS;
 use crate::stream::streams::Stream;
 use crate::structs::VideoError;
-use async_trait::async_trait;
-use tokio::sync::RwLock;
+
+#[cfg(feature = "ffmpeg")]
+use crate::{structs::FFmpegArgs, utils::ffmpeg_cmd_run};
 
 pub struct NonLiveStreamOptions {
     pub client: Option<reqwest_middleware::ClientWithMiddleware>,
@@ -11,6 +15,9 @@ pub struct NonLiveStreamOptions {
     pub dl_chunk_size: u64,
     pub start: u64,
     pub end: u64,
+
+    #[cfg(feature = "ffmpeg")]
+    pub ffmpeg_args: Option<FFmpegArgs>,
 }
 
 pub struct NonLiveStream {
@@ -21,6 +28,13 @@ pub struct NonLiveStream {
     end: RwLock<u64>,
 
     client: reqwest_middleware::ClientWithMiddleware,
+
+    #[cfg(feature = "ffmpeg")]
+    ffmpeg_args: Option<FFmpegArgs>,
+    #[cfg(feature = "ffmpeg")]
+    ffmpeg_start_byte: RwLock<Vec<u8>>,
+    #[cfg(feature = "ffmpeg")]
+    ffmpeg_end_byte: RwLock<usize>,
 }
 
 impl NonLiveStream {
@@ -52,6 +66,12 @@ impl NonLiveStream {
             dl_chunk_size: options.dl_chunk_size,
             start: RwLock::new(options.start),
             end: RwLock::new(options.end),
+            #[cfg(feature = "ffmpeg")]
+            ffmpeg_args: options.ffmpeg_args,
+            #[cfg(feature = "ffmpeg")]
+            ffmpeg_end_byte: RwLock::new(0),
+            #[cfg(feature = "ffmpeg")]
+            ffmpeg_start_byte: RwLock::new(vec![]),
         })
     }
 
@@ -66,6 +86,16 @@ impl NonLiveStream {
     async fn start_index(&self) -> u64 {
         *self.start.read().await
     }
+
+    #[cfg(feature = "ffmpeg")]
+    async fn ffmpeg_end_byte_index(&self) -> usize {
+        *self.ffmpeg_end_byte.read().await
+    }
+
+    #[cfg(feature = "ffmpeg")]
+    async fn ffmpeg_start_byte_index(&self) -> Vec<u8> {
+        self.ffmpeg_start_byte.read().await.to_vec()
+    }
 }
 
 #[async_trait]
@@ -100,13 +130,13 @@ impl Stream for NonLiveStream {
                 .unwrap(),
         );
 
-        let response = self.client.get(&self.link).headers(headers).send().await;
-
-        if response.is_err() {
-            return Err(VideoError::ReqwestMiddleware(response.err().unwrap()));
-        }
-
-        let mut response = response.expect("IMPOSSIBLE");
+        let mut response = self
+            .client
+            .get(&self.link)
+            .headers(headers)
+            .send()
+            .await
+            .map_err(VideoError::ReqwestMiddleware)?;
 
         let mut buf: Vec<u8> = vec![];
 
@@ -115,6 +145,39 @@ impl Stream for NonLiveStream {
             buf.extend(chunk.iter());
         }
 
+        #[cfg(feature = "ffmpeg")]
+        {
+            let ffmpeg_args = self
+                .ffmpeg_args
+                .clone()
+                .map(|x| x.build())
+                .unwrap_or_default();
+
+            if !ffmpeg_args.is_empty() {
+                let ffmpeg_start_byte_index = self.ffmpeg_start_byte_index().await;
+
+                let cmd_output = ffmpeg_cmd_run(
+                    &ffmpeg_args,
+                    &[&ffmpeg_start_byte_index, buf.as_slice()].concat(),
+                )
+                .await?;
+
+                let end_index = self.ffmpeg_end_byte_index().await;
+
+                let mut first_buffer_trim = 1;
+                if ffmpeg_start_byte_index.is_empty() {
+                    let mut start_byte = self.ffmpeg_start_byte.write().await;
+                    *start_byte = buf.clone();
+                    let mut end_byte = self.ffmpeg_end_byte.write().await;
+                    *end_byte = cmd_output.len();
+
+                    first_buffer_trim = 0;
+                }
+
+                buf = (cmd_output[(end_index + first_buffer_trim)..]).to_vec();
+            }
+        }
+
         if end != 0 {
             let mut start = self.start.write().await;
             *start = end + 1;

src/structs.rs

@@ -259,6 +259,10 @@ pub enum VideoError {
     /// Downloading live streams not supported, compile with `live` feature to enable
     #[error("Downloading live streams not supported, compile with `live` feature to enable")]
     LiveStreamNotSupported,
+    /// FFmpeg command error
+    #[error("FFmpeg command error: {0}")]
+    #[cfg(feature = "ffmpeg")]
+    FFmpeg(String),
 }
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@@ -650,3 +654,55 @@ impl<'de> Deserialize<'de> for MimeType {
         })
     }
 }
+
+#[cfg(feature = "ffmpeg")]
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct FFmpegArgs {
+    pub format: Option<String>,
+    pub audio_filter: Option<String>,
+    pub video_filter: Option<String>,
+}
+
+#[cfg(feature = "ffmpeg")]
+impl FFmpegArgs {
+    pub fn build(&self) -> Vec<String> {
+        let mut args: Vec<String> = vec![];
+
+        if let Some(format) = &self.format {
+            args.push("-f".to_string());
+            args.push(format.to_string());
+        }
+
+        if let Some(audio_filter) = &self.audio_filter {
+            args.push("-af".to_string());
+            args.push(audio_filter.to_string());
+        }
+
+        if let Some(video_filter) = &self.video_filter {
+            args.push("-vf".to_string());
+            args.push(video_filter.to_string());
+        }
+
+        if self.format.is_some() || self.audio_filter.is_some() || self.video_filter.is_some() {
+            args = [
+                vec![
+                    // input as stdin
+                    "-i".to_string(),
+                    // aliases of pipe:0
+                    "-".to_string(),
+                    // "-analyzeduration".to_string(),
+                    // "0".to_string(),
+                    // "-loglevel".to_string(),
+                    // "0".to_string(),
+                ],
+                args,
+            ]
+            .concat();
+
+            // pipe to stdout
+            args.push("pipe:1".to_string());
+        }
+
+        args
+    }
+}

src/utils.rs

@@ -2,6 +2,8 @@ use rand::Rng;
 use regex::Regex;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
+use std::process::Stdio;
+use tokio::{io::AsyncWriteExt, process::Command};
 use unicode_segmentation::UnicodeSegmentation;
 use urlencoding::decode;
 
@@ -15,6 +17,30 @@ use crate::structs::{
     VideoOptions, VideoQuality, VideoSearchOptions,
 };
 
+#[cfg(feature = "ffmpeg")]
+pub async fn ffmpeg_cmd_run(args: &Vec<String>, data: &[u8]) -> Result<Vec<u8>, VideoError> {
+    let mut cmd = Command::new("ffmpeg");
+    cmd.args(args)
+        .stdin(Stdio::piped())
+        .stdout(Stdio::piped())
+        .kill_on_drop(true);
+
+    let mut process = cmd.spawn().map_err(|x| VideoError::FFmpeg(x.to_string()))?;
+    let mut stdin = process
+        .stdin
+        .take()
+        .ok_or(VideoError::FFmpeg("Failed to open stdin".to_string()))?;
+    let cloned_data = data.to_owned();
+    tokio::spawn(async move { stdin.write_all(&cloned_data).await });
+
+    let output = process
+        .wait_with_output()
+        .await
+        .map_err(|x| VideoError::FFmpeg(x.to_string()))?;
+
+    Ok(output.stdout)
+}
+
 #[allow(dead_code)]
 pub fn get_cver(info: &serde_json::Value) -> &str {
     info.get("responseContext")