spreadsheet-to-json 0.2.3

use calamine::{open_workbook_auto, Data, Reader, Sheets};
use csv::{ReaderBuilder, StringRecord};
use heck::ToSnakeCase;
use indexmap::IndexMap;
use serde_json::{Number, Value};
use std::fs::File;
use std::io::BufReader;
use std::path::Path;
use std::str::FromStr;

use crate::data_set::*;
use crate::error::GenericError;
use alphanumeric::*;
use crate::headers::*;
use crate::helpers::float_value;
use crate::helpers::string_value;
use is_truthy::*;
use crate::round_decimal::RoundDecimal;
use crate::Extension;
use crate::Format;
use crate::OptionSet;
use crate::PathData;
use crate::RowOptionSet;
use fuzzy_datetime::{iso_fuzzy_to_date_string, iso_fuzzy_to_datetime_string};

/// Callback invoked once per row when saving asynchronously (e.g. --deferred mode)
pub type SaveRowFn = Box<dyn Fn(IndexMap<String, Value>) -> Result<(), GenericError> + Send + Sync>;

/// Output the result set with captured rows (up to the maximum allowed) directly.
/// This is now synchronous and calls the asynchronous function using a runtime.
pub fn process_spreadsheet_direct(opts: &OptionSet) -> Result<ResultSet, GenericError> {
    let rt = tokio::runtime::Runtime::new().unwrap();
    rt.block_on(process_spreadsheet_core(opts, None, None))
}

/// Output the result set with captured rows (up to the maximum allowed) immediately.
/// Use this in an async function using the tokio runtime if you direct results
/// without a save callback
pub async fn process_spreadsheet_immediate(opts: &OptionSet) -> Result<ResultSet, GenericError> {
    process_spreadsheet_core(opts, None, None).await
}

#[deprecated(
    since = "1.0.6",
    note = "This function is a wrapper for the renamed function `process_spreadsheet_inline`"
)]
pub async fn render_spreadsheet_direct(opts: &OptionSet) -> Result<ResultSet, GenericError> {
    process_spreadsheet_core(opts, None, None).await
}

/// Output the result set with deferred row saving and optional output reference
pub async fn process_spreadsheet_async(
    opts: &OptionSet,
    save_func: SaveRowFn,
    out_ref: Option<&str>,
) -> Result<ResultSet, GenericError> {
    process_spreadsheet_core(opts, Some(save_func), out_ref).await
}

/// Output the result set with captured rows (up to the maximum allowed) directly.
/// with optional asynchronous row save method and output reference
pub async fn process_spreadsheet_core(
    opts: &OptionSet,
    save_opt: Option<SaveRowFn>,
    out_ref: Option<&str>,
) -> Result<ResultSet, GenericError> {
    if let Some(filepath) = opts.path.clone() {
        let path = Path::new(&filepath);
        if !path.exists() {
            #[allow(dead_code)]
            return Err(GenericError("file_unavailable"));
        }
        let path_data = PathData::new(path);
        if path_data.is_valid() {
            if path_data.use_calamine() {
                read_workbook_core(&path_data, opts, save_opt, out_ref).await
            } else {
                read_csv_core(&path_data, opts, save_opt, out_ref).await
            }
        } else {
            Err(GenericError("unsupported_format"))
        }
    } else {
        Err(GenericError("no_filepath_specified"))
    }
}

#[deprecated(
    since = "1.0.6",
    note = "This function is a wrapper for the renamed function `process_spreadsheet_core`"
)]
pub async fn render_spreadsheet_core(
    opts: &OptionSet,
    save_opt: Option<SaveRowFn>,
    out_ref: Option<&str>,
) -> Result<ResultSet, GenericError> {
    process_spreadsheet_core(opts, save_opt, out_ref).await
}

