Initial cli

crates/pile/Cargo.toml

@@ -0,0 +1,27 @@
+[package]
+name = "pile"
+version = { workspace = true }
+rust-version = { workspace = true }
+edition = { workspace = true }
+
+[lints]
+workspace = true
+
+[dependencies]
+pile-toolbox = { workspace = true }
+pile-dataset = { workspace = true }
+pile-config = { workspace = true }
+
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true }
+tokio = { workspace = true }
+clap = { workspace = true }
+#clap_complete = { workspace = true }
+serde = { workspace = true }
+anyhow = { workspace = true }
+indicatif = { workspace = true }
+tracing-indicatif = { workspace = true }
+signal-hook = { workspace = true }
+anstyle = { workspace = true }
+toml = { workspace = true }
+tantivy = { workspace = true }

crates/pile/src/cli.rs

@@ -0,0 +1,74 @@
+use anstyle::{AnsiColor, Color, Style};
+
+pub fn clap_styles() -> clap::builder::Styles {
+	clap::builder::Styles::styled()
+		.usage(
+			Style::new()
+				.bold()
+				.fg_color(Some(Color::Ansi(AnsiColor::Blue))),
+		)
+		.header(
+			Style::new()
+				.bold()
+				.fg_color(Some(Color::Ansi(AnsiColor::Blue))),
+		)
+		.literal(
+			Style::new()
+				.bold()
+				.fg_color(Some(Color::Ansi(AnsiColor::BrightBlack))),
+		)
+		.invalid(
+			Style::new()
+				.bold()
+				.fg_color(Some(Color::Ansi(AnsiColor::Red))),
+		)
+		.error(
+			Style::new()
+				.bold()
+				.fg_color(Some(Color::Ansi(AnsiColor::Red))),
+		)
+		.valid(
+			Style::new()
+				.bold()
+				.underline()
+				.fg_color(Some(Color::Ansi(AnsiColor::Green))),
+		)
+		.placeholder(Style::new().fg_color(Some(Color::Ansi(AnsiColor::White))))
+}
+
+/*
+#[expect(clippy::unwrap_used)]
+pub fn progress_big() -> ProgressStyle {
+	return ProgressStyle::default_bar()
+		.template(
+			" {spinner:.green} [{elapsed_precise}] [{bar:40.green/dim}] {pos:>7}/{len:7} {msg:.dim} ({eta})",
+		)
+		.unwrap()
+		.progress_chars("=>-")
+		.tick_strings(&[
+			"⠉⠉", "⠈⠙", "⠀⠹", "⠀⢸", "⠀⣰", "⢀⣠", "⣀⣀", "⣄⡀", "⣆⠀", "⡇⠀", "⠏⠀", "⠋⠁", "⣏⣹",
+		]);
+}
+
+#[expect(clippy::unwrap_used)]
+pub fn spinner_small() -> ProgressStyle {
+	return ProgressStyle::default_bar()
+		.template("   {spinner:.red} {elapsed_precise:.dim} {msg:.dim}")
+		.unwrap()
+		.progress_chars("---")
+		.tick_strings(&["|", "/", "-", "\\", "|", "/", "-", "\\", "*"]);
+}
+
+#[expect(clippy::unwrap_used)]
+pub fn progress_bytes() -> ProgressStyle {
+	return ProgressStyle::default_bar()
+		.template(
+			"  {bar:16.red/white.dim} {elapsed_precise:.dim}  {bytes:>10}/{total_bytes:>10} {msg:.dim} ({eta})",
+		)
+		.unwrap()
+		.progress_chars("---")
+		.tick_strings(&[
+			"⠉⠉", "⠈⠙", "⠀⠹", "⠀⢸", "⠀⣰", "⢀⣠", "⣀⣀", "⣄⡀", "⣆⠀", "⡇⠀", "⠏⠀", "⠋⠁", "⣏⣹",
+		]);
+}
+*/

crates/pile/src/command/check.rs

