Add support for ytarchive-0.3.2

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "hoshinova"
-version = "0.2.2"
+version = "0.2.3"
 edition = "2021"
 repository = "https://github.com/HoloArchivists/hoshinova"
 homepage = "https://github.com/HoloArchivists/hoshinova"

Dockerfile

@@ -2,7 +2,7 @@
 FROM node:16 as web-deps
 WORKDIR /src/web
 COPY web/package.json web/yarn.lock ./
-RUN yarn install
+RUN yarn install --frozen-lockfile
 
 # Create base image for building Rust
 FROM rust:1.62-alpine AS rust-build-image
@@ -49,7 +49,7 @@ FROM alpine AS runner
 WORKDIR /app
 RUN set -ex; \
     apk add --no-cache ffmpeg wget unzip; \
-    wget -O /app/ytarchive.zip https://github.com/Kethsar/ytarchive/releases/download/latest/ytarchive_linux_amd64.zip; \
+    wget -O /app/ytarchive.zip https://github.com/Kethsar/ytarchive/releases/download/v0.3.2/ytarchive_linux_amd64.zip; \
     unzip /app/ytarchive.zip -d /usr/local/bin/; \
     rm /app/ytarchive.zip; \
     apk del wget unzip;

src/module/recorder.rs

@@ -398,6 +398,7 @@ pub enum YTAState {
     AlreadyProcessed,
     Ended,
     Interrupted,
+    Errored,
 }
 
 fn strip_ansi(s: &str) -> String {
@@ -459,7 +460,27 @@ impl YTAStatus {
             if let Some(x) = parts.next() {
                 self.total_size = Some(strip_ansi(x.trim()));
             };
-        } else if self.version == None && line.starts_with("ytarchive ") {
+            return;
+        } else if line.starts_with("Audio Fragments: ") {
+            self.state = YTAState::Recording;
+            let mut parts = line.split(';').map(|s| s.split(':').nth(1).unwrap_or(""));
+            if let Some(x) = parts.next() {
+                self.audio_fragments = x.trim().parse().ok();
+            };
+            if let Some(x) = parts.next() {
+                self.total_size = Some(strip_ansi(x.trim()));
+            };
+            return;
+        }
+
+        // New versions of ytarchive prepend a timestamp to the output
+        let line = if self.version == Some("0.3.2".into()) {
+            line[20..].trim()
+        } else {
+            line
+        };
+
+        if self.version == None && line.starts_with("ytarchive ") {
             self.version = Some(strip_ansi(&line[10..]));
         } else if self.video_quality == None && line.starts_with("Selected quality: ") {
             self.video_quality = Some(strip_ansi(&line[18..]));
@@ -483,7 +504,18 @@ impl YTAStatus {
             self.output_file = Some(strip_ansi(&line[12..]));
         } else if line.contains("User Interrupt") {
             self.state = YTAState::Interrupted;
-        } else if line.trim().is_empty() || line.contains("Loaded cookie file") {
+        } else if line.contains("Error retrieving player response")
+            || line.contains("unable to retrieve")
+            || line.contains("error writing the muxcmd file")
+            || line.contains("Something must have gone wrong with ffmpeg")
+            || line.contains("At least one error occurred")
+        {
+            self.state = YTAState::Errored;
+        } else if line.trim().is_empty()
+            || line.contains("Loaded cookie file")
+            || line.starts_with("Video Title: ")
+            || line.starts_with("Channel: ")
+        {
             // Ignore
         } else {
             warn!("Unknown ytarchive output: {}", line);

src/module/scraper.rs

@@ -63,6 +63,14 @@ impl RSS {
         let max_age =
             chrono::Duration::from_std(self.config.read().await.scraper.rss.ignore_older_than)
                 .context("Failed to convert ignore_older_than to chrono::Duration")?;
+        debug!(
+            "Ignoring videos older than {}",
+            max_age
+                .to_std()
+                .map(humantime::format_duration)
+                .map(|s| s.to_string())
+                .unwrap_or_else(|_| "???".into())
+        );
 
         // Fetch the RSS feed
         let url = format!(
@@ -86,16 +94,25 @@ impl RSS {
             .filter_map(move |entry| {
                 let mut scraped = scraped.lock().unwrap();
 
-                // Skip if video has already been scraped, or if it's too old
-                if scraped.contains(&entry.video_id)
-                    || entry.updated < chrono::Utc::now() - max_age
-                    || !channel.filters.iter().any(|filter| {
-                        // Or if the video doesn't match the filters
-                        filter.is_match(&entry.title)
-                            || (channel.match_description
-                                && filter.is_match(&entry.group.description))
-                    })
-                {
+                if scraped.contains(&entry.video_id) {
+                    // Skip if video has already been scraped
+                    debug!("Skipping {}: already scraped", entry.video_id);
+                    return None;
+                } else if entry.updated < chrono::Utc::now() - max_age {
+                    // Or if the video is too old
+                    debug!(
+                        "Skipping {}: too old ({} < {})",
+                        entry.video_id,
+                        entry.updated,
+                        chrono::Utc::now() - max_age
+                    );
+                    return None;
+                } else if !channel.filters.iter().any(|filter| {
+                    filter.is_match(&entry.title)
+                        || (channel.match_description && filter.is_match(&entry.group.description))
+                }) {
+                    // Or if the video doesn't match the filters
+                    debug!("Skipping {}: doesn't match filters", entry.video_id);
                     return None;
                 }
 

src/msgbus.rs

@@ -45,10 +45,10 @@ impl<T: Debug + Clone + Sync> MessageBus<T> {
     /// closed.
     pub async fn start(&mut self) {
         'out: while let Some(BusMessage::Message(msg)) = self.mix_rx.recv().await {
-            for tx in &mut self.mix_tx {
+            for (n, tx) in &mut self.mix_tx.iter().enumerate() {
                 match tx.try_send(msg.clone()) {
                     Err(e) => {
-                        error!("Failed to send message: {}", e);
+                        error!("Failed to send message to queue {}: {}", n, e);
                         break 'out;
                     }
                     _ => (),

web/src/pages/ConfigPage.tsx

@@ -77,7 +77,7 @@ const ConfigPage = () => {
               ? 'Reloading...'
               : isEditable
               ? 'Save and apply'
-              : 'Reload configuration'}
+              : 'Reload from disk'}
           </Button>
           {isEditable ? null : (
             <Text size="sm">

web/src/pages/TasksPage.tsx

@@ -25,7 +25,6 @@ import { closeAllModals, openModal } from '@mantine/modals';
 import { showNotification } from '@mantine/notifications';
 import { useQueryConfig } from '../api/config';
 import { YTAState } from '../bindings/YTAState';
-import { Task } from '../bindings/Task';
 
 const SleepingPanda = React.lazy(() => import('../lotties/SleepingPanda'));
 
@@ -40,7 +39,7 @@ const TaskStateBadge = ({ state }: { state: YTAState }) => (
         ? 'yellow'
         : state === 'Idle' || state === 'AlreadyProcessed' || state === 'Ended'
         ? 'gray'
-        : state === 'Interrupted'
+        : state === 'Interrupted' || state === 'Errored'
         ? 'red'
         : 'violet'
     }