siege/CjTui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Downloading 99% complete

Browse Source
This commit is contained in:
Chris Johnson
2026-02-16 15:37:06 -06:00
parent f466775cc4
commit 1a8ed911d5
1 changed files with 317 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files

446
src/core/downloads.rs
View File

@@ -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<u64>,
pub download_speed: u64, // bytes/sec
pub eta_secs: Option<u64>, // seconds remaining
// aria2
pub gid: Option<String>,
}
#[derive(Debug)]
@@ -52,7 +62,7 @@ pub enum DownloadCommand {
#[derive(Default)]
pub struct DownloadState {
pub queue: VecDeque<DownloadItem>,
pub active_pid: Option<u32>,
pub active_gid: Option<String>,
pub last_message: Option<String>,
}
@@ -231,6 +241,9 @@ fn make_download_item_movie(item: &JfItem) -> Result<DownloadItem, String> {
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<DownloadItem, String> {
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<DownloadItem, String> {
.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<DownloadItem, String> {
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<DownloadItem, String> {
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<DownloadItem, String> {
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<PathBuf, String> {
temp_dir_for(it)
}
@@ -350,7 +399,11 @@ fn find_real_downloaded_file(tmp_dir: &Path) -> Result<PathBuf, String> {
fn worker_loop(cfg: Config, state: Arc<Mutex<DownloadState>>, rx: mpsc::Receiver<DownloadCommand>) {
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<Mutex<DownloadState>>, rx: mpsc::Receiver
}
};
let token = cfg.access_token.clone().unwrap_or_default();
let mut shutdown = false;
let mut current_child: Option<Child> = 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<Mutex<DownloadState>>, 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<Mutex<DownloadState>>, 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<Mutex<DownloadState>>, 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;
}
}
}
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<Mutex<DownloadState>>, 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;
}
continue;
st.active_gid = None;
}
}
} else {
thread::sleep(Duration::from_millis(120));
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);
thread::sleep(Duration::from_millis(80));
continue;
}
match child.try_wait() {
Ok(Some(status)) => {
// finished
if status.success() {
// finalize/move
// -------------------------
// 3) Poll progress + handle completion
// -------------------------
poll_progress_for_front(&state);
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);
}
// drop item
if let Ok(mut st) = state.lock() {
let _ = st.queue.pop_front();
st.active_pid = None;
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]));
}
if let Ok(mut st) = state.lock() {
let _ = st.queue.pop_front();
st.active_pid = None;
st.active_gid = None;
st.last_message = Some("download complete".into());
}
}
} else {
let msg = format!("aria2 exited: {status}");
}
Some("error") => {
if let Ok(mut st) = state.lock() {
if let Some(front) = st.queue.front_mut() {
front.status = DownloadStatus::Failed(msg.clone());
front.status = DownloadStatus::Failed("aria2 status=error".into());
}
st.last_message = Some(msg);
st.last_message = Some("aria2 failed".into());
let _ = st.queue.pop_front();
st.active_pid = None;
st.active_gid = None;
}
}
current_child = None;
}
Ok(None) => {
Some("active") | Some("waiting") => {
thread::sleep(Duration::from_millis(200));
}
Err(_) => {
_ => {
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/<item_id>/
fn temp_dir_for(it: &DownloadItem) -> Result<PathBuf, String> {
@@ -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<Mutex<DownloadState>>,
) -> Result<Child, String> {
fn aria2_add_download(base_url: &str, token: &str, it: &DownloadItem) -> Result<String, String> {
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<Mutex<DownloadState>>) {
let (tmp_dir, idx) = {
fn poll_progress_for_front(state: &Arc<Mutex<DownloadState>>) {
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,
st.queue.front().and_then(|it| it.gid.clone())
};
let gid = match gid {
Some(g) => g,
None => return,
};
let res = match aria2_rpc_call("aria2.tellStatus", json!([gid, ["status","completedLength","totalLength","downloadSpeed"]])) {
Ok(v) => v,
Err(_) => return,
};
(tmp_dir, 0usize)
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::<u64>().ok()).unwrap_or(0);
let total = res.get("totalLength").and_then(|v| v.as_str()).and_then(|s| s.parse::<u64>().ok());
let speed = res.get("downloadSpeed").and_then(|v| v.as_str()).and_then(|s| s.parse::<u64>().ok()).unwrap_or(0);
let eta = match (total, speed) {
(Some(t), s) if t > completed && s > 0 => Some((t - completed) / s),
_ => None,
};
// Find the largest non-.aria2 file in temp dir (works for partial and final)
let mut best: Option<u64> = None;
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));
}
}
}
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;
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());
}
}
}
@@ -726,3 +770,91 @@ fn finalize_front(state: &Arc<Mutex<DownloadState>>) -> Result<(), String> {
Ok(())
}
const ARIA2_RPC_HOST: &str = "127.0.0.1";
const ARIA2_RPC_PORT: u16 = 6800;
fn aria2_http_post(body: &str) -> Result<String, String> {
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<Value, String> {
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<Child, String> {
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<Mutex<DownloadState>>) -> Option<String> {
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())
}