@@ -0,0 +1,46 @@
+use anyhow::{Context, Result, anyhow};
+use clap::Args;
+use pile_config::ConfigToml;
+use pile_toolbox::cancelabletask::{CancelFlag, CancelableTaskError};
+use std::{fmt::Debug, path::PathBuf};
+use tracing::{debug, error, info};
+
+use crate::{CliCmd, GlobalContext};
+
+#[derive(Debug, Args)]
+pub struct CheckCommand {
+	/// Path to dataset config
+	#[arg(long, short = 'c', default_value = "./pile.toml")]
+	config: PathBuf,
+}
+
+impl CliCmd for CheckCommand {
+	async fn run(
+		self,
+		_ctx: GlobalContext,
+		_flag: CancelFlag,
+	) -> Result<i32, CancelableTaskError<anyhow::Error>> {
+		debug!("Checking {}", self.config.display());
+
+		if !self.config.exists() {
+			return Err(anyhow!("{} does not exist", self.config.display()).into());
+		}
+
+		let config = std::fs::read_to_string(&self.config)
+			.with_context(|| format!("while reading {}", self.config.display()))?;
+		let config: Result<ConfigToml, _> = toml::from_str(&config);
+
+		match config {
+			Ok(config) => {
+				info!("Config in {} is valid", self.config.display());
+				debug!("{config:#?}");
+				return Ok(0);
+			}
+
+			Err(error) => {
+				error!("{} is invalid:\n{error}", self.config.display());
+				return Ok(1);
+			}
+		}
+	}
+}

crates/pile/src/command/index.rs

@@ -0,0 +1,110 @@
+use anyhow::{Context, Result, anyhow};
+use clap::Args;
+use pile_config::{ConfigToml, Source};
+use pile_dataset::{DataSource, index::DbFtsIndex, source::DirDataSource};
+use pile_toolbox::cancelabletask::{CancelFlag, CancelableTaskError};
+use std::{fmt::Debug, path::PathBuf};
+use tantivy::{Executor, Index, IndexWriter};
+use tracing::{error, info, trace, warn};
+
+use crate::{CliCmd, GlobalContext};
+
+#[derive(Debug, Args)]
+pub struct IndexCommand {
+	/// Path to dataset config
+	#[arg(long, short = 'c', default_value = "./pile.toml")]
+	config: PathBuf,
+}
+
+impl CliCmd for IndexCommand {
+	async fn run(
+		self,
+		_ctx: GlobalContext,
+		flag: CancelFlag,
+	) -> Result<i32, CancelableTaskError<anyhow::Error>> {
+		let parent = self
+			.config
+			.parent()
+			.with_context(|| format!("Config file {} has no parent", self.config.display()))?;
+
+		if !self.config.exists() {
+			return Err(anyhow!("{} does not exist", self.config.display()).into());
+		}
+
+		let config = {
+			let config = std::fs::read_to_string(&self.config)
+				.with_context(|| format!("while reading {}", self.config.display()))?;
+			let config: Result<ConfigToml, _> = toml::from_str(&config);
+
+			match config {
+				Ok(config) => {
+					trace!(message = "Loaded config", ?config);
+					config
+				}
+
+				Err(error) => {
+					error!("{} is invalid:\n{error}", self.config.display());
+					return Ok(1);
+				}
+			}
+		};
+
+		let working_dir = config
+			.dataset
+			.working_dir
+			.clone()
+			.unwrap_or(parent.join(".pile"))
+			.join(&config.dataset.name);
+		let fts_dir = working_dir.join("fts");
+
+		if fts_dir.is_dir() {
+			warn!("Removing existing index in {}", fts_dir.display());
+			std::fs::remove_dir_all(&fts_dir)
+				.with_context(|| format!("while removing {}", fts_dir.display()))?;
+		}
+
+		std::fs::create_dir_all(&fts_dir)
+			.with_context(|| format!("while creating dir {}", fts_dir.display()))?;
+
+		let mut sources = Vec::new();
+		for (name, source) in &config.dataset.source {
+			match source {
+				Source::Flac { path: dir } => {
+					let source = DirDataSource::new(name, dir.clone().to_vec());
+					sources.push(source);
+				}
+			}
+		}
+
+		let db_index = DbFtsIndex::new(&fts_dir, &config);
+		let mut index = Index::create_in_dir(&fts_dir, db_index.schema.clone()).unwrap();
+		index.set_executor(Executor::multi_thread(10, "build-fts").unwrap());
+		let mut index_writer: IndexWriter = index.writer(15_000_000).unwrap();
+
+		for s in sources {
+			info!("Processing source {:?}", s.name);
+
+			for i in s.iter() {
+				let (k, v) = i.unwrap();
+
+				let doc = match db_index.entry_to_document(&*v).unwrap() {
+					Some(x) => x,
+					None => {
+						warn!("Skipping {k:?}, document is empty");
+						continue;
+					}
+				};
+				index_writer.add_document(doc).unwrap();
+
+				if flag.is_cancelled() {
+					return Err(CancelableTaskError::Cancelled);
+				}
+			}
+		}
+
+		info!("Committing index");
+		index_writer.commit().unwrap();
+
+		return Ok(0);
+	}
+}