/// Parse spreadsheets with an optional callback method to save rows asynchronously and an optional output reference
/// that may be a file name or database identifier
pub async fn read_workbook_core<'a>(
    path_data: &PathData<'a>,
    opts: &OptionSet,
    save_opt: Option<SaveRowFn>,
    out_ref: Option<&str>,
) -> Result<ResultSet, GenericError> {
    if let Ok(mut workbook) = open_workbook_auto(path_data.path()) {
        let max_rows = opts.max_rows();
        let (selected_names, sheet_names, _sheet_indices) =
            match_sheet_name_and_index(&mut workbook, opts);

        if !selected_names.is_empty() {
            let info = WorkbookInfo::new(path_data, &selected_names, &sheet_names);

            if opts.multimode() {
                read_multiple_worksheets(&mut workbook, &sheet_names, opts, &info, max_rows).await
            } else {
                let sheet_ref = &selected_names[0];
                read_single_worksheet(workbook, sheet_ref, opts, &info, save_opt, out_ref).await
            }
        } else {
            Err(GenericError("workbook_with_no_sheets"))
        }
    } else {
        Err(GenericError("cannot_open_workbook"))
    }
}

/// Read multiple worksheets from a workbook in preview mode.
async fn read_multiple_worksheets(
    workbook: &mut Sheets<BufReader<File>>,
    sheet_names: &[String],
    opts: &OptionSet,
    info: &WorkbookInfo,
    max_rows: usize,
) -> Result<ResultSet, GenericError> {
    let mut sheets: Vec<SheetDataSet> = vec![];
    let capture_rows = opts.capture_rows();
    for (sheet_index, sheet_ref) in sheet_names.iter().enumerate() {
        let range = workbook.worksheet_range(&sheet_ref.clone())?;
        let mut headers: Vec<String> = vec![];
        let mut has_headers = false;
        let capture_headers = !opts.omit_header;
        let mut rows: Vec<IndexMap<String, Value>> =
            Vec::with_capacity(if capture_rows { max_rows } else { 0 });
        let mut row_index = 0;
        let header_row_index = opts.header_row_index();
        let mut col_keys: Vec<String> = vec![];
        let columns = if sheet_index == 0 {
            opts.rows.columns.clone()
        } else {
            vec![]
        };
        let mut resolved_row_opts = opts.rows.clone();
        let match_header_row_below = capture_headers && header_row_index > 0;
        if let Some(first_row) = range.headers() {
            let natural_keys = natural_column_keys(&first_row, &opts.field_mode);
            let resolved_columns = resolve_columns(&columns, &natural_keys);
            headers = build_header_keys(&first_row, &resolved_columns, &opts.field_mode);
            resolved_row_opts.columns = resolved_columns;
            has_headers = !match_header_row_below;
            col_keys = first_row;
        }
        let total = range.get_size().0;
        if capture_rows || match_header_row_below {
            let max_row_count = if capture_rows {
                max_rows
            } else {
                header_row_index + 2
            };
            let max_take = if total < max_row_count {
                total
            } else {
                max_row_count + 1
            };
            for row in range.rows().take(max_take) {
                if row_index > max_row_count {
                    break;
                }
                if match_header_row_below && (row_index + 1) == header_row_index {
                    let h_row = row
                        .iter()
                        .map(|c| c.to_string().to_snake_case())
                        .collect::<Vec<String>>();
                    let natural_keys = natural_column_keys(&h_row, &opts.field_mode);
                    let resolved_columns = resolve_columns(&columns, &natural_keys);
                    headers = build_header_keys(&h_row, &resolved_columns, &opts.field_mode);
                    resolved_row_opts.columns = resolved_columns;
                    has_headers = true;
                } else if (has_headers || !capture_headers) && capture_rows {
                    let raw_values: Vec<String> = row.iter().map(|c| c.to_string()).collect();
                    if is_not_header_row(&raw_values, row_index, &col_keys) {
                        let row_map = workbook_row_to_map(row, &resolved_row_opts, &headers);
                        rows.push(row_map);
                    }
                }
                row_index += 1;
            }
        }
        sheets.push(SheetDataSet::new(sheet_ref, &headers, &rows, total));
    }
    Ok(ResultSet::from_multiple(&sheets, info, opts))
}

