feat(yt/download/hooks): Show progress of post-processors

Otherwise, `yt` will just show everything downloaded, but does not continue.
Now users see, what is _actually_ happening.

8 files changed, 300 insertions, 97 deletions

diff --git a/crates/yt/src/commands/download/implm/download/download_options.rs b/crates/yt/src/commands/download/implm/download/download_options.rs
index 15fed7e..c9e5272 100644
--- a/crates/yt/src/commands/download/implm/download/download_options.rs
+++ b/crates/yt/src/commands/download/implm/download/download_options.rs
@@ -15,7 +15,7 @@ use yt_dlp::{YoutubeDL, options::YoutubeDLOptions};
 
 use crate::app::App;
 
-use super::progress_hook::wrapped_progress_hook;
+use super::hooks::{wrapped_progress_hook, wrapped_post_processor_hook};
 
 pub(crate) fn download_opts(
     app: &App,
@@ -23,6 +23,7 @@ pub(crate) fn download_opts(
 ) -> anyhow::Result<YoutubeDL> {
     YoutubeDLOptions::new()
         .with_progress_hook(wrapped_progress_hook)
+        .with_post_processor_hook(wrapped_post_processor_hook)
         .set("extract_flat", "in_playlist")
         .set(
             "extractor_args",
diff --git a/crates/yt/src/commands/download/implm/download/progress_hook.rs b/crates/yt/src/commands/download/implm/download/hooks.rs
index a414d4a..de52b98 100644
--- a/crates/yt/src/commands/download/implm/download/progress_hook.rs
+++ b/crates/yt/src/commands/download/implm/download/hooks.rs
@@ -12,15 +12,21 @@ use std::{
     collections::HashSet,
     io::{Write, stderr},
     process,
-    sync::{Mutex, atomic::Ordering},
+    sync::{
+        Mutex, OnceLock,
+        atomic::{AtomicUsize, Ordering},
+        mpsc::{self, Sender},
+    },
+    thread,
+    time::Duration,
 };
 
 use colors::{Colorize, IntoCanvas};
 use log::{Level, log_enabled};
-use yt_dlp::{json_cast, json_get, wrap_progress_hook};
+use yt_dlp::{json_cast, json_get, json_try_get, wrap_post_processor_hook, wrap_progress_hook};
 
 use crate::{
-    ansi_escape_codes::{clear_whole_line, move_to_col},
+    ansi_escape_codes::{clear_whole_line, cursor_up, erase_from_cursor_to_bottom, move_to_col},
     config::SHOULD_DISPLAY_COLOR,
     select::duration::MaybeDuration,
     shared::bytes::Bytes,
@@ -50,10 +56,31 @@ fn format_speed(speed: f64) -> String {
     let bytes = Bytes::new(speed.floor() as u64);
     format!("{bytes}/s")
 }
+fn get_title(info_dict: &serde_json::Map<String, serde_json::Value>) -> String {
+    if let Some(ext) = json_try_get!(info_dict, "ext", as_str) {
+        match ext {
+            "vtt" => {
+                format!(
+                    "Subtitles ({})",
+                    json_get_default!(info_dict, "name", as_str, "<No Subtitle Language>")
+                )
+            }
+            "webm" | "mp4" | "mp3" | "m4a" => {
+                json_get_default!(info_dict, "title", as_str, "<No title>").to_owned()
+            }
+            other => panic!("The extension '{other}' is not yet implemented"),
+        }
+    } else {
+        json_get_default!(info_dict, "title", as_str, "<No title>").to_owned()
+    }
+}
 
 /// # Panics
 /// If expectations fail.
-#[allow(clippy::needless_pass_by_value)]
+#[expect(
+    clippy::needless_pass_by_value,
+    reason = "We need to conform to the expected function signature"
+)]
 pub(crate) fn progress_hook(
     input: serde_json::Map<String, serde_json::Value>,
 ) -> Result<(), std::io::Error> {
@@ -65,21 +92,6 @@ pub(crate) fn progress_hook(
 
     let info_dict = json_get!(input, "info_dict", as_object);
 
-    let get_title = || -> String {
-        match json_get!(info_dict, "ext", as_str) {
-            "vtt" => {
-                format!(
-                    "Subtitles ({})",
-                    json_get_default!(info_dict, "name", as_str, "<No Subtitle Language>")
-                )
-            }
-            "webm" | "mp4" | "mp3" | "m4a" => {
-                json_get_default!(info_dict, "title", as_str, "<No title>").to_owned()
-            }
-            other => panic!("The extension '{other}' is not yet implemented"),
-        }
-    };
-
     match json_get!(input, "status", as_str) {
         "downloading" => {
             let elapsed = json_get_default!(input, "elapsed", as_f64, 0.0);
@@ -127,12 +139,12 @@ pub(crate) fn progress_hook(
                 }
             };
 
-            clear_whole_line();
-            move_to_col(1);
+            clear_whole_line(stderr());
+            move_to_col(stderr(), 1);
 
             let should_use_color = SHOULD_DISPLAY_COLOR.load(Ordering::Relaxed);
 
-            let title = get_title();
+            let title = get_title(info_dict);
 
             eprint!(
                 "{} [{}/{} at {}] -> [{} of {}{} {}] ",
@@ -175,7 +187,7 @@ pub(crate) fn progress_hook(
         }
         "finished" => {
             let should_use_color = SHOULD_DISPLAY_COLOR.load(Ordering::Relaxed);
-            let title = get_title();
+            let title = get_title(info_dict);
 
             let has_already_been_printed = {
                 let titles = TITLES.lock().expect("The lock should work");
@@ -198,7 +210,7 @@ pub(crate) fn progress_hook(
         "error" => {
             // TODO: This should probably return an Err. But I'm not so sure where the error would
             // bubble up to (i.e., who would catch it) <2025-01-21>
-            eprintln!("-> Error while downloading: {}", get_title());
+            eprintln!("-> Error while downloading: {}", get_title(info_dict));
             process::exit(1);
         }
         other => unreachable!("'{other}' should not be a valid state!"),
@@ -207,4 +219,141 @@ pub(crate) fn progress_hook(
     Ok(())
 }
 
+//  postprocessor_hooks:  A list of functions that get called on postprocessing
+//                        progress, with a dictionary with the entries
+//                        * status: One of "started", "processing", or "finished".
+//                                  Check this first and ignore unknown values.
+//                        * postprocessor: Name of the postprocessor
+//                        * info_dict: The extracted info_dict
+//
+//                        Progress hooks are guaranteed to be called at least twice
+//                        (with status "started" and "finished") if the processing is successful.
+#[expect(
+    clippy::needless_pass_by_value,
+    clippy::unnecessary_wraps,
+    reason = "The macro expects this function signature"
+)]
+pub(crate) fn post_processor_hook(
+    input: serde_json::Map<String, serde_json::Value>,
+) -> Result<(), std::io::Error> {
+    // Only add the handler, if the log-level is higher than Debug (this avoids covering debug
+    // messages).
+    if log_enabled!(Level::Debug) {
+        return Ok(());
+    }
+
+    let info_dict = json_get!(input, "info_dict", as_object);
+
+    match json_get!(input, "status", as_str) {
+        "started" => {
+            // Normally `yt_dlp` just gives us a `started` and a `finished` message, so we show all
+            // the post_processors that have currently been recorded as `started` and not yet as
+            // `finished`.
+            let post_processor = json_get!(input, "postprocessor", as_str);
+            maybe_setup_post_processor_progress(ProgressChange::Started(
+                post_processor.to_string(),
+            ));
+        }
+        "processing" => {
+            let post_processor = json_get!(input, "postprocessor", as_str);
+            let title = get_title(info_dict);
+
+            eprintln!("Processesing actually triggered, `{post_processor}` for `{title}`",);
+        }
+        "finished" => {
+            let post_processor = json_get!(input, "postprocessor", as_str);
+            maybe_setup_post_processor_progress(ProgressChange::Finished(
+                post_processor.to_string(),
+            ));
+
+            // let should_use_color = SHOULD_DISPLAY_COLOR.load(Ordering::Relaxed);
+            // let title = get_title(info_dict);
+            //
+            // eprintln!(
+            //     "{} ({}) finished.",
+            //     title.bold().blue().render(should_use_color),
+            //     post_processor.bold().green().render(should_use_color),
+            // );
+        }
+        other => unreachable!("'{other}' should not be a valid state!"),
+    }
+
+    Ok(())
+}
+
+enum ProgressChange {
+    Started(String),
+    Finished(String),
+}
+
+fn maybe_setup_post_processor_progress(pc: ProgressChange) {
+    fn new_state_id() -> usize {
+        static COUNTER: AtomicUsize = AtomicUsize::new(0);
+        COUNTER.fetch_add(1, Ordering::Relaxed)
+    }
+    static TX: OnceLock<Sender<ProgressChange>> = const { OnceLock::new() };
+
+    let should_use_color = SHOULD_DISPLAY_COLOR.load(Ordering::Relaxed);
+
+    let tx = TX.get_or_init(|| {
+        let (tx, rx) = mpsc::channel();
+        thread::spawn(move || {
+            fn process_new_data(data: &mut HashSet<String>, new_data: ProgressChange) {
+                match new_data {
+                    ProgressChange::Started(name) => {
+                        data.insert(name);
+                    }
+                    ProgressChange::Finished(name) => {
+                        data.remove(&name);
+                    }
+                }
+            }
+
+            let mut data = HashSet::new();
+
+            'main: loop {
+                let previous_length = data.len();
+
+                match rx.recv_timeout(Duration::from_millis(100)) {
+                    Ok(new_data) => process_new_data(&mut data, new_data),
+                    Err(err) => match err {
+                        mpsc::RecvTimeoutError::Timeout => (),
+                        mpsc::RecvTimeoutError::Disconnected => break 'main,
+                    },
+                }
+
+                // If we received multiple things at once, process them all at once.
+                while let Ok(new_data) = rx.try_recv() {
+                    process_new_data(&mut data, new_data);
+                }
+
+                let spinner = match new_state_id() % 4 {
+                    0 => '|',
+                    1 => '/',
+                    2 => '-',
+                    3 => '\\',
+                    _ => unreachable!("The `mod` operation does not permit these numbers"),
+                };
+
+                cursor_up(stderr(), previous_length);
+                erase_from_cursor_to_bottom(stderr());
+                for post_processor in &data {
+                    clear_whole_line(stderr());
+                    move_to_col(stderr(), 1);
+
+                    eprintln!(
+                        "[{spinner}] {} ",
+                        post_processor.blue().bold().render(should_use_color)
+                    );
+                }
+            }
+        });
+        tx
+    });
+
+    tx.send(pc)
+        .expect("The recieving end should not be dropped before we drop the tx");
+}
+
 wrap_progress_hook!(progress_hook, wrapped_progress_hook);
+wrap_post_processor_hook!(post_processor_hook, wrapped_post_processor_hook);
diff --git a/crates/yt/src/commands/download/implm/download/mod.rs b/crates/yt/src/commands/download/implm/download/mod.rs
index ab9de80..f761c70 100644
--- a/crates/yt/src/commands/download/implm/download/mod.rs
+++ b/crates/yt/src/commands/download/implm/download/mod.rs
@@ -29,7 +29,7 @@ use yt_dlp::YoutubeDL;
 
 #[allow(clippy::module_name_repetitions)]
 pub(crate) mod download_options;
-pub(crate) mod progress_hook;
+pub(crate) mod hooks;
 
 #[derive(Debug)]
 #[allow(clippy::module_name_repetitions)]
diff --git a/crates/yt_dlp/src/hooks.rs b/crates/yt_dlp/src/hooks.rs
new file mode 100644
index 0000000..df70ecd
--- /dev/null
+++ b/crates/yt_dlp/src/hooks.rs
@@ -0,0 +1,62 @@
+// yt - A fully featured command line YouTube client
+//
+// Copyright (C) 2025 Benedikt Peetz <benedikt.peetz@b-peetz.de>
+// SPDX-License-Identifier: GPL-3.0-or-later
+//
+// This file is part of Yt.
+//
+// You should have received a copy of the License along with this program.
+// If not, see <https://www.gnu.org/licenses/gpl-3.0.txt>.
+
+#[macro_export]
+macro_rules! wrap_progress_hook {
+    ($name:ident, $new_name:ident) => {
+        yt_dlp::wrap_hook! {"progress_hook", $name, $new_name}
+    };
+}
+
+#[macro_export]
+macro_rules! wrap_post_processor_hook {
+    ($name:ident, $new_name:ident) => {
+        yt_dlp::wrap_hook! {"post_processor_hook", $name, $new_name}
+    };
+}
+
+#[macro_export]
+macro_rules! wrap_hook {
+    ($ty:literal, $name:ident, $new_name:ident) => {
+        pub(crate) fn $new_name(
+            py: yt_dlp::hooks::__priv::pyo3::Python<'_>,
+        ) -> yt_dlp::hooks::__priv::pyo3::PyResult<
+            yt_dlp::hooks::__priv::pyo3::Bound<'_, yt_dlp::hooks::__priv::pyo3::types::PyCFunction>,
+        > {
+            #[yt_dlp::hooks::__priv::pyo3::pyfunction]
+            #[pyo3(crate = "yt_dlp::hooks::__priv::pyo3")]
+            fn inner(
+                input: yt_dlp::hooks::__priv::pyo3::Bound<
+                    '_,
+                    yt_dlp::hooks::__priv::pyo3::types::PyDict,
+                >,
+            ) -> yt_dlp::hooks::__priv::pyo3::PyResult<()> {
+                let processed_input = {
+                    let new_dict = yt_dlp::hooks::__priv::filter_dict(input);
+                    yt_dlp::hooks::__priv::json_dumps(&new_dict)
+                };
+
+                $name(processed_input)?;
+
+                Ok(())
+            }
+
+            let module = yt_dlp::hooks::__priv::pyo3::types::PyModule::new(py, $ty)?;
+            let fun = yt_dlp::hooks::__priv::pyo3::wrap_pyfunction!(inner, module)?;
+
+            Ok(fun)
+        }
+    };
+}
+
+pub mod __priv {
+    pub use crate::info_json::{json_dumps, json_loads, filter_dict};
+    pub use pyo3;
+}
diff --git a/crates/yt_dlp/src/info_json.rs b/crates/yt_dlp/src/info_json.rs
index 402acb4..df49218 100644
--- a/crates/yt_dlp/src/info_json.rs
+++ b/crates/yt_dlp/src/info_json.rs
@@ -54,3 +54,39 @@ pub fn json_dumps(input: &Bound<'_, PyDict>) -> serde_json::Map<String, serde_js
         _ => unreachable!("These should not be json.dumps output"),
     }
 }
+
+/// # Panics
+/// If expectation about python operations fail.
+#[must_use]
+pub fn filter_dict(input: Bound<'_, PyDict>) -> Bound<'_, PyDict> {
+    let new_dict = PyDict::new(input.py());
+
+    input
+        .into_iter()
+        .filter_map(|(name, value)| {
+            let real_name = name.extract::<String>().expect("Should always be a string");
+
+            if real_name.starts_with('_') {
+                None
+            } else {
+                let value = if value.is_instance_of::<PyDict>() {
+                    filter_dict(
+                        value
+                            .cast_into::<PyDict>()
+                            .expect("to be a dict, because we checked"),
+                    )
+                    .into_any()
+                } else {
+                    value
+                };
+                Some((real_name, value))
+            }
+        })
+        .for_each(|(key, value)| {
+            new_dict
+                .set_item(&key, value)
+                .expect("This is a transposition, should always be valid");
+        });
+
+    new_dict
+}
diff --git a/crates/yt_dlp/src/lib.rs b/crates/yt_dlp/src/lib.rs
index 4b252de..abe766d 100644
--- a/crates/yt_dlp/src/lib.rs
+++ b/crates/yt_dlp/src/lib.rs
@@ -27,7 +27,7 @@ use crate::{
 pub mod info_json;
 pub mod options;
 pub mod post_processors;
-pub mod progress_hook;
+pub mod hooks;
 pub mod python_error;
 
 #[macro_export]
diff --git a/crates/yt_dlp/src/options.rs b/crates/yt_dlp/src/options.rs
index 4b8906e..a87473d 100644
--- a/crates/yt_dlp/src/options.rs
+++ b/crates/yt_dlp/src/options.rs
@@ -21,7 +21,7 @@ use crate::{
     python_error::{IntoPythonError, PythonError},
 };
 
-pub type ProgressHookFunction = fn(py: Python<'_>) -> PyResult<Bound<'_, PyCFunction>>;
+pub type HookFunction = fn(py: Python<'_>) -> PyResult<Bound<'_, PyCFunction>>;
 pub type PostProcessorFunction = fn(py: Python<'_>) -> PyResult<Bound<'_, PyAny>>;
 
 /// Options, that are used to customize the download behaviour.
@@ -32,7 +32,8 @@ pub type PostProcessorFunction = fn(py: Python<'_>) -> PyResult<Bound<'_, PyAny>
 #[derive(Default, Debug)]
 pub struct YoutubeDLOptions {
     options: serde_json::Map<String, serde_json::Value>,
-    progress_hook: Option<ProgressHookFunction>,
+    progress_hook: Option<HookFunction>,
+    post_processor_hook: Option<HookFunction>,
     post_processors: Vec<PostProcessorFunction>,
 }
 
@@ -42,6 +43,7 @@ impl YoutubeDLOptions {
         let me = Self {
             options: serde_json::Map::new(),
             progress_hook: None,
+            post_processor_hook: None,
             post_processors: vec![],
         };
 
@@ -57,7 +59,7 @@ impl YoutubeDLOptions {
     }
 
     #[must_use]
-    pub fn with_progress_hook(self, progress_hook: ProgressHookFunction) -> Self {
+    pub fn with_progress_hook(self, progress_hook: HookFunction) -> Self {
         if let Some(_previous_hook) = self.progress_hook {
             todo!()
         } else {
@@ -67,6 +69,17 @@ impl YoutubeDLOptions {
             }
         }
     }
+    #[must_use]
+    pub fn with_post_processor_hook(self, post_processor_hook: HookFunction) -> Self {
+        if let Some(_previous_hook) = self.post_processor_hook {
+            todo!()
+        } else {
+            Self {
+                post_processor_hook: Some(post_processor_hook),
+                ..self
+            }
+        }
+    }
 
     #[must_use]
     pub fn with_post_processor(mut self, pp: PostProcessorFunction) -> Self {
@@ -135,6 +148,15 @@ signal.signal(signal.SIGINT, signal.SIG_DFL)
                     opts.set_item(intern!(py, "progress_hooks"), vec![ph(py).wrap_exc(py)?])
                         .wrap_exc(py)?;
                 }
+
+                // Setup the post_processor hook
+                if let Some(ph) = options.post_processor_hook {
+                    opts.set_item(
+                        intern!(py, "postprocessor_hooks"),
+                        vec![ph(py).wrap_exc(py)?],
+                    )
+                    .wrap_exc(py)?;
+                }
             }
 
             {
diff --git a/crates/yt_dlp/src/progress_hook.rs b/crates/yt_dlp/src/progress_hook.rs
deleted file mode 100644
index 7e5f8a5..0000000
--- a/crates/yt_dlp/src/progress_hook.rs
+++ /dev/null
@@ -1,67 +0,0 @@
-// yt - A fully featured command line YouTube client
-//
-// Copyright (C) 2025 Benedikt Peetz <benedikt.peetz@b-peetz.de>
-// SPDX-License-Identifier: GPL-3.0-or-later
-//
-// This file is part of Yt.
-//
-// You should have received a copy of the License along with this program.
-// If not, see <https://www.gnu.org/licenses/gpl-3.0.txt>.
-
-#[macro_export]
-macro_rules! wrap_progress_hook {
-    ($name:ident, $new_name:ident) => {
-        pub(crate) fn $new_name(
-            py: yt_dlp::progress_hook::__priv::pyo3::Python<'_>,
-        ) -> yt_dlp::progress_hook::__priv::pyo3::PyResult<
-            yt_dlp::progress_hook::__priv::pyo3::Bound<
-                '_,
-                yt_dlp::progress_hook::__priv::pyo3::types::PyCFunction,
-            >,
-        > {
-            #[yt_dlp::progress_hook::__priv::pyo3::pyfunction]
-            #[pyo3(crate = "yt_dlp::progress_hook::__priv::pyo3")]
-            fn inner(
-                input: yt_dlp::progress_hook::__priv::pyo3::Bound<
-                    '_,
-                    yt_dlp::progress_hook::__priv::pyo3::types::PyDict,
-                >,
-            ) -> yt_dlp::progress_hook::__priv::pyo3::PyResult<()> {
-                let processed_input = {
-                    let new_dict = yt_dlp::progress_hook::__priv::pyo3::types::PyDict::new(input.py());
-
-                    input
-                        .into_iter()
-                        .filter_map(|(name, value)| {
-                            let real_name = yt_dlp::progress_hook::__priv::pyo3::types::PyAnyMethods::extract::<String>(&name).expect("Should always be a string");
-
-                            if real_name.starts_with('_') {
-                                None
-                            } else {
-                                Some((real_name, value))
-                            }
-                        })
-                        .for_each(|(key, value)| {
-                            yt_dlp::progress_hook::__priv::pyo3::types::PyDictMethods::set_item(&new_dict, &key, value)
-                                .expect("This is a transpositions, should always be valid");
-                        });
-                    yt_dlp::progress_hook::__priv::json_dumps(&new_dict)
-                };
-
-                $name(processed_input)?;
-
-                Ok(())
-            }
-
-            let module = yt_dlp::progress_hook::__priv::pyo3::types::PyModule::new(py, "progress_hook")?;
-            let fun = yt_dlp::progress_hook::__priv::pyo3::wrap_pyfunction!(inner, module)?;
-
-            Ok(fun)
-        }
-    };
-}
-
-pub mod __priv {
-    pub use crate::info_json::{json_dumps, json_loads};
-    pub use pyo3;
-}