crates/pile/src/command/init.rs

@@ -0,0 +1,34 @@
+use anyhow::{Context, Result, anyhow};
+use clap::Args;
+use pile_config::INIT_DB_TOML;
+use pile_toolbox::cancelabletask::{CancelFlag, CancelableTaskError};
+use std::{fmt::Debug, fs::File, io::Write, path::PathBuf};
+
+use crate::{CliCmd, GlobalContext};
+
+#[derive(Debug, Args)]
+pub struct InitCommand {
+	/// Path to dataset config
+	#[arg(long, short = 'c', default_value = "./pile.toml")]
+	config: PathBuf,
+}
+
+impl CliCmd for InitCommand {
+	async fn run(
+		self,
+		_ctx: GlobalContext,
+		_flag: CancelFlag,
+	) -> Result<i32, CancelableTaskError<anyhow::Error>> {
+		if self.config.exists() {
+			return Err(anyhow!("{} already exists", self.config.display()).into());
+		}
+
+		let mut file = File::create(&self.config)
+			.with_context(|| format!("while creating {}", self.config.display()))?;
+
+		file.write_all(INIT_DB_TOML.as_bytes())
+			.with_context(|| format!("while writing {}", self.config.display()))?;
+
+		Ok(0)
+	}
+}

crates/pile/src/command/lookup.rs

@@ -0,0 +1,91 @@
+use anyhow::{Context, Result, anyhow};
+use clap::Args;
+use pile_config::ConfigToml;
+use pile_dataset::index::DbFtsIndex;
+use pile_toolbox::cancelabletask::{CancelFlag, CancelableTaskError};
+use std::{fmt::Debug, path::PathBuf, sync::Arc};
+use tantivy::collector::TopDocs;
+use tracing::{error, trace};
+
+use crate::{CliCmd, GlobalContext};
+
+#[derive(Debug, Args)]
+pub struct LookupCommand {
+	query: String,
+
+	/// Show each result's score
+	#[arg(long)]
+	score: bool,
+
+	/// Number of results to return
+	#[arg(long, short = 'n', default_value = "20")]
+	topn: usize,
+
+	/// Path to dataset config
+	#[arg(long, short = 'c', default_value = "./pile.toml")]
+	config: PathBuf,
+}
+
+impl CliCmd for LookupCommand {
+	async fn run(
+		self,
+		_ctx: GlobalContext,
+		_flag: CancelFlag,
+	) -> Result<i32, CancelableTaskError<anyhow::Error>> {
+		let parent = self
+			.config
+			.parent()
+			.with_context(|| format!("Config file {} has no parent", self.config.display()))?;
+
+		if !self.config.exists() {
+			return Err(anyhow!("{} does not exist", self.config.display()).into());
+		}
+
+		let config = {
+			let config = std::fs::read_to_string(&self.config)
+				.with_context(|| format!("while reading {}", self.config.display()))?;
+			let config: Result<ConfigToml, _> = toml::from_str(&config);
+
+			match config {
+				Ok(config) => {
+					trace!(message = "Loaded config", ?config);
+					config
+				}
+
+				Err(error) => {
+					error!("{} is invalid:\n{error}", self.config.display());
+					return Ok(1);
+				}
+			}
+		};
+
+		let working_dir = config
+			.dataset
+			.working_dir
+			.clone()
+			.unwrap_or(parent.join(".pile"))
+			.join(&config.dataset.name);
+		let fts_dir = working_dir.join("fts");
+
+		if !fts_dir.is_dir() {
+			return Err(anyhow!("fts index does not exist").into());
+		}
+
+		let db_index = DbFtsIndex::new(&fts_dir, &config);
+		let results = db_index
+			.lookup(self.query, Arc::new(TopDocs::with_limit(self.topn)))
+			.unwrap();
+
+		if self.score {
+			for res in results {
+				println!("{:0.02} {}", res.score, res.key);
+			}
+		} else {
+			for res in results {
+				println!("{}", res.key);
+			}
+		}
+
+		return Ok(0);
+	}
+}