/// Read a single worksheet from a workbook in immediate (sync) or asycnhronous modes
pub async fn read_single_worksheet(
    mut workbook: Sheets<BufReader<File>>,
    sheet_ref: &str,
    opts: &OptionSet,
    info: &WorkbookInfo,
    save_opt: Option<SaveRowFn>,
    out_ref: Option<&str>,
) -> Result<ResultSet, GenericError> {
    let range = workbook.worksheet_range(sheet_ref)?;
    let capture_rows = opts.capture_rows();
    let columns = opts.rows.columns.clone();
    let max_rows = opts.max_rows();
    let mut headers: Vec<String> = vec![];
    let mut col_keys: Vec<String> = vec![];
    let mut has_headers = false;
    let capture_headers = !opts.omit_header;
    let mut rows: Vec<IndexMap<String, Value>> =
        Vec::with_capacity(if capture_rows { max_rows } else { 0 });
    let mut row_index = 0;
    let header_row_index = opts.header_row_index();
    let match_header_row_below = capture_headers && header_row_index > 0;
    let mut resolved_row_opts = opts.rows.clone();

    if let Some(first_row) = range.headers() {
        let natural_keys = natural_column_keys(&first_row, &opts.field_mode);
        let resolved_columns = resolve_columns(&columns, &natural_keys);
        headers = build_header_keys(&first_row, &resolved_columns, &opts.field_mode);
        resolved_row_opts.columns = resolved_columns;
        has_headers = !match_header_row_below;
        col_keys = first_row;
    }
    let total = range.get_size().0;
    if capture_rows || match_header_row_below {
        let max_row_count = if capture_rows {
            max_rows
        } else {
            header_row_index + 2
        };
        let max_take = if total < max_row_count {
            total
        } else {
            max_row_count + 1
        };
        for row in range.rows().take(max_take) {
            if row_index > max_row_count {
                break;
            }
            if match_header_row_below && (row_index + 1) == header_row_index {
                let h_row = row
                    .iter()
                    .map(|c| c.to_string().to_snake_case())
                    .collect::<Vec<String>>();
                let natural_keys = natural_column_keys(&h_row, &opts.field_mode);
                let resolved_columns = resolve_columns(&columns, &natural_keys);
                headers = build_header_keys(&h_row, &resolved_columns, &opts.field_mode);
                resolved_row_opts.columns = resolved_columns;
                has_headers = true;
            } else if (has_headers || !capture_headers) && capture_rows {
                // only capture rows if headers are either omitted or have already been captured
                let raw_values: Vec<String> = row.iter().map(|c| c.to_string()).collect();
                if is_not_header_row(&raw_values, row_index, &col_keys) {
                    let row_map = workbook_row_to_map(row, &resolved_row_opts, &headers);
                    rows.push(row_map);
                }
            }
            row_index += 1;
        }
    }
    if let Some(save_method) = save_opt {
        let mut row_iter = range.rows();
        let mut save_count: usize = 0;
        if let Some(first_row) = row_iter.next() {
            let raw_values: Vec<String> = first_row.iter().map(|c| c.to_string()).collect();
            if is_not_header_row(&raw_values, 0, &col_keys) {
                let first_row_map = workbook_row_to_map(first_row, &resolved_row_opts, &headers);
                save_method(first_row_map)?;
                save_count += 1;
            }
        }
        for row in row_iter {
            if save_count >= max_rows {
                break;
            }
            let row_map = workbook_row_to_map(row, &resolved_row_opts, &headers);
            save_method(row_map)?;
            save_count += 1;
        }
    }

    let ds = DataSet::from_count_and_rows(total, rows, opts);
    Ok(ResultSet::new(info, &headers, ds, opts, out_ref))
}

