945 lines
30 KiB
Rust
945 lines
30 KiB
Rust
|
|
use std::{
|
|
collections::{VecDeque, HashSet},
|
|
fs,
|
|
path::{Path, PathBuf},
|
|
process::{Child, Command, Stdio},
|
|
sync::{mpsc, Arc, Mutex},
|
|
thread,
|
|
time::{Duration, Instant},
|
|
net::TcpStream,
|
|
io::{Read, Write},
|
|
};
|
|
|
|
use serde_json::{json, Value};
|
|
|
|
use crate::core::{config::Config, jellyfin::{JellyfinClient, JfItem, LibraryQuery}, paths};
|
|
|
|
#[derive(Clone, Debug)]
|
|
pub enum DownloadKind {
|
|
Movie,
|
|
Episode { series: String, season: Option<i32>, episode: Option<i32> },
|
|
Track { artist: Option<String>, album: Option<String>, track_no: Option<i32> },
|
|
Other,
|
|
}
|
|
|
|
#[derive(Clone, Debug)]
|
|
pub enum DownloadStatus {
|
|
Queued,
|
|
Downloading,
|
|
Finalizing,
|
|
Completed,
|
|
Failed(String),
|
|
Cancelled,
|
|
}
|
|
|
|
#[derive(Clone, Debug)]
|
|
pub struct DownloadItem {
|
|
pub item_id: String,
|
|
pub title: String,
|
|
pub kind: DownloadKind,
|
|
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)]
|
|
pub enum DownloadCommand {
|
|
Enqueue(Vec<DownloadItem>),
|
|
Cancel { index: usize }, // cancel by queue index (including current 0)
|
|
RemoveQueued { index: usize }, // remove if queued (not current)
|
|
Shutdown,
|
|
}
|
|
|
|
#[derive(Default)]
|
|
pub struct DownloadState {
|
|
pub queue: VecDeque<DownloadItem>,
|
|
pub active_gid: Option<String>,
|
|
pub last_message: Option<String>,
|
|
}
|
|
|
|
pub struct DownloadManager {
|
|
tx: mpsc::Sender<DownloadCommand>,
|
|
pub state: Arc<Mutex<DownloadState>>,
|
|
}
|
|
|
|
impl DownloadManager {
|
|
pub fn start(cfg: Config) -> Self {
|
|
let state = Arc::new(Mutex::new(DownloadState::default()));
|
|
let (tx, rx) = mpsc::channel::<DownloadCommand>();
|
|
|
|
// cleanup temp on start
|
|
let _ = cleanup_temp_dir();
|
|
|
|
let state_bg = Arc::clone(&state);
|
|
thread::spawn(move || worker_loop(cfg, state_bg, rx));
|
|
|
|
Self { tx, state }
|
|
}
|
|
|
|
pub fn send(&self, cmd: DownloadCommand) {
|
|
let _ = self.tx.send(cmd);
|
|
}
|
|
}
|
|
|
|
// ---------- directory helpers ----------
|
|
|
|
const DOWNLOAD_MARKER_EXT: &str = "cj-jftui.json";
|
|
|
|
fn marker_path_for_media(media_path: &Path) -> PathBuf {
|
|
// Example:
|
|
// Movie.mkv -> Movie.mkv.cj-jftui.json
|
|
let file_name = media_path
|
|
.file_name()
|
|
.and_then(|s| s.to_str())
|
|
.unwrap_or("download");
|
|
|
|
media_path
|
|
.parent()
|
|
.unwrap_or_else(|| Path::new("."))
|
|
.join(format!("{file_name}.{DOWNLOAD_MARKER_EXT}"))
|
|
}
|
|
|
|
fn write_download_marker(media_path: &Path, item_id: &str) -> Result<(), String> {
|
|
let marker = marker_path_for_media(media_path);
|
|
|
|
let payload = json!({
|
|
"item_id": item_id,
|
|
"saved_at_unix": std::time::SystemTime::now()
|
|
.duration_since(std::time::UNIX_EPOCH)
|
|
.ok()
|
|
.map(|d| d.as_secs())
|
|
.unwrap_or(0),
|
|
"media_file": media_path.file_name().and_then(|s| s.to_str()).unwrap_or(""),
|
|
});
|
|
|
|
fs::write(&marker, payload.to_string())
|
|
.map_err(|e| format!("write marker failed: {e}"))?;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
fn scan_dir_for_markers(dir: &Path, out: &mut HashSet<String>) -> Result<(), String> {
|
|
if !dir.exists() {
|
|
return Ok(());
|
|
}
|
|
|
|
for ent in fs::read_dir(dir).map_err(|e| format!("read_dir failed: {e}"))? {
|
|
let p = ent.map_err(|e| format!("dir entry failed: {e}"))?.path();
|
|
|
|
if p.is_dir() {
|
|
scan_dir_for_markers(&p, out)?;
|
|
continue;
|
|
}
|
|
|
|
// Look for *.cj-jftui.json
|
|
let name = p.file_name().and_then(|s| s.to_str()).unwrap_or("");
|
|
if !name.ends_with(DOWNLOAD_MARKER_EXT) {
|
|
continue;
|
|
}
|
|
|
|
let raw = match fs::read_to_string(&p) {
|
|
Ok(s) => s,
|
|
Err(_) => continue,
|
|
};
|
|
|
|
let v: Value = match serde_json::from_str(raw.trim()) {
|
|
Ok(v) => v,
|
|
Err(_) => continue,
|
|
};
|
|
|
|
if let Some(id) = v.get("item_id").and_then(|x| x.as_str()) {
|
|
if !id.trim().is_empty() {
|
|
out.insert(id.to_string());
|
|
}
|
|
}
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Public: scan the downloads root and return set of downloaded Jellyfin item ids.
|
|
pub fn scan_downloaded_ids() -> Result<HashSet<String>, String> {
|
|
let root = downloads_root()?;
|
|
let mut out = HashSet::new();
|
|
scan_dir_for_markers(&root, &mut out)?;
|
|
Ok(out)
|
|
}
|
|
|
|
|
|
pub fn downloads_root() -> Result<PathBuf, String> {
|
|
let dir = paths::app_config_dir().ok_or_else(|| "no config dir".to_string())?;
|
|
Ok(dir.join("downloads"))
|
|
}
|
|
|
|
pub fn temp_root() -> Result<PathBuf, String> {
|
|
let dir = paths::app_config_dir().ok_or_else(|| "no config dir".to_string())?;
|
|
Ok(dir.join("temp"))
|
|
}
|
|
|
|
pub fn ensure_dirs() -> Result<(), String> {
|
|
fs::create_dir_all(downloads_root().map_err(|e| e.to_string())?).map_err(|e| e.to_string())?;
|
|
fs::create_dir_all(temp_root().map_err(|e| e.to_string())?).map_err(|e| e.to_string())?;
|
|
Ok(())
|
|
}
|
|
|
|
pub fn cleanup_temp_dir() -> Result<(), String> {
|
|
let tmp = temp_root()?;
|
|
if tmp.exists() {
|
|
fs::remove_dir_all(&tmp).map_err(|e| format!("remove temp: {e}"))?;
|
|
}
|
|
fs::create_dir_all(&tmp).map_err(|e| format!("create temp: {e}"))?;
|
|
fs::create_dir_all(tmp.join("jobs")).map_err(|e| format!("create temp/jobs: {e}"))?;
|
|
Ok(())
|
|
}
|
|
|
|
// ---------- queue building (expansion) ----------
|
|
|
|
pub fn expand_for_enqueue(
|
|
jf: &JellyfinClient,
|
|
user_id: &str,
|
|
item: &JfItem,
|
|
) -> Result<Vec<DownloadItem>, String> {
|
|
let t = item.item_type.as_deref().unwrap_or("");
|
|
match t {
|
|
"Movie" => Ok(vec![make_download_item_movie(item)?]),
|
|
"Episode" => Ok(vec![make_download_item_episode(item)?]),
|
|
"Audio" => Ok(vec![make_download_item_track(item)?]),
|
|
|
|
// Containers:
|
|
"Season" => {
|
|
// episodes under Season (non-recursive)
|
|
list_all_items(jf, user_id, LibraryQuery::all().with_parent(item.id.clone()).with_item_types("Episode"), false)?
|
|
.into_iter()
|
|
.map(|ep| make_download_item_episode(&ep))
|
|
.collect()
|
|
}
|
|
"Series" => {
|
|
// episodes under Series (recursive)
|
|
let mut q = LibraryQuery::all()
|
|
.with_parent(item.id.clone())
|
|
.with_item_types("Episode");
|
|
q.recursive = true;
|
|
list_all_items(jf, user_id, q, true)?
|
|
.into_iter()
|
|
.map(|ep| make_download_item_episode(&ep))
|
|
.collect()
|
|
}
|
|
"MusicAlbum" => {
|
|
// tracks under album
|
|
list_all_items(jf, user_id, LibraryQuery::all().with_parent(item.id.clone()).with_item_types("Audio"), false)?
|
|
.into_iter()
|
|
.map(|tr| make_download_item_track(&tr))
|
|
.collect()
|
|
}
|
|
"Playlist" => {
|
|
// playlist items via special endpoint
|
|
list_all_playlist_items(jf, user_id, &item.id)?
|
|
.into_iter()
|
|
.map(|it| {
|
|
let tt = it.item_type.as_deref().unwrap_or("");
|
|
match tt {
|
|
"Audio" => make_download_item_track(&it),
|
|
"Episode" => make_download_item_episode(&it),
|
|
"Movie" => make_download_item_movie(&it),
|
|
_ => make_download_item_other(&it),
|
|
}
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
_ => Ok(vec![make_download_item_other(item)?]),
|
|
}
|
|
}
|
|
|
|
fn list_all_items(
|
|
jf: &JellyfinClient,
|
|
user_id: &str,
|
|
query: LibraryQuery,
|
|
_maybe_large: bool,
|
|
) -> Result<Vec<JfItem>, String> {
|
|
// user-driven action, so paging here is fine
|
|
const LIMIT: usize = 200;
|
|
let mut start = 0usize;
|
|
let mut out = Vec::new();
|
|
loop {
|
|
let page = jf.list_items(user_id, query.clone(), start, LIMIT)?;
|
|
out.extend(page.items.into_iter());
|
|
if out.len() >= page.total || page.total == 0 {
|
|
break;
|
|
}
|
|
start += LIMIT;
|
|
}
|
|
Ok(out)
|
|
}
|
|
|
|
fn list_all_playlist_items(jf: &JellyfinClient, user_id: &str, playlist_id: &str) -> Result<Vec<JfItem>, String> {
|
|
const LIMIT: usize = 200;
|
|
let mut start = 0usize;
|
|
let mut out = Vec::new();
|
|
loop {
|
|
let page = jf.playlist_items(user_id, playlist_id, start, LIMIT)?;
|
|
out.extend(page.items.into_iter());
|
|
if out.len() >= page.total || page.total == 0 {
|
|
break;
|
|
}
|
|
start += LIMIT;
|
|
}
|
|
Ok(out)
|
|
}
|
|
|
|
// ---------- naming ----------
|
|
|
|
fn sanitize_component(s: &str) -> String {
|
|
let mut out = String::with_capacity(s.len());
|
|
for ch in s.chars() {
|
|
// conservative sanitization for Linux/Windows compatibility
|
|
if matches!(ch, '<' | '>' | ':' | '"' | '/' | '\\' | '|' | '?' | '*' ) {
|
|
out.push('_');
|
|
} else {
|
|
out.push(ch);
|
|
}
|
|
}
|
|
let out = out.trim().to_string();
|
|
if out.is_empty() { "_".into() } else { out.chars().take(100).collect() }
|
|
}
|
|
|
|
fn make_download_item_movie(item: &JfItem) -> Result<DownloadItem, String> {
|
|
let name = sanitize_component(&item.name);
|
|
let rel = PathBuf::from("movies").join(format!("{name}.watch"));
|
|
Ok(DownloadItem {
|
|
item_id: item.id.clone(),
|
|
title: item.name.clone(),
|
|
kind: DownloadKind::Movie,
|
|
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_episode(item: &JfItem) -> Result<DownloadItem, String> {
|
|
let series = item.series_name.clone().unwrap_or_else(|| "Unknown Show".into());
|
|
let series_s = sanitize_component(&series);
|
|
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 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);
|
|
|
|
Ok(DownloadItem {
|
|
item_id: item.id.clone(),
|
|
title: format!("{series} - {}", item.name),
|
|
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,
|
|
})
|
|
}
|
|
|
|
fn make_download_item_track(item: &JfItem) -> Result<DownloadItem, String> {
|
|
let artist = item
|
|
.album_artist
|
|
.clone()
|
|
.or_else(|| item.artists.get(0).cloned());
|
|
|
|
let album = item.album.clone();
|
|
let track_no = item.index_number;
|
|
|
|
let artist_s = sanitize_component(artist.as_deref().unwrap_or("Unknown Artist"));
|
|
let album_s = sanitize_component(album.as_deref().unwrap_or("Unknown Album"));
|
|
let title_s = sanitize_component(&item.name);
|
|
|
|
let filename = match track_no {
|
|
Some(n) => format!("{:02} - {title_s}.listen", n),
|
|
None => format!("{title_s}.listen"),
|
|
};
|
|
|
|
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,
|
|
},
|
|
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(),
|
|
kind: DownloadKind::Other,
|
|
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)
|
|
}
|
|
|
|
// Find the actual downloaded media file in tmp dir.
|
|
// We ignore aria2 metadata (*.aria2) and partials (*.part).
|
|
fn find_real_downloaded_file(tmp_dir: &Path) -> Result<PathBuf, String> {
|
|
let mut best: Option<(u64, PathBuf)> = None;
|
|
|
|
let rd = fs::read_dir(tmp_dir).map_err(|e| format!("read temp dir: {e}"))?;
|
|
for ent in rd {
|
|
let p = ent.map_err(|e| format!("read dir entry: {e}"))?.path();
|
|
if !p.is_file() {
|
|
continue;
|
|
}
|
|
let name = p.file_name().and_then(|s| s.to_str()).unwrap_or("");
|
|
|
|
if name.ends_with(".aria2") || name.ends_with(".part") {
|
|
continue;
|
|
}
|
|
|
|
let sz = fs::metadata(&p).map_err(|e| format!("stat temp file: {e}"))?.len();
|
|
match &best {
|
|
None => best = Some((sz, p)),
|
|
Some((best_sz, _)) => {
|
|
if sz > *best_sz {
|
|
best = Some((sz, p));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
best.map(|(_, p)| p)
|
|
.ok_or_else(|| "no downloaded media file found in temp dir".to_string())
|
|
}
|
|
|
|
|
|
|
|
// ---------- worker loop ----------
|
|
|
|
fn worker_loop(cfg: Config, state: Arc<Mutex<DownloadState>>, rx: mpsc::Receiver<DownloadCommand>) {
|
|
let _ = ensure_dirs();
|
|
|
|
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() {
|
|
st.last_message = Some(format!("downloads disabled: {e}"));
|
|
}
|
|
return;
|
|
}
|
|
};
|
|
|
|
while !shutdown {
|
|
// -------------------------
|
|
// 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 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);
|
|
}
|
|
}
|
|
st.last_message = Some("queued".into());
|
|
}
|
|
}
|
|
|
|
DownloadCommand::Cancel { index } => {
|
|
if index == 0 {
|
|
// 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]));
|
|
}
|
|
|
|
// 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 {
|
|
// 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();
|
|
if index < v.len() {
|
|
let mut it = v.remove(index);
|
|
it.status = DownloadStatus::Cancelled;
|
|
let _ = delete_temp_for(&it);
|
|
}
|
|
st.queue = v.into();
|
|
st.last_message = Some("removed from queue".into());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
DownloadCommand::RemoveQueued { index } => {
|
|
if let Ok(mut st) = state.lock() {
|
|
if index > 0 && index < st.queue.len() {
|
|
let mut v: Vec<_> = st.queue.drain(..).collect();
|
|
let mut it = v.remove(index);
|
|
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 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,
|
|
};
|
|
|
|
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 !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 aria2_add_download(&cfg.base_url, &token, &it) {
|
|
Ok(gid) => {
|
|
if let Ok(mut st) = state.lock() {
|
|
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));
|
|
}
|
|
}
|
|
Err(e) => {
|
|
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(format!("download failed: {e}"));
|
|
let _ = st.queue.pop_front();
|
|
st.active_gid = None;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
thread::sleep(Duration::from_millis(80));
|
|
continue;
|
|
}
|
|
|
|
// -------------------------
|
|
// 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);
|
|
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]));
|
|
}
|
|
|
|
if let Ok(mut st) = state.lock() {
|
|
let _ = st.queue.pop_front();
|
|
st.active_gid = None;
|
|
st.last_message = Some("download complete".into());
|
|
}
|
|
}
|
|
}
|
|
|
|
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
|
|
// -------------------------
|
|
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> {
|
|
let tmp = temp_root()?;
|
|
Ok(tmp.join("jobs").join(&it.item_id))
|
|
}
|
|
|
|
fn delete_temp_for(it: &DownloadItem) -> Result<(), String> {
|
|
let dir = temp_dir_for(it)?;
|
|
if dir.exists() {
|
|
fs::remove_dir_all(&dir).map_err(|e| format!("rm temp: {e}"))?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
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);
|
|
|
|
let tmp_dir = temp_dir_for(it)?;
|
|
if tmp_dir.exists() {
|
|
let _ = fs::remove_dir_all(&tmp_dir);
|
|
}
|
|
fs::create_dir_all(&tmp_dir).map_err(|e| format!("mk temp dir: {e}"))?;
|
|
|
|
// 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}"),
|
|
});
|
|
|
|
// params: [ [uri], options ]
|
|
let result = aria2_rpc_call("aria2.addUri", json!([[url], opts]))?;
|
|
|
|
result.as_str()
|
|
.map(|s| s.to_string())
|
|
.ok_or_else(|| format!("aria2.addUri returned unexpected result: {result}"))
|
|
}
|
|
|
|
fn poll_progress_for_front(state: &Arc<Mutex<DownloadState>>) {
|
|
let gid = {
|
|
let st = match state.lock() {
|
|
Ok(s) => s,
|
|
Err(_) => return,
|
|
};
|
|
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,
|
|
};
|
|
|
|
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,
|
|
};
|
|
|
|
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());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
fn finalize_front(state: &Arc<Mutex<DownloadState>>) -> Result<(), String> {
|
|
let (it, tmp_dir, final_abs_watch) = {
|
|
let st = state.lock().map_err(|_| "lock poisoned".to_string())?;
|
|
let it = st.queue.front().ok_or_else(|| "no front".to_string())?.clone();
|
|
let tmp_dir = tmp_dir_of_item(&it)?;
|
|
let dl_root = downloads_root()?;
|
|
let final_abs_watch = dl_root.join(&it.final_rel_path);
|
|
(it, tmp_dir, final_abs_watch)
|
|
};
|
|
|
|
// Ensure final dir exists
|
|
let final_parent = final_abs_watch
|
|
.parent()
|
|
.ok_or_else(|| "no final parent".to_string())?;
|
|
fs::create_dir_all(final_parent).map_err(|e| format!("mk final dir: {e}"))?;
|
|
|
|
// Find real downloaded file (with real extension) inside tmp_dir
|
|
let real_file = find_real_downloaded_file(&tmp_dir)?;
|
|
|
|
// Build final filename:
|
|
// final_rel_path ends in "...something.watch/.listen"
|
|
// Keep stem, replace extension with actual media extension
|
|
let real_ext = real_file
|
|
.extension()
|
|
.and_then(|s| s.to_str())
|
|
.unwrap_or("")
|
|
.to_string();
|
|
|
|
let stem = final_abs_watch
|
|
.file_stem()
|
|
.and_then(|s| s.to_str())
|
|
.ok_or_else(|| "bad final filename stem".to_string())?
|
|
.to_string();
|
|
|
|
let final_abs = if real_ext.is_empty() {
|
|
final_abs_watch.clone()
|
|
} else {
|
|
final_parent.join(format!("{stem}.{real_ext}"))
|
|
};
|
|
|
|
// Move file into place
|
|
fs::rename(&real_file, &final_abs).map_err(|e| format!("move to final failed: {e}"))?;
|
|
|
|
// ✅ NEW: write marker so we can detect "downloaded" later
|
|
// (If marker fails, we still consider download successful.)
|
|
let _ = write_download_marker(&final_abs, &it.item_id);
|
|
|
|
// Cleanup temp folder fully
|
|
let _ = fs::remove_dir_all(&tmp_dir);
|
|
|
|
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())
|
|
}
|