crates/pile/src/command/mod.rs

@@ -0,0 +1,66 @@
+use anyhow::Result;
+use clap::Subcommand;
+use pile_toolbox::cancelabletask::{CancelFlag, CancelableTask, CancelableTaskError};
+
+mod check;
+mod index;
+mod init;
+mod lookup;
+
+use crate::GlobalContext;
+
+#[derive(Debug, Subcommand)]
+pub enum SubCommand {
+	/// Create an empty dataset
+	Init {
+		#[command(flatten)]
+		cmd: init::InitCommand,
+	},
+
+	/// Check dataset config
+	Check {
+		#[command(flatten)]
+		cmd: check::CheckCommand,
+	},
+
+	/// Rebuild all indices
+	Index {
+		#[command(flatten)]
+		cmd: index::IndexCommand,
+	},
+
+	/// Search all sources
+	Lookup {
+		#[command(flatten)]
+		cmd: lookup::LookupCommand,
+	},
+}
+
+impl CliCmdDispatch for SubCommand {
+	fn start(self, ctx: GlobalContext) -> Result<CancelableTask<Result<i32>>> {
+		match self {
+			Self::Init { cmd } => cmd.start(ctx),
+			Self::Check { cmd } => cmd.start(ctx),
+			Self::Index { cmd } => cmd.start(ctx),
+			Self::Lookup { cmd } => cmd.start(ctx),
+		}
+	}
+}
+
+pub(super) trait CliCmdDispatch: Sized + Send + 'static {
+	fn start(self, ctx: GlobalContext) -> Result<CancelableTask<Result<i32>>>;
+}
+
+impl<T: CliCmd> CliCmdDispatch for T {
+	fn start(self, ctx: GlobalContext) -> Result<CancelableTask<Result<i32>>> {
+		Ok(CancelableTask::spawn(|flag| self.run(ctx, flag)))
+	}
+}
+
+pub(super) trait CliCmd: Sized + Send + 'static {
+	fn run(
+		self,
+		ctx: GlobalContext,
+		flag: CancelFlag,
+	) -> impl std::future::Future<Output = Result<i32, CancelableTaskError<anyhow::Error>>> + Send;
+}

crates/pile/src/config/logging.rs

