feat: Extract mp3 and opus audio with ffmpeg

src/main.rs

@@ -105,8 +105,16 @@ async fn process_simple_embeds(page_body: &str, referer: &str, state: Arc<State>
                             async {
                                 let video = video.clone();
                                 tokio::spawn(async move {
-                                    info!("Download simple embed '{}'...", video.url());
-                                    video.download().await?;
+                                    let url = video.url();
+                                    info!("Download simple embed '{url}'...", );
+                                    video.clone().download().await?;
+
+                                    info!("Extract opus audio for simple embed '{url}'...");
+                                    video.clone().extract_audio("opus").await?;
+
+                                    info!("Extract mp3 audio for simple embed '{url}'...");
+                                    video.extract_audio("mp3").await?;
+
                                     Ok::<(), Report>(())
                                 })
                                 .await?
@@ -254,15 +262,23 @@ async fn process_showcase_clip(clip: &Value, referer: &str, state: Arc<State>) -
             debug!("embed_url_match: {embed_url_match:#?}");
 
             let embed_url = html_escape::decode_html_entities(embed_url_match.as_str());
-            info!("Download showcase clip '{embed_url}'...");
 
             let video = Arc::new(Video::new_with_title(
-                embed_url,
+                &*embed_url,
                 referer,
                 config.dot_get::<String>("video.title")?, // maybe_title
             ));
             (*state).push_video(video.clone()).await;
-            video.download().await?;
+
+            info!("Download showcase clip '{embed_url}'...");
+            video.clone().download().await?;
+
+            info!("Extract opus audio for showcase clip '{embed_url}'...");
+            video.clone().extract_audio("opus").await?;
+
+
+            info!("Extract mp3 audio for showcase clip '{embed_url}'...");
+            video.extract_audio("mp3").await?;
         }
         None => {
             bail!("Could not extract embed URL from config 'video.embed_code' string (embed_url not captured)");

src/state/video/mod.rs

@@ -1,16 +1,20 @@
 use color_eyre::{
-    eyre::{eyre, Result, WrapErr},
+    eyre::{Result, WrapErr},
     Report,
 };
 use lazy_static::lazy_static;
 use regex::Regex;
-use std::{process::Stdio, sync::Arc};
+use std::{path::Path, process::Stdio, sync::Arc};
 use tokio::{
-    io::{AsyncBufReadExt, BufReader},
+    io::{AsyncBufReadExt, AsyncRead, BufReader},
+    process::{Child, Command},
     sync::{RwLock, RwLockReadGuard},
+    task::JoinHandle,
 };
 use tracing::debug;
 
+use crate::util::maybe_join;
+
 use self::progress::ProgressDetail;
 
 pub mod progress;
@@ -27,6 +31,7 @@ pub struct Video {
 pub enum Stage {
     Initializing,
     Downloading,
+    ExtractingAudio,
     Finished,
 }
 
@@ -64,6 +69,10 @@ impl Video {
         *self.stage.write().await = Stage::Downloading;
     }
 
+    pub async fn set_stage_extracting_audio(&self) {
+        *self.stage.write().await = Stage::ExtractingAudio;
+    }
+
     pub async fn set_stage_finished(&self) {
         *self.stage.write().await = Stage::Finished;
     }
@@ -175,34 +184,113 @@ impl Video {
     pub async fn download(self: Arc<Self>) -> Result<()> {
         self.set_stage_downloading().await;
 
-        debug!(
-            "Spawn: yt-dlp --newline --no-colors --referer '{}' '{}'",
+        let cmd = format!(
+            "yt-dlp --newline --no-colors --referer '{}' '{}'",
             &self.referer,
             self.url()
         );
-        let mut child = tokio::process::Command::new("yt-dlp")
-            .kill_on_drop(true)
-            .stdout(Stdio::piped())
-            .arg("--newline")
-            .arg("--no-colors")
-            .arg("--referer")
-            .arg(&self.referer)
-            .arg(self.url())
-            .spawn()
-            .wrap_err("yt-dlp command failed to start")?;
-
-        let stdout = child.stdout.take().ok_or_else(|| {
-            eyre!(
-                "Child's stdout was None (yt-dlp --newline --no-colors --referer '{}' '{}')",
-                &self.referer,
-                self.url()
+
+        debug!("Spawn: {cmd}");
+        self.clone()
+            .child_read_to_end(
+                Command::new("yt-dlp")
+                    .kill_on_drop(true)
+                    .stdout(Stdio::piped())
+                    .stderr(Stdio::piped())
+                    .arg("--newline")
+                    .arg("--no-colors")
+                    .arg("--referer")
+                    .arg(&self.referer)
+                    .arg(self.url())
+                    .spawn()
+                    .wrap_err_with(|| "Command failed to start ({cmd})")?,
             )
-        })?;
+            .await?;
+
+        self.set_stage_finished().await;
+
+        Ok(())
+    }
+
+    pub async fn extract_audio(self: Arc<Self>, format: &str) -> Result<()> {
+        if let Some(ref output_file) = *self.output_file().await {
+            self.set_stage_extracting_audio().await;
+
+            let source = Path::new(output_file);
+            let destination = Path::new(output_file).with_extension(format);
+
+            let cmd = format!(
+                "ffmpeg -y -i '{}' '{}'",
+                source.to_string_lossy(),
+                destination.to_string_lossy()
+            );
+
+            debug!("Spawn: {cmd}");
+            self.clone()
+                .child_read_to_end(
+                    Command::new("ffmpeg")
+                        .kill_on_drop(true)
+                        .stdout(Stdio::piped())
+                        .stderr(Stdio::piped())
+                        .arg("-i")
+                        .arg(&source)
+                        .arg(&destination)
+                        .spawn()
+                        .wrap_err_with(|| "Command failed to start ({cmd})")?,
+                )
+                .await?;
+
+            self.set_stage_finished().await;
+
+            // TODO: Set stage finished
+        }
+
+        Ok(())
+    }
+
+    async fn child_read_to_end(self: Arc<Self>, mut child: Child) -> Result<()> {
+        let consume_stdout = child
+            .stdout
+            .take()
+            .map(|stdout| self.clone().consume_stream(stdout));
+
+        let consume_stderr = child
+            .stderr
+            .take()
+            .map(|stderr| self.clone().consume_stream(stderr));
+
+        let await_exit = async {
+            tokio::spawn(async move {
+                child
+                    .wait()
+                    .await
+                    .wrap_err("yt-dlp command failed to run")?;
+
+                Ok::<(), Report>(())
+            })
+            .await??;
+
+            Ok(())
+        };
+
+        tokio::try_join!(
+            maybe_join(consume_stdout),
+            maybe_join(consume_stderr),
+            await_exit,
+        )
+        .wrap_err("Could not join child consumers for stdout, stderr and awaiting child exit.")?;
 
-        let mut lines = BufReader::new(stdout).lines();
+        Ok(())
+    }
 
-        let video = self.clone();
-        let process_pipe = tokio::spawn(async move {
+    fn consume_stream<A: AsyncRead + Unpin + Send + 'static>(
+        self: Arc<Self>,
+        reader: A,
+    ) -> JoinHandle<Result<()>> {
+        let mut lines = BufReader::new(reader).lines();
+
+        let video = self;
+        tokio::spawn(async move {
             while let Some(next_line) = lines.next_line().await? {
                 video
                     .use_title(|title| {
@@ -220,22 +308,7 @@ impl Video {
             }
 
             Ok::<(), Report>(())
-        });
-
-        let process_wait = tokio::spawn(async move {
-            child
-                .wait()
-                .await
-                .wrap_err("yt-dlp command failed to run")?;
-
-            Ok::<(), Report>(())
-        });
-
-        tokio::try_join!(async { process_pipe.await? }, async { process_wait.await? },)?;
-
-        self.set_stage_finished().await;
-
-        Ok(())
+        })
     }
 
     // Acquire read guards for all fine-grained access-controlled fields.

src/ui/mod.rs

@@ -285,6 +285,7 @@ impl Ui {
                         match video.stage() {
                             VideoStage::Initializing => "Intializing...",
                             VideoStage::Downloading => "Downloading...",
+                            VideoStage::ExtractingAudio => "Extracting audio...",
                             VideoStage::Finished => "Finished!",
                         },
                         style::video_stage_style(video.stage()),

src/ui/style.rs

@@ -48,6 +48,7 @@ fn video_stage_color(video_stage: &Stage) -> Color {
     match video_stage {
         Stage::Initializing => Color::LightCyan,
         Stage::Downloading => Color::LightYellow,
+        Stage::ExtractingAudio => Color::LightBlue,
         Stage::Finished => Color::LightGreen,
     }
 }

src/util.rs

@@ -2,6 +2,7 @@ use std::str::FromStr;
 
 use color_eyre::{eyre::Result, Report};
 use reqwest::{header::HeaderValue, Client, Url};
+use tokio::task::JoinHandle;
 use tracing::debug;
 
 pub async fn fetch_with_referer(url: &str, referer: &str) -> Result<String> {
@@ -26,3 +27,10 @@ pub async fn fetch_with_referer(url: &str, referer: &str) -> Result<String> {
     })
     .await?
 }
+
+#[inline]
+pub async fn maybe_join(maybe_spawned: Option<JoinHandle<Result<()>>>) -> Result<()> {
+    maybe_spawned.map(|join: JoinHandle<Result<()>>| async { join.await? });
+
+    Ok(())
+}