use bnto_core::context::ProcessContext;
use bnto_core::errors::BntoError;
use bnto_core::processor::{FileData, NodeInput, NodeOutput, NodeProcessor, OutputFile};
use bnto_core::progress::ProgressReporter;
pub struct CleanSpreadsheet;
impl CleanSpreadsheet {
pub fn new() -> Self {
Self
}
}
impl Default for CleanSpreadsheet {
fn default() -> Self {
Self::new()
}
}
impl NodeProcessor for CleanSpreadsheet {
fn name(&self) -> &str {
"spreadsheet-clean"
}
fn metadata(&self) -> bnto_core::NodeMetadata {
use bnto_core::metadata::*;
NodeMetadata {
node_type: "spreadsheet-clean".to_string(),
name: "Clean CSV".to_string(),
description: "Remove empty rows, trim whitespace, and deduplicate CSV data".to_string(),
category: NodeCategory::Spreadsheet,
accepts: vec!["text/csv".to_string()],
platforms: vec!["browser".to_string()],
parameters: build_clean_parameters(),
input_cardinality: InputCardinality::PerFile,
requires: vec![],
}
}
fn process(
&self,
input: NodeInput,
progress: &ProgressReporter,
_ctx: &dyn ProcessContext,
) -> Result<NodeOutput, BntoError> {
progress.report(0, "Parsing CSV...");
let data = input
.data
.into_bytes()
.map_err(|e| BntoError::ProcessingFailed(format!("Failed to read input: {e}")))?;
let config = CleanConfig::from_params(&input.params);
let csv_text = parse_csv_input(&data)?;
progress.report(10, "Reading CSV records...");
let (cleaned_headers, num_columns) = read_and_clean_headers(csv_text, config.trim)?;
progress.report(20, "Cleaning records...");
let (mut cleaned_rows, original_row_count) =
collect_cleaned_rows(csv_text, config.trim, config.remove_empty, num_columns)?;
progress.report(60, "Removing duplicates...");
let duplicates_removed = deduplicate_rows(&mut cleaned_rows, config.deduplicate);
progress.report(80, "Writing cleaned CSV...");
let output_bytes = write_csv_output(&cleaned_headers, &cleaned_rows)?;
progress.report(90, "Building result...");
let metadata =
build_clean_metadata(original_row_count, cleaned_rows.len(), duplicates_removed);
progress.report(100, "Done!");
Ok(build_clean_output(output_bytes, &input.filename, metadata))
}
}
struct CleanConfig {
trim: bool,
remove_empty: bool,
deduplicate: bool,
}
impl CleanConfig {
fn from_params(params: &serde_json::Map<String, serde_json::Value>) -> Self {
Self {
trim: bool_param(params, "trimWhitespace", true),
remove_empty: bool_param(params, "removeEmptyRows", true),
deduplicate: bool_param(params, "removeDuplicates", true),
}
}
}
fn bool_param(
params: &serde_json::Map<String, serde_json::Value>,
key: &str,
default: bool,
) -> bool {
params.get(key).and_then(|v| v.as_bool()).unwrap_or(default)
}
fn build_clean_output(
data: Vec<u8>,
input_filename: &str,
metadata: serde_json::Map<String, serde_json::Value>,
) -> NodeOutput {
NodeOutput {
files: vec![OutputFile {
data: FileData::Bytes(data),
filename: generate_output_filename(input_filename),
mime_type: "text/csv".to_string(),
metadata: serde_json::Map::new(),
}],
metadata,
}
}
fn clean_bool_param(
name: &str,
label: &str,
description: &str,
) -> bnto_core::metadata::ParameterDef {
use bnto_core::metadata::*;
ParameterDef {
name: name.to_string(),
label: label.to_string(),
description: description.to_string(),
param_type: ParameterType::Boolean,
default: Some(serde_json::json!(true)),
..Default::default()
}
}
fn build_clean_parameters() -> Vec<bnto_core::metadata::ParameterDef> {
vec![
clean_bool_param(
"trimWhitespace",
"Trim Whitespace",
"Remove leading and trailing whitespace from every cell",
),
clean_bool_param(
"removeEmptyRows",
"Remove Empty Rows",
"Skip rows where every cell is blank",
),
clean_bool_param(
"removeDuplicates",
"Remove Duplicates",
"Remove duplicate rows, keeping the first occurrence",
),
]
}
fn parse_csv_input(data: &[u8]) -> Result<&str, BntoError> {
let csv_text = std::str::from_utf8(data).map_err(|e| {
BntoError::InvalidInput(format!(
"File is not valid UTF-8 text (is this really a CSV?): {e}"
))
})?;
if csv_text.trim().is_empty() {
return Err(BntoError::InvalidInput(
"CSV file is empty — no data to clean".to_string(),
));
}
Ok(csv_text)
}
fn read_and_clean_headers(csv_text: &str, trim: bool) -> Result<(Vec<String>, usize), BntoError> {
let mut reader = csv::ReaderBuilder::new()
.has_headers(true)
.flexible(true)
.from_reader(csv_text.as_bytes());
let headers = reader
.headers()
.map_err(|e| BntoError::ProcessingFailed(format!("Failed to read CSV headers: {e}")))?
.clone();
let cleaned: Vec<String> = if trim {
headers.iter().map(|h| h.trim().to_string()).collect()
} else {
headers.iter().map(|h| h.to_string()).collect()
};
let num_columns = cleaned.len();
Ok((cleaned, num_columns))
}
fn collect_cleaned_rows(
csv_text: &str,
trim: bool,
remove_empty: bool,
num_columns: usize,
) -> Result<(Vec<Vec<String>>, usize), BntoError> {
let mut reader = csv::ReaderBuilder::new()
.has_headers(true)
.flexible(true)
.from_reader(csv_text.as_bytes());
let mut cleaned_rows: Vec<Vec<String>> = Vec::new();
let mut original_row_count: usize = 0;
for result in reader.records() {
original_row_count += 1;
let record = match result {
Ok(rec) => rec,
Err(_) => continue, };
let row = normalize_row(&record, trim, num_columns);
if remove_empty && row.iter().all(|cell| cell.is_empty()) {
continue;
}
cleaned_rows.push(row);
}
Ok((cleaned_rows, original_row_count))
}
fn normalize_row(record: &csv::StringRecord, trim: bool, num_columns: usize) -> Vec<String> {
let mut row: Vec<String> = record
.iter()
.map(|cell| {
if trim {
cell.trim().to_string()
} else {
cell.to_string()
}
})
.collect();
while row.len() < num_columns {
row.push(String::new());
}
row.truncate(num_columns);
row
}
fn deduplicate_rows(rows: &mut Vec<Vec<String>>, enabled: bool) -> usize {
if !enabled {
return 0;
}
let mut seen = std::collections::HashSet::new();
let before = rows.len();
rows.retain(|row| seen.insert(row.join("\0")));
before - rows.len()
}
fn write_csv_output(headers: &[String], rows: &[Vec<String>]) -> Result<Vec<u8>, BntoError> {
let mut writer = csv::WriterBuilder::new().from_writer(Vec::new());
writer
.write_record(headers)
.map_err(|e| BntoError::ProcessingFailed(format!("Failed to write CSV headers: {e}")))?;
for row in rows {
writer
.write_record(row)
.map_err(|e| BntoError::ProcessingFailed(format!("Failed to write CSV row: {e}")))?;
}
writer
.into_inner()
.map_err(|e| BntoError::ProcessingFailed(format!("Failed to finalize CSV output: {e}")))
}
fn build_clean_metadata(
original_row_count: usize,
cleaned_row_count: usize,
duplicates_removed: usize,
) -> serde_json::Map<String, serde_json::Value> {
let rows_removed = original_row_count - cleaned_row_count;
let mut metadata = serde_json::Map::new();
metadata.insert(
"originalRows".to_string(),
serde_json::Value::Number(original_row_count.into()),
);
metadata.insert(
"cleanedRows".to_string(),
serde_json::Value::Number(cleaned_row_count.into()),
);
metadata.insert(
"rowsRemoved".to_string(),
serde_json::Value::Number(rows_removed.into()),
);
metadata.insert(
"duplicatesRemoved".to_string(),
serde_json::Value::Number(duplicates_removed.into()),
);
metadata
}
fn generate_output_filename(original: &str) -> String {
if let Some(dot_pos) = original.rfind('.') {
let stem = &original[..dot_pos];
let ext = &original[dot_pos..];
format!("{stem}-cleaned{ext}")
} else {
format!("{original}-cleaned")
}
}
#[cfg(test)]
mod tests {
use super::*;
use bnto_core::NoopContext;
fn make_csv_input(csv_text: &str) -> NodeInput {
NodeInput {
data: FileData::Bytes(csv_text.as_bytes().to_vec()),
filename: "test.csv".to_string(),
mime_type: Some("text/csv".to_string()),
params: serde_json::Map::new(),
}
}
fn make_csv_input_with_params(
csv_text: &str,
params: serde_json::Map<String, serde_json::Value>,
) -> NodeInput {
NodeInput {
data: FileData::Bytes(csv_text.as_bytes().to_vec()),
filename: "test.csv".to_string(),
mime_type: Some("text/csv".to_string()),
params,
}
}
fn output_csv_text(output: &NodeOutput) -> String {
let bytes = output.files[0]
.data
.clone()
.into_bytes()
.expect("Should read bytes");
String::from_utf8(bytes).expect("Output should be valid UTF-8")
}
fn count_data_rows(csv_text: &str) -> usize {
let mut reader = csv::ReaderBuilder::new()
.has_headers(true)
.flexible(true)
.from_reader(csv_text.as_bytes());
reader.records().count()
}
#[test]
fn test_name_returns_clean_csv() {
let processor = CleanSpreadsheet::new();
assert_eq!(processor.name(), "spreadsheet-clean");
}
#[test]
fn test_default_creates_same_as_new() {
#[allow(clippy::default_constructed_unit_structs)]
let _processor = CleanSpreadsheet::default();
}
#[test]
fn test_basic_csv_passthrough() {
let csv = "name,age,city\nAlice,30,NYC\nBob,25,LA\n";
let processor = CleanSpreadsheet::new();
let progress = ProgressReporter::new_noop();
let input = make_csv_input(csv);
let output = processor.process(input, &progress, &NoopContext).unwrap();
assert_eq!(output.files.len(), 1);
assert_eq!(output.files[0].mime_type, "text/csv");
let text = output_csv_text(&output);
let rows = count_data_rows(&text);
assert_eq!(rows, 2, "Should preserve both data rows");
assert!(text.contains("Alice"), "Should contain Alice");
assert!(text.contains("Bob"), "Should contain Bob");
}
#[test]
fn test_remove_empty_rows() {
let csv = "name,age\nAlice,30\n,,\nBob,25\n,,\n";
let processor = CleanSpreadsheet::new();
let progress = ProgressReporter::new_noop();
let input = make_csv_input(csv);
let output = processor.process(input, &progress, &NoopContext).unwrap();
let text = output_csv_text(&output);
let rows = count_data_rows(&text);
assert_eq!(rows, 2, "Should remove empty rows, keeping 2 data rows");
assert!(text.contains("Alice"));
assert!(text.contains("Bob"));
let removed = output
.metadata
.get("rowsRemoved")
.unwrap()
.as_u64()
.unwrap();
assert_eq!(removed, 2, "Should report 2 rows removed");
}
#[test]
fn test_trim_whitespace_from_cells() {
let csv = "name,age\n Alice , 30 \n Bob ,25\n";
let processor = CleanSpreadsheet::new();
let progress = ProgressReporter::new_noop();
let input = make_csv_input(csv);
let output = processor.process(input, &progress, &NoopContext).unwrap();
let text = output_csv_text(&output);
assert!(text.contains("Alice"), "Should contain trimmed 'Alice'");
assert!(text.contains("Bob"), "Should contain trimmed 'Bob'");
assert!(
!text.contains(" Alice"),
"Should NOT contain ' Alice' with leading spaces"
);
}
#[test]
fn test_remove_duplicate_rows() {
let csv = "name,age\nAlice,30\nBob,25\nAlice,30\n";
let processor = CleanSpreadsheet::new();
let progress = ProgressReporter::new_noop();
let input = make_csv_input(csv);
let output = processor.process(input, &progress, &NoopContext).unwrap();
let text = output_csv_text(&output);
let rows = count_data_rows(&text);
assert_eq!(rows, 2, "Should remove 1 duplicate, keeping 2 rows");
let dupes = output
.metadata
.get("duplicatesRemoved")
.unwrap()
.as_u64()
.unwrap();
assert_eq!(dupes, 1, "Should report 1 duplicate removed");
}
#[test]
fn test_preserves_header_row() {
let csv = " name , age \nAlice,30\n";
let processor = CleanSpreadsheet::new();
let progress = ProgressReporter::new_noop();
let input = make_csv_input(csv);
let output = processor.process(input, &progress, &NoopContext).unwrap();
let text = output_csv_text(&output);
assert!(
text.starts_with("name,age"),
"Header should be trimmed: got '{}'",
text.lines().next().unwrap_or("")
);
}
#[test]
fn test_variable_length_rows_padded() {
let csv = "name,age,city\nAlice,30\nBob,25,LA\n";
let processor = CleanSpreadsheet::new();
let progress = ProgressReporter::new_noop();
let input = make_csv_input(csv);
let output = processor.process(input, &progress, &NoopContext).unwrap();
let text = output_csv_text(&output);
let rows = count_data_rows(&text);
assert_eq!(rows, 2, "Should keep both rows");
assert!(text.contains("Alice"), "Should contain Alice's row");
}
#[test]
fn test_headers_only_csv() {
let csv = "name,age,city\n";
let processor = CleanSpreadsheet::new();
let progress = ProgressReporter::new_noop();
let input = make_csv_input(csv);
let output = processor.process(input, &progress, &NoopContext).unwrap();
let text = output_csv_text(&output);
let rows = count_data_rows(&text);
assert_eq!(rows, 0, "Should have zero data rows");
assert!(text.contains("name"), "Should still contain header");
let original = output
.metadata
.get("originalRows")
.unwrap()
.as_u64()
.unwrap();
assert_eq!(original, 0, "Original row count should be 0");
}
#[test]
fn test_empty_input_returns_error() {
let csv = "";
let processor = CleanSpreadsheet::new();
let progress = ProgressReporter::new_noop();
let input = make_csv_input(csv);
let result = processor.process(input, &progress, &NoopContext);
assert!(result.is_err(), "Empty input should return an error");
if let Err(err) = result {
assert!(
err.to_string().contains("empty"),
"Error should mention 'empty': got '{}'",
err
);
}
}
#[test]
fn test_non_utf8_input_returns_error() {
let bad_bytes = vec![0xFF, 0xFE, 0x00, 0x41]; let processor = CleanSpreadsheet::new();
let progress = ProgressReporter::new_noop();
let input = NodeInput {
data: FileData::Bytes(bad_bytes),
filename: "bad.csv".to_string(),
mime_type: None,
params: serde_json::Map::new(),
};
let result = processor.process(input, &progress, &NoopContext);
assert!(result.is_err(), "Non-UTF8 input should return an error");
if let Err(err) = result {
assert!(
err.to_string().contains("UTF-8"),
"Error should mention UTF-8: got '{}'",
err
);
}
}
#[test]
fn test_combined_trim_remove_empty_deduplicate() {
let csv = "name,age\n Alice , 30 \n,,\nBob,25\n Alice , 30 \n,,\n";
let processor = CleanSpreadsheet::new();
let progress = ProgressReporter::new_noop();
let input = make_csv_input(csv);
let output = processor.process(input, &progress, &NoopContext).unwrap();
let text = output_csv_text(&output);
let rows = count_data_rows(&text);
assert_eq!(
rows, 2,
"Should have 2 rows (Alice + Bob), got text:\n{}",
text
);
let removed = output
.metadata
.get("rowsRemoved")
.unwrap()
.as_u64()
.unwrap();
assert_eq!(
removed, 3,
"Should remove 3 rows total (2 empty + 1 duplicate)"
);
let dupes = output
.metadata
.get("duplicatesRemoved")
.unwrap()
.as_u64()
.unwrap();
assert_eq!(dupes, 1, "Should report 1 duplicate removed");
}
#[test]
fn test_remove_duplicates_false_preserves_duplicates() {
let csv = "name,age\nAlice,30\nBob,25\nAlice,30\n";
let processor = CleanSpreadsheet::new();
let progress = ProgressReporter::new_noop();
let mut params = serde_json::Map::new();
params.insert(
"removeDuplicates".to_string(),
serde_json::Value::Bool(false),
);
let input = make_csv_input_with_params(csv, params);
let output = processor.process(input, &progress, &NoopContext).unwrap();
let text = output_csv_text(&output);
let rows = count_data_rows(&text);
assert_eq!(
rows, 3,
"Should keep all 3 rows when removeDuplicates is false"
);
let dupes = output
.metadata
.get("duplicatesRemoved")
.unwrap()
.as_u64()
.unwrap();
assert_eq!(dupes, 0);
}
#[test]
fn test_trim_whitespace_false_preserves_whitespace() {
let csv = "name,age\n Alice , 30 \nBob,25\n";
let processor = CleanSpreadsheet::new();
let progress = ProgressReporter::new_noop();
let mut params = serde_json::Map::new();
params.insert("trimWhitespace".to_string(), serde_json::Value::Bool(false));
let input = make_csv_input_with_params(csv, params);
let output = processor.process(input, &progress, &NoopContext).unwrap();
let text = output_csv_text(&output);
assert!(
text.contains("Alice") && text.contains("Bob"),
"Should contain both names"
);
assert!(
text.contains(" Alice "),
"Should preserve whitespace around Alice: got:\n{}",
text
);
}
#[test]
fn test_remove_empty_rows_false_preserves_empty_rows() {
let csv = "name,age\nAlice,30\n,,\nBob,25\n";
let processor = CleanSpreadsheet::new();
let progress = ProgressReporter::new_noop();
let mut params = serde_json::Map::new();
params.insert(
"removeEmptyRows".to_string(),
serde_json::Value::Bool(false),
);
let input = make_csv_input_with_params(csv, params);
let output = processor.process(input, &progress, &NoopContext).unwrap();
let text = output_csv_text(&output);
let rows = count_data_rows(&text);
assert_eq!(
rows, 3,
"Should keep all 3 rows when removeEmptyRows is false"
);
}
#[test]
fn test_large_csv_1000_rows() {
let mut csv = String::from("id,name,value\n");
for i in 0..1200 {
csv.push_str(&format!("{i},item_{i},{}\n", i * 10));
}
let processor = CleanSpreadsheet::new();
let progress = ProgressReporter::new_noop();
let input = make_csv_input(&csv);
let output = processor.process(input, &progress, &NoopContext).unwrap();
let text = output_csv_text(&output);
let rows = count_data_rows(&text);
assert_eq!(rows, 1200, "Should process all 1200 rows");
let removed = output
.metadata
.get("rowsRemoved")
.unwrap()
.as_u64()
.unwrap();
assert_eq!(removed, 0, "No rows should be removed from clean data");
}
#[test]
fn test_output_filename_with_extension() {
let result = generate_output_filename("data.csv");
assert_eq!(result, "data-cleaned.csv");
}
#[test]
fn test_output_filename_without_extension() {
let result = generate_output_filename("data");
assert_eq!(result, "data-cleaned");
}
#[test]
fn test_output_filename_with_multiple_dots() {
let result = generate_output_filename("my.data.csv");
assert_eq!(result, "my.data-cleaned.csv");
}
#[test]
fn test_output_filename_in_result() {
let csv = "name\nAlice\n";
let processor = CleanSpreadsheet::new();
let progress = ProgressReporter::new_noop();
let input = NodeInput {
data: FileData::Bytes(csv.as_bytes().to_vec()),
filename: "employees.csv".to_string(),
mime_type: None,
params: serde_json::Map::new(),
};
let output = processor.process(input, &progress, &NoopContext).unwrap();
assert_eq!(output.files[0].filename, "employees-cleaned.csv");
}
#[test]
fn test_metadata_contains_all_fields() {
let csv = "name,age\nAlice,30\nBob,25\n";
let processor = CleanSpreadsheet::new();
let progress = ProgressReporter::new_noop();
let input = make_csv_input(csv);
let output = processor.process(input, &progress, &NoopContext).unwrap();
assert!(output.metadata.contains_key("originalRows"));
assert!(output.metadata.contains_key("cleanedRows"));
assert!(output.metadata.contains_key("rowsRemoved"));
assert!(output.metadata.contains_key("duplicatesRemoved"));
}
}