central config + download profiles + filter flags + other stuff

8 files changed, 133 insertions, 51 deletions

diff --git a/.gitignore b/.gitignore
index e65c413..c26e47a 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,2 +1,3 @@
 /target
-/*.json
\ No newline at end of file
+/*.json
+/*.yaml
\ No newline at end of file
diff --git a/scripts/config.ts b/scripts/config.ts
new file mode 100644
index 0000000..1068c92
--- /dev/null
+++ b/scripts/config.ts
@@ -0,0 +1,20 @@
+
+export interface Config {
+    enqueue: EnqueueTask[]
+    ytdlp_flatten: {
+        filters: { [key: string]: string }
+    }
+    ytdlp_download: {
+        output: string,
+        profiles: { [key: string]: string[] }
+    }
+}
+
+export interface EnqueueTask {
+    list_file: string
+    kind: string,
+    interval: number,
+    filter?: string,
+    oneshot?: boolean,
+    data: { [key: string]: unknown }
+}
diff --git a/scripts/enqueue.ts b/scripts/enqueue.ts
index 8d19e53..5dfb6b7 100644
--- a/scripts/enqueue.ts
+++ b/scripts/enqueue.ts
@@ -1,38 +1,59 @@
+import { Config, EnqueueTask } from "./config.ts";
 
 const ws = new WebSocket(Deno.args[0])
-const file = await Deno.readTextFile(Deno.args[1])
-const outdir = Deno.args.length >= 3 ? Deno.args[2] : "."
-const note_filter = Deno.args.length >= 4 ? Deno.args[3] : ""
-const requeue = Deno.env.has("REQUEUE")
+let config: Config = {} as unknown as Config
 
-function run_enqueue() {
-    let kind = "http"
+async function run_enqueue(eqt: EnqueueTask) {
+    const file = await Deno.readTextFile(eqt.list_file)
     for (const line of file.split("\n")) {
         if (!line.trim().length) continue
-        else if (line.startsWith("[") && line.endsWith("]"))
-            kind = line.substring(1, line.length - 1)
-        else {
-            const [name, rest] = line.split("=", 2)
-            const [id, note] = rest.split(";", 2)
-            if (note_filter.length && note != note_filter) continue
-            const key = `${kind}:${id}`;
-            ws.send(JSON.stringify({ t: "metadata", key, data: { output: outdir + "/" + name, title: name } }))
-            ws.send(JSON.stringify({ t: "enqueue", key, ignore_complete: requeue }))
-        }
+        let [name, rest] = line.split("=", 2)
+        let [id, flags_raw] = rest.split(";", 2)
+        let flags = flags_raw.split(" ")
+
+        if (eqt.filter && !flags.includes(eqt.filter)) continue
+        name = name.trim()
+        id = id.trim()
+        flags = flags.filter(e => e.length && e != eqt.filter)
+
+        const key = `${eqt.kind}:${id}`;
+        ws.send(JSON.stringify({
+            t: "metadata", key, data: {
+                ...eqt.data,
+                output: (eqt.data.output ?? ".") + "/" + name,
+                title: name,
+                flags,
+            }
+        }))
+        ws.send(JSON.stringify({ t: "enqueue", key, ignore_complete: !(eqt.oneshot ?? false) }))
     }
     ws.send(JSON.stringify({ t: "save" }))
     console.log("done");
+    setTimeout(() => run_enqueue(eqt), eqt.interval * 1000)
+}
+
+let started = false;
+function start() {
+    started = true
+    for (const t of config.enqueue)
+        run_enqueue(t)
 }
 
 ws.onerror = () => console.error("ws error")
-ws.onclose = () => console.error("ws closed")
+ws.onclose = () => {
+    console.error("ws closed")
+    Deno.exit(1)
+}
 ws.onopen = () => {
     console.log("ws open");
     ws.send(JSON.stringify({ t: "register", name: "enqueuer", task_kinds: [] }))
-    run_enqueue()
 }
 ws.onmessage = ev => {
     if (typeof ev.data != "string") return
     const p = JSON.parse(ev.data)
+    if (p.t == "config") {
+        config = p.config
+        if (!started) start()
+    }
     if (p.t == "error") console.error(`error: ${p.message}`);
 }
diff --git a/scripts/ytdlp_download.ts b/scripts/ytdlp_download.ts
index f82e3b8..760ac26 100644
--- a/scripts/ytdlp_download.ts
+++ b/scripts/ytdlp_download.ts
@@ -1,5 +1,7 @@
+import { Config } from "./config.ts";
 
 const ws = new WebSocket(Deno.args[0])
+let config: Config = {} as unknown as Config
 
 export class TextLineStream extends TransformStream<string, string> {
     line = "";
@@ -28,8 +30,15 @@ function key_to_url(key: string): string {
     throw new Error("unknown kind");
 }
 
-async function do_download(key: string, output: string) {
+async function do_download(key: string, data: { [key: string]: string }) {
     const url = key_to_url(key)
+
+    if (!data.output) throw new Error("no output");
+    const output = data.output as string;
+    if (!data.profile) throw new Error("no profile");
+    const args = config.ytdlp_download.profiles[data.profile as string]
+    if (!args) throw new Error(`unknown profile ${data.profile}`);
+
     await Deno.mkdir(output, { recursive: true })
     const child = new Deno.Command("yt-dlp", {
         args: [
@@ -37,16 +46,7 @@ async function do_download(key: string, output: string) {
             "--progress",
             "--progress-template", "%(progress)j",
             "--newline",
-            "--download-archive", "archive",
-            "-f", "bestvideo+bestaudio",
-            "--embed-metadata",
-            "--embed-thumbnail",
-            "--embed-info-json",
-            "--embed-subs",
-            "--embed-chapters",
-            "--ignore-no-formats",
-            "--remux", "mkv",
-            "-o", "%(id)s",
+            ...args,
             url
         ],
         stdout: "piped",
@@ -68,12 +68,15 @@ async function do_download(key: string, output: string) {
     }
     const status = await child.status
     if (!status.success) throw new Error("download failed");
-
+    ws.send(JSON.stringify({ t: "metadata", key, data: { progress: null, status: null } }))
     ws.send(JSON.stringify({ t: "complete", key }))
 }
 
 ws.onerror = () => console.error("ws error")
-ws.onclose = () => console.error("ws closed")
+ws.onclose = () => {
+    console.error("ws closed")
+    Deno.exit(1)
+}
 ws.onopen = () => {
     console.log("ws open");
     ws.send(JSON.stringify({ t: "register", name: "yt-dlp video downloader", task_kinds: ["youtube"] }))
@@ -82,10 +85,10 @@ ws.onopen = () => {
 ws.onmessage = async ev => {
     if (typeof ev.data != "string") return
     const p = JSON.parse(ev.data)
+    if (p.t == "config") config = p.config
     if (p.t == "error") console.error(`error: ${p.message}`);
     if (p.t == "work") {
-        if (!p.data.output) throw new Error("no output");
-        await do_download(p.key, p.data.output)
+        await do_download(p.key, p.data)
         ws.send(JSON.stringify({ t: "accept" }))
     }
 }
diff --git a/scripts/ytdlp_flatten.ts b/scripts/ytdlp_flatten.ts
index 84bbc8f..6d246da 100644
--- a/scripts/ytdlp_flatten.ts
+++ b/scripts/ytdlp_flatten.ts
@@ -1,6 +1,7 @@
+import { Config } from "./config.ts";
 
 const ws = new WebSocket(Deno.args[0])
-
+let config: Config = {} as unknown as Config
 
 function key_to_url(key: string): [string, string] {
     const [kind, id] = key.split(":", 2)
@@ -8,13 +9,16 @@ function key_to_url(key: string): [string, string] {
     throw new Error("unknown kind");
 }
 
-async function flat_playlist(url: string, kind: string, output: string) {
-    console.log(output, url);
+async function flat_playlist(url: string, kind: string, data: { [key: string]: unknown }) {
+    const flags = (data.flags as string[]) ?? []
+    if (!(flags instanceof Array)) throw new Error("flags is not an array");
+    const filter_parts = flags.map(e => config.ytdlp_flatten.filters[e]).join(" & ")
+    const filter_args = flags.length ? ["--match-filter", filter_parts] : []
     const o = await new Deno.Command("yt-dlp", {
         args: [
             "--flat-playlist",
             "--print-json",
-            "--match-filter", "availability=public & live_status=not_live",
+            ...filter_args,
             url
         ],
         stdout: "piped",
@@ -29,11 +33,11 @@ async function flat_playlist(url: string, kind: string, output: string) {
             t: "metadata",
             key,
             data: {
+                ...data,
                 title: ob.title,
                 subtitle: `by ${ob.playlist_uploader}; duration ${ob.duration_string}`,
-                description: ob.description,
                 thumbnail: ob.thumbnails[0]?.url,
-                output,
+                // description: ob.description,
             }
         }))
         ws.send(JSON.stringify({ t: "enqueue", key }))
@@ -42,7 +46,10 @@ async function flat_playlist(url: string, kind: string, output: string) {
 }
 
 ws.onerror = () => console.error("ws error")
-ws.onclose = () => console.error("ws closed")
+ws.onclose = () => {
+    console.error("ws closed")
+    Deno.exit(1)
+}
 ws.onopen = () => {
     console.log("ws open");
     ws.send(JSON.stringify({ t: "register", name: "yt-dlp playlist flattener", task_kinds: ["youtube-channel"] }))
@@ -51,11 +58,12 @@ ws.onopen = () => {
 ws.onmessage = async ev => {
     if (typeof ev.data != "string") return
     const p = JSON.parse(ev.data)
+    if (p.t == "config") config = p.config
     if (p.t == "error") console.error(`error: ${p.message}`);
     if (p.t == "work") {
         const [outkind, url] = key_to_url(p.key)
         if (!p.data.output) throw new Error("no output");
-        await flat_playlist(url, outkind, p.data.output)
+        await flat_playlist(url, outkind, p.data)
         ws.send(JSON.stringify({ t: "complete", key: p.key }))
         ws.send(JSON.stringify({ t: "save" }))
         ws.send(JSON.stringify({ t: "accept" }))
diff --git a/src/main.rs b/src/main.rs
index e4a17e8..0441cc6 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -4,13 +4,15 @@ pub mod webui;
 pub mod webui_ws;
 pub mod worker_ws;
 
-use anyhow::Result;
+use anyhow::{Result, anyhow};
 use api::{api_complete_json, api_loading_json, api_queue_json};
 use axum::{Router, routing::get};
-use log::{debug, info};
+use log::{debug, error, info};
 use serde_json::{Map, Value};
 use std::{
     collections::{HashMap, HashSet},
+    env::args,
+    path::Path,
     sync::Arc,
     time::Instant,
 };
@@ -40,6 +42,8 @@ pub struct State {
     workers: HashMap<WorkerID, Worker>,
     webui_broadcast: broadcast::Sender<Arc<WebuiEvent>>,
 
+    config: Value,
+
     metadata: HashMap<String, Map<String, Value>>,
     queue: HashSet<String>,
     loading: HashSet<String>,
@@ -50,6 +54,7 @@ pub struct State {
 async fn main() -> Result<()> {
     env_logger::init_from_env("LOG");
     let mut state = State::default();
+    state.load_config().await?;
     state.load().await?;
     let router = Router::new()
         .route("/", get(webui))
@@ -69,6 +74,7 @@ async fn main() -> Result<()> {
 impl Default for State {
     fn default() -> Self {
         Self {
+            config: Default::default(),
             worker_id_counter: Default::default(),
             workers: Default::default(),
             webui_broadcast: broadcast::channel(1024).0,
@@ -81,13 +87,28 @@ impl Default for State {
 }
 
 impl State {
+    pub async fn load_config(&mut self) -> Result<()> {
+        let path = args()
+            .nth(1)
+            .ok_or(anyhow!("first argument is config path"))?;
+
+        self.config = serde_yml::from_str::<serde_json::Value>(&read_to_string(path).await?)?;
+        Ok(())
+    }
     pub async fn load(&mut self) -> Result<()> {
         debug!("loading state");
         let t = Instant::now();
-        self.metadata = serde_json::from_str(&read_to_string("metadata.json").await?)?;
-        self.queue = serde_json::from_str(&read_to_string("queue.json").await?)?;
-        self.complete = serde_json::from_str(&read_to_string("complete.json").await?)?;
-        info!("state loaded (took {:?})", t.elapsed());
+        if AsRef::<Path>::as_ref("metadata.json").exists()
+            && AsRef::<Path>::as_ref("queue.json").exists()
+            && AsRef::<Path>::as_ref("complete.json").exists()
+        {
+            self.metadata = serde_json::from_str(&read_to_string("metadata.json").await?)?;
+            self.queue = serde_json::from_str(&read_to_string("queue.json").await?)?;
+            self.complete = serde_json::from_str(&read_to_string("complete.json").await?)?;
+            info!("state loaded (took {:?})", t.elapsed());
+        } else {
+            error!("some state files are missing, skipping load")
+        }
         Ok(())
     }
     pub async fn save(&mut self) -> Result<()> {
diff --git a/src/webui.rs b/src/webui.rs
index 73add5f..ca314c9 100644
--- a/src/webui.rs
+++ b/src/webui.rs
@@ -46,9 +46,9 @@ pub(crate) async fn webui(S(state): S<Arc<RwLock<State>>>) -> Html<String> {
                     }}
                 }
                 section.tasks {
-                    @Taskbin {title: "Queued", state: "queue", set: &g.queue, default, g }
-                    @Taskbin {title: "Loading", state: "loading", set: &g.loading, default, g }
-                    @Taskbin {title: "Completed", state: "complete", set: &g.complete, default, g }
+                    @Taskbin { title: "Queued", state: "queue", set: &g.queue, default, g }
+                    @Taskbin { title: "Loading", state: "loading", set: &g.loading, default, g }
+                    @Taskbin { title: "Completed", state: "complete", set: &g.complete, default, g }
                 }
             }
         }
diff --git a/src/worker_ws.rs b/src/worker_ws.rs
index 25ec7b0..f038100 100644
--- a/src/worker_ws.rs
+++ b/src/worker_ws.rs
@@ -56,6 +56,9 @@ pub enum WorkerResponse {
         key: String,
         data: Map<String, Value>,
     },
+    Config {
+        config: Value,
+    },
     Error {
         message: String,
     },
@@ -73,6 +76,11 @@ async fn worker_websocket_inner(ws: WebSocket, state: Arc<RwLock<State>>) {
 
     let worker = {
         let mut g = state.write().await;
+        tx.send(WorkerResponse::Config {
+            config: g.config.clone(),
+        })
+        .await
+        .unwrap();
         let id = g.worker_id_counter;
         g.worker_id_counter += 1;
         g.workers.insert(