@@ -0,0 +1,129 @@
+use clap::ValueEnum;
+use serde::Deserialize;
+use std::{fmt::Display, str::FromStr};
+use tracing_subscriber::EnvFilter;
+
+#[derive(Debug, Default)]
+pub enum LogLevel {
+	Trace,
+	Debug,
+	#[default]
+	Info,
+	Warn,
+	Error,
+}
+
+impl Display for LogLevel {
+	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+		match self {
+			Self::Trace => write!(f, "trace"),
+			Self::Debug => write!(f, "debug"),
+			Self::Info => write!(f, "info"),
+			Self::Warn => write!(f, "warn"),
+			Self::Error => write!(f, "error"),
+		}
+	}
+}
+
+#[derive(Debug, Deserialize, Clone, Copy, ValueEnum)]
+pub enum LoggingPreset {
+	Error,
+	Warn,
+	Info,
+	Debug,
+	Trace,
+}
+
+pub struct LoggingConfig {
+	other: LogLevel,
+
+	pile: LogLevel,
+	pile_audio: LogLevel,
+	pile_config: LogLevel,
+	pile_dataset: LogLevel,
+	pile_toolbox: LogLevel,
+}
+
+impl From<LoggingConfig> for EnvFilter {
+	fn from(conf: LoggingConfig) -> Self {
+		// Should never fail
+		#[expect(clippy::unwrap_used)]
+		EnvFilter::from_str(
+			&[
+				// Fixed sources
+				format!("html5ever={}", LogLevel::Error),
+				// Configurable sources
+				format!("pile={}", conf.pile),
+				format!("pile_audio={}", conf.pile_audio),
+				format!("pile_config={}", conf.pile_config),
+				format!("pile_dataset={}", conf.pile_dataset),
+				format!("pile_toolbox={}", conf.pile_toolbox),
+				conf.other.to_string(),
+			]
+			.join(","),
+		)
+		.unwrap()
+	}
+}
+
+impl Default for LoggingPreset {
+	fn default() -> Self {
+		return Self::Info;
+	}
+}
+
+impl LoggingPreset {
+	pub fn get_config(&self) -> LoggingConfig {
+		match self {
+			Self::Error => LoggingConfig {
+				other: LogLevel::Error,
+
+				pile: LogLevel::Error,
+				pile_audio: LogLevel::Error,
+				pile_config: LogLevel::Error,
+				pile_dataset: LogLevel::Error,
+				pile_toolbox: LogLevel::Error,
+			},
+
+			Self::Warn => LoggingConfig {
+				other: LogLevel::Error,
+
+				pile: LogLevel::Warn,
+				pile_audio: LogLevel::Warn,
+				pile_config: LogLevel::Warn,
+				pile_dataset: LogLevel::Warn,
+				pile_toolbox: LogLevel::Warn,
+			},
+
+			Self::Info => LoggingConfig {
+				other: LogLevel::Warn,
+
+				pile: LogLevel::Info,
+				pile_audio: LogLevel::Info,
+				pile_config: LogLevel::Info,
+				pile_dataset: LogLevel::Info,
+				pile_toolbox: LogLevel::Info,
+			},
+
+			Self::Debug => LoggingConfig {
+				other: LogLevel::Warn,
+
+				pile: LogLevel::Debug,
+				pile_audio: LogLevel::Debug,
+				pile_config: LogLevel::Debug,
+				pile_dataset: LogLevel::Debug,
+				pile_toolbox: LogLevel::Debug,
+			},
+
+			Self::Trace => LoggingConfig {
+				other: LogLevel::Trace,
+
+				pile: LogLevel::Trace,
+				pile_audio: LogLevel::Trace,
+				pile_config: LogLevel::Trace,
+				pile_dataset: LogLevel::Trace,
+				pile_toolbox: LogLevel::Trace,
+			},
+		}
+	}
+}

crates/pile/src/config/mod.rs

@@ -0,0 +1,2 @@
+mod logging;
+pub use logging::*;

crates/pile/src/main.rs