/// Process a CSV/TSV file asynchronously with an optional row save method
/// and output reference (file or database table reference)
pub async fn read_csv_core<'a>(
    path_data: &PathData<'a>,
    opts: &OptionSet,
    save_opt: Option<SaveRowFn>,
    out_ref: Option<&str>,
) -> Result<ResultSet, GenericError> {
    let separator = match path_data.mode() {
        Extension::Tsv => b't',
        _ => b',',
    };
    if let Ok(mut rdr) = ReaderBuilder::new()
        .delimiter(separator)
        .from_path(path_data.path())
    {
        let capture_header = !opts.omit_header;
        let capture_rows = opts.capture_rows();
        let max_line_usize = opts.max_rows();
        let mut rows: Vec<IndexMap<String, Value>> =
            Vec::with_capacity(if capture_rows { max_line_usize } else { 0 });
        let mut line_count = 0;

        let mut headers: Vec<String> = vec![];
        let mut resolved_row_opts = opts.rows.clone();
        if capture_header {
            if let Ok(hdrs) = rdr.headers() {
                headers = hdrs.into_iter().map(|s| s.to_owned()).collect();
            }
            let columns = opts.rows.columns.clone();
            let natural_keys = natural_column_keys(&headers, &opts.field_mode);
            let resolved_columns = resolve_columns(&columns, &natural_keys);
            headers = build_header_keys(&headers, &resolved_columns, &opts.field_mode);
            resolved_row_opts.columns = resolved_columns;
        }

        let mut total = 0;
        if capture_rows {
            for result in rdr.records() {
                if line_count >= max_line_usize {
                    break;
                }
                if let Some(row) = csv_row_result_to_values(result, &resolved_row_opts) {
                    rows.push(to_index_map(&row, &headers));
                    line_count += 1;
                }
            }
            total = line_count + rdr.records().count() + 1;
        } else if let Some(save_method) = save_opt {
            for result in rdr.records() {
                if let Some(row) = csv_row_result_to_values(result, &resolved_row_opts) {
                    let row_map = to_index_map(&row, &headers);
                    save_method(row_map)?;
                    total += 1;
                }
            }
        } else {
            if let Ok(mut count_rdr) = ReaderBuilder::new().from_path(path_data.path()) {
                total = count_rdr.records().count();
            }
        }
        let info = WorkbookInfo::simple(path_data);
        let ds = DataSet::from_count_and_rows(total, rows, opts);
        Ok(ResultSet::new(&info, &headers, ds, opts, out_ref))
    } else {
        let error_msg = match path_data.ext() {
            Extension::Tsv => "unreadable_tsv_file",
            _ => "unreadable_csv_file",
        };
        Err(GenericError(error_msg))
    }
}

// Convert an array of row data to an IndexMap of serde_json::Value objects
fn workbook_row_to_map(
    row: &[Data],
    opts: &RowOptionSet,
    headers: &[String],
) -> IndexMap<String, Value> {
    to_index_map(&workbook_row_to_values(row, opts), headers)
}

// Convert an array of row data to a vector of serde_json::Value objects
fn workbook_row_to_values(row: &[Data], opts: &RowOptionSet) -> Vec<Value> {
    row.iter()
        .enumerate()
        .map(|(c_index, cell)| workbook_cell_to_value(cell, opts, c_index))
        .collect()
}

/// Convert a spreadsheet data cell to a polymorphic serde_json::Value object
fn workbook_cell_to_value(cell: &Data, opts: &RowOptionSet, c_index: usize) -> Value {
    let col = opts.column(c_index);
    let format = col.map_or(Format::Auto, |c| c.format.to_owned());
    let def_val = col.and_then(|c| c.default.clone());

    let date_only = resolve_date_only(&format, opts.date_only);

    match cell {
        Data::Int(i) => Value::Number(Number::from_i128(*i as i128).unwrap()),
        Data::Float(f) => process_float_value(*f, format),
        Data::DateTimeIso(d) => process_iso_datetime_value(d, def_val, date_only),
        Data::DateTime(d) => process_excel_datetime_value(d, def_val, date_only),
        Data::Bool(b) => Value::Bool(*b),
        Data::String(s) => process_string_value(s, format, def_val),
        Data::Empty => def_val.unwrap_or(Value::Null),
        _ => Value::String(cell.to_string()),
    }
}

