orrisroot
a16f73543d
A review against the Python client, which shares this client's config and login cache, found the transfers holding whole files in memory and several commands answering differently from their Python counterparts. - Stream uploads and downloads instead of buffering the whole file, so memory no longer scales with file size times concurrency. Uploads still declare a Content-Length rather than going out chunked. - Write a download beside its destination and move it into place once complete, and refuse a destination that cannot be written, so a failed transfer leaves what was already there untouched. - Report the server's own error detail instead of the bare status. - List locked sub-folders in `ls --json --recursive` using the given password and skip only those that cannot be unlocked, instead of failing the whole listing. - Name each folder's laboratory in JSON output and sort its entries, matching the Python client's schema and order. - Ask the server who is logged in for `whoami`, rather than trusting the cached name. - Collapse repeated separators and refuse `..` in remote paths. - Fall back to the API when a laboratory is missing from the cache, so a newly added one no longer needs a fresh login. - Verify the login cache digest on read, as the Python client does. - Accept `-e` as the short form of `--exclude`. - Satisfy clippy and rustfmt across the crate.
554 lines
19 KiB
Rust
554 lines
19 KiB
Rust
use crate::cache::create_readonly_conn;
|
|
use crate::commands::shared::{
|
|
find_file_by_name, find_folder_by_doi, find_folder_limited, find_laboratory,
|
|
find_subfolder_by_name, is_doi, parse_doi_remote_path, parse_remote_path,
|
|
};
|
|
use crate::connection::{ApiRequestLimiter, MDRSConnection};
|
|
use anyhow::{anyhow, bail};
|
|
use std::path::PathBuf;
|
|
use std::sync::Arc;
|
|
use tokio::task::JoinSet;
|
|
|
|
pub async fn download(
|
|
remote_path: &str,
|
|
local_path: &str,
|
|
recursive: bool,
|
|
skip_if_exists: bool,
|
|
password: Option<&str>,
|
|
excludes: Vec<String>,
|
|
) -> Result<(), anyhow::Error> {
|
|
let limiter = ApiRequestLimiter::new(crate::settings::SETTINGS.concurrent);
|
|
|
|
// Validate that local_path is an existing directory (matching Python's behaviour).
|
|
let local_real = std::fs::canonicalize(local_path)
|
|
.map_err(|_| anyhow!("Local directory `{}` not found.", local_path))?;
|
|
if !local_real.is_dir() {
|
|
bail!("Local directory `{}` not found.", local_path);
|
|
}
|
|
|
|
// Detect DOI path: "remote:10.xxxx/prefix.ID[/optional/sub/path]"
|
|
if is_doi(remote_path.split_once(':').map(|x| x.1).unwrap_or("")) {
|
|
let (remote, doi, subpath) = parse_doi_remote_path(remote_path)?;
|
|
let (raw_conn, _cache) = create_readonly_conn(&remote).await?;
|
|
let (doi_folder, lab) = find_folder_by_doi(&raw_conn, &doi, password).await?;
|
|
let abs_path = format!("{}{}", doi_folder.path.trim_end_matches('/'), subpath);
|
|
let abs_path_clean = abs_path.trim_end_matches('/');
|
|
|
|
// Check if the target is a folder.
|
|
// If it is, we download the directory. If it is not, we download the file or subfolder.
|
|
let (folder, is_folder) = match find_folder_limited(
|
|
&raw_conn,
|
|
&limiter,
|
|
lab.id,
|
|
abs_path_clean,
|
|
password,
|
|
)
|
|
.await
|
|
{
|
|
Ok(f) => (f, true),
|
|
Err(_) => {
|
|
let (parent_path, _) = match abs_path_clean.rfind('/') {
|
|
Some(0) => ("/".to_string(), abs_path_clean[1..].to_string()),
|
|
Some(pos) => (
|
|
abs_path_clean[..pos].to_string(),
|
|
abs_path_clean[pos + 1..].to_string(),
|
|
),
|
|
None => ("/".to_string(), abs_path_clean.to_string()),
|
|
};
|
|
let parent =
|
|
find_folder_limited(&raw_conn, &limiter, lab.id, &parent_path, password)
|
|
.await?;
|
|
(parent, false)
|
|
}
|
|
};
|
|
|
|
let conn = Arc::new(raw_conn);
|
|
let excludes = Arc::new(excludes);
|
|
|
|
if is_folder {
|
|
let path = folder.path.clone();
|
|
let lab_name = Arc::new(lab.name);
|
|
let top_local = local_real.join(&folder.name);
|
|
let password_owned = password.map(str::to_string);
|
|
|
|
let mut folder_tasks: JoinSet<Result<DownloadFolderTaskResult, anyhow::Error>> =
|
|
JoinSet::new();
|
|
let mut download_tasks: JoinSet<Result<(), anyhow::Error>> = JoinSet::new();
|
|
let mut errors = Vec::new();
|
|
|
|
if !recursive {
|
|
// Single-folder, non-recursive: download the files inside the DOI folder.
|
|
let files = conn.list_all_files_limited(&folder.id, &limiter).await?;
|
|
for file in &files {
|
|
if is_excluded(&excludes, lab_name.as_str(), &path, Some(&file.name)) {
|
|
continue;
|
|
}
|
|
let dest = local_real.join(&file.name);
|
|
if skip_if_exists
|
|
&& dest.exists()
|
|
&& let Ok(meta) = std::fs::metadata(&dest)
|
|
&& meta.len() == file.size
|
|
{
|
|
println!("{}", dest.display());
|
|
continue;
|
|
}
|
|
let url = make_absolute_url(&conn, &file.download_url);
|
|
conn.download_file_limited(&url, &dest.to_string_lossy(), &limiter)
|
|
.await?;
|
|
println!("{}", dest.display());
|
|
}
|
|
return Ok(());
|
|
}
|
|
|
|
let session = DownloadSession {
|
|
conn: conn.clone(),
|
|
limiter,
|
|
lab_name,
|
|
excludes,
|
|
password: password_owned,
|
|
skip_if_exists,
|
|
};
|
|
spawn_download_folder_task(
|
|
&mut folder_tasks,
|
|
session.clone(),
|
|
folder.id.clone(),
|
|
top_local,
|
|
);
|
|
|
|
drive_download_tasks(&mut folder_tasks, &mut download_tasks, &mut errors, session)
|
|
.await;
|
|
|
|
if !errors.is_empty() {
|
|
bail!(errors.join("\n"));
|
|
}
|
|
return Ok(());
|
|
}
|
|
|
|
// Case: target is a file or subfolder inside the resolved folder.
|
|
let basename = match abs_path_clean.rfind('/') {
|
|
Some(pos) => abs_path_clean[pos + 1..].to_string(),
|
|
None => abs_path_clean.to_string(),
|
|
};
|
|
let files = conn.list_all_files_limited(&folder.id, &limiter).await?;
|
|
|
|
// Case 1: basename matches a file in the folder.
|
|
if let Some(file) = find_file_by_name(&files, &basename) {
|
|
if is_excluded(&excludes, &lab.name, &folder.path, Some(&file.name)) {
|
|
return Ok(());
|
|
}
|
|
let dest = local_real.join(&file.name);
|
|
if skip_if_exists
|
|
&& dest.exists()
|
|
&& let Ok(meta) = std::fs::metadata(&dest)
|
|
&& meta.len() == file.size
|
|
{
|
|
println!("{}", dest.display());
|
|
return Ok(());
|
|
}
|
|
let url = make_absolute_url(&conn, &file.download_url);
|
|
conn.download_file_limited(&url, &dest.to_string_lossy(), &limiter)
|
|
.await?;
|
|
println!("{}", dest.display());
|
|
return Ok(());
|
|
}
|
|
|
|
// Case 2: basename matches a sub-folder.
|
|
let subfolder = find_subfolder_by_name(&folder.sub_folders, &basename);
|
|
if let Some(sub) = subfolder {
|
|
if !recursive {
|
|
bail!("Cannot download `{}`: Is a folder.", abs_path_clean);
|
|
}
|
|
let top_local = local_real.join(&sub.name);
|
|
let mut folder_tasks: JoinSet<Result<DownloadFolderTaskResult, anyhow::Error>> =
|
|
JoinSet::new();
|
|
let mut download_tasks: JoinSet<Result<(), anyhow::Error>> = JoinSet::new();
|
|
let mut errors = Vec::new();
|
|
let lab_name = Arc::new(lab.name.clone());
|
|
let password_owned = password.map(str::to_string);
|
|
|
|
let session = DownloadSession {
|
|
conn: conn.clone(),
|
|
limiter,
|
|
lab_name,
|
|
excludes,
|
|
password: password_owned,
|
|
skip_if_exists,
|
|
};
|
|
spawn_download_folder_task(
|
|
&mut folder_tasks,
|
|
session.clone(),
|
|
sub.id.clone(),
|
|
top_local,
|
|
);
|
|
|
|
drive_download_tasks(&mut folder_tasks, &mut download_tasks, &mut errors, session)
|
|
.await;
|
|
|
|
if !errors.is_empty() {
|
|
bail!(errors.join("\n"));
|
|
}
|
|
return Ok(());
|
|
}
|
|
|
|
bail!("File or folder `{}` not found.", abs_path_clean);
|
|
}
|
|
|
|
// Normal path: "remote:/labname/path/..."
|
|
let (remote, labname, r_path) = parse_remote_path(remote_path)?;
|
|
let (raw_conn, cache) = create_readonly_conn(&remote).await?;
|
|
let conn = Arc::new(raw_conn);
|
|
let lab = find_laboratory(&conn, cache.as_ref(), &labname).await?;
|
|
|
|
// Split r_path into the parent directory path and the target basename.
|
|
// Trailing slashes are already stripped by parse_remote_path, so this is safe.
|
|
let r_path_clean = r_path.trim_end_matches('/');
|
|
let (parent_path, basename) = match r_path_clean.rfind('/') {
|
|
Some(0) => ("/".to_string(), r_path_clean[1..].to_string()),
|
|
Some(pos) => (
|
|
r_path_clean[..pos].to_string(),
|
|
r_path_clean[pos + 1..].to_string(),
|
|
),
|
|
None => ("/".to_string(), r_path_clean.to_string()),
|
|
};
|
|
|
|
let parent_folder =
|
|
find_folder_limited(&conn, &limiter, lab.id, &parent_path, password).await?;
|
|
let files = conn
|
|
.list_all_files_limited(&parent_folder.id, &limiter)
|
|
.await?;
|
|
|
|
// Case 1: basename matches a file in the parent folder.
|
|
if let Some(file) = find_file_by_name(&files, &basename) {
|
|
if is_excluded(&excludes, &lab.name, &parent_folder.path, Some(&file.name)) {
|
|
return Ok(());
|
|
}
|
|
// Python always places the downloaded file inside the local directory.
|
|
let dest = local_real.join(&file.name);
|
|
if skip_if_exists
|
|
&& dest.exists()
|
|
&& let Ok(meta) = std::fs::metadata(&dest)
|
|
&& meta.len() == file.size
|
|
{
|
|
println!("{}", dest.display());
|
|
return Ok(());
|
|
}
|
|
let url = make_absolute_url(&conn, &file.download_url);
|
|
conn.download_file_limited(&url, &dest.to_string_lossy(), &limiter)
|
|
.await?;
|
|
println!("{}", dest.display());
|
|
return Ok(());
|
|
}
|
|
|
|
// Case 2: basename matches a sub-folder.
|
|
let subfolder = find_subfolder_by_name(&parent_folder.sub_folders, &basename);
|
|
if let Some(sub) = subfolder {
|
|
if !recursive {
|
|
bail!("Cannot download `{}`: Is a folder.", r_path_clean);
|
|
}
|
|
// Python downloads into local_path/<remote_folder_name>/ (not directly into local_path).
|
|
// We create that subdirectory first, then recurse into it.
|
|
let top_local = local_real.join(&sub.name);
|
|
|
|
let mut folder_tasks: JoinSet<Result<DownloadFolderTaskResult, anyhow::Error>> =
|
|
JoinSet::new();
|
|
let mut download_tasks: JoinSet<Result<(), anyhow::Error>> = JoinSet::new();
|
|
let mut errors = Vec::new();
|
|
let excludes = Arc::new(excludes);
|
|
let lab_name = Arc::new(lab.name.clone());
|
|
let password = password.map(str::to_string);
|
|
|
|
let session = DownloadSession {
|
|
conn: conn.clone(),
|
|
limiter,
|
|
lab_name,
|
|
excludes,
|
|
password,
|
|
skip_if_exists,
|
|
};
|
|
spawn_download_folder_task(
|
|
&mut folder_tasks,
|
|
session.clone(),
|
|
sub.id.clone(),
|
|
top_local,
|
|
);
|
|
|
|
drive_download_tasks(&mut folder_tasks, &mut download_tasks, &mut errors, session).await;
|
|
|
|
if !errors.is_empty() {
|
|
bail!(errors.join("\n"));
|
|
}
|
|
return Ok(());
|
|
}
|
|
|
|
Err(anyhow!("File or folder `{}` not found.", r_path_clean))
|
|
}
|
|
|
|
/// Return true if the given lab/folder/file path matches any exclude pattern.
|
|
/// Constructs: `/{lab_name}{folder_path}{file_name}` lowercased, trailing slash stripped.
|
|
/// `folder_path` is expected to already start (and end) with "/".
|
|
fn is_excluded(
|
|
excludes: &[String],
|
|
lab_name: &str,
|
|
folder_path: &str,
|
|
file_name: Option<&str>,
|
|
) -> bool {
|
|
if excludes.is_empty() {
|
|
return false;
|
|
}
|
|
let path = format!("/{}{}{}", lab_name, folder_path, file_name.unwrap_or(""))
|
|
.trim_end_matches('/')
|
|
.to_lowercase();
|
|
excludes.iter().any(|e| e == &path)
|
|
}
|
|
|
|
/// Return an absolute URL for a file download.
|
|
/// If the API returns a relative path, prepend the connection's base URL.
|
|
fn make_absolute_url(conn: &MDRSConnection, url: &str) -> String {
|
|
if url.starts_with("http") {
|
|
url.to_string()
|
|
} else {
|
|
format!(
|
|
"{}/{}",
|
|
conn.url.trim_end_matches('/'),
|
|
url.trim_start_matches('/')
|
|
)
|
|
}
|
|
}
|
|
|
|
struct DownloadFolderTaskResult {
|
|
child_folders: Vec<(String, PathBuf)>,
|
|
download_jobs: Vec<DownloadJob>,
|
|
}
|
|
|
|
struct DownloadJob {
|
|
url: String,
|
|
dest_path: PathBuf,
|
|
}
|
|
|
|
/// What every folder of one recursive download shares: the connection and its request
|
|
/// budget, the laboratory the paths are named against, what to leave out, the password
|
|
/// for locked folders, and whether files already on disk may be left alone.
|
|
#[derive(Clone)]
|
|
struct DownloadSession {
|
|
conn: Arc<MDRSConnection>,
|
|
limiter: ApiRequestLimiter,
|
|
lab_name: Arc<String>,
|
|
excludes: Arc<Vec<String>>,
|
|
password: Option<String>,
|
|
skip_if_exists: bool,
|
|
}
|
|
|
|
fn spawn_download_folder_task(
|
|
folder_tasks: &mut JoinSet<Result<DownloadFolderTaskResult, anyhow::Error>>,
|
|
session: DownloadSession,
|
|
folder_id: String,
|
|
local_dir: PathBuf,
|
|
) {
|
|
folder_tasks.spawn(async move { process_download_folder(session, folder_id, local_dir).await });
|
|
}
|
|
|
|
fn spawn_download_task(
|
|
download_tasks: &mut JoinSet<Result<(), anyhow::Error>>,
|
|
conn: Arc<MDRSConnection>,
|
|
limiter: ApiRequestLimiter,
|
|
job: DownloadJob,
|
|
) {
|
|
download_tasks.spawn(async move {
|
|
let dest_str = job.dest_path.to_string_lossy().to_string();
|
|
match conn
|
|
.download_file_limited(&job.url, &dest_str, &limiter)
|
|
.await
|
|
{
|
|
Ok(()) => {
|
|
println!("{}", job.dest_path.display());
|
|
Ok(())
|
|
}
|
|
Err(err) => {
|
|
// Nothing to clear up: a failed transfer writes only to its own scratch
|
|
// file, and removes that itself.
|
|
Err(anyhow!(
|
|
"Failed to download {}: {}",
|
|
job.dest_path.display(),
|
|
err
|
|
))
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
async fn process_download_folder(
|
|
session: DownloadSession,
|
|
folder_id: String,
|
|
local_dir: PathBuf,
|
|
) -> Result<DownloadFolderTaskResult, anyhow::Error> {
|
|
let DownloadSession {
|
|
conn,
|
|
limiter,
|
|
lab_name,
|
|
excludes,
|
|
password,
|
|
skip_if_exists,
|
|
} = session;
|
|
let folder = conn.retrieve_folder_limited(&folder_id, &limiter).await?;
|
|
|
|
if is_excluded(excludes.as_slice(), lab_name.as_str(), &folder.path, None) {
|
|
return Ok(DownloadFolderTaskResult {
|
|
child_folders: Vec::new(),
|
|
download_jobs: Vec::new(),
|
|
});
|
|
}
|
|
|
|
tokio::fs::create_dir_all(&local_dir).await?;
|
|
println!("{}", local_dir.display());
|
|
|
|
let dir_files = conn.list_all_files_limited(&folder_id, &limiter).await?;
|
|
let mut download_jobs = Vec::new();
|
|
for file in &dir_files {
|
|
if is_excluded(
|
|
excludes.as_slice(),
|
|
lab_name.as_str(),
|
|
&folder.path,
|
|
Some(&file.name),
|
|
) {
|
|
continue;
|
|
}
|
|
let dest_path = local_dir.join(&file.name);
|
|
if skip_if_exists
|
|
&& dest_path.exists()
|
|
&& let Ok(meta) = std::fs::metadata(&dest_path)
|
|
&& meta.len() == file.size
|
|
{
|
|
println!("{}", dest_path.display());
|
|
continue;
|
|
}
|
|
download_jobs.push(DownloadJob {
|
|
url: make_absolute_url(&conn, &file.download_url),
|
|
dest_path,
|
|
});
|
|
}
|
|
|
|
let mut child_folder_tasks: JoinSet<Result<Option<(String, PathBuf)>, anyhow::Error>> =
|
|
JoinSet::new();
|
|
for sub_folder in folder.sub_folders {
|
|
let conn = conn.clone();
|
|
let limiter = limiter.clone();
|
|
let password = password.clone();
|
|
let sub_local = local_dir.join(&sub_folder.name);
|
|
child_folder_tasks.spawn(async move {
|
|
if sub_folder.lock {
|
|
match password.as_deref() {
|
|
Some(pw) => {
|
|
if conn
|
|
.folder_auth_limited(&sub_folder.id, pw, &limiter)
|
|
.await
|
|
.is_err()
|
|
{
|
|
return Ok(None);
|
|
}
|
|
}
|
|
None => return Ok(None),
|
|
}
|
|
}
|
|
Ok(Some((sub_folder.id, sub_local)))
|
|
});
|
|
}
|
|
|
|
let mut child_folders = Vec::new();
|
|
while let Some(result) = child_folder_tasks.join_next().await {
|
|
if let Some(child) = flatten_join_result(result)? {
|
|
child_folders.push(child);
|
|
}
|
|
}
|
|
|
|
Ok(DownloadFolderTaskResult {
|
|
child_folders,
|
|
download_jobs,
|
|
})
|
|
}
|
|
|
|
async fn drive_download_tasks(
|
|
folder_tasks: &mut JoinSet<Result<DownloadFolderTaskResult, anyhow::Error>>,
|
|
download_tasks: &mut JoinSet<Result<(), anyhow::Error>>,
|
|
errors: &mut Vec<String>,
|
|
session: DownloadSession,
|
|
) {
|
|
loop {
|
|
match (folder_tasks.is_empty(), download_tasks.is_empty()) {
|
|
(true, true) => break,
|
|
(false, true) => {
|
|
if let Some(result) = folder_tasks.join_next().await {
|
|
handle_download_folder_result(
|
|
result,
|
|
folder_tasks,
|
|
download_tasks,
|
|
errors,
|
|
&session,
|
|
);
|
|
}
|
|
}
|
|
(true, false) => {
|
|
if let Some(result) = download_tasks.join_next().await
|
|
&& let Err(err) = flatten_join_result(result)
|
|
{
|
|
errors.push(err.to_string());
|
|
}
|
|
}
|
|
(false, false) => {
|
|
tokio::select! {
|
|
result = folder_tasks.join_next() => {
|
|
if let Some(result) = result {
|
|
handle_download_folder_result(
|
|
result,
|
|
folder_tasks,
|
|
download_tasks,
|
|
errors,
|
|
&session,
|
|
);
|
|
}
|
|
}
|
|
result = download_tasks.join_next() => {
|
|
if let Some(result) = result
|
|
&& let Err(err) = flatten_join_result(result) {
|
|
errors.push(err.to_string());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
fn handle_download_folder_result(
|
|
result: Result<Result<DownloadFolderTaskResult, anyhow::Error>, tokio::task::JoinError>,
|
|
folder_tasks: &mut JoinSet<Result<DownloadFolderTaskResult, anyhow::Error>>,
|
|
download_tasks: &mut JoinSet<Result<(), anyhow::Error>>,
|
|
errors: &mut Vec<String>,
|
|
session: &DownloadSession,
|
|
) {
|
|
match flatten_join_result(result) {
|
|
Ok(task_result) => {
|
|
for (folder_id, local_dir) in task_result.child_folders {
|
|
spawn_download_folder_task(folder_tasks, session.clone(), folder_id, local_dir);
|
|
}
|
|
for job in task_result.download_jobs {
|
|
spawn_download_task(
|
|
download_tasks,
|
|
session.conn.clone(),
|
|
session.limiter.clone(),
|
|
job,
|
|
);
|
|
}
|
|
}
|
|
Err(err) => errors.push(err.to_string()),
|
|
}
|
|
}
|
|
|
|
fn flatten_join_result<T>(
|
|
result: Result<Result<T, anyhow::Error>, tokio::task::JoinError>,
|
|
) -> Result<T, anyhow::Error> {
|
|
match result {
|
|
Ok(inner) => inner,
|
|
Err(err) => Err(anyhow!("Task join failed: {}", err)),
|
|
}
|
|
}
|