static USAGE: &str = r#"
Validates CSV data using two main modes:
JSON SCHEMA VALIDATION MODE:
===========================
This mode is invoked if a JSON Schema file (draft 2020-12) is provided.
The CSV data is validated against the JSON Schema. If the CSV data is valid, no output
files are created and the command returns an exit code of 0.
If invalid records are found, they are put into an "invalid" file, with the rest of the
records put into a "valid"" file.
A "validation-errors.tsv" report is also created with the following columns:
* row_number: the row number of the invalid record
* field: the field name of the invalid field
* error: a validation error message detailing why the field is invalid
It uses the JSON Schema Validation Specification (draft 2020-12) to validate the CSV.
It validates the structure of the file, as well as the data types and domain/range of the fields.
See https://json-schema.org/draft/2020-12/json-schema-validation.html
qsv supports a custom format - `currency`. This format will only accept a valid currency, defined as:
1. ISO Currency Symbol (optional): This is the ISO 4217 three-character code or currency symbol
(e.g. USD, EUR, JPY, $, €, ¥, etc.)
2. Amount: This is the numerical value of the currency. More than 2 decimal places are allowed.
3. Formats: Valid currency formats include:
Standard: $1,000.00 or USD1000.00
Negative amounts: ($100.00) or -$100.00
Different styles: 1.000,00 (used in some countries for euros)
qsv also supports two custom keywords - `dynamicEnum` and `uniqueCombinedWith`.
dynamicEnum
===========
`dynamicEnum` allows for dynamic validation against a reference CSV file.
It can be used to validate against a set of values unknown at the time of schema creation or
when the set of valid values is dynamic or too large to hardcode into the JSON Schema with `enum`.
The reference CSV file can be local or a URL (http/https, dathere & ckan schemes supported).
The "dynamicEnum" value has the form:
// qsvlite binary variant only supports URIs which can be files on the local filesystem
// or remote files (http and https schemes supported)
dynamicEnum = "URI|colname" where colname is the column name or column index (0-based)
// use data.csv from the current working directory; use the 1st column for validation
dynamicEnum = "data.csv"
// use data.csv in /lookup_dir directory; use the column "Agency" for validation
dynamicEnum = "/lookupdir/data.csv|Agency"
// get data.csv; use the 3rd column for validation (2 as the col index is 0-based)
dynamicEnum = "https://example.com/data.csv|2"
// on other qsv binary variants, dynamicEnum has expanded caching functionality
dynamicEnum = "[cache_name;cache_age]|URI|colname" where cache_name and cache_age are optional
// use data.csv from current working directory; cache it as data with a default
// cache age of 3600 seconds i.e. the cached data.csv expires after 1 hour
dynamicEnum = "data.csv"
// get data.csv; cache it as custom_name, cache age 600 seconds
dynamicEnum = "custom_name;600|https://example.com/data.csv"
// get data.csv; cache it as data, cache age 800 seconds
dynamicEnum = ";800|https://example.com/data.csv"
// get the top matching result for nyc_neighborhoods (signaled by trailing ?),
// cache it as nyc_neighborhood_data.csv (NOTE: cache name is required when using CKAN scheme)
// with a default cache age of 3600 seconds
// be sure to set --ckan-api, otherwise it will default to datHere's CKAN (data.dathere.com)
dynamicEnum = "nyc_neighborhood_data|ckan:://nyc_neighborhoods?"
// get CKAN resource with id 1234567, cache it as resname, 3600 secs cache age
// note that if the resource is a private resource, you'll need to set --ckan-token
dynamicEnum = "resname|ckan:://1234567"
// same as above but with a cache age of 100 seconds; use the borough column for validation
dynamicEnum = "resname;100|ckan:://1234567|borough
// get us_states.csv from datHere lookup tables
dynamicEnum = "dathere://us_states.csv"
If colname is not specified, the first column of the CSV file is read and used for validation.
uniqueCombinedWith
==================
`uniqueCombinedWith` allows you to validate that combinations of values across specified columns
are unique. It can be used with either column names or column indices (0-based). For example:
// Validate that combinations of name and email are unique
uniqueCombinedWith = ["name", "email"]
// Validate that combinations of columns at indices 1 and 2 are unique
uniqueCombinedWith = [1, 2]
// Validate that the combinations of named and indexed columns are unique
uniqueCombinedWith = ["name", 2]
When a duplicate combination is found, the validation will fail and the error message will indicate
which columns had duplicate combinations (named columns first, then indexed columns). The invalid
records will be written to the .invalid file, while valid records will be written to the .valid file.
`uniqueCombinedWith` complements the standard `uniqueItems` keyword, which can only validate
uniqueness across a single column.
-------------------------------------------------------
You can create a JSON Schema file from a reference CSV file using the `qsv schema` command.
Once the schema is created, you can fine-tune it to your needs and use it to validate other CSV
files that have the same structure.
Be sure to select a "training" CSV file that is representative of the data you want to validate
when creating a schema. The data types, domain/range and regular expressions inferred from the
reference CSV file should be appropriate for the data you want to validate.
Typically, after creating a schema, you should edit it to fine-tune each field's inferred
validation rules.
For example, if we created a JSON schema file called "reference.schema.json" using the `schema` command.
And want to validate "mydata.csv" which we know has validation errors, the output files from running
`qsv validate mydata.csv reference.schema.json` are:
* mydata.csv.valid
* mydata.csv.invalid
* mydata.csv.validation-errors.tsv
With an exit code of 1 to indicate a validation error.
If we validate another CSV file, "mydata2.csv", which we know is valid, there are no output files,
and the exit code is 0.
If piped from stdin, the filenames will use `stdin.csv` as the base filename. For example:
`cat mydata.csv | qsv validate reference.schema.json`
* stdin.csv.valid
* stdin.csv.invalid
* stdin.csv.validation-errors.tsv
JSON SCHEMA SCHEMA VALIDATION SUBMODE:
---------------------------------------
`validate` also has a `schema` subcommand to validate JSON Schema files themselves. E.g.
`qsv validate schema myjsonschema.json`
// ignore format validation
`qsv validate schema --no-format-validation myjsonschema.json`
RFC 4180 VALIDATION MODE:
========================
If run without a JSON Schema file, the CSV is validated for RFC 4180 CSV standard compliance
(see https://github.com/dathere/qsv#rfc-4180-csv-standard).
It also confirms if the CSV is UTF-8 encoded.
For both modes, returns exit code 0 when the CSV file is valid, exitcode > 0 otherwise.
If all records are valid, no output files are produced.
Examples:
# Validate a CSV file. Use this to check if a CSV file is readable by qsv.
qsv validate data.csv
# Validate a TSV file against a JSON Schema
qsv validate data.tsv schema.json
# Validate multiple CSV files using various dialects against a JSON Schema
qsv validate data1.csv data2.tab data3.ssv schema.json
# Validate all CSV files in a directory against a JSON Schema
qsv validate /path/to/csv_directory schema.json
# Validate CSV files listed in a '.infile-list' file against a JSON Schema
qsv validate files.infile-list schema.json
For more examples, see the tests included in this file (denoted by '#[test]') or see
https://github.com/dathere/qsv/blob/master/tests/test_validate.rs.
Usage:
qsv validate schema [--no-format-validation] [<json-schema>]
qsv validate [options] [<input>...]
qsv validate [options] [<input>] <json-schema>
qsv validate --help
Validate arguments:
<input>... Input CSV file(s) to validate. If not provided, will read from stdin.
If input is a directory, all files in the directory will be validated.
If the input is a file with a '.infile-list' extension, the file will
be read as a list of input files. If the input are snappy-compressed
files(s), it will be decompressed automatically.
Extended Input Support is only available for RFC 4180 validation mode.
<json-schema> JSON Schema file to validate against. If not provided, `validate`
will run in RFC 4180 validation mode. The file can be a local file
or a URL (http and https schemes supported).
Validate options:
--trim Trim leading and trailing whitespace from fields before validating.
--no-format-validation Disable JSON Schema format validation. Ignores all JSON Schema
"format" keywords (e.g. date,email, uri, currency, etc.). This is
useful when you want to validate the structure of the CSV file
w/o worrying about the data types and domain/range of the fields.
--fail-fast Stops on first error.
--valid <suffix> Valid record output file suffix. [default: valid]
--invalid <suffix> Invalid record output file suffix. [default: invalid]
--json When validating without a JSON Schema, return the RFC 4180 check
as a JSON file instead of a message.
--pretty-json Same as --json, but pretty printed.
--valid-output <file> Change validation mode behavior so if ALL rows are valid, to pass it to
output, return exit code 1, and set stderr to the number of valid rows.
Setting this will override the default behavior of creating
a valid file only when there are invalid records.
To send valid records to stdout, use `-` as the filename.
-j, --jobs <arg> The number of jobs to run in parallel.
When not set, the number of jobs is set to the
number of CPUs detected.
-b, --batch <size> The number of rows per batch to load into memory,
before running in parallel. Automatically determined
for CSV files with more than 50000 rows.
Set to 0 to load all rows in one batch.
Set to 1 to force batch optimization even for files with
less than 50000 rows. [default: 50000]
FANCY REGEX OPTIONS:
--fancy-regex Use the fancy regex engine instead of the default regex engine
for validation.
The fancy engine supports advanced regex features such as
lookaround and backreferences, but is not as performant as
the default regex engine which guarantees linear-time matching,
prevents DoS attacks, and is more efficient for simple patterns.
--backtrack-limit <limit> Set the approximate number of backtracking steps allowed.
This is only used when --fancy-regex is set.
[default: 1000000]
OPTIONS FOR BOTH REGEX ENGINES:
--size-limit <mb> Set the approximate size limit, in megabytes, of a compiled regex.
[default: 50]
--dfa-size-limit <mb> Set the approximate capacity, in megabytes, of the cache of transitions
used by the engine's lazy Discrete Finite Automata.
[default: 10]
--timeout <seconds> Timeout for downloading json-schemas on URLs and for
'dynamicEnum' lookups on URLs. If 0, no timeout is used.
[default: 30]
--cache-dir <dir> The directory to use for caching downloaded dynamicEnum resources.
If the directory does not exist, qsv will attempt to create it.
If the QSV_CACHE_DIR envvar is set, it will be used instead.
Not available on qsvlite.
[default: ~/.qsv-cache]
--ckan-api <url> The URL of the CKAN API to use for downloading dynamicEnum
resources with the "ckan://" scheme.
If the QSV_CKAN_API envvar is set, it will be used instead.
Not available on qsvlite.
[default: https://data.dathere.com/api/3/action]
--ckan-token <token> The CKAN API token to use. Only required if downloading
private resources.
If the QSV_CKAN_TOKEN envvar is set, it will be used instead.
Not available on qsvlite.
EMAIL VALIDATION OPTIONS:
--email-required-tld Require the email to have a valid Top-Level Domain (TLD)
(e.g. .com, .org, .net, etc.).
e.g. "john.doe@example" is VALID if this option is NOT set.
--email-display-text Allow display text in emails.
e.g. "John Doe <john.doe@example.com>" is INVALID if this option is NOT set.
--email-min-subdomains <n> Minimum number of subdomains required in the email.
e.g. "jdoe@example.com" is INVALID if this option is set to 3,
but "jdoe@sub.example.com" is VALID.
[default: 2]
--email-domain-literal Allow domain literals in emails.
e.g. "john.doe@[127.0.0.1]" is VALID if this option is set.
Common options:
-h, --help Display this message
-n, --no-headers When set, the first row will not be interpreted
as headers. It will be validated with the rest
of the rows. Otherwise, the first row will always
appear as the header row in the output.
Note that this option is only valid when running
in RFC 4180 validation mode as JSON Schema validation
requires headers.
-d, --delimiter <arg> The field delimiter for reading CSV data.
Must be a single character.
-p, --progressbar Show progress bars. Not valid for stdin.
-q, --quiet Do not display validation summary message.
"#;
use std::{
env,
fs::File,
io::{BufReader, BufWriter, Read, Write},
path::PathBuf,
str,
sync::{
OnceLock,
atomic::{AtomicU16, Ordering},
},
};
use bitvec::prelude::*;
use csv::ByteRecord;
use foldhash::{HashSet, HashSetExt};
use indicatif::HumanCount;
#[cfg(any(feature = "feature_capable", feature = "lite"))]
use indicatif::{ProgressBar, ProgressDrawTarget};
use jsonschema::{
EmailOptions, Keyword, PatternOptions, ValidationError, Validator, paths::Location,
};
use log::debug;
use qsv_currency::Currency;
use rayon::{
iter::{IndexedParallelIterator, ParallelIterator},
prelude::IntoParallelRefIterator,
};
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value, json, value::Number};
#[cfg(feature = "lite")]
use tempfile::NamedTempFile;
#[cfg(not(feature = "lite"))]
use crate::lookup;
#[cfg(not(feature = "lite"))]
use crate::lookup::{LookupTableOptions, load_lookup_table};
use crate::{
CliError, CliResult,
config::{Config, DEFAULT_RDR_BUFFER_CAPACITY, DEFAULT_WTR_BUFFER_CAPACITY, Delimiter},
util,
};
static NULL_TYPE: OnceLock<Value> = OnceLock::new();
static TIMEOUT_SECS: AtomicU16 = AtomicU16::new(30);
#[cfg(not(feature = "lite"))]
static QSV_CACHE_DIR: OnceLock<String> = OnceLock::new();
#[cfg(not(feature = "lite"))]
static CKAN_API: OnceLock<String> = OnceLock::new();
#[cfg(not(feature = "lite"))]
static CKAN_TOKEN: OnceLock<Option<String>> = OnceLock::new();
static DELIMITER: OnceLock<Option<Delimiter>> = OnceLock::new();
macro_rules! fail_validation_error {
($($t:tt)*) => {{
use log::error;
let err = format!($($t)*);
error!("{err}");
Err(ValidationError::custom(err))
}};
}
#[derive(Deserialize)]
#[allow(dead_code)]
struct Args {
cmd_schema: bool,
flag_trim: bool,
flag_no_format_validation: bool,
flag_fail_fast: bool,
flag_valid: Option<String>,
flag_invalid: Option<String>,
flag_json: bool,
flag_pretty_json: bool,
flag_valid_output: Option<String>,
flag_jobs: Option<usize>,
flag_batch: usize,
flag_no_headers: bool,
flag_delimiter: Option<Delimiter>,
flag_progressbar: bool,
flag_quiet: bool,
arg_input: Vec<std::path::PathBuf>,
arg_json_schema: Option<String>,
flag_fancy_regex: bool,
flag_backtrack_limit: usize,
flag_size_limit: usize,
flag_dfa_size_limit: usize,
flag_timeout: u16,
flag_cache_dir: String,
flag_ckan_api: String,
flag_ckan_token: Option<String>,
flag_email_required_tld: bool,
flag_email_display_text: bool,
flag_email_min_subdomains: usize,
flag_email_domain_literal: bool,
}
enum JSONtypes {
String,
Number,
Integer,
Boolean,
Unsupported,
}
#[derive(Serialize, Deserialize)]
struct RFC4180Struct {
delimiter_char: char,
header_row: bool,
quote_char: char,
num_records: u64,
num_fields: u64,
fields: Vec<String>,
}
impl From<ValidationError<'_>> for CliError {
fn from(err: ValidationError) -> CliError {
CliError::Other(format!("{err}"))
}
}
#[inline]
fn currency_format_checker(s: &str) -> bool {
Currency::from_str(s).is_ok_and(|c| {
if c.symbol().is_empty() {
true } else {
qsv_currency::Currency::is_iso_currency(&c)
}
})
}
struct DynEnumValidator {
dynenum_set: HashSet<String>,
}
impl DynEnumValidator {
#[allow(dead_code)]
const fn new(dynenum_set: HashSet<String>) -> Self {
Self { dynenum_set }
}
}
impl Keyword for DynEnumValidator {
#[inline]
fn validate<'instance>(
&self,
instance: &'instance Value,
) -> Result<(), ValidationError<'instance>> {
if self.dynenum_set.contains(instance.as_str().unwrap()) {
Ok(())
} else {
let error =
ValidationError::custom(format!("{instance} is not a valid dynamicEnum value"));
Err(error)
}
}
#[inline]
fn is_valid(&self, instance: &Value) -> bool {
if let Value::String(s) = instance {
self.dynenum_set.contains(s)
} else {
false
}
}
}
struct UniqueCombinedWithValidator {
column_names: Vec<String>,
column_indices: Vec<usize>,
seen_combinations: std::sync::RwLock<HashSet<String>>,
}
impl UniqueCombinedWithValidator {
fn new(column_names: Vec<String>, column_indices: Vec<usize>) -> Self {
Self {
column_names,
column_indices,
seen_combinations: std::sync::RwLock::new(HashSet::new()),
}
}
}
impl Keyword for UniqueCombinedWithValidator {
fn validate<'instance>(
&self,
instance: &'instance Value,
) -> Result<(), ValidationError<'instance>> {
let obj = instance
.as_object()
.ok_or_else(|| ValidationError::custom("Instance must be an object"))?;
let mut values = Vec::with_capacity(self.column_names.len() + self.column_indices.len());
for name in &self.column_names {
if let Some(value) = obj.get(name) {
values.push(value.to_string());
}
}
if !self.column_indices.is_empty() {
let array: Vec<_> = obj.values().collect();
for &idx in &self.column_indices {
if let Some(value) = array.get(idx) {
values.push(value.to_string());
}
}
}
let combination = values.join("|");
let mut seen = self.seen_combinations.write().unwrap();
if seen.contains(&combination) {
let mut column_desc_parts =
Vec::with_capacity(self.column_names.len() + self.column_indices.len());
if !self.column_names.is_empty() {
column_desc_parts.extend(self.column_names.iter().cloned());
}
if !self.column_indices.is_empty() {
column_desc_parts.extend(
self.column_indices
.iter()
.map(std::string::ToString::to_string),
);
}
let column_desc = column_desc_parts.join(", ");
return Err(ValidationError::custom(format!(
"Combination of values for columns {column_desc} is not unique"
)));
}
seen.insert(combination);
drop(seen);
Ok(())
}
fn is_valid(&self, instance: &Value) -> bool {
let Some(obj) = instance.as_object() else {
return false;
};
let mut values = Vec::with_capacity(self.column_names.len() + self.column_indices.len());
for name in &self.column_names {
if let Some(value) = obj.get(name) {
values.push(value.to_string());
}
}
if !self.column_indices.is_empty() {
let array: Vec<_> = obj.values().collect();
for &idx in &self.column_indices {
if let Some(value) = array.get(idx) {
values.push(value.to_string());
}
}
}
let combination = values.join("|");
let seen = self.seen_combinations.read().unwrap();
!seen.contains(&combination)
}
}
#[allow(clippy::result_large_err)]
fn unique_combined_with_validator_factory<'a>(
_parent: &'a Map<String, Value>,
value: &'a Value,
_location: Location,
) -> Result<Box<dyn Keyword>, ValidationError<'a>> {
let columns = value.as_array().ok_or_else(|| {
ValidationError::custom("'uniqueCombinedWith' must be an array of column names or indices")
})?;
let col_len = columns.len();
let mut column_names = Vec::with_capacity(col_len);
let mut column_indices = Vec::with_capacity(col_len);
for col in columns {
if let Some(idx) = col.as_u64() {
column_indices.push(idx as usize);
} else {
let name = col.as_str().ok_or_else(|| {
ValidationError::custom("Column names must be strings or numbers")
})?;
column_names.push(name.to_string());
}
}
if column_names.is_empty() && column_indices.is_empty() {
return Err(ValidationError::custom(
"'uniqueCombinedWith' must specify at least one column",
));
}
Ok(Box::new(UniqueCombinedWithValidator::new(
column_names,
column_indices,
)))
}
#[cfg(not(feature = "lite"))]
fn parse_dynenum_uri(uri: &str) -> (String, String, i64, Option<String>) {
const DEFAULT_CACHE_AGE_SECS: i64 = 3600;
fn get_cache_name(uri: &str) -> String {
if uri.contains("://") {
let after_scheme = uri.split("://").nth(1).unwrap_or(uri);
after_scheme
.split('/')
.next_back()
.unwrap_or(after_scheme)
.trim_end_matches(".csv")
.to_string()
} else {
uri.split(['/', '\\'])
.next_back()
.unwrap_or(uri)
.trim_end_matches(".csv")
.to_string()
}
}
if !uri.contains('|') {
let final_uri = uri.to_string();
let cache_name = get_cache_name(&final_uri);
return (cache_name, final_uri, DEFAULT_CACHE_AGE_SECS, None);
}
let parts: Vec<&str> = uri.split('|').collect();
let (final_uri, cache_name, cache_age) = if parts[0].contains(';') {
let config_parts: Vec<&str> = parts[0].split(';').collect();
let name = if config_parts[0].is_empty() {
get_cache_name(parts[1])
} else {
config_parts[0].trim_end_matches(".csv").to_string()
};
let age = config_parts
.get(1)
.and_then(|s| s.parse::<i64>().ok())
.unwrap_or(DEFAULT_CACHE_AGE_SECS);
(parts[1].to_string(), name, age)
} else if parts[1].contains("://") {
(
parts[1].to_string(),
get_cache_name(parts[0]),
DEFAULT_CACHE_AGE_SECS,
)
} else {
(
parts[0].to_string(),
get_cache_name(parts[0]),
DEFAULT_CACHE_AGE_SECS,
)
};
let column = if parts.len() > 2 {
Some(parts[2].to_string())
} else if parts.len() == 2
&& !parts[1].contains("://")
&& !parts[1].to_lowercase().ends_with(".csv")
{
Some(parts[1].to_string())
} else {
None
};
(cache_name, final_uri, cache_age, column)
}
#[cfg(not(feature = "lite"))]
#[test]
fn test_parse_dynenum_uri() {
let (cache_name, uri, cache_age, column) = parse_dynenum_uri("https://example.com/data.csv");
assert_eq!(cache_name, "data");
assert_eq!(uri, "https://example.com/data.csv");
assert_eq!(cache_age, 3600);
assert_eq!(column, None);
let (cache_name, uri, cache_age, column) =
parse_dynenum_uri("custom_name;600|https://example.com/data.csv");
assert_eq!(cache_name, "custom_name");
assert_eq!(uri, "https://example.com/data.csv");
assert_eq!(cache_age, 600);
assert_eq!(column, None);
let (cache_name, uri, cache_age, column) = parse_dynenum_uri("lookup.csv|name");
assert_eq!(cache_name, "lookup");
assert_eq!(uri, "lookup.csv");
assert_eq!(cache_age, 3600);
assert_eq!(column, Some("name".to_string()));
let (cache_name, uri, cache_age, column) = parse_dynenum_uri("MyCache;1800|lookup.csv|code");
assert_eq!(cache_name, "MyCache");
assert_eq!(uri, "lookup.csv");
assert_eq!(cache_age, 1800);
assert_eq!(column, Some("code".to_string()));
let (cache_name, uri, cache_age, column) = parse_dynenum_uri(";1800|lookup.csv|code");
assert_eq!(cache_name, "lookup");
assert_eq!(uri, "lookup.csv");
assert_eq!(cache_age, 1800);
assert_eq!(column, Some("code".to_string()));
let (cache_name, uri, cache_age, column) = parse_dynenum_uri(";1800|lookup.csv");
assert_eq!(cache_name, "lookup");
assert_eq!(uri, "lookup.csv");
assert_eq!(cache_age, 1800);
assert_eq!(column, None);
let (cache_name, uri, cache_age, column) = parse_dynenum_uri("lookup.csv");
assert_eq!(cache_name, "lookup");
assert_eq!(uri, "lookup.csv");
assert_eq!(cache_age, 3600);
assert_eq!(column, None);
let (cache_name, uri, cache_age, column) = parse_dynenum_uri(r#"c:\Users\jdoe\lookup.csv"#);
assert_eq!(cache_name, "lookup");
assert_eq!(uri, r#"c:\Users\jdoe\lookup.csv"#);
assert_eq!(cache_age, 3600);
assert_eq!(column, None);
let (cache_name, uri, cache_age, column) = parse_dynenum_uri("/tmp/lookup.csv|first_col");
assert_eq!(cache_name, "lookup");
assert_eq!(uri, "/tmp/lookup.csv");
assert_eq!(cache_age, 3600);
assert_eq!(column, Some("first_col".to_string()));
let (cache_name, uri, cache_age, column) = parse_dynenum_uri("LookUp.csv");
assert_eq!(cache_name, "LookUp");
assert_eq!(uri, "LookUp.csv");
assert_eq!(cache_age, 3600);
assert_eq!(column, None);
let (cache_name, uri, cache_age, column) =
parse_dynenum_uri("NYC_neighborhood_data|ckan://nyc_neighborhoods?");
assert_eq!(cache_name, "NYC_neighborhood_data");
assert_eq!(uri, "ckan://nyc_neighborhoods?");
assert_eq!(cache_age, 3600);
assert_eq!(column, None);
let (cache_name, uri, cache_age, column) =
parse_dynenum_uri("NYC_neighborhood_data;5000|ckan://nyc_neighborhoods?");
assert_eq!(cache_name, "NYC_neighborhood_data");
assert_eq!(uri, "ckan://nyc_neighborhoods?");
assert_eq!(cache_age, 5000);
assert_eq!(column, None);
let (cache_name, uri, cache_age, column) =
parse_dynenum_uri("NYC_neighborhood_data;5000|ckan://nyc_neighborhoods?|Neighborhood_Col");
assert_eq!(cache_name, "NYC_neighborhood_data");
assert_eq!(uri, "ckan://nyc_neighborhoods?");
assert_eq!(cache_age, 5000);
assert_eq!(column, Some("Neighborhood_Col".to_string()));
let (cache_name, uri, cache_age, column) = parse_dynenum_uri("dathere://us_states.csv");
assert_eq!(cache_name, "us_states");
assert_eq!(uri, "dathere://us_states.csv");
assert_eq!(cache_age, 3600);
assert_eq!(column, None);
let (cache_name, uri, cache_age, column) =
parse_dynenum_uri("dathere://us_states.csv|state_col");
assert_eq!(cache_name, "us_states");
assert_eq!(uri, "dathere://us_states.csv");
assert_eq!(cache_age, 3600);
assert_eq!(column, Some("state_col".to_string()));
let (cache_name, uri, cache_age, column) =
parse_dynenum_uri("usl_lookup;6000|dathere://us_states.csv|state_col");
assert_eq!(cache_name, "usl_lookup");
assert_eq!(uri, "dathere://us_states.csv");
assert_eq!(cache_age, 6000);
assert_eq!(column, Some("state_col".to_string()));
}
#[cfg(not(feature = "lite"))]
#[allow(clippy::result_large_err)]
fn dyn_enum_validator_factory<'a>(
_parent: &'a Map<String, Value>,
value: &'a Value,
_location: Location,
) -> Result<Box<dyn Keyword>, ValidationError<'a>> {
let uri = value.as_str().ok_or_else(|| {
ValidationError::custom(
"'dynamicEnum' must be set to a CSV file on the local filesystem or on a URL.",
)
})?;
let (lookup_name, final_uri, cache_age_secs, column) = parse_dynenum_uri(uri);
let opts = LookupTableOptions {
name: lookup_name,
uri: final_uri,
cache_age_secs,
cache_dir: QSV_CACHE_DIR.get().unwrap().to_string(),
delimiter: DELIMITER.get().copied().flatten(),
ckan_api_url: CKAN_API.get().cloned(),
ckan_token: CKAN_TOKEN.get().and_then(std::clone::Clone::clone),
timeout_secs: TIMEOUT_SECS.load(Ordering::Relaxed),
};
let lookup_result = match load_lookup_table(&opts) {
Ok(result) => result,
Err(e) => return fail_validation_error!("Error loading dynamicEnum lookup table: {e}"),
};
let mut enum_set = HashSet::with_capacity(lookup_result.headers.len());
let rconfig = Config::new(Some(lookup_result.filepath).as_ref());
let mut rdr = match rconfig
.flexible(true)
.comment(Some(b'#'))
.skip_format_check(true)
.reader()
{
Ok(reader) => reader,
Err(e) => return fail_validation_error!("Error opening dynamicEnum file: {e}"),
};
let column_idx = if let Some(col_name) = column {
if let Ok(idx) = col_name.parse::<usize>() {
idx
} else {
match rdr.headers() {
Ok(headers) => {
let idx = headers.iter().position(|h| h == col_name);
match idx {
Some(i) => i,
None => {
return fail_validation_error!(
"Column '{}' not found in lookup table",
col_name
);
},
}
},
Err(e) => return fail_validation_error!("Error reading headers: {e}"),
}
}
} else {
0
};
for result in rdr.records() {
match result {
Ok(record) => {
if let Some(value) = record.get(column_idx) {
enum_set.insert(value.to_owned());
}
},
Err(e) => return fail_validation_error!("Error reading dynamicEnum file - {e}"),
}
}
Ok(Box::new(DynEnumValidator::new(enum_set)))
}
#[cfg(feature = "lite")]
#[allow(clippy::result_large_err)]
fn dyn_enum_validator_factory<'a>(
_parent: &'a Map<String, Value>,
value: &'a Value,
_location: Location,
) -> Result<Box<dyn Keyword>, ValidationError<'a>> {
if let Value::String(uri) = value {
let temp_download = match NamedTempFile::new() {
Ok(file) => file,
Err(e) => return fail_validation_error!("Failed to create temporary file: {e}"),
};
let parts: Vec<&str> = uri.split('|').collect();
let base_uri = parts[0];
let column = parts.get(1).map(std::string::ToString::to_string);
let dynenum_path = if base_uri.starts_with("http") {
let valid_url = reqwest::Url::parse(base_uri).map_err(|e| {
ValidationError::custom(format!("Error parsing dynamicEnum URL: {e}"))
})?;
let download_timeout = TIMEOUT_SECS.load(Ordering::Relaxed);
let future = util::download_file(
valid_url.as_str(),
temp_download.path().to_path_buf(),
false,
None,
Some(download_timeout),
None,
);
match tokio::runtime::Runtime::new() {
Ok(runtime) => {
if let Err(e) = runtime.block_on(future) {
return fail_validation_error!("Error downloading dynamicEnum file - {e}");
}
},
Err(e) => {
return fail_validation_error!("Error creating Tokio runtime - {e}");
},
}
temp_download.path().to_str().unwrap().to_string()
} else {
let uri_path = std::path::Path::new(base_uri);
let uri_exists = uri_path.exists();
if !uri_exists {
return fail_validation_error!("dynamicEnum file not found - {base_uri}");
}
uri_path.to_str().unwrap().to_string()
};
let mut enum_set = HashSet::with_capacity(50);
let rconfig = Config::new(Some(dynenum_path).as_ref());
let mut rdr = match rconfig
.flexible(true)
.comment(Some(b'#'))
.skip_format_check(true)
.reader()
{
Ok(reader) => reader,
Err(e) => return fail_validation_error!("Error opening dynamicEnum file: {e}"),
};
let column_idx = if let Some(col_name) = column {
if let Ok(idx) = col_name.parse::<usize>() {
idx
} else {
match rdr.headers() {
Ok(headers) => headers.iter().position(|h| h == col_name).unwrap_or(0),
Err(_) => 0,
}
}
} else {
0
};
for result in rdr.records() {
match result {
Ok(record) => {
if let Some(value) = record.get(column_idx) {
enum_set.insert(value.to_owned());
}
},
Err(e) => return fail_validation_error!("Error reading dynamicEnum file - {e}"),
};
}
Ok(Box::new(DynEnumValidator::new(enum_set)))
} else {
Err(ValidationError::custom(
"'dynamicEnum' must be set to a CSV file on the local filesystem or on a URL.",
))
}
}
pub fn run(argv: &[&str]) -> CliResult<()> {
let args: Args = util::get_args(USAGE, argv)?;
if args.cmd_schema {
if let Some(ref schema) = args.arg_json_schema {
let schema_json_string = load_json(schema)?;
let schema_json = serde_json::from_str(&schema_json_string)?;
if let Err(e) = jsonschema::meta::validate(&schema_json) {
return fail_clierror!("JSON Schema Meta-Reference Error: {e}");
}
let test_validator = if args.flag_no_format_validation {
Validator::options()
.should_validate_formats(false)
.should_ignore_unknown_formats(true)
.build(&schema_json)
} else {
Validator::options()
.should_validate_formats(true)
.should_ignore_unknown_formats(false)
.build(&schema_json)
};
if let Err(e) = test_validator {
return fail_clierror!("JSON Schema Format Validation Error: {e}");
}
if !args.flag_quiet {
winfo!("Valid JSON Schema.");
}
return Ok(());
}
return fail_clierror!("No JSON Schema file supplied.");
}
TIMEOUT_SECS.store(
util::timeout_secs(args.flag_timeout)? as u16,
Ordering::Relaxed,
);
let has_json_schema = if let Some(last_input) = args.arg_input.last() {
last_input
.extension()
.and_then(std::ffi::OsStr::to_str)
.is_some_and(|ext| ext.to_lowercase() == "json")
} else {
false
};
if !has_json_schema && args.arg_json_schema.is_none() {
return validate_rfc4180_mode(&args);
}
let (input_files, json_schema_path) = if has_json_schema {
let schema_path = args.arg_input.last().unwrap().clone();
let input_files = args.arg_input[..args.arg_input.len() - 1].to_vec();
(input_files, Some(schema_path))
} else {
(args.arg_input.clone(), None)
};
let json_schema_arg = if let Some(schema_path) = &json_schema_path {
Some(schema_path.to_string_lossy().to_string())
} else {
args.arg_json_schema.clone()
};
if input_files.len() > 1 {
return fail_clierror!(
"JSON Schema validation only supports a single input file. Use RFC 4180 validation \
mode for multiple files."
);
}
let input_path = input_files.first().ok_or_else(|| {
if has_json_schema && args.arg_input.len() == 1 {
CliError::Other(
"Only a JSON Schema file was provided, but no data file to validate. Please \
provide a data file to validate against the schema."
.to_string(),
)
} else {
CliError::Other("No input file provided for JSON Schema validation".to_string())
}
})?;
let mut rconfig = Config::new(Some(&input_path.to_string_lossy().to_string()))
.no_headers_flag(args.flag_no_headers)
.set_read_buffer(if std::env::var("QSV_RDR_BUFFER_CAPACITY").is_err() {
DEFAULT_RDR_BUFFER_CAPACITY * 10
} else {
DEFAULT_RDR_BUFFER_CAPACITY
});
if args.flag_delimiter.is_some() {
rconfig = rconfig.delimiter(args.flag_delimiter);
}
DELIMITER.set(args.flag_delimiter).unwrap();
let mut rdr = rconfig.reader()?;
if args.flag_no_headers {
return fail_clierror!("Cannot validate CSV without headers against a JSON Schema.");
}
#[cfg(any(feature = "feature_capable", feature = "lite"))]
let progress = ProgressBar::with_draw_target(None, ProgressDrawTarget::stderr_with_hz(5));
#[cfg(any(feature = "feature_capable", feature = "lite"))]
let show_progress =
(args.flag_progressbar || util::get_envvar_flag("QSV_PROGRESSBAR")) && !rconfig.is_stdin();
rconfig = rconfig.flexible(true);
let record_count = util::count_rows(&rconfig)?;
rconfig = rconfig.flexible(false);
#[cfg(any(feature = "feature_capable", feature = "lite"))]
if show_progress {
util::prep_progress(&progress, record_count);
} else {
progress.set_draw_target(ProgressDrawTarget::hidden());
}
let headers = rdr.byte_headers()?.clone();
let header_len = headers.len();
#[cfg(not(feature = "lite"))]
let qsv_cache_dir = lookup::set_qsv_cache_dir(&args.flag_cache_dir)?;
#[cfg(not(feature = "lite"))]
QSV_CACHE_DIR.set(qsv_cache_dir)?;
#[cfg(not(feature = "lite"))]
CKAN_API.set(if let Ok(api) = std::env::var("QSV_CKAN_API") {
api
} else {
args.flag_ckan_api.clone()
})?;
#[cfg(not(feature = "lite"))]
CKAN_TOKEN
.set(if let Ok(token) = std::env::var("QSV_CKAN_TOKEN") {
Some(token)
} else {
args.flag_ckan_token.clone()
})
.unwrap();
let json_schema_path =
json_schema_path.unwrap_or_else(|| PathBuf::from(json_schema_arg.as_ref().unwrap()));
let (schema_json, schema_compiled, has_unique_combined): (Value, Validator, bool) =
match load_json(&json_schema_path.to_string_lossy()) {
Ok(s) => {
let has_currency_format = s.contains(r#""format": "currency""#);
let has_dynamic_enum = s.contains("dynamicEnum");
let has_unique_combined = s.contains("uniqueCombinedWith");
let has_email_format = s.contains(r#""format": "email""#);
debug!("Custom formats/keywords: currency: {has_currency_format}, dynamicEnum: {has_dynamic_enum}");
debug!("uniqueCombinedWith: {has_unique_combined}, email: {has_email_format}");
#[cfg(target_endian = "big")]
let json_result = serde_json::from_str::<Value>(&s);
#[cfg(target_endian = "little")]
let json_result = {
let mut s_slice = s.as_bytes().to_vec();
simd_json::serde::from_slice::<Value>(&mut s_slice)
};
match json_result {
Ok(json) => {
let mut validator_options = Validator::options()
.should_validate_formats(!args.flag_no_format_validation);
if has_email_format {
let mut email_options = EmailOptions::default();
if args.flag_email_required_tld {
email_options = email_options.with_required_tld();
}
if !args.flag_email_display_text {
email_options = email_options.without_display_text();
}
if args.flag_email_min_subdomains > 2 {
email_options = email_options.with_minimum_sub_domains(args.flag_email_min_subdomains);
}
if !args.flag_email_domain_literal {
email_options = email_options.without_domain_literal();
}
validator_options = validator_options.with_email_options(email_options);
}
if has_currency_format {
validator_options = validator_options.with_format("currency", currency_format_checker);
}
if has_dynamic_enum {
validator_options = validator_options.with_keyword("dynamicEnum", dyn_enum_validator_factory);
}
if has_unique_combined {
validator_options = validator_options.with_keyword("uniqueCombinedWith", unique_combined_with_validator_factory);
}
if args.flag_fancy_regex {
let fancy_regex_options = PatternOptions::fancy_regex()
.backtrack_limit(args.flag_backtrack_limit)
.size_limit(args.flag_size_limit * (1 << 20))
.dfa_size_limit(args.flag_dfa_size_limit * (1 << 20));
validator_options = validator_options.with_pattern_options(fancy_regex_options);
} else {
let regex_options = PatternOptions::regex()
.size_limit(args.flag_size_limit * (1 << 20))
.dfa_size_limit(args.flag_dfa_size_limit * (1 << 20));
validator_options = validator_options.with_pattern_options(regex_options);
}
match validator_options.build(&json) {
Ok(schema) => (json, schema, has_unique_combined),
Err(e) => {
return fail_clierror!(r#"Cannot compile JSONschema. error: {e}
Try running `qsv validate schema {}` to check the JSON Schema file."#, json_schema_path.to_string_lossy());
},
}
},
Err(e) => {
return fail_clierror!(r#"Unable to parse JSONschema. error: {e}
Try running `qsv validate schema {}` to check the JSON Schema file."#, json_schema_arg.as_ref().unwrap());
},
}
},
Err(e) => {
return fail_clierror!("Unable to retrieve JSONschema. error: {e}");
},
};
if log::log_enabled!(log::Level::Debug) {
debug!("schema json: {:?}", &schema_json);
}
NULL_TYPE.set(Value::String("null".to_string())).unwrap();
let header_types = get_json_types(&headers, &schema_json)?;
let mut row_number: u64 = 0;
let mut invalid_count: u64 = 0;
let mut record = csv::ByteRecord::with_capacity(500, header_len);
let num_jobs = util::njobs(args.flag_jobs);
let mut valid_flags: BitVec = BitVec::with_capacity(record_count as usize);
let batch_size = util::optimal_batch_size(&rconfig, args.flag_batch, num_jobs);
let mut batch = Vec::with_capacity(batch_size);
let mut batch_validation_results: Vec<Option<String>> = Vec::with_capacity(batch_size);
let mut validation_error_messages: Vec<String> = Vec::with_capacity(50);
let flag_trim = args.flag_trim;
let flag_fail_fast = args.flag_fail_fast;
let mut itoa_buffer = itoa::Buffer::new();
let batch_pariter_min_len = batch_size / num_jobs;
'batch_loop: loop {
for _ in 0..batch_size {
match rdr.read_byte_record(&mut record) {
Ok(true) => {
row_number += 1;
record.push_field(itoa_buffer.format(row_number).as_bytes());
if flag_trim {
record.trim();
}
batch.push(std::mem::take(&mut record));
},
Ok(false) => break, Err(e) => {
return fail_clierror!("Error reading row: {row_number}: {e}");
},
}
}
if batch.is_empty() {
break 'batch_loop;
}
batch
.par_iter()
.with_min_len(batch_pariter_min_len)
.map(|record| {
let json_instance = match to_json_instance(&header_types, header_len, record) {
Ok(obj) => obj,
Err(e) => {
let row_number_string = unsafe {
simdutf8::basic::from_utf8(&record[header_len]).unwrap_unchecked()
};
return Some(format!("{row_number_string}\t<RECORD>\t{e}"));
},
};
let evaluation = if !has_unique_combined && schema_compiled.is_valid(&json_instance)
{
return None;
} else {
schema_compiled.evaluate(&json_instance)
};
if evaluation.flag().valid {
None
} else {
let row_number_string = unsafe {
simdutf8::basic::from_utf8(&record[header_len]).unwrap_unchecked()
};
let errors: Vec<_> = evaluation.iter_errors().collect();
let mut error_messages = Vec::with_capacity(errors.len());
for e in errors {
error_messages.push(format!(
"{row_number_string}\t{field}\t{error}",
field = e.instance_location.as_str().trim_start_matches('/'),
error = e.error
));
}
Some(error_messages.join("\n"))
}
})
.collect_into_vec(&mut batch_validation_results);
let start_idx = valid_flags.len();
valid_flags.extend(std::iter::repeat_n(true, batch_size));
for (i, result) in batch_validation_results.iter().enumerate() {
if let Some(validation_error_msg) = result {
invalid_count += 1;
unsafe { valid_flags.set_unchecked(start_idx + i, false) };
validation_error_messages.push(validation_error_msg.to_owned());
}
}
#[cfg(any(feature = "feature_capable", feature = "lite"))]
if show_progress {
progress.inc(batch.len() as u64);
}
batch.clear();
if flag_fail_fast && invalid_count > 0 {
break 'batch_loop;
}
}
#[cfg(any(feature = "feature_capable", feature = "lite"))]
if show_progress {
progress.set_message(format!(
" validated {} records.",
HumanCount(progress.length().unwrap())
));
util::finish_progress(&progress);
}
if invalid_count == 0 {
if let Some(valid_output) = args.flag_valid_output {
let valid_path = if valid_output == "-" {
None
} else {
Some(valid_output)
};
let mut valid_wtr = Config::new(valid_path.as_ref()).writer()?;
valid_wtr.write_byte_record(&headers)?;
let mut rdr = rconfig.reader()?;
let mut record = csv::ByteRecord::new();
while rdr.read_byte_record(&mut record)? {
valid_wtr.write_byte_record(&record)?;
}
valid_wtr.flush()?;
return fail_clierror!("{row_number}");
}
} else {
woutinfo!("Writing invalid/valid/error files...");
let input_path = args.arg_input.first().map_or_else(
|| "stdin.csv".to_string(),
|p| p.to_string_lossy().to_string(),
);
write_error_report(&input_path, validation_error_messages)?;
let valid_suffix = args.flag_valid.unwrap_or_else(|| "valid".to_string());
let invalid_suffix = args.flag_invalid.unwrap_or_else(|| "invalid".to_string());
split_invalid_records(
&rconfig,
&valid_flags[..],
&headers,
&input_path,
&valid_suffix,
&invalid_suffix,
)?;
let fail_fast_msg = if args.flag_fail_fast {
format!(
"fail-fast enabled. stopped after row {}.\n",
HumanCount(row_number)
)
} else {
String::new()
};
return fail_clierror!(
"{fail_fast_msg}{} out of {} records invalid.",
HumanCount(invalid_count),
HumanCount(row_number)
);
}
if !args.flag_quiet {
winfo!("All {} records valid.", HumanCount(row_number));
}
Ok(())
}
fn validate_rfc4180_mode(args: &Args) -> CliResult<()> {
use tempfile::tempdir;
let tmpdir = tempdir()?;
let processed_inputs = util::process_input(args.arg_input.clone(), &tmpdir, "")?;
let input_count = processed_inputs.len();
let flag_json = args.flag_json || args.flag_pretty_json;
let flag_pretty_json = args.flag_pretty_json;
let mut all_valid = true;
let mut total_files = 0;
let mut valid_files = 0;
for input_path in processed_inputs {
total_files += 1;
if !args.flag_quiet && input_count > 1 {
woutinfo!("Validating: {}", input_path.display());
}
let mut rconfig = Config::new(Some(&input_path.to_string_lossy().to_string()))
.no_headers_flag(args.flag_no_headers)
.set_read_buffer(if std::env::var("QSV_RDR_BUFFER_CAPACITY").is_err() {
DEFAULT_RDR_BUFFER_CAPACITY * 10
} else {
DEFAULT_RDR_BUFFER_CAPACITY
});
if args.flag_delimiter.is_some() {
rconfig = rconfig.delimiter(args.flag_delimiter);
}
let mut rdr = match rconfig.reader() {
Ok(reader) => reader,
Err(e) => {
if flag_json {
let file_error = json!({
"errors": [{
"title": "File validation error",
"detail": format!("Cannot read file {}: {}", input_path.display(), e),
"meta": {
"file": input_path.to_string_lossy()
}
}]
});
let json_error = if flag_pretty_json {
simd_json::to_string_pretty(&file_error).unwrap()
} else {
file_error.to_string()
};
return fail_clierror!("{json_error}");
}
return fail_clierror!("Cannot read file {}: {}", input_path.display(), e);
},
};
let validation_result = validate_single_file_rfc4180(
&mut rdr,
&rconfig,
flag_json,
flag_pretty_json,
args.flag_quiet,
);
match validation_result {
Ok(()) => {
valid_files += 1;
},
Err(e) => {
all_valid = false;
if !args.flag_quiet && input_count > 1 {
woutinfo!("❌ {}: {}", input_path.display(), e);
}
if input_count == 1 {
return Err(e);
}
},
}
}
if !args.flag_quiet && input_count > 1 {
if all_valid {
winfo!("✅ All {} files are valid.", total_files);
} else {
winfo!(
"❌ {} out of {} files are invalid.",
total_files - valid_files,
total_files
);
}
}
if all_valid {
Ok(())
} else if input_count > 1 {
fail_clierror!(
"{} out of {} files failed validation",
total_files - valid_files,
total_files
)
} else {
Err(CliError::Other("Validation failed".to_string()))
}
}
fn validate_single_file_rfc4180(
rdr: &mut csv::Reader<Box<dyn std::io::Read + Send + 'static>>,
rconfig: &Config,
flag_json: bool,
flag_pretty_json: bool,
quiet: bool,
) -> CliResult<()> {
let mut header_msg = String::new();
let mut header_len = 0_usize;
let mut field_vec: Vec<String> = Vec::new();
if !rconfig.no_headers {
let fields_result = rdr.headers();
match fields_result {
Ok(fields) => {
header_len = fields.len();
field_vec.reserve(header_len);
for field in fields {
field_vec.push(field.to_string());
}
let field_list = field_vec.join(r#"", ""#);
header_msg = format!(
"{} Columns: (\"{field_list}\");",
HumanCount(header_len as u64)
);
},
Err(e) => {
if flag_json {
if let csv::ErrorKind::Utf8 { pos, err } = e.kind() {
let header_error = json!({
"errors": [{
"title" : "Header UTF-8 validation error",
"detail" : format!("{e}"),
"meta": {
"record_position": format!("{pos:?}"),
"record_error": format!("{err}"),
}
}]
});
let json_error = if flag_pretty_json {
simd_json::to_string_pretty(&header_error).unwrap()
} else {
header_error.to_string()
};
return fail_encoding_clierror!("{json_error}");
}
let header_error = json!({
"errors": [{
"title" : "Header Validation error",
"detail" : format!("{e}"),
}]
});
let json_error = if flag_pretty_json {
simd_json::to_string_pretty(&header_error).unwrap()
} else {
header_error.to_string()
};
return fail_encoding_clierror!("{json_error}");
}
if let csv::ErrorKind::Utf8 { pos, err } = e.kind() {
return fail_encoding_clierror!(
"non-utf8 sequence detected in header, position {pos:?}.\n{err}\nUse `qsv \
input` to fix formatting and to handle non-utf8 sequences.\n
Alternatively, transcode your data to UTF-8 first using `iconv` or \
`recode`."
);
}
return fail_clierror!("Header Validation error: {e}.");
},
}
}
let mut record = csv::ByteRecord::with_capacity(500, header_len);
let mut result;
let mut record_idx: u64 = 0;
'rfc4180_check: loop {
result = rdr.read_byte_record(&mut record);
if let Err(e) = result {
if flag_json {
let validation_error = json!({
"errors": [{
"title" : "Validation error",
"detail" : format!("{e}"),
"meta": {
"last_valid_record": format!("{record_idx}"),
}
}]
});
let json_error = if flag_pretty_json {
simd_json::to_string_pretty(&validation_error).unwrap()
} else {
validation_error.to_string()
};
return fail!(json_error);
}
if let csv::ErrorKind::UnequalLengths {
expected_len: _,
len: _,
pos: _,
} = e.kind()
{
return fail_clierror!(
"Validation error: {e}.\nUse `qsv fixlengths` to fix record length issues."
);
}
return fail_clierror!("Validation error: {e}.\nLast valid record: {record_idx}");
}
if simdutf8::basic::from_utf8(record.as_slice()).is_err() {
if flag_json {
let validation_error = json!({
"errors": [{
"title" : "UTF-8 validation error",
"detail" : "Cannot parse CSV record as UTF-8",
"meta": {
"last_valid_record": format!("{record_idx}"),
"invalid_record": format!("{record:?}"),
}
}]
});
let json_error = if flag_pretty_json {
simd_json::to_string_pretty(&validation_error).unwrap()
} else {
validation_error.to_string()
};
return fail_encoding_clierror!("{json_error}");
}
return fail_encoding_clierror!(
r#"non-utf8 sequence at record {record_idx}.
Invalid record: {record:?}
Use `qsv input` to fix formatting and to handle non-utf8 sequences.
Alternatively, transcode your data to UTF-8 first using `iconv` or `recode`."#
);
}
if result.is_ok_and(|more_data| !more_data) {
break 'rfc4180_check;
}
record_idx += 1;
}
let msg = if flag_json {
let rfc4180 = RFC4180Struct {
delimiter_char: rconfig.get_delimiter() as char,
header_row: !rconfig.no_headers,
quote_char: rconfig.quote as char,
num_records: record_idx,
num_fields: header_len as u64,
fields: field_vec,
};
if flag_pretty_json {
simd_json::to_string_pretty(&rfc4180).unwrap()
} else {
simd_json::to_string(&rfc4180).unwrap()
}
} else {
let delim_display = if rconfig.get_delimiter() == b'\t' {
"TAB".to_string()
} else {
(rconfig.get_delimiter() as char).to_string()
};
format!(
"Valid: {header_msg} Records: {}; Delimiter: {delim_display}",
HumanCount(record_idx)
)
};
if !quiet {
woutinfo!("{msg}");
}
Ok(())
}
fn split_invalid_records(
rconfig: &Config,
valid_flags: &BitSlice,
headers: &ByteRecord,
input_path: &str,
valid_suffix: &str,
invalid_suffix: &str,
) -> CliResult<()> {
let mut split_row_num: usize = 0;
let mut valid_wtr =
Config::new(Some(input_path.to_owned() + "." + valid_suffix).as_ref()).writer()?;
valid_wtr.write_byte_record(headers)?;
let mut invalid_wtr =
Config::new(Some(input_path.to_owned() + "." + invalid_suffix).as_ref()).writer()?;
invalid_wtr.write_byte_record(headers)?;
let mut rdr = rconfig.reader()?;
let valid_flags_len = valid_flags.len();
let mut record = csv::ByteRecord::new();
while rdr.read_byte_record(&mut record)? {
if split_row_num > valid_flags_len {
break;
}
if valid_flags[split_row_num] {
valid_wtr.write_byte_record(&record)?;
} else {
invalid_wtr.write_byte_record(&record)?;
}
split_row_num += 1;
}
valid_wtr.flush()?;
invalid_wtr.flush()?;
Ok(())
}
fn write_error_report(input_path: &str, validation_error_messages: Vec<String>) -> CliResult<()> {
let wtr_capacitys = env::var("QSV_WTR_BUFFER_CAPACITY")
.unwrap_or_else(|_| DEFAULT_WTR_BUFFER_CAPACITY.to_string());
let wtr_buffer_size: usize = wtr_capacitys.parse().unwrap_or(DEFAULT_WTR_BUFFER_CAPACITY);
let output_file = File::create(input_path.to_owned() + ".validation-errors.tsv")?;
let mut output_writer = BufWriter::with_capacity(wtr_buffer_size, output_file);
output_writer.write_all(b"row_number\tfield\terror\n")?;
for error_msg in validation_error_messages {
output_writer.write_all(error_msg.as_bytes())?;
output_writer.write_all(b"\n")?;
}
output_writer.flush()?;
Ok(())
}
#[inline]
fn to_json_instance(
header_types: &[(String, JSONtypes)],
header_len: usize,
record: &ByteRecord,
) -> CliResult<Value> {
let mut json_object_map = Map::with_capacity(header_len);
let mut json_value;
for ((key, json_type), value) in header_types.iter().zip(record.iter()) {
if value.is_empty() {
json_object_map.insert(key.clone(), Value::Null);
continue;
}
json_value = match json_type {
JSONtypes::String => {
if let Ok(v) = simdutf8::basic::from_utf8(value) {
Value::String(v.to_owned())
} else {
Value::String(String::from_utf8_lossy(value).into_owned())
}
},
JSONtypes::Number => {
if let Ok(float) = fast_float2::parse(value) {
Value::Number(Number::from_f64(float).unwrap_or_else(|| Number::from(0)))
} else {
return fail_clierror!(
"Can't cast to Number. key: {key}, value: {}",
String::from_utf8_lossy(value)
);
}
},
JSONtypes::Integer => {
if let Ok(int) = atoi_simd::parse::<i64>(value) {
Value::Number(Number::from(int))
} else {
return fail_clierror!(
"Can't cast to Integer. key: {key}, value: {}",
String::from_utf8_lossy(value)
);
}
},
JSONtypes::Boolean => match value {
b"true" | b"1" => Value::Bool(true),
b"false" | b"0" => Value::Bool(false),
_ => {
return fail_clierror!(
"Can't cast to Boolean. key: {key}, value: {}",
String::from_utf8_lossy(value)
);
},
},
JSONtypes::Unsupported => unreachable!("we should never get an unsupported JSON type"),
};
json_object_map.insert(key.clone(), json_value);
}
Ok(Value::Object(json_object_map))
}
#[inline]
fn get_json_types(headers: &ByteRecord, schema: &Value) -> CliResult<Vec<(String, JSONtypes)>> {
let Some(schema_properties) = schema.get("properties") else {
return fail_clierror!("JSON Schema missing 'properties' object");
};
let null_type = NULL_TYPE.get().unwrap();
let mut field_def: &Value;
let mut field_type_def: &Value;
let mut json_type: JSONtypes;
let mut header_types: Vec<(String, JSONtypes)> = Vec::with_capacity(headers.len());
for header in headers {
let Ok(key) = simdutf8::basic::from_utf8(header) else {
let s = String::from_utf8_lossy(header);
return fail_encoding_clierror!("CSV header is not valid UTF-8: {s}");
};
field_def = schema_properties.get(key).unwrap_or(&Value::Null);
field_type_def = field_def.get("type").unwrap_or(&Value::Null);
json_type = match field_type_def {
Value::String(s) => match s.as_str() {
"string" => JSONtypes::String,
"number" => JSONtypes::Number,
"integer" => JSONtypes::Integer,
"boolean" => JSONtypes::Boolean,
_ => JSONtypes::Unsupported,
},
Value::Array(vec) => {
let mut return_val = JSONtypes::String;
for val in vec {
if *val == *null_type {
continue;
}
return_val = if let Some(s) = val.as_str() {
match s {
"string" => JSONtypes::String,
"number" => JSONtypes::Number,
"integer" => JSONtypes::Integer,
"boolean" => JSONtypes::Boolean,
_ => JSONtypes::Unsupported,
}
} else {
JSONtypes::String
};
}
return_val
},
_ => JSONtypes::String,
};
header_types.push((key.to_owned(), json_type));
}
Ok(header_types)
}
fn load_json(uri: &str) -> Result<String, String> {
let json_string = match uri {
url if url.to_lowercase().starts_with("http") => {
let client = match util::create_reqwest_blocking_client(
None,
TIMEOUT_SECS.load(Ordering::Relaxed),
Some(uri.to_string()),
) {
Ok(c) => c,
Err(e) => return fail_format!("Cannot build reqwest client: {e}."),
};
match client.get(url).send() {
Ok(response) => response.text().unwrap_or_default(),
Err(e) => return fail_format!("Cannot read JSON at url {url}: {e}."),
}
},
path => {
let mut buffer = String::new();
match File::open(path) {
Ok(p) => {
BufReader::new(p)
.read_to_string(&mut buffer)
.unwrap_or_default();
},
Err(e) => return fail_format!("Cannot read JSON file {path}: {e}."),
}
buffer
},
};
Ok(json_string)
}
#[cfg(test)]
fn validate_json_instance(
instance: &Value,
schema_compiled: &Validator,
) -> Option<Vec<(String, String)>> {
if schema_compiled.is_valid(instance) {
None
} else {
Some(
schema_compiled
.evaluate(instance)
.iter_errors()
.map(|e| (e.instance_location.to_string(), e.error.to_string()))
.collect(),
)
}
}
#[cfg(test)]
mod tests_for_csv_to_json_conversion {
use serde_json::json;
use super::*;
fn schema_json() -> Value {
serde_json::json!({
"$id": "https://example.com/test.schema.json",
"$schema": "https://json-schema.org/draft/2020-12/schema",
"title": "test",
"type": "object",
"properties": {
"A": {
"type": "string",
},
"B": {
"type": "number",
},
"C": {
"type": "integer",
},
"D": {
"type": "boolean",
},
"E": {
"type": ["string", "null"],
},
"F": {
"type": ["number", "null"],
},
"G": {
"type": ["integer", "null"],
},
"H": {
"type": ["boolean", "null"],
},
"I": {
"type": ["string", "null"],
},
"J": {
"type": ["number", "null"],
},
"K": {
"type": ["null", "integer"],
},
"L": {
"type": ["boolean", "null"],
},
}
})
}
#[test]
#[allow(clippy::approx_constant)]
fn test_to_json_instance() {
let _ = NULL_TYPE.get_or_init(|| Value::String("null".to_string()));
let csv = "A,B,C,D,E,F,G,H,I,J,K,L
hello,3.1415,300000000,true,,,,,hello,3.1415,300000000,true";
let mut rdr = csv::Reader::from_reader(csv.as_bytes());
let headers = rdr.byte_headers().unwrap().clone();
let header_types = get_json_types(&headers, &schema_json()).unwrap();
let mut record = rdr.byte_records().next().unwrap().unwrap();
record.trim();
assert_eq!(
to_json_instance(&header_types, headers.len(), &record)
.expect("can't convert csv to json instance"),
json!({
"A": "hello",
"B": 3.1415,
"C": 300_000_000,
"D": true,
"E": null,
"F": null,
"G": null,
"H": null,
"I": "hello",
"J": 3.1415,
"K": 300_000_000,
"L": true,
})
);
}
#[test]
fn test_to_json_instance_cast_integer_error() {
let _ = NULL_TYPE.get_or_init(|| Value::String("null".to_string()));
let csv = "A,B,C,D,E,F,G,H
hello,3.1415,3.0e8,true,,,,";
let mut rdr = csv::Reader::from_reader(csv.as_bytes());
let headers = rdr.byte_headers().unwrap().clone();
let header_types = get_json_types(&headers, &schema_json()).unwrap();
let result = to_json_instance(
&header_types,
headers.len(),
&rdr.byte_records().next().unwrap().unwrap(),
);
assert!(&result.is_err());
let error = result.err().unwrap().to_string();
assert_eq!("Can't cast to Integer. key: C, value: 3.0e8", error);
}
}
#[cfg(test)]
mod tests_for_schema_validation {
use super::*;
fn schema_json() -> Value {
serde_json::json!({
"$id": "https://example.com/person.schema.json",
"$schema": "https://json-schema.org/draft/2020-12/schema",
"title": "Person",
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "The person's title.",
"minLength": 2
},
"name": {
"type": "string",
"description": "The person's name.",
"minLength": 2
},
"age": {
"description": "Age in years which must be equal to or greater than 18.",
"type": "integer",
"minimum": 18
}
}
})
}
fn compiled_schema() -> Validator {
Validator::options()
.build(&schema_json())
.expect("Invalid schema")
}
#[test]
fn test_validate_with_no_errors() {
let _ = NULL_TYPE.get_or_init(|| Value::String("null".to_string()));
let csv = "title,name,age
Professor,Xaviers,60";
let mut rdr = csv::Reader::from_reader(csv.as_bytes());
let headers = rdr.byte_headers().unwrap().clone();
let header_types = get_json_types(&headers, &schema_json()).unwrap();
let record = &rdr.byte_records().next().unwrap().unwrap();
let instance = to_json_instance(&header_types, headers.len(), record).unwrap();
let result = validate_json_instance(&instance, &compiled_schema());
assert!(result.is_none());
}
#[test]
fn test_validate_with_error() {
let _ = NULL_TYPE.get_or_init(|| Value::String("null".to_string()));
let csv = "title,name,age
Professor,X,60";
let mut rdr = csv::Reader::from_reader(csv.as_bytes());
let headers = rdr.byte_headers().unwrap().clone();
let header_types = get_json_types(&headers, &schema_json()).unwrap();
let record = &rdr.byte_records().next().unwrap().unwrap();
let instance = to_json_instance(&header_types, headers.len(), record).unwrap();
let result = validate_json_instance(&instance, &compiled_schema());
assert!(result.is_some());
assert_eq!(
vec![(
"/name".to_string(),
"\"X\" is shorter than 2 characters".to_string()
)],
result.unwrap()
);
}
}
#[test]
fn test_validate_currency_email_dynamicenum_validator() {
#[cfg(not(feature = "lite"))]
let qsv_cache_dir = lookup::set_qsv_cache_dir("~/.qsv-cache").unwrap();
#[cfg(not(feature = "lite"))]
QSV_CACHE_DIR.get_or_init(|| qsv_cache_dir);
fn schema_currency_json() -> Value {
serde_json::json!({
"$id": "https://example.com/person.schema.json",
"$schema": "https://json-schema.org/draft/2020-12/schema",
"title": "Person",
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "The person's title.",
"minLength": 2
},
"name": {
"type": "string",
"description": "The person's name.",
"minLength": 2
},
"fee": {
"description": "The required fee to see the person.",
"type": "string",
"format": "currency",
},
"email": {
"description": "The person's email.",
"type": "string",
"format": "email",
},
"agency": {
"description": "The person's agency.",
"type": "string",
"dynamicEnum": "https://raw.githubusercontent.com/dathere/qsv/refs/heads/master/scripts/NYC_agencies.csv",
}
}
})
}
let _ = NULL_TYPE.get_or_init(|| Value::String("null".to_string()));
let csv = "title,name,fee
Professor,Xaviers,Ð 100.00";
let mut rdr = csv::Reader::from_reader(csv.as_bytes());
let headers = rdr.byte_headers().unwrap().clone();
let header_types = get_json_types(&headers, &schema_currency_json()).unwrap();
let record = &rdr.byte_records().next().unwrap().unwrap();
let instance = to_json_instance(&header_types, headers.len(), record).unwrap();
let compiled_schema = Validator::options()
.with_format("currency", currency_format_checker)
.with_keyword("dynamicEnum", dyn_enum_validator_factory)
.should_validate_formats(true)
.build(&schema_currency_json())
.expect("Invalid schema");
let result = validate_json_instance(&instance, &compiled_schema);
assert_eq!(
result,
Some(vec![(
"/fee".to_owned(),
"\"Ð 100.00\" is not a \"currency\"".to_owned()
)])
);
let csv = "title,name,fee,email
Professor,Xaviers,Ð 100.00,thisisnotanemail";
let mut rdr = csv::Reader::from_reader(csv.as_bytes());
let headers = rdr.byte_headers().unwrap().clone();
let header_types = get_json_types(&headers, &schema_currency_json()).unwrap();
let record = &rdr.byte_records().next().unwrap().unwrap();
let instance = to_json_instance(&header_types, headers.len(), record).unwrap();
let compiled_schema = Validator::options()
.with_format("currency", currency_format_checker)
.with_keyword("dynamicEnum", dyn_enum_validator_factory)
.should_validate_formats(true)
.build(&schema_currency_json())
.expect("Invalid schema");
let result = validate_json_instance(&instance, &compiled_schema);
assert_eq!(
result,
Some(vec![
(
"/fee".to_owned(),
"\"Ð 100.00\" is not a \"currency\"".to_owned()
),
(
"/email".to_owned(),
"\"thisisnotanemail\" is not a \"email\"".to_owned()
)
])
);
let csv = r#"title,name,fee,email,agency
Professor,Xaviers,"USD60.02",x@men.com,DOITT
He-man,Wolverine,"$100.00",claws@men.com,DPR
Mr,Deadpool,"¥1,000,000.00",landfill@nomail.net,DSNY
Mrs,T,"-€ 1.000.000,00",t+sheher@t.com,MODA
Madam,X,"(EUR 1.999.000,12)",x123@aol.com,DOB
SilicoGod,Vision,"1.000.000,00",singularity+is@here.ai,DOITT
Dr,Strange,"€ 1.000.000,00",stranger.danger@xmen.com,NYFD
Dr,Octopus,"WAX 100.000,00",octopussy@bond.net,DFTA
Mr,Robot,"B 1,000,000",71076.964-compuserve,ABCD"#;
let mut rdr = csv::Reader::from_reader(csv.as_bytes());
let headers = rdr.byte_headers().unwrap().clone();
let header_types = get_json_types(&headers, &schema_currency_json()).unwrap();
let compiled_schema = Validator::options()
.with_format("currency", currency_format_checker)
.with_keyword("dynamicEnum", dyn_enum_validator_factory)
.should_validate_formats(true)
.build(&schema_currency_json())
.expect("Invalid schema");
for (i, record) in rdr.byte_records().enumerate() {
let record = record.unwrap();
let instance = to_json_instance(&header_types, headers.len(), &record).unwrap();
let result = validate_json_instance(&instance, &compiled_schema);
match i {
0 => assert_eq!(result, None),
1 => assert_eq!(result, None),
2 => assert_eq!(result, None),
3 => assert_eq!(
result,
Some(vec![
(
"/name".to_owned(),
"\"T\" is shorter than 2 characters".to_owned()
),
(
"/agency".to_owned(),
"\"MODA\" is not a valid dynamicEnum value".to_owned()
)
])
),
4 => assert_eq!(
result,
Some(vec![(
"/name".to_owned(),
"\"X\" is shorter than 2 characters".to_owned()
)])
),
5 => assert_eq!(result, None),
6 => assert_eq!(
result,
Some(vec![(
"/agency".to_owned(),
"\"NYFD\" is not a valid dynamicEnum value".to_owned()
)])
),
7 => assert_eq!(
result,
Some(vec![(
"/fee".to_owned(),
"\"WAX 100.000,00\" is not a \"currency\"".to_owned()
)])
),
8 => assert_eq!(
result,
Some(vec![
(
"/fee".to_owned(),
"\"B 1,000,000\" is not a \"currency\"".to_owned()
),
(
"/email".to_owned(),
"\"71076.964-compuserve\" is not a \"email\"".to_owned()
),
(
"/agency".to_owned(),
"\"ABCD\" is not a valid dynamicEnum value".to_owned()
)
])
),
_ => unreachable!(),
}
}
}
#[test]
fn test_load_json_via_url() {
#[cfg(not(feature = "lite"))]
let qsv_cache_dir = lookup::set_qsv_cache_dir("~/.qsv-cache").unwrap();
#[cfg(not(feature = "lite"))]
QSV_CACHE_DIR.get_or_init(|| qsv_cache_dir);
let json_string_result = load_json("https://geojson.org/schema/FeatureCollection.json");
assert!(&json_string_result.is_ok());
let json_result: Result<Value, serde_json::Error> =
serde_json::from_str(&json_string_result.unwrap());
assert!(&json_result.is_ok());
}
#[test]
fn test_dyn_enum_validator() {
#[cfg(not(feature = "lite"))]
let qsv_cache_dir = lookup::set_qsv_cache_dir("~/.qsv-cache").unwrap();
#[cfg(not(feature = "lite"))]
QSV_CACHE_DIR.get_or_init(|| qsv_cache_dir);
let schema = json!({"dynamicEnum": "https://raw.githubusercontent.com/dathere/qsv/refs/heads/master/resources/test/fruits.csv", "type": "string"});
let validator = jsonschema::options()
.with_keyword("dynamicEnum", dyn_enum_validator_factory)
.build(&schema)
.unwrap();
assert!(validator.is_valid(&json!("banana")));
assert!(validator.is_valid(&json!("strawberry")));
assert!(validator.is_valid(&json!("apple")));
assert!(!validator.is_valid(&json!("Apple")));
assert!(!validator.is_valid(&json!("starapple")));
assert!(!validator.is_valid(&json!("bananana")));
assert!(!validator.is_valid(&json!("")));
assert!(!validator.is_valid(&json!(5)));
match validator.validate(&json!("lanzones")) {
Err(e) => {
assert_eq!(
format!("{e:?}"),
"ValidationError { repr: ValidationErrorRepr { instance: String(\"lanzones\"), \
kind: Custom { keyword: \"dynamicEnum\", message: \"\\\"lanzones\\\" is not a \
valid dynamicEnum value\" }, instance_path: Location(\"\"), schema_path: \
Location(\"/dynamicEnum\"), .. } }"
);
},
_ => {
unreachable!("Expected an error, but validation succeeded.");
},
}
}