/// A column's own Format::Date/Format::DateTime override takes precedence over the
/// row-wide --date-only default; any other format (including Format::Auto) falls back
/// to that row-wide default, unchanged from before this override existed.
fn resolve_date_only(format: &Format, row_date_only: bool) -> bool {
    match format {
        Format::Date => true,
        Format::DateTime => false,
        _ => row_date_only,
    }
}

fn process_float_value(value: f64, format: Format) -> Value {
    match format {
        Format::Integer => Value::Number(Number::from_i128(value as i128).unwrap()),
        Format::Boolean => Value::Bool(value >= 1.0),
        Format::Text => Value::String(value.to_string()),
        _ => Value::Number(Number::from_f64(value).unwrap()),
    }
}

fn process_excel_datetime_value(
    datetime: &calamine::ExcelDateTime,
    def_val: Option<Value>,
    date_only: bool,
) -> Value {
    let dt_ref = datetime.as_datetime().map_or_else(
        || def_val.unwrap_or(Value::Null),
        |dt| {
            let formatted_date = if date_only {
                dt.format("%Y-%m-%d").to_string()
            } else {
                dt.format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string()
            };
            Value::String(formatted_date)
        },
    );
    dt_ref
}

fn process_iso_datetime_value(dt_str: &str, def_val: Option<Value>, date_only: bool) -> Value {
    if date_only {
        iso_fuzzy_to_date_string(dt_str)
            .map_or_else(|| def_val.unwrap_or(Value::Null), Value::String)
    } else {
        iso_fuzzy_to_datetime_string(dt_str)
            .map_or_else(|| def_val.unwrap_or(Value::Null), Value::String)
    }
}

fn process_string_value(value: &str, format: Format, def_val: Option<Value>) -> Value {
    match format {
        Format::Boolean => process_truthy_value(value, def_val, |v, ef| v.is_truthy_core(ef)),
        Format::Truthy => process_truthy_value(value, def_val, |v, ef| v.is_truthy_standard(ef)),
        Format::TruthyCustom(opts) => process_truthy_value(value, def_val, |v, _| {
            v.is_truthy_custom(&opts)
        }),
        Format::Decimal(places) => {
            process_numeric_value(value, def_val, |n| float_value(n.round_decimal(places)))
        }
        Format::Float => process_numeric_value(value, def_val, float_value),
        Format::Date => process_date_value(value, def_val, iso_fuzzy_to_date_string),
        Format::DateTime => process_date_value(value, def_val, iso_fuzzy_to_datetime_string),
        _ => Value::String(value.to_owned()),
    }
}

fn process_truthy_value<F>(value: &str, def_val: Option<Value>, truthy_fn: F) -> Value
where
    F: Fn(&str, bool) -> Option<bool>,
{
    if let Some(is_true) = truthy_fn(value, false) {
        Value::Bool(is_true)
    } else {
        def_val.unwrap_or(Value::Null)
    }
}

fn process_numeric_value<F>(value: &str, def_val: Option<Value>, numeric_fn: F) -> Value
where
    F: Fn(f64) -> Value,
{
    if let Some(n) = value.to_first_number::<f64>() {
        numeric_fn(n)
    } else {
        def_val.unwrap_or(Value::Null)
    }
}

fn process_date_value<F>(value: &str, def_val: Option<Value>, date_fn: F) -> Value
where
    F: Fn(&str) -> Option<String>,
{
    if let Some(date_str) = date_fn(value) {
        string_value(&date_str)
    } else {
        def_val.unwrap_or(Value::Null)
    }
}

// Convert csv rows to value
fn csv_row_result_to_values(
    result: Result<StringRecord, csv::Error>,
    opts: &RowOptionSet,
) -> Option<Vec<Value>> {
    if let Ok(record) = result {
        let row = record
            .into_iter()
            .enumerate()
            .map(|(ci, cell)| csv_cell_to_json_value(cell, opts, ci))
            .collect();
        return Some(row);
    }
    None
}

