From c6b9b676a40d39bc37ccf57dbc17f19e313dafc5 Mon Sep 17 00:00:00 2001 From: enki Date: Tue, 19 May 2026 20:13:14 -0700 Subject: [PATCH] add Deluge Web UI integration for publisher Deluge uses a JSON-RPC HTTP API exposed by the Web UI plugin (port 8112), not the daemon binary protocol (port 58846). The integration logs in, fetches completed torrents, filters by configured labels, and enqueues them the same way as the qBittorrent poller. - DelugeConfig in config.rs (enabled, url, password, poll_interval_secs, labels) - src/publisher/deluge.rs: DelugeClient with login/_session_id cookie handling, core.get_torrents_status, label + is_finished filtering - Watcher: deluge_poll_loop alongside qb_poll_loop; each only starts if its respective client is configured/enabled - Example config updated with deluge section and Web UI port note --- deploy/kindexr.example.yaml | 11 ++- src/config.rs | 22 ++++- src/publisher/deluge.rs | 168 ++++++++++++++++++++++++++++++++++++ src/publisher/mod.rs | 1 + src/publisher/watcher.rs | 71 ++++++++++++--- 5 files changed, 257 insertions(+), 16 deletions(-) create mode 100644 src/publisher/deluge.rs diff --git a/deploy/kindexr.example.yaml b/deploy/kindexr.example.yaml index c237a64..e861216 100644 --- a/deploy/kindexr.example.yaml +++ b/deploy/kindexr.example.yaml @@ -84,4 +84,13 @@ publisher: password: "${QBIT_PASSWORD}" poll_interval_secs: 60 categories: - - "publish-nostr" # only publish torrents with this category label + - "publish-nostr" # only publish torrents with this qBittorrent category + + # Deluge Web UI integration (enable the Web UI plugin in Deluge preferences) + deluge: + enabled: false + url: "http://127.0.0.1:8112" # Web UI port (not the daemon port 58846) + password: "${DELUGE_WEB_PASSWORD}" + poll_interval_secs: 60 + labels: + - "publish-nostr" # Deluge Label plugin label to watch diff --git a/src/config.rs b/src/config.rs index 4ce68df..656acd5 100644 --- a/src/config.rs +++ b/src/config.rs @@ -70,6 +70,7 @@ pub struct PublisherConfig { pub publish_delay_secs: u64, pub identity: IdentityConfig, pub qbittorrent: QBittorrentConfig, + pub deluge: DelugeConfig, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -90,6 +91,18 @@ pub struct QBittorrentConfig { pub categories: Vec, } +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct DelugeConfig { + pub enabled: bool, + /// Deluge Web UI URL, e.g. "http://127.0.0.1:8112" + pub url: String, + pub password: String, + pub poll_interval_secs: u64, + /// Deluge Label plugin labels to watch for completed downloads. + /// Leave empty to watch all completed torrents. + pub labels: Vec, +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct NetworkConfig { pub proxy: ProxyConfig, @@ -201,7 +214,14 @@ impl Default for Config { username: "admin".into(), password: String::new(), poll_interval_secs: 60, - categories: vec!["publish-public".into()], + categories: vec!["publish-nostr".into()], + }, + deluge: DelugeConfig { + enabled: false, + url: "http://127.0.0.1:8112".into(), + password: String::new(), + poll_interval_secs: 60, + labels: vec!["publish-nostr".into()], }, }, } diff --git a/src/publisher/deluge.rs b/src/publisher/deluge.rs new file mode 100644 index 0000000..85befc7 --- /dev/null +++ b/src/publisher/deluge.rs @@ -0,0 +1,168 @@ +use anyhow::{bail, Context}; +use reqwest::Client; +use serde::Deserialize; +use serde_json::{json, Value}; +use std::collections::HashMap; +use tracing::debug; + +pub struct DelugeClient { + url: String, + password: String, + http: Client, +} + +#[derive(Debug, Clone)] +pub struct DelugeTorrent { + pub hash: String, + pub name: String, + pub label: String, + pub total_size: i64, + pub save_path: String, + pub files: Vec, + pub trackers: Vec, +} + +#[derive(Debug, Clone)] +pub struct DelugeFile { + pub path: String, + pub size: i64, +} + +#[derive(Deserialize)] +struct RpcResponse { + result: Option, + error: Option, +} + +impl DelugeClient { + pub fn new(url: &str, password: &str) -> Self { + DelugeClient { + url: url.trim_end_matches('/').to_owned(), + password: password.to_owned(), + http: Client::new(), + } + } + + fn endpoint(&self) -> String { + format!("{}/json", self.url) + } + + async fn rpc(&self, session: &str, method: &str, params: Value) -> anyhow::Result { + let body = json!({ + "method": method, + "params": params, + "id": 1, + }); + let resp = self.http + .post(self.endpoint()) + .header("Cookie", format!("_session_id={session}")) + .json(&body) + .send() + .await + .with_context(|| format!("deluge rpc {method}"))?; + + let rpc: RpcResponse = resp.json().await.context("deluge rpc parse")?; + if let Some(err) = rpc.error { + bail!("deluge rpc {method} error: {err}"); + } + Ok(rpc.result.unwrap_or(Value::Null)) + } + + /// Log in, return _session_id cookie value. + async fn login(&self) -> anyhow::Result { + let body = json!({ + "method": "auth.login", + "params": [self.password], + "id": 1, + }); + let resp = self.http + .post(self.endpoint()) + .json(&body) + .send() + .await + .context("deluge login")?; + + // Extract _session_id from Set-Cookie header. + let session = resp.headers() + .get("set-cookie") + .and_then(|v| v.to_str().ok()) + .and_then(|s| { + s.split(';').find_map(|part| { + part.trim().strip_prefix("_session_id=").map(|v| v.to_owned()) + }) + }); + + // Also check the response body — auth.login returns true on success. + let rpc: RpcResponse = resp.json().await.context("deluge login parse")?; + if rpc.result != Some(Value::Bool(true)) { + bail!("deluge login failed — check password"); + } + + session.ok_or_else(|| anyhow::anyhow!("deluge login: no _session_id cookie in response")) + } + + /// Return all completed torrents, optionally filtered by label list. + /// Pass an empty labels slice to return all completed. + pub async fn completed(&self, labels: &[String]) -> anyhow::Result> { + let session = self.login().await?; + + let fields = json!([ + "name", "hash", "label", "save_path", + "total_size", "files", "trackers", "is_finished" + ]); + + // Empty filter dict → all torrents. + let result = self.rpc(&session, "core.get_torrents_status", json!([{}, fields])).await?; + + let map = match result.as_object() { + Some(m) => m, + None => return Ok(vec![]), + }; + + let mut out = Vec::new(); + for (_hash, info) in map { + // Only finished torrents. + if info.get("is_finished").and_then(|v| v.as_bool()) != Some(true) { + continue; + } + + let hash = info["hash"].as_str().unwrap_or("").to_lowercase(); + let name = info["name"].as_str().unwrap_or("").to_owned(); + let label = info["label"].as_str().unwrap_or("").to_owned(); + + // Label filter — skip if we have a filter and this torrent doesn't match. + if !labels.is_empty() && !labels.iter().any(|l| l == &label) { + continue; + } + + let save_path = info["save_path"].as_str().unwrap_or("").to_owned(); + let total_size = info["total_size"].as_i64().unwrap_or(0); + + let files: Vec = info["files"] + .as_array() + .unwrap_or(&vec![]) + .iter() + .map(|f| DelugeFile { + path: f["path"].as_str().unwrap_or("").to_owned(), + size: f["size"].as_i64().unwrap_or(0), + }) + .collect(); + + let trackers: Vec = info["trackers"] + .as_array() + .unwrap_or(&vec![]) + .iter() + .filter_map(|t| t["url"].as_str().map(|s| s.to_owned())) + .collect(); + + if hash.is_empty() || name.is_empty() { + continue; + } + + out.push(DelugeTorrent { hash, name, label, total_size, save_path, files, trackers }); + } + + debug!("deluge: {} completed torrents (after label filter)", out.len()); + Ok(out) + } +} diff --git a/src/publisher/mod.rs b/src/publisher/mod.rs index e9bafe8..7940d3c 100644 --- a/src/publisher/mod.rs +++ b/src/publisher/mod.rs @@ -1,2 +1,3 @@ +pub mod deluge; pub mod qbittorrent; pub mod watcher; diff --git a/src/publisher/watcher.rs b/src/publisher/watcher.rs index 2dd6967..ca79b47 100644 --- a/src/publisher/watcher.rs +++ b/src/publisher/watcher.rs @@ -11,6 +11,7 @@ use crate::{ nostr::{signer::Signer, writer}, }; +use super::deluge::{DelugeClient, DelugeTorrent}; use super::qbittorrent::{QbClient, QbTorrent}; pub struct Watcher { @@ -34,15 +35,24 @@ impl Watcher { pub fn start(self) { let arc = Arc::new(self); - // Task 1: poll qBittorrent and enqueue newly completed torrents. - let w1 = arc.clone(); - tokio::spawn(async move { w1.poll_loop().await }); - // Task 2: drain the queue when delays expire. - let w2 = arc.clone(); - tokio::spawn(async move { w2.drain_loop().await }); + // Poll qBittorrent for newly completed torrents. + if arc.cfg.publisher.qbittorrent.poll_interval_secs > 0 + && !arc.cfg.publisher.qbittorrent.url.is_empty() + { + let w = arc.clone(); + tokio::spawn(async move { w.qb_poll_loop().await }); + } + // Poll Deluge for newly completed torrents. + if arc.cfg.publisher.deluge.enabled { + let w = arc.clone(); + tokio::spawn(async move { w.deluge_poll_loop().await }); + } + // Drain the publish queue when delays expire. + let w = arc.clone(); + tokio::spawn(async move { w.drain_loop().await }); } - async fn poll_loop(&self) { + async fn qb_poll_loop(&self) { let interval = Duration::from_secs(self.cfg.publisher.qbittorrent.poll_interval_secs.max(10)); let qb = QbClient::new( &self.cfg.publisher.qbittorrent.url, @@ -56,8 +66,8 @@ impl Watcher { match qb.completed_in_categories(categories).await { Ok(torrents) => { for t in torrents { - if let Err(e) = self.maybe_enqueue(&qb, &t, delay).await { - warn!(hash = t.hash, "enqueue failed: {e}"); + if let Err(e) = self.enqueue_qb(&t, delay).await { + warn!(hash = t.hash, "qb enqueue failed: {e}"); } } } @@ -67,18 +77,51 @@ impl Watcher { } } - async fn maybe_enqueue(&self, _qb: &QbClient, t: &QbTorrent, delay_secs: u64) -> anyhow::Result<()> { - let hash = t.hash.to_lowercase(); + async fn deluge_poll_loop(&self) { + let cfg = &self.cfg.publisher.deluge; + let interval = Duration::from_secs(cfg.poll_interval_secs.max(10)); + let client = DelugeClient::new(&cfg.url, &cfg.password); + let labels = &cfg.labels; + let delay = self.cfg.publisher.publish_delay_secs; - // Skip if already in DB or queue. + info!(url = %cfg.url, "deluge poller started"); + loop { + match client.completed(labels).await { + Ok(torrents) => { + for t in torrents { + if let Err(e) = self.enqueue_deluge(&t, delay).await { + warn!(hash = t.hash, "deluge enqueue failed: {e}"); + } + } + } + Err(e) => warn!("deluge poll failed: {e}"), + } + tokio::time::sleep(interval).await; + } + } + + async fn enqueue_qb(&self, t: &QbTorrent, delay_secs: u64) -> anyhow::Result<()> { + let hash = t.hash.to_lowercase(); if db::is_queued_or_published(&self.pool, &hash).await? { return Ok(()); } - let now = now_secs(); let scheduled_at = now + delay_secs as i64; db::enqueue(&self.pool, &hash, &t.name, &t.category, None, now, scheduled_at).await?; - info!(hash = %hash, title = %t.name, delay_secs, "queued for publishing"); + info!(hash = %hash, title = %t.name, delay_secs, "queued for publishing (qbittorrent)"); + Ok(()) + } + + async fn enqueue_deluge(&self, t: &DelugeTorrent, delay_secs: u64) -> anyhow::Result<()> { + let hash = t.hash.to_lowercase(); + if db::is_queued_or_published(&self.pool, &hash).await? { + return Ok(()); + } + let now = now_secs(); + let scheduled_at = now + delay_secs as i64; + // Use the label as the category (same concept, different client name). + db::enqueue(&self.pool, &hash, &t.name, &t.label, None, now, scheduled_at).await?; + info!(hash = %hash, title = %t.name, delay_secs, "queued for publishing (deluge)"); Ok(()) }