@@ -0,0 +1,128 @@
+use std::process::ExitCode;
+
+use anyhow::{Context, Result};
+use clap::Parser;
+use config::LoggingPreset;
+use indicatif::MultiProgress;
+use pile_toolbox::cancelabletask::CancelableTaskResult;
+use tracing::{error, warn};
+use tracing_indicatif::{IndicatifWriter, writer::Stderr};
+use tracing_subscriber::fmt::MakeWriter;
+
+use crate::{
+	command::{CliCmd, CliCmdDispatch, SubCommand},
+	signal::start_signal_task,
+};
+
+// TODO:
+//
+// Full FLAC fts support
+// - index threading
+// - validate names: dataset
+// - validate names: field
+// - error if dir not relative
+// - check if up-to-date
+// - api
+
+mod cli;
+mod command;
+mod config;
+mod signal;
+
+#[derive(Parser, Debug)]
+#[command(version, about, long_about = None, styles=cli::clap_styles())]
+struct Cli {
+	/// Increase verbosity (can be repeated)
+	#[arg(short, action = clap::ArgAction::Count,global = true)]
+	v: u8,
+
+	/// Decrease verbosity (can be repeated)
+	#[arg(short, action = clap::ArgAction::Count, global = true)]
+	q: u8,
+
+	#[command(subcommand)]
+	cmd: SubCommand,
+}
+
+#[derive(Clone)]
+pub struct GlobalContext {
+	#[expect(dead_code)]
+	mp: MultiProgress,
+}
+
+fn main() -> ExitCode {
+	#[expect(clippy::unwrap_used)]
+	let rt = tokio::runtime::Builder::new_multi_thread()
+		.enable_all()
+		.worker_threads(10)
+		.build()
+		.unwrap();
+
+	match rt.block_on(main_inner()) {
+		Ok(code) => {
+			std::process::exit(code);
+		}
+		Err(err) => {
+			for e in err.chain() {
+				error!("{}", e);
+			}
+			std::process::exit(1);
+		}
+	}
+}
+
+async fn main_inner() -> Result<i32> {
+	let cli = Cli::parse();
+
+	let level_i: i16 = cli.v as i16 - cli.q as i16;
+
+	let level;
+	if level_i <= -2 {
+		level = LoggingPreset::Error
+	} else if level_i == -1 {
+		level = LoggingPreset::Warn
+	} else if level_i == 0 {
+		level = LoggingPreset::Info
+	} else if level_i == 1 {
+		level = LoggingPreset::Debug
+	} else if level_i >= 2 {
+		level = LoggingPreset::Trace
+	} else {
+		unreachable!()
+	}
+
+	let mp = MultiProgress::new();
+	let writer: IndicatifWriter<Stderr> = IndicatifWriter::new(mp.clone());
+
+	tracing_subscriber::fmt()
+		.with_env_filter(level.get_config())
+		.without_time()
+		.with_ansi(true)
+		.with_writer(writer.make_writer())
+		.init();
+
+	let ctx = GlobalContext { mp };
+
+	let task = cli.cmd.start(ctx).context("while starting task")?;
+	let signal_task =
+		start_signal_task(task.flag().clone()).context("while starting signal task")?;
+
+	match task.join().await {
+		Ok(CancelableTaskResult::Finished(Ok(code))) => Ok(code),
+		Ok(CancelableTaskResult::Cancelled) => {
+			signal_task.abort();
+			warn!("Task cancelled successfully");
+			Ok(1)
+		}
+
+		Err(err) => {
+			signal_task.abort();
+			Err(err).context("while joining task")
+		}
+
+		Ok(CancelableTaskResult::Finished(Err(err))) => {
+			signal_task.abort();
+			Err(err).context("while running task")
+		}
+	}
+}

crates/pile/src/signal.rs

@@ -0,0 +1,24 @@
+use anyhow::{Context, Result};
+use pile_toolbox::cancelabletask::CancelFlag;
+use signal_hook::{consts::TERM_SIGNALS, iterator::Signals};
+use tokio::task::JoinHandle;
+use tracing::warn;
+
+/// Start an async task that listens for OS signals,
+/// setting `should_exit` to `true` when an exit signal
+/// is caught.
+pub fn start_signal_task(flag: CancelFlag) -> Result<JoinHandle<()>> {
+	let mut signals = Signals::new(TERM_SIGNALS).context("Failed to initialize signal handler")?;
+
+	let task = tokio::task::spawn_blocking(move || {
+		for sig in signals.forever() {
+			if TERM_SIGNALS.contains(&sig) {
+				warn!("Received signal {sig}, trying to exit cleanly");
+				flag.cancel();
+				break;
+			}
+		}
+	});
+
+	return Ok(task);
+}