Crude frontend added for controlling and monitoring indexing.
This commit is contained in:
parent
dcc9160548
commit
8248517d42
5 changed files with 394 additions and 66 deletions
16
Cargo.lock
generated
16
Cargo.lock
generated
|
|
@ -3611,16 +3611,6 @@ version = "1.13.2"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3c5e1a9a646d36c3599cd173a41282daf47c44583ad367b8e6837255952e5c67"
|
||||
|
||||
[[package]]
|
||||
name = "socket2"
|
||||
version = "0.5.8"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c970269d99b64e60ec3bd6ad27270092a5394c4e309314b18ae3fe575695fbe8"
|
||||
dependencies = [
|
||||
"libc",
|
||||
"windows-sys 0.52.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "soup3"
|
||||
version = "0.5.0"
|
||||
|
|
@ -3903,15 +3893,9 @@ checksum = "ba4f4a02a7a80d6f274636f0aa95c7e383b912d41fe721a31f29e29698585a4a"
|
|||
dependencies = [
|
||||
"backtrace",
|
||||
"bytes",
|
||||
"libc",
|
||||
"mio",
|
||||
"num_cpus",
|
||||
"parking_lot",
|
||||
"pin-project-lite",
|
||||
"signal-hook-registry",
|
||||
"socket2",
|
||||
"tokio-macros",
|
||||
"windows-sys 0.48.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
|
|||
|
|
@ -13,4 +13,4 @@ tqdm = "0.7.0"
|
|||
walkdir = "2.5.0"
|
||||
zip = "0.6"
|
||||
quick-xml = "0.31"
|
||||
tokio = { version = "1.0", features = ["full"] }
|
||||
tokio = { version = "1.0", features = ["time"] }
|
||||
|
|
|
|||
188
src/frontend.rs
188
src/frontend.rs
|
|
@ -1,15 +1,191 @@
|
|||
#![allow(non_snake_case)]
|
||||
|
||||
use std::sync::Arc;
|
||||
use dioxus::prelude::*;
|
||||
use crate::indexing::{IndexingService, IndexingStatus};
|
||||
|
||||
#[derive(Props, Clone)]
|
||||
pub struct AppProps {
|
||||
pub indexing_service: Arc<IndexingService>,
|
||||
}
|
||||
|
||||
// define a component that renders a div with the text "Hello, world!"
|
||||
pub fn App() -> Element {
|
||||
let mut count = use_signal(|| 0);
|
||||
impl PartialEq for AppProps {
|
||||
fn eq(&self, other: &Self) -> bool {
|
||||
Arc::ptr_eq(&self.indexing_service, &other.indexing_service)
|
||||
}
|
||||
}
|
||||
|
||||
pub fn App(props: AppProps) -> Element {
|
||||
let mut indexing_path = use_signal(|| "C:\\".to_string());
|
||||
let mut db_path = use_signal(|| "QuickSearch.db".to_string());
|
||||
let mut status_text = use_signal(|| "Idle".to_string());
|
||||
|
||||
let indexing_service_for_start = props.indexing_service.clone();
|
||||
let indexing_service_for_stop = props.indexing_service.clone();
|
||||
let indexing_service_for_refresh = props.indexing_service.clone();
|
||||
let indexing_service_for_timer = props.indexing_service.clone();
|
||||
|
||||
// Manual status refresh function
|
||||
let refresh_status = move |_| {
|
||||
let status = indexing_service_for_refresh.get_status();
|
||||
let status_str = match status {
|
||||
IndexingStatus::Idle => "Idle".to_string(),
|
||||
IndexingStatus::Running { files_processed, total_files, current_file, start_time } => {
|
||||
let elapsed = start_time.elapsed();
|
||||
let current_file_display = current_file
|
||||
.as_ref()
|
||||
.map(|f| format!("Current: {}", f.split('\\').last().unwrap_or(f)))
|
||||
.unwrap_or_default();
|
||||
|
||||
if let Some(total) = total_files {
|
||||
let percentage = if total > 0 {
|
||||
(files_processed as f64 / total as f64 * 100.0) as u32
|
||||
} else { 0 };
|
||||
format!(
|
||||
"Running: {}/{} files ({}%) - {:.1}s elapsed\n{}",
|
||||
files_processed,
|
||||
total,
|
||||
percentage,
|
||||
elapsed.as_secs_f64(),
|
||||
current_file_display
|
||||
)
|
||||
} else {
|
||||
format!(
|
||||
"Running: {} files processed - {:.1}s elapsed\n{}",
|
||||
files_processed,
|
||||
elapsed.as_secs_f64(),
|
||||
current_file_display
|
||||
)
|
||||
}
|
||||
}
|
||||
IndexingStatus::Stopping => "Stopping...".to_string(),
|
||||
IndexingStatus::Error(ref e) => format!("Error: {}", e),
|
||||
};
|
||||
status_text.set(status_str);
|
||||
};
|
||||
|
||||
// Automatic status updates every second
|
||||
{
|
||||
let mut status_text_clone = status_text.clone();
|
||||
let service_clone = indexing_service_for_timer.clone();
|
||||
use_future(move || {
|
||||
let service = service_clone.clone();
|
||||
async move {
|
||||
loop {
|
||||
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
|
||||
|
||||
let status = service.get_status();
|
||||
let status_str = match status {
|
||||
IndexingStatus::Idle => "Idle".to_string(),
|
||||
IndexingStatus::Running { files_processed, total_files, current_file, start_time } => {
|
||||
let elapsed = start_time.elapsed();
|
||||
let current_file_display = current_file
|
||||
.as_ref()
|
||||
.map(|f| format!("Current: {}", f.split('\\').last().unwrap_or(f)))
|
||||
.unwrap_or_default();
|
||||
|
||||
if let Some(total) = total_files {
|
||||
let percentage = if total > 0 {
|
||||
(files_processed as f64 / total as f64 * 100.0) as u32
|
||||
} else { 0 };
|
||||
format!(
|
||||
"Running: {}/{} files ({}%) - {:.1}s elapsed\n{}",
|
||||
files_processed,
|
||||
total,
|
||||
percentage,
|
||||
elapsed.as_secs_f64(),
|
||||
current_file_display
|
||||
)
|
||||
} else {
|
||||
format!(
|
||||
"Running: {} files processed - {:.1}s elapsed\n{}",
|
||||
files_processed,
|
||||
elapsed.as_secs_f64(),
|
||||
current_file_display
|
||||
)
|
||||
}
|
||||
}
|
||||
IndexingStatus::Stopping => "Stopping...".to_string(),
|
||||
IndexingStatus::Error(ref e) => format!("Error: {}", e),
|
||||
};
|
||||
status_text_clone.set(status_str);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
rsx! {
|
||||
h1 { "High-Five counter: {count}" }
|
||||
button { onclick: move |_| count += 1, "Up high!" }
|
||||
button { onclick: move |_| count -= 1, "Down low!" }
|
||||
div {
|
||||
style: "padding: 20px; font-family: Arial, sans-serif;",
|
||||
|
||||
h1 { "QuickSearch File Indexer" }
|
||||
|
||||
div {
|
||||
style: "margin-bottom: 20px;",
|
||||
h2 { "Indexing Controls" }
|
||||
|
||||
div {
|
||||
style: "margin-bottom: 10px;",
|
||||
label {
|
||||
style: "display: block; margin-bottom: 5px;",
|
||||
"Path to index:"
|
||||
}
|
||||
input {
|
||||
style: "width: 400px; padding: 5px;",
|
||||
r#type: "text",
|
||||
value: "{indexing_path}",
|
||||
oninput: move |evt| indexing_path.set(evt.value())
|
||||
}
|
||||
}
|
||||
|
||||
div {
|
||||
style: "margin-bottom: 10px;",
|
||||
label {
|
||||
style: "display: block; margin-bottom: 5px;",
|
||||
"Database path:"
|
||||
}
|
||||
input {
|
||||
style: "width: 400px; padding: 5px;",
|
||||
r#type: "text",
|
||||
value: "{db_path}",
|
||||
oninput: move |evt| db_path.set(evt.value())
|
||||
}
|
||||
}
|
||||
|
||||
div {
|
||||
style: "margin-bottom: 20px;",
|
||||
button {
|
||||
style: "margin-right: 10px; padding: 10px 20px; background-color: #4CAF50; color: white; border: none; cursor: pointer;",
|
||||
onclick: move |_| {
|
||||
let _ = indexing_service_for_start.start_indexing(
|
||||
indexing_path().clone(),
|
||||
db_path().clone()
|
||||
);
|
||||
},
|
||||
"Start Indexing"
|
||||
}
|
||||
button {
|
||||
style: "padding: 10px 20px; background-color: #f44336; color: white; border: none; cursor: pointer;",
|
||||
onclick: move |_| {
|
||||
let _ = indexing_service_for_stop.stop_indexing();
|
||||
},
|
||||
"Stop Indexing"
|
||||
}
|
||||
button {
|
||||
style: "margin-left: 10px; padding: 10px 20px; background-color: #2196F3; color: white; border: none; cursor: pointer;",
|
||||
onclick: refresh_status,
|
||||
"Refresh Status"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
div {
|
||||
h2 { "Status" }
|
||||
pre {
|
||||
style: "background-color: #f5f5f5; padding: 10px; border-radius: 5px; font-family: monospace;",
|
||||
"{status_text}"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
196
src/indexing.rs
Normal file
196
src/indexing.rs
Normal file
|
|
@ -0,0 +1,196 @@
|
|||
use std::sync::{Arc, Mutex, mpsc};
|
||||
use std::thread;
|
||||
use std::time::{Duration, Instant};
|
||||
use walkdir::WalkDir;
|
||||
use rusqlite::Connection;
|
||||
|
||||
use crate::file_handling;
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum IndexingStatus {
|
||||
Idle,
|
||||
Running {
|
||||
files_processed: usize,
|
||||
total_files: Option<usize>,
|
||||
current_file: Option<String>,
|
||||
start_time: Instant,
|
||||
},
|
||||
Stopping,
|
||||
Error(String),
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum IndexingCommand {
|
||||
Start {
|
||||
path: String,
|
||||
db_path: String,
|
||||
},
|
||||
Stop,
|
||||
}
|
||||
|
||||
pub struct IndexingService {
|
||||
status: Arc<Mutex<IndexingStatus>>,
|
||||
command_tx: mpsc::Sender<IndexingCommand>,
|
||||
_handle: thread::JoinHandle<()>,
|
||||
}
|
||||
|
||||
impl IndexingService {
|
||||
pub fn new() -> Self {
|
||||
let status = Arc::new(Mutex::new(IndexingStatus::Idle));
|
||||
let (command_tx, command_rx) = mpsc::channel();
|
||||
|
||||
let status_clone = status.clone();
|
||||
let handle = thread::spawn(move || {
|
||||
Self::indexing_thread(status_clone, command_rx);
|
||||
});
|
||||
|
||||
IndexingService {
|
||||
status,
|
||||
command_tx,
|
||||
_handle: handle,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn start_indexing(&self, path: String, db_path: String) -> Result<(), String> {
|
||||
self.command_tx
|
||||
.send(IndexingCommand::Start { path, db_path })
|
||||
.map_err(|e| format!("Failed to send start command: {}", e))
|
||||
}
|
||||
|
||||
pub fn stop_indexing(&self) -> Result<(), String> {
|
||||
self.command_tx
|
||||
.send(IndexingCommand::Stop)
|
||||
.map_err(|e| format!("Failed to send stop command: {}", e))
|
||||
}
|
||||
|
||||
pub fn get_status(&self) -> IndexingStatus {
|
||||
self.status.lock().unwrap().clone()
|
||||
}
|
||||
|
||||
fn indexing_thread(status: Arc<Mutex<IndexingStatus>>, command_rx: mpsc::Receiver<IndexingCommand>) {
|
||||
let stop_flag = Arc::new(Mutex::new(false));
|
||||
|
||||
while let Ok(command) = command_rx.recv() {
|
||||
match command {
|
||||
IndexingCommand::Start { path, db_path } => {
|
||||
if matches!(*status.lock().unwrap(), IndexingStatus::Running { .. }) {
|
||||
continue; // Already running
|
||||
}
|
||||
|
||||
*stop_flag.lock().unwrap() = false;
|
||||
*status.lock().unwrap() = IndexingStatus::Running {
|
||||
files_processed: 0,
|
||||
total_files: None,
|
||||
current_file: None,
|
||||
start_time: Instant::now(),
|
||||
};
|
||||
|
||||
// Run indexing
|
||||
if let Err(e) = Self::run_indexing(&status, &path, &db_path, &stop_flag) {
|
||||
*status.lock().unwrap() = IndexingStatus::Error(e);
|
||||
} else {
|
||||
*status.lock().unwrap() = IndexingStatus::Idle;
|
||||
}
|
||||
}
|
||||
IndexingCommand::Stop => {
|
||||
if matches!(*status.lock().unwrap(), IndexingStatus::Running { .. }) {
|
||||
*status.lock().unwrap() = IndexingStatus::Stopping;
|
||||
*stop_flag.lock().unwrap() = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn run_indexing(
|
||||
status: &Arc<Mutex<IndexingStatus>>,
|
||||
path: &str,
|
||||
db_path: &str,
|
||||
stop_flag: &Arc<Mutex<bool>>,
|
||||
) -> Result<(), String> {
|
||||
// Set up database
|
||||
let conn = Connection::open(db_path)
|
||||
.map_err(|e| format!("Failed to open database: {}", e))?;
|
||||
|
||||
conn.execute_batch(
|
||||
"PRAGMA journal_mode = OFF;
|
||||
PRAGMA synchronous = 0;
|
||||
PRAGMA cache_size = 10000;
|
||||
PRAGMA temp_store = MEMORY;",
|
||||
)
|
||||
.map_err(|e| format!("Failed to set PRAGMA: {}", e))?;
|
||||
|
||||
conn.execute(
|
||||
"CREATE TABLE IF NOT EXISTS files (
|
||||
name TEXT,
|
||||
path TEXT,
|
||||
size INTEGER,
|
||||
moddate INTEGER,
|
||||
hash BLOB);",
|
||||
(),
|
||||
)
|
||||
.map_err(|e| format!("Failed to create files table: {}", e))?;
|
||||
|
||||
conn.execute(
|
||||
"CREATE VIRTUAL TABLE IF NOT EXISTS searchabletext USING fts5 (name, path, text, tokenize = 'trigram');",
|
||||
(),
|
||||
)
|
||||
.map_err(|e| format!("Failed to create searchabletext table: {}", e))?;
|
||||
|
||||
let conn_mutex = Arc::new(Mutex::new(conn));
|
||||
|
||||
// Count total files for progress tracking
|
||||
let total_file_count = WalkDir::new(path)
|
||||
.into_iter()
|
||||
.filter_map(|entry| entry.ok())
|
||||
.filter(|entry| !entry.metadata().map(|m| m.is_dir()).unwrap_or(true))
|
||||
.count();
|
||||
|
||||
// Update status with total file count
|
||||
if let Ok(mut status_guard) = status.lock() {
|
||||
if let IndexingStatus::Running { ref mut total_files, .. } = *status_guard {
|
||||
*total_files = Some(total_file_count);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// Process files with periodic status updates
|
||||
let walker = WalkDir::new(path).into_iter();
|
||||
|
||||
// Use a custom parallel iterator that checks for stop condition
|
||||
let entries: Vec<_> = walker.filter_map(|entry| entry.ok()).collect();
|
||||
|
||||
for (i, entry) in entries.iter().enumerate() {
|
||||
if *stop_flag.lock().unwrap() {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
// Update current file in status
|
||||
if let Ok(mut status_guard) = status.lock() {
|
||||
if let IndexingStatus::Running { ref mut current_file, ref mut files_processed, .. } = *status_guard {
|
||||
*current_file = Some(entry.path().to_string_lossy().to_string());
|
||||
*files_processed = i;
|
||||
}
|
||||
}
|
||||
|
||||
// Process the entry
|
||||
file_handling::process_entry(&conn_mutex, entry.clone());
|
||||
|
||||
// Throttle status updates and check for stop signal more frequently
|
||||
if i % 10 == 0 {
|
||||
thread::sleep(Duration::from_millis(1));
|
||||
if *stop_flag.lock().unwrap() {
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for IndexingService {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
58
src/main.rs
58
src/main.rs
|
|
@ -1,56 +1,28 @@
|
|||
use std::sync::{Mutex, Arc};
|
||||
|
||||
use walkdir::WalkDir;
|
||||
use tqdm;
|
||||
use rusqlite::Connection;
|
||||
use std::sync::Arc;
|
||||
use dioxus::prelude::*;
|
||||
use dpc_pariter::IteratorExt as _;
|
||||
|
||||
mod frontend;
|
||||
mod file_handling;
|
||||
mod document_extraction;
|
||||
mod indexing;
|
||||
|
||||
fn main() {
|
||||
// launch(frontend::App);
|
||||
let path: &str = "G:\\datasets\\preprocessed_ch_100";
|
||||
let db_path: &str = "GDrive.db";
|
||||
// Launch the frontend with the indexing service
|
||||
launch(app);
|
||||
}
|
||||
|
||||
let conn = Connection::open(db_path).unwrap();
|
||||
// let conn = Connection::open_in_memory().unwrap();
|
||||
// PRAGMA cache_size is in number of pages with 1024 byte page size by default
|
||||
conn.execute_batch(
|
||||
"PRAGMA journal_mode = OFF;
|
||||
PRAGMA synchronous = 0;
|
||||
PRAGMA cache_size = 10000;
|
||||
PRAGMA temp_store = MEMORY;",
|
||||
)
|
||||
// PRAGMA locking_mode = EXCLUSIVE;
|
||||
.expect("PRAGMA");
|
||||
conn.execute("CREATE TABLE IF NOT EXISTS files (
|
||||
name TEXT,
|
||||
path TEXT,
|
||||
size INTEGER,
|
||||
moddate INTEGER,
|
||||
hash BLOB);", ()).unwrap();
|
||||
// https://sqlite.org/fts5.html
|
||||
conn.execute("CREATE VIRTUAL TABLE IF NOT EXISTS searchabletext USING fts5 (name, path, text, tokenize = 'trigram');", ()).unwrap();
|
||||
|
||||
let conn_mutex = Arc::new(Mutex::new(conn));
|
||||
|
||||
tqdm::tqdm(WalkDir::new(path).into_iter()).parallel_map(move |entry| {
|
||||
match entry {
|
||||
Ok(e) => file_handling::process_entry(&conn_mutex, e),
|
||||
Err(error) => {
|
||||
println!("Error with directory walk: {:?}", error);
|
||||
return;
|
||||
},
|
||||
};
|
||||
}).for_each(drop);
|
||||
fn app() -> Element {
|
||||
let indexing_service = Arc::new(indexing::IndexingService::new());
|
||||
|
||||
rsx! {
|
||||
frontend::App {
|
||||
indexing_service: indexing_service
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/*
|
||||
SELECT name, hash, count(*) as cnt FROM files GROUP BY hash ORDER BY cnt DESC;
|
||||
|
||||
SELECT name, path, text, snippet(searchabletext, 2 , "<b>", "</b>", "...", 64) as "snip" FROM searchabletext WHERE text MATCH 'Terrasound' LIMIT 100;
|
||||
*/
|
||||
|
||||
}
|
||||
*/
|
||||
Loading…
Add table
Reference in a new issue