pub mod canonical;
pub mod detect;
pub mod filter;
pub mod inspect;
pub mod ndjson;
pub mod query;
pub mod sort;
pub mod transform;
pub use canonical::{to_ods_workbook, CanonicalValue};
pub use detect::{detect_content, detect_extension};
pub use filter::Filter;
pub use inspect::inspect;
pub use ndjson::{stream_delimited_to_ndjson, stream_ndjson_to_delimited, stream_ndjson_to_ndjson};
pub use sort::{external_sort_delimited, SortSpec};
pub use transform::Transforms;
use std::io::Cursor;
fn delimited_to_json(input: &str, delim: u8) -> Result<String, Box<dyn std::error::Error>> {
let input = strip_bom(input);
let mut rdr = csv::ReaderBuilder::new()
.delimiter(delim)
.from_reader(Cursor::new(input));
let headers = rdr.headers()?.clone();
let mut seen = std::collections::HashSet::new();
for header in headers.iter() {
if !seen.insert(header) {
return Err(format!("duplicate header: {header}").into());
}
}
let mut records = Vec::new();
for result in rdr.records() {
let record = result?;
let mut map = serde_json::Map::new();
for (header, value) in headers.iter().zip(record.iter()) {
map.insert(
header.to_string(),
serde_json::Value::String(value.to_string()),
);
}
records.push(serde_json::Value::Object(map));
}
Ok(serde_json::to_string(&records)?)
}
fn json_to_delimited(input: &str, delim: u8) -> Result<String, Box<dyn std::error::Error>> {
let values: Vec<serde_json::Value> = serde_json::from_str(input)?;
if values.is_empty() {
return Ok(String::new());
}
let mut headers: Vec<String> = Vec::new();
for value in &values {
let obj = value.as_object().ok_or("JSON array must contain objects")?;
for key in obj.keys() {
if !headers.contains(key) {
headers.push(key.clone());
}
}
}
let mut wtr = csv::WriterBuilder::new()
.delimiter(delim)
.from_writer(vec![]);
wtr.write_record(&headers)?;
for value in values {
let obj = value.as_object().ok_or("JSON array must contain objects")?;
let mut record: Vec<String> = Vec::with_capacity(headers.len());
for h in &headers {
let cell = match obj.get(h) {
Some(serde_json::Value::String(s)) => s.clone(),
Some(serde_json::Value::Number(n)) => n.to_string(),
Some(serde_json::Value::Bool(b)) => b.to_string(),
Some(serde_json::Value::Null) => String::new(),
Some(serde_json::Value::Object(_)) | Some(serde_json::Value::Array(_)) => {
return Err(format!(
"nested value at key {h} cannot be represented in flat output"
)
.into());
}
None => String::new(),
};
record.push(cell);
}
wtr.write_record(&record)?;
}
let data = wtr.into_inner()?;
Ok(String::from_utf8(data)?)
}
pub fn csv_to_json(csv: &str) -> Result<String, Box<dyn std::error::Error>> {
delimited_to_json(csv, b',')
}
pub fn tsv_to_json(tsv: &str) -> Result<String, Box<dyn std::error::Error>> {
delimited_to_json(tsv, b'\t')
}
pub fn json_to_csv(json: &str) -> Result<String, Box<dyn std::error::Error>> {
json_to_delimited(json, b',')
}
pub fn json_to_tsv(json: &str) -> Result<String, Box<dyn std::error::Error>> {
json_to_delimited(json, b'\t')
}
fn strip_bom(input: &str) -> &str {
input.strip_prefix('\u{feff}').unwrap_or(input)
}
pub fn yaml_to_json(yaml: &str) -> Result<String, Box<dyn std::error::Error>> {
let opts = serde_saphyr::options! {
strict_booleans: true,
ignore_binary_tag_for_string: true,
};
let values: Vec<serde_json::Value> = serde_saphyr::from_str_with_options(yaml, opts)?;
for value in &values {
let obj = value
.as_object()
.ok_or("YAML sequence must contain mappings")?;
for (key, val) in obj {
if val.is_object() || val.is_array() {
return Err(format!(
"nested YAML value at key {key} cannot be represented in flat output"
)
.into());
}
}
}
Ok(serde_json::to_string(&values)?)
}
pub fn json_to_yaml(json: &str) -> Result<String, Box<dyn std::error::Error>> {
let values: Vec<serde_json::Value> = serde_json::from_str(json)?;
for value in &values {
let obj = value.as_object().ok_or("JSON array must contain objects")?;
for (key, val) in obj {
if val.is_object() || val.is_array() {
return Err(format!(
"nested JSON value at key {key} cannot be represented in flat YAML"
)
.into());
}
}
}
Ok(serde_saphyr::to_string(&values)?)
}
pub fn csv_to_yaml(csv: &str) -> Result<String, Box<dyn std::error::Error>> {
json_to_yaml(&csv_to_json(csv)?)
}
pub fn tsv_to_yaml(tsv: &str) -> Result<String, Box<dyn std::error::Error>> {
json_to_yaml(&tsv_to_json(tsv)?)
}
pub fn yaml_to_csv(yaml: &str) -> Result<String, Box<dyn std::error::Error>> {
json_to_csv(&yaml_to_json(yaml)?)
}
pub fn yaml_to_tsv(yaml: &str) -> Result<String, Box<dyn std::error::Error>> {
json_to_tsv(&yaml_to_json(yaml)?)
}
pub fn csv_to_tsv(csv: &str) -> Result<String, Box<dyn std::error::Error>> {
json_to_tsv(&csv_to_json(csv)?)
}
pub fn tsv_to_csv(tsv: &str) -> Result<String, Box<dyn std::error::Error>> {
json_to_csv(&tsv_to_json(tsv)?)
}
pub fn xml_to_json(xml: &str) -> Result<String, Box<dyn std::error::Error>> {
CanonicalValue::from_xml(xml)?.to_json()
}
pub fn json_to_xml(json: &str) -> Result<String, Box<dyn std::error::Error>> {
CanonicalValue::from_json(json)?.to_xml()
}
pub fn csv_to_xml(input: &str) -> Result<String, Box<dyn std::error::Error>> {
CanonicalValue::from_csv(input)?.to_xml()
}
pub fn tsv_to_xml(input: &str) -> Result<String, Box<dyn std::error::Error>> {
CanonicalValue::from_tsv(input)?.to_xml()
}
pub fn yaml_to_xml(input: &str) -> Result<String, Box<dyn std::error::Error>> {
CanonicalValue::from_yaml(input)?.to_xml()
}
pub fn xml_to_csv(xml: &str) -> Result<String, Box<dyn std::error::Error>> {
CanonicalValue::from_xml(xml)?.to_csv()
}
pub fn xml_to_tsv(xml: &str) -> Result<String, Box<dyn std::error::Error>> {
CanonicalValue::from_xml(xml)?.to_tsv()
}
pub fn xml_to_yaml(xml: &str) -> Result<String, Box<dyn std::error::Error>> {
CanonicalValue::from_xml(xml)?.to_yaml()
}
pub fn toml_to_json(toml: &str) -> Result<String, Box<dyn std::error::Error>> {
CanonicalValue::from_toml(toml)?.to_json()
}
pub fn toml_to_csv(toml: &str) -> Result<String, Box<dyn std::error::Error>> {
CanonicalValue::from_toml(toml)?.to_csv()
}
pub fn toml_to_tsv(toml: &str) -> Result<String, Box<dyn std::error::Error>> {
CanonicalValue::from_toml(toml)?.to_tsv()
}
pub fn toml_to_yaml(toml: &str) -> Result<String, Box<dyn std::error::Error>> {
CanonicalValue::from_toml(toml)?.to_yaml()
}
pub fn toml_to_xml(toml: &str) -> Result<String, Box<dyn std::error::Error>> {
CanonicalValue::from_toml(toml)?.to_xml()
}
pub fn json_to_toml(json: &str) -> Result<String, Box<dyn std::error::Error>> {
CanonicalValue::from_json(json)?.to_toml()
}
pub fn csv_to_toml(csv: &str) -> Result<String, Box<dyn std::error::Error>> {
CanonicalValue::from_csv(csv)?.to_toml()
}
pub fn tsv_to_toml(tsv: &str) -> Result<String, Box<dyn std::error::Error>> {
CanonicalValue::from_tsv(tsv)?.to_toml()
}
pub fn yaml_to_toml(yaml: &str) -> Result<String, Box<dyn std::error::Error>> {
CanonicalValue::from_yaml(yaml)?.to_toml()
}
pub fn xml_to_toml(xml: &str) -> Result<String, Box<dyn std::error::Error>> {
CanonicalValue::from_xml(xml)?.to_toml()
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum Format {
Csv,
Tsv,
Json,
Yaml,
Xml,
Toml,
Xlsx,
Sql,
Ndjson,
Ods,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[non_exhaustive]
pub enum SqlDialect {
#[default]
Generic,
PostgreSql,
Sqlite,
MySql,
}
impl std::str::FromStr for SqlDialect {
type Err = String;
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s.to_ascii_lowercase().as_str() {
"generic" | "ansi" => Ok(Self::Generic),
"postgres" | "postgresql" | "pg" => Ok(Self::PostgreSql),
"sqlite" => Ok(Self::Sqlite),
"mysql" | "mariadb" => Ok(Self::MySql),
other => Err(format!(
"unknown SQL dialect {other:?}; expected generic, postgresql, sqlite, or mysql"
)),
}
}
}
pub type OutputDoc = (String, Vec<u8>);
impl Format {
pub const ALL: [Format; 10] = [
Format::Csv,
Format::Tsv,
Format::Json,
Format::Yaml,
Format::Xml,
Format::Toml,
Format::Xlsx,
Format::Sql,
Format::Ndjson,
Format::Ods,
];
}
pub fn convert(
from: Format,
to: Format,
input: &str,
) -> Result<String, Box<dyn std::error::Error>> {
if from == to {
return Ok(input.to_string());
}
if from == Format::Xlsx || to == Format::Xlsx {
return Err("XLSX is binary; use convert_workbook instead".into());
}
if from == Format::Ods || to == Format::Ods {
return Err("ODS is binary; use convert_workbook instead".into());
}
let hub = match from {
Format::Csv => CanonicalValue::from_csv(input)?,
Format::Tsv => CanonicalValue::from_tsv(input)?,
Format::Json => CanonicalValue::from_json(input)?,
Format::Yaml => CanonicalValue::from_yaml(input)?,
Format::Xml => CanonicalValue::from_xml(input)?,
Format::Toml => CanonicalValue::from_toml(input)?,
Format::Sql => {
let mut sheets = CanonicalValue::from_sql_workbook(input, SqlDialect::Generic)?;
if sheets.len() != 1 {
return Err("multi-table SQL requires convert_workbook".into());
}
sheets.remove(0).1
}
Format::Ndjson => CanonicalValue::from_ndjson(input)?,
Format::Ods => unreachable!(),
Format::Xlsx => unreachable!(),
};
match to {
Format::Csv => hub.to_csv(),
Format::Tsv => hub.to_tsv(),
Format::Json => hub.to_json(),
Format::Yaml => hub.to_yaml(),
Format::Xml => hub.to_xml(),
Format::Toml => hub.to_toml(),
Format::Sql => hub.to_sql_with_table("data"),
Format::Ndjson => hub.to_ndjson(),
Format::Ods => unreachable!(),
Format::Xlsx => unreachable!(),
}
}
pub fn convert_workbook(
from: Format,
to: Format,
input: &[u8],
) -> Result<Vec<OutputDoc>, Box<dyn std::error::Error>> {
convert_workbook_with_options(from, to, input, SqlDialect::Generic, None)
}
pub fn convert_workbook_with_options(
from: Format,
to: Format,
input: &[u8],
dialect: SqlDialect,
table_name: Option<&str>,
) -> Result<Vec<OutputDoc>, Box<dyn std::error::Error>> {
if from == to && from != Format::Xlsx {
let text = std::str::from_utf8(input)?;
return Ok(vec![("Sheet1".to_string(), text.as_bytes().to_vec())]);
}
let sheets: Vec<(String, CanonicalValue)> = match from {
Format::Xlsx => CanonicalValue::from_xlsx_workbook(input)?,
Format::Ods => {
let v = CanonicalValue::from_ods(input)?;
match v {
CanonicalValue::Object(o) => o,
other => vec![("Sheet1".to_string(), other)],
}
}
Format::Sql => {
let text = std::str::from_utf8(input)?;
CanonicalValue::from_sql_workbook(text, dialect)?
}
Format::Ndjson => {
let text = std::str::from_utf8(input)?;
vec![("Sheet1".to_string(), CanonicalValue::from_ndjson(text)?)]
}
_ => {
let text = std::str::from_utf8(input)?;
let hub = match from {
Format::Csv => CanonicalValue::from_csv(text)?,
Format::Tsv => CanonicalValue::from_tsv(text)?,
Format::Json => CanonicalValue::from_json(text)?,
Format::Yaml => CanonicalValue::from_yaml(text)?,
Format::Xml => CanonicalValue::from_xml(text)?,
Format::Toml => CanonicalValue::from_toml(text)?,
Format::Sql | Format::Ndjson | Format::Xlsx | Format::Ods => unreachable!(),
};
vec![("Sheet1".to_string(), hub)]
}
};
if to == Format::Sql {
let text = if sheets.len() == 1 {
let table = table_name.unwrap_or("data");
sheets[0].1.to_sql_with_table(table)?
} else {
CanonicalValue::to_sql_workbook(&sheets)?
};
return Ok(vec![("Sheet1".to_string(), text.into_bytes())]);
}
if to == Format::Ods {
let bytes = to_ods_workbook(&sheets)?;
return Ok(vec![("Sheet1".to_string(), bytes)]);
}
if to == Format::Ndjson {
let text = if sheets.len() == 1 {
sheets[0].1.to_ndjson()?
} else {
let merged = CanonicalValue::Object(sheets);
merged.to_ndjson()?
};
return Ok(vec![("Sheet1".to_string(), text.into_bytes())]);
}
if to == Format::Xlsx {
let bytes = CanonicalValue::to_xlsx_workbook(&sheets)?;
return Ok(vec![("Sheet1".to_string(), bytes)]);
}
if sheets.len() == 1 {
let text = write_single(to, &sheets[0].1)?;
return Ok(vec![(sheets[0].0.clone(), text.into_bytes())]);
}
match to {
Format::Json | Format::Yaml | Format::Xml | Format::Toml => {
let merged = CanonicalValue::Object(sheets);
Ok(vec![(
"Sheet1".to_string(),
write_single(to, &merged)?.into_bytes(),
)])
}
Format::Csv | Format::Tsv => {
let mut out = Vec::with_capacity(sheets.len());
for (name, value) in &sheets {
out.push((name.clone(), write_single(to, value)?.into_bytes()));
}
Ok(out)
}
Format::Sql => unreachable!(),
Format::Ndjson => unreachable!(),
Format::Ods => unreachable!(),
Format::Xlsx => unreachable!(),
}
}
fn write_single(
format: Format,
value: &CanonicalValue,
) -> Result<String, Box<dyn std::error::Error>> {
match format {
Format::Csv => value.to_csv(),
Format::Tsv => value.to_tsv(),
Format::Json => value.to_json(),
Format::Yaml => value.to_yaml(),
Format::Xml => value.to_xml(),
Format::Toml => value.to_toml(),
Format::Sql => value.to_sql_with_table("data"),
Format::Ndjson => value.to_ndjson(),
Format::Ods => Err("ODS must go through to_ods_workbook".into()),
Format::Xlsx => Err("XLSX must go through to_xlsx_workbook".into()),
}
}
pub fn stream_delimited(
reader: impl std::io::Read,
from_delim: u8,
writer: impl std::io::Write,
to_delim: u8,
transforms: &Transforms,
) -> Result<(), Box<dyn std::error::Error>> {
let mut rdr = csv::ReaderBuilder::new()
.delimiter(from_delim)
.from_reader(reader);
let raw_headers: Vec<String> = rdr.headers()?.iter().map(|s| s.to_string()).collect();
let mut renamed_headers = raw_headers.clone();
for (from, to) in &transforms.rename {
for h in renamed_headers.iter_mut() {
if h == from {
*h = to.clone();
break;
}
}
}
let final_headers: Vec<String> = match &transforms.select {
Some(sel) => sel.clone(),
None => renamed_headers,
};
let mut wtr = csv::WriterBuilder::new()
.delimiter(to_delim)
.from_writer(writer);
wtr.write_record(&final_headers)?;
let limit = transforms.limit.unwrap_or(usize::MAX);
let mut emitted = 0usize;
let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
let dedup_cols = transforms.dedup.as_ref().cloned();
for result in rdr.records() {
if emitted >= limit {
break;
}
let record = result?;
let mut row: Vec<(String, CanonicalValue)> = Vec::with_capacity(raw_headers.len());
for (i, h) in raw_headers.iter().enumerate() {
let v = record.get(i).unwrap_or("").to_string();
row.push((h.clone(), CanonicalValue::String(v)));
}
if let Some(new_row) = transforms.apply_to_row(row) {
if let Some(cols) = &dedup_cols {
let key: String = if cols.is_empty() {
new_row
.iter()
.map(|(_, v)| match v {
CanonicalValue::String(s) => s.clone(),
_ => String::new(),
})
.collect::<Vec<_>>()
.join("\x1f")
} else {
cols.iter()
.map(|c| {
new_row
.iter()
.find(|(k, _)| k == c)
.map(|(_, v)| match v {
CanonicalValue::String(s) => s.clone(),
_ => String::new(),
})
.unwrap_or_default()
})
.collect::<Vec<_>>()
.join("\x1f")
};
if !seen.insert(key) {
continue;
}
}
let values: Vec<String> = new_row
.iter()
.map(|(_, v)| match v {
CanonicalValue::Null => String::new(),
CanonicalValue::String(s) => s.clone(),
CanonicalValue::Number(n) => n.to_string(),
CanonicalValue::Bool(b) => b.to_string(),
_ => String::new(),
})
.collect();
wtr.write_record(&values)?;
emitted += 1;
}
}
wtr.flush()?;
Ok(())
}