Move stream download logic to fix cms error/rate limiting

crunchy-cli-core/src/archive/command.rs

@@ -1,19 +1,22 @@
 use crate::archive::filter::ArchiveFilter;
 use crate::utils::context::Context;
-use crate::utils::download::MergeBehavior;
+use crate::utils::download::{DownloadBuilder, DownloadFormat, MergeBehavior};
 use crate::utils::ffmpeg::FFmpegPreset;
 use crate::utils::filter::Filter;
-use crate::utils::format::formats_visual_output;
+use crate::utils::format::{Format, SingleFormat};
 use crate::utils::locale::all_locale_in_locales;
 use crate::utils::log::progress;
 use crate::utils::os::{free_file, has_ffmpeg, is_special_file};
 use crate::utils::parse::parse_url;
+use crate::utils::video::variant_data_from_stream;
 use crate::Execute;
 use anyhow::bail;
 use anyhow::Result;
-use crunchyroll_rs::media::Resolution;
+use chrono::Duration;
+use crunchyroll_rs::media::{Resolution, Subtitle};
 use crunchyroll_rs::Locale;
 use log::debug;
+use std::collections::HashMap;
 use std::path::PathBuf;
 
 #[derive(Clone, Debug, clap::Parser)]
@@ -135,19 +138,33 @@ impl Execute for Archive {
 
         for (i, (media_collection, url_filter)) in parsed_urls.into_iter().enumerate() {
             let progress_handler = progress!("Fetching series details");
-            let archive_formats = ArchiveFilter::new(url_filter, self.clone())
+            let single_format_collection = ArchiveFilter::new(url_filter, self.clone())
                 .visit(media_collection)
                 .await?;
 
-            if archive_formats.is_empty() {
+            if single_format_collection.is_empty() {
                 progress_handler.stop(format!("Skipping url {} (no matching videos found)", i + 1));
                 continue;
             }
             progress_handler.stop(format!("Loaded series information for url {}", i + 1));
 
-            formats_visual_output(archive_formats.iter().map(|(_, f)| f).collect());
+            single_format_collection.full_visual_output();
+
+            let download_builder = DownloadBuilder::new()
+                .default_subtitle(self.default_subtitle.clone())
+                .ffmpeg_preset(self.ffmpeg_preset.clone().unwrap_or_default())
+                .output_format(Some("matroska".to_string()))
+                .audio_sort(Some(self.locale.clone()))
+                .subtitle_sort(Some(self.subtitle.clone()));
+
+            for single_formats in single_format_collection.into_iter() {
+                let (download_formats, mut format) = get_format(&self, &single_formats).await?;
+
+                let mut downloader = download_builder.clone().build();
+                for download_format in download_formats {
+                    downloader.add_format(download_format)
+                }
 
-            for (downloader, mut format) in archive_formats {
                 let formatted_path = format.format_path((&self.output).into(), true);
                 let (path, changed) = free_file(formatted_path.clone());
 
@@ -183,3 +200,104 @@ impl Execute for Archive {
         Ok(())
     }
 }
+
+async fn get_format(
+    archive: &Archive,
+    single_formats: &Vec<SingleFormat>,
+) -> Result<(Vec<DownloadFormat>, Format)> {
+    let mut format_pairs = vec![];
+    let mut single_format_to_format_pairs = vec![];
+
+    for single_format in single_formats {
+        let stream = single_format.stream().await?;
+        let Some((video, audio)) = variant_data_from_stream(&stream, &archive.resolution).await? else {
+            if single_format.is_episode() {
+                bail!(
+                    "Resolution ({}) is not available for episode {} ({}) of {} season {}",
+                    archive.resolution,
+                    single_format.episode_number,
+                    single_format.title,
+                    single_format.series_name,
+                    single_format.season_number,
+                )
+            } else {
+                bail!(
+                    "Resolution ({}) is not available for {} ({})",
+                    archive.resolution,
+                    single_format.source_type(),
+                    single_format.title
+                )
+            }
+        };
+
+        let subtitles: Vec<Subtitle> = archive
+            .subtitle
+            .iter()
+            .filter_map(|s| stream.subtitles.get(s).cloned())
+            .collect();
+
+        format_pairs.push((single_format, video.clone(), audio, subtitles.clone()));
+        single_format_to_format_pairs.push((single_format.clone(), video, subtitles))
+    }
+
+    let mut download_formats = vec![];
+
+    match archive.merge {
+        MergeBehavior::Video => {
+            for (single_format, video, audio, subtitles) in format_pairs {
+                download_formats.push(DownloadFormat {
+                    video: (video, single_format.audio.clone()),
+                    audios: vec![(audio, single_format.audio.clone())],
+                    subtitles,
+                })
+            }
+        }
+        MergeBehavior::Audio => download_formats.push(DownloadFormat {
+            video: (
+                (*format_pairs.first().unwrap()).1.clone(),
+                (*format_pairs.first().unwrap()).0.audio.clone(),
+            ),
+            audios: format_pairs
+                .iter()
+                .map(|(single_format, _, audio, _)| (audio.clone(), single_format.audio.clone()))
+                .collect(),
+            // mix all subtitles together and then reduce them via a map so that only one subtitle
+            // per language exists
+            subtitles: format_pairs
+                .iter()
+                .flat_map(|(_, _, _, subtitles)| subtitles.clone())
+                .map(|s| (s.locale.clone(), s))
+                .collect::<HashMap<Locale, Subtitle>>()
+                .into_values()
+                .collect(),
+        }),
+        MergeBehavior::Auto => {
+            let mut d_formats: HashMap<Duration, DownloadFormat> = HashMap::new();
+
+            for (single_format, video, audio, subtitles) in format_pairs {
+                if let Some(d_format) = d_formats.get_mut(&single_format.duration) {
+                    d_format.audios.push((audio, single_format.audio.clone()));
+                    d_format.subtitles.extend(subtitles)
+                } else {
+                    d_formats.insert(
+                        single_format.duration,
+                        DownloadFormat {
+                            video: (video, single_format.audio.clone()),
+                            audios: vec![(audio, single_format.audio.clone())],
+                            subtitles,
+                        },
+                    );
+                }
+            }
+
+            for d_format in d_formats.into_values() {
+                download_formats.push(d_format)
+            }
+        }
+    }
+
+    Ok((
+        download_formats,
+        Format::from_single_formats(single_format_to_format_pairs),
+    ))
+}

crunchy-cli-core/src/archive/filter.rs

@@ -1,24 +1,11 @@
 use crate::archive::command::Archive;
-use crate::utils::download::{DownloadBuilder, DownloadFormat, Downloader, MergeBehavior};
 use crate::utils::filter::{real_dedup_vec, Filter};
-use crate::utils::format::{Format, SingleFormat};
+use crate::utils::format::{Format, SingleFormat, SingleFormatCollection};
 use crate::utils::parse::UrlFilter;
-use crate::utils::video::variant_data_from_stream;
-use anyhow::{bail, Result};
-use chrono::Duration;
-use crunchyroll_rs::media::{Subtitle, VariantData};
+use anyhow::Result;
 use crunchyroll_rs::{Concert, Episode, Locale, Movie, MovieListing, MusicVideo, Season, Series};
 use log::warn;
-use std::collections::HashMap;
-use std::hash::Hash;
-
-pub(crate) struct FilterResult {
-    format: SingleFormat,
-    video: VariantData,
-    audio: VariantData,
-    duration: Duration,
-    subtitles: Vec<Subtitle>,
-}
+use std::collections::{BTreeMap, HashMap};
 
 enum Visited {
     Series,
@@ -48,8 +35,8 @@ impl ArchiveFilter {
 
 #[async_trait::async_trait]
 impl Filter for ArchiveFilter {
-    type T = Vec<FilterResult>;
-    type Output = (Downloader, Format);
+    type T = Vec<SingleFormat>;
+    type Output = SingleFormatCollection;
 
     async fn visit_series(&mut self, series: Series) -> Result<Vec<Season>> {
         // `series.audio_locales` isn't always populated b/c of crunchyrolls api. so check if the
@@ -168,11 +155,19 @@ impl Filter for ArchiveFilter {
         let mut episodes = vec![];
         if !matches!(self.visited, Visited::Series) && !matches!(self.visited, Visited::Season) {
             if self.archive.locale.contains(&episode.audio_locale) {
-                episodes.push(episode.clone())
+                episodes.push((episode.clone(), episode.subtitle_locales.clone()))
             }
-            episodes.extend(episode.version(self.archive.locale.clone()).await?);
-            let audio_locales: Vec<Locale> =
-                episodes.iter().map(|e| e.audio_locale.clone()).collect();
+            episodes.extend(
+                episode
+                    .version(self.archive.locale.clone())
+                    .await?
+                    .into_iter()
+                    .map(|e| (e.clone(), e.subtitle_locales.clone())),
+            );
+            let audio_locales: Vec<Locale> = episodes
+                .iter()
+                .map(|(e, _)| e.audio_locale.clone())
+                .collect();
             let missing_audio = missing_locales(&audio_locales, &self.archive.locale);
             if !missing_audio.is_empty() {
                 warn!(
@@ -186,11 +181,8 @@ impl Filter for ArchiveFilter {
                 )
             }
 
-            let mut subtitle_locales: Vec<Locale> = episodes
-                .iter()
-                .map(|e| e.subtitle_locales.clone())
-                .flatten()
-                .collect();
+            let mut subtitle_locales: Vec<Locale> =
+                episodes.iter().map(|(_, s)| s.clone()).flatten().collect();
             real_dedup_vec(&mut subtitle_locales);
             let missing_subtitles = missing_locales(&subtitle_locales, &self.archive.subtitle);
             if !missing_subtitles.is_empty()
@@ -210,81 +202,49 @@ impl Filter for ArchiveFilter {
                 self.season_subtitles_missing.push(episode.season_number)
             }
         } else {
-            episodes.push(episode.clone())
+            episodes.push((episode.clone(), episode.subtitle_locales.clone()))
         }
 
-        let mut formats = vec![];
-        for episode in episodes {
-            let stream = episode.streams().await?;
-            let (video, audio) = if let Some((video, audio)) =
-                variant_data_from_stream(&stream, &self.archive.resolution).await?
+        let relative_episode_number = if Format::has_relative_episodes_fmt(&self.archive.output) {
+            if self
+                .season_episode_count
+                .get(&episode.season_number)
+                .is_none()
             {
-                (video, audio)
-            } else {
-                bail!(
-                    "Resolution ({}) is not available for episode {} ({}) of {} season {}",
-                    &self.archive.resolution,
+                let season_episodes = episode.season().await?.episodes().await?;
+                self.season_episode_count.insert(
+                    episode.season_number,
+                    season_episodes.into_iter().map(|e| e.id).collect(),
+                );
+            }
+            let relative_episode_number = self
+                .season_episode_count
+                .get(&episode.season_number)
+                .unwrap()
+                .iter()
+                .position(|id| id == &episode.id);
+            if relative_episode_number.is_none() {
+                warn!(
+                    "Failed to get relative episode number for episode {} ({}) of {} season {}",
                     episode.episode_number,
                     episode.title,
                     episode.series_title,
                     episode.season_number,
-                );
-            };
-            let subtitles: Vec<Subtitle> = self
-                .archive
-                .subtitle
-                .iter()
-                .filter_map(|s| stream.subtitles.get(s).cloned())
-                .collect();
+                )
+            }
+            relative_episode_number
+        } else {
+            None
+        };
 
-            let relative_episode_number = if Format::has_relative_episodes_fmt(&self.archive.output)
-            {
-                if self
-                    .season_episode_count
-                    .get(&episode.season_number)
-                    .is_none()
-                {
-                    let season_episodes = episode.season().await?.episodes().await?;
-                    self.season_episode_count.insert(
-                        episode.season_number,
-                        season_episodes.into_iter().map(|e| e.id).collect(),
-                    );
-                }
-                let relative_episode_number = self
-                    .season_episode_count
-                    .get(&episode.season_number)
-                    .unwrap()
-                    .iter()
-                    .position(|id| id == &episode.id);
-                if relative_episode_number.is_none() {
-                    warn!(
-                        "Failed to get relative episode number for episode {} ({}) of {} season {}",
-                        episode.episode_number,
-                        episode.title,
-                        episode.series_title,
-                        episode.season_number,
-                    )
-                }
-                relative_episode_number
-            } else {
-                None
-            };
-
-            formats.push(FilterResult {
-                format: SingleFormat::new_from_episode(
-                    &episode,
-                    &video,
-                    subtitles.iter().map(|s| s.locale.clone()).collect(),
-                    relative_episode_number.map(|n| n as u32),
-                ),
-                video,
-                audio,
-                duration: episode.duration.clone(),
-                subtitles,
-            })
-        }
-
-        Ok(Some(formats))
+        Ok(Some(
+            episodes
+                .into_iter()
+                .map(|(e, s)| {
+                    SingleFormat::new_from_episode(e, s, relative_episode_number.map(|n| n as u32))
+                })
+                .collect(),
+        ))
     }
 
     async fn visit_movie_listing(&mut self, movie_listing: MovieListing) -> Result<Vec<Movie>> {
@@ -292,199 +252,37 @@ impl Filter for ArchiveFilter {
     }
 
     async fn visit_movie(&mut self, movie: Movie) -> Result<Option<Self::T>> {
-        let stream = movie.streams().await?;
-        let subtitles: Vec<&Subtitle> = self
-            .archive
-            .subtitle
-            .iter()
-            .filter_map(|l| stream.subtitles.get(l))
-            .collect();
-
-        let missing_subtitles = missing_locales(
-            &subtitles.iter().map(|&s| s.locale.clone()).collect(),
-            &self.archive.subtitle,
-        );
-        if !missing_subtitles.is_empty() {
-            warn!(
-                "Movie '{}' is not available with {} subtitles",
-                movie.title,
-                missing_subtitles
-                    .into_iter()
-                    .map(|l| l.to_string())
-                    .collect::<Vec<String>>()
-                    .join(", ")
-            )
-        }
-
-        let (video, audio) = if let Some((video, audio)) =
-            variant_data_from_stream(&stream, &self.archive.resolution).await?
-        {
-            (video, audio)
-        } else {
-            bail!(
-                "Resolution ({}) of movie {} is not available",
-                self.archive.resolution,
-                movie.title
-            )
-        };
-
-        Ok(Some(vec![FilterResult {
-            format: SingleFormat::new_from_movie(&movie, &video, vec![]),
-            video,
-            audio,
-            duration: movie.duration,
-            subtitles: vec![],
-        }]))
+        Ok(Some(vec![SingleFormat::new_from_movie(movie, vec![])]))
     }
 
     async fn visit_music_video(&mut self, music_video: MusicVideo) -> Result<Option<Self::T>> {
-        let stream = music_video.streams().await?;
-        let (video, audio) = if let Some((video, audio)) =
-            variant_data_from_stream(&stream, &self.archive.resolution).await?
-        {
-            (video, audio)
-        } else {
-            bail!(
-                "Resolution ({}) of music video {} is not available",
-                self.archive.resolution,
-                music_video.title
-            )
-        };
-
-        Ok(Some(vec![FilterResult {
-            format: SingleFormat::new_from_music_video(&music_video, &video),
-            video,
-            audio,
-            duration: music_video.duration,
-            subtitles: vec![],
-        }]))
+        Ok(Some(vec![SingleFormat::new_from_music_video(music_video)]))
     }
 
     async fn visit_concert(&mut self, concert: Concert) -> Result<Option<Self::T>> {
-        let stream = concert.streams().await?;
-        let (video, audio) = if let Some((video, audio)) =
-            variant_data_from_stream(&stream, &self.archive.resolution).await?
-        {
-            (video, audio)
-        } else {
-            bail!(
-                "Resolution ({}x{}) of music video {} is not available",
-                self.archive.resolution.width,
-                self.archive.resolution.height,
-                concert.title
-            )
-        };
-
-        Ok(Some(vec![FilterResult {
-            format: SingleFormat::new_from_concert(&concert, &video),
-            video,
-            audio,
-            duration: concert.duration,
-            subtitles: vec![],
-        }]))
+        Ok(Some(vec![SingleFormat::new_from_concert(concert)]))
     }
 
-    async fn finish(self, input: Vec<Self::T>) -> Result<Vec<Self::Output>> {
-        let flatten_input: Vec<FilterResult> = input.into_iter().flatten().collect();
+    async fn finish(self, input: Vec<Self::T>) -> Result<Self::Output> {
+        let flatten_input: Self::T = input.into_iter().flatten().collect();
 
-        #[derive(Hash, Eq, PartialEq)]
-        struct SortKey {
-            season: u32,
-            episode: String,
-        }
+        let mut single_format_collection = SingleFormatCollection::new();
 
-        let mut sorted: HashMap<SortKey, Vec<FilterResult>> = HashMap::new();
+        struct SortKey(u32, String);
+
+        let mut sorted: BTreeMap<(u32, String), Self::T> = BTreeMap::new();
         for data in flatten_input {
             sorted
-                .entry(SortKey {
-                    season: data.format.season_number,
-                    episode: data.format.episode_number.to_string(),
-                })
+                .entry((data.season_number, data.sequence_number.to_string()))
                 .or_insert(vec![])
                 .push(data)
         }
 
-        let mut values: Vec<Vec<FilterResult>> = sorted.into_values().collect();
-        values.sort_by(|a, b| {
-            a.first()
-                .unwrap()
-                .format
-                .sequence_number
-                .total_cmp(&b.first().unwrap().format.sequence_number)
-        });
-
-        let mut result = vec![];
-        for data in values {
-            let single_formats: Vec<SingleFormat> =
-                data.iter().map(|fr| fr.format.clone()).collect();
-            let format = Format::from_single_formats(single_formats);
-
-            let mut downloader = DownloadBuilder::new()
-                .default_subtitle(self.archive.default_subtitle.clone())
-                .ffmpeg_preset(self.archive.ffmpeg_preset.clone().unwrap_or_default())
-                .output_format(Some("matroska".to_string()))
-                .audio_sort(Some(self.archive.locale.clone()))
-                .subtitle_sort(Some(self.archive.subtitle.clone()))
-                .build();
-
-            match self.archive.merge.clone() {
-                MergeBehavior::Video => {
-                    for d in data {
-                        downloader.add_format(DownloadFormat {
-                            video: (d.video, d.format.audio.clone()),
-                            audios: vec![(d.audio, d.format.audio.clone())],
-                            subtitles: d.subtitles,
-                        })
-                    }
-                }
-                MergeBehavior::Audio => downloader.add_format(DownloadFormat {
-                    video: (
-                        data.first().unwrap().video.clone(),
-                        data.first().unwrap().format.audio.clone(),
-                    ),
-                    audios: data
-                        .iter()
-                        .map(|d| (d.audio.clone(), d.format.audio.clone()))
-                        .collect(),
-                    // mix all subtitles together and then reduce them via a map so that only one
-                    // subtitle per language exists
-                    subtitles: data
-                        .iter()
-                        .flat_map(|d| d.subtitles.clone())
-                        .map(|s| (s.locale.clone(), s))
-                        .collect::<HashMap<Locale, Subtitle>>()
-                        .into_values()
-                        .collect(),
-                }),
-                MergeBehavior::Auto => {
-                    let mut download_formats: HashMap<Duration, DownloadFormat> = HashMap::new();
-
-                    for d in data {
-                        if let Some(download_format) = download_formats.get_mut(&d.duration) {
-                            download_format.audios.push((d.audio, d.format.audio));
-                            download_format.subtitles.extend(d.subtitles)
-                        } else {
-                            download_formats.insert(
-                                d.duration,
-                                DownloadFormat {
-                                    video: (d.video, d.format.audio.clone()),
-                                    audios: vec![(d.audio, d.format.audio)],
-                                    subtitles: d.subtitles,
-                                },
-                            );
-                        }
-                    }
-
-                    for download_format in download_formats.into_values() {
-                        downloader.add_format(download_format)
-                    }
-                }
-            }
-
-            result.push((downloader, format))
+        for data in sorted.into_values() {
+            single_format_collection.add_single_formats(data)
         }
 
-        Ok(result)
+        Ok(single_format_collection)
     }
 }
 

crunchy-cli-core/src/download/command.rs

@@ -1,11 +1,13 @@
 use crate::download::filter::DownloadFilter;
 use crate::utils::context::Context;
+use crate::utils::download::{DownloadBuilder, DownloadFormat};
 use crate::utils::ffmpeg::FFmpegPreset;
 use crate::utils::filter::Filter;
-use crate::utils::format::formats_visual_output;
+use crate::utils::format::{Format, SingleFormat};
 use crate::utils::log::progress;
 use crate::utils::os::{free_file, has_ffmpeg};
 use crate::utils::parse::parse_url;
+use crate::utils::video::variant_data_from_stream;
 use crate::Execute;
 use anyhow::bail;
 use anyhow::Result;
@@ -116,19 +118,35 @@ impl Execute for Download {
 
         for (i, (media_collection, url_filter)) in parsed_urls.into_iter().enumerate() {
             let progress_handler = progress!("Fetching series details");
-            let download_formats = DownloadFilter::new(url_filter, self.clone())
+            let single_format_collection = DownloadFilter::new(url_filter, self.clone())
                 .visit(media_collection)
                 .await?;
 
-            if download_formats.is_empty() {
+            if single_format_collection.is_empty() {
                 progress_handler.stop(format!("Skipping url {} (no matching videos found)", i + 1));
                 continue;
             }
             progress_handler.stop(format!("Loaded series information for url {}", i + 1));
 
-            formats_visual_output(download_formats.iter().map(|(_, f)| f).collect());
+            single_format_collection.full_visual_output();
+
+            let download_builder = DownloadBuilder::new()
+                .default_subtitle(self.subtitle.clone())
+                .output_format(if self.output == "-" {
+                    Some("mpegts".to_string())
+                } else {
+                    None
+                });
+
+            for mut single_formats in single_format_collection.into_iter() {
+                // the vec contains always only one item
+                let single_format = single_formats.remove(0);
+
+                let (download_format, format) = get_format(&self, &single_format).await?;
+
+                let mut downloader = download_builder.clone().build();
+                downloader.add_format(download_format);
 
-            for (downloader, format) in download_formats {
                 let formatted_path = format.format_path((&self.output).into(), true);
                 let (path, changed) = free_file(formatted_path.clone());
 
@@ -149,3 +167,48 @@ impl Execute for Download {
         Ok(())
     }
 }
+
+async fn get_format(
+    download: &Download,
+    single_format: &SingleFormat,
+) -> Result<(DownloadFormat, Format)> {
+    let stream = single_format.stream().await?;
+    let Some((video, audio)) = variant_data_from_stream(&stream, &download.resolution).await? else {
+        if single_format.is_episode() {
+            bail!(
+                "Resolution ({}) is not available for episode {} ({}) of {} season {}",
+                download.resolution,
+                single_format.episode_number,
+                single_format.title,
+                single_format.series_name,
+                single_format.season_number,
+            )
+        } else {
+            bail!(
+                "Resolution ({}) is not available for {} ({})",
+                download.resolution,
+                single_format.source_type(),
+                single_format.title
+            )
+        }
+    };
+
+    let subtitle = if let Some(subtitle_locale) = &download.subtitle {
+        stream.subtitles.get(subtitle_locale).map(|s| s.clone())
+    } else {
+        None
+    };
+
+    let download_format = DownloadFormat {
+        video: (video.clone(), single_format.audio.clone()),
+        audios: vec![(audio, single_format.audio.clone())],
+        subtitles: subtitle.clone().map_or(vec![], |s| vec![s]),
+    };
+    let format = Format::from_single_formats(vec![(
+        single_format.clone(),
+        video,
+        subtitle.map_or(vec![], |s| vec![s]),
+    )]);
+
+    Ok((download_format, format))
+}

crunchy-cli-core/src/download/filter.rs

@@ -1,22 +1,12 @@
 use crate::download::Download;
-use crate::utils::download::{DownloadBuilder, DownloadFormat, Downloader};
 use crate::utils::filter::Filter;
-use crate::utils::format::{Format, SingleFormat};
+use crate::utils::format::{Format, SingleFormat, SingleFormatCollection};
 use crate::utils::parse::UrlFilter;
-use crate::utils::video::variant_data_from_stream;
 use anyhow::{bail, Result};
-use crunchyroll_rs::media::{Subtitle, VariantData};
 use crunchyroll_rs::{Concert, Episode, Movie, MovieListing, MusicVideo, Season, Series};
 use log::{error, warn};
 use std::collections::HashMap;
 
-pub(crate) struct FilterResult {
-    format: SingleFormat,
-    video: VariantData,
-    audio: VariantData,
-    subtitle: Option<Subtitle>,
-}
-
 pub(crate) struct DownloadFilter {
     url_filter: UrlFilter,
     download: Download,
@@ -37,8 +27,8 @@ impl DownloadFilter {
 
 #[async_trait::async_trait]
 impl Filter for DownloadFilter {
-    type T = FilterResult;
-    type Output = (Downloader, Format);
+    type T = SingleFormat;
+    type Output = SingleFormatCollection;
 
     async fn visit_series(&mut self, series: Series) -> Result<Vec<Season>> {
         // `series.audio_locales` isn't always populated b/c of crunchyrolls api. so check if the
@@ -165,32 +155,6 @@ impl Filter for DownloadFilter {
             }
         }
 
-        // get the correct video stream
-        let stream = episode.streams().await?;
-        let (video, audio) = if let Some((video, audio)) =
-            variant_data_from_stream(&stream, &self.download.resolution).await?
-        {
-            (video, audio)
-        } else {
-            bail!(
-                "Resolution ({}) is not available for episode {} ({}) of {} season {}",
-                self.download.resolution,
-                episode.episode_number,
-                episode.title,
-                episode.series_title,
-                episode.season_number,
-            )
-        };
-
-        // it is assumed that the subtitle, if requested, exists b/c the subtitle check above must
-        // be passed to reach this condition.
-        // the check isn't done in this if block to reduce unnecessary fetching of the stream
-        let subtitle = if let Some(subtitle_locale) = &self.download.subtitle {
-            stream.subtitles.get(subtitle_locale).map(|s| s.clone())
-        } else {
-            None
-        };
-
         // get the relative episode number. only done if the output string has the pattern to include
         // the relative episode number as this requires some extra fetching
         let relative_episode_number = if Format::has_relative_episodes_fmt(&self.download.output) {
@@ -225,17 +189,17 @@ impl Filter for DownloadFilter {
             None
         };
 
-        Ok(Some(FilterResult {
-            format: SingleFormat::new_from_episode(
-                &episode,
-                &video,
-                subtitle.clone().map_or(vec![], |s| vec![s.locale]),
-                relative_episode_number.map(|n| n as u32),
-            ),
-            video,
-            audio,
-            subtitle,
-        }))
+        Ok(Some(SingleFormat::new_from_episode(
+            episode.clone(),
+            self.download.subtitle.clone().map_or(vec![], |s| {
+                if episode.subtitle_locales.contains(&s) {
+                    vec![s]
+                } else {
+                    vec![]
+                }
+            }),
+            relative_episode_number.map(|n| n as u32),
+        )))
     }
 
     async fn visit_movie_listing(&mut self, movie_listing: MovieListing) -> Result<Vec<Movie>> {
@@ -243,113 +207,24 @@ impl Filter for DownloadFilter {
     }
 
     async fn visit_movie(&mut self, movie: Movie) -> Result<Option<Self::T>> {
-        let stream = movie.streams().await?;
-        let (video, audio) = if let Some((video, audio)) =
-            variant_data_from_stream(&stream, &self.download.resolution).await?
-        {
-            (video, audio)
-        } else {
-            bail!(
-                "Resolution ({}) of movie '{}' is not available",
-                self.download.resolution,
-                movie.title
-            )
-        };
-        let subtitle = if let Some(subtitle_locale) = &self.download.subtitle {
-            let Some(subtitle) = stream.subtitles.get(subtitle_locale) else {
-                error!(
-                    "Movie '{}' has no {} subtitles",
-                    movie.title,
-                    subtitle_locale
-                );
-                return Ok(None)
-            };
-            Some(subtitle.clone())
-        } else {
-            None
-        };
-
-        Ok(Some(FilterResult {
-            format: SingleFormat::new_from_movie(
-                &movie,
-                &video,
-                subtitle.clone().map_or(vec![], |s| vec![s.locale]),
-            ),
-            video,
-            audio,
-            subtitle,
-        }))
+        Ok(Some(SingleFormat::new_from_movie(movie, vec![])))
     }
 
     async fn visit_music_video(&mut self, music_video: MusicVideo) -> Result<Option<Self::T>> {
-        let stream = music_video.streams().await?;
-        let (video, audio) = if let Some((video, audio)) =
-            variant_data_from_stream(&stream, &self.download.resolution).await?
-        {
-            (video, audio)
-        } else {
-            bail!(
-                "Resolution ({}) of music video {} is not available",
-                self.download.resolution,
-                music_video.title
-            )
-        };
-
-        Ok(Some(FilterResult {
-            format: SingleFormat::new_from_music_video(&music_video, &video),
-            video,
-            audio,
-            subtitle: None,
-        }))
+        Ok(Some(SingleFormat::new_from_music_video(music_video)))
     }
 
     async fn visit_concert(&mut self, concert: Concert) -> Result<Option<Self::T>> {
-        let stream = concert.streams().await?;
-        let (video, audio) = if let Some((video, audio)) =
-            variant_data_from_stream(&stream, &self.download.resolution).await?
-        {
-            (video, audio)
-        } else {
-            bail!(
-                "Resolution ({}) of music video {} is not available",
-                self.download.resolution,
-                concert.title
-            )
-        };
-
-        Ok(Some(FilterResult {
-            format: SingleFormat::new_from_concert(&concert, &video),
-            video,
-            audio,
-            subtitle: None,
-        }))
+        Ok(Some(SingleFormat::new_from_concert(concert)))
     }
 
-    async fn finish(self, mut input: Vec<Self::T>) -> Result<Vec<Self::Output>> {
-        let mut result = vec![];
-        input.sort_by(|a, b| {
-            a.format
-                .sequence_number
-                .total_cmp(&b.format.sequence_number)
-        });
+    async fn finish(self, input: Vec<Self::T>) -> Result<Self::Output> {
+        let mut single_format_collection = SingleFormatCollection::new();
+
         for data in input {
-            let mut download_builder =
-                DownloadBuilder::new().default_subtitle(self.download.subtitle.clone());
-            // set the output format to mpegts / mpeg transport stream if the output file is stdout.
-            // mp4 isn't used here as the output file must be readable which isn't possible when
-            // writing to stdout
-            if self.download.output == "-" {
-                download_builder = download_builder.output_format(Some("mpegts".to_string()))
-            }
-            let mut downloader = download_builder.build();
-            downloader.add_format(DownloadFormat {
-                video: (data.video, data.format.audio.clone()),
-                audios: vec![(data.audio, data.format.audio.clone())],
-                subtitles: data.subtitle.map_or(vec![], |s| vec![s]),
-            });
-            result.push((downloader, Format::from_single_formats(vec![data.format])))
+            single_format_collection.add_single_formats(vec![data])
         }
 
-        Ok(result)
+        Ok(single_format_collection)
     }
 }

crunchy-cli-core/src/utils/download.rs

@@ -38,7 +38,7 @@ impl MergeBehavior {
     }
 }
 
-#[derive(derive_setters::Setters)]
+#[derive(Clone, derive_setters::Setters)]
 pub struct DownloadBuilder {
     ffmpeg_preset: FFmpegPreset,
     default_subtitle: Option<Locale>,

crunchy-cli-core/src/utils/filter.rs

@@ -18,7 +18,7 @@ pub trait Filter {
     async fn visit_music_video(&mut self, music_video: MusicVideo) -> Result<Option<Self::T>>;
     async fn visit_concert(&mut self, concert: Concert) -> Result<Option<Self::T>>;
 
-    async fn visit(mut self, media_collection: MediaCollection) -> Result<Vec<Self::Output>>
+    async fn visit(mut self, media_collection: MediaCollection) -> Result<Self::Output>
     where
         Self: Send + Sized,
     {
@@ -80,7 +80,7 @@ pub trait Filter {
         self.finish(result).await
     }
 
-    async fn finish(self, input: Vec<Self::T>) -> Result<Vec<Self::Output>>;
+    async fn finish(self, input: Vec<Self::T>) -> Result<Self::Output>;
 }
 
 /// Remove all duplicates from a [`Vec`].

crunchy-cli-core/src/utils/format.rs

@@ -1,9 +1,12 @@
 use crate::utils::filter::real_dedup_vec;
 use crate::utils::log::tab_info;
 use crate::utils::os::is_special_file;
-use crunchyroll_rs::media::{Resolution, VariantData};
-use crunchyroll_rs::{Concert, Episode, Locale, Movie, MusicVideo};
+use anyhow::Result;
+use chrono::Duration;
+use crunchyroll_rs::media::{Resolution, Stream, Subtitle, VariantData};
+use crunchyroll_rs::{Concert, Episode, Locale, MediaCollection, Movie, MusicVideo};
 use log::{debug, info};
+use std::cmp::Ordering;
 use std::collections::BTreeMap;
 use std::path::{Path, PathBuf};
 
@@ -15,9 +18,6 @@ pub struct SingleFormat {
     pub audio: Locale,
     pub subtitles: Vec<Locale>,
 
-    pub resolution: Resolution,
-    pub fps: f64,
-
     pub series_id: String,
     pub series_name: String,
 
@@ -29,12 +29,15 @@ pub struct SingleFormat {
     pub episode_number: String,
     pub sequence_number: f32,
     pub relative_episode_number: Option<u32>,
+
+    pub duration: Duration,
+
+    source: MediaCollection,
 }
 
 impl SingleFormat {
     pub fn new_from_episode(
-        episode: &Episode,
-        video: &VariantData,
+        episode: Episode,
         subtitles: Vec<Locale>,
         relative_episode_number: Option<u32>,
     ) -> Self {
@@ -43,8 +46,6 @@ impl SingleFormat {
             description: episode.description.clone(),
             audio: episode.audio_locale.clone(),
             subtitles,
-            resolution: video.resolution.clone(),
-            fps: video.fps,
             series_id: episode.series_id.clone(),
             series_name: episode.series_title.clone(),
             season_id: episode.season_id.clone(),
@@ -58,17 +59,17 @@ impl SingleFormat {
             },
             sequence_number: episode.sequence_number,
             relative_episode_number,
+            duration: episode.duration,
+            source: episode.into(),
         }
     }
 
-    pub fn new_from_movie(movie: &Movie, video: &VariantData, subtitles: Vec<Locale>) -> Self {
+    pub fn new_from_movie(movie: Movie, subtitles: Vec<Locale>) -> Self {
         Self {
             title: movie.title.clone(),
             description: movie.description.clone(),
             audio: Locale::ja_JP,
             subtitles,
-            resolution: video.resolution.clone(),
-            fps: video.fps,
             series_id: movie.movie_listing_id.clone(),
             series_name: movie.movie_listing_title.clone(),
             season_id: movie.movie_listing_id.clone(),
@@ -78,17 +79,17 @@ impl SingleFormat {
             episode_number: "1".to_string(),
             sequence_number: 1.0,
             relative_episode_number: Some(1),
+            duration: movie.duration,
+            source: movie.into(),
         }
     }
 
-    pub fn new_from_music_video(music_video: &MusicVideo, video: &VariantData) -> Self {
+    pub fn new_from_music_video(music_video: MusicVideo) -> Self {
         Self {
             title: music_video.title.clone(),
             description: music_video.description.clone(),
             audio: Locale::ja_JP,
             subtitles: vec![],
-            resolution: video.resolution.clone(),
-            fps: video.fps,
             series_id: music_video.id.clone(),
             series_name: music_video.title.clone(),
             season_id: music_video.id.clone(),
@@ -98,17 +99,17 @@ impl SingleFormat {
             episode_number: "1".to_string(),
             sequence_number: 1.0,
             relative_episode_number: Some(1),
+            duration: music_video.duration,
+            source: music_video.into(),
         }
     }
 
-    pub fn new_from_concert(concert: &Concert, video: &VariantData) -> Self {
+    pub fn new_from_concert(concert: Concert) -> Self {
         Self {
             title: concert.title.clone(),
             description: concert.description.clone(),
             audio: Locale::ja_JP,
             subtitles: vec![],
-            resolution: video.resolution.clone(),
-            fps: video.fps,
             series_id: concert.id.clone(),
             series_name: concert.title.clone(),
             season_id: concert.id.clone(),
@@ -118,8 +119,145 @@ impl SingleFormat {
             episode_number: "1".to_string(),
             sequence_number: 1.0,
             relative_episode_number: Some(1),
+            duration: concert.duration,
+            source: concert.into(),
         }
     }
+
+    pub async fn stream(&self) -> Result<Stream> {
+        let stream = match &self.source {
+            MediaCollection::Episode(e) => e.streams().await?,
+            MediaCollection::Movie(m) => m.streams().await?,
+            MediaCollection::MusicVideo(mv) => mv.streams().await?,
+            MediaCollection::Concert(c) => c.streams().await?,
+            _ => unreachable!(),
+        };
+        Ok(stream)
+    }
+
+    pub fn source_type(&self) -> String {
+        match &self.source {
+            MediaCollection::Episode(_) => "episode",
+            MediaCollection::Movie(_) => "movie",
+            MediaCollection::MusicVideo(_) => "music video",
+            MediaCollection::Concert(_) => "concert",
+            _ => unreachable!(),
+        }
+        .to_string()
+    }
+
+    pub fn is_episode(&self) -> bool {
+        match self.source {
+            MediaCollection::Episode(_) => true,
+            _ => false,
+        }
+    }
+}
+
+struct SingleFormatCollectionEpisodeKey(f32);
+
+impl PartialOrd for SingleFormatCollectionEpisodeKey {
+    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+        self.0.partial_cmp(&other.0)
+    }
+}
+impl Ord for SingleFormatCollectionEpisodeKey {
+    fn cmp(&self, other: &Self) -> Ordering {
+        self.0.total_cmp(&other.0)
+    }
+}
+impl PartialEq for SingleFormatCollectionEpisodeKey {
+    fn eq(&self, other: &Self) -> bool {
+        self.0.eq(&other.0)
+    }
+}
+impl Eq for SingleFormatCollectionEpisodeKey {}
+
+pub struct SingleFormatCollection(
+    BTreeMap<u32, BTreeMap<SingleFormatCollectionEpisodeKey, Vec<SingleFormat>>>,
+);
+
+impl SingleFormatCollection {
+    pub fn new() -> Self {
+        Self(BTreeMap::new())
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.0.is_empty()
+    }
+
+    pub fn add_single_formats(&mut self, single_formats: Vec<SingleFormat>) {
+        let format = single_formats.first().unwrap();
+        self.0
+            .entry(format.season_number)
+            .or_insert(BTreeMap::new())
+            .insert(
+                SingleFormatCollectionEpisodeKey(format.sequence_number),
+                single_formats,
+            );
+    }
+
+    pub fn full_visual_output(&self) {
+        debug!("Series has {} seasons", self.0.len());
+        for (season_number, episodes) in &self.0 {
+            info!(
+                "{} Season {}",
+                episodes
+                    .first_key_value()
+                    .unwrap()
+                    .1
+                    .first()
+                    .unwrap()
+                    .series_name
+                    .clone(),
+                season_number
+            );
+            for (i, (_, formats)) in episodes.iter().enumerate() {
+                let format = formats.first().unwrap();
+                if log::max_level() == log::Level::Debug {
+                    info!(
+                        "{} S{:02}E{:0>2}",
+                        format.title, format.season_number, format.episode_number
+                    )
+                } else {
+                    tab_info!(
+                        "{}. {} » S{:02}E{:0>2}",
+                        i + 1,
+                        format.title,
+                        format.season_number,
+                        format.episode_number
+                    )
+                }
+            }
+        }
+    }
+}
+
+impl IntoIterator for SingleFormatCollection {
+    type Item = Vec<SingleFormat>;
+    type IntoIter = SingleFormatCollectionIterator;
+
+    fn into_iter(self) -> Self::IntoIter {
+        SingleFormatCollectionIterator(self)
+    }
+}
+
+pub struct SingleFormatCollectionIterator(SingleFormatCollection);
+
+impl Iterator for SingleFormatCollectionIterator {
+    type Item = Vec<SingleFormat>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        let Some((_, episodes)) = self.0.0.iter_mut().next() else {
+            return None
+        };
+
+        let value = episodes.pop_first().unwrap().1;
+        if episodes.is_empty() {
+            self.0 .0.pop_first();
+        }
+        Some(value)
+    }
 }
 
 #[derive(Clone)]
@@ -146,28 +284,38 @@ pub struct Format {
 }
 
 impl Format {
-    pub fn from_single_formats(mut single_formats: Vec<SingleFormat>) -> Self {
+    pub fn from_single_formats(
+        mut single_formats: Vec<(SingleFormat, VariantData, Vec<Subtitle>)>,
+    ) -> Self {
         let locales: Vec<(Locale, Vec<Locale>)> = single_formats
             .iter()
-            .map(|sf| (sf.audio.clone(), sf.subtitles.clone()))
+            .map(|(single_format, _, subtitles)| {
+                (
+                    single_format.audio.clone(),
+                    subtitles
+                        .into_iter()
+                        .map(|s| s.locale.clone())
+                        .collect::<Vec<Locale>>(),
+                )
+            })
             .collect();
-        let first = single_formats.remove(0);
+        let (first_format, first_stream, _) = single_formats.remove(0);
 
         Self {
-            title: first.title,
-            description: first.description,
+            title: first_format.title,
+            description: first_format.description,
             locales,
-            resolution: first.resolution,
-            fps: first.fps,
-            series_id: first.series_id,
-            series_name: first.series_name,
-            season_id: first.season_id,
-            season_title: first.season_title,
-            season_number: first.season_number,
-            episode_id: first.episode_id,
-            episode_number: first.episode_number,
-            sequence_number: first.sequence_number,
-            relative_episode_number: first.relative_episode_number,
+            resolution: first_stream.resolution,
+            fps: first_stream.fps,
+            series_id: first_format.series_id,
+            series_name: first_format.series_name,
+            season_id: first_format.season_id,
+            season_title: first_format.season_title,
+            season_number: first_format.season_number,
+            episode_id: first_format.episode_id,
+            episode_number: first_format.episode_number,
+            sequence_number: first_format.sequence_number,
+            relative_episode_number: first_format.relative_episode_number,
         }
     }
 
@@ -262,60 +410,3 @@ impl Format {
         return s.as_ref().contains("{relative_episode_number}");
     }
 }
-
-pub fn formats_visual_output(formats: Vec<&Format>) {
-    if log::max_level() == log::Level::Debug {
-        let seasons = sort_formats_after_seasons(formats);
-        debug!("Series has {} seasons", seasons.len());
-        for (i, season) in seasons.into_iter().enumerate() {
-            info!("Season {}", i + 1);
-            for format in season {
-                info!(
-                    "{}: {}px, {:.02} FPS (S{:02}E{:0>2})",
-                    format.title,
-                    format.resolution,
-                    format.fps,
-                    format.season_number,
-                    format.episode_number,
-                )
-            }
-        }
-    } else {
-        for season in sort_formats_after_seasons(formats) {
-            let first = season.get(0).unwrap();
-            info!("{} Season {}", first.series_name, first.season_number);
-
-            for (i, format) in season.into_iter().enumerate() {
-                tab_info!(
-                    "{}. {} » {}px, {:.2} FPS (S{:02}E{:0>2})",
-                    i + 1,
-                    format.title,
-                    format.resolution,
-                    format.fps,
-                    format.season_number,
-                    format.episode_number
-                )
-            }
-        }
-    }
-}
-
-fn sort_formats_after_seasons(formats: Vec<&Format>) -> Vec<Vec<&Format>> {
-    let mut season_map = BTreeMap::new();
-
-    for format in formats {
-        season_map
-            .entry(format.season_number)
-            .or_insert(vec![])
-            .push(format)
-    }
-
-    season_map
-        .into_values()
-        .into_iter()
-        .map(|mut fmts| {
-            fmts.sort_by(|a, b| a.sequence_number.total_cmp(&b.sequence_number));
-            fmts
-        })
-        .collect()
-}