// convert CSV cell &str value to a polymorphic serde_json::VALUE
fn csv_cell_to_json_value(cell: &str, opts: &RowOptionSet, index: usize) -> Value {
    let has_number = cell.to_first_number::<f64>().is_some();
    // clean cell to check if it's numeric
    let col = opts.column(index);
    let (fmt, euro_num_mode) = if let Some(c) = col {
        (c.format.clone(), c.decimal_comma)
    } else {
        (Format::Auto, opts.decimal_comma)
    };
    let num_cell = if has_number {
        let euro_num_mode = uses_decimal_comma(cell, euro_num_mode);
        if euro_num_mode {
            cell.replace(",", ".").replace(",", ".")
        } else {
            cell.replace(",", "")
        }
    } else {
        cell.to_owned()
    };
    let mut new_cell = Value::Null;
    if !num_cell.is_empty() && num_cell.is_numeric() {
        if let Ok(float_val) = serde_json::Number::from_str(&num_cell) {
            match fmt {
                Format::Integer => {
                    // as_i128() only succeeds for Numbers that are already integer-valued
                    // internally, so it silently yields 0 for any decimal value (e.g. "58.2")
                    // via unwrap_or(0). Go through as_f64() and truncate instead, matching
                    // the equivalent xlsx/ods cell conversion in process_float_value above.
                    if let Some(f) = float_val.as_f64() {
                        if let Some(int_val) = Number::from_i128(f as i128) {
                            new_cell = Value::Number(int_val);
                        }
                    }
                }
                Format::Boolean => {
                    // only 1.0 or more will evaluate as true
                    new_cell = Value::Bool(float_val.as_f64().unwrap_or(0f64) >= 1.0);
                }
                _ => {
                    new_cell = Value::Number(float_val);
                }
            }
        }
    } else if let Some(is_true) = cell.is_truthy_core(false) {
        new_cell = Value::Bool(is_true);
    } else {
        new_cell = match fmt {
            Format::Truthy => {
                if let Some(is_true) = cell.is_truthy_standard(false) {
                    Value::Bool(is_true)
                } else {
                    Value::Null
                }
            }
            _ => Value::String(cell.to_string()),
        };
    }
    new_cell
}

