use crate::db::Hash;
use crate::errors::AppError;
use crate::fs::DirectoryReader;
use crate::options::{CacheOptions, Options, QueueAddArgs, SharedOptions};
use crate::queue::{Queue, QueueItem, QueueStatus};
use colored::Colorize;
use di::{injectable, Ref, RefMut};
use log::{info, trace};
use std::collections::BTreeMap;
use std::fs::File;
use std::io::BufReader;
use std::path::PathBuf;
#[injectable]
pub struct QueueAddCommand {
shared_options: Ref<SharedOptions>,
cache_options: Ref<CacheOptions>,
args: Ref<QueueAddArgs>,
queue: RefMut<Queue>,
}
impl QueueAddCommand {
pub async fn execute_cli(&mut self) -> Result<bool, AppError> {
if !self.shared_options.validate()
|| !self.cache_options.validate()
|| !self.args.validate()
{
return Ok(false);
}
let path = self
.args
.queue_add_path
.clone()
.expect("source should be set");
let status = self.execute(path).await?;
info!("{} {} items to the queue", "Added".bold(), status.added);
trace!(
"{} {} items already in the queue",
"Excluded".bold(),
status.excluded
);
Ok(true)
}
async fn execute(&mut self, path: PathBuf) -> Result<QueueStatus, AppError> {
if path.is_dir() {
self.execute_directory(path).await
} else if path.is_file() {
self.execute_file(path).await
} else {
AppError::explained(
"add to queue",
format!("Does not exist: {}", path.display()),
)
}
}
async fn execute_directory(&mut self, path: PathBuf) -> Result<QueueStatus, AppError> {
let mut queue = self.queue.write().expect("queue should be writeable");
let existing_paths: Vec<PathBuf> = queue
.get_all()
.await?
.values()
.map(|x| x.path.clone())
.collect();
trace!(
"{} {} existing paths",
"Skipping".bold(),
existing_paths.len()
);
trace!("Reading torrent directory: {}", path.display());
let paths = DirectoryReader::new()
.with_extension("torrent")
.with_max_depth(0)
.read(&path)
.or_else(|e| AppError::io(e, "read torrent directory"))?;
let found = paths.len();
trace!("{} {} torrent files", "Found".bold(), found);
let paths: Vec<PathBuf> = paths
.into_iter()
.filter(|x| !existing_paths.contains(x))
.collect();
let remaining = paths.len();
info!("{} {} new torrent files", "Found".bold(), remaining);
if remaining > 250 {
info!("This may take a while");
}
let added = queue.insert_new_torrent_files(paths).await?;
Ok(QueueStatus {
success: true,
added,
excluded: found - added,
})
}
async fn execute_file(&mut self, path: PathBuf) -> Result<QueueStatus, AppError> {
trace!("Reading queue file: {}", path.display());
let file = File::open(path).or_else(|e| AppError::io(e, "open chunk file"))?;
let reader = BufReader::new(file);
let items: BTreeMap<Hash<20>, QueueItem> =
serde_yaml::from_reader(reader).or_else(|e| AppError::yaml(e, "deserialize chunk"))?;
let found = items.len();
info!("{} {} items", "Found".bold(), found);
if found > 250 {
info!("This may take a while");
}
let queue = self.queue.write().expect("queue should be writeable");
let added = queue.set_many(items, true).await?;
Ok(QueueStatus {
success: true,
added,
excluded: found - added,
})
}
}