use bnto_core::context::ProcessContext;
use bnto_core::errors::BntoError;
use bnto_core::processor::{FileData, NodeInput, NodeOutput, NodeProcessor, OutputFile};
use bnto_core::progress::ProgressReporter;
const DEFAULT_MAX_ROWS: u64 = 100_000;
pub struct ReadSpreadsheet;
impl ReadSpreadsheet {
pub fn new() -> Self {
Self
}
}
impl Default for ReadSpreadsheet {
fn default() -> Self {
Self::new()
}
}
impl NodeProcessor for ReadSpreadsheet {
fn name(&self) -> &str {
"spreadsheet-read"
}
fn metadata(&self) -> bnto_core::NodeMetadata {
use bnto_core::metadata::*;
NodeMetadata {
node_type: "spreadsheet-read".to_string(),
name: "Read CSV Rows".to_string(),
description: "Explode a CSV into one item per row for loop iteration".to_string(),
category: NodeCategory::Spreadsheet,
accepts: vec!["text/csv".to_string()],
platforms: vec!["browser".to_string()],
parameters: build_read_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 = ReadConfig::from_params(&input.params);
let csv_text = parse_and_strip_bom(&data)?;
progress.report(10, "Reading CSV...");
let delimiter = resolve_delimiter(&config.delimiter);
let stem = file_stem(&input.filename);
let (headers, rows) = read_csv(csv_text, delimiter, config.has_headers)?;
if rows.len() as u64 > config.max_rows {
return Err(BntoError::InvalidInput(format!(
"CSV has {} rows, exceeds maxRows limit of {}. \
Increase the maxRows parameter to process larger files.",
rows.len(),
config.max_rows
)));
}
progress.report(50, &format!("Building {} row files...", rows.len()));
let files = build_row_files(&stem, &headers, &rows);
progress.report(90, "Building result...");
let metadata = build_read_metadata(headers.len(), rows.len());
progress.report(100, "Done!");
Ok(NodeOutput { files, metadata })
}
}
struct ReadConfig {
has_headers: bool,
delimiter: String,
max_rows: u64,
}
impl ReadConfig {
fn from_params(params: &serde_json::Map<String, serde_json::Value>) -> Self {
Self {
has_headers: params
.get("hasHeaders")
.and_then(|v| v.as_bool())
.unwrap_or(true),
delimiter: params
.get("delimiter")
.and_then(|v| v.as_str())
.unwrap_or("comma")
.to_string(),
max_rows: params
.get("maxRows")
.and_then(|v| v.as_u64())
.unwrap_or(DEFAULT_MAX_ROWS),
}
}
}
fn parse_and_strip_bom(data: &[u8]) -> Result<&str, BntoError> {
let data = if data.starts_with(&[0xEF, 0xBB, 0xBF]) {
&data[3..]
} else {
data
};
let 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 text.trim().is_empty() {
return Err(BntoError::InvalidInput(
"CSV file is empty — no data to read".to_string(),
));
}
Ok(text)
}
fn resolve_delimiter(delimiter: &str) -> u8 {
match delimiter {
"comma" | "," => b',',
"semicolon" | ";" => b';',
"tab" | "\t" => b'\t',
"pipe" | "|" => b'|',
other if other.len() == 1 => other.as_bytes()[0],
_ => b',',
}
}
fn read_csv(
csv_text: &str,
delimiter: u8,
has_headers: bool,
) -> Result<(Vec<String>, Vec<Vec<String>>), BntoError> {
let mut reader = csv::ReaderBuilder::new()
.has_headers(has_headers)
.delimiter(delimiter)
.flexible(true)
.from_reader(csv_text.as_bytes());
let headers = if has_headers {
let raw = reader
.headers()
.map_err(|e| BntoError::ProcessingFailed(format!("Failed to read CSV headers: {e}")))?
.clone();
raw.iter().map(|h| h.trim().to_string()).collect()
} else {
Vec::new() };
let mut rows = Vec::new();
let mut headers = headers;
for result in reader.records() {
let record = result
.map_err(|e| BntoError::ProcessingFailed(format!("Failed to read CSV row: {e}")))?;
if !has_headers && headers.is_empty() {
headers = (0..record.len()).map(|i| format!("col{i}")).collect();
}
let row: Vec<String> = (0..headers.len())
.map(|i| record.get(i).unwrap_or("").to_string())
.collect();
rows.push(row);
}
Ok((headers, rows))
}
fn build_row_files(stem: &str, headers: &[String], rows: &[Vec<String>]) -> Vec<OutputFile> {
rows.iter()
.enumerate()
.map(|(i, row)| {
let mut metadata = serde_json::Map::new();
for (col_idx, header) in headers.iter().enumerate() {
let value = row.get(col_idx).cloned().unwrap_or_default();
metadata.insert(header.clone(), serde_json::Value::String(value));
}
OutputFile {
data: FileData::Bytes(Vec::new()),
filename: format!("{stem}_row{i}.csv"),
mime_type: "text/csv".to_string(),
metadata,
}
})
.collect()
}
fn build_read_metadata(
column_count: usize,
row_count: usize,
) -> serde_json::Map<String, serde_json::Value> {
let mut metadata = serde_json::Map::new();
metadata.insert(
"columnCount".to_string(),
serde_json::Value::Number(column_count.into()),
);
metadata.insert(
"rowCount".to_string(),
serde_json::Value::Number(row_count.into()),
);
metadata
}
fn file_stem(filename: &str) -> String {
if let Some(dot_pos) = filename.rfind('.') {
filename[..dot_pos].to_string()
} else {
filename.to_string()
}
}
fn build_read_parameters() -> Vec<bnto_core::metadata::ParameterDef> {
use bnto_core::metadata::{Constraints, OptionEntry, ParameterDef, ParameterType};
vec![
ParameterDef {
name: "hasHeaders".to_string(),
label: "Has Headers".to_string(),
description: "First row contains column headers".to_string(),
param_type: ParameterType::Boolean,
default: Some(serde_json::json!(true)),
..Default::default()
},
ParameterDef {
name: "delimiter".to_string(),
label: "Delimiter".to_string(),
description: "Column separator character".to_string(),
param_type: ParameterType::Enum {
options: vec![
OptionEntry {
value: "comma".to_string(),
label: "Comma".to_string(),
},
OptionEntry {
value: "semicolon".to_string(),
label: "Semicolon".to_string(),
},
OptionEntry {
value: "tab".to_string(),
label: "Tab".to_string(),
},
OptionEntry {
value: "pipe".to_string(),
label: "Pipe".to_string(),
},
],
},
default: Some(serde_json::json!("comma")),
..Default::default()
},
ParameterDef {
name: "maxRows".to_string(),
label: "Max Rows".to_string(),
description: "Maximum rows to process. Error if CSV exceeds this limit.".to_string(),
param_type: ParameterType::Number,
default: Some(serde_json::json!(100000)),
constraints: Some(Constraints {
min: Some(1.0),
max: Some(10_000_000.0),
required: false,
}),
..Default::default()
},
]
}
#[cfg(test)]
mod tests {
use super::*;
use bnto_core::NoopContext;
fn make_input(csv_text: &str) -> NodeInput {
NodeInput {
data: FileData::Bytes(csv_text.as_bytes().to_vec()),
filename: "data.csv".to_string(),
mime_type: Some("text/csv".to_string()),
params: serde_json::Map::new(),
}
}
fn make_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: "data.csv".to_string(),
mime_type: Some("text/csv".to_string()),
params,
}
}
fn make_input_bytes(data: Vec<u8>) -> NodeInput {
NodeInput {
data: FileData::Bytes(data),
filename: "data.csv".to_string(),
mime_type: Some("text/csv".to_string()),
params: serde_json::Map::new(),
}
}
fn process_ok(input: NodeInput) -> NodeOutput {
let processor = ReadSpreadsheet::new();
let progress = ProgressReporter::new_noop();
processor
.process(input, &progress, &NoopContext)
.expect("should succeed")
}
fn process_err(input: NodeInput) -> BntoError {
let processor = ReadSpreadsheet::new();
let progress = ProgressReporter::new_noop();
match processor.process(input, &progress, &NoopContext) {
Err(e) => e,
Ok(_) => panic!("expected an error"),
}
}
#[test]
fn test_name() {
assert_eq!(ReadSpreadsheet::new().name(), "spreadsheet-read");
}
#[test]
fn test_metadata_node_type() {
let meta = ReadSpreadsheet::new().metadata();
assert_eq!(meta.node_type, "spreadsheet-read");
assert_eq!(
meta.category,
bnto_core::metadata::NodeCategory::Spreadsheet
);
}
#[test]
fn test_metadata_has_expected_params() {
let meta = ReadSpreadsheet::new().metadata();
let names: Vec<&str> = meta.parameters.iter().map(|p| p.name.as_str()).collect();
assert!(names.contains(&"hasHeaders"));
assert!(names.contains(&"delimiter"));
assert!(names.contains(&"maxRows"));
}
#[test]
fn test_metadata_accepts_csv() {
let meta = ReadSpreadsheet::new().metadata();
assert_eq!(meta.accepts, vec!["text/csv"]);
}
#[test]
fn test_metadata_browser_platform() {
let meta = ReadSpreadsheet::new().metadata();
assert_eq!(meta.platforms, vec!["browser"]);
}
#[test]
fn test_simple_csv_produces_per_row_files() {
let csv = "group,url\nAlpha,https://a.com\nBeta,https://b.com\nGamma,https://c.com\n";
let output = process_ok(make_input(csv));
assert_eq!(output.files.len(), 3, "3 rows should produce 3 files");
}
#[test]
fn test_row_metadata_has_column_values() {
let csv = "group,url,title\nAlpha,https://a.com,First\n";
let output = process_ok(make_input(csv));
let meta = &output.files[0].metadata;
assert_eq!(meta.get("group").unwrap(), "Alpha");
assert_eq!(meta.get("url").unwrap(), "https://a.com");
assert_eq!(meta.get("title").unwrap(), "First");
}
#[test]
fn test_output_data_is_empty() {
let csv = "name\nAlice\nBob\n";
let output = process_ok(make_input(csv));
for file in &output.files {
assert!(
file.data.is_empty().unwrap(),
"row files should have empty data"
);
}
}
#[test]
fn test_output_filename_includes_row_index() {
let csv = "name\nAlice\nBob\n";
let output = process_ok(make_input(csv));
assert_eq!(output.files[0].filename, "data_row0.csv");
assert_eq!(output.files[1].filename, "data_row1.csv");
}
#[test]
fn test_output_mime_type() {
let csv = "name\nAlice\n";
let output = process_ok(make_input(csv));
assert_eq!(output.files[0].mime_type, "text/csv");
}
#[test]
fn test_headers_only_produces_no_files() {
let csv = "name,age,city\n";
let output = process_ok(make_input(csv));
assert_eq!(
output.files.len(),
0,
"headers-only CSV should produce 0 files"
);
}
#[test]
fn test_single_row() {
let csv = "x\n42\n";
let output = process_ok(make_input(csv));
assert_eq!(output.files.len(), 1);
assert_eq!(output.files[0].metadata.get("x").unwrap(), "42");
}
#[test]
fn test_empty_cells_produce_empty_strings() {
let csv = "name,age,city\nAlice,,NYC\n";
let output = process_ok(make_input(csv));
assert_eq!(output.files[0].metadata.get("age").unwrap(), "");
}
#[test]
fn test_short_row_gets_empty_strings() {
let csv = "name,age,city\nAlice\n";
let output = process_ok(make_input(csv));
assert_eq!(output.files[0].metadata.get("name").unwrap(), "Alice");
assert_eq!(output.files[0].metadata.get("age").unwrap(), "");
assert_eq!(output.files[0].metadata.get("city").unwrap(), "");
}
#[test]
fn test_semicolon_delimiter() {
let csv = "name;age\nAlice;30\n";
let mut params = serde_json::Map::new();
params.insert("delimiter".into(), serde_json::json!("semicolon"));
let output = process_ok(make_input_with_params(csv, params));
assert_eq!(output.files[0].metadata.get("name").unwrap(), "Alice");
assert_eq!(output.files[0].metadata.get("age").unwrap(), "30");
}
#[test]
fn test_tab_delimiter() {
let csv = "name\tage\nAlice\t30\n";
let mut params = serde_json::Map::new();
params.insert("delimiter".into(), serde_json::json!("tab"));
let output = process_ok(make_input_with_params(csv, params));
assert_eq!(output.files[0].metadata.get("name").unwrap(), "Alice");
}
#[test]
fn test_bom_stripped() {
let csv_with_bom = b"\xEF\xBB\xBFname,age\nAlice,30\n";
let output = process_ok(make_input_bytes(csv_with_bom.to_vec()));
assert_eq!(output.files[0].metadata.get("name").unwrap(), "Alice");
assert!(output.files[0].metadata.get("\u{FEFF}name").is_none());
}
#[test]
fn test_no_headers_uses_synthetic_names() {
let csv = "Alice,30,NYC\nBob,25,LA\n";
let mut params = serde_json::Map::new();
params.insert("hasHeaders".into(), serde_json::json!(false));
let output = process_ok(make_input_with_params(csv, params));
assert_eq!(output.files.len(), 2);
assert_eq!(output.files[0].metadata.get("col0").unwrap(), "Alice");
assert_eq!(output.files[0].metadata.get("col1").unwrap(), "30");
assert_eq!(output.files[0].metadata.get("col2").unwrap(), "NYC");
}
#[test]
fn test_max_rows_rejects_large_csv() {
let mut csv = String::from("id\n");
for i in 0..10 {
csv.push_str(&format!("{i}\n"));
}
let mut params = serde_json::Map::new();
params.insert("maxRows".into(), serde_json::json!(5));
let err = process_err(make_input_with_params(&csv, params));
let msg = err.to_string();
assert!(
msg.contains("10 rows"),
"should mention actual count: {msg}"
);
assert!(msg.contains("5"), "should mention limit: {msg}");
}
#[test]
fn test_max_rows_at_limit_passes() {
let csv = "id\n1\n2\n3\n";
let mut params = serde_json::Map::new();
params.insert("maxRows".into(), serde_json::json!(3));
let output = process_ok(make_input_with_params(csv, params));
assert_eq!(output.files.len(), 3);
}
#[test]
fn test_default_max_rows() {
let params = serde_json::Map::new();
let config = ReadConfig::from_params(¶ms);
assert_eq!(config.max_rows, 100_000);
}
#[test]
fn test_empty_input() {
let err = process_err(make_input(""));
assert!(err.to_string().contains("empty"));
}
#[test]
fn test_whitespace_only_input() {
let err = process_err(make_input(" \n\n "));
assert!(err.to_string().contains("empty"));
}
#[test]
fn test_non_utf8_input() {
let err = process_err(make_input_bytes(vec![0xFF, 0xFE, 0x00, 0x41]));
assert!(err.to_string().contains("UTF-8"));
}
#[test]
fn test_metadata_counts() {
let csv = "name,age,city\nAlice,30,NYC\nBob,25,LA\n";
let output = process_ok(make_input(csv));
assert_eq!(output.metadata["columnCount"], 3);
assert_eq!(output.metadata["rowCount"], 2);
}
#[test]
fn test_fixture_manifest() {
let data = include_bytes!("../../../../test-fixtures/csv/manifest.csv");
let output = process_ok(make_input_bytes(data.to_vec()));
assert_eq!(output.files.len(), 3);
assert_eq!(output.files[0].metadata.get("group").unwrap(), "Alpha");
assert_eq!(
output.files[0].metadata.get("url").unwrap(),
"https://example.com/video1.mp4"
);
assert_eq!(output.files[2].metadata.get("group").unwrap(), "Beta");
}
#[test]
fn test_fixture_simple() {
let data = include_bytes!("../../../../test-fixtures/csv/simple.csv");
let output = process_ok(make_input_bytes(data.to_vec()));
assert_eq!(output.files.len(), 5);
assert_eq!(output.files[0].metadata.get("name").unwrap(), "Alice");
}
#[test]
fn test_fixture_headers_only() {
let data = include_bytes!("../../../../test-fixtures/csv/headers-only.csv");
let output = process_ok(make_input_bytes(data.to_vec()));
assert_eq!(output.files.len(), 0);
}
#[test]
fn test_large_csv_1000_rows() {
let mut csv = String::from("id,name,value\n");
for i in 0..1000 {
csv.push_str(&format!("{i},item_{i},{}\n", i * 10));
}
let output = process_ok(make_input(&csv));
assert_eq!(output.files.len(), 1000);
assert_eq!(output.metadata["rowCount"], 1000);
assert_eq!(output.metadata["columnCount"], 3);
assert_eq!(output.files[0].metadata.get("id").unwrap(), "0");
assert_eq!(output.files[999].metadata.get("id").unwrap(), "999");
}
}