pub async fn read_workbook_sheet_info<'a>(
    path_data: &PathData<'a>,
) -> Result<IndexMap<String, usize>, GenericError> {
    if let Ok(mut workbook) = open_workbook_auto(path_data.path()) {
        let mut im: IndexMap<String, usize> = IndexMap::new();
        for name in workbook.sheet_names() {
            if let Ok(range) = workbook.worksheet_range(&name) {
                im.insert(name, range.rows().count());
            }
        }
        Ok(im)
    } else {
        Err(GenericError("cannot_open_workbook"))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{helpers::*, Column};
    use serde_json::json;
    use std::path;

    #[test]
    fn test_direct_processing_xlsx() {
        let sample_path = "data/sample-data-1.xlsx";

        // instantiate the OptionSet with a sample path and a maximum row count of 1000 rows as the source file has 401 rows
        // (although )the default max is 10,000)
        let opts = OptionSet::new(sample_path).max_row_count(1_000);

        let result = process_spreadsheet_direct(&opts);

        // The source file should have 1 header row and 400 data rows
        assert_eq!(result.unwrap().num_rows, 401);
    }

    #[test]
    fn test_source_key_override_renames_and_reformats_without_position() {
        // End-to-end: overriding just one field out of many by its natural key,
        // without needing to enumerate/pad the columns ahead of it.
        let sample_path = "data/sample-data-1.csv";
        let mut opts = OptionSet::new(sample_path).max_row_count(2);
        opts.rows.columns = vec![
            Column::from_source_key_with_format("weight", Some("weight_lbs"), Format::Integer, None, false, false),
        ];

        let result = process_spreadsheet_direct(&opts).unwrap();
        // every other column keeps its natural, auto-detected name
        assert!(result.keys.contains(&"id".to_string()));
        assert!(result.keys.contains(&"first_name".to_string()));
        assert!(result.keys.contains(&"weight_lbs".to_string()));
        assert!(!result.keys.contains(&"weight".to_string()));

        let rows = result.to_vec();
        let first = rows.first().expect("at least one row");
        assert!(first.get("weight_lbs").is_some());
        assert!(first.get("weight").is_none());
        assert_eq!(first.get("id").unwrap(), 1);
    }

    #[test]
    fn test_resolve_date_only_prefers_column_format_over_row_default() {
        // A column's own Format::Date/Format::DateTime overrides the row-wide
        // --date-only default; Format::Auto (and anything else) falls back to it.
        assert!(resolve_date_only(&Format::Date, false));
        assert!(!resolve_date_only(&Format::DateTime, true));
        assert!(!resolve_date_only(&Format::Auto, false));
        assert!(resolve_date_only(&Format::Auto, true));
    }

    #[test]
    fn test_source_key_override_casts_native_datetime_cell_to_date_only() {
        // Regression test: workbook_cell_to_value computed the column's Format override
        // but only ever consulted the row-wide --date-only flag for Data::DateTime /
        // Data::DateTimeIso cells, so a per-column `Format::Date` override on a real
        // (non-string) datetime cell had no effect at all.
        let sample_path = "data/sample-data-1.xlsx";
        let mut opts = OptionSet::new(sample_path).max_row_count(1);
        opts.rows.columns = vec![
            Column::from_source_key_with_format("start_time", None, Format::Date, None, false, false),
        ];

        let result = process_spreadsheet_direct(&opts).unwrap();
        let rows = result.to_vec();
        let first = rows.first().expect("at least one row");
        let start_time = first.get("start_time").expect("start_time column").as_str().unwrap();
        assert_eq!(start_time, "2023-06-15");
        assert!(!start_time.contains('T'), "should be date-only, got: {}", start_time);
    }

    #[test]
    fn test_csv_cell_integer_format_truncates_decimal_values() {
        // Regression test: Number::as_i128() only succeeds for already-integer-valued
        // Numbers, so casting a decimal CSV cell like "58.2" to Format::Integer used to
        // silently produce 0 via unwrap_or(0) instead of the truncated value.
        let cols = vec![Column::new_format(Format::Integer, None)];
        let row_opts = RowOptionSet::simple(&cols);
        assert_eq!(csv_cell_to_json_value("58.2", &row_opts, 0), Value::Number(Number::from(58)));
        assert_eq!(csv_cell_to_json_value("82.5", &row_opts, 0), Value::Number(Number::from(82)));
        assert_eq!(csv_cell_to_json_value("100", &row_opts, 0), Value::Number(Number::from(100)));
    }

    #[test]
    fn test_csv_cell_does_not_coerce_ids_to_booleans() {
        // Regression test: these previously became `true`/`false` because their
        // embedded digit run (e.g. "SKU001" -> "001" -> 1) was fuzzily extracted
        // and matched against is_truthy_core's numeric range, even though the
        // column has no boolean intent (Format::Auto, the default).
        let row_opts = RowOptionSet::default();
        assert_eq!(csv_cell_to_json_value("SKU001", &row_opts, 0), Value::String("SKU001".to_string()));
        assert_eq!(csv_cell_to_json_value("A1", &row_opts, 0), Value::String("A1".to_string()));
        assert_eq!(csv_cell_to_json_value("01/06/2024", &row_opts, 0), Value::String("01/06/2024".to_string()));
        // literal boolean tokens should still be recognised
        assert_eq!(csv_cell_to_json_value("true", &row_opts, 0), Value::Bool(true));
        assert_eq!(csv_cell_to_json_value("false", &row_opts, 0), Value::Bool(false));
    }

    #[test]
    fn test_direct_processing_csv() {
        let sample_path = "data/sample-data-1.csv";

        // instantiate the OptionSet with a sample path and a maximum row count of 1000 rows as the source file has 401 rows
        // (although )the default max is 10,000)
        let opts = OptionSet::new(sample_path).max_row_count(1_000);

        let result = process_spreadsheet_direct(&opts);

        // The source file should have 1 header row and 400 data rows
        assert_eq!(result.unwrap().num_rows, 401);
    }

    #[test]
    fn test_multisheet_preview_ods() {
        let sample_path = "data/sample-data-2.ods";

        // instantiate the OptionSet with a sample path
        // a maximum row count returned of 10 rows
        // and read mode to *preview* to scan all sheets
        // It should correctly calculate
        let opts = OptionSet::new(sample_path)
            .max_row_count(10)
            .read_mode_preview();

        let result = process_spreadsheet_direct(&opts);

        // The source spreadsheet should have 2 sheets
        let dataset = result.unwrap();
        assert_eq!(dataset.sheets.len(), 2);
        // The source spreadsheet should have 101 + 17 (= 118) populated rows including headers
        assert_eq!(dataset.num_rows, 118);

        // The first sheet's data should only output 10 rows (including the header)
        assert_eq!(dataset.data.first_sheet().len(), 10);
    }

    #[test]
    fn test_column_override_1() {
        let sample_json = json!({
          "sku": "CHAIR16",
          "height": "112cm",
          "width": "69cm",
          "approved": "Y"
        });

        let rows = json_object_to_calamine_data(sample_json);

        let cols = vec![
            Column::new_format(Format::Text, Some(string_value(""))),
            Column::new_format(Format::Float, Some(float_value(95.0))),
            Column::new_format(Format::Float, Some(float_value(65.0))),
            Column::new_format(Format::Truthy, Some(bool_value(false))),
        ];

        // The first sheet's data should only output 10 rows (including the header)
        let opts = &RowOptionSet::simple(&cols);
        let result = workbook_row_to_values(&rows, opts);
        // the second column be cast to 112.0
        assert_eq!(result.get(1).unwrap(), 112.0);
        // the third column be cast to 69.0
        assert_eq!(result.get(2).unwrap(), 69.0);
        // the fourth column be cast to boolean
        assert_eq!(result.get(3).unwrap(), true);
    }

    #[test]
    fn test_column_override_2() {
        let sample_json = json!({
          "name": "Sophia",
          "dob": "2001-9-23",
          "weight": "62kg",
          "result": "GOOD"
        });

        let rows = json_object_to_calamine_data(sample_json);

        let cols = vec![
            Column::new_format(Format::Text, None),
            Column::new_format(Format::Date, None),
            Column::new_format(Format::Float, None),
            // the fourth column be cast to boolean
            Column::new_format(
                Format::truthy_custom("good", "bad"),
                Some(bool_value(false)),
            ),
        ];

        // The first sheet's data should only output 10 rows (including the header)
        let opts = &RowOptionSet::simple(&cols);
        let result = workbook_row_to_values(&rows, opts);
        assert_eq!(result.get(1).unwrap(), "2001-09-23");
        assert_eq!(result.get(2).unwrap(), 62.0);
        assert_eq!(result.get(3).unwrap(), true);
    }

    #[tokio::test]
    async fn test_read_workbook_info() {
        let sample_path = "data/sample-data-1.xlsx";
        let path_data = PathData::new(path::Path::new(sample_path));
        let info = read_workbook_sheet_info(&path_data).await;
        assert!(info.is_ok());
    }

    #[tokio::test]
    async fn test_large_csv_file() {
        let sample_path = "data/large-datasheet.csv";
        let max_rows = 100_000;
        let opts = OptionSet::new(sample_path).max_row_count(max_rows);
        let result = process_spreadsheet_core(&opts, None, None).await;
        if let Ok(data) = result.clone() {
            assert_eq!(data.data.first_sheet().len(), max_rows as usize);
        } else {
            panic!("Failed to process large CSV file");
        }
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_medium_excel_file() {
        let sample_path = "data/medium-spreadsheet-50_000.xlsx";
        let max_rows = 5_000;
        let opts = OptionSet::new(sample_path).max_row_count(max_rows);
        let result = process_spreadsheet_core(&opts, None, None).await;
        if let Ok(data) = result.clone() {
            assert_eq!(data.data.first_sheet().len(), max_rows as usize);
        } else {
            panic!("Failed to process large Excel file");
        }
        assert!(result.is_ok());
    }
}