From 1a8ed911d5312c252c9ef73ea572f7aa1418acf8 Mon Sep 17 00:00:00 2001 From: cwj3rcb Date: Mon, 16 Feb 2026 15:37:06 -0600 Subject: [PATCH] Downloading 99% complete --- src/core/downloads.rs | 502 ++++++++++++++++++++++++++---------------- 1 file changed, 317 insertions(+), 185 deletions(-) diff --git a/src/core/downloads.rs b/src/core/downloads.rs index c14b87b..da5ec3e 100644 --- a/src/core/downloads.rs +++ b/src/core/downloads.rs @@ -7,9 +7,13 @@ use std::{ sync::{mpsc, Arc, Mutex}, thread, time::{Duration, Instant}, + net::TcpStream, + io::{Read, Write}, }; -use crate::core::{config::Config, jellyfin::{JellyfinClient, JfItem, LibraryQuery, PagedItems}, paths}; +use serde_json::{json, Value}; + +use crate::core::{config::Config, jellyfin::{JellyfinClient, JfItem, LibraryQuery}, paths}; #[derive(Clone, Debug)] pub enum DownloadKind { @@ -37,8 +41,14 @@ pub struct DownloadItem { pub final_rel_path: PathBuf, // relative to downloads root pub status: DownloadStatus, pub added_at: Instant, + pub bytes_downloaded: u64, pub bytes_total: Option, + pub download_speed: u64, // bytes/sec + pub eta_secs: Option, // seconds remaining + + // aria2 + pub gid: Option, } #[derive(Debug)] @@ -52,7 +62,7 @@ pub enum DownloadCommand { #[derive(Default)] pub struct DownloadState { pub queue: VecDeque, - pub active_pid: Option, + pub active_gid: Option, pub last_message: Option, } @@ -231,6 +241,9 @@ fn make_download_item_movie(item: &JfItem) -> Result { added_at: Instant::now(), bytes_downloaded: 0, bytes_total: None, + download_speed: 0, + eta_secs: None, + gid: None, }) } @@ -240,24 +253,40 @@ fn make_download_item_episode(item: &JfItem) -> Result { let season_no = item.parent_index_number; let ep_no = item.index_number; - let season_folder = season_no.map(|n| format!("Season-{}", n)).unwrap_or_else(|| "Season-0".into()); + let season_folder = season_no + .map(|n| format!("Season-{}", n)) + .unwrap_or_else(|| "Season-0".into()); + let title_s = sanitize_component(&item.name); + let filename = match (season_no, ep_no) { (Some(s), Some(e)) => format!("S{:02}E{:02} - {title_s}.watch", s, e), _ => format!("{title_s}.watch"), }; - let rel = PathBuf::from("shows").join(series_s).join(season_folder).join(filename); + let rel = PathBuf::from("shows") + .join(series_s) + .join(season_folder) + .join(filename); Ok(DownloadItem { item_id: item.id.clone(), title: format!("{series} - {}", item.name), - kind: DownloadKind::Episode { series, season: season_no, episode: ep_no }, + kind: DownloadKind::Episode { + series, + season: season_no, + episode: ep_no, + }, final_rel_path: rel, status: DownloadStatus::Queued, added_at: Instant::now(), + bytes_downloaded: 0, bytes_total: None, + download_speed: 0, + eta_secs: None, + + gid: None, }) } @@ -266,6 +295,7 @@ fn make_download_item_track(item: &JfItem) -> Result { .album_artist .clone() .or_else(|| item.artists.get(0).cloned()); + let album = item.album.clone(); let track_no = item.index_number; @@ -278,23 +308,36 @@ fn make_download_item_track(item: &JfItem) -> Result { None => format!("{title_s}.listen"), }; - let rel = PathBuf::from("music").join(artist_s).join(album_s).join(filename); + let rel = PathBuf::from("music") + .join(artist_s) + .join(album_s) + .join(filename); Ok(DownloadItem { item_id: item.id.clone(), title: item.name.clone(), - kind: DownloadKind::Track { artist, album, track_no }, + kind: DownloadKind::Track { + artist, + album, + track_no, + }, final_rel_path: rel, status: DownloadStatus::Queued, added_at: Instant::now(), + bytes_downloaded: 0, bytes_total: None, + download_speed: 0, + eta_secs: None, + + gid: None, }) } fn make_download_item_other(item: &JfItem) -> Result { let name = sanitize_component(&item.name); let rel = PathBuf::from("other").join(format!("{name}.watch")); + Ok(DownloadItem { item_id: item.id.clone(), title: item.name.clone(), @@ -302,11 +345,17 @@ fn make_download_item_other(item: &JfItem) -> Result { final_rel_path: rel, status: DownloadStatus::Queued, added_at: Instant::now(), + bytes_downloaded: 0, bytes_total: None, + download_speed: 0, + eta_secs: None, + + gid: None, }) } + fn tmp_dir_of_item(it: &DownloadItem) -> Result { temp_dir_for(it) } @@ -350,7 +399,11 @@ fn find_real_downloaded_file(tmp_dir: &Path) -> Result { fn worker_loop(cfg: Config, state: Arc>, rx: mpsc::Receiver) { let _ = ensure_dirs(); - let jf = match JellyfinClient::from_config(&cfg) { + let token = cfg.access_token.clone().unwrap_or_default(); + + let mut shutdown = false; + + let mut aria2_daemon = match start_aria2_daemon() { Ok(c) => c, Err(e) => { if let Ok(mut st) = state.lock() { @@ -360,20 +413,19 @@ fn worker_loop(cfg: Config, state: Arc>, rx: mpsc::Receiver } }; - let token = cfg.access_token.clone().unwrap_or_default(); - - let mut shutdown = false; - let mut current_child: Option = None; - while !shutdown { - // handle commands (non-blocking) + // ------------------------- + // 1) Handle incoming commands + // ------------------------- while let Ok(cmd) = rx.try_recv() { match cmd { DownloadCommand::Enqueue(mut items) => { if let Ok(mut st) = state.lock() { // dedupe by item_id + final_rel_path - for mut it in items.drain(..) { - let already = st.queue.iter().any(|q| q.item_id == it.item_id && q.final_rel_path == it.final_rel_path); + for it in items.drain(..) { + let already = st.queue.iter().any(|q| { + q.item_id == it.item_id && q.final_rel_path == it.final_rel_path + }); if !already { st.queue.push_back(it); } @@ -381,31 +433,36 @@ fn worker_loop(cfg: Config, state: Arc>, rx: mpsc::Receiver st.last_message = Some("queued".into()); } } + DownloadCommand::Cancel { index } => { - // Cancel active (index 0) or any queued item. if index == 0 { - // 1) Kill aria2c - if let Some(child) = current_child.as_mut() { - let _ = child.kill(); + // cancel the active download (front) + let gid = state.lock().ok().and_then(|s| s.active_gid.clone()); + + if let Some(g) = gid { + let _ = aria2_rpc_call("aria2.forceRemove", json!([g.clone()])); + let _ = aria2_rpc_call("aria2.removeDownloadResult", json!([g])); } - // 2) Mark status + delete temp dir for the active item - let active_it = { - let st = state.lock().ok(); - st.and_then(|s| s.queue.front().cloned()) - }; - + // mark cancelled + delete temp + let active_it = state.lock().ok().and_then(|s| s.queue.front().cloned()); if let Some(it) = active_it { if let Ok(mut st) = state.lock() { if let Some(front) = st.queue.front_mut() { front.status = DownloadStatus::Cancelled; } st.last_message = Some("cancelled active download".into()); + st.active_gid = None; } let _ = delete_temp_for(&it); } + + // remove from queue + if let Ok(mut st) = state.lock() { + let _ = st.queue.pop_front(); + } } else { - // queued cancellation + // cancel queued item (not active) if let Ok(mut st) = state.lock() { if index < st.queue.len() { let mut v: Vec<_> = st.queue.drain(..).collect(); @@ -420,6 +477,7 @@ fn worker_loop(cfg: Config, state: Arc>, rx: mpsc::Receiver } } } + DownloadCommand::RemoveQueued { index } => { if let Ok(mut st) = state.lock() { if index > 0 && index < st.queue.len() { @@ -428,46 +486,61 @@ fn worker_loop(cfg: Config, state: Arc>, rx: mpsc::Receiver it.status = DownloadStatus::Cancelled; let _ = delete_temp_for(&it); st.queue = v.into(); + st.last_message = Some("removed from queue".into()); } } } + DownloadCommand::Shutdown => { shutdown = true; } } } - // if no active child, start next queued - let next = { - let mut st = match state.lock() { + if shutdown { + break; + } + + // ------------------------- + // 2) If nothing active, start next + // ------------------------- + let (has_queue, has_active_gid, next_item) = { + let st = match state.lock() { Ok(s) => s, Err(_) => break, }; - if st.queue.is_empty() { - None - } else { - // Ensure first is queued; we keep active at front with status Downloading/Finalizing - Some(st.queue.front().cloned().unwrap()) - } + + let has_queue = !st.queue.is_empty(); + let has_active_gid = st.active_gid.is_some(); + let next_item = st.queue.front().cloned(); + + (has_queue, has_active_gid, next_item) }; - if current_child.is_none() { - if let Some(mut it) = next { + if !has_queue { + thread::sleep(Duration::from_millis(120)); + continue; + } + + if !has_active_gid { + if let Some(it) = next_item { // mark downloading - { - if let Ok(mut st) = state.lock() { - if let Some(front) = st.queue.front_mut() { - front.status = DownloadStatus::Downloading; - } + if let Ok(mut st) = state.lock() { + if let Some(front) = st.queue.front_mut() { + front.status = DownloadStatus::Downloading; } } - match run_one_download(&cfg.base_url, &token, &it, &state) { - Ok(child) => { + match aria2_add_download(&cfg.base_url, &token, &it) { + Ok(gid) => { if let Ok(mut st) = state.lock() { - st.active_pid = Some(child.id()); + if let Some(front) = st.queue.front_mut() { + front.gid = Some(gid.clone()); + front.status = DownloadStatus::Downloading; + } + st.active_gid = Some(gid); + st.last_message = Some(format!("downloading {}", it.title)); } - current_child = Some(child); } Err(e) => { if let Ok(mut st) = state.lock() { @@ -475,88 +548,88 @@ fn worker_loop(cfg: Config, state: Arc>, rx: mpsc::Receiver front.status = DownloadStatus::Failed(e.clone()); } st.last_message = Some(format!("download failed: {e}")); - } - // drop failed item - if let Ok(mut st) = state.lock() { let _ = st.queue.pop_front(); - st.active_pid = None; + st.active_gid = None; } - continue; } } - } else { - thread::sleep(Duration::from_millis(120)); - continue; } + + thread::sleep(Duration::from_millis(80)); + continue; } - // update progress + handle completion - if let Some(child) = current_child.as_mut() { - // update bytes downloaded by looking at file in temp - update_progress_for_front(&state); + // ------------------------- + // 3) Poll progress + handle completion + // ------------------------- + poll_progress_for_front(&state); - match child.try_wait() { - Ok(Some(status)) => { - // finished - if status.success() { - // finalize/move - if let Ok(mut st) = state.lock() { - if let Some(front) = st.queue.front_mut() { - front.status = DownloadStatus::Finalizing; - } - } - if let Err(e) = finalize_front(&state) { - if let Ok(mut st) = state.lock() { - if let Some(front) = st.queue.front_mut() { - front.status = DownloadStatus::Failed(e.clone()); - } - st.last_message = Some(e); - } - // drop item - if let Ok(mut st) = state.lock() { - let _ = st.queue.pop_front(); - st.active_pid = None; - } - } else { - if let Ok(mut st) = state.lock() { - let _ = st.queue.pop_front(); - st.active_pid = None; - st.last_message = Some("download complete".into()); - } - } - } else { - let msg = format!("aria2 exited: {status}"); - if let Ok(mut st) = state.lock() { - if let Some(front) = st.queue.front_mut() { - front.status = DownloadStatus::Failed(msg.clone()); - } - st.last_message = Some(msg); - let _ = st.queue.pop_front(); - st.active_pid = None; + match aria2_front_status(&state).as_deref() { + Some("complete") => { + // mark finalizing + if let Ok(mut st) = state.lock() { + if let Some(front) = st.queue.front_mut() { + front.status = DownloadStatus::Finalizing; + } + } + + // finalize + if let Err(e) = finalize_front(&state) { + if let Ok(mut st) = state.lock() { + if let Some(front) = st.queue.front_mut() { + front.status = DownloadStatus::Failed(e.clone()); } + st.last_message = Some(e); + let _ = st.queue.pop_front(); + st.active_gid = None; + } + } else { + // remove aria2 result + let gid = state.lock().ok().and_then(|s| s.active_gid.clone()); + if let Some(g) = gid { + let _ = aria2_rpc_call("aria2.removeDownloadResult", json!([g])); } - current_child = None; + if let Ok(mut st) = state.lock() { + let _ = st.queue.pop_front(); + st.active_gid = None; + st.last_message = Some("download complete".into()); + } } - Ok(None) => { - thread::sleep(Duration::from_millis(200)); - } - Err(_) => { - thread::sleep(Duration::from_millis(200)); + } + + Some("error") => { + if let Ok(mut st) = state.lock() { + if let Some(front) = st.queue.front_mut() { + front.status = DownloadStatus::Failed("aria2 status=error".into()); + } + st.last_message = Some("aria2 failed".into()); + let _ = st.queue.pop_front(); + st.active_gid = None; } } + + Some("active") | Some("waiting") => { + thread::sleep(Duration::from_millis(200)); + } + + _ => { + thread::sleep(Duration::from_millis(200)); + } } } - // shutdown cleanup: kill child, clear temp - if let Some(mut child) = current_child { - let _ = child.kill(); - let _ = child.wait(); - } + // ------------------------- + // shutdown cleanup + // ------------------------- + let _ = aria2_rpc_call("aria2.shutdown", json!([])); + let _ = aria2_daemon.kill(); + let _ = aria2_daemon.wait(); let _ = cleanup_temp_dir(); } + // Each download gets its own isolated temp folder so deletes can't affect others. // temp/jobs// fn temp_dir_for(it: &DownloadItem) -> Result { @@ -572,103 +645,74 @@ fn delete_temp_for(it: &DownloadItem) -> Result<(), String> { Ok(()) } -fn run_one_download( - base_url: &str, - token: &str, - it: &DownloadItem, - state: &Arc>, -) -> Result { +fn aria2_add_download(base_url: &str, token: &str, it: &DownloadItem) -> Result { let base = base_url.trim_end_matches('/'); let url = format!("{}/Items/{}/Download", base, it.item_id); - // We download into a dedicated temp folder for this item. - // Keep your existing folder structure by deriving tmp from final_rel_path. let tmp_dir = temp_dir_for(it)?; - - // Ensure temp dir exists and is empty-ish for robustness if tmp_dir.exists() { - // If a previous attempt left junk, clear it so we don't pick wrong file on finalize. let _ = fs::remove_dir_all(&tmp_dir); } fs::create_dir_all(&tmp_dir).map_err(|e| format!("mk temp dir: {e}"))?; - // NOTE: - // - We do NOT use --out anymore. - // - We enable --content-disposition so aria2c uses server-provided filename (real extension). - let mut cmd = Command::new("aria2c"); - cmd.arg("--allow-overwrite=true") - .arg("--auto-file-renaming=false") - .arg("--continue=true") - .arg("--check-certificate=true") - .arg("--max-tries=0") - .arg("--retry-wait=5") - .arg("--timeout=60") - .arg("--connect-timeout=10") - .arg("--summary-interval=1") - .arg("--console-log-level=warn") - .arg("--file-allocation=none") - .arg("--content-disposition=true") - .arg("--dir") - .arg(&tmp_dir) - .arg("--header") - .arg(format!("X-Emby-Token: {token}")) - .arg(&url) - .stdin(Stdio::null()) - .stdout(Stdio::null()) - .stderr(Stdio::null()); + // aria2 options + let opts = json!({ + "dir": tmp_dir.to_string_lossy(), + "allow-overwrite": "true", + "auto-file-renaming": "false", + "continue": "true", + "check-certificate": "true", + "content-disposition": "true", + "header": format!("X-Emby-Token: {token}"), + }); - let child = cmd - .spawn() - .map_err(|e| format!("spawn aria2c failed (is aria2c installed?): {e}"))?; + // params: [ [uri], options ] + let result = aria2_rpc_call("aria2.addUri", json!([[url], opts]))?; - if let Ok(mut st) = state.lock() { - st.last_message = Some(format!("downloading {}", it.title)); - } - - Ok(child) + result.as_str() + .map(|s| s.to_string()) + .ok_or_else(|| format!("aria2.addUri returned unexpected result: {result}")) } -fn update_progress_for_front(state: &Arc>) { - let (tmp_dir, idx) = { +fn poll_progress_for_front(state: &Arc>) { + let gid = { let st = match state.lock() { Ok(s) => s, Err(_) => return, }; - if st.queue.is_empty() { - return; - } - let it = st.queue.front().unwrap(); - let tmp_dir = match temp_dir_for(it) { - Ok(p) => p, - Err(_) => return, - }; - (tmp_dir, 0usize) + st.queue.front().and_then(|it| it.gid.clone()) }; - // Find the largest non-.aria2 file in temp dir (works for partial and final) - let mut best: Option = None; + let gid = match gid { + Some(g) => g, + None => return, + }; - if let Ok(rd) = fs::read_dir(&tmp_dir) { - for ent in rd.flatten() { - let p = ent.path(); - if !p.is_file() { - continue; - } - let name = p.file_name().and_then(|s| s.to_str()).unwrap_or(""); - if name.ends_with(".aria2") { - continue; - } - if let Ok(m) = fs::metadata(&p) { - let sz = m.len(); - best = Some(best.map(|b| b.max(sz)).unwrap_or(sz)); - } - } - } + let res = match aria2_rpc_call("aria2.tellStatus", json!([gid, ["status","completedLength","totalLength","downloadSpeed"]])) { + Ok(v) => v, + Err(_) => return, + }; - if let Some(sz) = best { - if let Ok(mut st) = state.lock() { - if let Some(front) = st.queue.get_mut(idx) { - front.bytes_downloaded = sz; + let status = res.get("status").and_then(|v| v.as_str()).unwrap_or(""); + let completed = res.get("completedLength").and_then(|v| v.as_str()).and_then(|s| s.parse::().ok()).unwrap_or(0); + let total = res.get("totalLength").and_then(|v| v.as_str()).and_then(|s| s.parse::().ok()); + let speed = res.get("downloadSpeed").and_then(|v| v.as_str()).and_then(|s| s.parse::().ok()).unwrap_or(0); + + let eta = match (total, speed) { + (Some(t), s) if t > completed && s > 0 => Some((t - completed) / s), + _ => None, + }; + + if let Ok(mut st) = state.lock() { + if let Some(front) = st.queue.front_mut() { + front.bytes_downloaded = completed; + front.bytes_total = total; + front.download_speed = speed; + front.eta_secs = eta; + + // optional: if aria2 says error, reflect it + if status == "error" { + front.status = DownloadStatus::Failed("aria2 error".into()); } } } @@ -725,4 +769,92 @@ fn finalize_front(state: &Arc>) -> Result<(), String> { let _ = fs::remove_dir_all(&tmp_dir); Ok(()) -} \ No newline at end of file +} + + +const ARIA2_RPC_HOST: &str = "127.0.0.1"; +const ARIA2_RPC_PORT: u16 = 6800; + +fn aria2_http_post(body: &str) -> Result { + let mut stream = TcpStream::connect((ARIA2_RPC_HOST, ARIA2_RPC_PORT)) + .map_err(|e| format!("aria2 rpc connect failed: {e}"))?; + + let req = format!( + "POST /jsonrpc HTTP/1.1\r\nHost: {}:{}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + ARIA2_RPC_HOST, + ARIA2_RPC_PORT, + body.len(), + body + ); + + stream.write_all(req.as_bytes()).map_err(|e| format!("aria2 rpc write failed: {e}"))?; + + let mut resp = String::new(); + stream.read_to_string(&mut resp).map_err(|e| format!("aria2 rpc read failed: {e}"))?; + + // split headers/body + let (_hdrs, body) = resp.split_once("\r\n\r\n").ok_or("aria2 rpc bad http response")?; + Ok(body.to_string()) +} + +fn aria2_rpc_call(method: &str, params: Value) -> Result { + let payload = json!({ + "jsonrpc": "2.0", + "id": "cj-jftui", + "method": method, + "params": params, + }); + + let body = payload.to_string(); + let raw = aria2_http_post(&body)?; + let v: Value = serde_json::from_str(raw.trim()).map_err(|e| format!("aria2 rpc json parse failed: {e} ({raw})"))?; + + if let Some(err) = v.get("error") { + return Err(format!("aria2 rpc error: {err}")); + } + Ok(v.get("result").cloned().unwrap_or(Value::Null)) +} + +fn start_aria2_daemon() -> Result { + let mut cmd = Command::new("aria2c"); + cmd.arg("--enable-rpc=true") + .arg("--rpc-listen-all=false") + .arg(format!("--rpc-listen-port={}", ARIA2_RPC_PORT)) + .arg("--rpc-allow-origin-all=true") + .arg("--rpc-max-request-size=2M") + .arg("--check-certificate=true") + .arg("--continue=true") + .arg("--max-tries=0") + .arg("--retry-wait=5") + .arg("--timeout=60") + .arg("--connect-timeout=10") + .arg("--summary-interval=1") + .arg("--console-log-level=warn") + .arg("--file-allocation=none") + .arg("--content-disposition=true") + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + + let child = cmd.spawn().map_err(|e| format!("spawn aria2c rpc daemon failed: {e}"))?; + + // quick readiness check (try rpc a few times) + let start = Instant::now(); + while start.elapsed() < Duration::from_millis(800) { + if aria2_rpc_call("aria2.getVersion", json!([])).is_ok() { + return Ok(child); + } + thread::sleep(Duration::from_millis(40)); + } + + Err("aria2 rpc daemon did not become ready".into()) +} + +fn aria2_front_status(state: &Arc>) -> Option { + let gid = { + let st = state.lock().ok()?; + st.queue.front().and_then(|it| it.gid.clone()) + }?; + let res = aria2_rpc_call("aria2.tellStatus", json!([gid, ["status"]])).ok()?; + res.get("status").and_then(|v| v.as_str()).map(|s| s.to_string()) +}