use anyhow::{Result, anyhow, bail}; use common::proto::komp_ac::table_definition::{ ProfileTreeResponse, profile_tree_response::Profile, }; use common::proto::komp_ac::table_structure::{ GetTableImportDescriptorResponse, TableStructureResponse, }; use common::proto::komp_ac::tables_data::TableDataImportRow; use prost_types::{Value, value::Kind}; use sanitise_file_name::{Options, sanitise_with_options}; use std::collections::{HashMap, HashSet}; use std::io::BufRead; use std::path::Path; use crate::csv::CsvRecordReader; use crate::grpc::GrpcClient; use crate::value::convert_input_value; pub const CSV_IMPORT_PREVIEW_ROW_LIMIT: usize = 10; pub const CSV_EXPORT_PREVIEW_ROW_LIMIT: u64 = 20; pub const MAX_EXPORT_POSITION: u64 = i32::MAX as u64; #[derive(Debug, Clone)] pub struct ImportTable { pub name: String, pub columns: Vec, pub type_map: HashMap, pub link_columns: HashSet, pub revision: i64, } impl ImportTable { pub fn from_descriptor(descriptor: GetTableImportDescriptorResponse) -> Self { let writable = descriptor .fields .into_iter() .filter(|field| field.writable) .collect::>(); Self { name: descriptor.table_name, columns: writable.iter().map(|field| field.name.clone()).collect(), type_map: writable .iter() .map(|field| (field.name.clone(), field.storage_type.clone())) .collect(), link_columns: writable .iter() .filter(|field| field.link.is_some()) .map(|field| field.name.clone()) .collect(), revision: descriptor.table_revision, } } } #[derive(Debug, Clone, PartialEq, Eq)] pub struct ImportHeader { pub table_headers: Vec, pub columns: Vec, } #[derive(Debug, Clone)] pub struct ResolvedImportTarget { pub profile_name: String, pub tables: Vec, pub header: ImportHeader, } #[derive(Debug, Clone)] pub struct ExportTable { pub name: String, pub columns: Vec, pub count: u64, } pub fn export_table_header_row(tables: &[ExportTable]) -> Vec { tables .iter() .flat_map(|table| std::iter::repeat_n(table.name.clone(), table.columns.len())) .collect() } pub fn export_column_header_row(tables: &[ExportTable]) -> Vec { tables .iter() .flat_map(|table| table.columns.clone()) .collect() } pub fn maximum_export_row_count(tables: &[ExportTable]) -> u64 { tables.iter().map(|table| table.count).max().unwrap_or(0) } pub fn validate_output_filename(filename: &str) -> Result<()> { if filename.is_empty() || filename.trim() != filename || filename.ends_with('.') { bail!("Output filename must not be empty or start/end with whitespace or a dot"); } if sanitise_generated_filename(filename) != filename { bail!("Output filename '{filename}' is not portable across Windows, macOS, and Linux"); } if Path::new(filename).components().count() != 1 { bail!("Output filename must not contain a directory path"); } Ok(()) } pub fn sanitise_generated_filename(filename: &str) -> String { sanitise_with_options(filename, &portable_filename_options()) } fn portable_filename_options() -> Options> { Options { normalise_whitespace: false, trim_spaces_and_full_stops: false, trim_more_punctuation: false, six_measures_of_barley: "output", ..Options::DEFAULT } } pub fn parse_target_tables(value: &str) -> Vec { value .split(',') .map(str::trim) .filter(|table| !table.is_empty()) .map(ToString::to_string) .collect() } pub fn profile_header_row(profile_name: &str) -> Vec { vec![profile_name.to_string()] } pub fn table_header_row(tables: &[ImportTable]) -> Vec { tables .iter() .flat_map(|table| std::iter::repeat_n(table.name.clone(), table.columns.len())) .collect() } pub fn column_header_row(tables: &[ImportTable]) -> Vec { tables .iter() .flat_map(|table| table.columns.clone()) .collect() } pub fn parse_profile_header(row: &[String]) -> Option { if row.len() == 1 && !row[0].trim().is_empty() { Some(row[0].trim().to_string()) } else { None } } pub fn profile_has_tables(profile: &Profile, table_names: &[String]) -> bool { table_names .iter() .all(|table_name| profile.tables.iter().any(|table| table.name == *table_name)) } pub fn consecutive_table_names(values: &[String]) -> Vec { let mut names = Vec::new(); for value in values { if value.is_empty() { return Vec::new(); } if names.last() != Some(value) { names.push(value.clone()); } } names } pub async fn load_import_tables_for_profile( profile_name: &str, client: &mut GrpcClient, table_names: &[String], ) -> Result> { let mut tables = Vec::new(); for table_name in table_names { let descriptor = client .get_table_import_descriptor(profile_name.to_string(), table_name.clone()) .await?; tables.push(ImportTable::from_descriptor(descriptor)); } Ok(tables) } pub async fn resolve_import_target_from_reader( profile_tree: &ProfileTreeResponse, client: &mut GrpcClient, selected_profile: &str, selected_tables: &[String], reader: &mut CsvRecordReader, ) -> Result { let mut profile_name = selected_profile.trim().to_string(); let mut first = reader .next_record()? .ok_or_else(|| anyhow!("CSV file is empty"))?; if let Some(header_profile) = parse_profile_header(&first) { if !profile_name.is_empty() && profile_name != header_profile { bail!( "CSV profile '{}' does not match selected profile '{}'", header_profile, profile_name ); } profile_name = header_profile; first = reader .next_record()? .ok_or_else(|| anyhow!("CSV is missing a table or column header"))?; } let profiles = candidate_profiles(profile_tree, &profile_name)?; let mut second = None; let mut candidates = Vec::new(); if selected_tables.is_empty() { for profile in profiles { if let Some((tables, header)) = infer_multi_table_header( client, profile, &first, &mut second, reader, ) .await? { candidates.push((profile.name.clone(), tables, header)); } for table in &profile.tables { let tables = load_import_tables_for_profile( &profile.name, client, &[table.name.clone()], ) .await?; if first == column_header_row(&tables) { candidates.push(( profile.name.clone(), tables, ImportHeader { table_headers: Vec::new(), columns: first.clone(), }, )); } } } } else { for profile in profiles { if !profile_has_tables(profile, selected_tables) { continue; } let tables = load_import_tables_for_profile( &profile.name, client, selected_tables, ) .await?; if let Some(header) = match_selected_header( &first, &mut second, &tables, reader, )? { candidates.push((profile.name.clone(), tables, header)); } } } let (profile_name, tables, header) = match candidates.len() { 0 => bail!("CSV header does not match any available profile and table"), 1 => candidates.remove(0), _ => { let labels = candidates .iter() .map(|(profile_name, tables, _)| { format!( "{} / {}", profile_name, tables .iter() .map(|table| table.name.as_str()) .collect::>() .join(", ") ) }) .collect::>(); bail!( "CSV header is ambiguous. Select a profile and table. Matches: {}", labels.join("; ") ) } }; Ok(ResolvedImportTarget { profile_name, tables, header, }) } fn candidate_profiles<'a>( profile_tree: &'a ProfileTreeResponse, selected_profile: &str, ) -> Result> { if selected_profile.is_empty() { return Ok(profile_tree.profiles.iter().collect()); } let profiles = profile_tree .profiles .iter() .filter(|profile| profile.name == selected_profile) .collect::>(); if profiles.is_empty() { bail!("Selected profile '{}' is not available", selected_profile); } Ok(profiles) } fn read_second_header<'a, R: BufRead>( second: &'a mut Option>, reader: &mut CsvRecordReader, ) -> Result<&'a Vec> { if second.is_none() { *second = Some( reader .next_record()? .ok_or_else(|| anyhow!("CSV is missing the column header row"))?, ); } Ok(second.as_ref().expect("second header was inserted")) } fn match_selected_header( first: &[String], second: &mut Option>, tables: &[ImportTable], reader: &mut CsvRecordReader, ) -> Result> { if tables.len() == 1 { if first == table_header_row(tables) { let second = read_second_header(second, reader)?; return Ok( (second == &column_header_row(tables)).then(|| ImportHeader { table_headers: first.to_vec(), columns: second.clone(), }), ); } return Ok((first == column_header_row(tables)).then(|| ImportHeader { table_headers: Vec::new(), columns: first.to_vec(), })); } if first != table_header_row(tables) { return Ok(None); } let second = read_second_header(second, reader)?; Ok( (second == &column_header_row(tables)).then(|| ImportHeader { table_headers: first.to_vec(), columns: second.clone(), }), ) } async fn infer_multi_table_header( client: &mut GrpcClient, profile: &Profile, first: &[String], second: &mut Option>, reader: &mut CsvRecordReader, ) -> Result, ImportHeader)>> { let table_names = consecutive_table_names(first); if table_names.is_empty() || !profile_has_tables(profile, &table_names) { return Ok(None); } let tables = load_import_tables_for_profile(&profile.name, client, &table_names).await?; if first != table_header_row(&tables) { return Ok(None); } let second = read_second_header(second, reader)?; if second != &column_header_row(&tables) { return Ok(None); } Ok(Some(( tables, ImportHeader { table_headers: first.to_vec(), columns: second.clone(), }, ))) } pub fn row_to_table_data( table: &ImportTable, header: &ImportHeader, row: &[String], ) -> Result { let mut data = HashMap::new(); for (index, target_column) in header.columns.iter().enumerate() { if header .table_headers .get(index) .filter(|name| !name.is_empty()) .is_some_and(|name| name != &table.name) { continue; } if !table.columns.contains(target_column) { continue; } let Some(data_type) = table.type_map.get(target_column) else { continue; }; let raw = row.get(index).map(String::as_str).unwrap_or(""); let value = if table.link_columns.contains(target_column) && !raw.is_empty() { Value { kind: Some(Kind::StringValue(raw.to_string())), } } else { convert_input_value(raw, data_type, target_column).map_err(anyhow::Error::msg)? }; data.insert(target_column.clone(), value); } Ok(TableDataImportRow { data, link_display_columns: Vec::new() }) } pub fn exportable_columns(schema: &TableStructureResponse) -> Vec { schema .columns .iter() .filter(|column| { !column.is_primary_key && column.name != "id" && column.name != "deleted" && column.name != "created_at" && column.name != "row_revision" }) .map(|column| column.name.clone()) .collect() }