initial commit

.gitignore

@@ -0,0 +1,2 @@
+*.json
+/target

Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "ytd"
+version = "0.1.0"
+edition = "2024"
+license = "MIT"
+readme = "README.md"
+description = "YouTube daemon for crawling and processing channels with pre-configured commands"
+keywords = ["youtube", "rss", "daemon", "crawler", "aggregator"]
+categories = ["command-line-utilities", "parsing", "value-formatting"]
+repository = "https://codeberg.org/YGGverse/ytd"
+
+[dependencies]
+anyhow = "1.0.102"
+chrono = "0.4.44"
+clap = { version = "4.6", features = ["derive"] }
+env_logger = "0.11.10"
+log = "0.4.29"
+rustypipe = "0.11.4"
+serde = { version = "1.0.228", features = ["derive"] }
+tokio = { version = "1.51.0", features = ["rt-multi-thread"] }
+toml = "1.1.2"
+tracing-subscriber = { version = "0.3.23", features = ["env-filter"] }

README.md

@@ -1,3 +1,25 @@
 # ytd
 
-YouTube daemon for crawling and aggregating channels with pre-configured commands
+YouTube daemon for crawling and processing channels with pre-configured commands
+
+* this tool initially created as a tool for the [Pidpilne](https://codeberg.org/YGGverse/pidpilne) project.
+
+## Install
+
+``` bash
+cargo install ytd
+```
+
+## Usage
+
+Foreground / testing:
+
+``` bash
+RUST_LOG=trace cargo run -- -c example/config.toml
+```
+
+Background / daemon:
+
+``` bash
+RUST_LOG=ytd=warn NO_COLOR=1 /usr/local/bin/ytd -c /etc/config.toml
+```

example/config.toml

@@ -0,0 +1,29 @@
+# Update channels queue, in seconds (activates daemon mode)
+# * unset or comment to run once
+# update = 60
+
+# Channels queue config
+
+[channel.test]
+id = "UCl2mFZoRqjw_ELax4Yisf6w" # channel ID
+is_live = true
+is_short = false
+is_upcoming = false
+
+    # Channel item commands to apply (in order)
+
+    [[channel.test.command]]
+    exec = "/usr/bin/echo {ID}" # Supported macro replacements: * {ID} - parsed item URL
+    stdout_contains = "\n" # Check stdout for containing expected string, optional
+
+    # [[channel.test.command]]
+    # ..
+
+    # [[channel.test.command]]
+    # ..
+
+# [channel.test2]
+# ..
+
+# [channel.test3]
+# ..

src/argument.rs

@@ -0,0 +1,9 @@
+use clap::Parser;
+use std::path::PathBuf;
+
+#[derive(Parser, Debug)]
+#[command(version, about)]
+pub struct Argument {
+    #[arg(short, long)]
+    pub config: PathBuf,
+}

src/config.rs

@@ -0,0 +1,13 @@
+mod channel;
+
+use channel::Channel;
+use serde::Deserialize;
+use std::collections::HashMap;
+
+#[derive(Deserialize)]
+pub struct Config {
+    pub channel: HashMap<String, Channel>,
+    /// Repeat delay in seconds (activates daemon mode)
+    /// * None to run once
+    pub update: Option<u64>,
+}

src/config/channel.rs

@@ -0,0 +1,13 @@
+mod command;
+
+use command::Command;
+use serde::Deserialize;
+
+#[derive(Deserialize)]
+pub struct Channel {
+    pub id: String,
+    pub command: Vec<Command>,
+    pub is_live: bool,
+    pub is_short: bool,
+    pub is_upcoming: bool,
+}

src/config/channel/command.rs

@@ -0,0 +1,7 @@
+use serde::Deserialize;
+
+#[derive(Deserialize, PartialEq, Eq, Hash)]
+pub struct Command {
+    pub exec: String,
+    pub stdout_contains: Option<String>,
+}

src/main.rs

@@ -0,0 +1,93 @@
+mod argument;
+mod config;
+
+use argument::Argument;
+use chrono::Local;
+use clap::Parser;
+use config::Config;
+use log::*;
+use rustypipe::client::RustyPipe;
+use std::{env::var, process::Command, time::Duration};
+
+#[tokio::main]
+async fn main() {
+    if var("RUST_LOG").is_ok() {
+        use tracing_subscriber::{EnvFilter, fmt::*};
+        struct T;
+        impl time::FormatTime for T {
+            fn format_time(&self, w: &mut format::Writer<'_>) -> std::fmt::Result {
+                write!(w, "{}", Local::now())
+            }
+        }
+        fmt()
+            .with_timer(T)
+            .with_env_filter(EnvFilter::from_default_env())
+            .init()
+    }
+
+    let argument = Argument::parse();
+    let config: Config =
+        toml::from_str(&std::fs::read_to_string(&argument.config).unwrap()).unwrap();
+    let update = config.update.map(Duration::from_secs);
+
+    let rp = RustyPipe::new();
+
+    loop {
+        for (c, channel) in &config.channel {
+            match rp.query().channel_videos(&channel.id).await {
+                Ok(result) => {
+                    for item in result.content.items.iter().filter(|i| {
+                        i.is_live == channel.is_live
+                            && i.is_short == channel.is_short
+                            && i.is_upcoming == channel.is_upcoming
+                    }) {
+                        for command in &channel.command {
+                            let cmd = command.exec.replace("{ID}", &item.id);
+                            match Command::new("sh").arg("-c").arg(&cmd).output() {
+                                Ok(response) => {
+                                    if response.status.success() {
+                                        if command.stdout_contains.as_ref().is_none_or(|s| {
+                                            String::from_utf8_lossy(&response.stdout).contains(s)
+                                        }) {
+                                            debug!(
+                                                "command `{cmd}` for channel `{c}` successful: `{:?}`",
+                                                response
+                                            )
+                                        } else {
+                                            warn!(
+                                                "unexpected command `{cmd}` for channel `{c}`: `{:?}`",
+                                                response
+                                            )
+                                        }
+                                    } else {
+                                        warn!(
+                                            "command `{cmd}` for channel `{c}` failed: `{:?}`",
+                                            response
+                                        )
+                                    }
+                                }
+                                Err(e) => {
+                                    warn!("can't execute command `{cmd}` for channel `{c}`: `{e}`")
+                                }
+                            }
+                        }
+                    }
+                }
+                Err(e) => warn!("can't scrape channel `{c}`: `{e}`"),
+            }
+        }
+        match update {
+            Some(duration) => {
+                debug!(
+                    "queue completed; await {} seconds to continue...",
+                    duration.as_secs()
+                );
+                std::thread::sleep(duration)
+            }
+            None => {
+                debug!("all tasks completed; exit.");
+                break;
+            }
+        }
+    }
+}